560 lines
25 KiB
Python
560 lines
25 KiB
Python
# -*- 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
|
||
|
||
|
||
@case("[G5] 无码原因**先抹掉数字再归类** —— 价格各不相同也要聚成一组")
|
||
def t_g5():
|
||
# 2026-08-03 实机: 8 只票同一个原因、8 个不同价格, 原来按 text[:20] 硬截,
|
||
# 价格落在截断范围内 → 聚成 8 组一组一只 (聚合等于没做), 而且截断落在词中间,
|
||
# 印出「…按现价 18.68 买」「…按现价 6.47 买不」这种半句话。
|
||
rej = [{"ts_code": c, "reasons": [f"目标金额 60 元按现价 {p} 买不足一手 "
|
||
f"(已尝试: CUSTOM → BASE/FILL → BASE)"]}
|
||
for c, p in (("600764.SH", "18.68"), ("000800.SZ", "6.47"),
|
||
("600104.SH", "11.22"), ("002594.SZ", "95.77"))]
|
||
s = summarize_rejects(rej)
|
||
assert s.count("只") == 1, f"没聚成一组: {s}"
|
||
assert "4 只" in s, s
|
||
assert "买不足一手" in s, f"截断落在词中间了: {s}"
|
||
assert "18.68" not in s and "95.77" not in s, f"具体价格不该出现在聚合标签里: {s}"
|
||
assert "已尝试" not in s, f"括号尾注是排查细节, 不该进分类标签: {s}"
|
||
|
||
|
||
@case("[G6] 有码与无码混在一起时各归各的, 仍按条数降序")
|
||
def t_g6():
|
||
rej = [{"ts_code": "A", "reasons": ["目标金额 60 元按现价 1.00 买不足一手"]},
|
||
{"ts_code": "B", "reasons": ["目标金额 60 元按现价 2.00 买不足一手"]},
|
||
{"ts_code": "C", "reasons": ["目标金额 60 元按现价 3.00 买不足一手"]},
|
||
{"ts_code": "D", "reasons": ["SECTOR_RATIO: 行业[白酒]占组合将达 55.0% > 上限 40%"]},
|
||
{"ts_code": "E", "reasons": ["STOCK_CAP: 单股占规模将达 9.1% > 上限 8%"]}]
|
||
s = summarize_rejects(rej)
|
||
assert s.index("3 只") < s.index("SECTOR_RATIO"), f"没按条数降序: {s}"
|
||
assert "SECTOR_RATIO 1 只" in s and "STOCK_CAP 1 只" in s, s
|
||
|
||
|
||
# ================================================================ [H] 一手取整的记账口径
|
||
# 2026-07-31 实机: 命令进度写着 planned_amount=600,000 / gap=0 ("完全满足"), 而 15 条方案
|
||
# 的金额合计只有 573,883 —— 少买 26,117 元 (4.4%) 且账面上完全看不出来。
|
||
# 原因是 plan_increase_exposure 的②分支按**理论目标**累计, 而①分支按取整后的实际金额累计,
|
||
# 同一个函数隔二十行两种写法。
|
||
from app.core import planner as pl # noqa: E402
|
||
|
||
|
||
def _cx(**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=None, sector_names=0, sector_mv=0.0,
|
||
sector_max_names=4, sector_max_ratio=0.40)
|
||
d.update(kw)
|
||
return d
|
||
|
||
|
||
@case("[H1] planned_amount 必须等于各条方案金额之和")
|
||
def t_h1():
|
||
r = pl.plan_increase_exposure(
|
||
add_amount=0.30 * SCALE, positions=[],
|
||
candidates=[{"ts_code": f"股{i}.SH", "price": p, "score": 1.0 - i * 0.01}
|
||
for i, p in enumerate([24.83, 25.22, 35.21, 136.73, 35.04])],
|
||
ctx=_cx(), params={"stock_target_default": 0.06})
|
||
assert abs(sum(i["amount"] for i in r["items"]) - r["planned_amount"]) < 1e-6, (
|
||
f"账面 {r['planned_amount']:,.0f} vs 方案合计 "
|
||
f"{sum(i['amount'] for i in r['items']):,.0f}")
|
||
|
||
|
||
@case("[H2] 高价股的取整损失要记进 rounding_gap, 不能被吞掉")
|
||
def t_h2():
|
||
# 136.73 元一股, 一手就要 13,673 元 —— 目标 12 万只能买到 800 股 = 109,384
|
||
r = pl.plan_increase_exposure(
|
||
add_amount=0.06 * SCALE, positions=[],
|
||
candidates=[{"ts_code": "002353.SZ", "price": 136.73, "score": 1.0}],
|
||
ctx=_cx(), params={"stock_target_default": 0.06})
|
||
assert abs(r["planned_amount"] - 109_384) < 1.0, r["planned_amount"]
|
||
assert abs(r["rounding_gap"] - (120_000 - 109_384)) < 1.0, r["rounding_gap"]
|
||
assert any("一手取整" in n for n in r["notes"]), r["notes"]
|
||
|
||
|
||
@case("[H3] 取整零头不算命令失败, 候选不够才算")
|
||
def t_h3():
|
||
# 候选够: 目标全分配出去了, 只是取整少买一点 → ok
|
||
ok_r = pl.plan_increase_exposure(
|
||
add_amount=0.06 * SCALE, positions=[],
|
||
candidates=[{"ts_code": "002353.SZ", "price": 136.73, "score": 1.0}],
|
||
ctx=_cx(), params={"stock_target_default": 0.06})
|
||
assert ok_r["ok"] is True and ok_r["shortfall"] == 0 and ok_r["rounding_gap"] > 0
|
||
# 候选不够: 要 30% 却只有一只票能给 6% → 不 ok
|
||
bad_r = pl.plan_increase_exposure(
|
||
add_amount=0.30 * SCALE, positions=[],
|
||
candidates=[{"ts_code": "002353.SZ", "price": 136.73, "score": 1.0}],
|
||
ctx=_cx(), params={"stock_target_default": 0.06})
|
||
assert bad_r["ok"] is False and bad_r["shortfall"] > 0
|
||
assert any("候选与补仓空间不足" in n for n in bad_r["notes"])
|
||
|
||
|
||
@case("[H4] 组合市值按实际金额推进, 不按理论目标 —— 否则后面几只被虚高的仓位拦掉")
|
||
def t_h4():
|
||
r = pl.plan_increase_exposure(
|
||
add_amount=0.60 * SCALE, positions=[],
|
||
candidates=[{"ts_code": f"股{i}.SH", "price": 136.73, "score": 1.0 - i * 0.01}
|
||
for i in range(10)],
|
||
ctx=_cx(), params={"stock_target_default": 0.06})
|
||
names = {i["ts_code"] for i in r["items"]}
|
||
assert len(names) == 10, (
|
||
f"十只高价股每只 6%, 按实际金额推进才装得下 10 只; 按理论目标会提前收敛, "
|
||
f"实际装了 {len(names)} 只")
|
||
|
||
|
||
@case("[H5] 个股建仓命令的 gap 也要如实反映取整损失")
|
||
def t_h5():
|
||
r = pl.plan_open_target(ts_code="002353.SZ", price=136.73, target_pct=0.06,
|
||
ctx=_cx(), params={})
|
||
assert r["ok"] and r["gap"] > 0, f"原来写死 gap=0, 等于说'完全建到目标了': {r}"
|
||
assert abs(r["planned_amount"] - sum(i["amount"] for i in r["items"])) < 1e-6
|
||
|
||
|
||
|
||
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()
|