相关快讯：财联社电报点名条目进候选卡与实时计划，只当上下文不进判决（台账 050）

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

card.py

@@ -456,5 +456,20 @@ def events_short(ev) -> str:
     return f"{e['date'][5:]} {'与'.join(e.get('types') or [])}" + ("（复合）" if e.get("compound") else "")
 
 
+def news_view(nv) -> str:
+    """相关快讯的整句（2026-09-08 台账 050）：财联社电报点名这只票的条目，最新在前最多五条。只是上下文，不判利好利空。"""
+    if not isinstance(nv, dict) or not nv.get("items"):
+        return "相关快讯：近 3 天财联社电报没有点名这只票"
+    parts = [f"{it.get('time')} [{it.get('level')}] {it.get('title')}" for it in nv["items"]]
+    return f"相关快讯（近 3 天 {nv.get('count')} 条，财联社电报点名，不判利好利空）：" + "；".join(parts)
+
+
+def news_short(nv) -> str:
+    if not isinstance(nv, dict) or not nv.get("items"):
+        return "—"
+    it = nv["items"][0]
+    return f"{it.get('time')} {str(it.get('title') or '')[:24]}"
+
+
 def _pct(v) -> str:
     return "—" if v is None else f"{v:+.1%}"

plan.py

@@ -302,6 +302,9 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series,
     # 事件日字段从前复权行情表算，归成四情形。只展示加复盘分组，不进判决；任一读不到整体缺席不断产。
     events = sources.analyst_events(codes, ds)
     ev_fields = sources.event_day_fields(codes, ds, events)
+    # 相关快讯（2026-09-08 台账 050）：财联社电报近 3 天点名这只票的条目，只当上下文送人看与送研判，
+    # 不判正负、不进判决、不进定价状态。表读不到整体缺席不断产。
+    news = sources.news_flashes(codes, ds)
     if risk is None:                              # collect 会传入读过一次的名单；单独调用时自己读
         try:
             risk = factors._risk_set() or set()  # noqa: SLF001 —— 同仓自用
@@ -338,6 +341,7 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series,
             "valuation": valuation.get(k), "valuation_text": card.valuation_view(valuation.get(k)),
             "events": events.get(k), "events_text": card.events_view(events.get(k)),
             "pricing_state": card.pricing_state(ev_fields.get(k)),
+            "news": news.get(k), "news_text": card.news_view(news.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"],
@@ -470,6 +474,8 @@ def collect(date: str | None = None, top: int = 20, obs_top: int = 10,
                      events=c.get("events"), events_text=c.get("events_text"),
                      pricing_state=c.get("pricing_state"),
                      pricing_text=card.pricing_view(c.get("pricing_state")),
+                     # 相关快讯（2026-09-08 台账 050）：原值给程序，整句给人；只是上下文，不进判决。
+                     news=c.get("news"), news_text=c.get("news_text"),
                      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"),

sources.py

@@ -30,6 +30,8 @@ from __future__ import annotations
 
 import datetime as dt
 import json
+import re
+import logging
 import statistics
 from collections import defaultdict
 
@@ -1031,3 +1033,73 @@ def event_day_fields(codes, ds: str, events: dict, *, hist=None, read_mysql=None
 
 def _r4(v):
     return None if v is None else round(float(v), 4)
+
+# ---------------------------------------------------------------------------
+# 相关快讯（2026-09-08，台账 050）：财联社电报表 gp_cls_telegraph（153 库，与行情表同一连接）里
+# 用 stock_list_json 点名了这只票的条目。只当上下文，不当"正向事件"——电报说的是发生了什么，
+# 不说利好利空，正负由看卡的人与择时决策系统的研判去判；不进判决、不进定价状态、不进复盘分组。
+# 窗口按自然日往前数、不封顶到数据日：07:10 出计划时，昨晚与今晨的快讯都要在。
+# ---------------------------------------------------------------------------
+NEWS_WINDOW_DAYS = 3
+NEWS_MAX_PER_CODE = 5
+NEWS_TITLE_LEN = 60
+_news_log = logging.getLogger("bridge.news")
+
+
+def _norm_stock_id(sid) -> str | None:
+    """财联社的 StockID（sh688293 / sz000998）→ 前缀码（SH688293）。认不出返回 None。"""
+    s = str(sid or "").strip().upper()
+    if len(s) == 8 and s[:2] in ("SH", "SZ", "BJ") and s[2:].isdigit():
+        return s
+    return None
+
+
+def _news_title(title, content) -> str:
+    t = str(title or content or "").strip()
+    t = re.sub(r"^财联社\d{1,2}月\d{1,2}日电[，,]?", "", t).strip()
+    return t[:NEWS_TITLE_LEN] + ("…" if len(t) > NEWS_TITLE_LEN else "")
+
+
+def news_flashes(codes, ds: str, *, days: int = NEWS_WINDOW_DAYS, read_mysql=None,
+                 limit_per_code: int = NEWS_MAX_PER_CODE) -> dict:
+    """每只票近 days 个自然日被财联社电报点名的快讯，按前缀码索引；没被点名的票不在字典里。
+
+    返回 {k: {"count": 条数, "latest": "MM-DD HH:MM", "items": [{time, level, title, url}, ...] 最新在前}}。
+    level 是财联社自己的重要度（A 红色、B 加粗、C 普通），原样带出不解释。表读不到返回空字典，不断产。"""
+    rm = read_mysql or db.read_mysql
+    want = {c for c in (codes or []) if c}
+    if not want:
+        return {}
+    try:
+        since = (dt.date.fromisoformat(ds) - dt.timedelta(days=int(days))).isoformat()
+    except ValueError:
+        return {}
+    try:
+        df = rm("factor",
+                "SELECT cls_id, level, publish_time, title, content, stock_list_json, article_url "
+                "FROM gp_cls_telegraph WHERE publish_time >= %s AND JSON_LENGTH(stock_list_json) > 0 "
+                "ORDER BY publish_time DESC", (since,))
+    except Exception as e:  # noqa: BLE001 —— 表没建、连接失败：整体缺席
+        _news_log.warning("news_flashes 读不到 gp_cls_telegraph: %r", e)
+        return {}
+    out: dict = {}
+    for r in df.to_dict("records"):
+        raw = r.get("stock_list_json")
+        try:
+            lst = json.loads(raw) if isinstance(raw, (str, bytes)) else (raw or [])
+        except (TypeError, ValueError):
+            continue
+        ks = {k for k in (_norm_stock_id((it or {}).get("StockID")) for it in lst) if k and k in want}
+        if not ks:
+            continue
+        ts = r.get("publish_time")
+        when = ts.strftime("%m-%d %H:%M") if hasattr(ts, "strftime") else str(ts or "")[5:16]
+        item = {"time": when, "level": str(r.get("level") or "C"), "cls_id": r.get("cls_id"),
+                "title": _news_title(r.get("title"), r.get("content")), "url": r.get("article_url")}
+        for k in ks:
+            slot = out.setdefault(k, {"count": 0, "latest": when, "items": []})
+            slot["count"] += 1
+            if len(slot["items"]) < limit_per_code:
+                slot["items"].append(item)
+    return out
+

test_events_pricing.py

@@ -189,11 +189,53 @@ def test_pricing_state():
     t("没有事件的整句", "没有券商正向事件" in card.events_view(None))
 
 
+class _DF:
+    def __init__(self, rows): self._rows = rows
+    def to_dict(self, _kind): return self._rows
+
+
+def test_news():
+    """相关快讯（台账 050）：按财联社 StockID 标签落到前缀码，最新在前、最多五条、只当上下文。"""
+    import datetime as dt
+    import json
+    rows = [
+        {"cls_id": 3, "level": "C", "publish_time": dt.datetime(2026, 9, 8, 12, 39), "article_url": "u3",
+         "title": "燕东微：上半年8英寸SiN产线产能达3000片/月", "content": "",
+         "stock_list_json": json.dumps([{"StockID": "sh688172", "name": "燕东微"}])},
+        {"cls_id": 2, "level": "B", "publish_time": dt.datetime(2026, 9, 8, 11, 32), "article_url": "u2",
+         "title": "财联社9月8日电，农业板块再度拉升 金健米业17天9板", "content": "",
+         "stock_list_json": json.dumps([{"StockID": "sz000998"}, {"StockID": "sh688172"}, {"StockID": "bad"}])},
+        {"cls_id": 1, "level": "C", "publish_time": dt.datetime(2026, 9, 7, 9, 0), "article_url": "u1",
+         "title": None, "content": "财联社9月7日电，华鲁恒升：尿素价格未来走势存在较大不确定性",
+         "stock_list_json": "not json"},
+    ]
+    calls = []
+    def rm(which, sql, params=None):
+        calls.append((which, params)); return _DF(rows)
+    out = sources.news_flashes(["SH688172", "SZ000998", "SH600000"], "2026-09-08", read_mysql=rm)
+    t("用行情同一连接、窗口按自然日往前 3 天", calls[0][0] == "factor" and calls[0][1] == ("2026-09-05",))
+    t("燕东微两条、最新在前", out["SH688172"]["count"] == 2 and out["SH688172"]["items"][0]["cls_id"] == 3)
+    t("标题去掉「财联社X月X日电」前缀", out["SZ000998"]["items"][0]["title"].startswith("农业板块"))
+    t("没被点名的票不在字典里、坏 JSON 跳过", "SH600000" not in out and all(i["cls_id"] != 1 for i in out["SH688172"]["items"]))
+    t("时间落成 MM-DD HH:MM", out["SH688172"]["latest"] == "09-08 12:39")
+    big = [dict(rows[0], cls_id=10 + i, publish_time=dt.datetime(2026, 9, 8, 13, i)) for i in range(7)]
+    out2 = sources.news_flashes(["SH688172"], "2026-09-08", read_mysql=lambda *a, **k: _DF(big))
+    t("最多五条但计数照实", len(out2["SH688172"]["items"]) == 5 and out2["SH688172"]["count"] == 7)
+    def boom(*a, **k): raise RuntimeError("no table")
+    t("表读不到整体缺席不断产", sources.news_flashes(["SH688172"], "2026-09-08", read_mysql=boom) == {})
+    t("日期坏了也不断产", sources.news_flashes(["SH688172"], "bad", read_mysql=rm) == {})
+    v = card.news_view(out["SH688172"])
+    t("整句带条数与不判正负的说明", v.startswith("相关快讯（近 3 天 2 条") and "不判利好利空" in v and "[C]" in v)
+    t("没有快讯的整句", "没有点名" in card.news_view(None) and card.news_short(None) == "—")
+    t("短句是最新一条", card.news_short(out["SH688172"]).startswith("09-08 12:39 燕东微"))
+
+
 def main():
     test_events()
     test_event_day_fields()
     test_pricing_state()
-    print("ALL OK — 四类事件 / 事件日字段 / 定价状态四情形 / 卡上文字 全部通过")
+    test_news()
+    print("ALL OK — 四类事件 / 事件日字段 / 定价状态四情形 / 卡上文字 / 相关快讯 全部通过")
 
 
 if __name__ == "__main__":