逻辑状态四态：合成模块与离线单测（只写不接）

判断的不是"这条产业逻辑对不对"——那是研究员的活；只判断支撑它的证据还在不在。

四路证据各自归一成一个信号（向好、平稳、转弱、缺失，另加分歧表示这一路自己内部
方向相反）：甲路研报论断（方向受控必填，两个窗口——新材料窗口决定向好、陈旧线决定
还算不算数）、乙路产业研判（判采信倾向这个受控四值的迁移，不判自由文本）、丙路券商
行动（等长窗口的每股收益预测中位数与覆盖机构数）、丁路公司事件（恒定缺失，方向字段
九成五为空，照样出一个缺失让卡上看得见缺的是它）。

合成四条规则都落成了断言：每路必须带截止日（没有就降为缺失）、负向优先（转弱最多到
无法判断，只有两条路径允许落到逻辑存疑）、缺失既不算负面也不算正面（剔出合成但记名）、
慢信号不能单独定态。另有进出不对称：进入逻辑存疑即刻成立，退出要连续三个计划日不再
存疑；漏判一次证据走弱的代价比多留一次人工复核大得多。

作用于候选卡是单调的：逻辑强化只能提前卡内序、永远不升判决；逻辑存疑只能改分流通道、
不能把仅展示变成可执行。唯一的新语义是候选加逻辑存疑——可交易口径成立而研究证据在
走弱，正是杀逻辑与杀估值那条边界，处置是强制人工确认。

本次只写不接进计划装配：乙路依赖行业观点快照攒够几天才有迁移可判。
测试 59 例通过，不连库。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

logic_state.py

@@ -0,0 +1,432 @@
+"""逻辑状态四态：判断支撑一只票的研究证据还在不在（纯函数，不连库、不读配置）。
+
+## 它回答什么，不回答什么
+
+它不判断"这条产业逻辑对不对"——那是研究员的活，系统做不了也不该做。
+它只看支撑逻辑的证据还在不在、有没有在走弱：预测在不在下修、覆盖在不在收缩、
+有没有出现方向相反的论断、产业研判的采信倾向有没有翻过来。
+
+与候选卡的三态是正交的两维，不合并。候选卡回答"今天要不要买"，是当日无状态重算的
+入场判决；这里回答"支撑它的研究证据还在不在"，是跨日的持有期跟踪。两者不能合并还有
+一个硬理由：持仓票在候选筛选里被整行剔掉且剔在判决检查之前，所以候选卡三态对持仓
+天然无效，而持仓恰恰是逻辑四态唯一真正有用的地方。
+
+## 四路证据
+
+  甲路 研报论断   方向受控必填（利好、利空、中性三选一，缺方向的断言落库时直接丢弃），
+                  行业级环节级公司级三种粒度都有，随研报入库即时。中速。
+  乙路 产业研判   主题级。判据用采信倾向这个四选一的受控枚举（偏多、偏空、中性、证据
+                  不足）的迁移，不用那段自由文本——判"上修还是分化"会退化成中文匹配。慢速。
+  丙路 券商行动   个股级周度：同财年同预测期的每股收益预测中位数变化，与覆盖机构数变化。
+                  看的是券商的行动不是言辞——券商极少明说不看好某个行业。中速。
+  丁路 公司事件   目前不可用：全库事件断言里九成五的方向字段为空，有值的也是动作词
+                  （减持、增持、预增）不是利好利空的语义判断。这一路照样出一个"缺失"，
+                  让卡上看得见缺的是它，而不是静悄悄地不存在。
+
+## 每路先归一成一个信号
+
+向好、平稳、转弱、缺失，另加一个分歧——分歧是"这一路自己内部方向相反"，设计里把它
+单列为证据矛盾的一个来源，所以它不能被压进那四个里的任何一个。
+
+一条硬要求：每路必须带自己的截止日。没有截止日的证据一律按缺失处理，因为"这条证据
+是哪天的"决定了它还算不算数，不知道就不能采信。
+
+## 合成的四条规则
+
+  一，每路各自归一，各带截止日与覆盖计数。
+  二，负向优先：任何一路转弱，整体最多到无法判断；只有两条路径允许整体落到逻辑存疑，
+      见 DOUBT_TRIGGERS。
+  三，缺失既不是负面也不是正面证据，从合成里剔除并记明缺哪一路。缺省态永远是
+      无法判断加证据不足——从来没有行业观点、券商也算不出来的票，不算逻辑成立。
+  四，慢信号定底色、中速定当前态、日度事件只作一次性复核触发不能单独定态。抖动不靠
+      平滑系数，靠三件事：等长窗口本身就是低通、状态迁移要求连续若干计划日同向、
+      退出负态比进入负态慢（settle）。
+
+## 阈值
+
+全部由调用方传入，一次定死，每个在台账里写明与哪个既有口径同源，绝不按复盘收益读数
+回调——那是在历史样本上挑参数。本模块只给默认值，不读配置文件。
+
+离线单测见 test_logic_state.py，不连库。
+"""
+from __future__ import annotations
+
+import datetime as dt
+
+# ---- 四态 ----
+STATE_STRONG = "逻辑强化"
+STATE_HOLD = "逻辑成立"
+STATE_UNKNOWN = "无法判断"
+STATE_DOUBT = "逻辑存疑"
+
+# 无法判断的两个子因，动作不同：证据矛盾进人工复核清单（提示"需要裁决"并列出矛盾两路的
+# 日期与出处）；证据不足不进清单，只在卡上标注缺哪一路（提示"需要补材料"）。
+WHY_THIN = "证据不足"
+WHY_CONFLICT = "证据矛盾"
+
+# ---- 每路的信号 ----
+SIG_UP = "向好"
+SIG_FLAT = "平稳"
+SIG_DOWN = "转弱"
+SIG_NONE = "缺失"
+SIG_SPLIT = "分歧"
+
+# ---- 四路 ----
+PATH_CLAIM = "研报论断"
+PATH_JUDGE = "产业研判"
+PATH_BROKER = "券商行动"
+PATH_EVENT = "公司事件"
+
+# 每路的节奏，合成规则四要用。慢信号不能单独定态。
+SPEED = {PATH_JUDGE: "慢", PATH_CLAIM: "中", PATH_BROKER: "中", PATH_EVENT: "日"}
+
+# 只有这两条路径的转弱允许整体落到逻辑存疑，别的转弱最多到无法判断。
+DOUBT_TRIGGERS = {
+    PATH_BROKER: "每股收益预测下修且覆盖机构收缩，两个条件同时满足",
+    PATH_CLAIM: "论断方向跨期翻转（先前利好、最新一条转利空）",
+    PATH_JUDGE: "采信倾向从偏多迁移到偏空，且材料指纹变过",
+}
+
+# 采信倾向的受控四值（数据基座抽取端的提示词写死的，见 backend/workers/tasks.py）。
+LEAN_BULL = "偏多"
+LEAN_BEAR = "偏空"
+LEAN_NEUTRAL = "中性"
+LEAN_THIN = "证据不足"
+
+
+def signal(path: str, sig: str, *, as_of: str | None = None, coverage: int = 0,
+           why: str = "", hard: bool = False, refs=None) -> dict:
+    """一路证据归一后的形状。
+
+    hard 表示这一路的转弱达到了逻辑存疑的进入条件（见 DOUBT_TRIGGERS）；转弱而 hard
+    为假，整体最多到无法判断。没有截止日的证据一律降为缺失——不知道是哪天的证据，
+    就不知道它还算不算数。
+    """
+    if sig != SIG_NONE and not as_of:
+        return {"path": path, "signal": SIG_NONE, "as_of": None, "coverage": 0,
+                "why": "这一路没有截止日，不能采信", "hard": False, "refs": []}
+    return {"path": path, "signal": sig, "as_of": as_of, "coverage": int(coverage or 0),
+            "why": why, "hard": bool(hard and sig == SIG_DOWN), "refs": list(refs or [])}
+
+
+# ============================================================================
+# 甲路：研报因果论断
+# ============================================================================
+
+def from_claims(claims, ds: str, *, fresh_days: int = 30, stale_days: int = 90) -> dict:
+    """把一只票的因果论断归一成一路信号。
+
+    两个窗口，各管一件事：
+
+      fresh_days  新材料窗口。向好要求这个窗口里有新的利好论断——强化说的是"变好了"
+                  而不是"现在是好的"，只有新材料才算变化。默认 30 天，与候选卡确认线的
+                  三十个交易日在同一量级；这里取自然日，因为披露日本来就是自然日。
+      stale_days  还算数的上界。全部论断都超过它，这一路按缺失处理（陈旧等同于没有，
+                  不是负面证据——规则三）。默认 90 天，与候选卡的论断陈旧线同源。
+
+    跨期方向翻转是近似判定：按披露日排序，最新一条转利空而在它之前的论断以利好为主，
+    就算翻转。真正的跨期改口判定在数据基座的融合层，那里能对上同一对主客体；这里只有
+    按票聚起来的论断，对不上主客体，所以这是个偏宽的近似——它会把"讲的不是同一件事的
+    两条论断"也算成翻转。之所以仍然用它：翻转只是把状态压到逻辑存疑，而逻辑存疑的动作
+    是出减持提议强制交人裁决，绝不自动卖，误判的代价是多一次人工复核。
+    """
+    rows = [c for c in (claims or []) if isinstance(c, dict) and c.get("disclosure_date")]
+    if not rows:
+        return signal(PATH_CLAIM, SIG_NONE, why="没有研报论断")
+    rows.sort(key=lambda c: c["disclosure_date"])
+    latest = rows[-1]["disclosure_date"]
+    age = _gap(latest, ds)
+    if age is None:
+        return signal(PATH_CLAIM, SIG_NONE, why="论断的披露日认不出来")
+    if age > int(stale_days):
+        return signal(PATH_CLAIM, SIG_NONE, coverage=len(rows),
+                      why=f"{len(rows)} 条论断最新一条已过 {age} 天，按缺失处理")
+
+    fresh = [c for c in rows if (_gap(c["disclosure_date"], ds) or 10 ** 6) <= int(fresh_days)]
+    up = [c for c in fresh if c.get("direction") == "利好"]
+    down = [c for c in fresh if c.get("direction") == "利空"]
+    refs = _refs(fresh or rows)
+
+    if up and down:
+        return signal(PATH_CLAIM, SIG_SPLIT, as_of=latest, coverage=len(fresh), refs=refs,
+                      why=f"{fresh_days} 天内同时有 {len(up)} 条利好与 {len(down)} 条利空论断，方向分歧")
+    if down:
+        flipped = _flipped(rows)
+        why = (f"{fresh_days} 天内有 {len(down)} 条利空论断"
+               + ("，且在此之前的论断以利好为主，方向跨期翻转" if flipped else ""))
+        return signal(PATH_CLAIM, SIG_DOWN, as_of=latest, coverage=len(fresh),
+                      why=why, hard=flipped, refs=refs)
+    if up:
+        return signal(PATH_CLAIM, SIG_UP, as_of=latest, coverage=len(fresh), refs=refs,
+                      why=f"{fresh_days} 天内有 {len(up)} 条新的利好论断")
+    return signal(PATH_CLAIM, SIG_FLAT, as_of=latest, coverage=len(rows), refs=_refs(rows),
+                  why=f"{len(rows)} 条论断都还在 {stale_days} 天之内，但 {fresh_days} 天内没有新材料")
+
+
+def _flipped(rows_sorted) -> bool:
+    """最新一条转利空，而在它之前（严格早于它的披露日）的论断以利好为主。"""
+    last = rows_sorted[-1]
+    if last.get("direction") != "利空":
+        return False
+    before = [c for c in rows_sorted if c["disclosure_date"] < last["disclosure_date"]]
+    if not before:
+        return False
+    return sum(1 for c in before if c.get("direction") == "利好") > \
+           sum(1 for c in before if c.get("direction") == "利空")
+
+
+def _refs(rows, limit: int = 3) -> list:
+    out = []
+    for c in rows[-limit:][::-1]:
+        out.append({"date": c.get("disclosure_date"), "doc": c.get("doc_title"),
+                    "id": c.get("claim_id"), "direction": c.get("direction"),
+                    "text": c.get("mechanism")})
+    return out
+
+
+# ============================================================================
+# 乙路：产业研判
+# ============================================================================
+
+def from_judgement(cur, prev=None, *, stale_days: int = 60) -> dict:
+    """把一个主题的产业研判快照归一成一路信号。cur 与 prev 是 judgement.build_rows 的行。
+
+    判的是采信倾向这个受控枚举的迁移，不是它当前的取值。理由是强化说的是"变好了"：
+    一个一直偏多的主题每天都算向好的话，强化会变得到处都是，也就不再说明任何事。
+    所以稳定偏多归平稳——它是有效的非负面证据，够撑住逻辑成立，但撑不起逻辑强化。
+
+    只有材料指纹变过才算一次迁移。指纹没变而日子变老，只累加陈旧天数；超过 stale_days
+    按缺失处理，因为产业研判是人工点按钮触发的，长期不变说明没人重新看过它，
+    不是"看过了没变"。
+    """
+    if not cur:
+        return signal(PATH_JUDGE, SIG_NONE, why="这个环节对不上任何产业研判主题")
+    as_of = _s(cur.get("review_date")) or _s(cur.get("plan_date"))
+    cov = int(cur.get("n_materials") or 0)
+    if cur.get("verified") == 0:
+        return signal(PATH_JUDGE, SIG_NONE, coverage=cov, why="产业研判的自我校验未通过")
+    lean = _s(cur.get("leaning"))
+    if lean == LEAN_THIN:
+        return signal(PATH_JUDGE, SIG_NONE, coverage=cov, why="产业研判自己给的是证据不足")
+    stale = int(cur.get("stale_days") or 0)
+    migrated = cur.get("migrated")
+    if migrated != 1 and stale > int(stale_days):
+        return signal(PATH_JUDGE, SIG_NONE, coverage=cov,
+                      why=f"产业研判的材料已 {stale} 天没变过，按缺失处理")
+
+    prev_lean = _s(cur.get("leaning_prev")) or (_s(prev.get("leaning")) if prev else None)
+    refs = [{"theme": _s(cur.get("subject_name")) or _s(cur.get("cluster_key")),
+             "leaning": lean, "leaning_prev": prev_lean, "date": as_of,
+             "n_bull": cur.get("n_bull"), "n_bear": cur.get("n_bear")}]
+    if migrated == 1 and prev_lean and prev_lean != lean:
+        if prev_lean == LEAN_BULL and lean == LEAN_BEAR:
+            return signal(PATH_JUDGE, SIG_DOWN, as_of=as_of, coverage=cov, hard=True, refs=refs,
+                          why="产业研判的采信倾向从偏多迁移到偏空")
+        if lean == LEAN_BEAR:
+            return signal(PATH_JUDGE, SIG_DOWN, as_of=as_of, coverage=cov, refs=refs,
+                          why=f"产业研判的采信倾向从{prev_lean}迁移到偏空")
+        if lean == LEAN_BULL:
+            return signal(PATH_JUDGE, SIG_UP, as_of=as_of, coverage=cov, refs=refs,
+                          why=f"产业研判的采信倾向从{prev_lean}迁移到偏多")
+        return signal(PATH_JUDGE, SIG_FLAT, as_of=as_of, coverage=cov, refs=refs,
+                      why=f"产业研判的采信倾向从{prev_lean}迁移到{lean}，不是多空方向的变化")
+    if lean == LEAN_BEAR:
+        return signal(PATH_JUDGE, SIG_DOWN, as_of=as_of, coverage=cov, refs=refs,
+                      why="产业研判的采信倾向是偏空")
+    return signal(PATH_JUDGE, SIG_FLAT, as_of=as_of, coverage=cov, refs=refs,
+                  why=f"产业研判的采信倾向是{lean}，与上一版相同")
+
+
+# ============================================================================
+# 丙路：券商行动
+# ============================================================================
+
+def from_broker(now, prev, *, as_of: str | None = None, min_firms: int = 2,
+                eps_drop: float = 0.05, eps_rise: float = 0.05,
+                cover_drop: float = 0.30) -> dict:
+    """把同一财年同一预测期的每股收益预测中位数与覆盖机构数，归一成一路信号。
+
+    now 与 prev 是两个等长窗口的读数，各是 {"eps": 中位数, "firms": 机构数}。
+    两个窗口必须等长——窗口不等长会让八成的票假显示覆盖收缩（实测：前 135 天对近 45 天
+    时，907 只票误报）。等长本身也是抗抖动的低通，见合成规则四。
+
+    每股收益从正转负或从负转正时，变化率没有意义，单独标记为转亏或扭亏而不算百分比。
+    转亏方向明确是坏消息，但它不满足逻辑存疑那个双条件，所以只是转弱不带 hard。
+    """
+    n0, n1 = int((prev or {}).get("firms") or 0), int((now or {}).get("firms") or 0)
+    if n0 < int(min_firms) or n1 < int(min_firms):
+        return signal(PATH_BROKER, SIG_NONE, coverage=min(n0, n1),
+                      why=f"两个窗口里能算的机构不足 {min_firms} 家（前 {n0} 家、后 {n1} 家）")
+    e0, e1 = (prev or {}).get("eps"), (now or {}).get("eps")
+    if e0 is None or e1 is None:
+        return signal(PATH_BROKER, SIG_NONE, coverage=min(n0, n1), why="每股收益预测中位数缺失")
+    shrink = (n0 - n1) / n0 if n0 else 0.0
+    refs = [{"eps_prev": e0, "eps_now": e1, "firms_prev": n0, "firms_now": n1, "date": as_of}]
+
+    if (e0 > 0) != (e1 > 0):
+        turned = e1 <= 0
+        return signal(PATH_BROKER, SIG_DOWN if turned else SIG_UP, as_of=as_of,
+                      coverage=n1, refs=refs,
+                      why=("每股收益预测由盈转亏，变化率不适用" if turned
+                           else "每股收益预测由亏转盈，变化率不适用"))
+    chg = (e1 - e0) / abs(e0) if e0 else 0.0
+    if chg <= -float(eps_drop):
+        hard = shrink > float(cover_drop)
+        why = f"每股收益预测中位数下修 {abs(chg):.0%}"
+        if hard:
+            why += f"，同时覆盖机构从 {n0} 家收缩到 {n1} 家（减 {shrink:.0%}）"
+        return signal(PATH_BROKER, SIG_DOWN, as_of=as_of, coverage=n1, hard=hard,
+                      why=why, refs=refs)
+    if chg >= float(eps_rise):
+        return signal(PATH_BROKER, SIG_UP, as_of=as_of, coverage=n1, refs=refs,
+                      why=f"每股收益预测中位数上修 {chg:.0%}")
+    return signal(PATH_BROKER, SIG_FLAT, as_of=as_of, coverage=n1, refs=refs,
+                  why=f"每股收益预测中位数变化 {chg:+.0%}，在阈值之内")
+
+
+# ============================================================================
+# 丁路：公司事件（目前不可用）
+# ============================================================================
+
+def from_events(*_a, **_k) -> dict:
+    """这一路目前不可用，恒定返回缺失。
+
+    照样出一个缺失而不是干脆不实现，是为了让卡上看得见缺的是它：缺失要分得清
+    "无数据源"和"读不到"，静悄悄地不存在会让人以为系统判过了。
+
+    不可用有两个结构性原因，都不是接一下就能解决的。一是方向缺失：全库三十八万条
+    事件断言里三十六万五千条的方向字段为空，占九成五，有值的也是动作词（减持、增持、
+    预增、预减、回购、授予）不是利好利空的语义判断。二是粒度错位：十类事件闭集全部是
+    公司级的，没有一类是行业级催化，而事件谓词的客体类型受控集只有公司与实体、抽取时
+    主体也写死成公司类型，所以行业级催化在结构上不可能从这一路出来——它走甲路。
+    """
+    return signal(PATH_EVENT, SIG_NONE, why="公司事件这一路目前无数据源（方向字段九成五为空）")
+
+
+# ============================================================================
+# 合成
+# ============================================================================
+
+def compose(paths, *, strong_needs=(PATH_CLAIM, PATH_JUDGE)) -> dict:
+    """把各路信号合成一个状态。返回 state / why / paths / usable / missing / reasons。
+
+    strong_needs 是"逻辑强化必须包含其中至少一路"的名单，默认是甲路与乙路——
+    强化不能只靠券商预测上修单独成立，那只是卖方在调数字，不是产业逻辑变好了。
+    """
+    rows = [p for p in (paths or []) if isinstance(p, dict) and p.get("path")]
+    usable = [p for p in rows if p["signal"] != SIG_NONE]
+    missing = [p for p in rows if p["signal"] == SIG_NONE]
+    miss_note = [f"{p['path']}：{p['why']}" for p in missing]
+
+    if not usable:
+        return _out(STATE_UNKNOWN, WHY_THIN, rows, usable, missing,
+                    ["四路证据一路都不可用"] + miss_note)
+
+    ups = [p for p in usable if p["signal"] == SIG_UP]
+    downs = [p for p in usable if p["signal"] == SIG_DOWN]
+    splits = [p for p in usable if p["signal"] == SIG_SPLIT]
+
+    # 规则二，负向优先。分歧与"一路向好另一路转弱"都是证据矛盾，动作是交人裁决。
+    if splits or (ups and downs):
+        rs = [f"{p['path']}（{p['as_of']}）：{p['why']}" for p in splits + ups + downs]
+        return _out(STATE_UNKNOWN, WHY_CONFLICT, rows, usable, missing,
+                    ["不同证据指向相反，需要裁决"] + rs + miss_note)
+    if downs:
+        hard = [p for p in downs if p.get("hard")]
+        rs = [f"{p['path']}（{p['as_of']}）：{p['why']}" for p in downs]
+        if hard:
+            return _out(STATE_DOUBT, None, rows, usable, missing,
+                        [f"{p['path']}的转弱达到进入条件：{DOUBT_TRIGGERS.get(p['path'], '')}"
+                         for p in hard] + rs + miss_note)
+        return _out(STATE_UNKNOWN, WHY_CONFLICT, rows, usable, missing,
+                    ["有证据在走弱，但没到逻辑存疑的进入条件"] + rs + miss_note)
+
+    # 规则四，慢信号不能单独定态。
+    if all(SPEED.get(p["path"]) == "慢" for p in usable):
+        return _out(STATE_UNKNOWN, WHY_THIN, rows, usable, missing,
+                    ["只剩产业研判这一路慢信号，不足以单独定态"] + miss_note)
+
+    if len(ups) >= 2 and any(p["path"] in strong_needs for p in ups):
+        rs = [f"{p['path']}（{p['as_of']}）：{p['why']}" for p in ups]
+        return _out(STATE_STRONG, None, rows, usable, missing, rs + miss_note)
+
+    rs = [f"{p['path']}（{p['as_of']}）：{p['why']}" for p in usable]
+    return _out(STATE_HOLD, None, rows, usable, missing, rs + miss_note)
+
+
+def _out(state, why, rows, usable, missing, reasons) -> dict:
+    return {"state": state, "why": why, "paths": rows,
+            "usable": [p["path"] for p in usable], "missing": [p["path"] for p in missing],
+            "reasons": reasons,
+            "as_of": max([p["as_of"] for p in usable if p.get("as_of")], default=None)}
+
+
+def settle(prev_state, raw_state, recent_raw, *, confirm_days: int = 2,
+           exit_doubt_days: int = 3) -> tuple:
+    """状态迁移的抗抖动与进出不对称。返回（落定的状态，一句话说明）。
+
+    recent_raw 是含今天在内、最近若干个计划日的原始合成状态，最新的排在最后。
+
+    进入逻辑存疑即刻成立，退出要求连续 exit_doubt_days 个计划日都不再存疑——
+    这个不对称是有意的：漏判一次证据走弱的代价，比多留一次人工复核大得多。
+    其余方向的迁移要求连续 confirm_days 个计划日同向，避免一天一变。
+    """
+    seq = [s for s in (recent_raw or []) if s]
+    if raw_state == STATE_DOUBT:
+        return STATE_DOUBT, "进入逻辑存疑即刻成立"
+    if prev_state == STATE_DOUBT:
+        tail = seq[-int(exit_doubt_days):]
+        if len(tail) >= int(exit_doubt_days) and all(s != STATE_DOUBT for s in tail):
+            return raw_state, f"连续 {exit_doubt_days} 个计划日不再存疑，退出逻辑存疑"
+        got = sum(1 for s in tail if s != STATE_DOUBT)
+        return STATE_DOUBT, f"退出逻辑存疑要连续 {exit_doubt_days} 个计划日不存疑，现在 {got} 天"
+    if prev_state is None or raw_state == prev_state:
+        return raw_state, "维持"
+    tail = seq[-int(confirm_days):]
+    if len(tail) >= int(confirm_days) and all(s == raw_state for s in tail):
+        return raw_state, f"连续 {confirm_days} 个计划日同向，迁移到{raw_state}"
+    return prev_state, f"迁移要连续 {confirm_days} 个计划日同向，今天先维持{prev_state}"
+
+
+def apply_to_card(verdict: str, state: str) -> dict:
+    """逻辑状态怎么作用于候选卡的判决。收敛规则是单调的，两条都不许越界：
+
+      逻辑强化只能把卡内序提前、在提议卡上多写一行，永远不能把判决往上升一档。
+      逻辑存疑只能改分流通道，不能把仅展示变成可执行。
+
+    真正需要新语义的只有一格：候选加逻辑存疑，含义是可交易口径成立而研究证据在走弱，
+    这正是杀逻辑与杀估值那条边界。处置是强制人工确认，复用现有的强制确认字段、与关注
+    判决走同一条队列，页面文案区分来源。
+    """
+    out = {"verdict": verdict, "rank_bonus": 0, "force_confirm": False,
+           "pause_add": False, "note": ""}
+    if state == STATE_STRONG:
+        out["rank_bonus"] = 1
+        out["note"] = "研究证据在走强，卡内序提前；判决不变"
+    elif state == STATE_DOUBT:
+        out["pause_add"] = True
+        if verdict == "候选":
+            out["force_confirm"] = True
+            out["note"] = "可交易口径成立而研究证据在走弱，强制人工确认；停增持侧自主动作"
+        else:
+            out["note"] = "研究证据在走弱，停增持侧自主动作；判决不变"
+    elif state == STATE_UNKNOWN:
+        out["note"] = "研究证据不足以定论；判决不变"
+    return out
+
+
+# ---- 小工具 ----
+
+def _s(v):
+    if v is None:
+        return None
+    s = str(v).strip()
+    return s or None
+
+
+def _gap(a: str, b: str):
+    """a 到 b 的自然日数；认不出返回 None。"""
+    try:
+        return (dt.date.fromisoformat(str(b)[:10]) - dt.date.fromisoformat(str(a)[:10])).days
+    except (ValueError, TypeError):
+        return None

test_logic_state.py

@@ -0,0 +1,244 @@
+"""logic_state 的离线单测（不连库、不读配置）。
+
+覆盖：四路各自的归一（甲路两个窗口与跨期翻转、乙路迁移与三种缺失、丙路双条件与
+盈亏翻转、丁路恒缺失）、合成四条规则（截止日必填、负向优先、缺失不算证据、慢信号
+不能单独定态）、逻辑强化的两个限制（要两路且必须含甲或乙）、状态迁移的进出不对称、
+以及作用于候选卡时的单调性（强化不升判决、存疑不把仅展示变成可执行）。
+
+跑法：python3 test_logic_state.py   或   pytest test_logic_state.py
+"""
+import logic_state as ls
+
+DS = "2026-09-03"
+
+
+def t(name, cond):
+    assert cond, name
+    print("  ok", name)
+
+
+def claim(date, direction="利好", **kw):
+    d = {"disclosure_date": date, "direction": direction, "mechanism": "机制",
+         "doc_title": "研报", "claim_id": "c" + date.replace("-", "")}
+    d.update(kw)
+    return d
+
+
+def test_claims():
+    print("甲路 研报论断")
+    t("没有论断 -> 缺失", ls.from_claims([], DS)["signal"] == ls.SIG_NONE)
+    t("论断没有披露日 -> 缺失",
+      ls.from_claims([{"direction": "利好"}], DS)["signal"] == ls.SIG_NONE)
+
+    r = ls.from_claims([claim("2026-02-01")], DS)
+    t("全部超陈旧线 -> 缺失（陈旧等同于没有，不是负面证据）",
+      r["signal"] == ls.SIG_NONE and "已过" in r["why"])
+
+    r = ls.from_claims([claim("2026-08-25")], DS)
+    t("新材料窗口内的利好 -> 向好", r["signal"] == ls.SIG_UP and r["as_of"] == "2026-08-25")
+    t("向好带截止日与出处", r["coverage"] == 1 and r["refs"][0]["id"] == "c20260825")
+
+    r = ls.from_claims([claim("2026-07-01")], DS)
+    t("还在陈旧线内但没有新材料 -> 平稳（强化说的是变好了，不是现在是好的）",
+      r["signal"] == ls.SIG_FLAT)
+
+    r = ls.from_claims([claim("2026-08-25"), claim("2026-08-28", "利空")], DS)
+    t("新材料窗口内多空同现 -> 分歧", r["signal"] == ls.SIG_SPLIT)
+
+    r = ls.from_claims([claim("2026-08-28", "利空")], DS)
+    t("窗口内只有利空、之前没有论断 -> 转弱但不到存疑门槛",
+      r["signal"] == ls.SIG_DOWN and not r["hard"])
+
+    r = ls.from_claims([claim("2026-07-10"), claim("2026-07-20"), claim("2026-08-28", "利空")], DS)
+    t("先前以利好为主、最新一条转利空 -> 转弱且达到存疑门槛",
+      r["signal"] == ls.SIG_DOWN and r["hard"] and "跨期翻转" in r["why"])
+
+    r = ls.from_claims([claim("2026-07-10", "利空"), claim("2026-08-28", "利空")], DS)
+    t("先前就是利空、最新还是利空 -> 转弱但不算翻转", r["signal"] == ls.SIG_DOWN and not r["hard"])
+
+    r = ls.from_claims([claim("2026-08-01")], DS, fresh_days=60)
+    t("新材料窗口旋钮生效", r["signal"] == ls.SIG_UP)
+    r = ls.from_claims([claim("2026-02-01")], DS, stale_days=365)
+    t("陈旧线旋钮生效", r["signal"] == ls.SIG_FLAT)
+
+
+def jrow(leaning=ls.LEAN_BULL, prev=None, migrated=0, stale=3, verified=1, **kw):
+    d = {"leaning": leaning, "leaning_prev": prev, "migrated": migrated, "stale_days": stale,
+         "verified": verified, "review_date": "2026-08-19", "n_materials": 35,
+         "subject_name": "钨产业链", "n_bull": 3, "n_bear": 3}
+    d.update(kw)
+    return d
+
+
+def test_judgement():
+    print("乙路 产业研判")
+    t("对不上任何主题 -> 缺失", ls.from_judgement(None)["signal"] == ls.SIG_NONE)
+    r = ls.from_judgement(jrow(verified=0))
+    t("自我校验未通过 -> 缺失", r["signal"] == ls.SIG_NONE and "自我校验" in r["why"])
+    r = ls.from_judgement(jrow(leaning=ls.LEAN_THIN))
+    t("研判自己给的就是证据不足 -> 缺失", r["signal"] == ls.SIG_NONE)
+    r = ls.from_judgement(jrow(stale=90))
+    t("材料长期没变过 -> 缺失（人工触发，不变说明没人重看过）",
+      r["signal"] == ls.SIG_NONE and "没变过" in r["why"])
+    r = ls.from_judgement(jrow(stale=90, migrated=1))
+    t("指纹刚变过就不算陈旧", r["signal"] != ls.SIG_NONE)
+
+    t("稳定偏多 -> 平稳，不是向好",
+      ls.from_judgement(jrow())["signal"] == ls.SIG_FLAT)
+    r = ls.from_judgement(jrow(leaning=ls.LEAN_BULL, prev=ls.LEAN_NEUTRAL, migrated=1))
+    t("采信倾向迁移到偏多 -> 向好", r["signal"] == ls.SIG_UP)
+    r = ls.from_judgement(jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BULL, migrated=1))
+    t("偏多迁移到偏空 -> 转弱且达到存疑门槛", r["signal"] == ls.SIG_DOWN and r["hard"])
+    r = ls.from_judgement(jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_NEUTRAL, migrated=1))
+    t("中性迁移到偏空 -> 转弱但不到存疑门槛", r["signal"] == ls.SIG_DOWN and not r["hard"])
+    r = ls.from_judgement(jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BULL, migrated=0))
+    t("指纹没变就不算迁移，当前偏空仍记转弱但不是硬触发",
+      r["signal"] == ls.SIG_DOWN and not r["hard"])
+    r = ls.from_judgement(jrow(leaning=ls.LEAN_NEUTRAL, prev=ls.LEAN_BULL, migrated=1))
+    t("迁到中性 -> 平稳（不是多空方向的变化）", r["signal"] == ls.SIG_FLAT)
+
+
+def test_broker():
+    print("丙路 券商行动")
+    r = ls.from_broker({"eps": 2.3, "firms": 1}, {"eps": 2.4, "firms": 5}, as_of=DS)
+    t("任一窗口机构不足两家 -> 缺失", r["signal"] == ls.SIG_NONE)
+    r = ls.from_broker({"eps": 2.30, "firms": 5}, {"eps": 2.34, "firms": 5}, as_of=DS)
+    t("变化在阈值之内 -> 平稳", r["signal"] == ls.SIG_FLAT)
+    r = ls.from_broker({"eps": 2.6, "firms": 5}, {"eps": 2.0, "firms": 5}, as_of=DS)
+    t("只有预测上修 -> 向好", r["signal"] == ls.SIG_UP)
+    r = ls.from_broker({"eps": 1.8, "firms": 5}, {"eps": 2.4, "firms": 5}, as_of=DS)
+    t("只下修、覆盖没收缩 -> 转弱但不到存疑门槛",
+      r["signal"] == ls.SIG_DOWN and not r["hard"])
+    r = ls.from_broker({"eps": 1.8, "firms": 3}, {"eps": 2.4, "firms": 9}, as_of=DS)
+    t("下修且覆盖收缩，两个条件同现 -> 转弱且达到存疑门槛",
+      r["signal"] == ls.SIG_DOWN and r["hard"] and "收缩" in r["why"])
+    r = ls.from_broker({"eps": 2.4, "firms": 3}, {"eps": 2.4, "firms": 9}, as_of=DS)
+    t("只覆盖收缩而预测没动 -> 平稳（覆盖只作否决位，不单独定负）",
+      r["signal"] == ls.SIG_FLAT)
+    r = ls.from_broker({"eps": -0.3, "firms": 5}, {"eps": 0.8, "firms": 5}, as_of=DS)
+    t("由盈转亏 -> 转弱，不给百分比", r["signal"] == ls.SIG_DOWN and "不适用" in r["why"])
+    r = ls.from_broker({"eps": 0.5, "firms": 5}, {"eps": -0.4, "firms": 5}, as_of=DS)
+    t("由亏转盈 -> 向好，不给百分比", r["signal"] == ls.SIG_UP and "不适用" in r["why"])
+    t("丁路恒定缺失", ls.from_events()["signal"] == ls.SIG_NONE)
+
+
+def sig(path, s, hard=False, as_of=DS):
+    return ls.signal(path, s, as_of=as_of, coverage=3, why="测试", hard=hard)
+
+
+def test_compose():
+    print("合成四条规则")
+    r = ls.compose([])
+    t("一路都没有 -> 无法判断加证据不足（缺省态）",
+      r["state"] == ls.STATE_UNKNOWN and r["why"] == ls.WHY_THIN)
+    r = ls.compose([ls.from_events()])
+    t("只有一路不可用 -> 仍是证据不足，且记明缺哪一路",
+      r["state"] == ls.STATE_UNKNOWN and r["missing"] == [ls.PATH_EVENT])
+
+    # 规则一：没有截止日的证据不能采信
+    bad = ls.signal(ls.PATH_BROKER, ls.SIG_UP, as_of=None, coverage=9, why="上修")
+    t("规则一：没有截止日一律降为缺失", bad["signal"] == ls.SIG_NONE)
+
+    # 规则三：缺失既不是负面也不是正面
+    r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_UP), sig(ls.PATH_BROKER, ls.SIG_UP),
+                    ls.from_events()])
+    t("规则三：缺失被剔出合成，不影响状态", r["state"] == ls.STATE_STRONG)
+    t("缺失仍记在名单里", ls.PATH_EVENT in r["missing"] and ls.PATH_EVENT not in r["usable"])
+
+    # 规则四：慢信号不能单独定态
+    r = ls.compose([sig(ls.PATH_JUDGE, ls.SIG_FLAT)])
+    t("规则四：只剩产业研判这一路慢信号 -> 无法判断加证据不足",
+      r["state"] == ls.STATE_UNKNOWN and r["why"] == ls.WHY_THIN)
+    r = ls.compose([sig(ls.PATH_JUDGE, ls.SIG_FLAT), sig(ls.PATH_BROKER, ls.SIG_FLAT)])
+    t("配上一路中速信号就能定态", r["state"] == ls.STATE_HOLD)
+
+    print("四态各自的入口")
+    r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_UP), sig(ls.PATH_JUDGE, ls.SIG_UP)])
+    t("两路向好且含甲乙 -> 逻辑强化", r["state"] == ls.STATE_STRONG)
+    r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_UP)])
+    t("只有一路向好 -> 逻辑成立而不是强化", r["state"] == ls.STATE_HOLD)
+    r = ls.compose([sig(ls.PATH_BROKER, ls.SIG_UP), sig(ls.PATH_EVENT, ls.SIG_UP)])
+    t("两路向好但不含甲乙 -> 不给强化（券商上修不能单独撑起强化）",
+      r["state"] == ls.STATE_HOLD)
+    r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_FLAT)])
+    t("一路平稳无负面 -> 逻辑成立", r["state"] == ls.STATE_HOLD)
+
+    r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_SPLIT)])
+    t("一路内部分歧 -> 无法判断加证据矛盾",
+      r["state"] == ls.STATE_UNKNOWN and r["why"] == ls.WHY_CONFLICT)
+    r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_UP), sig(ls.PATH_BROKER, ls.SIG_DOWN)])
+    t("两路方向相反 -> 无法判断加证据矛盾",
+      r["state"] == ls.STATE_UNKNOWN and r["why"] == ls.WHY_CONFLICT)
+
+    print("规则二 负向优先")
+    r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_DOWN)])
+    t("转弱但没到进入条件 -> 最多到无法判断，不落存疑",
+      r["state"] == ls.STATE_UNKNOWN and r["why"] == ls.WHY_CONFLICT)
+    r = ls.compose([sig(ls.PATH_BROKER, ls.SIG_DOWN, hard=True)])
+    t("转弱且达到进入条件 -> 逻辑存疑", r["state"] == ls.STATE_DOUBT)
+    t("存疑的理由写明是哪条路径触发的",
+      any(ls.DOUBT_TRIGGERS[ls.PATH_BROKER] in x for x in r["reasons"]))
+    r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_FLAT), sig(ls.PATH_BROKER, ls.SIG_DOWN, hard=True)])
+    t("一路平稳压不住另一路的硬触发", r["state"] == ls.STATE_DOUBT)
+    r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_UP), sig(ls.PATH_BROKER, ls.SIG_DOWN, hard=True)])
+    t("一路向好与硬触发同现 -> 是矛盾交人裁决，不直接落存疑",
+      r["state"] == ls.STATE_UNKNOWN and r["why"] == ls.WHY_CONFLICT)
+    t("硬标记只对转弱有效", ls.signal(ls.PATH_CLAIM, ls.SIG_UP, as_of=DS, hard=True)["hard"] is False)
+
+
+def test_settle():
+    print("迁移的抗抖动与进出不对称")
+    s, _ = ls.settle(ls.STATE_HOLD, ls.STATE_DOUBT, [ls.STATE_HOLD, ls.STATE_DOUBT])
+    t("进入存疑即刻成立，不等确认", s == ls.STATE_DOUBT)
+    s, why = ls.settle(ls.STATE_DOUBT, ls.STATE_HOLD, [ls.STATE_DOUBT, ls.STATE_HOLD])
+    t("退出存疑：只有一天不存疑，维持存疑", s == ls.STATE_DOUBT and "现在 1 天" in why)
+    s, _ = ls.settle(ls.STATE_DOUBT, ls.STATE_HOLD,
+                     [ls.STATE_HOLD, ls.STATE_HOLD, ls.STATE_HOLD])
+    t("退出存疑：连续三天不存疑才放行", s == ls.STATE_HOLD)
+    s, _ = ls.settle(ls.STATE_DOUBT, ls.STATE_HOLD,
+                     [ls.STATE_HOLD, ls.STATE_DOUBT, ls.STATE_HOLD])
+    t("三天里夹一天存疑就不放行", s == ls.STATE_DOUBT)
+
+    s, _ = ls.settle(ls.STATE_HOLD, ls.STATE_STRONG, [ls.STATE_HOLD, ls.STATE_STRONG])
+    t("其余迁移：只有一天同向，先维持原状", s == ls.STATE_HOLD)
+    s, _ = ls.settle(ls.STATE_HOLD, ls.STATE_STRONG, [ls.STATE_STRONG, ls.STATE_STRONG])
+    t("连续两天同向才迁移", s == ls.STATE_STRONG)
+    s, _ = ls.settle(None, ls.STATE_HOLD, [ls.STATE_HOLD])
+    t("第一次见到这只票，直接采用", s == ls.STATE_HOLD)
+    s, _ = ls.settle(ls.STATE_HOLD, ls.STATE_HOLD, [ls.STATE_HOLD])
+    t("状态没变时不需要确认", s == ls.STATE_HOLD)
+    s, _ = ls.settle(ls.STATE_HOLD, ls.STATE_STRONG, [ls.STATE_STRONG], confirm_days=1)
+    t("确认天数旋钮生效", s == ls.STATE_STRONG)
+
+
+def test_apply():
+    print("作用于候选卡的单调性")
+    r = ls.apply_to_card("关注", ls.STATE_STRONG)
+    t("强化不把关注升成候选", r["verdict"] == "关注" and r["rank_bonus"] == 1)
+    r = ls.apply_to_card("仅展示", ls.STATE_STRONG)
+    t("强化也不把仅展示升上来", r["verdict"] == "仅展示" and not r["force_confirm"])
+    r = ls.apply_to_card("候选", ls.STATE_DOUBT)
+    t("候选加存疑 -> 强制人工确认（杀逻辑与杀估值那条边界）",
+      r["force_confirm"] and r["pause_add"] and r["verdict"] == "候选")
+    r = ls.apply_to_card("仅展示", ls.STATE_DOUBT)
+    t("存疑不把仅展示变成可执行", r["verdict"] == "仅展示" and not r["force_confirm"])
+    r = ls.apply_to_card("候选", ls.STATE_UNKNOWN)
+    t("无法判断不改判决也不强制确认",
+      r["verdict"] == "候选" and not r["force_confirm"] and not r["pause_add"])
+    r = ls.apply_to_card("候选", ls.STATE_HOLD)
+    t("逻辑成立什么都不做", r["rank_bonus"] == 0 and not r["force_confirm"] and not r["note"])
+
+
+def main():
+    test_claims()
+    test_judgement()
+    test_broker()
+    test_compose()
+    test_settle()
+    test_apply()
+    print("ALL OK — 四路归一 / 合成四条规则 / 四态入口 / 负向优先 / 进出不对称 / "
+          "作用于候选卡的单调性 全部通过")
+
+
+if __name__ == "__main__":
+    main()