# -*- coding: utf-8 -*-
"""
第二十一批模块单测 (减持的自动执行边界, 零外部依赖, 不连库不触网)
==================================================================
运行: 在 tradingSystem 仓库根目录执行  python scripts/test_batch21_units.py

背景 (2026-09-03 安全修复): 提议分流原来把自动执行写成
    auto_exec = (side == "sell") or (autonomy == full and not force_queue)
卖出方向在或运算的左边, 把 force_queue (强制入人工队列) 整个短路了 —— 方向是卖,
这个标记就不起作用。当时没出事, 是因为走到分流的卖出候选只有保垫减仓一种, 它既不强制
确认也不送研判, force_queue 恒为假。但设计里明确要求「研究证据走弱触发的减持必须交人
裁决、绝不自动卖」, 那类减持一旦接上来, 结果会是自动卖出。

本批把修完之后的四件事钉死:
  * 规则算出来的保垫减仓仍然自动执行 (老行为一个字不变);
  * 带了强制入队标记的减持不再被方向短路 —— 强制入队是一票否决, 排在方向与档位之前;
  * 研究证据走弱这个来源必定入人工队列 (预留通道: 目前还没有上游在产出它);
  * 两条真正需要自动卖出的路一个字没碰 —— 决策系统高置信风控卖出的自动清仓、用户命令
    驱动的清仓, 它们根本不经过提议分流。这两条用「分流函数上装绊线」的办法证明。
约定同前: 全过输出 "ALL PASS (n cases)" 退出码 0。
"""
import os
import sys
import traceback
from datetime import datetime

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.core import tradedays as td                                           # noqa: E402

RESULTS = []


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


NOW = datetime(2026, 9, 3, 10, 30)


# ================================================================ 夹具
class _Patch:
    """临时替换若干模块属性, 退出时原样还回去 —— 单测不连库不触网。"""

    def __init__(self):
        self.saved = []

    def __call__(self, mod, name, value):
        self.saved.append((mod, name, getattr(mod, name)))
        setattr(mod, name, value)

    def __enter__(self):
        return self

    def __exit__(self, *a):
        for mod, name, old in reversed(self.saved):
            setattr(mod, name, old)
        return False


def _patch_params(p, mod, values):
    """把某模块引用的 param_store 取值函数换成一份字典 (口径同第二十批)。"""
    v = values

    def _get_list(k, d=None, sep=","):
        x = v.get(k)
        if x in (None, ""):
            return list(d or [])
        return [s.strip() for s in str(x).split(sep) if s.strip()]
    p(mod, "get", lambda k, d=None: v.get(k, d))
    p(mod, "get_int", lambda k, d=0: int(v.get(k, d)))
    p(mod, "get_float", lambda k, d=0.0: float(v.get(k, d)))
    p(mod, "get_bool", lambda k, d=False: bool(v.get(k, d)))
    p(mod, "get_list", _get_list)


PARAMS_BASE = {"PMS_EXEC_WINDOW_TDAYS": 3, "PMS_PROPOSAL_TTL_HOURS": 24,
               "PMS_JUDGE_TICK_BUDGET_SEC": 150}

# 研判不可用的应答 (judge.request 的降级形状)
JUDGE_DEGRADED = {"verdict": "UNAVAILABLE", "reason": "决策系统未接通", "degraded": True,
                  "raw": None, "confidence": None}
JUDGE_PASS = {"verdict": "PASS", "reason": "理由仍成立", "degraded": False,
              "raw": {"verdict": "PASS"}, "confidence": 70.0}


def trim_cand(qty=1000, price=10.0):
    """一条真的由动作引擎算出来的保垫减仓候选 (不手搓, 口径与生产一致)。"""
    p = {"ts_code": "600000.SH", "cushion_peak": 0.10, "cushion_pct": 0.04,
         "total_qty": qty * 3, "price": price}
    c = ae.eval_trim(p, {"trim_peak": 0.06, "trim_giveback": 0.5})
    assert c and c["action"] == "TRIM" and c["side"] == "sell" and c["qty"] == qty, c
    return c


def buy_cand(action="ADD", qty=1000, price=10.0, **kw):
    """一条买入侧候选 (加仓)。判定表用它跟卖出侧对照。"""
    c = {"ts_code": "600000.SH", "action": action, "side": "buy", "qty": qty,
         "reason": "测试用加仓候选", "hard_numbers": {"price": price},
         "needs_user_confirm": False, "judge_required": False, "source": ae.SRC_ENGINE}
    c.update(kw)
    return c


def _route(c, *, autonomy="propose_only", judge_resp=JUDGE_PASS, dry_run=False,
           values=None, price=10.0):
    """把一条候选送进 proposal_service._route_one, 规则闸/研判/落表全换成桩。

    回 (out, got): out 是分流结论, got 是这一路真的往库里写了什么。
    """
    from app.core import rule_gate
    from app.repo import pms_repo
    from app.services import judge, param_store, portfolio, proposal_service as psvc
    got = {"ledger": [], "instructions": [], "proposals": [], "judge_calls": []}
    with _Patch() as p:
        p(rule_gate, "check", lambda **kw: {"passed": True, "failed": [], "warnings": []})
        p(judge, "request", lambda cand_, context=None, **kw: (
            got["judge_calls"].append(cand_.get("action")) or dict(judge_resp)))
        p(portfolio, "caps_ctx", lambda *a, **kw: {})
        p(pms_repo, "insert_ledger", lambda **kw: got["ledger"].append(kw) or 1)
        p(pms_repo, "insert_instruction", lambda **kw: got["instructions"].append(kw) or 1)
        p(pms_repo, "insert_proposal", lambda **kw: got["proposals"].append(kw) or 1)
        p(pms_repo, "list_ledger", lambda **kw: [])
        p(pms_repo, "update_position", lambda code, **kw: 1)
        _patch_params(p, param_store, {**PARAMS_BASE, **(values or {})})
        out = {"autonomy": autonomy, "open_autonomy": autonomy, "executed": [], "queued": [],
               "rejected": [], "skipped": [], "errors": [], "degraded": False}
        held = [{"ts_code": c["ts_code"], "total_qty": 3000, "avail_qty": 3000,
                 "price": price, "frozen_reason": "NONE"}]
        view = {"positions": held, "held": held, "sector_ready": False,
                "params": {}, "totals": {}}
        psvc._route_one(c, view, {"_mkt": {}}, {}, False, NOW, dry_run, out)
    return out, got


# ================================================================ 一, 老行为不变
@case("保垫减仓·规则算出来的减持仍然自动执行 (两个档位都是, 老行为一个字不变)")
def _():
    for autonomy in ("propose_only", "full"):
        out, got = _route(trim_cand(), autonomy=autonomy)
        assert len(out["executed"]) == 1 and not out["queued"], (autonomy, out)
        assert "减持自动执行" in out["executed"][0]["why"], out["executed"]
        assert len(got["instructions"]) == 1 and not got["proposals"], (autonomy, got)
        ins = got["instructions"][0]
        assert ins["action"] == "TRIM" and ins["side"] == "sell" and ins["qty"] == 1000, ins
        assert ins["progress"]["is_command"] is False and ins["progress"]["auto"] is True, ins
        # 账本记一条放行, 仲裁人是规则 (减仓不送研判, 见下条用例)
        assert got["ledger"][0]["verdict"] == "PASS" and got["ledger"][0]["arbiter"] == "rule"


@case("保垫减仓·不送研判闸 (JUDGE_ACTIONS 只有买入侧四类), 分流不因此改判")
def _():
    c = trim_cand()
    assert c["judge_required"] is False, c
    assert ae.A_TRIM not in ae.JUDGE_ACTIONS and ae.A_OPEN in ae.JUDGE_ACTIONS
    out, got = _route(c)
    assert got["judge_calls"] == [], got["judge_calls"]
    assert len(out["executed"]) == 1, out


@case("判定表·买入侧四种组合逐条对照 (档位与强制入队的老口径一个字不变)")
def _():
    table = [
        # (档位, 强制确认, 期望自动执行)
        ("full", False, True),
        ("full", True, False),
        ("propose_only", False, False),
        ("propose_only", True, False),
    ]
    for autonomy, confirm, want_auto in table:
        c = buy_cand(needs_user_confirm=confirm)
        out, _ = _route(c, autonomy=autonomy)
        got_auto = bool(out["executed"])
        assert got_auto is want_auto, (autonomy, confirm, want_auto, out)


# ================================================================ 二, 短路修掉了
@case("修复要点·带强制确认标记的减持不再被方向短路 (从前 side==sell 直接自动卖)")
def _():
    c = trim_cand()
    c["needs_user_confirm"] = True
    c["confirm_why"] = "测试用: 这条减持必须交人"
    for autonomy in ("propose_only", "full"):
        out, got = _route(c, autonomy=autonomy)
        assert not out["executed"] and len(out["queued"]) == 1, (autonomy, out)
        assert out["queued"][0]["why"] == "测试用: 这条减持必须交人", out["queued"]
        assert not got["instructions"] and len(got["proposals"]) == 1, (autonomy, got)
        pr = got["proposals"][0]
        assert pr["action"] == "TRIM" and pr["qty"] == 1000, pr
        assert pr["hard_numbers"]["needs_user_confirm"] is True, pr["hard_numbers"]


@case("修复要点·研判不可用的减持同样入队 (卖出方向不再绕开降级入队这条规矩)")
def _():
    # 今天减仓不送研判, 所以这条走的是「将来减仓也进研判范围」的那种处境:
    # 候选自己声明要研判, 而研判回不可用 —— 降级入队对卖出必须同样有效。
    c = trim_cand()
    c["judge_required"] = True
    out, got = _route(c, autonomy="full", judge_resp=JUDGE_DEGRADED)
    assert got["judge_calls"] == ["TRIM"], got["judge_calls"]
    assert not out["executed"] and len(out["queued"]) == 1, out
    assert out["queued"][0]["why"] == "研判不可用, 降级人工确认", out["queued"]
    assert not got["instructions"] and len(got["proposals"]) == 1, got
    # 研判真回了通过时照旧自动执行 (降级入队只针对拿不到结论)
    out2, got2 = _route(c, autonomy="full", judge_resp=JUDGE_PASS)
    assert len(out2["executed"]) == 1 and not out2["queued"], out2
    assert got2["ledger"][0]["arbiter"] == "judge", got2["ledger"]


@case("修复要点·试算口径与真跑一致: 强制入队的减持在 dry_run 里也归入队一列")
def _():
    c = trim_cand()
    c["needs_user_confirm"] = True
    out, got = _route(c, autonomy="full", dry_run=True)
    assert not out["executed"] and len(out["queued"]) == 1, out
    assert out["queued"][0]["route"] == "queue" and out["queued"][0]["dry_run"] is True, out
    assert not got["instructions"] and not got["proposals"], got     # 试算滴水不写
    # 规则触发的那条在试算里仍是自动执行
    out2, _ = _route(trim_cand(), autonomy="full", dry_run=True)
    assert out2["executed"] and out2["executed"][0]["route"] == "auto", out2


# ================================================================ 三, 研究证据走弱的预留通道
@case("预留通道·来源常量与强制入队判定 (认不出的来源不影响老行为)")
def _():
    assert ae.SRC_ENGINE == "engine" and ae.SRC_RESEARCH_WEAK == "research_weak"
    # 名单只查「研究证据走弱在里面、动作引擎自己不在里面」, 不查它一共有几条 ——
    # 这份名单本来就是给后来的来源加行用的, 断言写成全等于会拦住正当的新增。
    assert ae.FORCE_QUEUE_SOURCES[ae.SRC_RESEARCH_WEAK] == ae.WHY_RESEARCH_WEAK_CONFIRM
    assert ae.SRC_ENGINE not in ae.FORCE_QUEUE_SOURCES, ae.FORCE_QUEUE_SOURCES
    # 名单里的每一条都必须给得出一句交人的原因 (不许只登记键、原因留空)
    for src, why in ae.FORCE_QUEUE_SOURCES.items():
        assert isinstance(why, str) and why.strip(), (src, why)
        assert ae.source_confirm_why(src) == why, src
    assert ae.source_confirm_why(ae.SRC_RESEARCH_WEAK) == ae.WHY_RESEARCH_WEAK_CONFIRM
    assert ae.source_confirm_why(" research_weak ") == ae.WHY_RESEARCH_WEAK_CONFIRM
    # 没有来源 / 动作引擎自己 / 认不出的来源: 一律不强制, 走原有的方向与档位判定
    for x in (None, "", "engine", "signal", "某个将来的来源"):
        assert ae.source_confirm_why(x) is None, x
    assert "交人裁决" in ae.WHY_RESEARCH_WEAK_CONFIRM
    assert "不自动卖出" in ae.WHY_RESEARCH_WEAK_CONFIRM


@case("预留通道·候选构造器产出的形状: 卖出方向 / 强制确认 / 带来源与交人原因")
def _():
    c = ae.reduce_on_weak_research("600000.SH", qty=1000,
                                   reason="研究证据走弱: 三条买入理由有两条不再成立",
                                   hard_numbers={"price": 10.0})
    assert c["ts_code"] == "600000.SH" and c["action"] == "TRIM" and c["side"] == "sell"
    assert c["qty"] == 1000 and c["source"] == ae.SRC_RESEARCH_WEAK, c
    assert c["needs_user_confirm"] is True, c
    assert c["confirm_why"] == ae.WHY_RESEARCH_WEAK_CONFIRM, c
    assert c["hard_numbers"]["price"] == 10.0, c
    # 清仓口径也能用同一个构造器
    c2 = ae.reduce_on_weak_research("600000.SH", qty=3000, reason="清", action="EXIT")
    assert c2["action"] == "EXIT" and c2["side"] == "sell" and c2["source"] == ae.SRC_RESEARCH_WEAK


@case("预留通道·研究证据走弱的减持必定入人工队列 (full 档位也不许自动卖)")
def _():
    c = ae.reduce_on_weak_research("600000.SH", qty=1000, reason="研究证据走弱",
                                   hard_numbers={"price": 10.0})
    for autonomy in ("propose_only", "full"):
        out, got = _route(c, autonomy=autonomy)
        assert not out["executed"] and len(out["queued"]) == 1, (autonomy, out)
        assert out["queued"][0]["why"] == ae.WHY_RESEARCH_WEAK_CONFIRM, out["queued"]
        assert not got["instructions"] and len(got["proposals"]) == 1, (autonomy, got)
        assert got["proposals"][0]["hard_numbers"]["source"] == ae.SRC_RESEARCH_WEAK


@case("预留通道·只带来源、没带强制确认标记的减持一样入队 (来源自己就是一票否决)")
def _():
    c = trim_cand()
    c["source"] = ae.SRC_RESEARCH_WEAK          # 只改来源, 强制确认标记仍是假
    assert c["needs_user_confirm"] is False and "confirm_why" not in c
    out, got = _route(c, autonomy="full")
    assert not out["executed"] and len(out["queued"]) == 1, out
    assert out["queued"][0]["why"] == ae.WHY_RESEARCH_WEAK_CONFIRM, out["queued"]
    assert not got["instructions"] and len(got["proposals"]) == 1, got


@case("留痕·提议硬数字带来源, 人在等我拍板里看得出这条减持是哪来的")
def _():
    c = trim_cand()
    c["needs_user_confirm"] = True               # 先让它入队才有提议可看
    _, got = _route(c, autonomy="full")
    assert got["proposals"][0]["hard_numbers"]["source"] == ae.SRC_ENGINE, got["proposals"]
    c2 = ae.reduce_on_weak_research("600000.SH", qty=1000, reason="研究证据走弱")
    _, got2 = _route(c2, autonomy="full")
    assert got2["proposals"][0]["hard_numbers"]["source"] == ae.SRC_RESEARCH_WEAK


# ================================================================ 四, 两条自动卖出的路没被碰到
@case("自动止损·风控高置信卖出照旧直接落清仓指令, 且根本不经过提议分流")
def _():
    from app.repo import pms_repo
    from app.services import param_store, proposal_service as psvc, signal_service as ssvc
    got = {"instructions": [], "ledger": [], "proposals": [], "routed": []}
    with _Patch() as p:
        # 绊线: 这条路要是走进了提议分流, 用例立刻炸 —— 这次改动改的就是那一段
        p(psvc, "_route_one", lambda *a, **kw: got["routed"].append(a) or None)
        p(pms_repo, "list_instructions", lambda **kw: [])
        p(pms_repo, "list_proposals", lambda **kw: [])
        p(pms_repo, "insert_instruction", lambda **kw: got["instructions"].append(kw) or 1)
        p(pms_repo, "insert_ledger", lambda **kw: got["ledger"].append(kw) or 1)
        p(pms_repo, "insert_proposal", lambda **kw: got["proposals"].append(kw) or 1)
        _patch_params(p, param_store, PARAMS_BASE)
        pos = {"ts_code": "600000.SH", "total_qty": 3000, "avail_qty": 3000, "price": 9.5}
        view = {"positions": [pos], "held": [pos]}
        prm = {"sell_conf_min": 0.75, "auto_exit_conf": 0.85, "trim_ratio": 1 / 3}
        out = {"ok": True, "read": 0, "exits": [], "proposals": [], "recorded": 0,
               "ignored": 0, "errors": [], "dry_run": False}
        sig = {"ts_code": "600000.SH", "action": "SELL", "confidence": 0.92,
               "source": "bionic_risk", "reason": "风控: 形态破位", "msg_id": "1-1"}
        ssvc._handle(sig, view, prm, {}, td.ymd(), False, out)
    assert got["routed"] == [], "风控卖出不该经过提议分流"
    assert len(out["exits"]) == 1 and not out["proposals"], out
    assert len(got["instructions"]) == 1 and not got["proposals"], got
    ins = got["instructions"][0]
    assert ins["action"] == "EXIT" and ins["side"] == "sell" and ins["qty"] == 3000, ins
    assert ins["progress"]["urgent"] is True and ins["progress"]["from_signal"] is True, ins
    assert ins["progress"]["is_command"] is False, ins


@case("自动止损·中等置信的风控卖出照旧只落提议 (门槛分档没被这次改动碰到)")
def _():
    from app.repo import pms_repo
    from app.services import param_store, proposal_service as psvc, signal_service as ssvc
    got = {"instructions": [], "ledger": [], "proposals": [], "routed": []}
    with _Patch() as p:
        p(psvc, "_route_one", lambda *a, **kw: got["routed"].append(a) or None)
        p(pms_repo, "list_instructions", lambda **kw: [])
        p(pms_repo, "list_proposals", lambda **kw: [])
        p(pms_repo, "insert_instruction", lambda **kw: got["instructions"].append(kw) or 1)
        p(pms_repo, "insert_ledger", lambda **kw: got["ledger"].append(kw) or 1)
        p(pms_repo, "insert_proposal", lambda **kw: got["proposals"].append(kw) or 1)
        _patch_params(p, param_store, PARAMS_BASE)
        pos = {"ts_code": "600000.SH", "total_qty": 3000, "avail_qty": 3000, "price": 9.5}
        view = {"positions": [pos], "held": [pos]}
        prm = {"sell_conf_min": 0.75, "auto_exit_conf": 0.85, "trim_ratio": 1 / 3}
        out = {"ok": True, "read": 0, "exits": [], "proposals": [], "recorded": 0,
               "ignored": 0, "errors": [], "dry_run": False}
        sig = {"ts_code": "600000.SH", "action": "SELL", "confidence": 0.80,
               "source": "bionic_risk", "reason": "风控: 量能转弱", "msg_id": "1-2"}
        ssvc._handle(sig, view, prm, {}, td.ymd(), False, out)
    assert got["routed"] == [], "风控卖出不该经过提议分流"
    assert not out["exits"] and len(out["proposals"]) == 1, out
    assert not got["instructions"] and len(got["proposals"]) == 1, got
    assert got["proposals"][0]["action"] == "TRIM" and got["proposals"][0]["qty"] == 1000


@case("命令清仓·一键清仓照常出方案, 且根本不经过提议分流")
def _():
    from app.core import command_spec as cs
    from app.repo import pms_repo
    from app.services import (command_service as csvc, param_store, portfolio,
                              proposal_service as psvc, strategy_service)
    got = {"plans": [], "ledger": [], "routed": []}
    with _Patch() as p:
        p(psvc, "_route_one", lambda *a, **kw: got["routed"].append(a) or None)
        p(psvc, "scan_and_route", lambda **kw: got["routed"].append(("scan",)) or {})
        p(pms_repo, "update_command", lambda *a, **kw: 1)
        p(pms_repo, "insert_plans", lambda rows: got["plans"].extend(rows) or len(rows))
        p(pms_repo, "list_plans", lambda **kw: [])
        p(pms_repo, "list_instructions", lambda **kw: [])
        p(pms_repo, "list_strategies", lambda **kw: [])
        p(pms_repo, "list_proposals", lambda **kw: [])
        p(pms_repo, "insert_ledger", lambda **kw: got["ledger"].append(kw) or 1)
        p(strategy_service, "set_status",
          lambda sid, status, by="user": {"ok": True, "status": status})
        held = [{"ts_code": "600000.SH", "total_qty": 3000, "avail_qty": 3000, "price": 9.5,
                 "price_ok": True, "base_qty": 3000}]
        p(portfolio, "positions_view", lambda **kw: {
            "held": held, "positions": held,
            "params": {"scale": 2_000_000, "weak_neg_days": 5},
            "totals": {"scale": 2_000_000}, "sector_ready": True})
        _patch_params(p, param_store, PARAMS_BASE)
        r = csvc.plan_command({"cmd_type": "LIQUIDATE_ALL",
                               "command_id": "CMD_20260903_0001",
                               "params": {"window_tdays": 1, "confirm": "YES"}})
    assert got["routed"] == [], "用户命令驱动的清仓不该经过提议分流"
    assert r["status"] == cs.ST_EXECUTING, r
    acts = {x["action"] for x in got["plans"]}
    assert "EXIT" in acts, got["plans"]
    exits = [x for x in got["plans"] if x["action"] == "EXIT"]
    assert exits[0]["ts_code"] == "600000.SH" and exits[0]["qty"] == 3000, exits


# ================================================================ 五, 源码守卫
@case("源码守卫·分流里不许再出现「卖出方向短路强制入队」那种写法")
def _():
    path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
                        "app", "services", "proposal_service.py")
    with open(path, encoding="utf-8") as f:
        src = f.read()
    # 只看真正的赋值行, 说明性注释里可以原样引用旧写法 (那段注释本身就是这次修复的说明)
    bad = [ln for ln in src.splitlines()
           if ln.strip().startswith('auto_exec = (side == "sell") or')]
    assert not bad, f"卖出方向又被放回或运算左边, 强制入队会再次失效: {bad}"
    assert "src_why = ae.source_confirm_why(c.get(\"source\"))" in src, "来源判定不见了"
    assert 'auto_exec = (not force_queue) and (side == "sell" or autonomy == AUTONOMY_FULL)' \
        in src, "强制入队不再是一票否决"


# ================================================================ 跑
def main():
    ok = fail = 0
    for name, fn in RESULTS:
        try:
            fn()
            ok += 1
            print(f"  PASS  {name}")
        except Exception as e:
            fail += 1
            print(f"  FAIL  {name}: {type(e).__name__}: {e}")
            traceback.print_exc()
    print(f"\n通过 {ok} 例, 失败 {fail} 例")
    if fail:
        sys.exit(1)
    print(f"ALL PASS ({ok} cases)")


if __name__ == "__main__":
    main()