# -*- coding: utf-8 -*- """ 第三批模块单测 (实机运行, 零外部依赖) ====================================== 运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch3_units.py 覆盖: exec_timing 分日配额/分笔/买卖出手判定/兜底/顺延/窗口收口; rule_gate 减持放行口径、增持全约束、命令与自主的刹车差别。 约定同前: 全过输出 "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 exec_timing as et # noqa: E402 from app.core import rule_gate as rg # noqa: E402 RESULTS = [] def case(name): def deco(fn): RESULTS.append((name, fn)) return fn return deco PRM = {"sell_avoid_open_min": 30, "buy_halt_dayup": 0.05, "eod_force_time": "14:45", "eod_force_discount": 0.998} def day(**kw): d = {"price": 10.0, "vwap": 10.0, "high": 10.5, "low": 9.5, "open": 10.0, "day_chg_from_open": 0.0, "bars": 60} d.update(kw) return d # ================================================================ exec_timing @case("分日配额·整除/上取整到一手/最后一日全出/零股尾巴并入") def _(): assert et.daily_quota(6000, 3) == 2000 assert et.daily_quota(5000, 3) == 1700 # 1666.7 → 上取整到一手 assert et.daily_quota(1000, 1) == 1000 # 最后一日全出 assert et.daily_quota(150, 3) == 150 # 尾巴不足一手 → 一次出完 assert et.daily_quota(0, 3) == 0 assert et.daily_quota(100, 5) == 100 # 整票清仓允许零股 assert et.daily_quota(14050, 3, allow_odd_tail=True) == 4700 assert et.daily_quota(14050, 1, allow_odd_tail=True) == 14050 @case("分笔·配额切成 N 笔, 余数并入最后一笔") def _(): assert et.slice_qty(3000, 3) == [1000, 1000, 1000] assert et.slice_qty(1000, 3) == [300, 300, 400] assert et.slice_qty(100, 3) == [100] assert et.slice_qty(0, 3) == [] assert et.slice_qty(2500, 1) == [2500] @case("卖出择时·避开开盘30分钟 → 站上均价才出手 → 限价打折") def _(): r = et.decide(side="sell", now="09:45", day=day(), params=PRM, is_last_day=False, quota=2000) assert r["action"] == et.ACT_WAIT and "避开开盘" in r["reason"], r r = et.decide(side="sell", now="10:05", day=day(price=10.2, vwap=10.0), params=PRM, is_last_day=False, quota=2000) assert r["action"] == et.ACT_FIRE and r["limit_price"] == 10.18, r # 10.2×0.998 r = et.decide(side="sell", now="10:05", day=day(price=9.8, vwap=10.0), params=PRM, is_last_day=False, quota=2000) assert r["action"] == et.ACT_WAIT and "等更好的价" in r["reason"], r @case("卖出择时·14:45 兜底强制出手 (不管均价)") def _(): r = et.decide(side="sell", now="14:45", day=day(price=9.5, vwap=10.0), params=PRM, is_last_day=False, quota=2000) assert r["action"] == et.ACT_FIRE and r["forced"] is True, r assert r["limit_price"] == 9.48, r # 9.5×0.998 assert et.decide(side="sell", now="14:44", day=day(price=9.5, vwap=10.0), params=PRM, is_last_day=False, quota=2000)["action"] == et.ACT_WAIT @case("买入择时·跌破均价才买 / 回踩带买 / 不追高停手") def _(): r = et.decide(side="buy", now="10:05", day=day(price=9.8, vwap=10.0), params=PRM, is_last_day=False, quota=3000) assert r["action"] == et.ACT_FIRE and r["limit_price"] == 9.82, r # 9.8×1.002 r = et.decide(side="buy", now="10:05", day=day(price=10.3, vwap=10.0), params=PRM, is_last_day=False, quota=3000) assert r["action"] == et.ACT_WAIT and "等回调" in r["reason"], r r = et.decide(side="buy", now="10:05", day=day(price=10.3, vwap=10.0, support=10.2), params=PRM, is_last_day=False, quota=3000) assert r["action"] == et.ACT_FIRE and "回踩带" in r["reason"], r r = et.decide(side="buy", now="10:05", day=day(price=11.0, vwap=10.0, day_chg_from_open=0.06), params=PRM, is_last_day=False, quota=3000) assert r["action"] == et.ACT_STOP and "不追高" in r["reason"], r @case("买入择时·窗口末日 14:45 强制完成, 非末日则顺延") def _(): r = et.decide(side="buy", now="14:50", day=day(price=10.5, vwap=10.0), params=PRM, is_last_day=True, quota=3000) assert r["action"] == et.ACT_FIRE and r["forced"] and r["limit_price"] == 10.52, r r = et.decide(side="buy", now="14:50", day=day(price=10.5, vwap=10.0), params=PRM, is_last_day=False, quota=3000) assert r["action"] == et.ACT_WAIT and "顺延次日" in r["reason"], r @case("择时·停牌/一字板/非交易时段/配额出完 一律不动") def _(): assert et.decide(side="sell", now="10:05", day=day(price=0), params=PRM, is_last_day=False, quota=100)["action"] == et.ACT_SKIP assert et.decide(side="sell", now="10:05", day=day(halted=True), params=PRM, is_last_day=False, quota=100)["action"] == et.ACT_SKIP assert et.decide(side="buy", now="10:05", day=day(limit_up=True), params=PRM, is_last_day=False, quota=100)["action"] == et.ACT_SKIP r = et.decide(side="sell", now="10:05", day=day(limit_down=True), params=PRM, is_last_day=False, quota=100) assert r["action"] == et.ACT_SKIP and "顺延" in r["reason"], r assert et.decide(side="sell", now="12:00", day=day(), params=PRM, is_last_day=False, quota=100)["action"] == et.ACT_WAIT r = et.decide(side="sell", now="10:05", day=day(price=10.2), params=PRM, is_last_day=False, quota=1000, fired_today=1000) assert r["action"] == et.ACT_WAIT and "配额已出完" in r["reason"], r @case("择时·时点解析与交易时段判定") def _(): assert et.hm_to_min("14:45") == 885 and et.hm_to_min((9, 30)) == 570 assert et.in_session(570) and et.in_session(690) and et.in_session(780) assert not et.in_session(569) and not et.in_session(700) and not et.in_session(901) @case("窗口收口·命令类置部分完成, 自主类作废, 未到期继续") def _(): assert et.window_verdict(remaining_qty=0, tdays_left=0, is_command=True)["verdict"] == "DONE" assert et.window_verdict(remaining_qty=500, tdays_left=2, is_command=True)["verdict"] == "RUNNING" v = et.window_verdict(remaining_qty=500, tdays_left=0, is_command=True) assert v["verdict"] == "PARTIAL" and "告警" in v["note"], v assert et.window_verdict(remaining_qty=500, tdays_left=0, is_command=False)["verdict"] == "EXPIRED" # ================================================================ rule_gate def ctx(**kw): d = {"ts_code": "600000.SH", "position": {"total_qty": 6000, "avail_qty": 6000, "frozen_reason": "NONE"}, "day": {"price": 10.0, "vwap": 10.0, "ma5": 10.0, "day_chg_from_open": 0.01}, "params": {"no_chase_ma5": 0.06, "buy_halt_dayup": 0.05}, "flags": {"buy_halt": False, "exec_halt": False, "brake_active": False, "blacklisted": False, "is_command": False}, "caps": dict(scale=2_000_000, portfolio_cap=0.60, stock_cap=0.08, max_names=15, portfolio_mv=800_000, names_count=5, stock_mv=60_000, is_new_name=False, sector=None, sector_names=0, sector_mv=0.0, sector_max_names=4, sector_max_ratio=0.40, cash_reserve=0.0, sector_source_ready=True)} for k, v in kw.items(): if isinstance(v, dict) and isinstance(d.get(k), dict): d[k] = {**d[k], **v} else: d[k] = v return d @case("规则闸·卖出正常放行 + 硬数字留痕") def _(): r = rg.check(side="sell", action="TRIM", qty=2000, price=10.0, ctx=ctx()) assert r["passed"] and r["failed"] == [], r assert r["hard_numbers"]["avail_qty"] == 6000 and r["hard_numbers"]["qty"] == 2000 @case("规则闸·卖出超持仓/超可卖/零股非清仓 三种拦截") def _(): r = rg.check(side="sell", action="EXIT", qty=9000, price=10.0, ctx=ctx()) assert any(x.startswith("OVER_SELL") for x in r["failed"]), r r = rg.check(side="sell", action="TRIM", qty=5000, price=10.0, ctx=ctx(position={"avail_qty": 3000})) assert any(x.startswith("T1_UNAVAILABLE") for x in r["failed"]), r r = rg.check(side="sell", action="TRIM", qty=1050, price=10.0, ctx=ctx()) assert any(x.startswith("LOT_INVALID") for x in r["failed"]), r # 清仓允许零股 r = rg.check(side="sell", action="EXIT", qty=6050, price=10.0, ctx=ctx(position={"total_qty": 6050, "avail_qty": 6050})) assert r["passed"], r @case("规则闸·减持不受冻结/刹车/上限影响 (只挡增持)") def _(): c = ctx(position={"frozen_reason": "COMMAND_HALT"}, flags={"brake_active": True, "buy_halt": True}) r = rg.check(side="sell", action="EXIT", qty=6000, price=10.0, ctx=c) assert r["passed"], r r2 = rg.check(side="buy", action="ADD", qty=1000, price=10.0, ctx=c) assert not r2["passed"] and any(x.startswith("FROZEN") for x in r2["failed"]), r2 @case("规则闸·买入 全局暂停/黑名单/非整百 拦截") def _(): r = rg.check(side="buy", action="ADD", qty=1000, price=10.0, ctx=ctx(flags={"buy_halt": True})) assert any(x.startswith("BUY_HALT") for x in r["failed"]), r r = rg.check(side="buy", action="OPEN", qty=1000, price=10.0, ctx=ctx(flags={"blacklisted": True})) assert any(x.startswith("BLACKLIST") for x in r["failed"]), r r = rg.check(side="buy", action="OPEN", qty=150, price=10.0, ctx=ctx()) assert any(x.startswith("LOT_INVALID") for x in r["failed"]), r @case("规则闸·刹车对自主是拦截, 对命令只是提示 (命令至上)") def _(): auto = rg.check(side="buy", action="ADD", qty=1000, price=10.0, ctx=ctx(flags={"brake_active": True})) assert not auto["passed"] and any(x.startswith("BRAKE_ACTIVE") for x in auto["failed"]) cmd = rg.check(side="buy", action="OPEN", qty=1000, price=10.0, ctx=ctx(flags={"brake_active": True, "is_command": True})) assert cmd["passed"], cmd assert any(w.startswith("BRAKE_ACTIVE") for w in cmd["warnings"]), cmd @case("规则闸·不追高两道 (当日涨幅 / 距 MA5) 与 MA5 缺失告警") def _(): r = rg.check(side="buy", action="OPEN", qty=1000, price=10.0, ctx=ctx(day={"day_chg_from_open": 0.07})) assert any(x.startswith("NO_CHASE_DAYUP") for x in r["failed"]), r r = rg.check(side="buy", action="OPEN", qty=1000, price=11.0, ctx=ctx(day={"ma5": 10.0})) assert any(x.startswith("NO_CHASE_MA5") for x in r["failed"]), r # 距 MA5 10% > 6% r = rg.check(side="buy", action="OPEN", qty=1000, price=10.3, ctx=ctx(day={"ma5": 10.0})) assert r["passed"], r # 3% 在容忍内 r = rg.check(side="buy", action="OPEN", qty=1000, price=10.0, ctx=ctx(day={"ma5": None})) assert r["passed"] and any(w.startswith("MA5_MISSING") for w in r["warnings"]), r @case("规则闸·买入过组合上限 (复用 planner 同一份口径)") def _(): r = rg.check(side="buy", action="ADD", qty=12000, price=10.0, ctx=ctx()) assert any(x.startswith("STOCK_CAP") for x in r["failed"]), r # 6万+12万 > 16万 r = rg.check(side="buy", action="ADD", qty=1000, price=10.0, ctx=ctx(caps=None)) assert any(x.startswith("CAPS_MISSING") for x in r["failed"]), r @case("规则闸·必要输入缺失与全局暂停执行一律拒绝 (宁可不动)") def _(): r = rg.check(side="sell", action="EXIT", qty=1000, price=0, ctx=ctx()) assert any(x.startswith("PRICE_MISSING") for x in r["failed"]), r r = rg.check(side="sell", action="EXIT", qty=0, price=10.0, ctx=ctx()) assert any(x.startswith("QTY_INVALID") for x in r["failed"]), r r = rg.check(side="sell", action="EXIT", qty=1000, price=10.0, ctx=ctx(flags={"exec_halt": True})) assert any(x.startswith("EXEC_HALT") for x in r["failed"]), r r = rg.check(side="sell", action="EXIT", qty=1000, price=10.0, ctx=ctx(day={"halted": True})) assert any(x.startswith("HALTED") for x in r["failed"]), r r = rg.check(side="hold", action="X", qty=100, price=10.0, ctx=ctx()) assert any(x.startswith("SIDE_INVALID") for x in r["failed"]), r @case("规则闸·批量统计 (日报关注区用)") def _(): rs = [rg.check(side="sell", action="EXIT", qty=1000, price=10.0, ctx=ctx()), rg.check(side="buy", action="OPEN", qty=150, price=10.0, ctx=ctx()), rg.check(side="buy", action="OPEN", qty=100, price=10.0, ctx=ctx(flags={"buy_halt": True}))] s = rg.summarize(rs) assert s["total"] == 3 and s["passed"] == 1 and s["rejected"] == 2, s assert s["by_reason"].get("LOT_INVALID") == 1 and s["by_reason"].get("BUY_HALT") == 1, s # ---------------------------------------------------------------- runner def main(): 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()