# -*- coding: utf-8 -*- """ 第九批模块单测 (零外部依赖, 不连库不触网) ========================================== 运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch9_units.py 覆盖: 接管既有持仓前的成本价体检 (app/core/rebuild_check.py)。 这一批守的是一个**只发生一次、且不可逆**的决定: 账本清空后按「以下游为准」认领真实持仓, 那一刻定死每只票的开仓价 → 摊薄成本 → 安全垫 → 补仓/加仓/保垫减仓的共同判据。 2026-07-29 首次接管 22 只持仓时踩过一次 (全按现价开仓, 安全垫齐刷刷是 0)。 所以用例的重点不是"算得对不对", 而是**该拦的拦不拦得住**, 以及**不该拦的会不会误伤** —— 误伤的代价是多等一天, 漏拦的代价是一本每个数都错、却和真账长得一模一样的账。 """ import os import sys import traceback sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.core import rebuild_check as rb # noqa: E402 RESULTS = [] def case(name): def deco(fn): RESULTS.append((name, fn)) return fn return deco def _p(code, qty=1000, cost=10.0, avail=None, price=None): """一行下游持仓 (downstream_repo.fetch_positions 的行形态)。""" return {"ts_code": code, "qty": qty, "cost": cost, "avail_qty": qty if avail is None else avail, "price": price} # ================================================================ [A] 单只票的判定 @case("[A1] 成本价正常 → OK, 并算出安全垫") def t_a1(): r = rb.check_row(_p("600000.SH", cost=20.0), price=10.0) assert r["verdict"] == rb.OK assert abs(r["cushion_pct"] - (-0.5)) < 1e-9, "真实成本 20、现价 10 就是实亏 50%" @case("[A2] 成本价缺失 / 为 0 / 为负 → MISSING") def t_a2(): for bad in (None, 0, 0.0, -1.0, ""): r = rb.check_row(_p("600000.SH", cost=bad), price=10.0) assert r["verdict"] == rb.MISSING, f"cost={bad!r} 应判 MISSING" assert "安全垫恒为 0" in rb.check_row(_p("600000.SH", cost=0), price=10.0)["why"] @case("[A3] 成本 ≈ 现价 → EQ_PRICE (安全垫≈0)") def t_a3(): r = rb.check_row(_p("600000.SH", cost=10.02), price=10.0) # 差 0.2% < 0.5% assert r["verdict"] == rb.EQ_PRICE r2 = rb.check_row(_p("600000.SH", cost=10.30), price=10.0) # 差 3% > 0.5% assert r2["verdict"] == rb.OK, "正常的小幅浮盈不该被当成'拿现价充数'" @case("[A4] 成本与现价差两个数量级 → ABSURD (多半是单位错)") def t_a4(): assert rb.check_row(_p("600000.SH", cost=1000.0), price=10.0)["verdict"] == rb.ABSURD assert rb.check_row(_p("600000.SH", cost=0.05), price=10.0)["verdict"] == rb.ABSURD # 分/元 混用是最典型的一种: 成本记成 1002 分而现价是 10.02 元 r = rb.check_row(_p("600000.SH", cost=1002.0), price=10.02) assert r["verdict"] == rb.ABSURD and "单位" in r["why"] @case("[A5] 可用量 > 总量 或为负 → AVAIL_BAD") def t_a5(): assert rb.check_row(_p("600000.SH", qty=1000, avail=1500), price=10.0)["verdict"] \ == rb.AVAIL_BAD assert rb.check_row(_p("600000.SH", qty=1000, avail=-1), price=10.0)["verdict"] \ == rb.AVAIL_BAD # 当日买入是合法的: 可用 < 总量 (成本给个与现价不同的值, 免得撞上 EQ_PRICE) assert rb.check_row(_p("600000.SH", qty=1000, avail=0, cost=20.0), price=10.0)["verdict"] == rb.OK @case("[A6] 取不到现价 → NO_PRICE, 但成本本身仍算可用") def t_a6(): r = rb.check_row(_p("600000.SH", cost=20.0), price=None) assert r["verdict"] == rb.NO_PRICE and r["cushion_pct"] is None assert "成本值本身可用" in r["why"] @case("[A7] 可用量的判定排在成本之前 —— 两个都坏时先报可用量") def t_a7(): r = rb.check_row(_p("600000.SH", qty=100, avail=999, cost=0), price=10.0) assert r["verdict"] == rb.AVAIL_BAD # ================================================================ [B] 整批的阻断判据 @case("[B1] 全部正常 → 不阻断") def t_b1(): rows = [_p("600000.SH", cost=20.0), _p("600519.SH", cost=1500.0)] out = rb.check_costs(rows, {"600000.SH": 10.0, "600519.SH": 1800.0}) assert out["blocking"] is False and out["reasons"] == [] assert "可以建账" in out["hint"] @case("[B2] 少数几只没成本价不阻断, 但要单独列出来") def t_b2(): # 既有设计对"某一只没成本价"有逐只回退现价的路径 (build_recon_fixes, 带 price_source # 留痕)。闸不该推翻它 —— 闸管的是"一本从头就错的账", 不是替既有设计做二次判断。 rows = [_p("600000.SH", cost=20.0), _p("600519.SH", cost=0)] out = rb.check_costs(rows, {"600000.SH": 10.0, "600519.SH": 1800.0}) assert out["blocking"] is False assert out["estimated"] == ["600519.SH"] assert "改不回来" in out["hint"], "估出来的成本后续对账不会再碰, 这句必须说出来" @case("[B2b] 全部没成本价 → 阻断 (对端那一列整个没填)") def t_b2b(): rows = [_p("600000.SH", cost=0), _p("600519.SH", cost=None)] out = rb.check_costs(rows, {"600000.SH": 10.0, "600519.SH": 1800.0}) assert out["blocking"] is True and out["estimated"] == [] assert "全部" in out["reasons"][0] @case("[B2c] 数据是**错的**而不是缺的 → 一只就阻断") def t_b2c(): # ABSURD/AVAIL_BAD 说明这批数据的生产方式有问题, 不该只怀疑那一只 for bad in ({"cost": 1000.0}, {"avail": 9999}): rows = [_p("600000.SH", cost=20.0), _p("600519.SH", **bad)] out = rb.check_costs(rows, {"600000.SH": 10.0, "600519.SH": 10.0}) assert out["blocking"] is True, bad assert "错的" in out["reasons"][0] @case("[B3] 整组成本≈现价 → 阻断 (这就是 07-29 那个坑的模样)") def t_b3(): rows = [_p(f"60000{i}.SH", cost=10.0) for i in range(1, 6)] out = rb.check_costs(rows, {f"60000{i}.SH": 10.0 for i in range(1, 6)}) assert out["blocking"] is True and out["eq_ratio"] == 1.0 assert "安全垫会是 0" in out["reasons"][0] @case("[B4] 少数几只成本≈现价 → 不阻断 (当日买入是正常的)") def t_b4(): rows = [_p("600000.SH", cost=20.0), _p("600519.SH", cost=1800.0), _p("600036.SH", cost=30.0), _p("601318.SH", cost=10.0)] px = {"600000.SH": 10.0, "600519.SH": 1500.0, "600036.SH": 40.0, "601318.SH": 10.0} out = rb.check_costs(rows, px) assert out["counts"].get(rb.EQ_PRICE) == 1 assert out["blocking"] is False, "四只里一只当日买入很正常, 不该拦" @case("[B5] 只有一只票时不按占比阻断 —— 样本太小, 它可能真是当日买的") def t_b5(): out = rb.check_costs([_p("600000.SH", cost=10.0)], {"600000.SH": 10.0}) assert out["counts"].get(rb.EQ_PRICE) == 1 assert out["blocking"] is False @case("[B6] 取不到现价的票不稀释 EQ_PRICE 占比") def t_b6(): # 两只成本≈现价 + 八只取不到现价。若拿 10 做分母, 占比 20% 不阻断 —— 那是错的: # 能判的两只全是 EQ_PRICE, 该阻断。 rows = [_p("600001.SH", cost=10.0), _p("600002.SH", cost=10.0)] + \ [_p(f"6001{i:02d}.SH", cost=5.0) for i in range(1, 9)] px = {"600001.SH": 10.0, "600002.SH": 10.0} # 其余取不到价 out = rb.check_costs(rows, px) assert out["counts"].get(rb.NO_PRICE) == 8 assert out["eq_ratio"] == 1.0 and out["blocking"] is True @case("[B7] 零持仓行不参与体检") def t_b7(): rows = [_p("600000.SH", qty=0, cost=0), _p("600519.SH", cost=1800.0)] out = rb.check_costs(rows, {"600519.SH": 1500.0}) assert out["n"] == 1 and out["blocking"] is False, "qty=0 的行是历史残留, 不该拖累建账" @case("[B8] 下游一只持仓都没有 → 不阻断但说清是没账可建") def t_b8(): out = rb.check_costs([], {}) assert out["n"] == 0 and out["blocking"] is False assert "无账可建" in out["hint"] @case("[B9] 阻断时的 hint 要给出可执行的下一步") def t_b9(): out = rb.check_costs([_p("600000.SH", cost=0)], {"600000.SH": 10.0}) # 唯一一只且缺失 assert "cost_price" in out["hint"], "得说清要对端改哪一列, 不是只说'数据有问题'" assert "长得一模一样" in out["hint"], "得说清为什么不能将就着建" # ================================================================ [C] 情形覆盖 (不阻断) @case("[C1] 四种情形齐全 → enough") def t_c1(): rows = [_p("600000.SH", cost=10.0), # 浮盈 +50% _p("600519.SH", cost=20.0), # 浮亏 −25% _p("600036.SH", cost=10.0, qty=1000, avail=0), # 当日买入 _p("601318.SH", cost=10.0)] px = {"600000.SH": 15.0, "600519.SH": 15.0, "600036.SH": 11.0, "601318.SH": 11.0} cov = rb.coverage(rb.check_costs(rows, px)["rows"]) assert cov["enough"] is True and cov["missing"] == [] @case("[C2] 全是不赚不亏 → 报出这轮验不到哪些纪律") def t_c2(): rows = [_p(f"60000{i}.SH", cost=10.0) for i in range(1, 4)] cov = rb.coverage(rb.check_costs(rows, {f"60000{i}.SH": 10.05 for i in range(1, 4)})["rows"]) assert cov["enough"] is False assert any("浮盈" in m for m in cov["missing"]) assert any("浮亏" in m for m in cov["missing"]) @case("[C3] 只有一只持仓 → 报持仓不足 3 只") def t_c3(): cov = rb.coverage(rb.check_costs([_p("600000.SH", cost=20.0)], {"600000.SH": 10.0})["rows"]) assert cov["enough"] is False and any("不足 3 只" in m for m in cov["missing"]) @case("[C4] 覆盖不全绝不阻断建账 —— 数据是真的就该建") def t_c4(): rows = [_p("600000.SH", cost=20.0), _p("600519.SH", cost=3000.0)] out = rb.check_costs(rows, {"600000.SH": 10.0, "600519.SH": 1500.0}) cov = rb.coverage(out["rows"]) assert out["blocking"] is False and cov["enough"] is False # ================================================================ [D] 健壮性 @case("[D1] 脏数据不炸") def t_d1(): for bad in ({"ts_code": "600000.SH"}, {"ts_code": None, "qty": "x", "cost": "y"}, {"ts_code": "600000.SH", "qty": "1000", "cost": "20.0", "avail_qty": "1000"}): rb.check_row(bad, price=10.0) out = rb.check_costs([{"ts_code": "600000.SH", "qty": "1000", "cost": "20.0"}], {"600000.SH": "10.0"}) assert out["n"] == 1, "字符串数字要能吃进去 —— 库里 DECIMAL 列取出来常是字符串" @case("[D2] rows 为 None 不炸") def t_d2(): assert rb.check_costs(None, None)["n"] == 0 assert rb.coverage(None)["enough"] is False @case("[D3] 判定常量互不相同 —— 别把两种故障混成一个码") def t_d3(): vs = [rb.OK, rb.MISSING, rb.EQ_PRICE, rb.ABSURD, rb.AVAIL_BAD, rb.NO_PRICE] assert len(set(vs)) == len(vs) # ================================================================ [E] 连续不一致按日推进 # 2026-07-31 实机暴露: 日报关注区写出「连续 175 日不一致」, 而项目 07-14 才开工。 # 原因是每调一次 reconcile() 就 +1, 而盘中轻对账每分钟调一次 —— 设计里「连续 3 日 → ERROR # 待人工」实际成了「连续 3 分钟」。假警报天天响, 真告警就被埋掉。 from app.core import recon as rc # noqa: E402 @case("[E1] 同一天内反复对账不重复计数") def t_e1(): s = rc.advance_streak(0, 0, 20260731, True) assert s["streak"] == 1 and s["changed"] is True for _ in range(5): # 手工点五次「对账」 s = rc.advance_streak(s["streak"], s["ymd"], 20260731, True) assert s["streak"] == 1, "同一天点几次都只算一天 —— 否则三分钟就升 ERROR" assert s["changed"] is False @case("[E2] 跨交易日才 +1") def t_e2(): s = rc.advance_streak(1, 20260731, 20260801, True) assert s["streak"] == 2 and s["ymd"] == 20260801 s = rc.advance_streak(s["streak"], s["ymd"], 20260803, True) # 跨周末 assert s["streak"] == 3 @case("[E3] 差异消失立刻归零, 不必等下一天") def t_e3(): s = rc.advance_streak(7, 20260731, 20260731, False) assert s["streak"] == 0 and s["changed"] is True and s["ymd"] == 20260731 @case("[E4] 本来就是 0 且没差异 → 什么都没变, 不必写库") def t_e4(): assert rc.advance_streak(0, 20260731, 20260731, False)["changed"] is False @case("[E5] 三日门槛与 severity 对得上") def t_e5(): assert rc.recon_severity(0) == rc.SEV_OK assert rc.recon_severity(1) == rc.SEV_WARN and rc.recon_severity(2) == rc.SEV_WARN assert rc.recon_severity(3) == rc.SEV_ERROR # 走满三个交易日才该到 ERROR —— 这条串起来验, 免得两边各改一半 s = {"streak": 0, "ymd": 0} for i, d in enumerate((20260731, 20260801, 20260803), start=1): s = rc.advance_streak(s["streak"], s["ymd"], d, True) assert rc.recon_severity(s["streak"]) == (rc.SEV_ERROR if i >= 3 else rc.SEV_WARN) @case("[E6] 缺日期时退化成每次都推进, 但不会把已有计数弄丢") def t_e6(): s = rc.advance_streak(2, 0, 0, True) # 两个 ymd 都拿不到 assert s["streak"] == 3, "判不了是不是同一天就按保守走(照常推进), 别把计数清零" # ================================================================ [F] 行业占比闸的双判据 # 2026-07-31 实机暴露: 空账本 + 10 只强传导候选 + 一条 60% 升仓命令 → 一条方案都出不来。 # 因为只看"占组合"的话, 第一只票按定义就是组合的 100%, 必然超任何小于 100% 的上限; # 而它被拒后组合市值不推进, 后面每一只面对的还是 100% —— 哪怕分属十个不同行业。 # 行业源是当天才通的 (此前 sector 恒为 None、整段跳过), 所以这个洞一直藏着。 from app.core.sizer import check_caps # noqa: E402 SCALE = 2_000_000 def _ctx(**kw): d = dict(scale=SCALE, portfolio_cap=0.60, stock_cap=0.08, max_names=15, portfolio_mv=0.0, names_count=0, stock_mv=0.0, is_new_name=True, sector="储能", sector_names=0, sector_mv=0.0, sector_max_names=4, sector_max_ratio=0.40) d.update(kw) return d @case("[F1] 空账本的第一笔买入不该被行业占比闸拦下") def t_f1(): bad = check_caps(ts_code="600000.SH", add_amount=0.06 * SCALE, ctx=_ctx()) assert bad == [], f"第一只票必然是组合的 100%, 拦它等于建不了仓: {bad}" @case("[F2] 候选分属不同行业时, 能一路建到总仓上限") def t_f2(): c, n = _ctx(), 0 for i in range(1, 15): c["sector"], c["sector_mv"], c["sector_names"] = f"行业{i}", 0.0, 0 if check_caps(ts_code=f"股{i}", add_amount=0.06 * SCALE, ctx=c): break c["portfolio_mv"] += 0.06 * SCALE c["names_count"] += 1 n += 1 assert n == 10 and abs(c["portfolio_mv"] / SCALE - 0.60) < 1e-9, ( f"6% 一只、总仓上限 60% → 应正好进 10 只, 实际 {n} 只") @case("[F3] 组合建到接近上限时, 行业占比闸照常拦") def t_f3(): # 组合 52%(=104万), 储能已占 38%(=76万), 再买 6% → 组合 58% (未撞总仓闸), # 储能占组合 75.9% > 40%, 占规模 44% > 绝对线 24% —— 两条都成立, 该拦 c = _ctx(portfolio_mv=0.52 * SCALE, sector_mv=0.38 * SCALE, names_count=9, sector_names=3) bad = check_caps(ts_code="X", add_amount=0.06 * SCALE, ctx=c) assert any("SECTOR_RATIO" in b for b in bad), bad assert not any("PORTFOLIO_CAP" in b for b in bad), "这条用例要单独验行业判据" @case("[F4] 占比超但绝对敞口小 → 不拦 (集中度是风险的放大器, 不是风险本身)") def t_f4(): # 组合只有 12%(=24万) 且全在储能: 占组合 100% 超上限, 但只占规模 18% < 绝对线 24% c = _ctx(portfolio_mv=0.12 * SCALE, sector_mv=0.12 * SCALE, names_count=2, sector_names=2) assert check_caps(ts_code="X", add_amount=0.06 * SCALE, ctx=c) == [] @case("[F5] 绝对线不会单独触发 —— 它只用来豁免建仓初期, 不会额外拦人") def t_f5(): """绝对线取 `sector_max_ratio × portfolio_cap` 是有讲究的: 组合在总仓上限以内时, "占规模超绝对线" 必然蕴含 "占组合超上限"。所以这条判据**只会放宽、不会收紧** —— 它把建仓初期那段不可满足的区间豁免掉, 而不改变组合建起来之后的口径。 取值再大一点 (比如直接用 sector_max_ratio) 就会变成一道独立的、更严的闸。""" ratio, cap = 0.40, 0.60 for port_pct in (0.06, 0.12, 0.24, 0.36, 0.48, 0.60): for sec_pct in (0.02, 0.06, 0.12, 0.20, 0.28, 0.36): if sec_pct > port_pct: continue over_scale = sec_pct > ratio * cap + 1e-9 over_port = (sec_pct / port_pct) > ratio + 1e-9 assert not (over_scale and not over_port), ( f"组合 {port_pct:.0%} 行业 {sec_pct:.0%}: 绝对线单独触发了") @case("[F6] 同一行业的只数上限仍然管用 (占比放宽不等于行业闸失效)") def t_f6(): c, n = _ctx(), 0 for i in range(1, 9): if check_caps(ts_code=f"股{i}", add_amount=0.06 * SCALE, ctx=c): break c["portfolio_mv"] += 0.06 * SCALE c["sector_mv"] += 0.06 * SCALE c["names_count"] += 1 c["sector_names"] += 1 n += 1 assert n == 4, f"同一行业最多 4 只 (SECTOR_NAMES), 实际 {n}" @case("[F7] 行业源没配时整段跳过 (约束停用而不是误拦)") def t_f7(): assert check_caps(ts_code="X", add_amount=0.06 * SCALE, ctx=_ctx(sector=None)) == [] # ================================================================ [G] 拒绝原因聚合 from app.core.planner import summarize_rejects # noqa: E402 @case("[G1] 把 rejects 聚合成一行人话, 按条数降序") def t_g1(): rej = [{"ts_code": f"股{i}", "reasons": ["SECTOR_RATIO: 行业[储能]占组合将达 100.0% > 上限 40%"]} for i in range(5)] + \ [{"ts_code": "股X", "reasons": ["STOCK_CAP: 股X 加后 9.0% > 单股上限 8%"]}] s = summarize_rejects(rej) assert s.startswith("SECTOR_RATIO 5 只"), s assert "STOCK_CAP 1 只" in s @case("[G2] 一只票撞多条判据时每条都计数") def t_g2(): s = summarize_rejects([{"ts_code": "股A", "reasons": ["STOCK_CAP: …", "SECTOR_RATIO: …"]}]) assert "STOCK_CAP 1 只" in s and "SECTOR_RATIO 1 只" in s @case("[G3] 没有拒绝时返回空串 (调用方据此区分'拒了'和'本来就没候选')") def t_g3(): assert summarize_rejects([]) == "" and summarize_rejects(None) == "" @case("[G4] 没有判据码的原因 (一手不可行那类) 也归得了类") def t_g4(): s = summarize_rejects([{"ts_code": "股A", "reasons": ["目标金额 3000 元按现价 45.00 买不足一手"]}]) assert "1 只" in s and "股A" in s 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()