# -*- 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": "公司深度：质地好…"}]} st = pf.parse_logic_states(payload)["SH601126"] 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": "SH601126", "total_qty": 0}] with _Patch() as p: p(lss, "state_map", lambda: {"SH601126": 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())