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


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