tradingSystem/scripts/test_batch23_units.py

# -*- coding: utf-8 -*-
"""公司质地与建议方案接入 (2026-09-09 接入方案, 台账 053)。全部离线, 不连库不调模型。

  1. 解析层收公司深度两键 (固定键、缺质地归 None), 候选选择透传, 硬数字带它。
  2. 整句进送研判白名单; 开关关掉就不送; 原值字典不送。
  3. 逐票逻辑状态解析收公司深度两键; 持仓页 company_view 一行。
  4. 建议方案矩阵: 六种情形各一例; 研判补档; 开关开着时评估参数按建议。
"""
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 advice_service as adv                                 # noqa: E402
from app.services import judge as jd                                           # noqa: E402
from app.services import logic_state_service as lss                            # noqa: E402
from app.services import param_store                                           # noqa: E402
from app.services import plan_feed as pf                                       # noqa: E402

RESULTS = []


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


class _Patch:
    def __init__(self): self._saved = []
    def __enter__(self): return self
    def __call__(self, obj, name, value):
        self._saved.append((obj, name, getattr(obj, name))); setattr(obj, name, value)
    def __exit__(self, *a):
        for obj, name, v in reversed(self._saved):
            setattr(obj, name, v)


def _cr(overall="好", valuation="贵", doubt=False):
    return {"period": "2026Q1", "overall": overall, "valuation": valuation, "doubt_hard": doubt,
            "groups": {"回报与护城河": "好", "盈余质量与财务安全": "好", "成长与含金量": "好"},
            "grade_counts": {"好": 5, "中": 3, "差": 1, "证据不足": 1}, "thesis": "质地好但估值贵", "confidence": "中",
            "invalidation": "下一期看现金流", "antithesis": "覆盖少", "catalyst": "无法判断", "brain_status": "ok",
            "ran_at": "2026-09-08", "age_days": 1, "report_url": "http://192.168.16.178:8000/company/601126.SH/review", "extra": 1}


@case("解析层收公司深度两键: 固定键、缺质地归 None、候选透传、硬数字带它")
def test_passthrough():
    row = pf._rows([{"code": "601126.SH", "rank": 1, "score": 210, "company_review": _cr(), "company_review_text": "公司深度：质地好…",
                     "company_gate": ""}], "main")[0]
    assert row["company_review"]["overall"] == "好" and "extra" not in row["company_review"], row["company_review"]
    assert row["company_review"]["grade_counts"]["好"] == 5 and row["company_review"]["groups"]["回报与护城河"] == "好"
    assert row["company_review_text"].startswith("公司深度")
    empty = pf._rows([{"code": "601126.SH", "rank": 1, "score": 210, "company_review": {"overall": None}}], "main")[0]
    assert empty["company_review"] is None and empty["company_review_text"] is None
    plan = {"date": "2026-09-09", "main": [row], "observe": []}
    item = pf.select_candidates(plan, top_n=10)["items"][0]
    assert item["company_review"]["valuation"] == "贵"
    c = {**item, "price": 10.0}
    params = {"scale": 1_000_000, "stock_target_default": 0.06, "batch_split": (0.5, 0.25, 0.25)}
    with _Patch() as p:
        p(ae, "check_all_caps", lambda **kw: [])
        p(ae, "_new_name_ctx", lambda caps, c: caps)
        cand, why = ae.eval_open(c, params, {"names_count": 0, "max_names": 10}, 500_000)
    hn = cand["hard_numbers"]
    assert hn["company_review"]["overall"] == "好" and hn["company_review_text"] == row["company_review_text"], why


@case("整句送研判且开关可关; 原值字典不送")
def test_judge_whitelist():
    hn = {"price": 10.0, "company_review": _cr(), "company_review_text": "公司深度：质地好…", "advice": {"tier": "减半仓"}}
    with _Patch() as p:
        p(param_store, "get_bool", lambda k, d=None: True)
        sent = jd._judge_hard_numbers("OPEN", hn)
    assert sent["company_review_text"] == hn["company_review_text"] and "company_review" not in sent and "advice" not in sent, sent
    with _Patch() as p:
        p(param_store, "get_bool", lambda k, d=None: False if k == "PMS_JUDGE_COMPANY_TEXT" else True)
        sent2 = jd._judge_hard_numbers("OPEN", hn)
    assert "company_review_text" not in sent2, sent2
    assert "company_review_text" in jd.OPEN_JUDGE_KEYS


@case("逐票逻辑状态解析收公司深度; 持仓页 company_view 一行")
def test_positions_view():
    payload = {"date": "2026-09-09", "states": [{"code": "SH601126", "state": "逻辑成立", "why": None, "as_of": "2026-09-08",
                                                  "reasons": ["a"], "company_review": _cr(), "company_review_text": "公司深度：质地好…"}]}
    parsed = pf.parse_logic_states(payload)
    key, st = next(iter(parsed.items()))            # 键是 PMS 自己的代码形态 (normalize_code), 不猜
    assert st["company_review"]["overall"] == "好" and st["company_review_text"].startswith("公司深度")
    v = lss.company_view(st)
    assert v["overall"] == "好" and "质地好" in v["line"] and v["invalidation"] == "下一期看现金流" and v["report_url"]
    assert lss.company_view({"state": "逻辑成立"}) is None
    rows = [{"ts_code": key, "total_qty": 0}]
    with _Patch() as p:
        p(lss, "state_map", lambda: {key: st})
        lss.decorate_positions(rows)
    assert rows[0]["company"]["overall"] == "好"


@case("建议方案矩阵六情形与研判补档")
def test_advice_matrix():
    params = {"stock_target_default": 0.06, "batch_split": (0.5, 0.25, 0.25)}
    val_good = {"neut": 12.0, "pess": 9.5, "price": 10.0}      # 上行 20%、下行 5% → 赔率 4
    val_bad = {"neut": 10.5, "pess": 8.0, "price": 10.0}       # 上行 5%、下行 20% → 赔率 0.25
    a = adv.advise({"company_review": _cr("好", "中"), "logic_state": {"state": "逻辑成立"}, "valuation": val_good}, params)
    assert a["tier"] == "标准仓" and a["splits"] == (0.5, 0.25, 0.25) and abs(a["target_pct"] - 0.06) < 1e-9, a
    b = adv.advise({"company_review": _cr("好", "贵"), "logic_state": {"state": "逻辑成立"}, "valuation": val_good}, params)
    assert b["tier"] == "减半仓" and b["splits"] == (0.5, 0.5) and abs(b["target_pct"] - 0.03) < 1e-9, b
    c = adv.advise({"company_review": _cr("中", "中"), "logic_state": {"state": "逻辑成立"}, "valuation": val_good}, params)
    assert c["tier"] == "标准仓" and c["splits"] == (0.3, 0.35, 0.35) and "二十日头看多" in "".join(c["conditions"]), c
    d = adv.advise({"company_review": _cr("好", "中"), "logic_state": {"state": "逻辑成立"}, "valuation": val_bad}, params)
    assert d["tier"] == "试探仓" and abs(d["target_pct"] - 0.01) < 1e-9 and "人工确认" in "".join(d["conditions"]), d
    e = adv.advise({"company_review": _cr("好", "中"), "logic_state": {"state": "逻辑成立"}, "valuation": val_good,
                    "pricing_state": {"state": "高位兑现"}}, params)
    assert e["delay_first"] and e["tier"] == "标准仓", e
    f = adv.advise({"company_review": _cr("差", "中", True), "logic_state": {"state": "逻辑存疑"}}, params)
    assert f["tier"] == "不建" and f["target_pct"] == 0.0
    g = adv.advise({"company_review": None, "valuation": val_good}, params)
    assert g["tier"] == "标准仓" and "沿用机械方案" in "".join(g["why"]) and g["splits"] == (0.5, 0.25, 0.25)
    # 研判补档
    a2 = adv.apply_judge(a, {"verdict": "APPROVE", "confidence": 45}, 60)
    assert a2["tier"] == "试探仓" and a2["needs_user_confirm"] and "研判" in "".join(a2["why"]), a2
    a3 = adv.apply_judge(a, {"verdict": "APPROVE", "confidence": 80}, 60)
    assert a3["tier"] == "标准仓"
    a4 = adv.apply_judge(a, {"verdict": "UNAVAILABLE", "confidence": None}, 60)
    assert a4["tier"] == "试探仓"
    # 评估参数
    p2 = adv.params_for(b, params)
    assert abs(p2["stock_target_default"] - 0.03) < 1e-9 and p2["batch_split"] == (0.5, 0.5)
    assert adv.params_for(f, params) is params
    assert "建议方案：减半仓" in adv.text(b) and "机械方案 6.0%" in adv.text(b)


@case("扫描时建议方案进硬数字, 开关开着时按建议仓位评估")
def test_scan_open_uses_advice():
    item = {"ts_code": "601126.SH", "code": "601126.SH", "rank": 1, "score": 210, "price": 10.0, "verdict": "候选",
            "company_review": _cr("好", "贵"), "company_review_text": "x", "logic_state": {"state": "逻辑成立"},
            "valuation": {"neut": 12.0, "pess": 9.5, "price": 10.0}}
    params = {"scale": 1_000_000, "stock_target_default": 0.06, "batch_split": (0.5, 0.25, 0.25), "plan_by_advice": True,
              "open_route_by_verdict": False, "open_route_by_logic": False}
    seen = {}
    def fake_eval(c, p, ctx, left):
        seen["target"] = p.get("stock_target_default"); seen["split"] = p.get("batch_split")
        return {"ts_code": c["ts_code"], "action": ae.A_OPEN, "qty": 100, "hard_numbers": {}}, None
    with _Patch() as p:
        p(ae, "eval_open", fake_eval)
        out = ae.scan_open(candidates=[item], params=params, caps={"names_count": 0, "max_names": 10}, room_amt=500_000, slots=5)
    assert abs(seen["target"] - 0.03) < 1e-9 and seen["split"] == (0.5, 0.5), (seen, out.get("skipped"))
    cands = out["candidates"]
    hn = cands[0]["hard_numbers"]
    assert hn["advice"]["tier"] == "减半仓" and hn["advice_text"].startswith("建议方案") and hn["plan_by_advice"] is True


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


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