452 lines
22 KiB
Python
452 lines
22 KiB
Python
# -*- 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, cmd=True):
|
||
# cmd 默认 True = 命令口径 = 2026-08-06 之前的行为, 所以下面这批等价性用例的判据没变。
|
||
# 那天给窗口末日的强制完成分了岔 (命令照旧强制完成, 自主到期作废), hard_gate 的
|
||
# is_command 是**必填**的 —— 漏传立刻 TypeError, 不许静默按某一侧走, 所以这里补上。
|
||
# 分岔本身的行为在 test_batch12_units.py 的「窗口末日」那组里锁着。
|
||
return dict(side=side, now=now, day=d or day(), params=PRM,
|
||
is_last_day=last, is_command=cmd, 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()
|