tradingSystem/scripts/test_batch30_units.py

# -*- coding: utf-8 -*-
"""三源合议 · 工作包三离场纪律 (2026-09-11 方案第五节, 台账 010)。全部离线, 不连库。

入场靠投票, 离场靠纪律: SAR 转空像保垫减仓一样按规则执行, 不进入场那套合议投票。
  A eval_tech_exit: 相位转空才评/确认清仓·未确认减三分之一/翻向超期不评/同一翻空一次/
    无读数弃权/propose_only 交人/可卖量夹紧/不足一手不评/开关关掉逐字回旧。
  B 卖出优先级: 目标价 < 研究走弱 < 技术面转空 < 保垫减仓。
  C scan 端到端: 产出来源 tech_exit / 同轮与保垫减仓并现留转空 / 目标价优先于转空。
  D 策略票只看目标价 (按函数判, 不按动作名) —— 技术面转空对策略票不评。
  E 参数登记。
"""
import os
import sys
import traceback

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.core import action_engine as ae          # noqa: E402
from app.services import param_store as ps         # noqa: E402

RESULTS = []


def case(name):
    def deco(fn):
        RESULTS.append((name, fn))
        return fn
    return deco


def pos(**kw):
    p = {"ts_code": "600000.SH", "price": 10.0, "avg_cost": 10.0, "total_qty": 6000,
         "base_qty": 6000, "add_qty": 0, "dca_qty": 0, "market_value": 60_000,
         "cushion_pct": 0.0, "cushion_peak": 0.0, "target_pct": 0.06,
         "support_ref": None, "pressure_ref": None, "stop_ref": None,
         "fill_count": 0, "dca_count": 0, "frozen_reason": "NONE", "avail_qty": 6000}
    p.update(kw)
    return p


def tech(phase="转空", flip=1, confirm=False, sar_value=9.5, reason="SAR 翻空"):
    return {"phase": phase, "sar_flip_days": flip, "confirm": confirm,
            "sar_value": sar_value, "reason": reason, "stance": "看空"}


def tpos(t, **kw):
    p = pos(**kw)
    p["tech"] = t
    return p


def tparams(**kw):
    p = {"scale": 2_000_000, "stock_target_default": 0.06, "batch_split": (0.5, 0.25, 0.25),
         "cushion_solid": 0.03, "trim_peak": 0.06, "trim_giveback": 0.5,
         "dca_triggers": (-0.08, -0.15), "dca_deep_confirm": -0.15, "dca_max_ratio": 0.5,
         "no_chase_ma5": 0.06, "build_window_tdays": 10, "fill_max_loss": -0.03,
         "tech_exit_on": True, "tech_exit_propose_only": False,
         "tech_exit_trim_ratio": 1.0 / 3, "tech_exit_fresh_days": 2, "tech_exit_done": set()}
    p.update(kw)
    return p


# ================================================================ A eval_tech_exit
@case("A 转空离场·开关关掉逐字回旧 (tech_exit_on 假 → 不评)")
def _():
    assert ae.eval_tech_exit(tpos(tech(confirm=True)), tparams(tech_exit_on=False)) is None


@case("A 转空离场·相位非转空不评")
def _():
    assert ae.eval_tech_exit(tpos(tech(phase="趋势空", confirm=True)), tparams()) is None
    assert ae.eval_tech_exit(tpos(tech(phase="趋势多", confirm=False)), tparams()) is None


@case("A 转空离场·确认转空清仓全部可卖量 (EXIT, 来源 tech_exit)")
def _():
    c = ae.eval_tech_exit(tpos(tech(confirm=True), total_qty=6000, avail_qty=6000), tparams())
    assert c and c["action"] == "EXIT" and c["side"] == "sell"
    assert c["qty"] == 6000 and c["source"] == ae.SRC_TECH_EXIT


@case("A 转空离场·未确认减三分之一 (TRIM)")
def _():
    c = ae.eval_tech_exit(tpos(tech(confirm=False), total_qty=6000, avail_qty=6000), tparams())
    assert c and c["action"] == "TRIM" and c["qty"] == 2000


@case("A 转空离场·翻向超期不评 (超过 fresh_days)")
def _():
    assert ae.eval_tech_exit(tpos(tech(flip=3, confirm=True)), tparams()) is None
    assert ae.eval_tech_exit(tpos(tech(flip=2, confirm=True)), tparams()) is not None


@case("A 转空离场·同一翻空已处理过不再评")
def _():
    p = tpos(tech(confirm=True), ts_code="600000.SH")
    assert ae.eval_tech_exit(p, tparams(tech_exit_done={"600000.SH"})) is None
    assert ae.eval_tech_exit(p, tparams(tech_exit_done=set())) is not None


@case("A 转空离场·无技术面读数不评 (无读数弃权)")
def _():
    p = pos()
    p.pop("tech", None)
    assert ae.eval_tech_exit(p, tparams()) is None
    p2 = pos()
    p2["tech"] = None
    assert ae.eval_tech_exit(p2, tparams()) is None


@case("A 转空离场·propose_only 打交人标记")
def _():
    c = ae.eval_tech_exit(tpos(tech(confirm=True), total_qty=6000, avail_qty=6000),
                          tparams(tech_exit_propose_only=True))
    assert c and c.get("needs_user_confirm") is True and "propose_only" in (c.get("confirm_why") or "")


@case("A 转空离场·按 T+1 可卖量夹紧 (确认时清可卖量而非总持仓)")
def _():
    c = ae.eval_tech_exit(tpos(tech(confirm=True), total_qty=6000, avail_qty=3000), tparams())
    assert c and c["qty"] == 3000


@case("A 转空离场·不足一手不评")
def _():
    assert ae.eval_tech_exit(tpos(tech(confirm=True), total_qty=50, avail_qty=50), tparams()) is None


# ================================================================ B 卖出优先级
@case("B 卖出优先级·目标价 < 研究走弱 < 技术面转空 < 保垫减仓")
def _():
    pr = lambda a, s: ae._sell_priority({"action": a, "source": s})
    assert (pr("EXIT", ae.SRC_TARGET_PRICE) < pr("TRIM", ae.SRC_RESEARCH_WEAK)
            < pr("EXIT", ae.SRC_TECH_EXIT) < pr("TRIM", ae.SRC_ENGINE))


# ================================================================ C scan 端到端
@case("C scan·确认转空产出 EXIT 候选")
def _():
    p = tpos(tech(confirm=True), total_qty=6000, avail_qty=6000)
    r = ae.scan(positions=[p], params=tparams(), market={p["ts_code"]: {}})
    exits = [c for c in r["candidates"] if c.get("source") == ae.SRC_TECH_EXIT]
    assert len(exits) == 1 and exits[0]["action"] == "EXIT"


@case("C scan·同轮与保垫减仓并现留技术面转空 (优先级更高)")
def _():
    p = tpos(tech(confirm=True), total_qty=6000, avail_qty=6000,
             cushion_peak=0.08, cushion_pct=0.04)
    r = ae.scan(positions=[p], params=tparams(), market={p["ts_code"]: {}})
    sells = [c for c in r["candidates"] if c["side"] == "sell"]
    assert len(sells) == 1 and sells[0]["source"] == ae.SRC_TECH_EXIT


@case("C scan·目标价到价优先于技术面转空")
def _():
    p = tpos(tech(confirm=True), total_qty=6000, avail_qty=6000, price=13.0)
    r = ae.scan(positions=[p], params=tparams(), market={p["ts_code"]: {}},
                stock_params={"600000.SH": {"target_price": 12.5}})
    sells = [c for c in r["candidates"] if c["side"] == "sell"]
    assert len(sells) == 1 and sells[0]["source"] == ae.SRC_TARGET_PRICE


@case("C scan·开关关掉逐字回旧: 挂了转空读数也不产出转空离场")
def _():
    p = tpos(tech(confirm=True), total_qty=6000, avail_qty=6000)
    r = ae.scan(positions=[p], params=tparams(tech_exit_on=False), market={p["ts_code"]: {}})
    assert not any(c.get("source") == ae.SRC_TECH_EXIT for c in r["candidates"])


# ================================================================ D 策略票只看目标价 (Part 2)
@case("D 策略票·目标价照评但技术面转空不评 (按函数判)")
def _():
    p = tpos(tech(confirm=True), total_qty=6000, avail_qty=6000, price=13.0)
    r = ae.scan(positions=[p], params=tparams(), market={p["ts_code"]: {}},
                strategy_codes={"600000.SH"}, stock_params={"600000.SH": {"target_price": 12.5}})
    assert any(c.get("source") == ae.SRC_TARGET_PRICE for c in r["candidates"]), "策略票目标价照评"
    assert not any(c.get("source") == ae.SRC_TECH_EXIT for c in r["candidates"]), "策略票技术面转空不评"


@case("D 策略票·无目标价时技术面转空也不产出 (策略层接管)")
def _():
    p = tpos(tech(confirm=True), total_qty=6000, avail_qty=6000)
    r = ae.scan(positions=[p], params=tparams(), market={p["ts_code"]: {}},
                strategy_codes={"600000.SH"})
    assert not any(c.get("source") == ae.SRC_TECH_EXIT for c in r["candidates"])


# ================================================================ E 参数登记
@case("E 参数·转空离场三键登记, 档位默认 full")
def _():
    for k in ("PMS_TECH_EXIT_AUTONOMY", "PMS_TECH_EXIT_TRIM_RATIO", "PMS_TECH_EXIT_DONE"):
        assert k in ps.RUNTIME_EXTRA, k
    assert ps.RUNTIME_EXTRA["PMS_TECH_EXIT_AUTONOMY"][0] == "full"
    assert ps.RUNTIME_EXTRA["PMS_TECH_EXIT_AUTONOMY"][1] is str
    assert abs(ps.RUNTIME_EXTRA["PMS_TECH_EXIT_TRIM_RATIO"][0] - 1.0 / 3) < 1e-3


# ================================================================ F 跟踪止盈盘中 SAR 止损线 (part 3)
def _trail_sar(price, sar_line, now, avail=6000, avg=10.0, sold_day=0):
    from app.services import strategy_runner as srun
    st = {"ts_code": "600000.SH",
          "params": {"giveback": 0.05, "sell_ratio": 1.0, "sar_line": sar_line}}
    pos = {"avg_cost": avg, "avail_qty": avail, "total_qty": avail, "cushion_pct": price / avg - 1}
    ctx = {"state": {"armed": False, "high_water": price, "sar_stop_day": sold_day}, "notes": []}
    d = srun._eval_trail(st, pos, {"price": price}, now, ctx)
    return d, ctx


@case("F SAR 线·09:45 后跌破 SAR 缓冲即全清 (leg=trail_sar, 当日记一次)")
def _():
    from datetime import datetime
    from app.services import strategy_runner as srun
    d, ctx = _trail_sar(9.9, 10.0, datetime(2026, 9, 11, 10, 0))   # 9.9 < 10×0.997=9.97
    assert d and d["action"] == srun.A_EXIT and d["leg"] == "trail_sar" and d["qty"] == 6000
    from app.core import tradedays as td
    assert ctx["state"]["sar_stop_day"] == td.ymd(datetime(2026, 9, 11, 10, 0))


@case("F SAR 线·09:45 前不触发")
def _():
    from datetime import datetime
    d, _ = _trail_sar(9.5, 10.0, datetime(2026, 9, 11, 9, 40))
    assert d is None


@case("F SAR 线·现价在缓冲之上不触发")
def _():
    from datetime import datetime
    d, _ = _trail_sar(9.98, 10.0, datetime(2026, 9, 11, 10, 0))   # 9.98 > 9.97
    assert d is None


@case("F SAR 线·当日只触发一次 (已记 sar_stop_day)")
def _():
    from datetime import datetime
    from app.core import tradedays as td
    now = datetime(2026, 9, 11, 10, 0)
    d, _ = _trail_sar(9.5, 10.0, now, sold_day=td.ymd(now))
    assert d is None


@case("F SAR 线·没刷进 sar_line 时不判 (加不改, 拿不到 SAR 绝不当跌破)")
def _():
    from datetime import datetime
    from app.services import strategy_runner as srun
    st = {"ts_code": "600000.SH", "params": {"giveback": 0.05, "sell_ratio": 1.0}}  # 无 sar_line
    pos = {"avg_cost": 10.0, "avail_qty": 6000, "total_qty": 6000, "cushion_pct": -0.05}
    ctx = {"state": {"armed": False, "high_water": 9.5}, "notes": []}
    assert srun._eval_trail(st, pos, {"price": 9.5}, datetime(2026, 9, 11, 10, 0), ctx) is None


@case("F SAR 线·开关关掉不触发")
def _():
    from datetime import datetime
    saved = ps.get_bool
    ps.get_bool = lambda k, d=False: (False if k == "PMS_TECH_SAR_STOP_ON_TRAIL" else saved(k, d))
    try:
        d, _ = _trail_sar(9.5, 10.0, datetime(2026, 9, 11, 10, 0))
        assert d is None
    finally:
        ps.get_bool = saved


# ================================================================ G SAR 线刷新腿 (part 3)
@case("G 刷新腿·只动自动挂载的 ACTIVE 跟踪止盈, 网格与人工不碰")
def _():
    from app.services import strategy_advisor as adv, tech_service
    from app.repo import pms_repo
    calls = []
    s_sm, s_up = tech_service.state_map, pms_repo.update_strategy
    tech_service.state_map = lambda: {"600000.SH": {"sar_value": 9.55}}
    pms_repo.update_strategy = lambda sid, **kw: (calls.append((sid, kw)) or 1)
    try:
        strat_all = [
            {"strategy_id": "A", "ts_code": "600000.SH", "type": "TRAIL", "status": "ACTIVE",
             "note": "自动挂载: x", "params": {"giveback": 0.05}},
            {"strategy_id": "B", "ts_code": "600000.SH", "type": "GRID", "status": "ACTIVE",
             "note": "自动挂载: x", "params": {}},
            {"strategy_id": "C", "ts_code": "600000.SH", "type": "TRAIL", "status": "ACTIVE",
             "note": "手工挂的", "params": {}},
        ]
        out = {"errors": []}
        adv._refresh_sar_lines(strat_all, out)
        assert len(calls) == 1 and calls[0][0] == "A"
        assert abs(calls[0][1]["params"]["sar_line"] - 9.55) < 1e-6
    finally:
        tech_service.state_map, pms_repo.update_strategy = s_sm, s_up


@case("G 刷新腿·无读数撤掉旧 SAR 线 (不拿旧读数当今天)")
def _():
    from app.services import strategy_advisor as adv, tech_service
    from app.repo import pms_repo
    calls = []
    s_sm, s_up = tech_service.state_map, pms_repo.update_strategy
    tech_service.state_map = lambda: {}          # 全无读数
    pms_repo.update_strategy = lambda sid, **kw: (calls.append((sid, kw)) or 1)
    try:
        strat_all = [{"strategy_id": "A", "ts_code": "600000.SH", "type": "TRAIL",
                      "status": "ACTIVE", "note": "自动挂载: x",
                      "params": {"giveback": 0.05, "sar_line": 9.9}}]
        out = {"errors": []}
        adv._refresh_sar_lines(strat_all, out)
        assert len(calls) == 1 and "sar_line" not in calls[0][1]["params"]
    finally:
        tech_service.state_map, pms_repo.update_strategy = s_sm, s_up


@case("G 参数·SAR 止损线两键登记")
def _():
    assert ps.RUNTIME_EXTRA["PMS_TECH_SAR_STOP_ON_TRAIL"][0] is True
    assert abs(ps.RUNTIME_EXTRA["PMS_TECH_SAR_STOP_BUFFER"][0] - 0.003) < 1e-9


# ================================================================ H 弱基本面试探仓紧止盈自动挂载 (part 4)
@case("H 紧止盈·入场账本 advice.tight_trail 为真才认")
def _():
    from app.services import strategy_advisor as adv
    from app.repo import pms_repo
    s_lots, s_ins, s_led = pms_repo.list_lots, pms_repo.get_instruction, pms_repo.ledger_by_ref
    pms_repo.list_lots = lambda code, **kw: [{"instruction_id": "I1"}]
    pms_repo.get_instruction = lambda iid: {"origin_id": "P1"}
    try:
        pms_repo.ledger_by_ref = lambda refs: [{"verdict": "PASS",
                                                "hard_numbers": {"advice": {"tight_trail": True}}}]
        assert adv._entry_tight_trail("600000.SH") is True
        pms_repo.ledger_by_ref = lambda refs: [{"verdict": "PASS",
                                                "hard_numbers": {"advice": {"tight_trail": False}}}]
        assert adv._entry_tight_trail("600000.SH") is False
        pms_repo.ledger_by_ref = lambda refs: [{"verdict": "PASS", "hard_numbers": {}}]
        assert adv._entry_tight_trail("600000.SH") is False
    finally:
        pms_repo.list_lots, pms_repo.get_instruction, pms_repo.ledger_by_ref = s_lots, s_ins, s_led


@case("H 紧止盈·非试探仓不挂 (返回 False, 走常规边)")
def _():
    from app.services import strategy_advisor as adv
    s = adv._entry_tight_trail
    adv._entry_tight_trail = lambda code: False
    try:
        out = {"attached": [], "skipped": [], "blocked": []}
        r = adv._attach_tight_trail({"ts_code": "600000.SH"}, {"optout_cooldown_tdays": 10},
                                    set(), set(), True, out, None)
        assert r is False and not out["attached"]
    finally:
        adv._entry_tight_trail = s


@case("H 紧止盈·试探仓 dry_run 挂 回撤3%/硬目标8%/带SAR线")
def _():
    from app.services import strategy_advisor as adv, tech_service
    s_e, s_sm = adv._entry_tight_trail, tech_service.state_map
    adv._entry_tight_trail = lambda code: True
    tech_service.state_map = lambda: {"600000.SH": {"sar_value": 9.6}}
    try:
        out = {"attached": [], "skipped": [], "blocked": []}
        r = adv._attach_tight_trail({"ts_code": "600000.SH", "price": 10.0},
                                    {"optout_cooldown_tdays": 10}, set(), set(), True, out, None)
        assert r is True and len(out["attached"]) == 1
        pr = out["attached"][0]["params"]
        assert abs(pr["giveback"] - 0.03) < 1e-9 and abs(pr["hard_target"] - 0.08) < 1e-9
        assert abs(pr["sar_line"] - 9.6) < 1e-6
    finally:
        adv._entry_tight_trail, tech_service.state_map = s_e, s_sm


@case("H 紧止盈·真挂调 strategy_service.attach 且 note 标弱基本面")
def _():
    from app.services import strategy_advisor as adv, tech_service
    from app.repo import pms_repo
    s_e, s_sm, s_il = adv._entry_tight_trail, tech_service.state_map, pms_repo.insert_ledger
    adv._entry_tight_trail = lambda code: True
    tech_service.state_map = lambda: {}
    pms_repo.insert_ledger = lambda **kw: "L"
    calls = []

    class SS:
        def attach(self, spec, by=None):
            calls.append(spec)
            return {"ok": True, "strategy_id": "T9"}
    try:
        out = {"attached": [], "skipped": [], "blocked": [], "errors": []}
        r = adv._attach_tight_trail({"ts_code": "600000.SH", "price": 10.0},
                                    {"optout_cooldown_tdays": 10}, set(), set(), False, out, SS())
        assert r is True and calls and calls[0]["type"] == "TRAIL"
        assert "试探仓" in calls[0]["note"] and "紧止盈" in calls[0]["note"]
        assert out["attached"][0]["strategy_id"] == "T9"
    finally:
        adv._entry_tight_trail, tech_service.state_map, pms_repo.insert_ledger = s_e, s_sm, s_il


@case("H 紧止盈·人工撤下冷却期内占位让路 (返回 True 不走常规边, 不挂)")
def _():
    from app.services import strategy_advisor as adv
    s_e = adv._entry_tight_trail
    adv._entry_tight_trail = lambda code: True
    try:
        out = {"attached": [], "skipped": [], "blocked": []}
        r = adv._attach_tight_trail({"ts_code": "600000.SH"}, {"optout_cooldown_tdays": 10},
                                    set(), {("600000.SH", adv.R_TRAIL)}, True, out, None)
        assert r is True and len(out["skipped"]) == 1 and not out["attached"]
    finally:
        adv._entry_tight_trail = s_e


@case("H 参数·紧止盈两键登记")
def _():
    assert abs(ps.RUNTIME_EXTRA["PMS_TECH_TIGHT_TRAIL_GIVEBACK"][0] - 0.03) < 1e-9
    assert abs(ps.RUNTIME_EXTRA["PMS_TECH_TIGHT_TRAIL_TARGET"][0] - 0.08) < 1e-9


def main():
    ok = 0
    for name, fn in RESULTS:
        try:
            fn()
            ok += 1
            print("  ok  " + name)
        except Exception:
            print("  FAIL " + name)
            traceback.print_exc()
    print("-" * 60)
    if ok == len(RESULTS):
        print("ALL PASS (%d cases)" % ok)
        return 0
    print("FAILED %d/%d" % (len(RESULTS) - ok, len(RESULTS)))
    return 1


if __name__ == "__main__":
    sys.exit(main())