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

764 lines
34 KiB
Python
Raw Normal View History

#7 和 #12
2026-07-31 13:28:10 +08:00
# -*- coding: utf-8 -*-
"""
第八批模块单测 (零外部依赖, 不连库不触网)
==========================================
运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch8_units.py
覆盖: 上游榜单的版本比对 (app/core/plan_diff.py) —— 名册与指纹、三种比对语义、
进出榜的尾部闸、档位升降与券商覆盖翻转、名次跳变、持仓票视图。
这一批的重点不是"能不能算出差异", 而是**不报假变化**: 榜单是按 top 截断过的, 尾部的进出
是截断噪音而不是上游观点变化。假警报天天响, 提示就等于没有 —— 与 `_capped` 治的是同一个病。
"""
import os
import sys
import traceback
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.core import plan_diff as pd # noqa: E402
RESULTS = []
def case(name):
def deco(fn):
RESULTS.append((name, fn))
return fn
return deco
# ================================================================ fixture
def _r(code, rank, score, tier="强传导", theme="储能", bucket="main", name=None):
"""名册单行 (plan_diff 内部形态)。"""
return {"c": code, "n": name or code[:6], "r": rank, "s": score,
"t": tier, "h": theme, "b": bucket}
def roster(*rows) -> dict:
return {r["c"]: r for r in rows}
def plan(main=(), observe=(), date="2026-07-30", capped_main=False, capped_obs=False) -> dict:
"""plan_feed.parse_plan() 结果的最小形态。
`returned` / `truncated` 也要带上 —— 快照落库时靠它们记条数与截断标志, 少了的话
"上一版坏没坏"这类判据在单测里恒为假, 用例看着过了实际什么都没验。
"""
def row(x, bucket):
return {"ts_code": x["c"], "name": x["n"], "rank": x["r"], "score": x["s"],
"tier": x["t"], "theme": x["h"], "bucket": bucket}
return {"date": date,
"main": [row(x, "main") for x in main],
"observe": [row(x, "observe") for x in observe],
"returned": {"main": len(main), "observe": len(observe)},
"truncated": {"main": bool(capped_main), "observe": bool(capped_obs)}}
# ================================================================ [A] 名册与指纹
@case("[A1] roster_of 从计划结构取出名册")
def t_a1():
ro = pd.roster_of(plan(main=[_r("600418.SH", 1, 242.2)],
observe=[_r("600877.SH", 1, 101.1, tier=None, bucket="observe")]))
assert set(ro) == {"600418.SH", "600877.SH"}
assert ro["600418.SH"]["b"] == "main" and ro["600877.SH"]["b"] == "observe"
assert ro["600418.SH"]["t"] == "强传导" and ro["600877.SH"]["t"] is None
@case("[A2] 同一只票两档都出现时以主榜为准")
def t_a2():
p = plan(main=[_r("600418.SH", 3, 240.0)],
observe=[_r("600418.SH", 1, 99.0, tier=None, bucket="observe")])
ro = pd.roster_of(p)
assert len(ro) == 1
assert ro["600418.SH"]["b"] == "main", "主榜在前, 观察档那条不该覆盖它"
@case("[A3] 指纹稳定: 同一名册两次算结果一致")
def t_a3():
a = roster(_r("600000.SH", 1, 10.0), _r("600418.SH", 2, 9.0))
b = roster(_r("600418.SH", 2, 9.0), _r("600000.SH", 1, 10.0)) # 插入次序不同
assert pd.digest_of(a) == pd.digest_of(b), "指纹不能依赖 dict 的插入次序"
@case("[A4] 指纹不含 score —— 分数抖动但次序没变, 不算新版本")
def t_a4():
a = roster(_r("600000.SH", 1, 242.2801))
b = roster(_r("600000.SH", 1, 242.2799))
assert pd.digest_of(a) == pd.digest_of(b), (
"/plan 是实时算的, score 末位天天抖; 算进指纹会让快照表白涨而信息量为零")
@case("[A5] 指纹含 rank —— 次序变了就是新版本")
def t_a5():
assert pd.digest_of(roster(_r("600000.SH", 1, 10.0))) != \
pd.digest_of(roster(_r("600000.SH", 2, 10.0)))
@case("[A6] 指纹含 tier / theme / bucket")
def t_a6():
base = roster(_r("600000.SH", 1, 10.0))
assert pd.digest_of(base) != pd.digest_of(roster(_r("600000.SH", 1, 10.0, tier="弱传导")))
assert pd.digest_of(base) != pd.digest_of(roster(_r("600000.SH", 1, 10.0, theme="整机制造")))
assert pd.digest_of(base) != pd.digest_of(roster(_r("600000.SH", 1, 10.0, bucket="observe")))
@case("[A7] 名册 ↔ 落库行 往返不丢字段")
def t_a7():
ro = roster(_r("600418.SH", 7, 233.3, tier="弱传导", theme="整车"))
back = pd.roster_from_rows(pd.roster_rows(ro))
assert back == ro
assert pd.digest_of(back) == pd.digest_of(ro)
@case("[A8] 落库行里的坏数据跳过而不是整份废掉")
def t_a8():
back = pd.roster_from_rows([{"c": "600000.SH", "r": 1}, "坏行", None, {}, {"r": 5}])
assert list(back) == ["600000.SH"]
@case("[A9] 名册行按代码序输出 —— 落库文本稳定")
def t_a9():
rows = pd.roster_rows(roster(_r("600418.SH", 2, 9.0), _r("600000.SH", 1, 10.0)))
assert [r["c"] for r in rows] == ["600000.SH", "600418.SH"]
# ================================================================ [B] 档位序
@case("[B1] 档位强弱: 强传导 > 弱传导 > 无传导")
def t_b1():
assert pd.tier_rank("强传导") > pd.tier_rank("弱传导") > pd.tier_rank("无传导")
@case("[B2] 未知档位返回 None (判不了就不判)")
def t_b2():
assert pd.tier_rank("超强传导") is None and pd.tier_rank(None) is None
@case("[B3] 未知档位不参与升降判定 —— 宁可不报, 不能报反")
def t_b3():
d = pd.diff(roster(_r("600000.SH", 1, 10.0, tier="超强传导")),
roster(_r("600000.SH", 1, 10.0, tier="强传导")),
prev_date="2026-07-29", curr_date="2026-07-30")
assert d["counts"]["tier_up"] == 0 and d["counts"]["tier_down"] == 0
# ================================================================ [C] 首次
@case("[C1] 没有上一版时一条变化都不报")
def t_c1():
d = pd.diff(None, roster(*[_r(f"60000{i}.SH", i, 100 - i) for i in range(1, 9)]),
curr_date="2026-07-30")
assert d["kind"] == pd.KIND_FIRST
assert sum(d["counts"].values()) == 0, "首次落快照若报'新进榜', 会是整整一榜的假变化"
assert d["roster_size"]["curr"] == 8
@case("[C2] 首次的 note 说明从下一次开始才有比对")
def t_c2():
d = pd.diff({}, roster(_r("600000.SH", 1, 10.0)))
assert "首次" in d["note"] and d["prev_broken"] is False
@case("[C3] 上一版读不出来要说成故障, 不能长得跟'首次'一样")
def t_c3():
# 快照的 roster_json 坏了 → 解析成空。若与"首次"同形, 一条真实变化都不报却显示正常。
d = pd.diff({}, roster(_r("600000.SH", 1, 10.0)),
prev_date="2026-07-29", curr_date="2026-07-30", prev_broken=True)
assert d["prev_broken"] is True
assert "读不出来" in d["note"] and "比不了" in d["note"]
assert "首次" not in d["note"]
@case("[C4] 首次也要回显调用方给的参数, 不能写死默认值")
def t_c4():
d = pd.diff(None, roster(_r("600000.SH", 1, 10.0)), rank_jump=80, tail_guard=0.2)
assert d["params"] == {"rank_jump": 80, "tail_guard": 0.2}, "写死会让页面显示错的口径"
# ================================================================ [D] 三种比对语义
@case("[D1] 计划日不同 = 跨日的正常升降档")
def t_d1():
d = pd.diff(roster(_r("600000.SH", 1, 10.0)), roster(_r("600000.SH", 1, 10.0)),
prev_date="2026-07-29", curr_date="2026-07-30")
assert d["kind"] == pd.KIND_CROSS_DAY
@case("[D2] 同一计划日的不同版本要单独成一类")
def t_d2():
d = pd.diff(roster(_r("600000.SH", 1, 10.0)), roster(_r("600000.SH", 2, 10.0)),
prev_date="2026-07-30", curr_date="2026-07-30")
assert d["kind"] == pd.KIND_REVISION
@case("[D3] 同日新版本的结论要点明'候选池与此前不是同一份'")
def t_d3():
d = pd.diff(roster(_r("600000.SH", 1, 10.0)), roster(_r("600418.SH", 1, 11.0)),
prev_date="2026-07-30", curr_date="2026-07-30")
assert "重算" in d["note"] and "候选池" in d["note"]
# ================================================================ [E] 进出榜与尾部闸
@case("[E1] 新进榜与掉榜各归各位")
def t_e1():
prev = roster(_r("600000.SH", 1, 10.0), _r("600418.SH", 2, 9.0))
curr = roster(_r("600000.SH", 1, 10.0), _r("600519.SH", 2, 9.5))
d = pd.diff(prev, curr, prev_date="2026-07-29", curr_date="2026-07-30")
assert [x["ts_code"] for x in d["entered"]] == ["600519.SH"]
assert [x["ts_code"] for x in d["exited"]] == ["600418.SH"]
@case("[E2] 掉榜那条带的是**上一版**的排名与档位")
def t_e2():
prev = roster(_r("600418.SH", 7, 233.3, tier="弱传导"))
d = pd.diff(prev, roster(_r("600000.SH", 1, 10.0)),
prev_date="2026-07-29", curr_date="2026-07-30")
x = d["exited"][0]
assert x["rank"] == 7 and x["tier"] == "弱传导", "掉榜的票在这一版没有数据, 只能示上一版"
@case("[E3] 两版都没吃满 top 时不设尾部闸 —— 消失就是真消失")
def t_e3():
prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
curr = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 10)]) # 第 10 名没了
d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30",
prev_capped=False, curr_capped=False)
assert d["counts"]["exited"] == 1 and d["tail_churn"]["exited"] == 0
# tail_guarded 报的是各档闸位在第几名, None = 该档没设闸 (让"为什么这条没报"可查)
assert d["tail_guarded"] == {"entered": {"main": None, "observe": None},
"exited": {"main": None, "observe": None}}
@case("[E4] 这一版吃满 top 时, 榜尾掉榜归 tail_churn 不列名")
def t_e4():
prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
curr = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 10)]) # 原第 10 名没了
d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30",
curr_capped=True) # 榜长 9, 闸位 4.5, 上一版 rank 10 在闸外
assert d["counts"]["exited"] == 0
assert d["tail_churn"]["exited"] == 1, "它可能只是掉到我们要的条数之外, 不是上游摘了它"
@case("[E5] 吃满 top 也照报榜首掉榜")
def t_e5():
prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
curr = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(2, 11)]) # 榜首没了
d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", curr_capped=True)
assert [x["ts_code"] for x in d["exited"]] == ["600001.SH"]
assert d["tail_churn"]["exited"] == 0
@case("[E6] 新进榜的尾部闸看的是**上一版**吃没吃满")
def t_e6():
prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
curr = dict(prev)
curr["600099.SH"] = _r("600099.SH", 9, 91.5) # 新面孔排在榜尾
d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", prev_capped=True)
assert d["counts"]["entered"] == 0
assert d["tail_churn"]["entered"] == 1, "上一版吃满了, 这张脸当时可能就在榜上只是没露面"
@case("[E7] 上一版吃满时, 榜首的新面孔照报")
def t_e7():
prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
curr = dict(prev)
curr["600099.SH"] = _r("600099.SH", 1, 200.0) # 新面孔直接冲到第一
d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", prev_capped=True)
assert [x["ts_code"] for x in d["entered"]] == ["600099.SH"]
@case("[E8] 两道闸互不干扰: 这一版截断不该影响新进榜的判定")
def t_e8():
prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
curr = dict(prev)
curr["600099.SH"] = _r("600099.SH", 9, 91.5)
d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30",
prev_capped=False, curr_capped=True)
assert d["counts"]["entered"] == 1, "上一版没吃满, 那这张新面孔就是真的新"
assert d["tail_churn"]["entered"] == 0
@case("[E9] 整档消失不被尾部闸吞掉")
def t_e9():
prev = roster(_r("600877.SH", 1, 101.0, tier=None, bucket="observe"),
_r("600000.SH", 1, 10.0))
curr = roster(_r("600000.SH", 1, 10.0)) # 观察档整档没了
d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", curr_capped=True)
assert [x["ts_code"] for x in d["exited"]] == ["600877.SH"], (
"这一版观察档一条都没有, 谈不上被 top 截断 —— 整档消失本身就是要报的大事")
@case("[E10] 尾部闸按档各算各的 (闸位取自各自那一档的长度)")
def t_e10():
prev = roster(*([_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)]
+ [_r("600877.SH", 1, 101.0, tier=None, bucket="observe")]))
curr = roster(*([_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 10)]
+ [_r(f"6009{i:02d}.SH", i, 90 - i, tier=None, bucket="observe")
for i in range(1, 11)]))
d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30",
curr_capped={"main": True, "observe": True})
# 主榜榜尾那只走 churn; 观察档那只 rank=1、这一版观察档有 10 条(闸位 5), 在闸内 → 照报
assert [x["ts_code"] for x in d["exited"]] == ["600877.SH"]
assert d["tail_churn"]["exited"] == 1
@case("[E11] 主榜吃满不代表观察档也吃满 —— 别拿主榜的截断解释观察档的摘牌")
def t_e11():
# 上游两个独立参数: top=10 吃满了, obs_top=100 只回了 4 条 (没吃满)。
# 观察档那只真被摘了 —— 若两档共用主榜那一个标志, 它会被记成"截断噪音"永远看不见。
prev = roster(*([_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)]
+ [_r(f"6009{i:02d}.SH", i, 90 - i, tier=None, bucket="observe")
for i in range(1, 5)]))
curr = roster(*([_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)]
+ [_r(f"6009{i:02d}.SH", i, 90 - i, tier=None, bucket="observe")
for i in range(1, 4)])) # 观察档第 4 名被摘
d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30",
curr_capped={"main": True, "observe": False})
assert [x["ts_code"] for x in d["exited"]] == ["600904.SH"], (
"观察档没吃满就不该设闸 —— 拿主榜的截断解释观察档, 会吞掉一次真实摘牌")
assert d["tail_churn"]["exited"] == 0
# 反过来: 两档都按吃满算, 这条就会被吞掉 (证明上面那条断言不是白写的)
d2 = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", curr_capped=True)
assert d2["counts"]["exited"] == 0 and d2["tail_churn"]["exited"] == 1
@case("[E12] 持仓票不受榜尾闸约束 —— 漏报一条就是一个仓位")
def t_e12():
prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
curr = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 10)])
# 非持仓: 上一版 rank 10 在闸外 → 吞掉
d0 = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", curr_capped=True)
assert d0["counts"]["exited"] == 0
# 同一只票是持仓: 照报, 且记一笔豁免
d1 = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30",
curr_capped=True, held=["600010.SH"])
assert [x["ts_code"] for x in d1["exited"]] == ["600010.SH"]
assert d1["tail_churn"]["exited"] == 0 and d1["tail_churn"]["held_exempt"] == 1
assert d1["held"]["n_watch"] == 1
@case("[E13] rank 判不出来时照报, 不当噪音吞掉")
def t_e13():
prev = roster(_r("600000.SH", None, 10.0), _r("600001.SH", 1, 99.0))
curr = roster(_r("600001.SH", 1, 99.0))
d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", curr_capped=True)
assert [x["ts_code"] for x in d["exited"]] == ["600000.SH"], (
"这个函数的'是'代表吞掉一条变化, 所以拿不准时必须放行")
# ================================================================ [F] 档位与券商覆盖
@case("[F1] 升档与降档")
def t_f1():
prev = roster(_r("600000.SH", 1, 10.0, tier="弱传导"), _r("600418.SH", 2, 9.0, tier="强传导"))
curr = roster(_r("600000.SH", 1, 10.0, tier="强传导"), _r("600418.SH", 2, 9.0, tier="弱传导"))
d = pd.diff(prev, curr, prev_date="2026-07-29", curr_date="2026-07-30")
assert [x["ts_code"] for x in d["tier_up"]] == ["600000.SH"]
assert d["tier_up"][0]["tier_from"] == "弱传导" and d["tier_up"][0]["tier_to"] == "强传导"
assert [x["ts_code"] for x in d["tier_down"]] == ["600418.SH"]
@case("[F2] 主榜 ↔ 观察档 互换 = 券商覆盖翻转, 单独一段")
def t_f2():
prev = roster(_r("600000.SH", 1, 10.0))
curr = roster(_r("600000.SH", 1, 10.0, tier=None, bucket="observe"))
d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30")
assert len(d["bucket_moved"]) == 1
m = d["bucket_moved"][0]
assert m["moved_from"] == "main" and m["moved_to"] == "observe"
assert d["counts"]["exited"] == 0, "换档不是掉榜 —— 票还在, 只是没券商覆盖了"
@case("[F3] 换了档就不算名次跳变 —— 两档的 rank 不可比")
def t_f3():
prev = roster(_r("600000.SH", 300, 10.0))
curr = roster(_r("600000.SH", 1, 10.0, tier=None, bucket="observe"))
d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30", rank_jump=10)
assert d["counts"]["rank_jump"] == 0
assert d["counts"]["bucket_moved"] == 1
# ================================================================ [G] 名次跳变
@case("[G1] 跳变到阈值才报")
def t_g1():
prev = roster(_r("600000.SH", 100, 10.0), _r("600418.SH", 20, 20.0))
curr = roster(_r("600000.SH", 20, 30.0), _r("600418.SH", 15, 22.0))
d = pd.diff(prev, curr, prev_date="2026-07-29", curr_date="2026-07-30", rank_jump=50)
assert [x["ts_code"] for x in d["rank_jump"]] == ["600000.SH"]
assert d["rank_jump"][0]["rank_from"] == 100 and d["rank_jump"][0]["rank_delta"] == -80
@case("[G2] 按跳变幅度降序")
def t_g2():
prev = roster(_r("600000.SH", 200, 1.0), _r("600418.SH", 60, 2.0))
curr = roster(_r("600000.SH", 100, 5.0), _r("600418.SH", 1, 9.0))
d = pd.diff(prev, curr, prev_date="2026-07-29", curr_date="2026-07-30", rank_jump=50)
assert [x["ts_code"] for x in d["rank_jump"]] == ["600000.SH", "600418.SH"]
@case("[G3] rank 缺失时不算跳变")
def t_g3():
prev = roster(_r("600000.SH", None, 1.0))
curr = roster(_r("600000.SH", 1, 9.0))
d = pd.diff(prev, curr, prev_date="2026-07-29", curr_date="2026-07-30", rank_jump=1)
assert d["counts"]["rank_jump"] == 0
# ================================================================ [H] 持仓票视图
@case("[H1] 持仓票掉榜进 n_watch")
def t_h1():
prev = roster(_r("600000.SH", 1, 10.0), _r("600418.SH", 2, 9.0))
curr = roster(_r("600418.SH", 1, 9.0))
d = pd.diff(prev, curr, prev_date="2026-07-29", curr_date="2026-07-30",
held=["600000.SH"])
assert [x["ts_code"] for x in d["held"]["exited"]] == ["600000.SH"]
assert d["held"]["n_watch"] == 1
assert d["exited"][0]["held"] is True
@case("[H2] 持仓票降档也进 n_watch")
def t_h2():
d = pd.diff(roster(_r("600000.SH", 1, 10.0, tier="强传导")),
roster(_r("600000.SH", 1, 10.0, tier="弱传导")),
prev_date="2026-07-29", curr_date="2026-07-30", held=["600000.SH"])
assert d["held"]["n_watch"] == 1 and len(d["held"]["tier_down"]) == 1
@case("[H2b] 持仓票掉进观察档 = 丢了券商覆盖, 也要进 n_watch")
def t_h2b():
# 观察档的行没有 tier, 所以这既不算 exited 也不算 tier_down —— 不显式算进来的话,
# 一只持仓票丢了估值锚, 页面红条不亮、探活脚本还会打一句"没被摘也没降档"。
d = pd.diff(roster(_r("600000.SH", 3, 240.0, tier="强传导")),
roster(_r("600000.SH", 5, 99.0, tier=None, bucket="observe")),
prev_date="2026-07-29", curr_date="2026-07-30", held=["600000.SH"])
assert d["counts"]["exited"] == 0 and d["counts"]["tier_down"] == 0
assert [x["ts_code"] for x in d["held"]["coverage_lost"]] == ["600000.SH"]
assert d["held"]["n_watch"] == 1
assert "券商覆盖" in d["note"]
@case("[H2c] 掉进主榜(拿到券商覆盖)不算要盯的事")
def t_h2c():
d = pd.diff(roster(_r("600000.SH", 5, 99.0, tier=None, bucket="observe")),
roster(_r("600000.SH", 3, 240.0, tier="强传导")),
prev_date="2026-07-29", curr_date="2026-07-30", held=["600000.SH"])
assert d["counts"]["bucket_moved"] == 1
assert d["held"]["coverage_lost"] == [] and d["held"]["n_watch"] == 0
@case("[H3] 持仓票升档只进 n_total, 不进 n_watch")
def t_h3():
d = pd.diff(roster(_r("600000.SH", 1, 10.0, tier="弱传导")),
roster(_r("600000.SH", 1, 10.0, tier="强传导")),
prev_date="2026-07-29", curr_date="2026-07-30", held=["600000.SH"])
assert d["held"]["n_watch"] == 0 and d["held"]["n_total"] == 1
@case("[H4] 非持仓票不进持仓视图")
def t_h4():
d = pd.diff(roster(_r("600000.SH", 1, 10.0)), roster(_r("600418.SH", 1, 10.0)),
prev_date="2026-07-29", curr_date="2026-07-30", held=["600519.SH"])
assert d["held"]["n_total"] == 0 and d["counts"]["exited"] == 1
@case("[H5] 持仓票掉榜要在结论句里显眼")
def t_h5():
d = pd.diff(roster(_r("600000.SH", 1, 10.0)), roster(_r("600418.SH", 1, 10.0)),
prev_date="2026-07-29", curr_date="2026-07-30", held=["600000.SH"])
assert "持仓票" in d["note"] and "掉榜" in d["note"]
# ================================================================ [I] 健壮性
@case("[I1] 空对空不炸")
def t_i1():
d = pd.diff({}, {}, prev_date="2026-07-30", curr_date="2026-07-30")
assert d["kind"] == pd.KIND_FIRST and sum(d["counts"].values()) == 0
@case("[I2] 名册全没了也报得出来")
def t_i2():
d = pd.diff(roster(_r("600000.SH", 1, 10.0)), {},
prev_date="2026-07-29", curr_date="2026-07-30")
assert d["counts"]["exited"] == 1 and d["roster_size"]["curr"] == 0
@case("[I3] tail_guard 越界值被夹回 [0,1]")
def t_i3():
prev = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11)])
curr = roster(*[_r(f"6000{i:02d}.SH", i, 100 - i) for i in range(1, 11) if i != 5])
for given, clamped, reported in ((5.0, 1.0, 1), (-1.0, 0.0, 0)):
d = pd.diff(prev, curr, prev_date="2026-07-30", curr_date="2026-07-30",
curr_capped=True, tail_guard=given)
assert d["params"]["tail_guard"] == clamped
# 闸开到底(1.0): 榜内的第 5 名掉了要报; 闸关到底(0.0): 一律当截断噪音
assert d["counts"]["exited"] == reported
@case("[I4] counts 与各段长度始终一致")
def t_i4():
prev = roster(_r("600000.SH", 1, 10.0, tier="强传导"), _r("600418.SH", 200, 2.0))
curr = roster(_r("600000.SH", 1, 10.0, tier="弱传导"), _r("600519.SH", 3, 8.0))
d = pd.diff(prev, curr, prev_date="2026-07-29", curr_date="2026-07-30")
for k, n in d["counts"].items():
assert len(d[k]) == n, f"{k} 的 counts 与实际条数对不上"
# ================================================================ [J] 服务层落库往返
# 纯逻辑之外唯一容易藏 bug 的地方: 名册 → JSON 落库 → 读回来 → 比对 这条往返。
# 单测不连库 —— 拿内存桩顶掉 pms_repo 与 param_store (plan_feed 是函数内 import, 装在
# sys.modules 上就生效; 与 test_batch7 用假 requests 是同一套路)。
def _with_fake_repo(params=None):
import json as _j
import types
store = {"rows": [], "seq": 0}
repo = types.ModuleType("app.repo.pms_repo")
def insert_plan_snapshot(*, plan_date, digest, roster, meta=None, n_main=0, n_observe=0,
capped_main=False, capped_obs=False, fetched_at=None):
# 与生产同一条判据: 只跟**最新那行**比, 不是"曾经出现过的任何一行"。
# 早先桩里写的是后者 (模拟唯一键), 于是 A→B→A 这个 case 在桩上永远复现不出来 ——
# 桩比生产更严, 单测就成了给错实现背书。
last = max(store["rows"], key=lambda x: x["id"]) if store["rows"] else None
if last and last["plan_date"] == plan_date and last["digest"] == digest:
return 0
store["seq"] += 1
store["rows"].append({
"id": store["seq"], "plan_date": plan_date, "digest": digest,
"fetched_ymd": 20260731, "fetched_at": f"t{store['seq']}",
"n_main": n_main, "n_observe": n_observe,
"capped_main": bool(capped_main), "capped_obs": bool(capped_obs),
# 真表是 MEDIUMTEXT, 落进去的是 JSON 文本、读回来才转对象。桩必须照做 ——
# 否则"名册里塞了不可序列化的东西"这类事故单测里根本看不见。
"roster_json": _j.dumps(roster, ensure_ascii=False), "meta": meta or {}})
return 1
def latest_plan_snapshots(limit=2):
out = []
for r in sorted(store["rows"], key=lambda x: -x["id"])[:limit]:
r2 = dict(r)
r2["roster"] = _j.loads(r2.pop("roster_json"))
out.append(r2)
return out
def list_plan_snapshots(*, plan_date=None, limit=50):
rows = [dict(r) for r in sorted(store["rows"], key=lambda x: -x["id"])]
if plan_date:
rows = [r for r in rows if r["plan_date"] == plan_date]
for r in rows:
r.pop("roster_json", None)
return rows[:limit]
repo.insert_plan_snapshot = insert_plan_snapshot
repo.latest_plan_snapshots = latest_plan_snapshots
repo.list_plan_snapshots = list_plan_snapshots
repo.prune_plan_snapshots = lambda keep=200: 0
repo.list_positions = lambda only_open=False: []
ps = types.ModuleType("app.services.param_store")
vals = dict(params or {})
ps.get = lambda k, d=None: vals.get(k, d)
ps.get_int = lambda k, d=0: int(vals.get(k, d))
ps.get_float = lambda k, d=0.0: float(vals.get(k, d))
ps.get_bool = lambda k, d=False: bool(vals.get(k, d))
ps.get_list = lambda k, d=(): list(vals.get(k, d))
saved = {n: sys.modules.get(n) for n in ("app.repo.pms_repo", "app.services.param_store")}
sys.modules["app.repo.pms_repo"] = repo
sys.modules["app.services.param_store"] = ps
def restore():
for n, m in saved.items():
if m is None:
sys.modules.pop(n, None)
else:
sys.modules[n] = m
return store, restore
@case("[J1] 落库往返: 名册经 JSON 存取后指纹不变")
def t_j1():
import json as _j
from app.services import plan_feed as pf
store, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
try:
p = plan(main=[_r("600418.SH", 1, 242.24, theme="整车"),
_r("600000.SH", 2, 241.55, tier="弱传导", theme=None)],
observe=[_r("600877.SH", 1, 101.1, tier=None, bucket="observe")],
date="2026-07-30")
out = pf.snapshot(p)
assert out["stored"] is True and out["rows"] == 3
back = pd.roster_from_rows(_j.loads(store["rows"][0]["roster_json"]))
assert pd.digest_of(back) == out["digest"], "存取一轮指纹就变了, 比对全废"
assert back["600000.SH"]["h"] is None, "theme 为空的票也要原样存回来"
finally:
restore()
@case("[J2] 同一份榜重复落只留一行")
def t_j2():
from app.services import plan_feed as pf
store, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
try:
p = plan(main=[_r("600418.SH", 1, 242.24)], date="2026-07-30")
assert pf.snapshot(p)["stored"] is True
assert pf.snapshot(p)["stored"] is False, "内容没变还新增一行, 表会白涨"
assert len(store["rows"]) == 1
finally:
restore()
@case("[J3] score 变了但次序没变, 不算新版本")
def t_j3():
from app.services import plan_feed as pf
store, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
try:
pf.snapshot(plan(main=[_r("600418.SH", 1, 242.2801)], date="2026-07-30"))
pf.snapshot(plan(main=[_r("600418.SH", 1, 242.2799)], date="2026-07-30"))
assert len(store["rows"]) == 1
finally:
restore()
@case("[J4] 只读预览不写库, 且认得出'与库里那份相同'")
def t_j4():
from app.services import plan_feed as pf
store, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
try:
p = plan(main=[_r("600418.SH", 1, 242.24)], date="2026-07-30")
pf.snapshot(p)
n = len(store["rows"])
out = pf.preview_changes(p)
assert out["ok"] and out["same_as_stored"] is True
assert len(store["rows"]) == n, "preview 必须只读 —— 探活脚本的只读契约靠它"
finally:
restore()
@case("[J5] 只读预览能比出手上这份与库里那份的差异")
def t_j5():
from app.services import plan_feed as pf
store, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
try:
pf.snapshot(plan(main=[_r("600418.SH", 1, 242.24)], date="2026-07-30"))
out = pf.preview_changes(plan(main=[_r("600519.SH", 1, 250.0)], date="2026-07-30"))
d = out["diff"]
assert d["kind"] == pd.KIND_REVISION
assert [x["ts_code"] for x in d["entered"]] == ["600519.SH"]
assert [x["ts_code"] for x in d["exited"]] == ["600418.SH"]
assert len(store["rows"]) == 1
finally:
restore()
@case("[J6] 快照关掉时 changes 明说不可用, 而不是装作没变化")
def t_j6():
from app.services import plan_feed as pf
_, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": False})
try:
out = pf.changes()
assert out["ok"] is False and out["diff"] is None
assert "PMS_PLAN_SNAPSHOT" in out["hint"], "得说清是关掉了, 不是没有变化"
finally:
restore()
@case("[J7] 快照落库失败不阻断候选池, 但要把错带出来")
def t_j7():
from app.services import plan_feed as pf
_, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
try:
def _boom(**kw):
raise RuntimeError("DB 挂了")
sys.modules["app.repo.pms_repo"].insert_plan_snapshot = _boom
out = pf._snapshot_quiet(plan(main=[_r("600418.SH", 1, 1.0)], date="2026-07-30"))
assert out["stored"] is False and "DB 挂了" in out["error"]
finally:
restore()
@case("[J7b] 上游 A→B→A 改回去时, 比对方向不能反")
def t_j7b():
# 曾经用 (plan_date, digest) 唯一键去重, 于是第三次的 A 落不下去、库里最新仍是 B,
# 页面把 B→A 这次真实回退显示成 A→B —— 方向整个反了, 还顺带诬告一句"落库失败"。
from app.services import plan_feed as pf
store, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
try:
a = plan(main=[_r("600418.SH", 1, 242.24)], date="2026-07-30")
b = plan(main=[_r("600519.SH", 1, 250.00)], date="2026-07-30")
pf.snapshot(a)
pf.snapshot(b)
assert pf.snapshot(a)["stored"] is True, "改回去也是一次真实变化, 必须落得下去"
assert len(store["rows"]) == 3
out = pf.preview_changes(a)
assert out["same_as_stored"] is True
d = pf.changes()["diff"]
assert [x["ts_code"] for x in d["entered"]] == ["600418.SH"]
assert [x["ts_code"] for x in d["exited"]] == ["600519.SH"]
finally:
restore()
@case("[J7c] 上一版名册坏了要报故障, 不能显示成'首次'")
def t_j7c():
import json as _j
from app.services import plan_feed as pf
store, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
try:
pf.snapshot(plan(main=[_r("600418.SH", 1, 1.0)], date="2026-07-30"))
store["rows"][0]["roster_json"] = _j.dumps([]) # 名册没了, n_main 还写着 1
out = pf.preview_changes(plan(main=[_r("600519.SH", 1, 2.0)], date="2026-07-31"))
assert out["diff"]["prev_broken"] is True
assert "读不出来" in out["hint"]
finally:
restore()
@case("[J7d] 持仓代码是前缀式也要认得出来")
def t_j7d():
from app.services import plan_feed as pf
_, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
try:
sys.modules["app.repo.pms_repo"].list_positions = \
lambda only_open=False: [{"ts_code": "SH600418"}] # 前缀式
pf.snapshot(plan(main=[_r("600418.SH", 1, 1.0)], date="2026-07-30"))
out = pf.preview_changes(plan(main=[_r("600519.SH", 1, 2.0)], date="2026-07-31"))
assert out["diff"]["held"]["n_watch"] == 1, "不归一的后果是持仓视图静默为空"
finally:
restore()
@case("[J8] 快照台账数得出同一计划日有几版")
def t_j8():
from app.services import plan_feed as pf
_, restore = _with_fake_repo({"PMS_PLAN_SNAPSHOT": True})
try:
pf.snapshot(plan(main=[_r("600418.SH", 1, 1.0)], date="2026-07-30"))
pf.snapshot(plan(main=[_r("600418.SH", 2, 1.0)], date="2026-07-30"))
pf.snapshot(plan(main=[_r("600418.SH", 1, 1.0)], date="2026-07-29"))
log = pf.snapshot_log()
assert log["ok"] and log["versions_per_date"]["2026-07-30"] == 2
assert log["revised"] == {"2026-07-30": 2}, "同一 date 多版是 §7.3 那个现象, 要数出来"
finally:
restore()
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()