tradingSystem/scripts/test_batch29_units.py

# -*- coding: utf-8 -*-
"""三源合议 · 工作包二接入下单链路 (2026-09-11 方案第五节, 台账 006/009)。全部离线, 不连库。

工作包二的三块「大脑」(纯逻辑 / 装配服务 / 仓位矩阵) 由第二十八批守; 本批守它们**接进下单链路**
那一段, 也就是交接文档第五节的活。每一组都带一条「开关关掉逐字回旧」——这是「加不改」的机器证明。

  A scan_open 合议分流: 跳过/观察不占名额、交人强制确认、放行照常、缺合议退回旧路、开关关掉回旧。
  B 定档改用 advise_v2 与六键进硬数字, 且六键一个都不进 judge.OPEN_JUDGE_KEYS。
  C 持仓增持门 consensus_gate_why 矩阵与 scan 端到端 (减持不受影响、开关关掉回旧)。
  D 装配服务 _attach_consensus 与持仓 decorate_positions。
  E 候选按质地排序 (质地看多分桶在前、技术面不参与、观察档进池开关、开关关掉回旧)。
  F 三个新参数登记。
"""
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 judge as jd              # noqa: E402
from app.services import plan_feed as pf          # 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 params(**kw):
    p = {"scale": 2_000_000.0, "stock_target_default": 0.06,
         "batch_split": (0.5, 0.25, 0.25), "min_lot_merge": True}
    p.update(kw)
    return p


def caps(**kw):
    c = {"scale": 2_000_000.0, "portfolio_cap": 0.70, "stock_cap": 0.08,
         "max_names": 20, "portfolio_mv": 0.0, "names_count": 0,
         "stock_mv": 0.0, "is_new_name": False, "sector": None,
         "sector_names": 0, "sector_mv": 0.0, "sector_names_map": {}, "sector_mv_map": {},
         "sector_max_names": 4, "sector_max_ratio": 0.40, "cash_reserve": 0.0,
         "sector_source_ready": True, "cash_avail": None, "cash_source": "estimate"}
    c.update(kw)
    return c


def cand(code, price=10.0, score=100.0, rank=1, **kw):
    d = {"ts_code": code, "price": price, "score": score, "sector": kw.pop("sector", None),
         "theme": kw.pop("theme", None), "tier": kw.pop("tier", "强传导"),
         "upside": kw.pop("upside", 0.3), "heat": kw.pop("heat", 0.5),
         "rank": rank, "bucket": "main", "src": "plan_api"}
    d.update(kw)
    return d


def with_con(c, route, *, fund="看多", tech="看多", timing="看多", reason=None):
    """给候选手挂一份合议 (直接控制路由, 不经装配服务) —— 单测 scan_open 分流本身。"""
    c = dict(c)
    con = {"route": route, "route_reason": reason or ("合议判" + route),
           "direction": "看多", "votes": {"fund": fund, "tech": tech, "timing": timing},
           "strength": "强", "reason": "三票 → 看多"}
    c["consensus"] = con
    c["consensus_blocks"] = {"fund": {"stance": fund}, "tech": {"stance": tech},
                             "timing": {"stance": timing}, "consensus": con}
    c["consensus_hard"] = {"fund_stance": {"stance": fund}, "tech": {"stance": tech},
                           "tech_text": "技术面理由", "timing": {"stance": timing},
                           "consensus": con, "consensus_text": "合议理由"}
    return c


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


def _blocks(direction="中性", tech="中性", fund="中性"):
    return {"consensus": {"direction": direction}, "tech": {"stance": tech}, "fund": {"stance": fund}}


# ================================================================ A scan_open 合议分流
@case("A 分流·开关关掉逐字回旧: 挂了跳过的合议也照出候选, 与不挂时一字不差")
def _():
    base = cand("600000.SH")
    r_plain = ae.scan_open(candidates=[dict(base)], params=params(consensus_route=False),
                           caps=caps(), room_amt=1_400_000, slots=20)
    r_con = ae.scan_open(candidates=[with_con(base, "跳过")], params=params(consensus_route=False),
                         caps=caps(), room_amt=1_400_000, slots=20)
    assert r_plain == r_con, "关掉合议时, 挂不挂合议、路由是不是跳过, 结果必须一字不差"
    assert len(r_plain["candidates"]) == 1


@case("A 分流·跳过: 不产候选、不占名额 (后面的票补上)")
def _():
    cs = [with_con(cand("600000.SH", score=200), "跳过", reason="没有买方评析"),
          with_con(cand("000001.SZ", score=100), "放行")]
    r = ae.scan_open(candidates=cs, params=params(consensus_route=True),
                     caps=caps(), room_amt=1_400_000, slots=1)
    assert [c["ts_code"] for c in r["candidates"]] == ["000001.SZ"], "跳过的票不该占掉唯一名额"
    assert any(s["ts_code"] == "600000.SH" and "买方评析" in s["why"] for s in r["skipped"])


@case("A 分流·观察: 不产候选、带处置词 wait_tech")
def _():
    r = ae.scan_open(candidates=[with_con(cand("600000.SH"), "观察", reason="等技术面开口")],
                     params=params(consensus_route=True), caps=caps(), room_amt=1_400_000, slots=20)
    assert r["candidates"] == []
    s = next(s for s in r["skipped"] if s["ts_code"] == "600000.SH")
    assert s.get("disp") == ae.DISP_WAIT_TECH and "技术面" in s["why"]


@case("A 分流·交人: 照常产候选、占名额, 但强制人工确认并带原因")
def _():
    r = ae.scan_open(candidates=[with_con(cand("600000.SH"), "交人", reason="基本面看空、技术面看多，试探仓")],
                     params=params(consensus_route=True), caps=caps(), room_amt=1_400_000, slots=20)
    assert len(r["candidates"]) == 1
    c = r["candidates"][0]
    assert c.get("needs_user_confirm") is True and "试探" in (c.get("confirm_why") or "")


@case("A 分流·放行: 正常产候选, 不强制确认")
def _():
    r = ae.scan_open(candidates=[with_con(cand("600000.SH"), "放行")],
                     params=params(consensus_route=True), caps=caps(), room_amt=1_400_000, slots=20)
    assert len(r["candidates"]) == 1 and not r["candidates"][0].get("needs_user_confirm")


@case("A 分流·缺合议但开关开: 退回旧路照常产出 (装不上不拦)")
def _():
    r = ae.scan_open(candidates=[cand("600000.SH")], params=params(consensus_route=True),
                     caps=caps(), room_amt=1_400_000, slots=20)
    assert len(r["candidates"]) == 1


# ================================================================ B advise_v2 定档与六键
@case("B 定档·合议开着用 advise_v2 (矩阵, version=2); 关着用旧 advise (version=1)")
def _():
    on = ae.scan_open(candidates=[with_con(cand("600000.SH"), "放行", fund="看多", tech="看多")],
                      params=params(consensus_route=True), caps=caps(), room_amt=1_400_000, slots=20)
    adv_on = on["candidates"][0]["hard_numbers"]["advice"]
    assert adv_on.get("version") == "2", "合议开着定档应改用 advise_v2"
    off = ae.scan_open(candidates=[cand("600000.SH")], params=params(consensus_route=False),
                       caps=caps(), room_amt=1_400_000, slots=20)
    assert off["candidates"][0]["hard_numbers"]["advice"].get("version") == "1"


@case("B 六键·合议开着六键进硬数字")
def _():
    r = ae.scan_open(candidates=[with_con(cand("600000.SH"), "放行")],
                     params=params(consensus_route=True), caps=caps(), room_amt=1_400_000, slots=20)
    hn = r["candidates"][0]["hard_numbers"]
    for k in ("fund_stance", "tech", "tech_text", "timing", "consensus", "consensus_text"):
        assert k in hn, f"六键 {k} 应进硬数字"


@case("B 六键·一个都不进 judge.OPEN_JUDGE_KEYS (基本面技术面不重判)")
def _():
    six = {"fund_stance", "tech", "tech_text", "timing", "consensus", "consensus_text"}
    assert six.isdisjoint(set(jd.OPEN_JUDGE_KEYS)), "六键混进送研判白名单会让决策系统重判基本面/技术面"


@case("B 六键·送研判裁剪后确实不含这六键 (机器证明)")
def _():
    r = ae.scan_open(candidates=[with_con(cand("600000.SH"), "放行")],
                     params=params(consensus_route=True), caps=caps(), room_amt=1_400_000, slots=20)
    sent = jd._judge_hard_numbers("OPEN", r["candidates"][0]["hard_numbers"])
    for k in ("fund_stance", "tech", "tech_text", "timing", "consensus", "consensus_text"):
        assert k not in sent


# ================================================================ C 持仓增持门 (纯逻辑矩阵)
@case("C 增持门·合议看空停三类 (回踩补足/盈利加仓/补仓)")
def _():
    for act in (ae.A_FILL, ae.A_ADD, ae.A_DCA):
        assert ae.consensus_gate_why(_blocks(direction="看空"), act), act


@case("C 增持门·技术面看空停回踩补足与盈利加仓, 不停补仓")
def _():
    b = _blocks(direction="中性", tech="看空", fund="中性")
    assert ae.consensus_gate_why(b, ae.A_FILL) and ae.consensus_gate_why(b, ae.A_ADD)
    assert ae.consensus_gate_why(b, ae.A_DCA) is None


@case("C 增持门·基本面看空停补仓, 不停回踩补足与盈利加仓")
def _():
    b = _blocks(direction="中性", tech="中性", fund="看空")
    assert ae.consensus_gate_why(b, ae.A_DCA)
    assert ae.consensus_gate_why(b, ae.A_FILL) is None and ae.consensus_gate_why(b, ae.A_ADD) is None


@case("C 增持门·三方非看空不设门; 缺块或无读数不设门 (弃权)")
def _():
    b = _blocks(direction="看多", tech="看多", fund="看多")
    for act in (ae.A_FILL, ae.A_ADD, ae.A_DCA):
        assert ae.consensus_gate_why(b, act) is None
    assert ae.consensus_gate_why(None, ae.A_FILL) is None
    assert ae.consensus_gate_why({}, ae.A_DCA) is None
    assert ae.consensus_gate_why(_blocks("无读数", "无读数", "无读数"), ae.A_DCA) is None


@case("C 增持门·scan 端到端: 合议看空真拦下 FILL")
def _():
    p = pos(price=9.8, cushion_pct=-0.02, support_ref=9.7, market_value=58_800)
    p["consensus_blocks"] = _blocks(direction="看空", tech="看空", fund="看空")
    r = ae.scan(positions=[p], params={**PARAMS, "tech_gate_increase": True},
                market={p["ts_code"]: mkt()})
    assert not any(c["side"] == "buy" for c in r["candidates"]), "合议看空时不该出增持候选"
    assert any(s["action"] == ae.A_FILL and "看空" in s["why"] for s in r["skipped"])


@case("C 增持门·scan 端到端: 减持侧 (保垫减仓) 不受合议看空影响")
def _():
    p = pos(cushion_peak=0.08, cushion_pct=0.04)
    p["consensus_blocks"] = _blocks(direction="看空", tech="看空", fund="看空")
    r = ae.scan(positions=[p], params={**PARAMS, "tech_gate_increase": True},
                market={p["ts_code"]: mkt()})
    assert any(c["action"] == ae.A_TRIM for c in r["candidates"]), "减持不该被增持门拦"


@case("C 增持门·开关关掉逐字回旧: 挂了合议看空也照出 FILL")
def _():
    base = pos(price=9.8, cushion_pct=-0.02, support_ref=9.7, market_value=58_800)
    r_plain = ae.scan(positions=[dict(base)], params=PARAMS, market={base["ts_code"]: mkt()})
    p2 = dict(base)
    p2["consensus_blocks"] = _blocks(direction="看空", tech="看空", fund="看空")
    r_con = ae.scan(positions=[p2], params=PARAMS, market={base["ts_code"]: mkt()})  # PARAMS 无 tech_gate_increase
    assert [c["action"] for c in r_plain["candidates"]] == [c["action"] for c in r_con["candidates"]]
    assert any(c["action"] == ae.A_FILL for c in r_con["candidates"])


# ================================================================ D 装配服务
_ASM_P = {"fund_required": True, "fund_stale_days": 120, "fund_consensus_good_min": 8,
          "consensus_route": True, "timing_stale_tdays": 3}


def _cr(overall="好"):
    return {"overall": overall, "valuation": "便宜",
            "grade_counts": {"好": 8, "中": 2, "差": 0}, "groups": {}, "doubt_hard": False, "age_days": 0}


@case("D 装配·decorate_positions 给持仓挂 consensus_blocks, 评析从 company_reviews 带出")
def _():
    from app.services import consensus_service as csv
    rows = [{"ts_code": "600000.SH"}, {"ts_code": "000001.SZ"}]
    csv.decorate_positions(rows, nightly={}, tech_states={},
                           company_reviews={"600000.SH": _cr("好")}, params=_ASM_P)
    assert rows[0]["consensus_blocks"]["fund"]["stance"] == "看多"
    assert rows[1]["consensus_blocks"]["fund"]["stance"] == "无读数"  # 没带评析 = 无读数, 绝不折看空


@case("D 装配·decorate_positions 单行异常不塌 (缺 ts_code 抹掉键)")
def _():
    from app.services import consensus_service as csv
    rows = [{"ts_code": ""}]
    csv.decorate_positions(rows, nightly={}, tech_states={}, company_reviews={}, params=_ASM_P)
    assert "consensus_blocks" not in rows[0]


# ================================================================ E 候选按质地排序
def _row(code, score, overall=None, rank=1):
    r = {"ts_code": code, "name": "股票", "score": score, "rank": rank, "tier": "强传导",
         "bucket": "main"}
    if overall:
        r["company_review"] = {"overall": overall, "valuation": "便宜", "age_days": 0,
                               "grade_counts": {}, "groups": {}}
    return r


@case("E 排序·开关关掉逐字回旧: 纯按分数降序")
def _():
    plan = {"date": "2026-09-11", "main": [_row("600000.SH", 100), _row("000001.SZ", 90, "好")],
            "observe": []}
    out = pf.select_candidates(plan, rank_by_quality=False)
    assert [x["ts_code"] for x in out["items"]] == ["600000.SH", "000001.SZ"]


@case("E 排序·按质地: 质地看多分桶在前, 低分看多排在高分无评析之前")
def _():
    plan = {"date": "2026-09-11", "main": [_row("600000.SH", 100), _row("000001.SZ", 90, "好")],
            "observe": []}
    out = pf.select_candidates(plan, rank_by_quality=True)
    assert [x["ts_code"] for x in out["items"]] == ["000001.SZ", "600000.SH"]


@case("E 排序·同桶内仍按分数 (质地差与无评析同桶, 高分在前)")
def _():
    plan = {"date": "2026-09-11",
            "main": [_row("600000.SH", 80, "差"), _row("000001.SZ", 90)], "observe": []}
    out = pf.select_candidates(plan, rank_by_quality=True)
    assert [x["ts_code"] for x in out["items"]] == ["000001.SZ", "600000.SH"]


@case("E 排序·观察档进池开关: 默认关不进, 开了质地看多的进")
def _():
    plan = {"date": "2026-09-11", "main": [_row("600000.SH", 100)],
            "observe": [_row("000001.SZ", 50, "好"), _row("000002.SZ", 60)]}
    off = pf.select_candidates(plan, rank_by_quality=False, observe_if_fund_bull=False)
    assert [x["ts_code"] for x in off["items"]] == ["600000.SH"]
    on = pf.select_candidates(plan, rank_by_quality=False, observe_if_fund_bull=True)
    codes = {x["ts_code"] for x in on["items"]}
    assert "000001.SZ" in codes and "000002.SZ" not in codes  # 只进质地看多的观察行


# ================================================================ F 参数登记
@case("F 参数·三个新开关登记进 RUNTIME_EXTRA, 类型与默认对")
def _():
    for k in ("PMS_TECH_GATE_INCREASE", "PMS_PLAN_RANK_BY_QUALITY", "PMS_PLAN_OBSERVE_IF_FUND_BULL"):
        assert k in ps.RUNTIME_EXTRA, k
        assert ps.RUNTIME_EXTRA[k][1] is bool
    assert ps.RUNTIME_EXTRA["PMS_TECH_GATE_INCREASE"][0] is True
    assert ps.RUNTIME_EXTRA["PMS_PLAN_RANK_BY_QUALITY"][0] is True
    assert ps.RUNTIME_EXTRA["PMS_PLAN_OBSERVE_IF_FUND_BULL"][0] is False  # 守宁缺毋滥, 默认关


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