From 296fc96909b5207eda06254f78af1da827bc5584 Mon Sep 17 00:00:00 2001 From: zlt Date: Wed, 9 Sep 2026 12:05:03 +0800 Subject: [PATCH] =?UTF-8?q?=E5=85=AC=E5=8F=B8=E8=B4=A8=E5=9C=B0=E6=8E=A5?= =?UTF-8?q?=E5=85=A5=EF=BC=9A=E5=85=AC=E5=8F=B8=E6=B7=B1=E5=BA=A6=E5=8F=96?= =?UTF-8?q?=E6=95=B0=E3=80=81=E9=80=BB=E8=BE=91=E7=8A=B6=E6=80=81=E7=AC=AC?= =?UTF-8?q?=E4=BA=94=E8=B7=AF=E3=80=81=E5=88=A4=E5=86=B3=E6=94=B6=E6=95=9B?= =?UTF-8?q?=E5=8F=AA=E9=99=8D=E4=B8=8D=E5=8D=87=E3=80=81=E5=8D=A1=E4=B8=8E?= =?UTF-8?q?=E8=AE=A1=E5=88=92=E4=B8=A4=E9=94=AE=E3=80=81=E5=80=99=E9=80=89?= =?UTF-8?q?=E5=8D=95=E8=A1=A8=E5=88=97=E3=80=81=E5=A4=8D=E7=9B=98=E5=88=86?= =?UTF-8?q?=E7=BB=84=E3=80=81=E9=80=90=E7=A5=A8=E6=8E=A5=E5=8F=A3=E5=9B=9E?= =?UTF-8?q?=E4=B8=A4=E9=94=AE=E3=80=81=E5=8D=95=E6=B5=8B=EF=BC=88=E5=8F=B0?= =?UTF-8?q?=E8=B4=A6=20053=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5.1 --- api.py | 10 ++ card.py | 48 ++++++- config.py | 5 + docs/个股深度评析接入方案_2026-09-09.md | 180 ++++++++++++++++++++++++ logic_state.py | 31 +++- plan.py | 32 ++++- plan_review.py | 10 ++ sources.py | 60 ++++++++ test_company_quality.py | 113 +++++++++++++++ 9 files changed, 479 insertions(+), 10 deletions(-) create mode 100644 docs/个股深度评析接入方案_2026-09-09.md create mode 100644 test_company_quality.py diff --git a/api.py b/api.py index cb01f7e..d3f8f45 100644 --- a/api.py +++ b/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} diff --git a/card.py b/card.py index 0505811..fc3063e 100644 --- a/card.py +++ b/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 "—" diff --git a/config.py b/config.py index 7e9a3cd..3e8d654 100644 --- a/config.py +++ b/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")) # 读上一次落定态与近几日原始态时往前看几个自然日。要盖住国庆这种连休八天再加两个周末的空档； diff --git a/docs/个股深度评析接入方案_2026-09-09.md b/docs/个股深度评析接入方案_2026-09-09.md new file mode 100644 index 0000000..69091c3 --- /dev/null +++ b/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）"。 +- 薛云奎业绩栏删"经营现金流对净利润（最新期）"，非年报期在数据说明里写明年初至今口径。 +- 盈利稳定改用财务指标表年度每股收益为正的年数，重构表缺年度行不再断档。 +- 主营构成说明改为"取表头行（营业收入合计）"，两套分类相差的旧文字删除。 +- 净现金公司在数据说明与彭曼、格林布拉特两个视角加"净经营资产回报率偏高"的说明。 diff --git a/logic_state.py b/logic_state.py index 1d9d31c..d26bff9 100644 --- a/logic_state.py +++ b/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] diff --git a/plan.py b/plan.py index e9e740b..c70d657 100644 --- a/plan.py +++ b/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)} 个环节：定位对不对看这里，挑票看候选单）") diff --git a/plan_review.py b/plan_review.py index 99d96ae..9d96f2e 100644 --- a/plan_review.py +++ b/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))): diff --git a/sources.py b/sources.py index 9046b5c..9ed5a20 100644 --- a/sources.py +++ b/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 diff --git a/test_company_quality.py b/test_company_quality.py new file mode 100644 index 0000000..8354351 --- /dev/null +++ b/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()