zlt
/
tradingSystem
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
tradingSystem/scripts/test_batch11_units.py

448 lines
21 KiB
Python
Raw Normal View History

加入决策系统
2026-08-03 12:49:01 +08:00
# -*- coding: utf-8 -*-
"""
第十一批模块单测: 择时实现A (委托决策系统, 待办 #9)
================================================================
运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch11_units.py
覆盖:
[A] 硬闸拆分后与实现B的等价性 (hard_gate 是从 decide 里拆出来的, 不许拆漂)
[B] apply_advice: 决策系统应答 → 本地决策的折算与建议价保护
[C] exec_advisor: 档位短路 / 硬闸先行 / 咨询 / 缓存TTL / 失败冷却 / UNAVAILABLE 退B
[D] 参数校验与 run_tick 全链路 (实现A出手、退B出手, 内存桩)
约定同前: 全过输出 "ALL PASS (n cases)" 退出码 0。
这批的三条纪律 (与四条项目铁律对齐):
* 拿不到不等于有答案 —— 咨询失败/对端给不出结论, 一律整轮退实现B, 绝不把「拿不到」当 FIRE/WAIT。
* 配额与收尾兜底始终留在 PMS 本地 —— 本地检查命中时连咨询都不发生 (用 HTTP 计数器钉死)。
* 默认档位 B 必须与接通前一字不差 —— 实现A是加装不是改装。
"""
import os
import sys
import traceback
from datetime import datetime
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
from app.core import exec_timing as et # 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,
"advice_limit_band": 0.03}
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
def _dec(side="buy", now="10:30", d=None, quota=1000, fired=0, last=False):
return dict(side=side, now=now, day=d or day(), params=PRM,
is_last_day=last, fired_today=fired, quota=quota)
# ================================================================
# [A] 硬闸拆分后的等价性
# ================================================================
@case("[A1] 硬闸命中的场景: hard_gate 与 decide 逐字段一致 (拆分不许拆漂)")
def _():
scenarios = [
_dec(d=day(price=0)), # 无价
_dec(d=day(halted=True)), # 停牌
_dec(quota=1000, fired=1000), # 配额尽
_dec(now="09:10"), # 非时段
_dec(now="12:00"), # 午休
_dec(side="sell", d=day(limit_down=True)), # 跌停一字板
_dec(side="buy", d=day(limit_up=True)), # 涨停一字板
_dec(side="buy", d=day(day_chg_from_open=0.06)), # 不追高 STOP
_dec(side="sell", now="14:50"), # 卖出兜底
_dec(side="buy", now="14:50", last=True), # 买入末日兜底
_dec(side="buy", now="14:50", last=False), # 兜底后不新开买单
_dec(side="hold"), # 非法方向
]
for kw in scenarios:
h = et.hard_gate(**kw)
d = et.decide(**kw)
assert h is not None, f"该场景硬闸必须有决定: {kw}"
assert h == d, f"硬闸与 decide 不一致: {kw}\n hard={h}\n decide={d}"
@case("[A2] 看法区间 (盘中非兜底、无一字板): hard_gate 放行为 None, decide 给出看法")
def _():
for kw in (_dec(side="buy", now="10:30"),
_dec(side="sell", now="10:30"),
_dec(side="sell", now="09:40"), # 避开开盘属看法, 不属硬闸
_dec(side="buy", now="10:30", d=day(price=10.2, vwap=10.0))):
assert et.hard_gate(**kw) is None, kw
assert et.decide(**kw)["action"] in (et.ACT_FIRE, et.ACT_WAIT), kw
@case("[A3] 卖出跌停一字板在末日不跳过 (顺延无日可顺), 两个口径同步")
def _():
kw = _dec(side="sell", d=day(limit_down=True), last=True, now="10:30")
assert et.hard_gate(**kw) is None # 末日不 SKIP → 进看法/研判
assert et.decide(**kw)["action"] in (et.ACT_FIRE, et.ACT_WAIT)
# ================================================================
# [B] apply_advice: 研判 → 决策
# ================================================================
@case("[B1] 对端答 WAIT → WAIT, 理由带 [实现A] 前缀")
def _():
r = et.apply_advice(side="buy", day=day(), params=PRM,
advice={"verdict": "WAIT", "reason": "现价高于买入区间上沿"}, left=800)
assert r["action"] == et.ACT_WAIT and r["qty_hint"] == 800
assert r["reason"].startswith("[实现A]") and "买入区间上沿" in r["reason"]
@case("[B2] 对端答 FIRE → FIRE, 应答里的建议价直接采用")
def _():
r = et.apply_advice(side="buy", day=day(price=10.0), params=PRM,
advice={"verdict": "FIRE", "limit_price": 10.05,
"reason": "现价在买入区间内"}, left=600)
assert r["action"] == et.ACT_FIRE and r["limit_price"] == 10.05, r
assert r["forced"] is False
@case("[B3] 建议价缺失/为0 → 按本地口径 (买 ×1.002 / 卖 ×0.998)")
def _():
rb = et.apply_advice(side="buy", day=day(price=10.0), params=PRM,
advice={"verdict": "FIRE"}, left=100)
assert rb["limit_price"] == round(10.0 * 1.002, 2), rb
rs = et.apply_advice(side="sell", day=day(price=10.0), params=PRM,
advice={"verdict": "FIRE", "limit_price": 0}, left=100)
assert rs["limit_price"] == round(10.0 * 0.998, 2), rs
@case("[B4] 建议价偏离现价超 band → 本地口径重定并在理由里留痕")
def _():
r = et.apply_advice(side="buy", day=day(price=10.0), params=PRM,
advice={"verdict": "FIRE", "limit_price": 12.0}, left=100)
assert r["limit_price"] == round(10.0 * 1.002, 2), r
assert "偏离现价" in r["reason"], r["reason"]
# band 可调: 放宽到 25% 后 12.0 (偏 20%) 就该被采用
r2 = et.apply_advice(side="buy", day=day(price=10.0),
params={**PRM, "advice_limit_band": 0.25},
advice={"verdict": "FIRE", "limit_price": 12.0}, left=100)
assert r2["limit_price"] == 12.0, r2
@case("[B5] verdict 无法识别 → None (拿不到不等于有答案, 由调用方退实现B)")
def _():
for v in ("", None, "MAYBE", "PASS", "REJECT"):
assert et.apply_advice(side="buy", day=day(), params=PRM,
advice={"verdict": v}, left=100) is None, v
# ================================================================
# [C] exec_advisor 服务层 (HTTP 打桩)
# ================================================================
def _advisor(params=None, resp=None, exc=None):
"""装桩并返回 (exec_advisor, calls) —— calls 记录每次 HTTP 咨询的 payload。"""
from test_wiring import install_fakes
install_fakes(prices={}, params=params or {})
from app.services import exec_advisor as ea
calls = []
def fake_post(url, payload, timeout):
calls.append({"url": url, "payload": payload, "timeout": timeout})
if exc:
raise exc
return dict(resp or {})
ea._post = fake_post
return ea, calls
def _adecide(ea, *, side="buy", now="10:30", d=None, quota=1000, fired=0,
last=False, prog=None, pos=None):
return ea.decide(side=side, action="OPEN" if side == "buy" else "TRIM",
ts_code="600000.SH", now=now, day=d or day(), params=PRM,
is_last_day=last, fired_today=fired, quota=quota,
pos=pos or {"support_ref": 9.5, "pressure_ref": 11.0,
"total_qty": 0, "avail_qty": 0},
tdays_left=3, prog=prog)
@case("[C1] 默认档位 B: 决策与 et.decide 一字不差, 零咨询 (实现A是加装不是改装)")
def _():
ea, calls = _advisor(resp={"verdict": "FIRE"})
for kw in (dict(side="buy", now="10:30"), dict(side="sell", now="10:30"),
dict(side="sell", now="14:50"), dict(side="buy", now="09:10")):
got = _adecide(ea, **kw)
src = got.pop("source")
want = et.decide(side=kw["side"], now=kw["now"], day=day(), params=PRM,
is_last_day=False, fired_today=0, quota=1000)
assert got == want, (got, want)
assert src == "B"
assert not calls, "档位 B 不该有任何 HTTP 咨询"
@case("[C2] 档位A但接口地址全空 → 仍走 B (available=False), 状态一句话说破")
def _():
ea, calls = _advisor(params={"PMS_EXEC_IMPL": "A"})
r = _adecide(ea)
assert r["source"] == "B" and not calls
st = ea.status()
assert st["available"] is False and "地址为空" in st["note"], st
@case("[C3] 档位A: FIRE 研判 → 出手, 建议价采用, 结论写入 prog.exec_advice")
def _():
ea, calls = _advisor(params={"PMS_EXEC_IMPL": "A", "PMS_JUDGE_API_BASE": "http://b"},
resp={"verdict": "FIRE", "limit_price": 10.02,
"reason": "现价在买入区间内, 限价挂区间上沿", "valid_min": 6,
"confidence": 70})
prog = {}
r = _adecide(ea, prog=prog)
assert r["action"] == et.ACT_FIRE and r["limit_price"] == 10.02, r
assert r["source"] == "A" and "[实现A]" in r["reason"]
assert len(calls) == 1 and calls[0]["url"] == "http://b/api/intraday/pms_exec"
adv = prog["exec_advice"]
assert adv["verdict"] == "FIRE" and adv["ymd"] and adv["valid_until_min"], adv
# valid_min=6 < TTL 10 → 有效期按 6 分钟算
assert adv["valid_until_min"] == et.add_trade_minutes(et.hm_to_min("10:30"), 6), adv
@case("[C4] 有效期内复用缓存不再咨询 (source=A缓存), 过期后重新咨询")
def _():
ea, calls = _advisor(params={"PMS_EXEC_IMPL": "A", "PMS_JUDGE_API_BASE": "http://b"},
resp={"verdict": "WAIT", "reason": "等回踩", "valid_min": 5})
prog = {}
r1 = _adecide(ea, now="10:30", prog=prog)
assert r1["action"] == et.ACT_WAIT and r1["source"] == "A" and len(calls) == 1
r2 = _adecide(ea, now="10:33", prog=prog) # 5 分钟内
assert r2["source"] == "A缓存" and len(calls) == 1, (r2, len(calls))
r3 = _adecide(ea, now="10:36", prog=prog) # 过期
assert r3["source"] == "A" and len(calls) == 2, (r3, len(calls))
@case("[C5] 缓存是昨天的 → 当天首跳直接重新咨询 (隔日不吃旧结论)")
def _():
from app.core import tradedays as td
ea, calls = _advisor(params={"PMS_EXEC_IMPL": "A", "PMS_JUDGE_API_BASE": "http://b"},
resp={"verdict": "WAIT", "reason": "x"})
prog = {"exec_advice": {"ymd": td.ymd() - 1, "verdict": "FIRE",
"valid_until_min": 24 * 60, "reason": "昨日结论"}}
r = _adecide(ea, prog=prog)
assert len(calls) == 1 and r["source"] == "A", (len(calls), r)
assert prog["exec_advice"]["ymd"] == td.ymd()
@case("[C6] 咨询异常 → 本轮退实现B并进入冷却, 冷却内不再咨询")
def _():
ea, calls = _advisor(params={"PMS_EXEC_IMPL": "A", "PMS_JUDGE_API_BASE": "http://b"},
exc=RuntimeError("connect refused"))
prog = {}
r1 = _adecide(ea, now="10:30", prog=prog)
assert r1["source"].startswith("B(实现A不可用"), r1
assert "connect refused" in r1["source"]
assert len(calls) == 1
b = et.decide(side="buy", now="10:30", day=day(), params=PRM,
is_last_day=False, fired_today=0, quota=1000)
assert r1["action"] == b["action"] and r1["reason"] == b["reason"] # 真在走B
assert prog["exec_advice"]["fail_until_min"], prog
r2 = _adecide(ea, now="10:32", prog=prog) # 冷却 (默认5分钟) 内
assert len(calls) == 1 and "冷却至" in r2["source"], (len(calls), r2)
r3 = _adecide(ea, now="10:40", prog=prog) # 冷却过后恢复咨询
assert len(calls) == 2, len(calls)
@case("[C7] 对端回 UNAVAILABLE (给不出结论) → 退B + 冷却, 不当 FIRE 也不当 WAIT")
def _():
ea, calls = _advisor(params={"PMS_EXEC_IMPL": "A", "PMS_JUDGE_API_BASE": "http://b"},
resp={"verdict": "UNAVAILABLE", "reason": "昨夜结论缺失, 请退内置择时"})
prog = {}
r = _adecide(ea, prog=prog)
assert r["source"].startswith("B(实现A不可用: UNAVAILABLE"), r
assert "UNAVAILABLE" in prog["exec_advice"]["error"], prog
@case("[C8] 本地检查先行: 配额尽/兜底/不追高时零咨询 (这些永远不问决策系统)")
def _():
ea, calls = _advisor(params={"PMS_EXEC_IMPL": "A", "PMS_JUDGE_API_BASE": "http://b"},
resp={"verdict": "WAIT", "reason": "决策系统说等"})
r1 = _adecide(ea, quota=1000, fired=1000) # 配额尽
assert r1["action"] == et.ACT_WAIT and r1["source"] == "guard" and not calls
r2 = _adecide(ea, side="sell", now="14:50") # 兜底: 对端说什么都不算
assert r2["action"] == et.ACT_FIRE and r2["forced"] and r2["source"] == "guard"
assert not calls
r3 = _adecide(ea, side="buy", d=day(day_chg_from_open=0.08)) # 不追高
assert r3["action"] == et.ACT_STOP and r3["source"] == "guard" and not calls
@case("[C9] valid_min 只缩不放: 对端给 999 分钟按本端 TTL 上限截")
def _():
ea, calls = _advisor(params={"PMS_EXEC_IMPL": "A", "PMS_JUDGE_API_BASE": "http://b",
"PMS_EXEC_ADVICE_TTL_MIN": "10"},
resp={"verdict": "WAIT", "reason": "x", "valid_min": 999})
prog = {}
_adecide(ea, now="10:30", prog=prog)
assert prog["exec_advice"]["valid_until_min"] == \
et.add_trade_minutes(et.hm_to_min("10:30"), 10), prog
@case("[C10] prog=None (试算语义): 咨询照发但不缓存, 两跳两问")
def _():
ea, calls = _advisor(params={"PMS_EXEC_IMPL": "A", "PMS_JUDGE_API_BASE": "http://b"},
resp={"verdict": "WAIT", "reason": "x"})
_adecide(ea, prog=None)
_adecide(ea, prog=None)
assert len(calls) == 2, len(calls)
@case("[C11] 咨询 payload 带齐现场: 行情快照/参考位/持仓/窗口 (bionic 靠它免重取)")
def _():
ea, calls = _advisor(params={"PMS_EXEC_IMPL": "A", "PMS_JUDGE_API_BASE": "http://b"},
resp={"verdict": "WAIT", "reason": "x"})
_adecide(ea, side="sell", now="10:31",
pos={"support_ref": 9.5, "pressure_ref": 11.0, "stop_ref": 9.2,
"ref_source": "bionic", "total_qty": 3000, "avail_qty": 3000,
"avg_cost": 9.0, "cushion_pct": 0.11})
p = calls[0]["payload"]
assert p["direction"] == "PMS_EXEC" and p["side"] == "sell"
assert p["day"]["price"] == 10.0 and p["day"]["vwap"] == 10.0
assert p["refs"]["support"] == 9.5 and p["refs"]["pressure"] == 11.0
assert p["position"]["avg_cost"] == 9.0 and p["qty_left"] == 1000
assert p["now"] == "10:31" and p["tdays_left"] == 3
@case("[C12] PMS_EXEC_API_BASE 独立配置时优先于 PMS_JUDGE_API_BASE")
def _():
ea, calls = _advisor(params={"PMS_EXEC_IMPL": "A", "PMS_JUDGE_API_BASE": "http://judge",
"PMS_EXEC_API_BASE": "http://exec"},
resp={"verdict": "WAIT", "reason": "x"})
_adecide(ea, prog={})
assert calls[0]["url"].startswith("http://exec/"), calls[0]["url"]
# ================================================================
# [D] 参数校验 + run_tick 全链路
# ================================================================
@case("[D1] PMS_EXEC_IMPL 只认 A/B (大小写归一), 其他值被参数中心拒绝")
def _():
from test_wiring import install_fakes
install_fakes(prices={})
from app.services import param_store
assert param_store.set_param("PMS_EXEC_IMPL", "A")["ok"]
assert param_store.set_param("PMS_EXEC_IMPL", "b")["ok"] # 归一后合法
bad = param_store.set_param("PMS_EXEC_IMPL", "C")
assert bad["ok"] is False and "只能是" in bad["error"], bad
from app.services import exec_advisor as ea
param_store.set_param("PMS_EXEC_IMPL", "a")
assert ea.impl() == "A"
@case("[D2] run_tick·实现A出手全链路: 研判FIRE → 规则闸 → 下发, exec_advice 落表")
def _():
from test_wiring import install_fakes
fake = install_fakes(prices={"600000.SH": 10.0},
params={"PMS_TOTAL_SCALE": "2000000", "PMS_EXEC_IMPL": "A",
"PMS_JUDGE_API_BASE": "http://b"},
positions=[{"ts_code": "600000.SH", "total_qty": 6000,
"avail_qty": 6000, "avg_cost": 9.0}])
from app.services import exec_advisor as ea, executor
calls = []
def fake_post(url, payload, timeout):
calls.append(payload)
return {"verdict": "FIRE", "limit_price": 9.99, "reason": "现价在卖出区间内, 放行分批卖出",
"valid_min": 10, "confidence": 75}
ea._post = fake_post
fake.insert_instruction(instruction_id="INS_A", origin_type="plan", origin_id="P1",
ts_code="600000.SH", action="TRIM", side="sell", qty=3000,
window_tdays=3, status="PROPOSED",
progress={"deadline": "2026-07-29", "is_command": True,
"children": []})
r = executor.run_tick(now=datetime(2026, 7, 27, 10, 5))
assert r["ok"] and len(r["fired"]) == 1, r
assert r["fired"][0]["limit"] == 9.99 # 建议价被采用
assert "[实现A]" in r["fired"][0]["reason"]
ins = fake.instructions["INS_A"]
assert ins["status"] == "DISPATCHED"
assert ins["progress"]["exec_advice"]["verdict"] == "FIRE" # 结论随落表持久化
assert len(calls) == 1
# 同日第二跳: 配额已出完 → 硬闸 WAIT, 不再咨询
r2 = executor.run_tick(now=datetime(2026, 7, 27, 10, 6))
assert r2["waited"] and len(calls) == 1, (r2, len(calls))
@case("[D3] run_tick·实现A故障不停出手: 咨询挂了照走实现B, 单照下")
def _():
from test_wiring import install_fakes
fake = install_fakes(prices={"600000.SH": 10.0},
params={"PMS_TOTAL_SCALE": "2000000", "PMS_EXEC_IMPL": "A",
"PMS_JUDGE_API_BASE": "http://b"},
positions=[{"ts_code": "600000.SH", "total_qty": 6000,
"avail_qty": 6000, "avg_cost": 9.0}])
from app.services import exec_advisor as ea, executor
def dead_post(url, payload, timeout):
raise RuntimeError("bionic down")
ea._post = dead_post
fake.insert_instruction(instruction_id="INS_F", origin_type="plan", origin_id="P1",
ts_code="600000.SH", action="TRIM", side="sell", qty=3000,
window_tdays=3, status="PROPOSED",
progress={"deadline": "2026-07-29", "is_command": True,
"children": []})
# 10:05 价==vwap → 实现B本来就该出手; 实现A故障不能把这一单卡住
r = executor.run_tick(now=datetime(2026, 7, 27, 10, 5))
assert r["ok"] and len(r["fired"]) == 1, r
ins = fake.instructions["INS_F"]
assert ins["status"] == "DISPATCHED"
assert ins["progress"]["exec_advice"]["error"], ins["progress"] # 冷却已记
assert "实现A不可用" in ins["progress"]["last_decision"]["source"]
@case("[D4] run_tick·默认档位B下 exec_advice 不出现 (不接通就没有新状态)")
def _():
from test_wiring import install_fakes
fake = install_fakes(prices={"600000.SH": 10.0},
params={"PMS_TOTAL_SCALE": "2000000"},
positions=[{"ts_code": "600000.SH", "total_qty": 6000,
"avail_qty": 6000, "avg_cost": 9.0}])
from app.services import executor
fake.insert_instruction(instruction_id="INS_B0", origin_type="plan", origin_id="P1",
ts_code="600000.SH", action="TRIM", side="sell", qty=3000,
window_tdays=3, status="PROPOSED",
progress={"deadline": "2026-07-29", "is_command": True,
"children": []})
r = executor.run_tick(now=datetime(2026, 7, 27, 10, 5))
assert r["ok"] and r["fired"], r
assert "exec_advice" not in fake.instructions["INS_B0"]["progress"]
assert fake.instructions["INS_B0"]["progress"]["last_decision"]["source"] == "B"
# ---------------------------------------------------------------- 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()