公司质地接入：公司深度取数、逻辑状态第五路、判决收敛只降不升、卡与计划两键、候选单表列、复盘分组、逐票接口回两键、单测（台账 053）

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

api.py

@@ -21,6 +21,7 @@ from fastapi import FastAPI, HTTPException, Request
 from fastapi.responses import PlainTextResponse
 
 import card
+import sources
 import config
 import db
 import logic_state_daily
@@ -196,4 +197,13 @@ def logic_state_lookup(codes: str | None = None, code: str | None = None,
             item["valuation_text"] = card.valuation_view(v)
     except Exception as e:  # noqa: BLE001 —— 估值算不出不该拖垮状态查询
         _access.warning("logic_state valuation failed date=%s err=%r", d, e)
+    # 公司深度两键随状态一起回（2026-09-09 接入方案）：持仓页"证据还在不在"旁显示质地与失效条件。读不到整批不带。
+    try:
+        crs = sources.company_reviews(list(keyed), d)
+        for item in out:
+            cr = crs.get(item["code"])
+            item["company_review"] = cr
+            item["company_review_text"] = card.company_review_view(cr)
+    except Exception as e:  # noqa: BLE001
+        _access.warning("logic_state company_review failed date=%s err=%r", d, e)
     return {"date": d, "count": len(out) + len(bad), "states": out + bad}

card.py

@@ -73,7 +73,8 @@ def _num(v) -> float | None:
 
 def judge(ev: dict[str, Any], *, start_pct: float = 3.0,
           accum_max_age: int = 30, neg_tol: float = 0.0,
-          logic_stale_days: int = 90, require_started: bool = True) -> dict[str, Any]:
+          logic_stale_days: int = 90, require_started: bool = True,
+          company_review: dict | None = None, quality_gate: bool = True) -> dict[str, Any]:
     """一只票的证据字典 -> 判决字典。
 
     ev 里认的键（缺键按缺失处理，不报错）：
@@ -198,8 +199,19 @@ def judge(ev: dict[str, Any], *, start_pct: float = 3.0,
                                        covered=bool(ev.get("covered")), upside=upside,
                                        neg_tol=neg_tol))
 
+    # 公司质地（2026-09-09 接入方案第四节）：判决收敛只降不升——质地差且判决为候选，降为关注交人裁决；
+    # 质地好不升档；证据不足、没报告不动。三门槛一确认三风险本身不动。
+    company_gate = ""
+    cr = company_review if isinstance(company_review, dict) else None
+    if quality_gate and cr and cr.get("overall") == "差" and verdict == VERDICT_CANDIDATE:
+        verdict = VERDICT_WATCH
+        why_q = "；".join((cr.get("quality_reasons") or [])[:2]) or "基座代码合成的质地档为差"
+        company_gate = f"公司质地评为差（{why_q}），候选降为关注交人裁决"
+        basis = f"{basis}；{company_gate}"
+
     return {"verdict": verdict, "reasons": reasons, "missing": missing, "risk": risk,
             "gates": gates, "confirm": confirm, "failed_gates": failed, "basis": basis,
+            "company_gate": company_gate,
             "logic": logic_lines(ev.get("logic")),
             "logic_quality": logic_quality(ev.get("logic"))}
 
@@ -484,6 +496,40 @@ def industry_catalyst_short(lst) -> str:
     return f"{x.get('magnitude')}·{x.get('horizon')} {str(x.get('title') or '')[:20]}"
 
 
+def company_review_view(cr) -> str:
+    """公司深度的整句（2026-09-09 接入方案）：质地档与三类、十视角计数、买方论点一句、失效条件、报告链接。只展示。"""
+    if not isinstance(cr, dict) or not cr.get("overall"):
+        return "公司深度：暂无（数据基座还没出这家的报告）"
+    g = cr.get("groups") or {}
+    gc = cr.get("grade_counts") or {}
+    parts = [f"公司深度（{cr.get('period')} 期报告，{cr.get('age_days') if cr.get('age_days') is not None else '?'} 天前）："
+             f"质地{cr['overall']}（回报{g.get('回报与护城河', '缺')}、盈余{g.get('盈余质量与财务安全', '缺')}、成长{g.get('成长与含金量', '缺')}），"
+             f"估值{cr.get('valuation') or '无'}；十视角 好{gc.get('好', 0)}/中{gc.get('中', 0)}/差{gc.get('差', 0)}/证据不足{gc.get('证据不足', 0)}"]
+    if cr.get("thesis"):
+        parts.append(f"买方论点：{cr['thesis']}（置信{cr.get('confidence') or '?'}）")
+    if cr.get("invalidation"):
+        parts.append(f"失效条件：{cr['invalidation']}")
+    if cr.get("antithesis"):
+        parts.append(f"反方：{cr['antithesis']}")
+    if cr.get("report_url"):
+        parts.append(f"报告 {cr['report_url']}")
+    return "；".join(parts)
+
+
+def company_review_short(cr) -> str:
+    if not isinstance(cr, dict) or not cr.get("overall"):
+        return "—"
+    gc = cr.get("grade_counts") or {}
+    return f"{cr['overall']}·{cr.get('valuation') or '无'} 好{gc.get('好', 0)}中{gc.get('中', 0)}差{gc.get('差', 0)}"
+
+
+def company_review_label(cr) -> str:
+    """复盘分组用：好 / 中 / 差 / 无（证据不足与缺席都归无）。"""
+    if not isinstance(cr, dict):
+        return "无"
+    return cr.get("overall") if cr.get("overall") in ("好", "中", "差") else "无"
+
+
 def news_short(nv) -> str:
     if not isinstance(nv, dict) or not nv.get("items"):
         return "—"

config.py

@@ -279,6 +279,11 @@ JUDGEMENT_PREV_LOOKBACK_DAYS = int(os.environ.get("JUDGEMENT_PREV_LOOKBACK_DAYS"
 LOGIC_STATE_TABLE = os.environ.get("LOGIC_STATE_TABLE", "t_akg_logic_state_daily")
 # 抗抖动的两个天数（计划日数），与 logic_state.settle 的默认值同源，一次定死、不按复盘读数回调：
 # 其余方向的迁移要连续几个计划日同向；退出逻辑存疑要连续几个计划日不再存疑（进入存疑即刻成立）。
+# 公司质地（2026-09-09 接入方案第四节，台账 053）：三处开关默认开。
+LOGIC_PATH_COMPANY = os.environ.get("LOGIC_PATH_COMPANY", "1") == "1"            # 逻辑状态第五路
+CARD_COMPANY_QUALITY_GATE = os.environ.get("CARD_COMPANY_QUALITY_GATE", "1") == "1"  # 候选卡"质地差降一档"
+COMPANY_REVIEW_STALE_DAYS = int(os.environ.get("COMPANY_REVIEW_STALE_DAYS", "120"))   # 报告运行日超过它算陈旧
+AKG_API_BASE = os.environ.get("AKG_API_BASE", "http://192.168.16.178:8000")           # 报告链接前缀
 LOGIC_SETTLE_CONFIRM_DAYS = int(os.environ.get("LOGIC_SETTLE_CONFIRM_DAYS", "2"))
 LOGIC_SETTLE_EXIT_DAYS = int(os.environ.get("LOGIC_SETTLE_EXIT_DAYS", "3"))
 # 读上一次落定态与近几日原始态时往前看几个自然日。要盖住国庆这种连休八天再加两个周末的空档；

docs/个股深度评析接入方案_2026-09-09.md

@@ -0,0 +1,180 @@
+# 个股深度评析接入方案（2026-09-09，已批准）
+
+> 接《个股深度评析方案_2026-09-09》第一批。批准后落盘为两个仓库的 docs/个股深度评析接入方案_2026-09-09.md，台账记 053。
+> 09-09 三条拍板已吸收：**公司质地参与决策，三处全开**（逻辑状态第五路含存疑触发、候选卡"质地差降一档"、评议文字送新建仓研判）；**云端评议每晚上限二十家自动跑**；对话入口的 skill 以后再说。
+> 用户 09-09 原话："盈余质量是不是太狭隘……这部分实际上也是主观选股观点的主要部分……不要浪费我们这个阶段的成果。"这份方案照此把公司质地当成主观选股的第四条证据线，不当一条风险文本。
+
+---
+
+## 零、一句话
+
+主观选股是行业逻辑、公司质地、估值预期差、时点四件事的合成。前三件里公司质地一直空着，第一批把它算出来了；这一步把它接进系统：**质地由代码合成，进逻辑状态、进判决收敛、进 PMS 提议卡与持仓页、进复盘分组；云端大模型做买方式评议，写论点、反方、失效条件，给人看、送研判，不定判决。**
+
+---
+
+## 一、这一步买什么
+
+1. **公司质地成为第四条证据线。** 逻辑状态原有四路（研报论断、产业研判、券商行动、公司事件缺失），加第五路"公司质地"；候选卡判决收敛加一条"质地差降一档"；持仓票财报期一换质地转差，PMS 现成的停增持与强制确认自动接上。
+2. **买方评议补上研究深度。** 云端评议按论点卡七字段出论点、证据编号、催化剂与时间窗、预期差、失效条件、置信度、版本，外加逐视角复核、反方论点、无法判断清单。一致性检查表第七行"研究深度"由此有了数据源。
+3. **复盘能回答"质地有没有用"。** 周报按公司质地四档分组，两周后决定要不要把门槛再往前提。
+
+---
+
+## 二、原则
+
+- **质地由代码合成，模型只写字。** 十个视角的判据归三类再合成一档；评议不参与合成。
+- **加，不改。** 第五路是并列新增；判决收敛只降不升（质地好不升档，证据不足不惩罚）；三处各有开关，记台账；评析表、判决三门槛、择时决策系统一个不动。
+- **描述性估值。** 估值与预期差单列"贵、中、便宜"只展示，不进质地、不进判决。
+- **评议不能引入表外事实。** 匿名化（不给公司名与代码）、证据只认编号、数字逐个回查、未核比例超三成记半成品。
+- **费用有护栏。** 每晚上限二十家，单家输出上限，截断不重试，索引表记用量，日检报读数。
+
+---
+
+## 三、公司质地怎么合成（代码，一次定死，单测钉住）
+
+十个视角先归三类，每类评好、中、差、证据不足（该类能判的视角不足一个即证据不足）：
+
+| 类 | 用到的视角与指标 | 好 | 差 |
+|---|---|---|---|
+| 回报与护城河 | 巴菲特与芒格、特里·史密斯、彭曼、杜邦 | 至少两个视角为好且没有差 | 至少两个视角为差 |
+| 盈余质量与财务安全 | 盈余质量三分、F 分数、薛云奎财务栏 | 盈余质量三分为好且 F 分数不低于 6 | 盈余质量三分为差，或 F 分数不高于 3 且应计比率超过 10% |
+| 成长与含金量 | 薛云奎经营栏与业绩栏、收入五年复合增速、经营现金流对净利润 | 经营栏为好、业绩栏不为差、增速不低于 8%、现金转化不低于 0.8 | 经营栏为差，或现金转化低于 0.5 |
+
+合成：任一类为差则质地差；三类都不差且回报类为好则质地好；两类以上证据不足则质地证据不足；其余为中。估值单列：滚动市盈率十年分位与隐含市盈率对一致预期中位，分贵、中、便宜，只展示。
+
+三家试跑公司按这套规则：四方股份三类都好，质地**好**，估值贵；澜起科技回报类中（巴菲特差、特里好）、其余中，质地**中**，估值贵；立讯精密 F 分数差但盈余质量好、应计 0.2%，不触发差，质地**中**，估值中。三家都不会因为这一步被降档，第五路只有四方股份给出"向好"。
+
+---
+
+## 四、接进决策的三处（三处全开，各有开关）
+
+**逻辑状态第五路"公司质地"（选股系统 logic_state.py）。** 信号：好为向好，中为平稳，差为转弱，没报告、证据不足或报告陈旧（最新期比当前财报期落后一期以上，或运行日超过 120 天）为缺失。速度归"慢"：单独不能定态，逻辑强化仍必须含研报论断或产业研判那一路。存疑触发：质地差且盈余质量类为差，与券商下修、论断翻转并列进 DOUBT_TRIGGERS。截止日用报告运行日。开关 LOGIC_PATH_COMPANY，默认开。
+
+**候选卡判决收敛（card.py）。** 三门槛一确认三风险不动；在判决算完之后加一条：质地差且判决为候选，降为关注交人裁决，依据句写"公司质地评为差：盈余质量类差（应计 x、F 分数 y）"；质地好不升档；证据不足不动。开关 CARD_COMPANY_QUALITY_GATE，默认开。
+
+**评议文字送新建仓研判（PMS judge.OPEN_JUDGE_KEYS 加 company_review_text）。** 照 09-02 送判决与理由的先例：送的是买方论点一句、置信度、失效条件，让研判拿自己的量价结论对照；质地档与评价计数也随硬数字进提议卡与账本。开关 PMS_JUDGE_COMPANY_TEXT，默认开。
+
+**持仓票。** 逐票逻辑状态接口随状态回 company_review 两键（照安全边际的先例），PMS 持仓页"证据还在不在"旁多一行"公司质地 · 失效条件"。财报期换后基座重跑报告，质地转差走第五路进存疑：停增持、强制确认，减持开关照旧默认关。
+
+**复盘。** 周报候选单与档位表各按"公司质地=好/中/差/无"分组。
+
+---
+
+## 五、云端评议层（数据基座）
+
+- 新模块 synthesis.py：把侧车编号化成指标 [M..] 与判据 [C..]，不带公司名与代码；一次调用 brain 档位（当前 deepseek-v4-pro），JSON 输出；三条纪律照环节评析的写法（截断不重试、非截断空答案只重试一次、单家输出总额上限）。
+- 输出：逐视角复核（哪些判据在这家公司情形下失真，偏严偏松，还剩什么有效信息）、综合三句（回报来源、含金量、财务安全）、买方论点七字段（证据只写编号）、反方论点最多三条、无法判断清单最多六条。
+- 后处理：编号回填、催化剂无编号改写为"无法判断"、预期不引一致预期编号记警告、字段截断、版本由代码填。
+- 数字回查：文本里的数字逐个匹配允许集合（指标值的小数、百分数、亿元、万元、倍、阈值数、一致预期各档），年份与十二以内小整数白名单；未命中在报告里就地标记，未核比例超三成记半成品。
+- 落地：侧车加 synthesis 键；报告第五段替换占位文字（段头写模型、版本、用量、回查结果，段尾编号对照表）；索引表加 summary、brain_used、brain_status、synthesis_version 四列，摘要含质地档、三类评价、十视角计数、论点一句、置信度、反方一句、催化剂、无法判断条数、未核数字条数、报告路径。
+- 指纹加评议版本：没评议的旧行开了开关自然重跑；评议成功后不重花钱；失败次日再试，同一家同一期失败三次放弃。
+- 护栏：每晚上限二十家、每晚输出总额上限、单家上限；批任务按候选名次分配评议名额；索引记用量；日检加一行。
+- 费用估算：输入约一万 token、输出两千到六千，一家约 0.15 元、最坏 0.3 元；二十家一晚 3 到 6 元；一百二十家首轮 20 到 40 元，之后只随财报期与版本变动。墙钟每家一到三分钟，二十家在 00:45 到 05:30 窗口内串行跑得完。
+
+---
+
+## 六、接入后 PMS 的整体流程（以三家试跑公司为例）
+
+**每晚 00:45（数据基座）。** 对当天候选名单前一百二十家跑报告：纯代码部分几秒一家；云端评议按名次给前二十家，指纹没变的不重跑。四方股份、澜起科技、立讯精密这类已有报告的票，只有财报期换或版本升才重算。
+
+**每早 07:10（选股系统）。** 出计划时每张卡多三样：公司质地（四方好、澜起中、立讯中）、估值标签（四方贵、澜起贵、立讯中）、评议摘要（论点一句、置信度、失效条件、报告链接）。逻辑状态五路合成：四方股份第五路向好，若研报论断或产业研判也向好则逻辑强化，卡内序提前，判决不变；澜起与立讯第五路平稳，不影响。判决收敛：假设某只候选的质地评为差（盈余质量类差），判决从候选降为关注，依据句写明原因，交人裁决。
+
+**每早 08:40（PMS 拉计划）。** 候选照旧进建仓方案；关注强制人工确认；仅展示不出提议。提议卡上多一行"公司深度：好 4 中 3 差 1 · 论点：……（置信中）· 失效：下一期财报看经营现金流对净利润与毛利率 · 报告"。
+
+**盘中（新建仓研判）。** 请求体多带评议一句与失效条件；择时决策系统的研判拿它与量价两个头对照后给通过、驳回或不可用。四方股份这类质地好、估值贵的票，研判可能给"通过但注意估值分位 92%"；驳回行照旧记把握度。人在提议卡上看到质地、论点、反方、失效条件一起裁决。
+
+**持仓期。** 每早逐票逻辑状态刷新，持仓页"证据还在不在"旁显示质地与失效条件。财报期换后，若某持仓票的质地从中转差且盈余质量类为差，第五路转弱达到存疑条件：停增持、新增仓强制确认，减持开关默认关；人在持仓页看到"公司质地转差：应计 x、F 分数 y"这句原因。
+
+**周五复盘。** 候选单按质地四档分组读数；两周后拿读数决定要不要把"质地差降一档"升级成门槛、要不要开减持。
+
+---
+
+## 六之二、再进一步：综合信息给出建议建仓方案（09-09 用户追问后加）
+
+可以，而且应该做，但分两层，落在两处，都不是数据基座。数据基座定过不给买卖建议；建仓是 PMS 的事，时点是择时决策系统的事。
+
+**第一层，规则合成的建仓建议（PMS，代码，本方案第三段一起做）。** 输入六样：公司质地（好中差）、估值标签（贵中便宜）、赔率与预期差、逻辑状态（成立强化存疑无法判断）、研判结论与把握度、定价状态与量价两个头。输出一份"建议方案"：
+
+| 情形 | 仓位档 | 分批 | 附加条件 |
+|---|---|---|---|
+| 质地好、估值中或便宜、逻辑成立或强化、研判通过 | 标准仓（现行 6%） | 5 成、2.5 成、2.5 成 | 第二批要求二十日头不看空 |
+| 质地好、估值贵 | 减半仓 | 两批各半 | 第二批等下一期财报失效条件未触发再加 |
+| 质地中 | 标准仓 | 第一批只 3 成，余下两批各 3.5 成 | 后两批要求二十日头看多且定价状态不是高位兑现 |
+| 研判把握度低或不可用、赔率低于 1 | 试探仓（1%） | 一批 | 只在人工确认后下 |
+| 定价状态高位兑现 | 不变 | 首批推迟 | 等回撤到支撑再下第一批 |
+| 质地差、逻辑存疑 | 不建 | — | 已在判决与分流里拦住 |
+
+止损定性与风控卖出不动（那是设计内的自动止损）；持有期限按质地写"长期持有类"或"事件驱动类"，只作标签。这份建议先与现行机械方案并排显示在提议卡上；开关 PMS_PLAN_BY_ADVICE 打开后，建议方案直接作为提议的仓位档与分批（模拟侧，正式库不部署），账本记两份方案便于复盘对照。建议默认在模拟侧打开，两周复盘后看"按建议建的仓"与"按机械方案建的仓"的名单级读数再定正式口径。
+
+**采纳之后，择时决策系统照旧参与每一批的下单时点，不参与仓位大小。** 现行链路不改：提议采纳后，PMS 按方案分批，每一批下单前都调择时决策系统的盘中择时接口（读昨夜支撑压力推出买入区间：支撑下沿百分之一到支撑加三成幅度），现价在区间内才下单且限价挂区间上沿，高于上沿不追、低于下沿不买，择时拿不到就退 PMS 内置口径；资金、上限、一手、可卖日这些硬检查留在 PMS。建议方案改的只是"建多少、分几批、加批要什么条件"：后两批的附加条件（二十日头不看空、定价状态不是高位兑现、失效条件未触发）由 PMS 在每批触发前用最新一份昨夜结论检查，过了再交盘中择时定时点。持仓期的止损止盈、关注条件仍是择时决策系统的。所以完整的分工是：数据基座出质地与评议，选股系统出判决与逻辑状态，PMS 出建仓方案与建议档位，择时决策系统出每一批的时点与持仓期的风控。
+
+**第二层，研判写的建仓建议（择时决策系统，09-15 量价研判链判收后再做）。** 新建仓研判的输出多一个键：仓位倾向（标准、减半、试探、不建）、分批节奏、入场条件（价格区间、量能确认）、一句理由。它拿的是同一批输入加盘中量价，与规则版并排显示，分歧交人。观察期内不动那条链的提示词，所以排在判收之后。
+
+**不做的。** 不让数据基座的评议给仓位与价格；不把建议方案接到正式库；不做回测挑参数，仓位档矩阵一次定死记台账，只按复盘读数换档位。
+
+## 七、分批与判收
+
+**第一段（数据基座，先于模型）。** 第一批五处小修（可得年份不足判"算不了"、删一季度失真的最新期现金流判据、盈利稳定改用年度每股收益、主营构成说明改为表头行取法、净现金公司回报率虚高加说明）；质地合成模块与摘要；索引表新列；视角版本升级；单测。
+怎么验：合成数据单测全绿；三家试跑质地档与本方案第三节一致。
+
+**第二段（云端评议）。** synthesis.py、报告第五段、runner 接法、任务与接口参数、日检行、脚本 --with-brain；三家试跑。
+怎么验：三家评议的完成原因都是正常结束（不截断）；数字回查未命中不超过一成；论点与反方每条带编号；催化剂三家都应是"无法判断"；重跑不带 force 为指纹跳过；索引表记用量。人读三份评议，记"比代码汇总多说了什么"。
+
+**第三段（选股系统与 PMS 接线，含建议建仓方案）。** 选股系统取数、卡上整句与短句、第五路、判决收敛、复盘分组、逐票接口回两键、单测；PMS 解析、硬数字、送研判白名单、提议卡一行、持仓页一行、规则合成的建议方案（六之二的矩阵，模块 app/services/advice_service.py，纯函数加单测）、提议卡并排显示两份方案、账本记两份、开关 PMS_PLAN_BY_ADVICE 模拟侧默认开；接口容器重启与镜像重建经同意、避开开盘。
+怎么验：实时计划里三家的两键与质地档正确；PMS 拉取解析到；送研判的请求体含评议一句；持仓页有那一行；提议卡有两份方案且仓位档与矩阵一致（用三家试跑公司各造一条提议核）；复盘周报有分组行；全套测试通过。
+
+**第四段（启用）。** 评议开关与每晚上限打开；次日核日检行、提议卡、账本。台账 053。
+
+**判收。** 四条：一，候选名单里的公司报告与评议不落后最新财报期一个月；二，三处开关打开后一致性检查表第七行"研究深度"由不一致改为一致；三，两周复盘周报有质地分组读数；四，人读十份评议至少七份"比代码汇总多说了东西"。是否升级门槛、是否开减持，判收后另开拍板。
+
+---
+
+## 八、明确不做
+
+不让模型定质地、不让评议改判决；不做规范性估值、不给目标价；不改三门槛与三风险；不动择时决策系统代码；不做对话入口 skill（以后再说）。
+
+---
+
+## 九、风险
+
+一，第五路把持仓票推进存疑会触发停增持：触发条件已收紧到"质地差且盈余质量类差"，且减持开关默认关。二，评议截断：思考型模型的思考计入输出，上限先设 16k，首晚截断率高再升到 24k。三，报告陈旧：财报期换后一个月内报告未更新即算缺失，不拿旧质地当新证据。四，基座与选股系统的时区：夜间上限按最近二十小时计，不按日历日。五，一致性：质地规则只在基座一处算，选股系统只读结果，避免两边各算一套。
+
+---
+
+## 附录甲 · 实现骨架（可不读）
+
+**数据基座（astock-kg）。**
+- `backend/app/analysis/company_review/quality.py`（新）：`classify(lens_results, metrics) -> {overall, groups: {回报与护城河, 盈余质量与财务安全, 成长与含金量}, valuation_label, reasons}`；纯函数，单测钉住第三节的规则。
+- `backend/app/analysis/company_review/synthesis.py`（新）：`build_input(sidecar)`（编号化、允许集合）、`run(sidecar, client, ...)`（一次调用、三条纪律、后处理）、`numbers_check(obj, allowed)`；常量 SYNTHESIS_VERSION、系统提示词、上限。
+- `backend/app/analysis/company_review/summary.py`（新）：`build(code, period, quality, lens_results, syn) -> dict`，不超过 1.5 KB。
+- `runner.py`：签名加 `with_brain`、`brain_client`；指纹加评议版本；夜间用量四道闸；落库带 summary 与四个新列。`report.py`：第五段按 synthesis 渲染，段尾编号对照。
+- `company_review_store.py`：`ALTER TABLE ... ADD COLUMN IF NOT EXISTS summary JSONB / brain_used / brain_status / synthesis_version`；`brain_usage(hours=20)`、`latest_many(codes)`、`recent_codes(days, brain_only)`；`summary()` 多回评议读数。
+- `lenses/graham.py`、`buffett.py`、`xue_siwei.py`、`penman.py`、`greenblatt.py`、`metrics.py`：第七节第一段的五处小修；REVIEW_VERSION 升到 2026.09-2。
+- `workers/tasks.py`：`company_review(..., with_brain)`、`company_review_batch(..., with_brain=None, max_brain=None)`；环境变量 AKG_COMPANY_REVIEW_BRAIN（默认关，试跑满意后置 1）、AKG_COMPANY_REVIEW_BRAIN_MAX（20）、AKG_COMPANY_REVIEW_BRAIN_TOKENS_NIGHTLY（500000）。
+- `main.py`：两个触发接口加 with_brain 与 max_brain；新增 `GET /company/{ts_code}/review/summary`。`daily_check.py` 加评议读数。
+- `scripts/company_review_run.py` 加 `--with-brain`。
+- 测试：`backend/tests/company_review/test_synthesis.py`（编号化稳定、数字回查八例、编号回填、催化剂改写、假客户端三种结局、报告第五段、摘要四态）、`test_quality.py`（三类合成规则与三家试跑读数）、`test_company_review.py` 补小修六例。
+
+**选股系统（akg-factor-bridge）。**
+- `sources.company_reviews(codes, *, read_pg=None)`：读 company_review_runs 最新 ok 行（DISTINCT ON ts_code），带 age_days 与报告链接；表读不到整体缺席。
+- `card.company_review_view/short/label`；`card.judge` 末尾加质地差降一档（开关 CARD_COMPANY_QUALITY_GATE）。
+- `logic_state.py`：PATH_COMPANY、SPEED 慢、DOUBT_TRIGGERS 加一条、`from_company(review, ds)`；`plan._logic_inputs` 取数、`_logic_state_of` 装第五路（开关 LOGIC_PATH_COMPANY）。
+- `plan.py`：卡与顶层行加 company_review / company_review_text / quality；候选单表加"公司质地"列。`plan_review.py`：分组"公司质地=…"。`api.py` 逐票逻辑状态接口随状态回两键。
+- 测试：`test_events_pricing.py::test_company_review`、`test_logic_state.py` 第五路与存疑触发、`test_card.py` 降档规则。
+
+**PMS（tradingSystem）。**
+- `plan_feed.py`：`_company_review_or_none`、`_rows` 与 `select_candidates` 两键、`parse_logic_states` 收两键；`action_engine.eval_open` 硬数字；`judge.OPEN_JUDGE_KEYS` 加 company_review_text（开关 PMS_JUDGE_COMPANY_TEXT）；`logic_state_service.decorate_positions` 加 company 块；`index.html` 提议卡一行、持仓页一行；`scripts/test_batch23_units.py`；`BIONIC_PMS_INTERFACE.md` 记新键。
+
+**验证命令（178，容器内）。**
+```bash
+docker compose exec -T -e PYTHONPATH=/app backend python /app/scripts/company_review_run.py --code 601126.SH --force --with-brain --print
+```
+预期：报告第五段带"云端评议"段头，论点与反方每条带 [M..]/[C..]，数字回查未命中数为 0 或个位数，索引表新行 brain_status=ok。
+
+## 附录乙 · 第一批小修清单（可不读）
+
+- 格雷厄姆"盈利稳定"、巴菲特"每年不低于 15%"：可得年份不足阈值时判"算不了（history_short）"。
+- 薛云奎业绩栏删"经营现金流对净利润（最新期）"，非年报期在数据说明里写明年初至今口径。
+- 盈利稳定改用财务指标表年度每股收益为正的年数，重构表缺年度行不再断档。
+- 主营构成说明改为"取表头行（营业收入合计）"，两套分类相差的旧文字删除。
+- 净现金公司在数据说明与彭曼、格林布拉特两个视角加"净经营资产回报率偏高"的说明。

logic_state.py

@@ -76,15 +76,17 @@ PATH_CLAIM = "研报论断"
 PATH_JUDGE = "产业研判"
 PATH_BROKER = "券商行动"
 PATH_EVENT = "公司事件"
+PATH_COMPANY = "公司质地"      # 第五路（2026-09-09 接入方案第四节）：数据基座个股深度评析合成的质地档
 
 # 每路的节奏，合成规则四要用。慢信号不能单独定态。
-SPEED = {PATH_JUDGE: "慢", PATH_CLAIM: "中", PATH_BROKER: "中", PATH_EVENT: "日"}
+SPEED = {PATH_JUDGE: "慢", PATH_CLAIM: "中", PATH_BROKER: "中", PATH_EVENT: "日", PATH_COMPANY: "慢"}
 
 # 只有这两条路径的转弱允许整体落到逻辑存疑，别的转弱最多到无法判断。
 DOUBT_TRIGGERS = {
     PATH_BROKER: "每股收益预测下修且覆盖机构收缩，两个条件同时满足",
     PATH_CLAIM: "论断方向跨期翻转（先前利好、最新一条转利空）",
     PATH_JUDGE: "采信倾向从偏多迁移到偏空，且材料指纹变过",
+    PATH_COMPANY: "公司质地评为差且盈余质量类为差（基座代码合成，不由模型定）",
 }
 
 # 采信倾向的受控四值（数据基座抽取端的提示词写死的，见 backend/workers/tasks.py）。
@@ -329,6 +331,31 @@ def from_broker(now, prev, *, as_of: str | None = None, min_firms: int = 2,
 # 丁路：公司事件（目前不可用）
 # ============================================================================
 
+def from_company(review, ds: str, *, stale_days: int = 120) -> dict:
+    """第五路：数据基座个股深度评析合成的质地档归一成一路信号（2026-09-09 接入方案第四节）。
+
+    好 → 向好；中 → 平稳；差 → 转弱（质地差且盈余质量类差达到逻辑存疑的进入条件，hard）；
+    没报告、证据不足、报告陈旧（运行日距数据日超过 stale_days）→ 缺失。截止日用报告运行日。
+    慢信号：单独不能定态；逻辑强化仍必须含研报论断或产业研判那一路。"""
+    if not isinstance(review, dict) or not review.get("overall"):
+        return signal(PATH_COMPANY, SIG_NONE, why="数据基座还没有这家公司的深度评析报告")
+    overall = str(review.get("overall"))
+    as_of = _s(review.get("ran_at"))
+    age = review.get("age_days")
+    if age is not None and age > stale_days:
+        return signal(PATH_COMPANY, SIG_NONE, why=f"报告运行日 {as_of} 距数据日 {age} 天，超过 {stale_days} 天算陈旧")
+    if overall == "证据不足":
+        return signal(PATH_COMPANY, SIG_NONE, as_of=as_of, why="报告评为证据不足")
+    groups = review.get("groups") or {}
+    why = (f"质地{overall}：回报{groups.get('回报与护城河', '缺')}、盈余{groups.get('盈余质量与财务安全', '缺')}、"
+           f"成长{groups.get('成长与含金量', '缺')}（报告期 {review.get('period')}）")
+    if overall == "好":
+        return signal(PATH_COMPANY, SIG_UP, as_of=as_of, coverage=1, why=why)
+    if overall == "差":
+        return signal(PATH_COMPANY, SIG_DOWN, as_of=as_of, coverage=1, why=why, hard=bool(review.get("doubt_hard")))
+    return signal(PATH_COMPANY, SIG_FLAT, as_of=as_of, coverage=1, why=why)
+
+
 def from_events(*_a, **_k) -> dict:
     """这一路目前不可用，恒定返回缺失。
 
@@ -361,7 +388,7 @@ def compose(paths, *, strong_needs=(PATH_CLAIM, PATH_JUDGE)) -> dict:
 
     if not usable:
         return _out(STATE_UNKNOWN, WHY_THIN, rows, usable, missing,
-                    ["四路证据一路都不可用"] + miss_note)
+                    ["各路证据一路都不可用"] + miss_note)
 
     ups = [p for p in usable if p["signal"] == SIG_UP]
     downs = [p for p in usable if p["signal"] == SIG_DOWN]

plan.py

@@ -186,7 +186,7 @@ def _segments_of(ds: str) -> dict:
 
 
 def _logic_state_of(k: str, evd: dict, seg_of: dict, seg_view: dict,
-                    broker: dict, ds: str, full_logic=None, seg_hist=None) -> dict:
+                    broker: dict, ds: str, full_logic=None, seg_hist=None, company=None) -> dict:
     """一只票的逻辑状态四态。四路各自归一，再按合成规则合成。
 
     乙路的取法：一只票可能挂在多个被指向的环节上，取第一个有行业观点的那个。
@@ -203,7 +203,12 @@ def _logic_state_of(k: str, evd: dict, seg_of: dict, seg_view: dict,
     c = broker.get(k) or logic_state.signal(
         logic_state.PATH_BROKER, logic_state.SIG_NONE,
         why="两个等长窗口里算不出可比的每股收益预测")
-    return logic_state.compose([a, b, c, logic_state.from_events()])
+    paths = [a, b, c, logic_state.from_events()]
+    # 第五路 公司质地（2026-09-09 接入方案第四节，开关 LOGIC_PATH_COMPANY）：慢信号，单独不定态；
+    # 质地差且盈余质量类差才达到存疑的进入条件。关掉开关这一路根本不进合成。
+    if config.LOGIC_PATH_COMPANY:
+        paths.append(logic_state.from_company((company or {}).get(k), ds, stale_days=config.COMPANY_REVIEW_STALE_DAYS))
+    return logic_state.compose(paths)
 
 
 def _logic_inputs(codes: list, ds: str) -> dict:
@@ -238,6 +243,8 @@ def _logic_inputs(codes: list, ds: str) -> dict:
         "broker": sources.broker_actions(codes, ds, rows=broker_rows),
         "seg_of": _segments_of(ds),
         "hist": logic_state_daily.history(ds, codes=None if len(codes) > 50 else codes),
+        # 公司质地（2026-09-09 接入方案）：基座个股深度评析索引表的摘要，第五路与卡上一行共用。读不到为空字典。
+        "company": sources.company_reviews(codes, ds),
     }
 
 
@@ -254,7 +261,7 @@ def logic_states_for(codes: list, ds: str) -> dict:
     out = {}
     for k in codes:
         raw = _logic_state_of(k, {"logic": []}, inp["seg_of"], inp["seg_view"], inp["broker"], ds,
-                              full_logic=inp["logic_full"].get(k) or [], seg_hist=inp["seg_hist"])
+                              full_logic=inp["logic_full"].get(k) or [], seg_hist=inp["seg_hist"], company=inp["company"])
         out[k] = logic_state_daily.settle_one(k, raw, inp["hist"].get(k))
     return out
 
@@ -291,6 +298,7 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series,
     # logic_state.apply_to_card，是单调的——强化只能提前卡内序、永远不升判决。
     inp = _logic_inputs(codes, ds)
     logic_full, seg_view, seg_hist = inp["logic_full"], inp["seg_view"], inp["seg_hist"]
+    company = inp.get("company") or {}
     broker, seg_of, hist = inp["broker"], inp["seg_of"], inp["hist"]
     logic = {k: v[:config.LOGIC_CLAIMS_PER_STOCK] for k, v in logic_full.items()}
     # 安全边际三情景（2026-09-07 第四件）：按同一预测期的每股收益与市盈率预测算悲观、中性、乐观
@@ -332,7 +340,7 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series,
                "accum_age": n.get("accum_age"), "y_signal": n.get("signal"),
                "stale_snapshot": stale, "logic": logic.get(k) or []}
         raw_state = _logic_state_of(k, evd, seg_of, seg_view, broker, ds,
-                                    full_logic=logic_full.get(k) or [], seg_hist=seg_hist)
+                                    full_logic=logic_full.get(k) or [], seg_hist=seg_hist, company=company)
         # 抗抖动（2026-09-07 第三件桥侧前置）：配上这只票在逐票日频表里的近日行，落定今天的状态。
         # 进入逻辑存疑即刻成立，退出要连续几天不再存疑，其余迁移要连续几天同向；表里没有
         # 这只票时落定态就是原始态。落定态进 state，原始态另存 raw_state，两个都发给下游。
@@ -341,7 +349,8 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series,
                        accum_max_age=config.CARD_ACCUM_MAX_AGE,
                        neg_tol=config.UPSIDE_NEG_TOLERANCE,
                        logic_stale_days=config.LOGIC_STALE_DAYS,
-                       require_started=config.CARD_REQUIRE_STARTED)
+                       require_started=config.CARD_REQUIRE_STARTED,
+                       company_review=company.get(k), quality_gate=config.CARD_COMPANY_QUALITY_GATE)
         cards[k] = {
             **j, "logic_state": state,
             "valuation": valuation.get(k), "valuation_text": card.valuation_view(valuation.get(k)),
@@ -349,6 +358,7 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series,
             "pricing_state": card.pricing_state(ev_fields.get(k)),
             "news": news.get(k), "news_text": card.news_view(news.get(k)),
             "industry_catalyst": ind_cat.get(k), "industry_catalyst_text": card.industry_catalyst_view(ind_cat.get(k)),
+            "company_review": company.get(k), "company_review_text": card.company_review_view(company.get(k)),
             "theme": theme, "n_sources": n_sources, "chain_fit": evd["chain_fit"],
             "started_source": "moved_view" if mv else None,
             "logic_claims": evd["logic"],
@@ -486,6 +496,10 @@ def collect(date: str | None = None, top: int = 20, obs_top: int = 10,
                      # 行业催化（2026-09-08 台账 051）：环节级，原值给程序，整句给人；只展示不进判决。
                      industry_catalyst=c.get("industry_catalyst"),
                      industry_catalyst_text=c.get("industry_catalyst_text"),
+                     # 公司深度（2026-09-09 接入方案）：质地档与评议摘要，原值给程序，整句给人；
+                     # 质地进逻辑状态第五路与判决收敛（只降不升），评议文字只给人看与送研判。
+                     company_review=c.get("company_review"), company_review_text=c.get("company_review_text"),
+                     company_gate=c.get("company_gate"),
                      card={"pct0": c.get("pct0"), "net_z": c.get("net_z"),
                            "heat_chg": c.get("heat_chg"), "accum": c.get("accum"),
                            "night": c.get("night"), "gates": c.get("gates"),
@@ -725,8 +739,8 @@ def render_md(d: dict) -> str:
         L.append("（今日无候选——候选为空不是故障：环节没被指向、成员没启动或没有明确吸筹，都会为空。）")
     else:
         L.append("| # | 代码 | 名称 | 环节 | 源数 | 当日涨幅 | 吸筹 | 预期空间 | 安全边际（悲观下行/中性上行，赔率） "
-                 "| 催化事件 | 定价状态 | 理由 | 因果论断（出处） |")
+                 "| 公司质地 | 催化事件 | 定价状态 | 理由 | 因果论断（出处） |")
-        L.append("|---|------|------|------|------|----------|------|----------|----------|----------|----------|------|------------------|")
+        L.append("|---|------|------|------|------|----------|------|----------|----------|----------|----------|----------|------|------------------|")
         for r in cands:
             c = r.get("card") or {}
             ac = c.get("accum") or {}
@@ -734,6 +748,7 @@ def render_md(d: dict) -> str:
             L.append(f"| {r['rank']} | {r['code']} | {r['name'] or '—'} | {ev_.get('theme') or '—'} "
                      f"| {ev_.get('n_sources') or '—'} | {_fmt_pct0(c.get('pct0'))} "
                      f"| {_fmt_accum(ac)} | {_fmt_pct(r.get('upside'))} | {card.valuation_short(r.get('valuation'))} "
+                     f"| {card.company_review_short(r.get('company_review'))} "
                      f"| {card.events_short(r.get('events'))} | {card.pricing_short(r.get('pricing_state'))} "
                      f"| {'；'.join(r.get('reasons') or [])} | {_fmt_logic(r.get('logic'))} |")
         L.append("")
@@ -744,6 +759,9 @@ def render_md(d: dict) -> str:
                  "定价状态按最近一次事件日的事件前涨幅、事件日跳空、收盘位置与量比归成四情形："
                  "价格发现（事件前没涨、事件日放量收高）、趋势延续（事件前已涨、事件日仍放量收高）、"
                  "高位兑现（事件前大涨、事件日放量冲高回落）、震荡消化（其余）。两者都只展示、不进判决。")
+        L.append("公司质地是数据基座个股深度评析按代码合成的档（回报与护城河、盈余质量与财务安全、成长与含金量三类），"
+                 "后面是估值标签（贵中便宜，只展示）与十个大师视角的好中差计数。质地进逻辑状态第五路；质地差的候选降为关注交人；"
+                 "评议文字见候选卡整句与报告链接。")
     L.append("")
     segs = d.get("segments_pointed") or []
     L.append(f"## 关注环节（今日被传导指向的 {len(segs)} 个环节：定位对不对看这里，挑票看候选单）")

plan_review.py

@@ -60,6 +60,7 @@ import re
 import pandas as pd
 
 import common
+import card
 import config
 import db
 import plan
@@ -449,6 +450,15 @@ def run(since: str, until: str | None, horizons: tuple, start: str, out_dir: str
                     if s:
                         rows.append({"date": day, "h": h, "list": lst_name, "group": f"催化事件={label}",
                                      "regime_post": regime_post, "regime_pre": regime_pre, **s})
+            # 公司质地四档（2026-09-09 接入方案）：好 / 中 / 差 / 无，候选单与档位表各分一次。
+            cq_of = {r["code"]: card.company_review_label(r.get("company_review")) for r in main_rows + obs_rows}
+            for lst_name, lst_codes in (("候选单", cands), ("档位表", list(cq_of))):
+                for label in ("好", "中", "差", "无"):
+                    codes = [c for c in lst_codes if cq_of.get(c, "无") == label]
+                    s = summarize(ret, codes, base_all, base_main, caps, cap_base)
+                    if s:
+                        rows.append({"date": day, "h": h, "list": lst_name, "group": f"公司质地={label}",
+                                     "regime_post": regime_post, "regime_pre": regime_pre, **s})
             # 行业催化有无（2026-09-08 台账 051）：环节级催化，只作观察分组；材料未到时全部落"无"。
             ind_of = {r["code"]: bool(r.get("industry_catalyst")) for r in main_rows + obs_rows}
             for lst_name, lst_codes in (("候选单", cands), ("档位表", list(ind_of))):

sources.py

@@ -1168,3 +1168,63 @@ def catalysts_for_codes(seg_of: dict, cat_by_seg: dict, *, limit: int = 3) -> di
             out[k] = lst[:limit]
     return out
 
+
+# ---------------------------------------------------------------------------
+# 公司深度评析（2026-09-09 接入方案，台账 053）：数据基座 company_review_runs 索引表最新一行的摘要。
+# 质地由基座代码合成，这里只读不算；报告正文在对象存储，链接给人。表读不到整体缺席不断产。
+# ---------------------------------------------------------------------------
+def company_reviews(codes, ds: str, *, read_pg=None) -> dict:
+    """{前缀码: {"period", "review_version", "overall", "groups", "valuation", "doubt_hard", "quality_reasons",
+    "grade_counts", "grade_line", "thesis", "confidence", "invalidation", "antithesis", "catalyst", "brain_status",
+    "ran_at", "age_days", "report_url"}}；没报告的票不在字典里。"""
+    rp = read_pg or db.read_pg
+    want = sorted({c for c in (codes or []) if c})
+    if not want:
+        return {}
+    dots = [_to_dot(c) for c in want]
+    try:
+        df = rp("SELECT DISTINCT ON (ts_code) ts_code, period, review_version, grades, summary, brain_used, brain_status, ran_at "
+                "FROM company_review_runs WHERE status = 'ok' AND ts_code = ANY(%s) ORDER BY ts_code, ran_at DESC", (dots,))
+    except Exception as e:  # noqa: BLE001
+        logging.getLogger("bridge.company_review").warning("company_reviews 读不到: %r", e)
+        return {}
+    out: dict = {}
+    for r in df.to_dict("records"):
+        sm = r.get("summary")
+        if isinstance(sm, str):
+            try:
+                sm = json.loads(sm)
+            except ValueError:
+                sm = None
+        sm = sm or {}
+        grades = r.get("grades")
+        if isinstance(grades, str):
+            try:
+                grades = json.loads(grades)
+            except ValueError:
+                grades = {}
+        ran = r.get("ran_at")
+        try:
+            ran_day = ran.date() if hasattr(ran, "date") else dt.date.fromisoformat(str(ran)[:10])
+            age = (dt.date.fromisoformat(ds) - ran_day).days
+        except (TypeError, ValueError, AttributeError):
+            ran_day, age = None, None
+        k = _prefix_any(str(r.get("ts_code")))
+        brain = sm.get("brain") or {}
+        out[k] = {"period": r.get("period"), "review_version": r.get("review_version"),
+                  "overall": sm.get("overall"), "groups": sm.get("groups"), "valuation": sm.get("valuation"),
+                  "doubt_hard": bool(sm.get("doubt_hard")), "quality_reasons": sm.get("quality_reasons") or [],
+                  "grade_counts": sm.get("grade_counts") or _count_grades(grades), "grade_line": sm.get("grade_line"),
+                  "thesis": sm.get("thesis"), "confidence": sm.get("confidence"), "invalidation": sm.get("invalidation"),
+                  "antithesis": sm.get("antithesis"), "catalyst": sm.get("catalyst"),
+                  "brain_status": r.get("brain_status") or brain.get("status"), "brain_used": bool(r.get("brain_used")),
+                  "ran_at": ran_day.isoformat() if ran_day else None, "age_days": age,
+                  "report_url": f"{config.AKG_API_BASE}{sm.get('report_path') or '/company/' + str(r.get('ts_code')) + '/review'}"}
+    return out
+
+
+def _count_grades(grades) -> dict:
+    out = {"好": 0, "中": 0, "差": 0, "证据不足": 0}
+    for v in (grades or {}).values():
+        out[v] = out.get(v, 0) + 1
+    return out

test_company_quality.py

@@ -0,0 +1,113 @@
+"""公司质地接入（2026-09-09 接入方案第四节）：取数、第五路信号与合成、判决收敛只降不升、卡上文字。离线，不连库。"""
+import datetime as dt
+import json
+import sys
+import types
+
+for _n in ("pandas", "pymysql", "psycopg"):
+    if _n not in sys.modules:
+        try:
+            __import__(_n)
+        except Exception:  # noqa: BLE001
+            _m = types.ModuleType(_n)
+            if _n == "pandas":
+                _m.DataFrame = type("DataFrame", (), {})
+            sys.modules[_n] = _m
+
+import card          # noqa: E402
+import logic_state as ls  # noqa: E402
+import sources       # noqa: E402
+
+
+def t(name, cond):
+    print(("  ok  " if cond else "  FAIL") + name)
+    assert cond, name
+
+
+class _DF:
+    def __init__(self, rows): self._rows = rows
+    def to_dict(self, _kind): return self._rows
+
+
+def _review(overall="好", doubt=False, ran="2026-09-08", groups=None):
+    return {"overall": overall, "groups": groups or {"回报与护城河": "好", "盈余质量与财务安全": "好", "成长与含金量": "好"},
+            "doubt_hard": doubt, "ran_at": ran, "age_days": 1, "period": "2026Q1",
+            "grade_counts": {"好": 5, "中": 3, "差": 1, "证据不足": 1}, "valuation": "贵",
+            "thesis": "质地好但估值贵", "confidence": "中", "invalidation": "下一期看现金流", "quality_reasons": ["回报类好"],
+            "report_url": "http://192.168.16.178:8000/company/601126.SH/review"}
+
+
+def test_sources():
+    calls = []
+    def rp(sql, params=None):
+        calls.append(params)
+        return _DF([{"ts_code": "601126.SH", "period": "2026Q1", "review_version": "2026.09-2", "grades": json.dumps({"penman": "好"}),
+                     "summary": json.dumps({"overall": "好", "groups": {"回报与护城河": "好"}, "valuation": "贵", "doubt_hard": False,
+                                            "thesis": "x", "confidence": "中", "grade_counts": {"好": 5, "中": 3, "差": 1, "证据不足": 1},
+                                            "report_path": "/company/601126.SH/review"}, ensure_ascii=False),
+                     "brain_used": True, "brain_status": "ok", "ran_at": dt.datetime(2026, 9, 8, 10, 0)},
+                    {"ts_code": "688008.SH", "period": "2026Q1", "review_version": "2026.09-2", "grades": {"buffett": "差"},
+                     "summary": None, "brain_used": False, "brain_status": None, "ran_at": dt.datetime(2026, 5, 1, 10, 0)}])
+    out = sources.company_reviews(["SH601126", "SH688008", "SZ000001"], "2026-09-09", read_pg=rp)
+    t("按点后缀码查", calls[0][0] == ["601126.SH", "688008.SH", "000001.SZ"])
+    t("摘要解析与报告链接", out["SH601126"]["overall"] == "好" and out["SH601126"]["report_url"].endswith("/company/601126.SH/review"))
+    t("运行日距数据日", out["SH601126"]["age_days"] == 1 and out["SH688008"]["age_days"] == 131)
+    t("没摘要的老行只有评价计数", out["SH688008"]["overall"] is None and out["SH688008"]["grade_counts"]["差"] == 1)
+    t("没报告的票不在字典里", "SZ000001" not in out)
+    def boom(*a, **k): raise RuntimeError("no table")
+    t("表读不到整体缺席", sources.company_reviews(["SH601126"], "2026-09-09", read_pg=boom) == {})
+
+
+def test_fifth_path():
+    up = ls.from_company(_review("好"), "2026-09-09")
+    t("质地好向好且带截止日", up["signal"] == ls.SIG_UP and up["as_of"] == "2026-09-08" and up["path"] == ls.PATH_COMPANY)
+    flat = ls.from_company(_review("中"), "2026-09-09")
+    t("质地中平稳", flat["signal"] == ls.SIG_FLAT)
+    down = ls.from_company(_review("差", doubt=True), "2026-09-09")
+    t("质地差且盈余差：转弱且达到存疑条件", down["signal"] == ls.SIG_DOWN and down["hard"])
+    soft = ls.from_company(_review("差", doubt=False), "2026-09-09")
+    t("质地差但盈余不差：转弱不硬", soft["signal"] == ls.SIG_DOWN and not soft["hard"])
+    t("没报告缺失", ls.from_company(None, "2026-09-09")["signal"] == ls.SIG_NONE)
+    stale = _review("好"); stale["age_days"] = 200
+    t("报告陈旧缺失", ls.from_company(stale, "2026-09-09")["signal"] == ls.SIG_NONE)
+    t("证据不足缺失", ls.from_company(_review("证据不足"), "2026-09-09")["signal"] == ls.SIG_NONE)
+    # 合成：质地好单独不能定态（慢信号）；配研报论断向好则强化；质地差硬转弱单独就存疑
+    alone = ls.compose([ls.signal(ls.PATH_CLAIM, ls.SIG_NONE, why="缺"), up])
+    t("只剩公司质地这一路慢信号不能单独定态", alone["state"] == ls.STATE_UNKNOWN)
+    with_claim = ls.compose([ls.signal(ls.PATH_CLAIM, ls.SIG_UP, as_of="2026-09-08", coverage=3, why="利好"), up])
+    t("研报论断向好加质地好为逻辑强化", with_claim["state"] == ls.STATE_STRONG)
+    doubt = ls.compose([ls.signal(ls.PATH_CLAIM, ls.SIG_UP, as_of="2026-09-08", coverage=3, why="利好"), down])
+    t("一路向好一路转弱是证据矛盾交人", doubt["state"] == ls.STATE_UNKNOWN)
+    doubt2 = ls.compose([ls.signal(ls.PATH_CLAIM, ls.SIG_NONE, why="缺"), down])
+    t("质地差达到进入条件则逻辑存疑", doubt2["state"] == ls.STATE_DOUBT)
+    t("存疑触发表里有第五路", ls.PATH_COMPANY in ls.DOUBT_TRIGGERS and ls.SPEED[ls.PATH_COMPANY] == "慢")
+
+
+def test_card_gate_and_views():
+    ev = {"pointed": True, "theme": "端侧AI芯片", "n_sources": 3, "pct0": 4.0, "covered": True, "upside": 0.2, "risk_name": False,
+          "accum_state": "明确吸筹·量在价先", "accum_score": 80, "accum_age": 1, "y_signal": "WATCH", "stale_snapshot": False, "logic": []}
+    base = card.judge(ev)
+    t("底子是候选", base["verdict"] == "候选")
+    down = card.judge(ev, company_review=_review("差", doubt=True))
+    t("质地差候选降为关注且依据写明", down["verdict"] == "关注" and "公司质地评为差" in down["basis"] and down["company_gate"])
+    t("质地中不动", card.judge(ev, company_review=_review("中"))["verdict"] == "候选")
+    t("质地好不升档、证据不足不动", card.judge(ev, company_review=_review("好"))["verdict"] == "候选"
+      and card.judge(ev, company_review=_review("证据不足"))["verdict"] == "候选")
+    t("开关关掉不降", card.judge(ev, company_review=_review("差"), quality_gate=False)["verdict"] == "候选")
+    watch_ev = {**ev, "accum_state": ""}
+    t("本来就是关注的不再变", card.judge(watch_ev, company_review=_review("差"))["verdict"] == "关注")
+    v = card.company_review_view(_review("好"))
+    t("整句带质地三类、十视角计数、论点、失效条件、链接", all(x in v for x in ("质地好", "回报好", "好5/中3/差1", "买方论点", "失效条件", "报告 http")))
+    t("短句与分组标签", card.company_review_short(_review("好")).startswith("好·贵") and card.company_review_label(_review("证据不足")) == "无"
+      and card.company_review_label(None) == "无" and card.company_review_view(None).startswith("公司深度：暂无"))
+
+
+def main():
+    test_sources()
+    test_fifth_path()
+    test_card_gate_and_views()
+    print("ALL OK — 公司质地：取数 / 第五路 / 判决收敛 / 卡上文字")
+
+
+if __name__ == "__main__":
+    main()