From b7abd8c3fe09135bc1d87c807eed3ba060b79ed7 Mon Sep 17 00:00:00 2001 From: zlt Date: Thu, 6 Aug 2026 13:55:43 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E8=87=AA=E5=8A=A8=E5=BB=BA?= =?UTF-8?q?=E4=BB=93=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/test_batch12_units.py | 392 ++++++++++++++++++++++++++++++++++ 1 file changed, 392 insertions(+) create mode 100644 scripts/test_batch12_units.py diff --git a/scripts/test_batch12_units.py b/scripts/test_batch12_units.py new file mode 100644 index 0000000..f3296c3 --- /dev/null +++ b/scripts/test_batch12_units.py @@ -0,0 +1,392 @@ +# -*- coding: utf-8 -*- +""" +第十二批模块单测 (实机运行, 零外部依赖, 不连库) +================================================== +运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch12_units.py +覆盖: 自主新建仓 (动作引擎 OPEN) 的选票、数量、名额与金额的滚动扣减、行业集中度滚动; + 研判请求的硬数字裁剪 (仓位数字不送决策系统); + 参考位漂移的当日首答锁定与偏离即停; + 研判的单轮时间预算。 +约定同前: 全过输出 "ALL PASS (n cases)" 退出码 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 action_engine as ae # noqa: E402 +from app.services import exec_advisor as ea # noqa: E402 +from app.services import judge as jd # noqa: E402 +from app.services import param_store as ps # noqa: E402 + +RESULTS = [] + + +def case(name): + def deco(fn): + RESULTS.append((name, fn)) + return fn + return deco + + +# ================================================================ 夹具 +def params(**kw): + """action_engine 那侧要的参数快照 (口径同 param_store.sizing_params)。""" + p = {"scale": 2_000_000.0, "stock_target_default": 0.06, + "batch_split": (0.5, 0.25, 0.25), "min_lot_merge": True} + p.update(kw) + return p + + +def caps(**kw): + """portfolio.caps_ctx 的产出形状 (组合快照)。""" + c = {"scale": 2_000_000.0, "portfolio_cap": 0.70, "stock_cap": 0.08, + "max_names": 20, "portfolio_mv": 0.0, "names_count": 0, + "stock_mv": 0.0, "is_new_name": False, "sector": None, + "sector_names": 0, "sector_mv": 0.0, + "sector_names_map": {}, "sector_mv_map": {}, + "sector_max_names": 4, "sector_max_ratio": 0.40, + "cash_reserve": 0.0, "sector_source_ready": True, + "cash_avail": None, "cash_source": "estimate"} + c.update(kw) + return c + + +def cand(code, price=10.0, score=100.0, sector=None, rank=1, **kw): + d = {"ts_code": code, "price": price, "score": score, "sector": sector, + "theme": kw.pop("theme", None), "tier": kw.pop("tier", "强传导"), + "upside": kw.pop("upside", 0.3), "heat": kw.pop("heat", 0.5), + "rank": rank, "bucket": "main", "src": "plan_api"} + d.update(kw) + return d + + +class _PatchParams: + """把 param_store 的取值函数临时换成一份字典 —— 单测不连库。""" + + def __init__(self, values): + self.values = values + self.saved = {} + + def __enter__(self): + for name in ("get", "get_int", "get_float", "get_bool"): + self.saved[name] = getattr(ps, name) + v = self.values + ps.get = lambda k, d=None: v.get(k, d) + ps.get_int = lambda k, d=0: int(v.get(k, d)) + ps.get_float = lambda k, d=0.0: float(v.get(k, d)) + ps.get_bool = lambda k, d=False: bool(v.get(k, d)) + return self + + def __exit__(self, *a): + for name, fn in self.saved.items(): + setattr(ps, name, fn) + return False + + +# ================================================================ 动作引擎: 新建仓 +@case("新建仓·正常产出: 只提底仓批, 数量按 50% 取整到一手") +def _(): + # 目标 6% × 200 万 = 12 万; 底仓 50% = 6 万; 现价 10 元 → 6000 股 + r = ae.scan_open(candidates=[cand("600000.SH")], params=params(), caps=caps(), + room_amt=1_400_000, slots=20) + assert len(r["candidates"]) == 1, r + c = r["candidates"][0] + assert c["action"] == "OPEN" and c["side"] == "buy", c + assert c["qty"] == 6000, c + assert c["judge_required"] is True, "OPEN 必须有资格送研判闸" + assert c["needs_user_confirm"] is False, "新建仓不强制人工确认 (档位说了算)" + # 现价与行业必须挂在候选上 —— 新票在账本里没有行, 服务层取不到这两样 + assert c["price"] == 10.0 and "sector" in c, c + hn = c["hard_numbers"] + assert hn["score"] == 100.0 and hn["price"] == 10.0, hn + assert hn["target_amount"] == 120000.0, hn # 三批取整后正好 12 万 + assert hn["base_amount"] == 60000.0, hn + + +@case("新建仓·没有名额时一条不产, 且说得出为什么") +def _(): + r = ae.scan_open(candidates=[cand("600000.SH")], params=params(), + caps=caps(names_count=20), room_amt=1_400_000, slots=0) + assert r["candidates"] == [], r + assert "名额" in r["skipped"][0]["why"], r["skipped"] + + +@case("新建仓·没有可投金额时一条不产, 且说得出为什么") +def _(): + r = ae.scan_open(candidates=[cand("600000.SH")], params=params(), caps=caps(), + room_amt=0, slots=20) + assert r["candidates"] == [], r + assert "可投金额" in r["skipped"][0]["why"], r["skipped"] + + +@case("新建仓·取不到实时价整只跳过并留痕 (昨收不能拿来下单)") +def _(): + r = ae.scan_open(candidates=[cand("600000.SH", price=None)], params=params(), + caps=caps(), room_amt=1_400_000, slots=20) + assert r["candidates"] == [], r + assert "实时价" in r["skipped"][0]["why"], r["skipped"] + + +@case("新建仓·底仓批买不足一手 → 不提, 原因是一手不是别的") +def _(): + # 目标 6% × 10 万 = 6000 元; 现价 500 元一手就要 5 万 → 拆不出 + r = ae.scan_open(candidates=[cand("600000.SH", price=500.0)], + params=params(scale=100_000.0), caps=caps(scale=100_000.0), + room_amt=70_000, slots=20) + assert r["candidates"] == [], r + assert "一手" in r["skipped"][0]["why"], r["skipped"] + + +@case("新建仓·名额边走边扣: 只剩一个名额就只产一条, 其余留到下一跳") +def _(): + cs = [cand("600000.SH", score=300, rank=1), cand("600001.SH", score=200, rank=2), + cand("600002.SH", score=100, rank=3)] + r = ae.scan_open(candidates=cs, params=params(), caps=caps(names_count=19), + room_amt=1_400_000, slots=1) + assert [c["ts_code"] for c in r["candidates"]] == ["600000.SH"], r["candidates"] + whys = [s["why"] for s in r["skipped"]] + assert any("名额已用完" in w for w in whys), whys + + +@case("新建仓·金额边走边扣: 三条候选只够两条, 第三条被金额挡下") +def _(): + # 每条整只目标 12 万; 可投 25 万 → 两条 (24 万) 之后只剩 1 万, 不够一整只 + cs = [cand("600000.SH", score=300, rank=1), cand("600001.SH", score=200, rank=2), + cand("600002.SH", score=100, rank=3)] + r = ae.scan_open(candidates=cs, params=params(), caps=caps(), + room_amt=250_000, slots=20) + assert [c["ts_code"] for c in r["candidates"]] == ["600000.SH", "600001.SH"], r["candidates"] + whys = " ".join(s["why"] for s in r["skipped"]) + assert "不开半截新仓" in whys, whys + + +@case("新建仓·钱不够一整只就不开 (不拿名额去换零头仓位)") +def _(): + # 目标 12 万, 只剩 5 万 —— 拆得出一手批次, 但那是个补不到目标的零头仓位 + r = ae.scan_open(candidates=[cand("600000.SH")], params=params(), caps=caps(), + room_amt=50_000, slots=20) + assert r["candidates"] == [], r + assert "不开半截新仓" in r["skipped"][0]["why"], r["skipped"] + + +@case("新建仓·滚动扣减真的生效: 组合上限只容得下两条时不会三条都放出来") +def _(): + # 总仓上限 70% × 200 万 = 140 万, 已有 116 万 → 只剩 24 万 = 两只 + # 如果不滚动 (每条都拿同一份旧快照算), 三条会各自都通过 —— 这一例就是为它写的 + cs = [cand(f"60000{i}.SH", score=300 - i, rank=i + 1) for i in range(3)] + r = ae.scan_open(candidates=cs, params=params(), + caps=caps(portfolio_mv=1_160_000, names_count=5), + room_amt=240_000, slots=20) + assert len(r["candidates"]) == 2, [c["ts_code"] for c in r["candidates"]] + + +@case("新建仓·行业集中度也滚动: 同行业名额只剩一个时第二只被拦") +def _(): + cs = [cand("600000.SH", score=300, rank=1, sector="储能"), + cand("600001.SH", score=200, rank=2, sector="储能")] + r = ae.scan_open(candidates=cs, params=params(), + caps=caps(sector_max_names=1), room_amt=1_400_000, slots=20) + assert len(r["candidates"]) == 1, [c["ts_code"] for c in r["candidates"]] + whys = " ".join(s["why"] for s in r["skipped"]) + assert "SECTOR_NAMES" in whys, whys + + +@case("新建仓·行业数据源没配时不误拦 (sector 一律按空处理)") +def _(): + cs = [cand("600000.SH", score=300, rank=1, sector="储能"), + cand("600001.SH", score=200, rank=2, sector="储能")] + r = ae.scan_open(candidates=cs, params=params(), + caps=caps(sector_max_names=1, sector_source_ready=False), + room_amt=1_400_000, slots=20) + assert len(r["candidates"]) == 2, [c["ts_code"] for c in r["candidates"]] + + +@case("新建仓·已有在途或今日被拒过的票不重复提") +def _(): + r = ae.scan_open(candidates=[cand("600000.SH")], params=params(), caps=caps(), + room_amt=1_400_000, slots=20, skip={("600000.SH", "OPEN")}) + assert r["candidates"] == [], r + assert "在途" in r["skipped"][0]["why"], r["skipped"] + + +@case("新建仓·单票评估异常不拖垮整轮, 后面的票照常评") +def _(): + orig = ae.eval_open + + def boom(c, *a, **kw): + if c["ts_code"] == "600000.SH": + raise RuntimeError("求值炸了") + return orig(c, *a, **kw) + + ae.eval_open = boom + try: + cs = [cand("600000.SH", score=300, rank=1), cand("600001.SH", score=200, rank=2)] + r = ae.scan_open(candidates=cs, params=params(), caps=caps(), + room_amt=1_400_000, slots=20) + finally: + ae.eval_open = orig + assert [c["ts_code"] for c in r["candidates"]] == ["600001.SH"], r["candidates"] + whys = " ".join(s["why"] for s in r["skipped"]) + assert "评估异常" in whys and "RuntimeError" in whys, whys + + +@case("回归·既有四类动作的判据与研判范围没被动过") +def _(): + assert ae.JUDGE_ACTIONS == {"FILL", "ADD", "DCA", "OPEN"}, ae.JUDGE_ACTIONS + assert "TRIM" not in ae.JUDGE_ACTIONS, "减持永远不送研判" + assert [a for a, _ in ae.EVALUATORS] == ["TRIM", "ADD", "FILL", "DCA"] + # 持仓那条扫描的输入与产出一个字没改: 空持仓进去, 空候选出来, 不会去碰候选池 + r = ae.scan(positions=[], params=params(), market={}) + assert r == {"candidates": [], "skipped": []}, r + + +# ================================================================ 研判请求的硬数字裁剪 +@case("研判请求·新建仓只送定性材料, 仓位数字一个不送") +def _(): + hard = {"price": 10.0, "score": 300.0, "theme": "储能", "tier": "强传导", + "upside": 0.3, "heat": 0.5, "plan_rank": 1, "plan_bucket": "main", + "plan_src": "plan_api", "sector": "电力设备", + # 下面这些是仓位口径, 决策系统本来就不管仓位 + "target_pct": 0.06, "target_amount": 120000.0, "base_amount": 60000.0, + "batch_scheme": "0.5,0.25,0.25", "names_before": 3, "max_names": 20, + "room_amt_before": 1_400_000.0} + got = jd._judge_hard_numbers("OPEN", hard) + assert set(got) == set(jd.OPEN_JUDGE_KEYS), got + for k in ("target_amount", "names_before", "max_names", "room_amt_before", + "batch_scheme", "base_amount", "target_pct"): + assert k not in got, f"{k} 不该送到决策系统" + assert got["score"] == 300.0 and got["price"] == 10.0, got + + +@case("研判请求·加仓类原样送 (它们的硬数字本身就是判据)") +def _(): + hard = {"cushion_pct": -0.085, "stage": 1, "base_qty": 1000} + for act in ("FILL", "ADD", "DCA", "SWITCH"): + assert jd._judge_hard_numbers(act, hard) == hard, act + assert jd._judge_hard_numbers("DCA", None) == {} + + +# ================================================================ 参考位漂移 +@case("参考位漂移·只管新建仓, 别的动作一律不比对") +def _(): + with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.03}): + prog = {"ref_lock": {"ymd": 20260806, "support": 31.27, "pressure": 40.0}} + adv = {"support": 29.00, "pressure": 40.0} + assert ea._check_ref_drift(action="FILL", prog=prog, adv=adv, today=20260806) == "" + assert ea._check_ref_drift(action="ADD", prog=prog, adv=adv, today=20260806) == "" + assert ea._check_ref_drift(action=None, prog=prog, adv=adv, today=20260806) == "" + + +@case("参考位漂移·当日首答只锁定不比对") +def _(): + with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.03}): + prog = {} + adv = {"support": 31.27, "pressure": 40.0, "buy_band": [30.96, 33.89], + "consulted_at": "09:35"} + assert ea._check_ref_drift(action="OPEN", prog=prog, adv=adv, today=20260806) == "" + assert prog["ref_lock"]["ymd"] == 20260806 + assert prog["ref_lock"]["support"] == 31.27 + assert prog["ref_lock"]["at"] == "09:35" + + +@case("参考位漂移·没漂过阈值就放行") +def _(): + with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.03}): + prog = {"ref_lock": {"ymd": 20260806, "support": 31.27, "pressure": 40.0}} + adv = {"support": 31.50, "pressure": 40.0} # 差 0.7% + assert ea._check_ref_drift(action="OPEN", prog=prog, adv=adv, today=20260806) == "" + + +@case("参考位漂移·漂过阈值就报出来 (002335 实证: 支撑 31.27→29.00)") +def _(): + with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.03}): + prog = {"ref_lock": {"ymd": 20260806, "support": 31.27, "pressure": 40.0, + "at": "09:35"}} + adv = {"support": 29.00, "pressure": 40.0} + why = ea._check_ref_drift(action="OPEN", prog=prog, adv=adv, today=20260806) + assert why, "漂了 7.3% 必须报出来" + assert "31.27" in why and "29.0" in why, why + assert "当日暂停" in why, why + + +@case("参考位漂移·压力位单独漂也算 (000035 实证: 压力 5.2→5.15 未过阈值)") +def _(): + with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.03}): + prog = {"ref_lock": {"ymd": 20260806, "support": 5.0, "pressure": 5.2}} + assert ea._check_ref_drift(action="OPEN", prog=prog, + adv={"support": 5.0, "pressure": 5.15}, + today=20260806) == "" # 差 0.96%, 不到阈值 + why = ea._check_ref_drift(action="OPEN", prog=prog, + adv={"support": 5.0, "pressure": 4.5}, + today=20260806) # 差 13.5% + assert "压力" in why, why + + +@case("参考位漂移·锁按日重置, 次日的新结论不算漂移") +def _(): + with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.03}): + prog = {"ref_lock": {"ymd": 20260806, "support": 31.27, "pressure": 40.0}} + adv = {"support": 29.00, "pressure": 40.0, "consulted_at": "09:31"} + assert ea._check_ref_drift(action="OPEN", prog=prog, adv=adv, today=20260807) == "" + assert prog["ref_lock"]["ymd"] == 20260807, "次日首答应该重新锁定" + assert prog["ref_lock"]["support"] == 29.00 + + +@case("参考位漂移·阈值设成 0 等于关掉这道防护") +def _(): + with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.0}): + prog = {"ref_lock": {"ymd": 20260806, "support": 31.27, "pressure": 40.0}} + adv = {"support": 1.0, "pressure": 40.0} + assert ea._check_ref_drift(action="OPEN", prog=prog, adv=adv, today=20260806) == "" + + +@case("参考位漂移·拿不到支撑压力时不锁也不误报") +def _(): + with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.03}): + prog = {} + assert ea._check_ref_drift(action="OPEN", prog=prog, + adv={"support": None, "pressure": None}, + today=20260806) == "" + assert "ref_lock" not in prog, "没有位就不该锁一份空的进去" + + +# ================================================================ 研判的单轮时间预算 +@case("研判预算·没设预算就一律放行") +def _(): + from app.services import proposal_service as prs + assert prs._judge_budget_left(None) is True + + +@case("研判预算·够一次就放行, 不够就留到下一跳") +def _(): + import time + from app.services import proposal_service as prs + with _PatchParams({"PMS_JUDGE_TIMEOUT": 90}): + assert prs._judge_budget_left(time.monotonic() + 150) is True # 150 > 90 + assert prs._judge_budget_left(time.monotonic() + 30) is False # 30 < 90 + assert prs._judge_budget_left(time.monotonic() - 1) is False # 已经超了 + + +# ================================================================ 跑 +def main(): + passed, failed = 0, [] + for name, fn in RESULTS: + try: + fn() + passed += 1 + print(f" ✓ {name}") + except Exception as e: + failed.append((name, e)) + print(f" ✗ {name}: {type(e).__name__}: {e}") + traceback.print_exc() + print() + if failed: + print(f"FAILED {len(failed)}/{len(RESULTS)}") + sys.exit(1) + print(f"ALL PASS ({passed} cases)") + + +if __name__ == "__main__": + main()