tradingSystem/scripts/test_batch4_units.py

# -*- coding: utf-8 -*-
"""
第四批模块单测 (实机运行, 零外部依赖)
======================================
运行: 在 tradingSystem 仓库根目录执行  python scripts/test_batch4_units.py
覆盖: action_engine 四类自主动作的触发边界与数量口径 (FILL/ADD/DCA/TRIM) 及扫描剪枝。
"""
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

RESULTS = []


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


PARAMS = {"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}


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"}
    p.update(kw)
    return p


def mkt(**kw):
    m = {"ma5": 10.0, "high5": 10.0, "tdays_since_open": 3, "tdays_since_last_add": 5}
    m.update(kw)
    return m


# ================================================================ FILL
@case("回踩补足·建仓期内浅亏未破支撑 → 按补足批额度补到目标")
def _():
    c = ae.eval_fill(pos(price=9.8, cushion_pct=-0.02, support_ref=9.7, market_value=58_800),
                     PARAMS, mkt())
    assert c and c["action"] == "FILL" and c["side"] == "buy", c
    assert c["qty"] == 3000, c            # 补足批 3 万 ÷ 9.8 → 3000 股
    assert c["judge_required"] and not c["needs_user_confirm"]
    assert "未破支撑" in c["reason"]


@case("回踩补足·每票一次 / 出建仓期 / 亏太深 / 破支撑 / 无支撑 一律不提")
def _():
    base = dict(price=9.8, cushion_pct=-0.02, support_ref=9.7, market_value=58_800)
    assert ae.eval_fill(pos(**base, fill_count=1), PARAMS, mkt()) is None
    assert ae.eval_fill(pos(**base), PARAMS, mkt(tdays_since_open=11)) is None
    assert ae.eval_fill(pos(price=9.5, cushion_pct=-0.05, support_ref=9.0,
                            market_value=57_000), PARAMS, mkt()) is None
    assert ae.eval_fill(pos(price=9.5, cushion_pct=-0.02, support_ref=9.7,
                            market_value=57_000), PARAMS, mkt()) is None   # 破支撑
    assert ae.eval_fill(pos(**{**base, "support_ref": None}), PARAMS, mkt()) is None
    # 已转盈就不是「回踩补足」的场景了
    assert ae.eval_fill(pos(price=10.5, cushion_pct=0.05, support_ref=9.7,
                            market_value=63_000), PARAMS, mkt()) is None


# ================================================================ ADD
@case("盈利加仓·厚垫 + 创5日新高 → 按加仓批额度加")
def _():
    c = ae.eval_add(pos(price=11.0, cushion_pct=0.10, market_value=66_000),
                    PARAMS, mkt(ma5=10.6, high5=11.0))
    assert c and c["action"] == "ADD" and c["qty"] == 2700, c   # 3万 ÷ 11 → 2700 股
    assert "创 5 日新高" in c["reason"], c
    # 站上压力位也算
    c2 = ae.eval_add(pos(price=11.0, cushion_pct=0.10, pressure_ref=10.9,
                         market_value=66_000), PARAMS, mkt(ma5=10.6, high5=12.0))
    assert c2 and "压力位" in c2["reason"], c2


@case("盈利加仓·薄垫/未突破/距上次<2日/追高 一律不提")
def _():
    assert ae.eval_add(pos(price=11.0, cushion_pct=0.02, market_value=66_000),
                       PARAMS, mkt(high5=11.0)) is None                    # 垫子不厚
    assert ae.eval_add(pos(price=10.5, cushion_pct=0.10, market_value=63_000),
                       PARAMS, mkt(high5=12.0)) is None                    # 没新高没压力位
    assert ae.eval_add(pos(price=11.0, cushion_pct=0.10, market_value=66_000),
                       PARAMS, mkt(high5=11.0, tdays_since_last_add=1)) is None
    assert ae.eval_add(pos(price=11.0, cushion_pct=0.10, market_value=66_000),
                       PARAMS, mkt(ma5=10.0, high5=11.0)) is None          # 距 MA5 10% > 6%


# ================================================================ DCA
@case("补仓·触及 -8% 首档 → 按底仓一半提, 不强制确认")
def _():
    c = ae.eval_dca(pos(price=9.1, cushion_pct=-0.09, market_value=54_600), PARAMS, mkt())
    assert c and c["action"] == "DCA" and c["qty"] == 3000, c      # 底仓 6000 × 50%
    assert c["needs_user_confirm"] is False and c["judge_required"], c
    assert c["hard_numbers"]["stage"] == 1


@case("补仓·-15% 及更深 永远需用户确认 + 研判须答杀逻辑还是杀情绪")
def _():
    c = ae.eval_dca(pos(price=8.4, cushion_pct=-0.16, market_value=50_400), PARAMS, mkt())
    assert c and c["needs_user_confirm"] is True, c
    assert c["hard_numbers"]["stage"] == 2 and "必须用户确认" in c["reason"], c


@case("补仓·各档只评估一次 / 终身只执行一次 / 浮盈不评估")
def _():
    assert ae.eval_dca(pos(cushion_pct=-0.09, dca_count=1), PARAMS, mkt()) is None
    assert ae.eval_dca(pos(cushion_pct=-0.16, dca_count=1), PARAMS, mkt()) is not None
    assert ae.eval_dca(pos(cushion_pct=-0.16, dca_count=2), PARAMS, mkt()) is None
    assert ae.eval_dca(pos(cushion_pct=-0.20, dca_qty=1000), PARAMS, mkt()) is None
    assert ae.eval_dca(pos(cushion_pct=-0.05), PARAMS, mkt()) is None
    assert ae.eval_dca(pos(cushion_pct=0.05), PARAMS, mkt()) is None


# ================================================================ TRIM
@case("保垫减仓·峰值≥6%且回吐过半 → 减 1/3, 不需研判不需确认")
def _():
    c = ae.eval_trim(pos(cushion_peak=0.08, cushion_pct=0.04), PARAMS)
    assert c and c["action"] == "TRIM" and c["side"] == "sell", c
    assert c["qty"] == 2000, c                                    # 6000 的 1/3
    assert not c["judge_required"] and not c["needs_user_confirm"], c


@case("保垫减仓·峰值不足或回吐不到一半 不提")
def _():
    assert ae.eval_trim(pos(cushion_peak=0.05, cushion_pct=-0.01), PARAMS) is None
    assert ae.eval_trim(pos(cushion_peak=0.08, cushion_pct=0.0401), PARAMS) is None
    assert ae.eval_trim(pos(cushion_peak=0.08, cushion_pct=None), PARAMS) is None
    assert ae.eval_trim(pos(cushion_peak=0.08, cushion_pct=0.04, total_qty=200),
                        PARAMS) is None                           # 1/3 不足一手


# ================================================================ 目标价到价 (EXIT)
@case("目标价·到价产出清仓候选: 全部持仓 / 不送研判 / 强制交人裁决")
def _():
    c = ae.eval_target(pos(price=13.0, target_price=12.5, total_qty=6050,
                           avail_qty=6050, cushion_pct=0.30), PARAMS)
    assert c and c["action"] == ae.A_EXIT and c["side"] == "sell", c
    assert c["qty"] == 6050, c                       # 全部持仓, 零股一并 (与清仓命令同口径)
    assert not c["judge_required"], c                # 减持不送研判
    assert c["needs_user_confirm"] is True, c        # 必定交人
    assert c["source"] == ae.SRC_TARGET_PRICE, c
    assert c["confirm_why"] == ae.WHY_TARGET_PRICE_CONFIRM, c
    assert c["hard_numbers"]["target_price"] == 12.5, c["hard_numbers"]
    # 恰好等于目标价也算到价 (不设缓冲、不做平滑 —— 那个数是用户自己填的)
    assert ae.eval_target(pos(price=12.5, target_price=12.5), PARAMS) is not None


@case("目标价·未到价 / 没设 / 取不到现价 / 空仓 一律不产出")
def _():
    assert ae.eval_target(pos(price=12.4, target_price=12.5), PARAMS) is None   # 差一分不算
    assert ae.eval_target(pos(price=13.0), PARAMS) is None                      # 没设目标价
    assert ae.eval_target(pos(price=13.0, target_price=0), PARAMS) is None
    # 取不到现价: price 是拿摊薄成本顶的假价, 绝不能折成「到价了」
    assert ae.eval_target(pos(price=13.0, target_price=12.5, price_ok=False), PARAMS) is None
    assert ae.eval_target(pos(price=0, target_price=12.5), PARAMS) is None
    assert ae.eval_target(pos(price=None, target_price=12.5), PARAMS) is None
    assert ae.eval_target(pos(price=13.0, target_price=12.5, total_qty=0), PARAMS) is None


@case("扫描·目标价从个股参数命令读 (不读持仓行投影) / 冻结不挡 / 同轮买入让路")
def _():
    ps = [pos(ts_code="600000.SH", price=13.0, cushion_pct=0.30, market_value=78_000,
              frozen_reason="COMMAND_HALT")]
    m = {"600000.SH": mkt(ma5=12.0, high5=13.0)}
    # 不传 stock_params → 这条动作根本不评估, 其余四类行为与从前一字不差
    r0 = ae.scan(positions=ps, params=PARAMS, market=m)
    assert not any(c["action"] == ae.A_EXIT for c in r0["candidates"]), r0["candidates"]
    # 传了命令表里的目标价 → 到价产出清仓候选; 冻结只禁增持, 挡不住它
    r = ae.scan(positions=ps, params=PARAMS, market=m,
                stock_params={"600000.SH": {"target_price": 12.5}})
    acts = {(c["ts_code"], c["action"]) for c in r["candidates"]}
    assert ("600000.SH", ae.A_EXIT) in acts, acts
    # 同轮买卖互斥: 一边提议清仓一边提议加仓是自相矛盾, 买入侧让路
    ps2 = [pos(ts_code="000001.SZ", price=13.0, cushion_pct=0.30, market_value=78_000)]
    m2 = {"000001.SZ": mkt(ma5=12.5, high5=13.0)}      # 不追高那道过得去, 才真会出 ADD
    r2 = ae.scan(positions=ps2, params=PARAMS, market=m2,
                 stock_params={"000001.SZ": {"target_price": 12.5}})
    acts2 = {c["action"] for c in r2["candidates"]}
    assert ae.A_EXIT in acts2 and "ADD" not in acts2, acts2
    assert any("买卖互斥" in (s.get("why") or "") for s in r2["skipped"]), r2["skipped"]
    # 传进来的持仓行不许被改写 (合并出的是本轮副本)
    assert "target_price" not in ps2[0], ps2[0]


@case("扫描·同轮只发一条减持: 到价清仓优先于保垫减仓, 免得人点了采纳却卖不成")
def _():
    # 复现被修掉的那个场景: 峰值安全垫 20% 回落到 9% 会触发保垫减仓, 同时现价已到目标价。
    # 修之前两条会一起留下 —— 保垫减仓不用确认、当场卖掉一部分, 而到价清仓记的是扫描那一刻
    # 的全部持仓、要等人拍板; 等人第二天点采纳, 持仓已经少了, 落指令时数量超过实际持仓,
    # 卖出前的检查会整条驳回, 人点了「清掉」结果一股没卖。
    ps = [pos(ts_code="600000.SH", price=13.0, total_qty=6000, avail_qty=6000,
              cushion_peak=0.20, cushion_pct=0.09, market_value=78_000)]
    m = {"600000.SH": mkt(ma5=12.0, high5=13.0)}
    r = ae.scan(positions=ps, params=PARAMS, market=m,
                stock_params={"600000.SH": {"target_price": 12.5}})
    sells = [c for c in r["candidates"] if c["side"] == ae.SELL]
    assert len(sells) == 1, sells                      # 一轮只发一条
    assert sells[0]["action"] == ae.A_EXIT, sells      # 留下的是人设的目标价那条
    assert sells[0]["qty"] == 6000, sells              # 数量仍是全部持仓
    assert any("一轮只发一条减持" in (s.get("why") or "") for s in r["skipped"]), r["skipped"]
    # 没设目标价时保垫减仓照旧, 老行为一个字不变
    r2 = ae.scan(positions=ps, params=PARAMS, market=m)
    sells2 = [c for c in r2["candidates"] if c["side"] == ae.SELL]
    assert len(sells2) == 1 and sells2[0]["action"] == ae.A_TRIM, sells2


# ================================================================ 扫描
@case("扫描·冻结票只评减仓 / 已在途不重复提 / 单票异常不拖垮整轮")
def _():
    ps = [pos(ts_code="600000.SH", cushion_peak=0.08, cushion_pct=0.04,
              frozen_reason="COMMAND_HALT"),
          pos(ts_code="000001.SZ", price=11.0, cushion_pct=0.10, market_value=66_000),
          pos(ts_code="600519.SH", price=None, cushion_pct=0.10, market_value=66_000)]
    r = ae.scan(positions=ps, params=PARAMS, market={
        "600000.SH": mkt(), "000001.SZ": mkt(ma5=10.6, high5=11.0), "600519.SH": mkt()})
    acts = {(c["ts_code"], c["action"]) for c in r["candidates"]}
    assert ("600000.SH", "TRIM") in acts, acts                    # 冻结不挡减仓
    assert not any(c[0] == "600000.SH" and c[1] != "TRIM" for c in acts), acts
    assert ("000001.SZ", "ADD") in acts, acts
    # 2026-09-04 起这句话按冻结的三种来路分开写, 不再印枚举值。三种来路对交易员是
    # 三件不同的事: 你下的命令、你手设的、系统刹车期间。
    assert all("只卖不买" in s["why"] and "不评估加仓" in s["why"]
               for s in r["skipped"]), r["skipped"]
    assert any("你下过命令" in s["why"] for s in r["skipped"]), r["skipped"]
    for s in r["skipped"]:           # 枚举值不许再出现在给人看的句子里
        for w in ("COMMAND_HALT", "MANUAL", "BRAKE"):
            assert w not in s["why"], s["why"]

    r2 = ae.scan(positions=ps, params=PARAMS,
                 market={"600000.SH": mkt(), "000001.SZ": mkt(ma5=10.6, high5=11.0),
                         "600519.SH": mkt()},
                 skip={("000001.SZ", "ADD")})
    assert ("000001.SZ", "ADD") not in {(c["ts_code"], c["action"]) for c in r2["candidates"]}
    # 2026-08-06: skip 改成 {(代码, 动作): 原因} 的字典, 好让 skipped 说得出是哪一种跳过
    # (在途提议 / 在途指令 / 今日被规则闸拒过 / 今日被研判闸驳回过 —— 前两种明天照样被挡,
    # 后两种日切就重新评估)。这里传的仍是**集合**, 走的是向后兼容那条路, 所以拿到兜底话。
    assert any(s.get("why") == ae.SKIP_WHY_FALLBACK for s in r2["skipped"]), r2["skipped"]

    # 空仓票直接跳过, 不进候选
    r3 = ae.scan(positions=[pos(ts_code="300750.SZ", total_qty=0)], params=PARAMS, market={})
    assert r3["candidates"] == []


@case("跨轮·同票已有减持在途时, 本轮其余减持一律让路 (审查 2026-09-07 第 1 条)")
def _():
    # 第一分钟: 到价清仓 (等人拍板) + 保垫减仓同时触发, 同轮只留到价那条。
    # 第二分钟: 到价清仓已成在途、按 (代码, EXIT) 进了跳过集合 —— 修之前保垫减仓单独
    # 产出、不需确认、当场自动卖掉 1/3, 系统刚说「卖不卖由你定」一分钟后自己先卖了。
    ps = [pos(ts_code="600000.SH", price=13.0, total_qty=6000, avail_qty=6000,
              cushion_peak=0.20, cushion_pct=0.09, market_value=78_000)]
    m = {"600000.SH": mkt(ma5=12.0, high5=13.0)}
    sp = {"600000.SH": {"target_price": 12.5}}
    r1 = ae.scan(positions=ps, params=PARAMS, market=m, stock_params=sp)
    assert [c["action"] for c in r1["candidates"] if c["side"] == ae.SELL] == [ae.A_EXIT]
    # 第二轮: 模拟 proposal_service 把在途的到价清仓放进跳过集合
    r2 = ae.scan(positions=ps, params=PARAMS, market=m, stock_params=sp,
                 skip={("600000.SH", ae.A_EXIT): "已有在途提议 PRP_x 在等人确认"})
    sells2 = [c for c in r2["candidates"] if c["side"] == ae.SELL]
    assert sells2 == [], f"在途清仓等人拍板时保垫减仓不许单独产出: {sells2}"
    whys = [s["why"] for s in r2["skipped"] if s["ts_code"] == "600000.SH"]
    assert any("在处理中" in w and "抢在人前面" in w for w in whys), whys
    # 在途的是 TRIM 时同理: 目标价到价也让路 (人已经在处理这只票的减持)
    r3 = ae.scan(positions=ps, params=PARAMS, market=m, stock_params=sp,
                 skip={("600000.SH", ae.A_TRIM): "已有在途指令 INS_x (DISPATCHED)"})
    assert [c for c in r3["candidates"] if c["side"] == ae.SELL] == []
    # 跳过集合里只有买入侧时, 减持不受影响 (老行为)
    r4 = ae.scan(positions=ps, params=PARAMS, market=m, stock_params=sp,
                 skip={("600000.SH", ae.A_ADD): "x"})
    assert [c["action"] for c in r4["candidates"] if c["side"] == ae.SELL] == [ae.A_EXIT]


@case("策略票·挂了交易方案的票仍评目标价, 其余四类不评 (审查 2026-09-07 第 5 条)")
def _():
    # 修之前 scan() 对策略票整只 continue, 目标价命令显示生效、到价那天什么都不发生。
    ps = [pos(ts_code="600000.SH", price=13.0, total_qty=6000, avail_qty=6000,
              cushion_peak=0.20, cushion_pct=0.09, market_value=78_000)]
    m = {"600000.SH": mkt(ma5=12.0, high5=13.0)}
    r = ae.scan(positions=ps, params=PARAMS, market=m, strategy_codes={"600000.SH"},
                stock_params={"600000.SH": {"target_price": 12.5}})
    acts = [c["action"] for c in r["candidates"]]
    assert acts == [ae.A_EXIT], f"策略票只该产出到价清仓: {acts}"
    assert r["candidates"][0]["needs_user_confirm"] is True
    assert any("你设的目标价仍照常看" in s["why"] for s in r["skipped"]), r["skipped"]
    # 没设目标价时策略票什么都不产出 (老行为一个字不变)
    r2 = ae.scan(positions=ps, params=PARAMS, market=m, strategy_codes={"600000.SH"})
    assert r2["candidates"] == [], r2["candidates"]


@case("到价清仓·按 T+1 可卖量报数, 可卖为零不产出 (审查 2026-09-07 第 4 条)")
def _():
    # 到价当天有买入成交时可卖量小于总持仓; 原先按总持仓报, 规则闸「卖出 > T+1 可卖」
    # 整条拒掉再被当日去重挡住, 人当天看不到这条到价提议。
    m = {"600000.SH": mkt(ma5=12.0, high5=13.0)}
    sp = {"600000.SH": {"target_price": 12.5}}
    p1 = pos(ts_code="600000.SH", price=13.0, total_qty=1000, avail_qty=800,
             cushion_pct=0.30, cushion_peak=0.30, market_value=13_000)
    r = ae.scan(positions=[p1], params=PARAMS, market=m, stock_params=sp)
    ex = [c for c in r["candidates"] if c["action"] == ae.A_EXIT]
    assert len(ex) == 1 and ex[0]["qty"] == 800, ex
    assert ex[0]["hard_numbers"]["total_qty"] == 1000 and ex[0]["hard_numbers"]["avail_qty"] == 800
    assert "今天只能卖 T+1 可卖的 800 股" in ex[0]["reason"], ex[0]["reason"]
    p0 = pos(ts_code="600000.SH", price=13.0, total_qty=1000, avail_qty=0,
             cushion_pct=0.30, cushion_peak=0.30, market_value=13_000)
    r0 = ae.scan(positions=[p0], params=PARAMS, market=m, stock_params=sp)
    assert not [c for c in r0["candidates"] if c["action"] == ae.A_EXIT], r0["candidates"]
    # 可卖量缺失 (None) 按总持仓, 不当成零 —— 旧数据行没有这个字段
    assert ae.clamp_sell_qty(6000, {"avail_qty": None}) == 6000
    assert ae.clamp_sell_qty(6000, {"avail_qty": 2000}) == 2000
    assert ae.clamp_sell_qty(6000, {"avail_qty": 0}) == 0
    assert ae.clamp_sell_qty(0, {"avail_qty": 9999}) == 0


@case("扫描·批次额度与距目标空间口径")
def _():
    p = pos(market_value=100_000)
    assert ae.batch_amount(p, PARAMS, 0) == 60_000.0       # 底仓 50%
    assert ae.batch_amount(p, PARAMS, 1) == 30_000.0       # 补足 25%
    assert ae.batch_amount(p, PARAMS, 2) == 30_000.0       # 加仓 25%
    assert ae.room_to_target(p, PARAMS) == 20_000.0
    assert ae.room_to_target(pos(market_value=130_000), PARAMS) == 0.0


# ---------------------------------------------------------------- runner
def main():
    passed, failed = 0, 0
    for name, fn in RESULTS:
        try:
            fn()
            print(f"  PASS  {name}")
            passed += 1
        except Exception:
            print(f"  FAIL  {name}")
            traceback.print_exc()
            failed += 1
    print("-" * 60)
    if failed:
        print(f"FAILED: {failed} / {passed + failed}")
        sys.exit(1)
    print(f"ALL PASS ({passed} cases)")


if __name__ == "__main__":
    main()