# -*- coding: utf-8 -*- """ 第十三批模块单测 (实机运行, 零外部依赖) ======================================== 运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch13_units.py 覆盖 (2026-08-18 决策系统卖出采纳复核后的三处改动): * 消化门槛降到 0.60 后 digest 的分档 (纯逻辑); * 清仓完成清场的"每次闭仓只清一次"标记 (_cleanup_todo 纯逻辑); * 确认即加速: 决策系统卖出命中在途清仓 → 只把卖出**指令**升级紧急 (打桩 repo); * 闭仓清场: 撤策略/在途建仓/买单/提议, 幂等第二遍无动作 (打桩 repo/service)。 """ import os import sys import traceback sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.core import signal_rules as sr # noqa: E402 from app.services import command_service as csvc # noqa: E402 from app.services import signal_service as ssvc # noqa: E402 from app.repo import pms_repo # noqa: E402 RESULTS = [] def case(name): def deco(fn): RESULTS.append((name, fn)) return fn return deco HELD = {"ts_code": "300474.SZ", "total_qty": 3000, "avail_qty": 3000, "price": 8.0} PRM60 = {"sell_conf_min": 0.60, "auto_exit_conf": 0.85, "trim_ratio": 1 / 3} # ================================================================ 门槛分档 @case("门槛 0.60·持仓票卖出: <60 忽略 / 60~84 提议 / ≥85 清仓") def _(): def sell(conf): return {"source": "risk_sell", "ts_code": "300474.SZ", "action": "SELL", "confidence": conf} assert sr.digest(sell(0.55), HELD, PRM60)["action"] == sr.ACT_IGNORE assert sr.digest(sell(0.60), HELD, PRM60)["action"] == sr.ACT_PROPOSE # 恰好到线, 采纳 assert sr.digest(sell(0.72), HELD, PRM60)["action"] == sr.ACT_PROPOSE # 实测这一带最多 assert sr.digest(sell(0.84), HELD, PRM60)["action"] == sr.ACT_PROPOSE assert sr.digest(sell(0.85), HELD, PRM60)["action"] == sr.ACT_EXIT assert sr.digest(sell(0.92), HELD, PRM60)["action"] == sr.ACT_EXIT # 未持有仍一律忽略 (不因降门槛就对没持仓的票动手) assert sr.digest(sell(0.95), {"total_qty": 0}, PRM60)["action"] == sr.ACT_IGNORE # ================================================================ 清场标记 (纯逻辑) @case("清场标记·每次闭仓只清一次, 重开的票掉出标记下次再清") def _(): # 首轮: A/B 刚闭仓, 都要清 todo, done = csvc._cleanup_todo({"A", "B"}, set()) assert todo == ["A", "B"] and done == {"A", "B"} # 次轮: 还是 A/B 闭仓且已清过 → 不再清 todo, done = csvc._cleanup_todo({"A", "B"}, {"A", "B"}) assert todo == [] and done == {"A", "B"} # A 重新建仓(掉出闭仓集), 新来 C 闭仓 → 只清 C; A 从标记移除 todo, done = csvc._cleanup_todo({"B", "C"}, {"A", "B"}) assert todo == ["C"] and done == {"B", "C"} and "A" not in done # A 又闭仓 → 因已移出标记, 会被重新清 todo, done = csvc._cleanup_todo({"A", "B", "C"}, {"B", "C"}) assert todo == ["A"] and done == {"A", "B", "C"} # ---------------------------------------------------------------- 打桩工具 class _Rec: """记录型假 repo/service: 存住调用, 供断言。""" def __init__(self): self.instr = [] # 在途指令 self.updates = {} # instruction_id -> progress self.ledger = [] self.plans = [] self.strategies = [] self.proposals = [] self.positions = [] self.plan_cancelled = [] self.prop_declined = [] self.strat_cancelled = [] self.instr_cancelled = [] self.param = {} def _patch_signal(rec): pms_repo.list_instructions = lambda **kw: [i for i in rec.instr if (kw.get("ts_code") in (None, i["ts_code"])) and (kw.get("side") in (None, i.get("side")))] def _upd(iid, **kw): if "progress" in kw: rec.updates[iid] = kw["progress"] return 1 pms_repo.update_instruction = _upd pms_repo.insert_ledger = lambda **kw: rec.ledger.append(kw) @case("确认加速·只升级在途卖出指令, 买单与已紧急的不动") def _(): rec = _Rec() rec.instr = [ {"instruction_id": "S1", "ts_code": "600150.SH", "side": "sell", "progress": {}}, {"instruction_id": "S2", "ts_code": "600150.SH", "side": "sell", "progress": {"urgent": True}}, # 已紧急, 不重复 {"instruction_id": "B1", "ts_code": "600150.SH", "side": "buy", "progress": {}}, ] _patch_signal(rec) ids = ssvc._escalate_inflight_sells("600150.SH", "决策系统确认卖出") assert ids == ["S1"], ids # 只 S1 被升级 assert rec.updates["S1"]["urgent"] is True assert "escalated_by_signal" in rec.updates["S1"] assert "B1" not in rec.updates and "S2" not in rec.updates assert any(l.get("action") == "EXIT" and "紧急直通" in l.get("reason", "") for l in rec.ledger) def _patch_cleanup(rec): import app.services.executor as ex import app.services.strategy_service as strat pms_repo.list_positions = lambda **kw: rec.positions pms_repo.list_strategies = lambda **kw: [s for s in rec.strategies if s.get("status") in (kw.get("statuses") or [s.get("status")])] pms_repo.list_plans = lambda **kw: rec.plans pms_repo.list_instructions = lambda **kw: [i for i in rec.instr if kw.get("side") in (None, i.get("side"))] pms_repo.list_proposals = lambda **kw: rec.proposals pms_repo.update_plan = lambda pid, **kw: rec.plan_cancelled.append(pid) or 1 pms_repo.decide_proposal = lambda pid, st: rec.prop_declined.append((pid, st)) or 1 pms_repo.insert_ledger = lambda **kw: rec.ledger.append(kw) pms_repo.get_param = lambda k: rec.param.get(k) pms_repo.set_param = lambda k, v, by=None: rec.param.__setitem__(k, v) strat.set_status = lambda sid, st, by=None: (rec.strat_cancelled.append((sid, st)) or {"ok": True}) ex.cancel_instruction = lambda iid, reason=None: (rec.instr_cancelled.append(iid) or {"ok": True}) @case("闭仓清场·撤策略/建仓方案/买单/提议, 幂等第二遍无动作") def _(): rec = _Rec() rec.positions = [ {"ts_code": "300474.SZ", "status": "CLOSED", "total_qty": 0}, # 刚闭仓, 要清 {"ts_code": "600150.SH", "status": "HOLDING", "total_qty": 1700}, # 在持, 不碰 ] rec.strategies = [{"strategy_id": "G1", "ts_code": "300474.SZ", "status": "ACTIVE"}, {"strategy_id": "G2", "ts_code": "600150.SH", "status": "ACTIVE"}] rec.plans = [{"plan_id": "P1", "ts_code": "300474.SZ", "action": "OPEN"}, {"plan_id": "P2", "ts_code": "300474.SZ", "action": "EXIT"}] # 卖出方案不撤 rec.instr = [{"instruction_id": "IB", "ts_code": "300474.SZ", "side": "buy"}] rec.proposals = [{"proposal_id": "PR1", "ts_code": "300474.SZ", "action": "ADD"}] _patch_cleanup(rec) r = csvc.cleanup_exited_positions() assert len(r["cleaned"]) == 1 and r["cleaned"][0]["ts_code"] == "300474.SZ", r assert rec.strat_cancelled == [("G1", "CANCELLED")] # 只撤闭仓票的策略 assert rec.plan_cancelled == ["P1"] # 只撤建仓方案, 不撤 EXIT assert rec.instr_cancelled == ["IB"] assert rec.prop_declined == [("PR1", "DECLINED")] assert any(l.get("action") == "CLEANUP" for l in rec.ledger) assert "300474.SZ" in rec.param.get(csvc.CLEANUP_MARK_KEY, "") # 幂等: 第二遍 (标记已记 300474) → 不再动任何东西 rec.strat_cancelled.clear(); rec.plan_cancelled.clear() rec.instr_cancelled.clear(); rec.prop_declined.clear() r2 = csvc.cleanup_exited_positions() assert r2["cleaned"] == [] and not rec.strat_cancelled and not rec.plan_cancelled, r2 # ================================================================ runner def main(): passed = failed = 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()