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

735 lines
36 KiB
Python
Raw Normal View History

添加交易逻辑
2026-08-25 13:45:51 +08:00
# -*- coding: utf-8 -*-
"""
第十七批: 策略自动挂载 (个股打法状态机) —— STRATEGY_AUTO_ATTACH_PLAN.md 步骤二
==============================================================================
零外部依赖, 不连库。四块内容:
1. 纯逻辑: 定性归类 (子串包含+保守优先级) / 网格参数生成与放弃路径 /
连边矩阵 (双命中取止盈、负垫不挂) / 日上限计数 (接力不占) / 冷却推导 / 接力判定;
2. note 约定字面量钉死 —— 冷却与计数全靠它, 改一个字就把无状态推导改坏;
3. strategy_service.clear_buypause 的来源匹配 (只清自己停的, 不放开风控停的);
4. 编排冒烟 (全打桩): dry_run 滴水不写 / 真挂走 attach+留痕 / 名额满挡下 /
边三停买腿与恢复 / 边四接力全链 (撤网格→打标记→挂止盈→初始化高水位)。
运行: python scripts/test_batch17_units.py
"""
import os
import sys
import traceback
from types import SimpleNamespace
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.services import param_store # noqa: E402
from app.services import strategy_advisor as adv # noqa: E402
from app.services import strategy_service as svc # noqa: E402
from app.repo import pms_repo # noqa: E402
RESULTS = []
TODAY = 20260825
def case(name):
def deco(fn):
RESULTS.append((name, fn))
return fn
return deco
def _prm(**kw):
d = {"enabled": True,
"rules": {adv.R_GRID, adv.R_TRAIL, adv.R_EXIT, adv.R_HANDOFF},
"daily_max": 2, "accum_stale_tdays": 3, "grid_band": 0.08,
"grid_step_pct": 0.02, "grid_cap_ratio": 0.5, "heat_th": 0.80,
"trail_giveback": 0.05, "trail_sell_ratio": 0.5, "handoff_enabled": True,
"handoff_cooldown_tdays": 10, "optout_cooldown_tdays": 10}
d.update(kw)
return d
def _out():
return {"ok": True, "checked": 0, "attached": [], "handoffs": [], "paused": [],
"resumed": [], "blocked": [], "skipped": [], "errors": [],
"unknown_states": []}
def _pos(**kw):
d = {"ts_code": "600000.SH", "total_qty": 1000, "avail_qty": 1000,
"price": 10.0, "price_ok": True, "market_value": 10000.0,
"cushion_pct": 0.05, "support_ref": 9.0, "pressure_ref": 11.0,
"frozen_reason": "NONE"}
d.update(kw)
return d
# ================================================================
# [一] 定性归类: 子串包含 + 保守优先级
# ================================================================
@case("[归类] 五档标准词各归各位 (契约钉死)")
def _():
assert adv.classify_accum("明确吸筹") == adv.CLS_CLEAR
assert adv.classify_accum("潜在吸筹") == adv.CLS_MAYBE
assert adv.classify_accum("无吸筹迹象") == adv.CLS_NONE_SIGN
assert adv.classify_accum("高位派发") == adv.CLS_DISTRIB
assert adv.classify_accum("信号不明") == adv.CLS_UNCLEAR
@case("[归类] 子串包含: 带前后缀修饰照样认 (探测实测 62 只词表外的根因)")
def _():
# 数据底座 feed.py 的 _ACCUM_KEEP 用 any(k in state), 前缀匹配会把这些全漏掉
assert adv.classify_accum("明确吸筹(强)") == adv.CLS_CLEAR
assert adv.classify_accum("近期呈潜在吸筹迹象") == adv.CLS_MAYBE
assert adv.classify_accum("→ 高位派发风险") == adv.CLS_DISTRIB
assert adv.classify_accum("走势信号不明朗") == adv.CLS_UNCLEAR
@case("[归类] 两词同现按保守方向: 派发优先于明确吸筹")
def _():
assert adv.classify_accum("前期明确吸筹, 现转高位派发") == adv.CLS_DISTRIB
assert adv.classify_accum("高位派发后再现潜在吸筹") == adv.CLS_DISTRIB
@case("[归类] 空值→无字段, 认不出→词表外 (调用方一律按无标志处理)")
def _():
assert adv.classify_accum(None) == adv.CLS_NOFIELD
assert adv.classify_accum("") == adv.CLS_NOFIELD
assert adv.classify_accum(" ") == adv.CLS_NOFIELD
assert adv.classify_accum("横盘整理") == adv.CLS_UNKNOWN
assert adv.classify_accum("chip_concentration_up") == adv.CLS_UNKNOWN
# ================================================================
# [二] 网格参数生成 (方案附录二)
# ================================================================
@case("[网格] 支撑压力锚得住: 上界=压力×1.01 下界=支撑×0.99, 手数为整百")
def _():
gp, why = adv.grid_params(price=10.0, support=9.0, pressure=11.0, band=0.08,
step_pct=0.02, cap_room=70000.0, cap_ratio=0.5)
assert gp, why
assert gp["lower"] == 8.91 and gp["upper"] == 11.11 and gp["center"] == 10.0, gp
assert gp["max_capital"] == 35000.0, gp
assert gp["per_lot"] % 100 == 0 and gp["per_lot"] >= 100, gp
# 下方档数 5 档 (1.09 / 0.2), 每档预算 7000 → 700 股
assert gp["per_lot"] == 700, gp
@case("[网格] 锚不住 (支撑压力缺失/不合形) 退百分比带")
def _():
gp, _ = adv.grid_params(price=10.0, support=0, pressure=0, band=0.08,
step_pct=0.02, cap_room=70000.0, cap_ratio=0.5)
assert gp and gp["lower"] == 9.2 and gp["upper"] == 10.8, gp
# 支撑在现价上方 (不合形) 同样退带宽, 不硬凑
gp2, _ = adv.grid_params(price=10.0, support=12.0, pressure=11.0, band=0.08,
step_pct=0.02, cap_room=70000.0, cap_ratio=0.5)
assert gp2 and gp2["lower"] == 9.2 and gp2["upper"] == 11.11, gp2
@case("[网格] 三条放弃路径: 无实时价 / 买不起一手 / 区间不成立")
def _():
gp, why = adv.grid_params(price=None, support=9, pressure=11, band=0.08,
step_pct=0.02, cap_room=70000, cap_ratio=0.5)
assert gp is None and "实时价" in why, why
gp, why = adv.grid_params(price=10.0, support=9, pressure=11, band=0.08,
step_pct=0.02, cap_room=1500.0, cap_ratio=0.5)
assert gp is None and "买不起一手" in why, why # 750 元 < 一手 1000 元
gp, why = adv.grid_params(price=10.0, support=0, pressure=0, band=1.5,
step_pct=0.02, cap_room=70000, cap_ratio=0.5)
assert gp is None and "区间不成立" in why, why # 带宽>1 → 下界为负 (防御路径)
@case("[网格] 档距下限 0.5%: 传入更小的步长被抬起来")
def _():
gp, _ = adv.grid_params(price=10.0, support=9.0, pressure=11.0, band=0.08,
step_pct=0.001, cap_room=70000.0, cap_ratio=0.5)
assert gp and gp["step_pct"] == 0.005, gp
添加交易逻辑
2026-08-25 13:57:16 +08:00
@case("[网格] 科创板一手=200: 买得起判断与 per_lot 下限都按 200 (dry-run 实盘发现)")
def _():
assert adv.lot_of("688802.SH") == 200 and adv.lot_of("689009.SH") == 200
assert adv.lot_of("600000.SH") == 100 and adv.lot_of("300750.SZ") == 100
# 预算够 100 股不够 200 股: 主板挂得出, 科创板必须放弃并把 200 写进理由
gp, why = adv.grid_params(price=10.0, support=9.0, pressure=11.0, band=0.08,
step_pct=0.02, cap_room=3000.0, cap_ratio=0.5, lot=200)
assert gp is None and "一手(200股)" in why, (gp, why)
gp, _ = adv.grid_params(price=10.0, support=9.0, pressure=11.0, band=0.08,
step_pct=0.02, cap_room=3000.0, cap_ratio=0.5, lot=100)
assert gp and gp["per_lot"] == 100, gp
# 科创板预算充足但每档折出来不足 200 → 抬到 200 (仍是 100 的整数倍, runner 不会磨掉)
gp, _ = adv.grid_params(price=10.0, support=9.0, pressure=11.0, band=0.08,
step_pct=0.02, cap_room=16000.0, cap_ratio=0.5, lot=200)
assert gp and gp["per_lot"] == 200 and gp["per_lot"] % 100 == 0, gp
添加交易逻辑
2026-08-25 13:45:51 +08:00
# ================================================================
# [三] 连边矩阵 (plan_edge)
# ================================================================
@case("[连边] 双命中 (明确吸筹+高热+正垫) 取止盈 —— 保利润优先于做波段")
def _():
e, why = adv.plan_edge(cls=adv.CLS_CLEAR, fresh=True, heat=0.9, cushion=0.05,
prm=_prm())
assert e == adv.R_TRAIL and "双命中" in why, (e, why)
@case("[连边] 单命中各走各边: 高热正垫→止盈; 明确吸筹新鲜→网格")
def _():
e, why = adv.plan_edge(cls=adv.CLS_NOFIELD, fresh=False, heat=0.85,
cushion=0.02, prm=_prm())
assert e == adv.R_TRAIL and "超阈值" in why, (e, why)
e, why = adv.plan_edge(cls=adv.CLS_CLEAR, fresh=True, heat=0.3, cushion=-0.02,
prm=_prm())
assert e == adv.R_GRID, (e, why)
@case("[连边] 负垫永不挂止盈 (把位置留给深亏补仓评估) —— 小口径钉死")
def _():
e, why = adv.plan_edge(cls=adv.CLS_NOFIELD, fresh=False, heat=0.95,
cushion=-0.08, prm=_prm())
assert e is None and "补仓评估" in why, (e, why)
# 垫子缺失 (盘前无价) 同样不挂
e, _ = adv.plan_edge(cls=adv.CLS_NOFIELD, fresh=False, heat=0.95,
cushion=None, prm=_prm())
assert e is None
@case("[连边] 吸筹结论超日龄视为无标志; 潜在吸筹不触发 (拍板①只认明确)")
def _():
e, why = adv.plan_edge(cls=adv.CLS_CLEAR, fresh=False, heat=None, cushion=0.05,
prm=_prm())
assert e is None and "超日龄" in why, (e, why)
e, _ = adv.plan_edge(cls=adv.CLS_MAYBE, fresh=True, heat=None, cushion=0.05,
prm=_prm())
assert e is None
@case("[连边] 边清单裁剪生效: 去掉 heat_trail 后双命中落回网格")
def _():
p = _prm(rules={adv.R_GRID, adv.R_EXIT})
e, _ = adv.plan_edge(cls=adv.CLS_CLEAR, fresh=True, heat=0.9, cushion=0.05, prm=p)
assert e == adv.R_GRID, e
p = _prm(rules={adv.R_TRAIL})
e, _ = adv.plan_edge(cls=adv.CLS_CLEAR, fresh=True, heat=0.3, cushion=0.05, prm=p)
assert e is None, e
# ================================================================
# [四] note 约定与无状态推导 (日上限 / 两种冷却)
# ================================================================
@case("[约定] note 字面量钉死 —— 冷却与计数全靠它, 改一个字就坏")
def _():
assert adv.NOTE_AUTO == "自动挂载: "
assert adv.NOTE_HANDOFF == "自动挂载(接力): "
assert adv.MARK_HANDOFF_OUT == "[接力撤下]"
assert adv.NOTE_HANDOFF.startswith("自动挂载") # 冷却推导认「自动挂载」开头
@case("[计数] 今日新挂只数常规挂计入、接力与人工与昨日不计入")
def _():
rows = [
{"note": "自动挂载: 明确吸筹→网格", "created_at": "2026-08-25 09:40:01"},
{"note": "自动挂载: 高热→止盈", "created_at": "2026-08-25 09:40:02"},
{"note": "自动挂载(接力): 站上上界", "created_at": "2026-08-25 09:41:00"}, # 接力不算
{"note": "手工挂的网格", "created_at": "2026-08-25 10:00:00"}, # 人工不算
{"note": "自动挂载: 昨天挂的", "created_at": "2026-08-24 09:40:00"}, # 昨日不算
{"note": None, "created_at": "2026-08-25 09:40:00"},
]
assert adv.count_auto_today(rows, TODAY) == 2
@case("[冷却] 人工撤下→同票同规则进冷却; 接力标记→只进接力冷却不算人工")
def _():
rows = [
# 人工撤下的自动网格, 票还持有, 5 天前 → optout
{"ts_code": "600000.SH", "type": "GRID", "note": "自动挂载: 网格",
"updated_at": "2026-08-20 10:00:00"},
# 接力撤下的网格 → 只进 handoff_cool
{"ts_code": "600519.SH", "type": "GRID", "note": "自动挂载: 网格 [接力撤下]",
"updated_at": "2026-08-20 11:00:00"},
]
opt, hand = adv.cooldowns_from_cancelled(rows, ["600000.SH", "600519.SH"], TODAY,
optout_tdays=10, handoff_tdays=10)
assert ("600000.SH", adv.R_GRID) in opt, opt
assert "600519.SH" not in {c for c, _ in opt}, opt
assert hand == {"600519.SH"}, hand
@case("[冷却] 票已不持有不算人工撤下 (清仓清场撤的不该罚) / 超窗口出冷却 / 人工策略不掺和")
def _():
rows = [
{"ts_code": "600000.SH", "type": "GRID", "note": "自动挂载: 网格",
"updated_at": "2026-08-20 10:00:00"}, # 不在持仓 → 不算
{"ts_code": "600519.SH", "type": "TRAIL", "note": "自动挂载: 止盈",
"updated_at": "2026-07-10 10:00:00"}, # 46 天 > 10td×2 → 出冷却
{"ts_code": "000001.SZ", "type": "GRID", "note": "手工网格",
"updated_at": "2026-08-24 10:00:00"}, # 人工挂的, 撤了也不进冷却
]
opt, hand = adv.cooldowns_from_cancelled(rows, ["600519.SH", "000001.SZ"], TODAY,
optout_tdays=10, handoff_tdays=10)
assert opt == set() and hand == set(), (opt, hand)
# ================================================================
# [五] 接力判定 (handoff_ready)
# ================================================================
@case("[接力] 齐活才走: 站上上界 + 热度达标 + 正垫 → True")
def _():
ok, why = adv.handoff_ready(price=10.6, price_ok=True, upper=10.5, heat=0.85,
cushion=0.06, prm=_prm())
assert ok and "站上网格上界" in why, (ok, why)
@case("[接力] 没站上上界静默不动; 站上了但热度不够/垫不正要说清")
def _():
ok, why = adv.handoff_ready(price=10.2, price_ok=True, upper=10.5, heat=0.9,
cushion=0.05, prm=_prm())
assert not ok and why == "", (ok, why)
ok, why = adv.handoff_ready(price=10.6, price_ok=True, upper=10.5, heat=0.5,
cushion=0.05, prm=_prm())
assert not ok and "未达阈值" in why, why
ok, why = adv.handoff_ready(price=10.6, price_ok=True, upper=10.5, heat=0.9,
cushion=-0.01, prm=_prm())
assert not ok and "不接力" in why, why
@case("[接力] 独立开关与边清单双闸; 价格拿成本顶的 (price_ok=False) 不判")
def _():
ok, _ = adv.handoff_ready(price=10.6, price_ok=True, upper=10.5, heat=0.9,
cushion=0.05, prm=_prm(handoff_enabled=False))
assert not ok
ok, _ = adv.handoff_ready(price=10.6, price_ok=True, upper=10.5, heat=0.9,
cushion=0.05,
prm=_prm(rules={adv.R_GRID, adv.R_TRAIL, adv.R_EXIT}))
assert not ok
ok, _ = adv.handoff_ready(price=10.6, price_ok=False, upper=10.5, heat=0.9,
cushion=0.05, prm=_prm())
assert not ok
# ================================================================
# [六] clear_buypause: 只清自己停的
# ================================================================
def _with_buypause(entry, fn):
import json as _json
saved = {}
orig_get, orig_set = pms_repo.get_param, pms_repo.set_param
pms_repo.get_param = lambda k: _json.dumps(entry) if entry is not None else None
pms_repo.set_param = lambda k, v, by="system": saved.update({"key": k, "val": v}) or {"ok": True}
try:
r = fn()
finally:
pms_repo.get_param, pms_repo.set_param = orig_get, orig_set
return r, saved
@case("[停买] clear_buypause 来源匹配才清: accum 停的清得掉, 风控停的不动")
def _():
r, saved = _with_buypause({"600000.SH": {"source": "accum", "reason": "x"}},
lambda: svc.clear_buypause("600000.SH", only_source="accum"))
assert r == {"ok": True, "cleared": True, "ts_code": "600000.SH"}, r
assert saved and "600000.SH" not in saved["val"], saved
r, saved = _with_buypause({"600000.SH": {"source": "signal", "reason": "风控预警"}},
lambda: svc.clear_buypause("600000.SH", only_source="accum"))
assert r["ok"] and not r["cleared"] and "signal" in r.get("why", ""), r
assert not saved, saved # 没清 → 一个字都没写
@case("[停买] 没停过→ok 且 cleared=False; 空码→ok=False")
def _():
r, saved = _with_buypause({}, lambda: svc.clear_buypause("600000.SH",
only_source="accum"))
assert r["ok"] and not r["cleared"] and not saved, (r, saved)
r, _ = _with_buypause({}, lambda: svc.clear_buypause(""))
assert not r["ok"], r
# ================================================================
# [七] 参数登记 (param_store)
# ================================================================
@case("[参数] 总开关进 FAIL_CLOSED=False; 13 个键全有中文说明")
def _():
assert param_store.FAIL_CLOSED.get("PMS_AUTO_STRATEGY_ENABLED") is False
for k in ("PMS_AUTO_STRATEGY_ENABLED", "PMS_AUTO_STRATEGY_RULES",
"PMS_AUTO_STRATEGY_DAILY_MAX", "PMS_AUTO_ACCUM_STALE_TDAYS",
"PMS_AUTO_GRID_BAND", "PMS_AUTO_GRID_STEP_PCT", "PMS_AUTO_GRID_CAP_RATIO",
"PMS_AUTO_HEAT_TH", "PMS_AUTO_TRAIL_GIVEBACK", "PMS_AUTO_TRAIL_SELL_RATIO",
"PMS_AUTO_HANDOFF_ENABLED", "PMS_AUTO_HANDOFF_COOLDOWN_TDAYS",
"PMS_AUTO_OPTOUT_COOLDOWN_TDAYS"):
assert k in param_store.DESC, k
from config.settings import settings as _s
assert hasattr(_s, k), k
@case("[参数] 边清单写入口校验: 打错边名被拦, 合法值与越界数值各归各")
def _():
err = param_store._range_check("PMS_AUTO_STRATEGY_RULES", "accum_grid,typo_edge")
assert err and "typo_edge" in err, err
assert param_store._range_check("PMS_AUTO_STRATEGY_RULES",
"accum_grid,heat_trail,accum_exit,handoff") is None
assert param_store._range_check("PMS_AUTO_HEAT_TH", 1.5), "热度阈值 1.5 应越界"
assert param_store._range_check("PMS_AUTO_HEAT_TH", 0.8) is None
# ================================================================
# [八] 编排冒烟 (全打桩, 不连库)
# ================================================================
class _Rec:
"""记录调用的假 strategy_service / pms_repo 面板。"""
def __init__(self):
self.attach_calls, self.status_calls, self.pause_calls = [], [], []
self.clear_calls, self.ledger, self.upd_calls = [], [], []
def _scan_stubbed(rec, *, held, accum, heat, strategies=None, cancelled=None,
dry_run, prm=None, attach_ret=None):
"""把 scan() 的所有外部依赖打桩后跑一轮, 返回 out。"""
from app.services import command_service, portfolio, strategy_service
orig = {}
def keep(mod, name, fake):
orig[(mod, name)] = getattr(mod, name)
setattr(mod, name, fake)
view = {"held": held, "positions": held,
"params": {"scale": 1000000.0, "stock_cap": 0.08}}
n_attach = {"n": 0}
def fake_attach(cfg, by="user"):
rec.attach_calls.append((cfg, by))
n_attach["n"] += 1
return dict(attach_ret or {"ok": True, "strategy_id": f"S_NEW{n_attach['n']}"})
def fake_list_strategies(**kw):
sts = kw.get("statuses") or []
if "CANCELLED" in sts:
return list(cancelled or [])
return list(strategies or [])
keep(adv, "_params", lambda: prm or _prm())
keep(param_store, "get_bool",
lambda k, d=False: {"PMS_STRATEGY_ENABLED": True,
"PMS_GLOBAL_EXEC_HALT": False}.get(k, d))
keep(portfolio, "positions_view", lambda: view)
keep(adv, "accum_of", lambda codes: dict(accum))
keep(adv, "heat_of", lambda codes: (dict(heat), {"stale": False, "td": TODAY}))
keep(command_service, "blacklist", lambda: set())
keep(pms_repo, "list_strategies", fake_list_strategies)
keep(pms_repo, "list_instructions", lambda **kw: [])
keep(pms_repo, "list_plans", lambda **kw: [])
keep(pms_repo, "insert_ledger",
lambda **kw: rec.ledger.append(kw) or 1)
keep(pms_repo, "update_strategy",
lambda sid, **kw: rec.upd_calls.append((sid, kw)) or 1)
keep(strategy_service, "buypause_map", lambda: {})
keep(strategy_service, "attach", fake_attach)
keep(strategy_service, "set_status",
lambda sid, st, by="user": rec.status_calls.append((sid, st, by)) or
{"ok": True, "strategy_id": sid})
keep(strategy_service, "pause_buy",
lambda code, reason="", source="signal":
rec.pause_calls.append((code, source)) or ["SID"])
keep(strategy_service, "clear_buypause",
lambda code, only_source=None:
rec.clear_calls.append((code, only_source)) or {"ok": True, "cleared": True})
try:
return adv.scan(dry_run=dry_run)
finally:
for (mod, name), fn in orig.items():
setattr(mod, name, fn)
@case("[冒烟] dry_run 滴水不写: 报出会挂网格, 但 attach/留痕/停买全没动")
def _():
rec = _Rec()
out = _scan_stubbed(rec, held=[_pos()], heat={},
accum={"600000.SH": {"cls": adv.CLS_CLEAR, "state": "明确吸筹",
"score": 80, "ymd": TODAY, "age_days": 1}},
dry_run=True)
assert out["ok"], out["errors"]
assert out["attached"] and out["attached"][0]["dry_run"], out["attached"]
assert out["attached"][0]["ts_code"] == "600000.SH"
assert not rec.attach_calls and not rec.ledger and not rec.pause_calls, "dry_run 写了东西"
@case("[冒烟] 真挂网格: attach 收到区间与上限, note 以「自动挂载: 」开头, 留痕 ATTACH/PASS")
def _():
rec = _Rec()
out = _scan_stubbed(rec, held=[_pos()], heat={},
accum={"600000.SH": {"cls": adv.CLS_CLEAR, "state": "明确吸筹",
"score": 80, "ymd": TODAY, "age_days": 1}},
dry_run=False)
assert out["ok"] and out["attached"], (out["errors"], out["blocked"], out["skipped"])
cfg, by = rec.attach_calls[0]
assert by == "auto" and cfg["type"] == "GRID" and cfg["autonomy"] == "auto"
assert cfg["note"].startswith(adv.NOTE_AUTO), cfg["note"]
assert 0 < cfg["params"]["lower"] < 10.0 < cfg["params"]["upper"], cfg["params"]
assert cfg["params"]["max_capital"] == 35000.0, cfg["params"]
led = [x for x in rec.ledger if x.get("action") == "ATTACH"]
assert led and led[0]["verdict"] == "PASS" and led[0]["ref_id"] == "S_NEW1", led
@case("[冒烟] 每日名额: 第三只被挡并留痕说明, dry_run 同样受限")
def _():
held = [_pos(ts_code=c) for c in ("600000.SH", "600519.SH", "000001.SZ")]
accum = {c: {"cls": adv.CLS_CLEAR, "state": "明确吸筹", "score": 80,
"ymd": TODAY, "age_days": 1} for c in ("600000.SH", "600519.SH",
"000001.SZ")}
rec = _Rec()
out = _scan_stubbed(rec, held=held, accum=accum, heat={}, dry_run=True)
assert len(out["attached"]) == 2, out
assert len(out["blocked"]) == 1 and "名额已满" in out["blocked"][0]["why"], out["blocked"]
rec2 = _Rec()
out2 = _scan_stubbed(rec2, held=held, accum=accum, heat={}, dry_run=False)
assert len(rec2.attach_calls) == 2, rec2.attach_calls
assert any("名额已满" in x.get("reason", "") for x in rec2.ledger), rec2.ledger
@case("[冒烟] 排除项: 冻结票与热度停更都不挂; 词表外定性浮到 unknown_states")
def _():
held = [_pos(frozen_reason="RISK_FREEZE")]
accum = {"600000.SH": {"cls": adv.CLS_CLEAR, "state": "明确吸筹", "score": 80,
"ymd": TODAY, "age_days": 1}}
rec = _Rec()
out = _scan_stubbed(rec, held=held, accum=accum, heat={}, dry_run=False)
assert not out["attached"] and not rec.attach_calls
assert any("冻结" in s.get("why", "") for s in out["skipped"]), out["skipped"]
# 词表外
rec2 = _Rec()
out2 = _scan_stubbed(rec2, held=[_pos()], heat={},
accum={"600000.SH": {"cls": adv.CLS_UNKNOWN, "state": "横盘整理",
"score": None, "ymd": TODAY, "age_days": 1}},
dry_run=False)
assert out2["unknown_states"] and out2["unknown_states"][0]["state"] == "横盘整理"
assert not rec2.attach_calls
@case("[冒烟] 边三: 已挂自动网格遇派发 → pause_buy(source=accum) 一次, 已停不重复")
def _():
st = {"strategy_id": "S_G", "ts_code": "600000.SH", "type": "GRID",
"note": "自动挂载: 明确吸筹→网格", "params": {"upper": 10.5}, "state": {}}
rec = _Rec()
fake_svc = SimpleNamespace(
pause_buy=lambda code, reason="", source="signal":
rec.pause_calls.append((code, source)) or ["S_G"],
clear_buypause=lambda code, only_source=None:
rec.clear_calls.append((code, only_source)) or {"ok": True, "cleared": True})
orig_led = pms_repo.insert_ledger
pms_repo.insert_ledger = lambda **kw: rec.ledger.append(kw) or 1
try:
out = _out()
adv._tend_existing(st, _pos(price=10.0), {"cls": adv.CLS_DISTRIB, "state": "高位派发"},
True, 0.3, _prm(), {}, set(), TODAY, False, out, fake_svc)
assert rec.pause_calls == [("600000.SH", "accum")], rec.pause_calls
assert out["paused"] and any(x.get("action") == "NOTE" for x in rec.ledger)
# 已停着 (无论谁停的) 不重复停
out2 = _out()
adv._tend_existing(st, _pos(price=10.0), {"cls": adv.CLS_DISTRIB}, True, 0.3,
_prm(), {"600000.SH": {"source": "signal"}}, set(), TODAY,
False, out2, fake_svc)
assert len(rec.pause_calls) == 1 and not out2["paused"]
finally:
pms_repo.insert_ledger = orig_led
@case("[冒烟] 边三解除: 定性回明确且新鲜, 只解除 accum 来源的停买")
def _():
st = {"strategy_id": "S_G", "ts_code": "600000.SH", "type": "GRID",
"note": "自动挂载: 网格", "params": {"upper": 10.5}, "state": {}}
rec = _Rec()
fake_svc = SimpleNamespace(
pause_buy=lambda *a, **k: [],
clear_buypause=lambda code, only_source=None:
rec.clear_calls.append((code, only_source)) or {"ok": True, "cleared": True})
orig_led = pms_repo.insert_ledger
pms_repo.insert_ledger = lambda **kw: rec.ledger.append(kw) or 1
try:
out = _out()
adv._tend_existing(st, _pos(price=10.0),
{"cls": adv.CLS_CLEAR, "state": "明确吸筹"}, True, 0.3,
_prm(), {"600000.SH": {"source": "accum"}}, set(), TODAY,
False, out, fake_svc)
assert rec.clear_calls == [("600000.SH", "accum")], rec.clear_calls
assert out["resumed"] == [{"ts_code": "600000.SH"}], out["resumed"]
# 风控停的不归边三管
out2 = _out()
adv._tend_existing(st, _pos(price=10.0), {"cls": adv.CLS_CLEAR}, True, 0.3,
_prm(), {"600000.SH": {"source": "signal"}}, set(), TODAY,
False, out2, fake_svc)
assert len(rec.clear_calls) == 1 and not out2["resumed"]
finally:
pms_repo.insert_ledger = orig_led
@case("[冒烟] 边四全链: 撤网格→老 note 打[接力撤下]→挂止盈(接力 note)→高水位从现价起算")
def _():
st = {"strategy_id": "S_G", "ts_code": "600000.SH", "type": "GRID",
"note": "自动挂载: 网格", "params": {"upper": 10.5}, "state": {}}
rec = _Rec()
fake_svc = SimpleNamespace(
set_status=lambda sid, s, by="user": rec.status_calls.append((sid, s, by)) or
{"ok": True},
attach=lambda cfg, by="user": rec.attach_calls.append((cfg, by)) or
{"ok": True, "strategy_id": "S_T"},
pause_buy=lambda *a, **k: [], clear_buypause=lambda *a, **k: {"ok": True})
orig_led, orig_upd = pms_repo.insert_ledger, pms_repo.update_strategy
pms_repo.insert_ledger = lambda **kw: rec.ledger.append(kw) or 1
pms_repo.update_strategy = lambda sid, **kw: rec.upd_calls.append((sid, kw)) or 1
try:
out = _out()
adv._tend_existing(st, _pos(price=10.6), {"cls": adv.CLS_CLEAR}, True, 0.9,
_prm(), {}, set(), TODAY, False, out, fake_svc)
assert rec.status_calls == [("S_G", "CANCELLED", "auto")], rec.status_calls
marks = [kw for sid, kw in rec.upd_calls if sid == "S_G"]
assert marks and adv.MARK_HANDOFF_OUT in marks[0]["note"], rec.upd_calls
cfg, by = rec.attach_calls[0]
assert cfg["type"] == "TRAIL" and cfg["note"].startswith(adv.NOTE_HANDOFF)
hw = [kw for sid, kw in rec.upd_calls if sid == "S_T"]
assert hw and hw[0]["state"] == {"high_water": 10.6}, rec.upd_calls
assert out["handoffs"] and out["handoffs"][0]["to"] == "S_T"
assert any(x.get("action") == "HANDOFF" and x.get("verdict") == "PASS"
for x in rec.ledger)
finally:
pms_repo.insert_ledger, pms_repo.update_strategy = orig_led, orig_upd
@case("[冒烟] 边四让路: 接力冷却期 / 有在途委托 / dry_run 都不真动")
def _():
st = {"strategy_id": "S_G", "ts_code": "600000.SH", "type": "GRID",
"note": "自动挂载: 网格", "params": {"upper": 10.5}, "state": {}}
rec = _Rec()
fake_svc = SimpleNamespace(
set_status=lambda *a, **k: rec.status_calls.append(a) or {"ok": True},
attach=lambda *a, **k: rec.attach_calls.append(a) or {"ok": True,
"strategy_id": "X"},
pause_buy=lambda *a, **k: [], clear_buypause=lambda *a, **k: {"ok": True})
# 冷却期
out = _out()
adv._tend_existing(st, _pos(price=10.6), {"cls": adv.CLS_CLEAR}, True, 0.9,
_prm(), {}, {"600000.SH"}, TODAY, False, out, fake_svc)
assert not rec.status_calls and any("冷却" in s["why"] for s in out["skipped"])
# 在途委托
st2 = dict(st, state={"pending": {"iid": "INS_1"}})
orig_get = pms_repo.get_instruction
pms_repo.get_instruction = lambda iid: {"instruction_id": iid, "status": "DISPATCHED"}
try:
out2 = _out()
adv._tend_existing(st2, _pos(price=10.6), {"cls": adv.CLS_CLEAR}, True, 0.9,
_prm(), {}, set(), TODAY, False, out2, fake_svc)
assert not rec.status_calls and any("在途" in s["why"] for s in out2["skipped"])
finally:
pms_repo.get_instruction = orig_get
# dry_run 只报不动
out3 = _out()
adv._tend_existing(st, _pos(price=10.6), {"cls": adv.CLS_CLEAR}, True, 0.9,
_prm(), {}, set(), TODAY, True, out3, fake_svc)
assert out3["handoffs"] and out3["handoffs"][0]["dry_run"] and not rec.status_calls
@case("[冒烟] 人工策略不碰; 自动止盈无事; 接力半途而废要报错并留痕 REJECT")
def _():
rec = _Rec()
fake_svc = SimpleNamespace(
set_status=lambda sid, s, by="user": {"ok": True},
attach=lambda cfg, by="user": {"ok": False, "errors": ["跟踪止盈触发时要卖出, 但当前 T+1 可卖为 0"]},
pause_buy=lambda *a, **k: [], clear_buypause=lambda *a, **k: {"ok": True})
manual = {"strategy_id": "S_M", "ts_code": "600000.SH", "type": "GRID",
"note": "手工网格", "params": {"upper": 10.5}, "state": {}}
out = _out()
adv._tend_existing(manual, _pos(price=10.6), {"cls": adv.CLS_CLEAR}, True, 0.9,
_prm(), {}, set(), TODAY, False, out, fake_svc)
assert any("人工策略" in s["why"] for s in out["skipped"]), out["skipped"]
auto_trail = dict(manual, note="自动挂载: 止盈", type="TRAIL")
out2 = _out()
adv._tend_existing(auto_trail, _pos(price=10.6), {"cls": adv.CLS_CLEAR}, True, 0.9,
_prm(), {}, set(), TODAY, False, out2, fake_svc)
assert any("本轮无事" in s["why"] for s in out2["skipped"]), out2["skipped"]
# 半途而废: 撤成挂败 → errors 里有「人工处理」, 留痕 HANDOFF/REJECT
orig_led, orig_upd = pms_repo.insert_ledger, pms_repo.update_strategy
pms_repo.insert_ledger = lambda **kw: rec.ledger.append(kw) or 1
pms_repo.update_strategy = lambda sid, **kw: 1
try:
grid = dict(manual, note="自动挂载: 网格")
out3 = _out()
adv._tend_existing(grid, _pos(price=10.6), {"cls": adv.CLS_CLEAR}, True, 0.9,
_prm(), {}, set(), TODAY, False, out3, fake_svc)
assert any("人工处理" in e for e in out3["errors"]), out3["errors"]
assert any(x.get("action") == "HANDOFF" and x.get("verdict") == "REJECT"
for x in rec.ledger), rec.ledger
finally:
pms_repo.insert_ledger, pms_repo.update_strategy = orig_led, orig_upd
@case("[冒烟] 三道总闸: 自动开关关 / 策略层关 / 休假模式, 各自明说并整轮不动")
def _():
rec = _Rec()
out = _scan_stubbed(rec, held=[_pos()], accum={}, heat={}, dry_run=False,
prm=_prm(enabled=False))
assert not rec.attach_calls and any("总开关关闭" in s["why"] for s in out["skipped"])
orig = param_store.get_bool
param_store.get_bool = lambda k, d=False: {"PMS_STRATEGY_ENABLED": False,
"PMS_GLOBAL_EXEC_HALT": False}.get(k, d)
orig_p = adv._params
adv._params = lambda: _prm()
try:
out2 = adv.scan(dry_run=False)
assert any("策略层总开关" in s["why"] for s in out2["skipped"]), out2
param_store.get_bool = lambda k, d=False: {"PMS_STRATEGY_ENABLED": True,
"PMS_GLOBAL_EXEC_HALT": True}.get(k, d)
out3 = adv.scan(dry_run=False)
assert any("休假" in s["why"] for s in out3["skipped"]), out3
finally:
param_store.get_bool, adv._params = orig, orig_p
添加交易逻辑
2026-08-25 13:57:16 +08:00
@case("[判分] report_strategy_score: 成交聚合按边分侧 / 对照只收名额挡下的 / 涨跌口径空值安全")
def _():
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import report_strategy_score as rep
orig = rep.fetch_all
try:
rep.fetch_all = lambda sql, p=None: [
{"side": "buy", "exec_qty": 300, "exec_avg_price": 10.0},
{"side": "buy", "exec_qty": 100, "exec_avg_price": 11.0},
{"side": "sell", "exec_qty": 200, "exec_avg_price": 10.8},
{"side": "weird", "exec_qty": 999, "exec_avg_price": 1.0}, # 未知边丢弃
]
agg = rep.fills_of("S_X")
assert agg["buy"] == {"qty": 400, "amt": 4100.0, "n": 2}, agg
assert agg["sell"]["qty"] == 200 and abs(agg["sell"]["amt"] - 2160.0) < 1e-6, agg
rep.fetch_all = lambda sql, p=None: [
{"ts_code": "A", "verdict": "PASS", "price_at": 10, "reason": "挂了"},
{"ts_code": "B", "verdict": "NOTE", "price_at": 9,
"reason": "想挂 accum_grid 但今日新挂名额已满, 留到明天"},
{"ts_code": "C", "verdict": "NOTE", "price_at": 8,
"reason": "想挂网格但放弃: 买不起一手"}, # 排除项, 不是对照
]
att, ctl = rep.ledger_window(30)
assert [r["ts_code"] for r in att] == ["A"], att
assert [r["ts_code"] for r in ctl] == ["B"], ctl
finally:
rep.fetch_all = orig
assert rep._pct(10.0, 11.0) is not None and abs(rep._pct(10.0, 11.0) - 0.1) < 1e-9
assert rep._pct(None, 11.0) is None and rep._pct(10.0, None) is None
assert rep._fmt_pct(None) == "—" and rep._fmt_pct(0.1) == "+10.00%"
添加交易逻辑
2026-08-25 13:45:51 +08:00
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()