约 {{ p.qty }} 股 · 参考价 {{ (p.hard_numbers||{}).price==null ? '—' : (p.hard_numbers||{}).price }}
@@ -1602,6 +1606,9 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
证据截止 {{ s.row.logic_now.as_of || '—' }}；{{ s.row.logic_now.settle_note }}
+
+
+ {{ s.row.company.line }}；失效条件：{{ s.row.company.invalidation }}
{{ (s.row.logic_now && s.row.logic_now.text) || '今天没有读数' }}
diff --git a/scripts/run_tests.py b/scripts/run_tests.py
index 532d949..2071dec 100644
--- a/scripts/run_tests.py
+++ b/scripts/run_tests.py
@@ -56,6 +56,8 @@
强制入队对卖出同样有效/研判不可用的减持入队/研究证据走弱
来源必定交人裁决/风控高置信卖出与命令清仓不经提议分流/
到价提议挂着时止损照落并作废该提议 (17 例)
+ test_batch23_units.py 公司质地与建议方案接入 (2026-09-09 接入方案): 解析层收公司深度两键/整句送研判可关/建议
+ 方案矩阵与研判补档/持仓页公司块。
test_batch22_units.py 逻辑状态四态接入 (2026-09-07 第三件): 解析层收逻辑状态/硬数字带它而研判
白名单不收/判决候选而逻辑存疑强制交人/持仓存疑停增持侧/研究走弱
减持默认关且开了必交人/同轮减持优先级/早上取回与映射新鲜度/策略
@@ -110,7 +112,7 @@ SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py",
"test_batch13_units.py", "test_batch14_units.py", "test_batch15_units.py",
"test_batch16_units.py", "test_batch17_units.py",
"test_batch18_units.py", "test_batch19_units.py", "test_batch20_units.py",
- "test_batch21_units.py", "test_batch22_units.py",
+ "test_batch21_units.py", "test_batch22_units.py", "test_batch23_units.py",
"test_page_enum_guard.py",
"test_wiring.py"]
diff --git a/scripts/test_batch23_units.py b/scripts/test_batch23_units.py
new file mode 100644
index 0000000..2bac6e3
--- /dev/null
+++ b/scripts/test_batch23_units.py
@@ -0,0 +1,175 @@
+# -*- 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())