858 lines
42 KiB
Python
858 lines
42 KiB
Python
# -*- 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
|
||
|
||
|
||
@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
|
||
|
||
|
||
# ================================================================
|
||
# [三] 连边矩阵 (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 = [], [], []
|
||
self.param_writes = []
|
||
|
||
|
||
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(param_store, "set_param",
|
||
lambda k, v, by="user": rec.param_writes.append((k, v)) or {"ok": True})
|
||
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
|
||
# 2026-08-28 补: 真跑那半段的返回值也要锁住 (原来 out2 算了没断言)
|
||
assert out2["ok"], out2
|
||
assert len(out2["blocked"]) == 1 and "名额已满" in out2["blocked"][0]["why"], out2["blocked"]
|
||
|
||
|
||
@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
|
||
|
||
|
||
@case("[仪表] 真跑每个出口都写一跳摘要 (含总开关关着那次), 预演一笔不写")
|
||
def _():
|
||
import json as _json
|
||
assert "PMS_AUTO_LAST_SCAN" in param_store.RUNTIME_EXTRA # 没登记 set_param 会拒写
|
||
rec = _Rec()
|
||
_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 len(rec.param_writes) == 1, rec.param_writes
|
||
k, v = rec.param_writes[0]
|
||
summ = _json.loads(v)
|
||
assert k == "PMS_AUTO_LAST_SCAN" and summ["checked"] == 1
|
||
assert summ["counts"]["attached"] == 1 and "ymd" in summ and "at" in summ
|
||
# 预演滴水不写
|
||
rec2 = _Rec()
|
||
_scan_stubbed(rec2, held=[_pos()], accum={}, heat={}, dry_run=True)
|
||
assert not rec2.param_writes
|
||
# 总开关关着的真跑也要留一笔 —— 不然「跑了但关着」和「根本没跑」分不清
|
||
rec3 = _Rec()
|
||
_scan_stubbed(rec3, held=[_pos()], accum={}, heat={}, dry_run=False,
|
||
prm=_prm(enabled=False))
|
||
assert len(rec3.param_writes) == 1 and _json.loads(rec3.param_writes[0][1])["enabled"] is False
|
||
|
||
|
||
@case("[阶段] stage_of 五阶段判定矩阵: 策略优先于信号, 派发优先于深亏, 都不是归中性")
|
||
def _():
|
||
def s(**kw):
|
||
d = dict(cls=None, fresh=False, heat=None, heat_th=0.80, cushion=None,
|
||
stype=None, is_auto=False)
|
||
d.update(kw)
|
||
return adv.stage_of(**d)[0]
|
||
# 已挂自动策略: 阶段以策略为准, 信号什么样都不改判
|
||
assert s(stype="GRID", is_auto=True, cls=adv.CLS_DISTRIB) == adv.STG_ACCUM
|
||
assert s(stype="TRAIL", is_auto=True, cushion=-0.1) == adv.STG_LAUNCH
|
||
# 人工策略不算数 (is_auto=False), 按信号判
|
||
assert s(stype="GRID", is_auto=False, cls=adv.CLS_DISTRIB) == adv.STG_DISTRIB
|
||
# 信号判定次序: 派发 > 吸筹 > 拉升 > 深亏 > 中性
|
||
assert s(cls=adv.CLS_DISTRIB, cushion=-0.2) == adv.STG_DISTRIB
|
||
assert s(cls=adv.CLS_CLEAR, fresh=True, cushion=-0.05) == adv.STG_ACCUM
|
||
assert s(cls=adv.CLS_CLEAR, fresh=False, cushion=-0.05) == adv.STG_REPAIR
|
||
assert s(heat=0.9, cushion=0.03) == adv.STG_LAUNCH
|
||
assert s(heat=0.9, cushion=-0.03) == adv.STG_REPAIR # 热但亏, 不算拉升
|
||
assert s(heat=0.5, cushion=0.03) == adv.STG_NEUTRAL
|
||
assert s() == adv.STG_NEUTRAL
|
||
|
||
|
||
@case("[阶段] 扫描给每只持仓各出一行阶段, 人工策略票注明不碰")
|
||
def _():
|
||
held = [_pos(ts_code="600000.SH"), # 明确吸筹 → 吸筹震荡
|
||
_pos(ts_code="600519.SH", cushion_pct=-0.06), # 深亏修复
|
||
_pos(ts_code="000001.SZ"), # 无信号 → 中性
|
||
_pos(ts_code="002415.SZ")] # 人工策略
|
||
accum = {"600000.SH": {"cls": adv.CLS_CLEAR, "state": "明确吸筹", "score": 80,
|
||
"ymd": TODAY, "age_days": 1}}
|
||
manual = [{"strategy_id": "S_M", "ts_code": "002415.SZ", "type": "T0",
|
||
"note": "手工做T", "params": {}, "state": {}}]
|
||
rec = _Rec()
|
||
out = _scan_stubbed(rec, held=held, accum=accum, heat={}, strategies=manual,
|
||
dry_run=True)
|
||
stg = {x["ts_code"]: x for x in out["stages"]}
|
||
assert len(out["stages"]) == 4, out["stages"]
|
||
assert stg["600000.SH"]["stage"] == adv.STG_ACCUM
|
||
assert stg["600519.SH"]["stage"] == adv.STG_REPAIR
|
||
assert stg["000001.SZ"]["stage"] == adv.STG_NEUTRAL
|
||
assert "人工策略" in stg["002415.SZ"]["why"], stg["002415.SZ"]
|
||
|
||
|
||
@case("[科创板] runner 止盈卖出: 200 股起, 不足 200 只许一次性全清, 主板行为一字不变")
|
||
def _():
|
||
from app.services import strategy_runner as srun
|
||
assert srun._min_lot("688802.SH") == 200 and srun._min_lot("600000.SH") == 100
|
||
|
||
def trail(code, avail, ratio, avg=8.0, price=10.0):
|
||
st = {"ts_code": code, "params": {"giveback": 0.05, "sell_ratio": ratio}}
|
||
pos = {"avg_cost": avg, "avail_qty": avail, "cushion_pct": price / avg - 1}
|
||
ctx = {"state": {"armed": True, "high_water": 12.0}, "notes": []}
|
||
return srun._eval_trail(st, pos, {"price": price}, None, ctx)
|
||
|
||
d = trail("688802.SH", 150, 1.0) # 全清: 不足 200 按例外一次性全卖
|
||
assert d and d["qty"] == 150 and d["action"] == srun.A_EXIT, d
|
||
d = trail("688802.SH", 700, 0.25) # 部分卖 175→整百 100 不合法 → 抬到 200
|
||
assert d and d["qty"] == 200, d
|
||
assert trail("688802.SH", 150, 0.5) is None # 可卖不足 200, 部分卖放弃
|
||
d = trail("600000.SH", 700, 0.25) # 主板照旧 100
|
||
assert d and d["qty"] == 100, d
|
||
d = trail("600000.SH", 150, 1.0) # 主板全清照旧整百
|
||
assert d and d["qty"] == 100, d
|
||
|
||
|
||
@case("[科创板] runner 网格: 手工 100 股档抬到 200 / 卖出量不足 200 不推进档位")
|
||
def _():
|
||
from app.services import strategy_runner as srun
|
||
prm = {"lower": 9.0, "upper": 11.0, "center": 10.0, "step_pct": 0.02,
|
||
"per_lot": 100, "max_capital": 50000}
|
||
|
||
def grid(code, price, avail, state, add_qty=None):
|
||
st = {"ts_code": code, "params": prm}
|
||
pos = {"avail_qty": avail, "add_qty": avail if add_qty is None else add_qty}
|
||
ctx = {"state": state, "notes": [], "buy_paused": False}
|
||
return srun._eval_grid(st, pos, {"price": price}, None, ctx)
|
||
|
||
lv = srun._grid_levels(prm)
|
||
# 下行跌破一档 → 买: 科创板把 100 股/档抬成 200
|
||
b0 = srun._band(lv, 9.5)
|
||
d = grid("688802.SH", 9.5, 0, {"last_band": b0 + 1, "filled_levels": {}})
|
||
assert d and d["side"] == "buy" and d["qty"] == 200, d
|
||
d = grid("600000.SH", 9.5, 0, {"last_band": b0 + 1, "filled_levels": {}})
|
||
assert d and d["qty"] == 100, d # 主板照旧
|
||
# 上行涨破 → 卖: 网格股有 200 但今天可卖只有 100, 科创板卖 100 不合法, 原地等
|
||
b1 = srun._band(lv, 10.5)
|
||
st1 = {"last_band": b1 - 1,
|
||
"filled_levels": {str(b1 - 1): {"qty": 200, "price": 9.8}}}
|
||
assert grid("688802.SH", 10.5, 100, st1, add_qty=200) is None
|
||
assert st1.get("last_band") == b1 - 1, st1 # 档位没被推进, 下跳凑够了还能卖
|
||
|
||
|
||
@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%"
|
||
|
||
|
||
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()
|