zlt
/
tradingSystem
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
tradingSystem/scripts/test_batch7_units.py

627 lines
31 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# -*- coding: utf-8 -*-
"""
第七批模块单测 (零外部依赖, 不连库不触网)
==========================================
运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch7_units.py
覆盖: 上游选股计划接口 (plan_feed) 的解析、新鲜度校验、候选筛选三段纯逻辑。
fixture 用的是 2026-07-30 上游给的真实应答样例 (截取前若干条, 结构一字未改),
坏数据用例另造 —— 「上游改了字段/漏了字段」这类事故必须由单测先炸, 而不是盘中静默少票。
"""
import os
import sys
import traceback
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.services import plan_feed as pf # noqa: E402
RESULTS = []
def case(name):
def deco(fn):
RESULTS.append((name, fn))
return fn
return deco
def raises(exc, fn, *a, **kw):
try:
fn(*a, **kw)
except exc:
return True
except Exception as e:
raise AssertionError(f"期望 {exc.__name__}, 实际 {type(e).__name__}: {e}")
raise AssertionError(f"期望抛 {exc.__name__}, 但没抛")
# ================================================================ fixture
def _m(rank, code, name, score, theme, heat, upside, tier="强传导", n_sources=7, moved=0.0):
return {"rank": rank, "code": code, "name": name, "score": score,
"evidence": {"theme": theme, "n_sources": n_sources, "moved_ratio": moved},
"heat": heat, "upside": upside, "tier": tier}
def _o(rank, code, name, score, theme, heat, n_sources=9, moved=0.075):
return {"rank": rank, "code": code, "name": name, "score": score,
"evidence": {"theme": theme, "n_sources": n_sources, "moved_ratio": moved},
"heat": heat, "upside": None}
SAMPLE = {
"date": "2026-07-29",
"counts": {"main": 961, "observe": 107, "gate_covered": 2386},
"market_snapshot_days": ["2026-07-28"],
"heat_date": "2026-07-27",
"theme_cap": 5,
"main": [
_m(1, "SH600418", "江淮汽车", 242.24, "整车", 0.2249, 2.1203202525935945),
_m(2, "SH688717", "艾罗能源", 242.05, "储能", 0.4165, 1.5817543859649121),
_m(3, "SZ300952", "恒辉安防", 241.91, "传感器", 0.3391, 1.4061712846347607),
_m(4, "SH605598", "上海港湾", 241.72, "储能", 0.214, 1.292594091460947),
_m(6, "SH600875", "东方电气", 241.34, "整机", 0.4074, 1.2340787236824915,
n_sources=9, moved=0.075),
_m(10, "SH600104", "上汽集团", 241.18, "整车", 0.1239, 0.8065614360879017),
],
"observe": [
_o(1, "SH600877", "电科芯片", 101.18, "集成电路设计", 0.4743, moved=0.0),
_o(2, "SH603611", "诺力股份", 101.12, "整机", 0.5868),
_o(3, "SZ000768", "中航西飞", 101.06, "整机", 0.121),
],
"changes": None,
"encoding": "主榜分=200+传导档位×20+组内分（还没热、还便宜）；观察档分=100+0.6z(传导)+0.4z(−热度)",
}
# ================================================================ 解析
@case("解析·真实样例: 计数/主题上限/热度日全部落位, 返回条数与 counts 分开记")
def _():
p = pf.parse_plan(SAMPLE)
assert p["date"] == "2026-07-29" and p["heat_date"] == "2026-07-27"
assert p["theme_cap"] == 5
assert p["counts"] == {"main": 961, "observe": 107, "gate_covered": 2386}
# counts 是上游全量, returned 是这次应答里真的给了几条 —— 混用会看不出上游截断了
assert p["returned"] == {"main": 6, "observe": 3}
assert p["market_snapshot_days"] == ["2026-07-28"]
assert p["encoding"].startswith("主榜分=200")
@case("解析·代码从前缀式归一为点式 (沪/深/科创/创业板四种都要对)")
def _():
p = pf.parse_plan(SAMPLE)
codes = [r["ts_code"] for r in p["main"]]
assert codes[:4] == ["600418.SH", "688717.SH", "300952.SZ", "605598.SH"], codes
assert [r["ts_code"] for r in p["observe"]][2] == "000768.SZ"
@case("解析·字段口径: score/heat/upside 转 float, evidence 拆平, bucket 打标")
def _():
p = pf.parse_plan(SAMPLE)
r = p["main"][0]
assert r["name"] == "江淮汽车" and r["rank"] == 1 and r["tier"] == "强传导"
assert abs(r["score"] - 242.24) < 1e-9 and abs(r["heat"] - 0.2249) < 1e-9
assert abs(r["upside"] - 2.1203202525935945) < 1e-12
assert r["theme"] == "整车" and r["n_sources"] == 7 and r["moved_ratio"] == 0.0
assert r["bucket"] == pf.BUCKET_MAIN and p["observe"][0]["bucket"] == pf.BUCKET_OBSERVE
# 观察档没有 tier、upside 是 null —— 都得是 None 而不是 0
assert p["observe"][0]["tier"] is None and p["observe"][0]["upside"] is None
# 2026-09-02 起的候选卡五键: 这份旧样例没有, 每行都必须是 None (不是 0 / 空串 / 空列表),
# 顶层的版本与生成时刻同理 —— 「旧版计划」与「新版但判为空」在下游必须分得开
for k in ("verdict", "reasons", "missing", "risk", "card_rank"):
assert k in r and r[k] is None, (k, r.get(k))
assert p["observe"][0][k] is None, (k, p["observe"][0][k])
assert p["plan_version"] is None and p["generated_at"] is None
@case("解析·themes 映射: 主榜优先于观察档, 同码不被观察档覆盖")
def _():
d = dict(SAMPLE)
d["observe"] = list(SAMPLE["observe"]) + [_o(9, "SH600418", "江淮汽车", 100.1, "汽车零部件", 0.2)]
p = pf.parse_plan(d)
assert p["themes"]["600418.SH"] == "整车", p["themes"]["600418.SH"]
assert p["themes"]["600877.SH"] == "集成电路设计"
@case("解析·变形应答一律抛 PlanFeedError (不是 JSON 对象 / 缺 date / 两档全空)")
def _():
raises(pf.PlanFeedError, pf.parse_plan, ["不是对象"])
raises(pf.PlanFeedError, pf.parse_plan, None)
raises(pf.PlanFeedError, pf.parse_plan, {"main": SAMPLE["main"]}) # 缺 date
raises(pf.PlanFeedError, pf.parse_plan, {"date": "2026-07-29", "main": [], "observe": []})
raises(pf.PlanFeedError, pf.parse_plan, {"date": " ", "main": SAMPLE["main"]})
@case("解析·脏行跳过而不整体失效: 非 dict / 缺 code / 同码重复 / evidence 非 dict")
def _():
d = {"date": "2026-07-29", "main": [
"我是脏行", None, 42,
{"rank": 1, "code": "SH600418", "score": 1.0, "evidence": "不是对象"},
{"rank": 2, "code": "", "score": 2.0},
{"rank": 3, "code": "600418", "score": 3.0}, # 与第一条同码 (归一后相同)
{"code": "SZ300952", "score": 4.0}, # 缺 rank
]}
p = pf.parse_plan(d)
codes = [r["ts_code"] for r in p["main"]]
assert codes == ["600418.SH", "300952.SZ"], codes
assert p["main"][0]["theme"] is None and p["main"][0]["n_sources"] is None
# rank 缺失用出现序号补 (第 7 个元素 → 7), 免得排序键出现 None
assert p["main"][1]["rank"] == 7, p["main"][1]["rank"]
assert p["counts"] == {"main": None, "observe": None, "gate_covered": None}
# 脏行里也没有候选卡五键 → 全 None, 不许因 evidence 变形连带把这五项算成别的东西
for k in ("verdict", "reasons", "missing", "risk", "card_rank"):
assert p["main"][0][k] is None and p["main"][1][k] is None, k
@case("解析·候选卡五键透传: 判决/理由/缺失/风险/卡内序原样带到候选, 顶层带版本与生成时刻")
def _():
row = dict(_m(1, "SH600418", "江淮汽车", 242.24, "整车", 0.2249, 1.2),
verdict="候选", reasons=["强传导链首环", "券商覆盖 7 家", "目标价高于现价"],
missing=[], risk=[], card_rank=2)
watch = dict(_m(2, "SH688717", "艾罗能源", 242.05, "储能", 0.4, 1.1),
verdict="关注", reasons=["强传导"], missing=["确认线缺失"], risk=["昨夜信号陈旧"],
card_rank=5)
d = {"date": "2026-09-02", "plan_version": "2026-09-02.3",
"generated_at": "2026-09-02T06:31:07", "main": [row, watch]}
p = pf.parse_plan(d)
assert p["plan_version"] == "2026-09-02.3" and p["generated_at"] == "2026-09-02T06:31:07"
r = p["main"][0]
assert r["verdict"] == "候选" and r["card_rank"] == 2, r
assert r["reasons"] == ["强传导链首环", "券商覆盖 7 家", "目标价高于现价"], r["reasons"]
assert r["missing"] == [] and r["risk"] == [], r # 空列表保留为空列表, 不折成 None
w = p["main"][1]
assert w["verdict"] == "关注" and w["missing"] == ["确认线缺失"] and w["risk"] == ["昨夜信号陈旧"]
# 筛选后的候选原样带着这五项 (动作引擎靠它们进硬数字)
items = {x["ts_code"]: x for x in pf.select_candidates(p, top_n=5)["items"]}
assert items["600418.SH"]["verdict"] == "候选" and items["600418.SH"]["card_rank"] == 2
assert items["600418.SH"]["reasons"] == r["reasons"] and items["600418.SH"]["risk"] == []
assert items["688717.SH"]["verdict"] == "关注" and items["688717.SH"]["missing"] == ["确认线缺失"]
# 默认不按判决分流: 关注照样进池, 也没有 display_only
assert set(items) == {"600418.SH", "688717.SH"} and pf.select_candidates(p)["display_only"] == []
@case("解析·候选卡键缺失一律 None (旧版计划一个键都没有); 空串同缺失, 空列表要保留")
def _():
d = {"date": "2026-09-02", "main": [
_m(1, "SH600418", "甲", 100.0, "整车", 0.1, 1.0), # 一个键都没有
dict(_m(2, "SH600419", "乙", 99.0, "整车", 0.1, 1.0),
verdict=" ", reasons=None, missing="", risk=[], card_rank=""), # 键在但是空的
dict(_m(3, "SH600420", "丙", 98.0, "整车", 0.1, 1.0),
verdict="候选", reasons={"不是": "列表"}, missing=42, risk="单条风险", card_rank="7")]}
p = pf.parse_plan(d)
a, b, c = p["main"]
for k in ("verdict", "reasons", "missing", "risk", "card_rank"):
assert a[k] is None, (k, a[k])
assert b["verdict"] is None and b["reasons"] is None and b["missing"] is None, b
assert b["risk"] == [] and b["card_rank"] is None, b # [] 是「上游说没有风险」, 要留
assert c["verdict"] == "候选" and c["reasons"] is None and c["missing"] is None, c
assert c["risk"] == ["单条风险"] and c["card_rank"] == 7, c # 单个字符串当一项, 数字串可用
@case("解析·reasons 只留前四条 (硬数字/账本/研判/提议卡都只要最要紧的几条); missing 与 risk 不截")
def _():
assert pf.REASONS_KEEP == 4
many = ["理由一", "", "理由二", None, " 理由三 ", "理由四", "理由五", "理由六"]
d = {"date": "2026-09-02", "main": [
dict(_m(1, "SH600418", "甲", 100.0, "整车", 0.1, 1.0),
verdict="候选", reasons=many, missing=list(many), risk=[f"风险{i}" for i in range(6)])]}
r = pf.parse_plan(d)["main"][0]
assert r["reasons"] == ["理由一", "理由二", "理由三", "理由四"], r["reasons"] # 空项去掉、空白去掉、只留四条
assert r["missing"] == ["理由一", "理由二", "理由三", "理由四", "理由五", "理由六"], r["missing"]
assert len(r["risk"]) == 6, r["risk"]
x = pf.select_candidates(pf.parse_plan(d), top_n=1)["items"][0]
assert x["reasons"] == r["reasons"] and len(x["risk"]) == 6
@case("解析·数值容错: 字符串数字可用, 空串/None/非数字一律 None 而不是 0")
def _():
d = {"date": "2026-07-29", "main": [
{"rank": "5", "code": "SH600418", "score": "242.24", "heat": "", "upside": None,
"evidence": {"theme": " 整车 ", "n_sources": "7", "moved_ratio": "abc"}}]}
r = pf.parse_plan(d)["main"][0]
assert r["rank"] == 5 and abs(r["score"] - 242.24) < 1e-9
assert r["heat"] is None and r["upside"] is None and r["moved_ratio"] is None
assert r["n_sources"] == 7 and r["theme"] == "整车" # theme 两头空白要去掉
@case("解析·漏斗: counts 是打分池, returned 是过筛后 —— 两者相减没有意义, 只做展示")
def _():
p = pf.parse_plan(SAMPLE)
assert p["funnel"] == {"scored_main": 961, "returned_main": 6,
"scored_observe": 107, "returned_observe": 3}, p["funnel"]
@case("解析·吃满才叫截断: 拿 counts 判会永远报警 (961 vs 55 是两把尺子)")
def _():
# 上游签名默认 top=20 / obs_top=10。样例只有 6+3 条 —— 没吃满, 就是上游能给的全部,
# 尽管 counts 写着 961。**这正是不能拿 counts 判截断的原因。**
p = pf.parse_plan(SAMPLE)
assert p["requested"] == {"top": 20, "obs_top": 10, "theme_cap": None}, p["requested"]
assert p["truncated"] == {"main": False, "observe": False}, p["truncated"]
# 要了 6 条正好给 6 条 → 吃满, 可能还有更多
q = pf.parse_plan(SAMPLE, requested={"top": 6, "obs_top": 3, "theme_cap": 999})
assert q["truncated"] == {"main": True, "observe": True}, q["truncated"]
assert q["requested"]["theme_cap"] == 999
# 要 100 条只给 6 条 → 没吃满
r = pf.parse_plan(SAMPLE, requested={"top": 100, "obs_top": 100})
assert r["truncated"] == {"main": False, "observe": False}
# top=0 (不传) 时无从判断, 不许瞎报
z = pf.parse_plan(SAMPLE, requested={"top": 0, "obs_top": 0})
assert z["truncated"] == {"main": False, "observe": False}
@case("取数·附加查询串: 解析 / 合并进请求 / date 不可被覆盖 / 坏串忽略不炸")
def _():
assert pf.parse_query_extra("limit=1000") == {"limit": "1000"}
assert pf.parse_query_extra("?limit=1000&offset=20") == {"limit": "1000", "offset": "20"}
assert pf.parse_query_extra("") == {} and pf.parse_query_extra(None) == {}
assert pf.parse_query_extra(" ") == {}
assert pf.parse_query_extra("这不是查询串") == {} # 解析不出 k=v → 空, 不抛
calls, restore = _with_fake_requests(lambda *a: _FakeResp(SAMPLE))
try:
pf.fetch(base="http://x:8300", path="/plan", timeout=3,
extra_params={"limit": "1000", "date": "被覆盖的"}, date="2026-07-29")
finally:
restore()
# date 是 PMS 自己的语义, extra 里同名键不许赢
assert calls[0]["params"] == {"limit": "1000", "date": "2026-07-29"}, calls[0]["params"]
# ================================================================ 新鲜度
@case("新鲜度·日龄: 当天 0 / 上一交易日 1 / 未来日期按 0 (为下一交易日出的计划)")
def _():
assert pf.plan_age_tdays("2026-07-30", today="2026-07-30") == 0
assert pf.plan_age_tdays("2026-07-29", today="2026-07-30") == 1
assert pf.plan_age_tdays("2026-07-27", today="2026-07-30") == 3
assert pf.plan_age_tdays("2026-07-31", today="2026-07-30") == 0
@case("新鲜度·跨周末只算交易日: 周五的计划到周一仍是 1 个交易日龄")
def _():
# 2026-07-24 周五, 2026-07-27 周一 —— 中间两天不是交易日, 不该被算进日龄
assert pf.plan_age_tdays("2026-07-24", today="2026-07-27") == 1
assert pf.plan_age_tdays("2026-07-24", today="2026-07-28") == 2
@case("新鲜度·超期抛错并说清日龄 (上游停更时拿旧榜当今天比没候选更危险)")
def _():
p = pf.parse_plan(SAMPLE) # date = 2026-07-29
assert pf.assert_fresh(p, max_stale_tdays=1, today="2026-07-30") == 1
assert pf.assert_fresh(p, max_stale_tdays=0, today="2026-07-29") == 0
raises(pf.PlanFeedError, pf.assert_fresh, p, max_stale_tdays=0, today="2026-07-30")
try:
pf.assert_fresh(p, max_stale_tdays=1, today="2026-08-05")
except pf.PlanFeedError as e:
assert "2026-07-29" in str(e) and "交易日" in str(e), str(e)
else:
raise AssertionError("超期计划没有抛 PlanFeedError (2026-08-28 补: 原来不抛也静默通过)")
# ================================================================ 筛选
@case("筛选·按 score 降序取前 N, 同分按上游 rank 稳定次序")
def _():
p = pf.parse_plan(SAMPLE)
r = pf.select_candidates(p, top_n=3)
assert [x["ts_code"] for x in r["items"]] == ["600418.SH", "688717.SH", "300952.SZ"]
assert r["considered"] == 6 and r["eligible"] == 6 and r["dropped"]["capped"] == 3
# 同分: rank 小的在前
d = {"date": "2026-07-29", "main": [_m(9, "SH600001", "甲", 100.0, "整车", 0.1, 1.0),
_m(2, "SH600002", "乙", 100.0, "整车", 0.1, 1.0)]}
rr = pf.select_candidates(pf.parse_plan(d), top_n=2)
assert [x["ts_code"] for x in rr["items"]] == ["600002.SH", "600001.SH"]
@case("筛选·top_n=0 视为不截断 (别把'不限'写成'一只都不要')")
def _():
p = pf.parse_plan(SAMPLE)
assert len(pf.select_candidates(p, top_n=0)["items"]) == 6
assert len(pf.select_candidates(p, top_n=999)["items"]) == 6
@case("筛选·已持有与黑名单剔除, 且两者分开计数 (输入接受前缀式)")
def _():
p = pf.parse_plan(SAMPLE)
r = pf.select_candidates(p, held=["SH600418", "688717.SH"], black=["SZ300952"], top_n=10)
codes = [x["ts_code"] for x in r["items"]]
assert "600418.SH" not in codes and "688717.SH" not in codes and "300952.SZ" not in codes
assert r["dropped"]["held"] == 2 and r["dropped"]["black"] == 1
assert r["eligible"] == 3
@case("筛选·tier 白名单只对带 tier 的行生效; 观察档的闸门是 include_observe")
def _():
d = dict(SAMPLE)
d["main"] = list(SAMPLE["main"]) + [
_m(500, "SH600519", "弱票", 220.0, "白酒", 0.1, 0.5, tier="弱传导")]
p = pf.parse_plan(d)
r = pf.select_candidates(p, tiers=["强传导"], top_n=50)
assert "600519.SH" not in [x["ts_code"] for x in r["items"]]
assert r["dropped"]["tier"] == 1
# 观察档没有 tier —— tiers 非空时也不该被 tier 规则误杀, 但默认根本不进池
assert all(x["bucket"] == "main" for x in r["items"])
r2 = pf.select_candidates(p, tiers=["强传导"], include_observe=True, top_n=50)
obs = [x["ts_code"] for x in r2["items"] if x["bucket"] == "observe"]
assert obs == ["600877.SH", "603611.SH", "000768.SZ"], obs
assert r2["dropped"]["tier"] == 1
@case("筛选·跨档同码只留一次, 且留主榜那份 (主榜分高, 排序天然优先)")
def _():
d = dict(SAMPLE)
d["observe"] = list(SAMPLE["observe"]) + [_o(9, "SH600418", "江淮汽车", 100.1, "汽车零部件", 0.2)]
r = pf.select_candidates(pf.parse_plan(d), include_observe=True, top_n=50)
hit = [x for x in r["items"] if x["ts_code"] == "600418.SH"]
assert len(hit) == 1 and hit[0]["bucket"] == "main" and hit[0]["theme"] == "整车"
assert r["dropped"]["dup"] == 1
@case("筛选·score 下限与 n_sources 下限各自独立计数")
def _():
p = pf.parse_plan(SAMPLE)
r = pf.select_candidates(p, min_score=241.7, top_n=50)
assert [x["ts_code"] for x in r["items"]] == ["600418.SH", "688717.SH", "300952.SZ",
"605598.SH"]
assert r["dropped"]["score"] == 2
r2 = pf.select_candidates(p, min_sources=9, top_n=50)
assert [x["ts_code"] for x in r2["items"]] == ["600875.SH"], r2["items"]
assert r2["dropped"]["sources"] == 5
@case("筛选·upside 下限 (2.12=+212%): 对所有行生效, 观察档因 upside=null 被一并挡掉")
def _():
d = dict(SAMPLE)
d["observe"] = list(SAMPLE["observe"])
p = pf.parse_plan(d)
r = pf.select_candidates(p, min_upside=1.3, top_n=50)
assert [x["ts_code"] for x in r["items"]] == ["600418.SH", "688717.SH", "300952.SZ"], r["items"]
assert r["dropped"]["upside"] == 3 # 1.2926 / 1.2341 / 0.8066 三只在门槛下
# 观察档 upside 恒为 null → 按 0 算, 设了下限就全挡掉 (方向保守: 宁可少票)
r2 = pf.select_candidates(p, min_upside=1.3, include_observe=True, top_n=50)
assert all(x["bucket"] == "main" for x in r2["items"]), r2["items"]
assert r2["dropped"]["upside"] == 6
# 不设下限时观察档照旧能进 (闸门只有 include_observe)
r3 = pf.select_candidates(p, include_observe=True, top_n=50)
assert any(x["bucket"] == "observe" for x in r3["items"])
assert r3["dropped"]["upside"] == 0
@case("筛选·upside 永不参与排序 (券商目标价噪音大, 排序只认 score)")
def _():
d = {"date": "2026-07-29", "main": [
_m(1, "SH600001", "高分低空间", 242.0, "整车", 0.1, 0.30),
_m(2, "SH600002", "低分高空间", 200.0, "整车", 0.1, 5.00)]}
r = pf.select_candidates(pf.parse_plan(d), top_n=2)
assert [x["ts_code"] for x in r["items"]] == ["600001.SH", "600002.SH"], r["items"]
@case("筛选·ST 剔除 (上游明确不剔, 这道闸归 PMS): ST/*ST/退市 都要挡, 名字缺失不误杀")
def _():
assert pf.is_st("ST喜临门") and pf.is_st("*ST辉丰") and pf.is_st("S*ST前锋")
assert pf.is_st("SST华塑") and pf.is_st("*st 辉丰") # 大小写与空格都要吃掉
assert pf.is_st("退市海润") and pf.is_st("海润退") # 退市整理期两种写法
assert not pf.is_st("江淮汽车") and not pf.is_st("") and not pf.is_st(None)
# 「圣农发展」「深赛格」这类 S 开头的中文名不许误杀 (A 股名称都是中文, 拉丁 ST 前缀安全)
assert not pf.is_st("圣农发展") and not pf.is_st("深赛格") and not pf.is_st("三安光电")
d = {"date": "2026-07-29", "main": [
_m(1, "SH600001", "ST喜临门", 240.0, "整机", 0.1, 1.0),
_m(2, "SH600002", "*ST辉丰", 239.0, "整机", 0.1, 1.0),
_m(3, "SH600003", "江淮汽车", 238.0, "整车", 0.1, 1.0),
{"rank": 4, "code": "SH600004", "score": 237.0, "tier": "强传导",
"evidence": {"theme": "整车"}}]} # 没有 name
p = pf.parse_plan(d)
r = pf.select_candidates(p, top_n=10, exclude_st=True)
assert [x["ts_code"] for x in r["items"]] == ["600003.SH", "600004.SH"], r["items"]
assert r["dropped"]["st"] == 2
assert r["st_unknown"] == ["600004.SH"], r["st_unknown"] # 判不了要报出来, 不静默
assert len(pf.select_candidates(p, top_n=10, exclude_st=False)["items"]) == 4
@case("筛选·白名单档位取全判据: 看见白名单之外的档位才算齐 (吃满 top 不等于漏了)")
def _():
strong = [_m(i, f"SH60{i:04d}", f"强{i}", 240 - i, "整车", 0.1, 1.0) for i in range(1, 4)]
weak = [_m(9, "SH609999", "弱", 220.0, "整车", 0.1, 1.0, tier="弱传导")]
# 只有强传导 → 判不出后面还有没有, tier_complete=False (被 top 卡住了)
only_strong = pf.select_candidates(pf.parse_plan({"date": "2026-07-29", "main": strong}),
tiers=["强传导"], top_n=10)
assert only_strong["tier_complete"] is False
# 出现弱传导 → 强传导已经全在前面了
both = pf.select_candidates(pf.parse_plan({"date": "2026-07-29", "main": strong + weak}),
tiers=["强传导"], top_n=10)
assert both["tier_complete"] is True
# 没设白名单时无从判断
assert pf.select_candidates(pf.parse_plan({"date": "2026-07-29", "main": strong}),
top_n=10)["tier_complete"] is None
@case("筛选·候选级主题限额在 top_n 截断之前生效 (否则前 N 名被单一主题垄断)")
def _():
# 造一个"宽池子": 储能 6 只分最高, 传感器 3 只, 整车 2 只
rows = ([_m(i, f"SH60{i:04d}", f"储{i}", 300 - i, "储能", 0.2, 1.0) for i in range(1, 7)]
+ [_m(10 + i, f"SH61{i:04d}", f"传{i}", 200 - i, "传感器", 0.2, 1.0) for i in range(1, 4)]
+ [_m(20 + i, f"SH62{i:04d}", f"整{i}", 100 - i, "整车", 0.2, 1.0) for i in range(1, 3)])
p = pf.parse_plan({"date": "2026-07-29", "main": rows})
# 不限主题: 前 5 名全是储能 —— 规则闸按 PMS_SECTOR_MAX_NAMES 一拦就废掉大半
plain = pf.select_candidates(p, top_n=5)
assert {x["theme"] for x in plain["items"]} == {"储能"}, plain["items"]
# 限 2 只/主题: 先摊开再截断, 5 个名额分到三个主题
capped = pf.select_candidates(p, top_n=5, theme_cap=2)
got = [(x["theme"], x["ts_code"]) for x in capped["items"]]
assert [t for t, _ in got] == ["储能", "储能", "传感器", "传感器", "整车"], got
assert capped["dropped"]["theme"] == 5 # 储能多 4 只 + 传感器多 1 只
# 主题为空的行归到「(无主题)」一档, 不跟着别的主题挤
p2 = pf.parse_plan({"date": "2026-07-29", "main": [
{"rank": 1, "code": "SH600001", "score": 9}, {"rank": 2, "code": "SH600002", "score": 8}]})
assert len(pf.select_candidates(p2, top_n=5, theme_cap=1)["items"]) == 1
@case("取数·build_query: 显式 top/obs_top/theme_cap 覆盖 QUERY_EXTRA 同名键; 0=不传")
def _():
assert pf.build_query({"top": 300, "obs_top": 100, "theme_cap": 999, "query_extra": {}}) == \
{"top": "300", "obs_top": "100", "theme_cap": "999"}
# 0 = 不传该参数 (用上游默认)
assert pf.build_query({"top": 0, "obs_top": 0, "theme_cap": 0, "query_extra": {}}) == {}
# QUERY_EXTRA 里的同名键输给显式参数; 非同名键照旧透传
q = pf.build_query({"top": 300, "obs_top": 0, "theme_cap": 0,
"query_extra": {"top": "20", "foo": "bar"}})
assert q == {"top": "300", "foo": "bar"}, q
@case("筛选·输出字段: sector 用 theme 灌 (planner 吃这个), score 缺失兜 0.0 不留 None")
def _():
d = {"date": "2026-07-29", "main": [{"rank": 1, "code": "SH600418",
"evidence": {"theme": "整车"}}]}
x = pf.select_candidates(pf.parse_plan(d), top_n=5)["items"][0]
assert x["score"] == 0.0 and x["sector"] == "整车" and x["theme"] == "整车"
assert x["src"] == "plan_api" and x["ts_code"] == "600418.SH"
assert "price" not in x, "计划不带价格, 这里绝不能凭空造一个价出来"
# 候选卡五键: 键必须在 (下游按键取, 不按 in 判), 旧版计划里值是 None
for k in ("verdict", "reasons", "missing", "risk", "card_rank"):
assert k in x and x[k] is None, (k, x.get(k))
@case("筛选·计数自洽: considered = eligible + 各项丢弃 (含仅展示); items = eligible − capped")
def _():
d = dict(SAMPLE)
d["main"] = list(SAMPLE["main"]) + [
_m(500, "SH600519", "弱票", 220.0, "白酒", 0.1, 0.5, tier="弱传导"),
# 2026-09-02 起的判决: 仅展示的行按判决分流时在候选阶段剔掉, 单独计数
dict(_m(600, "SH600600", "展示票", 242.50, "整车", 0.1, 0.9), verdict="仅展示")] # 分最高
p = pf.parse_plan(d)
r = pf.select_candidates(p, held=["SH600418"], black=["SZ300952"], tiers=["强传导"],
min_sources=7, top_n=2, include_observe=True, theme_cap=2,
route_by_verdict=True)
dr = r["dropped"]
assert r["considered"] == 11, r["considered"]
assert r["considered"] == r["eligible"] + dr["held"] + dr["black"] + dr["tier"] \
+ dr["score"] + dr["sources"] + dr["upside"] + dr["theme"] + dr["dup"] \
+ dr["display"], (r, dr)
assert dr["display"] == 1 and r["display_only"] == ["600600.SH"], (dr, r["display_only"])
assert len(r["items"]) == r["eligible"] - dr["capped"] == 2
# 分流开关关着: 那只仅展示的票照常进池 (分数高, 会排到第一), 计数里 display 是 0
r0 = pf.select_candidates(p, held=["SH600418"], black=["SZ300952"], tiers=["强传导"],
min_sources=7, top_n=2, include_observe=True, theme_cap=2)
assert r0["dropped"]["display"] == 0 and r0["display_only"] == []
assert r0["items"][0]["ts_code"] == "600600.SH", r0["items"]
@case("筛选·空计划不炸 (main/observe 缺键) 且 date 透传")
def _():
r = pf.select_candidates({"date": "2026-07-29"}, top_n=5)
assert r["items"] == [] and r["considered"] == 0 and r["eligible"] == 0
assert r["date"] == "2026-07-29"
# ================================================================ 取数守卫
class _FakeResp:
def __init__(self, payload, status=200):
self._p, self.status_code = payload, status
def raise_for_status(self):
if self.status_code >= 400:
raise RuntimeError(f"HTTP {self.status_code}")
def json(self):
if isinstance(self._p, Exception):
raise self._p
return self._p
def _with_fake_requests(handler):
"""把 requests 换成假模块 (装在 sys.modules 上, plan_feed 是函数内 import)。"""
import types
calls = []
fake = types.ModuleType("requests")
def _get(url, params=None, timeout=None):
calls.append({"url": url, "params": params, "timeout": timeout})
return handler(url, params, timeout)
fake.get = _get
prev = sys.modules.get("requests")
sys.modules["requests"] = fake
def restore():
if prev is None:
sys.modules.pop("requests", None)
else:
sys.modules["requests"] = prev
return calls, restore
@case("取数·base 为空立即抛 PlanFeedError, 一个 HTTP 请求都不发")
def _():
calls, restore = _with_fake_requests(
lambda *a: (_ for _ in ()).throw(AssertionError("不该发请求")))
try:
raises(pf.PlanFeedError, pf.fetch, base="", path="/plan", timeout=1)
raises(pf.PlanFeedError, pf.fetch, base=" ", path="/plan", timeout=1)
finally:
restore()
assert calls == [], calls
@case("取数·成功路径: URL 拼接/date 传参/超时透传, 并盖上 url 与 fetched_at")
def _():
calls, restore = _with_fake_requests(lambda *a: _FakeResp(SAMPLE))
try:
p = pf.fetch(base="http://192.168.16.155:8300/", path="plan", timeout=7,
date="2026-07-29")
finally:
restore()
assert calls[0]["url"] == "http://192.168.16.155:8300/plan", calls
assert calls[0]["params"] == {"date": "2026-07-29"} and calls[0]["timeout"] == 7
assert p["date"] == "2026-07-29" and p["requested_date"] == "2026-07-29"
assert p["url"].endswith("/plan") and p["fetched_at"] > 0
# 不传 date 时不能带一个 date=None 的空参上去
calls2, restore2 = _with_fake_requests(lambda *a: _FakeResp(SAMPLE))
try:
pf.fetch(base="http://x:8300", path="/plan", timeout=3)
finally:
restore2()
assert calls2[0]["params"] is None, calls2
@case("取数·HTTP 报错/JSON 坏掉一律包装成 PlanFeedError 并带上 URL")
def _():
for handler in (lambda *a: _FakeResp(SAMPLE, status=500),
lambda *a: _FakeResp(ValueError("Expecting value")),
lambda *a: (_ for _ in ()).throw(OSError("Connection refused"))):
calls, restore = _with_fake_requests(handler)
try:
try:
pf.fetch(base="http://192.168.16.155:8300", path="/plan", timeout=1)
raise AssertionError("应当抛 PlanFeedError")
except pf.PlanFeedError as e:
assert "192.168.16.155:8300/plan" in str(e), str(e)
finally:
restore()
@case("取数·来源常量与旧表来源都在: plan_api / buy_plan / both")
def _():
assert (pf.SRC_PLAN_API, pf.SRC_BUY_PLAN, pf.SRC_BOTH) == ("plan_api", "buy_plan", "both")
assert pf.FAIL_CACHE_SEC > 0, "失败也要缓存一会儿, 否则每分钟的调度位会把超时叠成雪崩"
def main():
import logging
logging.disable(logging.CRITICAL)
passed, failed = 0, 0
for name, fn in RESULTS:
try:
fn()
print(f" PASS {name}")
passed += 1
except Exception:
print(f" FAIL {name}")
traceback.print_exc()
failed += 1
print("-" * 60)
if failed:
print(f"FAILED: {failed} / {passed + failed}")
sys.exit(1)
print(f"ALL PASS ({passed} cases)")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink