第零件丙组：四态的三处输入缺陷

独立审查（09-07）第 8、9、10 条，不修的话第三件一接入就是分流错。

一，甲路只喂了展示用的三条论断。跨期翻转与同期分歧都在三条上算，"之前的论断"最多
两条，翻转两头都会判错：历史八空两好加最新一空，全量判无法判断、三条判逻辑存疑；
历史五好近三空则反过来。现在全量取一次（上限由新配置项 LOGIC_CLAIMS_FULL 定，
默认 200，与覆盖率读数脚本同口径），展示从里面切前三条。

二，乙路的硬触发只维持一个计划日。迁移只在指纹变化那天记 migrated=1，次日归零，
而采信倾向仍偏空并未回稳，存疑却自动解除。加一份每簇近几个计划日的历史（一次查询
取全部簇，不逐簇查），规则是近 N 日内出现过偏多到偏空的迁移且此后一直偏空，硬触发
就仍成立；倾向一离开偏空立刻作废。N 取 5（新配置项 JUDGEMENT_HOLD_DAYS）。

三，落在法定假日的快照会吞掉硬触发。快照按工作日投递、plan_date 取当天，假日那行
夹在两个数据日之间，节后计划读到的是与假日行比对得到的 migrated=0。加交易日守卫：
今天不是交易日就不落行。日历读不到时不拦，不让守卫本身成为断产点。

测试：四态模块加五例，快照模块加八例，计划装配加三例。

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

config.py

@@ -240,6 +240,17 @@ MARKET_MYSQL_SOURCE = os.environ.get("MARKET_MYSQL_SOURCE", "price").strip().low
 # 只展示、不作门槛、不进判决；设 0 表示不读该视图（视图未建时也可用它关掉那一行告警）。
 LOGIC_CLAIMS_PER_STOCK = int(os.environ.get("LOGIC_CLAIMS_PER_STOCK", "3"))
 
+# 四态甲路吃的论断上限（2026-09-07）。它与上面那个"卡上展示几条"是两回事：展示三条够看，
+# 但判跨期翻转与同期分歧要看全量——三条里"之前的论断"最多两条，翻转两头都会判错。
+# 200 与覆盖率读数脚本同口径（实测单票最多 44 条，200 是护栏不是预期）。
+LOGIC_CLAIMS_FULL = int(os.environ.get("LOGIC_CLAIMS_FULL", "200"))
+
+# 乙路硬触发的维持窗口（计划日数，2026-09-07）。采信倾向从偏多迁移到偏空那一天记
+# migrated=1，次日归零；不带历史的话硬触发只维持一个计划日，而倾向仍偏空并未回稳。
+# 规则是：近 N 个计划日内出现过这次迁移且此后一直偏空，硬触发就仍成立。N 取 5，
+# 与四态退出存疑要求"连续三个计划日不再存疑"同一量级、再留两天余量。一次定死。
+JUDGEMENT_HOLD_DAYS = int(os.environ.get("JUDGEMENT_HOLD_DAYS", "5"))
+
 # 论断陈旧线（自然日）。超过它就在候选卡的缺失项里写明"研报论断已过 N 天"，只提示不作门槛。
 # 取 90 天的理由是与吸筹评分的日龄上限可比：那一条是三十个交易日，约合六周多的自然日，
 # 而研报的更新节奏本来就比资金结构评分慢得多（行业深度按季度、跟踪报告按事件），

judgement.py

@@ -186,6 +186,36 @@ def by_segment_name(snaps: dict) -> dict:
     return out
 
 
+def recent_rows(day: str, days: int = 5, read_mysql=None) -> dict:
+    """每个簇最近 days 个计划日的快照行，按簇键索引、行按计划日升序，只取严格早于 day 的。
+
+    给四态乙路判"硬触发要不要维持"用（2026-09-07）：迁移只在指纹变化那天记 migrated=1，
+    次日归零，光看当天那一行硬触发只能维持一个计划日。一次查询取回全部簇的近几行，
+    不逐簇查——计划装配一轮要看几百只票，逐簇查是几百次往返。
+    读不到返回空字典，那样乙路退回"只看当天一行"，不断产。
+    """
+    table = config.JUDGEMENT_SNAPSHOT_TABLE
+    reader = read_mysql or db.read_mysql
+    try:
+        start = (dt.date.fromisoformat(day) - dt.timedelta(days=int(days) * 2 + 7)).isoformat()
+    except ValueError:
+        return {}
+    try:
+        rows = sources._records(reader(  # noqa: SLF001 —— 同仓自用
+            "factor",
+            f"SELECT * FROM {table} WHERE plan_date >= %s AND plan_date < %s "
+            f"ORDER BY cluster_key, plan_date", (start, day)))
+    except Exception as e:  # noqa: BLE001
+        print(f"  （{table} 读不到近日行，乙路只看当天一行: {e!r}）")
+        return {}
+    out: dict = {}
+    for r in rows:
+        key = str(r.get("cluster_key") or "").strip()
+        if key:
+            out.setdefault(key, []).append({k: _blank_to_none(v) for k, v in dict(r).items()})
+    return {k: v[-int(days):] for k, v in out.items()}
+
+
 def load_previous(day: str, read_mysql=None) -> dict:
     """本计划日之前、回看窗口之内，每个簇最近一次快照的行，按簇键索引。
 
@@ -239,15 +269,29 @@ def save(day: str, rows: list, conn_factory=None) -> None:
         conn.commit()
 
 
-def snapshot(day: str | None = None, fetch=None, load_prev=None, write=None) -> dict:
+def snapshot(day: str | None = None, fetch=None, load_prev=None, write=None,
+             calendar=None) -> dict:
     """一个计划日的完整一步：取最新一版结论、配上一版比对、幂等写当日行、打印读数。
 
     day 不传取今天。补写一个早于今天的计划日要当心：视图给的永远是此刻的最新一版，
     把它记在过去某一天名下，等于把今天的观点写成那天的观察结果，版本史会失真——
     所以这种情况下会多打印一行提醒，但仍照写（重跑与补跑都幂等）。
-    三个函数都可注入（离线单测）：取数、读上一版、落库。"""
+    三个函数都可注入（离线单测）：取数、读上一版、落库。calendar 注入交易日历（升序 ISO 串）。
-    day = day or dt.date.today().isoformat()
+
-    if day != dt.date.today().isoformat():
+    非交易日不落行（2026-09-07 审查修）。快照按工作日 20:40 投递、plan_date 取当天，
+    法定假日（工作日）那行 plan_date 大于节前的数据日、小于节后首个数据日；计划读的是
+    plan_date 不晚于数据日的最新一行，节后计划读到的是与假日行比对得到的 migrated=0，
+    指纹迁移若首次被假日快照记录，任何计划都读不到那个 migrated=1，乙路的硬触发就丢了。
+    国庆那一周连续五个工作日假日，几乎必然命中。所以：今天不是交易日就不写，直接返回。"""
+    today = dt.date.today().isoformat()
+    if day is None:
+        cal = calendar if calendar is not None else sources.trading_days(today, back_days=21)
+        if cal and today not in cal:
+            print(f"  行业观点快照：{today} 不是交易日，不落行（最近交易日 {cal[-1]}）")
+            return {"date": today, "rows": 0, "skipped": "非交易日", "last_trading_day": cal[-1],
+                    "table": config.JUDGEMENT_SNAPSHOT_TABLE}
+        day = today
+    if day != today:
         print(f"  ⚠️ 计划日 {day} 不是今天：抄下来的是此刻的最新一版结论，"
               f"记在过去的日期名下会让版本史失真，确认是有意为之再用。")
     rows_in = (fetch or sources.judgement_rows)()

logic_state.py

@@ -188,7 +188,8 @@ def _refs(rows, limit: int = 3) -> list:
 # 乙路：产业研判
 # ============================================================================
 
-def from_judgement(cur, prev=None, *, stale_days: int = 60) -> dict:
+def from_judgement(cur, prev=None, *, stale_days: int = 60,
+                   history=None, hold_days: int = 5) -> dict:
     """把一个主题的产业研判快照归一成一路信号。cur 与 prev 是 judgement.build_rows 的行。
 
     判的是采信倾向这个受控枚举的迁移，不是它当前的取值。理由是强化说的是"变好了"：
@@ -198,6 +199,12 @@ def from_judgement(cur, prev=None, *, stale_days: int = 60) -> dict:
     只有材料指纹变过才算一次迁移。指纹没变而日子变老，只累加陈旧天数；超过 stale_days
     按缺失处理，因为产业研判是人工点按钮触发的，长期不变说明没人重新看过它，
     不是"看过了没变"。
+
+    history 是这个簇最近若干个计划日的快照行（按日升序，不含 cur），2026-09-07 加的。
+    没有它，硬触发只维持一个计划日：迁移只在指纹变化那天记 migrated=1，次日就归零，
+    而采信倾向仍是偏空、并未回稳，存疑却自动解除。方案定的是"进入即刻成立，退出要
+    连续若干日同向"。这里的规则：近 hold_days 个计划日内出现过偏多到偏空的迁移，
+    且此后一直偏空，硬触发就仍然成立；一旦倾向离开偏空，立刻不再算。
     """
     if not cur:
         return signal(PATH_JUDGE, SIG_NONE, why="这个环节对不上任何产业研判主题")
@@ -231,12 +238,35 @@ def from_judgement(cur, prev=None, *, stale_days: int = 60) -> dict:
         return signal(PATH_JUDGE, SIG_FLAT, as_of=as_of, coverage=cov, refs=refs,
                       why=f"产业研判的采信倾向从{prev_lean}迁移到{lean}，不是多空方向的变化")
     if lean == LEAN_BEAR:
+        # 近几个计划日内翻过空、此后一直偏空 → 硬触发仍然成立（2026-09-07）。
+        flip_day = _recent_bear_flip(history, hold_days)
+        if flip_day:
+            return signal(PATH_JUDGE, SIG_DOWN, as_of=as_of, coverage=cov, hard=True, refs=refs,
+                          why=f"产业研判 {flip_day} 从偏多迁移到偏空，此后一直偏空")
         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 _recent_bear_flip(history, hold_days: int):
+    """近 hold_days 行里最后一次"偏多→偏空"的迁移日；迁移之后若倾向离开过偏空，返回 None。
+
+    history 按日升序、不含当天。只看末尾 hold_days 行：更早的翻转已经超出方案定的
+    退出窗口，不该再撑着硬触发。
+    """
+    rows = [r for r in (history or []) if isinstance(r, dict)][-int(hold_days):]
+    flip = None
+    for r in rows:
+        lean = _s(r.get("leaning"))
+        if r.get("migrated") == 1 and _s(r.get("leaning_prev")) == LEAN_BULL and lean == LEAN_BEAR:
+            flip = _s(r.get("plan_date")) or _s(r.get("review_date")) or "近日"
+            continue
+        if flip and lean != LEAN_BEAR:
+            flip = None                      # 翻空之后又离开了偏空，硬触发作废
+    return flip
+
+
 # ============================================================================
 # 丙路：券商行动
 # ============================================================================

plan.py

@@ -180,16 +180,20 @@ def _segments_of(ds: str) -> dict:
 
 
 def _logic_state_of(k: str, evd: dict, seg_of: dict, seg_view: dict,
-                    broker: dict, ds: str) -> dict:
+                    broker: dict, ds: str, full_logic=None, seg_hist=None) -> dict:
     """一只票的逻辑状态四态。四路各自归一，再按合成规则合成。
 
     乙路的取法：一只票可能挂在多个被指向的环节上，取第一个有行业观点的那个。
     取不到就是缺失，卡上会写明缺的是哪一路——缺失既不算负面也不算正面证据，
     但必须让人看得见系统缺的是什么，不能让人以为系统判过了。
     """
-    a = logic_state.from_claims(evd.get("logic"), ds, stale_days=config.LOGIC_STALE_DAYS)
+    # 甲路吃全量论断（2026-09-07）；没传全量时退回卡上那几条，只为让旧调用与单测不断。
+    claims = full_logic if full_logic is not None else evd.get("logic")
+    a = logic_state.from_claims(claims, ds, stale_days=config.LOGIC_STALE_DAYS)
     row = next((seg_view[t] for t in seg_of.get(k, []) if t in seg_view), None)
-    b = logic_state.from_judgement(row)
+    # 乙路带上这个簇近几个计划日的历史，让"偏多到偏空"的硬触发能维持住（2026-09-07）。
+    hist = (seg_hist or {}).get(str(row.get("cluster_key") or "")) if row else None
+    b = logic_state.from_judgement(row, history=hist, hold_days=config.JUDGEMENT_HOLD_DAYS)
     c = broker.get(k) or logic_state.signal(
         logic_state.PATH_BROKER, logic_state.SIG_NONE,
         why="两个等长窗口里算不出可比的每股收益预测")
@@ -211,7 +215,12 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series,
     daily = sources.stock_daily(ds)
     night = sources.night_conclusions(codes, ds)
     # 因果论断（2026-09-03）：数据基座抽取的论断挂在卡上作证据线，只展示不进判决；视图未建时为空。
-    logic = sources.logic_claims(codes, ds)
+    # 2026-09-07 审查修：四态的甲路要吃**全量**论断，卡上展示只取最近几条。原先两者共用
+    # 一份截断到三条的列表，跨期翻转与同期分歧都只在三条上算，翻转两头都会判错——
+    # 历史八空两好加最新一空，全量判无法判断、三条判逻辑存疑；历史五好近三空则反过来。
+    # 全量取一次，展示从里面切前几条，取数层在截断前算的质量画像照旧挂在每条上。
+    logic_full = sources.logic_claims(codes, ds, per_stock=config.LOGIC_CLAIMS_FULL)
+    logic = {k: v[:config.LOGIC_CLAIMS_PER_STOCK] for k, v in logic_full.items()}
     # 逻辑状态四态的三路输入（丁路公司事件无数据源，logic_state 那边恒出缺失）。
     # 这三路都是"研究证据还在不在"的跟踪，与候选卡的三门槛判决是正交的两维：
     # 判决回答今天要不要买，四态回答支撑它的研究证据还在不在。收敛规则在
@@ -223,6 +232,8 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series,
     # 调度中心一直在按时跑）。load_previous(D+1) 取的正是 plan_date <= D 的每簇最新一行，
     # 顺带也扛得住某个晚上没跑：会退回上一版，陈旧天数照常累加。
     seg_view = judgement.by_segment_name(judgement.load_previous(_next_day(ds)))
+    # 每簇近几个计划日的快照，给乙路判"硬触发要不要维持"（2026-09-07 审查修）。
+    seg_hist = judgement.recent_rows(_next_day(ds), days=config.JUDGEMENT_HOLD_DAYS)
     broker = sources.broker_actions(codes, ds)
     seg_of = _segments_of(ds)
     if risk is None:                              # collect 会传入读过一次的名单；单独调用时自己读
@@ -245,7 +256,8 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series,
                "accum_state": n.get("accum_state"), "accum_score": n.get("accum_score"),
                "accum_age": n.get("accum_age"), "y_signal": n.get("signal"),
                "stale_snapshot": stale, "logic": logic.get(k) or []}
-        state = _logic_state_of(k, evd, seg_of, seg_view, broker, ds)
+        state = _logic_state_of(k, evd, seg_of, seg_view, broker, ds,
+                                full_logic=logic_full.get(k) or [], seg_hist=seg_hist)
         j = card.judge(evd, start_pct=config.CARD_START_PCT,
                        accum_max_age=config.CARD_ACCUM_MAX_AGE,
                        neg_tol=config.UPSIDE_NEG_TOLERANCE,

test_judgement_snapshot.py

@@ -273,8 +273,45 @@ def test_migration_on_fingerprint_change():
     t("空版本史不报错", judgement.migrations([]) == [] and judgement.migrations(None) == [])
 
 
+def test_recent_rows_and_holiday_guard():
+    print("近日行读取与非交易日守卫（审查 2026-09-07 第 9、10 条）")
+    rows = [
+        {"plan_date": "2026-09-01", "cluster_key": "Segment::甲", "leaning": "偏多", "migrated": 0},
+        {"plan_date": "2026-09-02", "cluster_key": "Segment::甲", "leaning": "偏空", "migrated": 1,
+         "leaning_prev": "偏多"},
+        {"plan_date": "2026-09-03", "cluster_key": "Segment::甲", "leaning": "偏空", "migrated": 0},
+        {"plan_date": "2026-09-03", "cluster_key": "Segment::乙", "leaning": "偏多", "migrated": None,
+         "n_bull": float("nan")},
+    ]
+    got = judgement.recent_rows("2026-09-04", days=2, read_mysql=lambda src, sql, params: rows)
+    t("按簇索引、只留最近 days 行、按日升序",
+      [r["plan_date"] for r in got["Segment::甲"]] == ["2026-09-02", "2026-09-03"])
+    t("NaN 归一成 None", got["Segment::乙"][0]["n_bull"] is None)
+    t("读失败返回空字典不抛错", judgement.recent_rows("2026-09-04", read_mysql=_boom) == {})
+    t("日期认不出返回空字典", judgement.recent_rows("不是日期", read_mysql=lambda *a, **k: rows) == {})
+
+    wrote = []
+    r = judgement.snapshot(None, fetch=lambda: [], load_prev=lambda d: {},
+                           write=lambda d, rs: wrote.append((d, rs)), calendar=["2000-01-03"])
+    t("今天不在交易日历里 -> 不落行、返回跳过", r.get("skipped") == "非交易日" and wrote == [])
+    import datetime as _dt
+    today = _dt.date.today().isoformat()
+    r = judgement.snapshot(None, fetch=lambda: [], load_prev=lambda d: {},
+                           write=lambda d, rs: wrote.append((d, rs)), calendar=["2000-01-03", today])
+    t("今天是交易日 -> 照常落行", r.get("rows") == 0 and wrote and wrote[0][0] == today)
+    wrote.clear()
+    r = judgement.snapshot(None, fetch=lambda: [], load_prev=lambda d: {},
+                           write=lambda d, rs: wrote.append((d, rs)), calendar=[])
+    t("日历读不到 -> 不拦，按今天落行（不让守卫本身成为断产点）", wrote and wrote[0][0] == today)
+    wrote.clear()
+    r = judgement.snapshot("2026-09-01", fetch=lambda: [], load_prev=lambda d: {},
+                           write=lambda d, rs: wrote.append((d, rs)), calendar=["2000-01-03"])
+    t("显式给了计划日 -> 不查日历，照写", wrote and wrote[0][0] == "2026-09-01")
+
+
 def main():
     test_fetch()
+    test_recent_rows_and_holiday_guard()
     test_fetch_failure()
     test_prev_read_failure()
     test_same_day_idempotent()

test_logic_state.py

@@ -98,6 +98,28 @@ def test_judgement():
     t("迁到中性 -> 平稳（不是多空方向的变化）", r["signal"] == ls.SIG_FLAT)
 
 
+def test_judgement_history():
+    print("乙路 硬触发的维持窗口（审查 2026-09-07 第 9 条）")
+    # 迁移只在指纹变化那天记 migrated=1，次日归零；不带历史时硬触发只维持一个计划日。
+    flip = jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BULL, migrated=1, plan_date="2026-09-02")
+    after1 = jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BEAR, migrated=0, plan_date="2026-09-03")
+    today = jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BEAR, migrated=0, plan_date="2026-09-04")
+    r = ls.from_judgement(today)
+    t("不带历史：迁移次日之后只是转弱、不硬", r["signal"] == ls.SIG_DOWN and not r["hard"])
+    r = ls.from_judgement(today, history=[flip, after1])
+    t("带历史：近几日翻过空且此后一直偏空 -> 硬触发维持", r["hard"] and "此后一直偏空" in r["why"])
+    back = jrow(leaning=ls.LEAN_BULL, prev=ls.LEAN_BEAR, migrated=1, plan_date="2026-09-03")
+    r = ls.from_judgement(today, history=[flip, back])
+    t("翻空之后又回到偏多过一次 -> 硬触发作废", not r["hard"])
+    old = [jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BULL, migrated=1, plan_date="2026-08-20")] \
+        + [jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BEAR, migrated=0, plan_date=f"2026-08-2{i}")
+           for i in range(1, 7)]
+    r = ls.from_judgement(today, history=old, hold_days=5)
+    t("翻空早于维持窗口 -> 不再撑着硬触发", not r["hard"])
+    r = ls.from_judgement(today, history=[None, "坏行", flip])
+    t("历史里的坏行被跳过、不抛错", r["hard"])
+
+
 def test_broker():
     print("丙路 券商行动")
     r = ls.from_broker({"eps": 2.3, "firms": 1}, {"eps": 2.4, "firms": 5}, as_of=DS)
@@ -241,11 +263,12 @@ def test_apply():
 def main():
     test_claims()
     test_judgement()
+    test_judgement_history()
     test_broker()
     test_compose()
     test_settle()
     test_apply()
-    print("ALL OK — 四路归一 / 合成四条规则 / 四态入口 / 负向优先 / 进出不对称 / "
+    print("ALL OK — 四路归一 / 乙路硬触发维持窗口 / 合成四条规则 / 四态入口 / 负向优先 / 进出不对称 / "
           "作用于候选卡的单调性 全部通过")
 
 

test_plan_logic_state.py

@@ -85,7 +85,24 @@ def main():
     t("次日推算正确", plan._next_day("2026-09-03") == "2026-09-04")  # noqa: SLF001
     t("认不出的日期原样返回，不抛异常", plan._next_day("不是日期") == "不是日期")  # noqa: SLF001
 
-    print("ALL OK — 四态随卡产出 / 缺席不断产 / 下游形状 / 判决不被改动 / 计划日推算 全部通过")
+    print("乙路历史传到计划装配（审查 2026-09-07 第 9 条）")
+    cur = jrow(leaning="偏空")
+    cur["cluster_key"] = "Segment::固态电解质"
+    hist = {"Segment::固态电解质": [
+        {"plan_date": "2026-09-01", "leaning": "偏空", "leaning_prev": "偏多", "migrated": 1},
+        {"plan_date": "2026-09-02", "leaning": "偏空", "leaning_prev": "偏空", "migrated": 0}]}
+    st = plan._logic_state_of("SH600000", {"logic": []}, {"SH600000": ["固态电解质"]},  # noqa: SLF001
+                              {"固态电解质": cur}, {}, DS, seg_hist=hist)
+    t("近日翻空且一直偏空 -> 硬触发维持 -> 逻辑存疑", st["state"] == ls.STATE_DOUBT)
+    st = plan._logic_state_of("SH600000", {"logic": []}, {"SH600000": ["固态电解质"]},  # noqa: SLF001
+                              {"固态电解质": cur}, {}, DS)
+    t("不传历史时退回只看当天一行（不断产）", st["state"] != ls.STATE_DOUBT)
+    full = [claim("2026-07-10"), claim("2026-07-20"), claim("2026-08-28", "利空")]
+    st = plan._logic_state_of("SH600000", {"logic": full[-1:]}, {}, {}, {}, DS, full_logic=full)  # noqa: SLF001
+    t("甲路吃全量论断：三条里判得出跨期翻转", any(p["path"] == ls.PATH_CLAIM and p["signal"] == ls.SIG_DOWN
+                                          for p in st["paths"]) and st["state"] == ls.STATE_DOUBT)
+
+    print("ALL OK — 四态随卡产出 / 缺席不断产 / 下游形状 / 判决不被改动 / 计划日推算 / 乙路历史与甲路全量 全部通过")
 
 
 if __name__ == "__main__":