tradingSystem/scripts/test_batch17_units.py

# -*- 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


@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()