diff --git a/card.py b/card.py index 2426a09..4ffd3a9 100644 --- a/card.py +++ b/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%}" diff --git a/plan.py b/plan.py index 0760d90..62c3c26 100644 --- a/plan.py +++ b/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"), diff --git a/sources.py b/sources.py index baaed6a..3ac82da 100644 --- a/sources.py +++ b/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 + diff --git a/test_events_pricing.py b/test_events_pricing.py index a19d7c0..17b47db 100644 --- a/test_events_pricing.py +++ b/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__":