"""候选卡的取数层：把各条证据线从三处库读成"按前缀码索引的字典"（全部只读）。 ## 为什么要有它 候选卡的规则在 card.py，是纯函数；证据从哪来、怎么对齐代码格式、缺了怎么办， 全部收在这里，plan.py 只做装配。三处来源： 基座 PG v_factor_transmission_moved 已启动成员及其所在环节的传导证据（只认 Segment 目标） v_factor_stock_daily 数据日涨幅、主力净额异常值、热度变化 153 代理 strategy_daily_results 决策系统昨夜结论：信号、支撑压力位、吸筹块 （吸筹的评分、状态、评分日三项一次读齐；基座落库的吸筹版本没有评分日，所以不从基座取） 平台 MySQL gp_day_data 交易日历（算评分日龄用；取一只长期存在的票的日期序列） 基座 PG v_factor_logic 数据基座抽取的因果论断（方向、机制、时效、出处），每票最多几条， 只展示不作门槛（2026-09-03 方案第 3.3 节） 平台 MySQL zs_day_data / eastmoney_rzrq_data / fear_greed_index 计划环境段的市场四项：两市成交额、融资余额、恐贪指数； 市场广度从基座 v_factor_stock_daily 当日行自算（同一节） 基座 PG v_factor_judgement 产业研判与环节评析的最新一版结论（采信倾向、多空条数、 自我校验、材料指纹），每个计划日抄一份存版本史（judgement.py） 代码格式：基座是点后缀式 600000.SH，决策系统与桥是前缀式 SH600000，进出都过 common.to_prefix。 读失败的语义：每一路读不到都返回空字典并打印一行原因，候选卡按"缺失"处理（进关注或仅展示）， 不让计划断产——与 pool.py 的安全边界一致。 离线单测：logic_claims 与 market_context 都接受注入的读函数（read_pg / read_mysql）， test_market_context.py 用假数据函数替换真实连接，不连库；两者内部只对"记录列表"做计算， 读函数返回 DataFrame 或普通的字典列表都可以（见 _records）。 """ from __future__ import annotations import datetime as dt import json import re import logging import statistics from collections import defaultdict import pandas as pd import common import config import db # 上证指数与深证成指在指数日线表 zs_day_data 里的代码，两市成交额取二者当日 amount 之和。 MARKET_INDEX_CODES = ("000001.SH", "399001.SZ") # 市场广度里"涨停家数"的近似口径：涨幅达到 9.8%（不读涨跌停价表，与环节日行情视图注释一致）。 LIMIT_UP_PCT = 9.8 # 融资表与恐贪指数表的日期列名不在本仓库内核实过，按候选名逐个匹配（第一个命中的用）。 _DATE_COL_CANDIDATES = ("trade_date", "date", "stat_date", "data_date", "report_date", "dt", "timestamp", "update_date", "record_date", "created_at") def moved_members(ds: str) -> dict[str, dict]: """数据日 ds 被传导指向的环节里，已启动（不在未动名单）的成员。 同一票挂在多个被指向环节上时，取源数最多、其次链符最高的那条作卡上的证据。""" try: df = db.read_pg( "SELECT ts_code, target, n_sources, chain_fit, members_total, moved, " "moved_ratio, mkt_trade_date FROM v_factor_transmission_moved " "WHERE scan_date = %s", (ds,)) except Exception as e: # noqa: BLE001 print(f" （已动成员视图读取失败，候选卡的传导门槛整体缺席: {e!r}）") return {} if df.empty: return {} df["k"] = df["ts_code"].map(lambda s: common.to_prefix(str(s).strip())) df["n_sources"] = pd.to_numeric(df["n_sources"], errors="coerce").fillna(0) df["chain_fit"] = pd.to_numeric(df["chain_fit"], errors="coerce").fillna(0) df = df.sort_values(["n_sources", "chain_fit"], ascending=False).drop_duplicates("k") out = {} for r in df.itertuples(): out[r.k] = {"theme": str(r.target), "n_sources": int(r.n_sources), "chain_fit": float(r.chain_fit), "members_total": None if pd.isna(r.members_total) else int(r.members_total), "moved": None if pd.isna(r.moved) else int(r.moved), "mkt_trade_date": None if pd.isna(r.mkt_trade_date) else str(r.mkt_trade_date)} return out def stock_daily(ds: str) -> dict[str, dict]: """数据日 ds 的个股行情三列：涨幅（百分数）、主力净额异常值、热度五日变化。""" try: df = db.read_pg( "SELECT ts_code, pct_change, net_z, heat_chg FROM v_factor_stock_daily " "WHERE trade_date = %s", (ds,)) except Exception as e: # noqa: BLE001 print(f" （个股日行情视图读取失败，候选卡的已启动门槛整体缺席: {e!r}）") return {} out = {} for r in df.itertuples(): k = common.to_prefix(str(r.ts_code).strip()) out[k] = {"pct0": _f(r.pct_change), "net_z": _f(r.net_z), "heat_chg": _f(r.heat_chg)} return out def trading_days(upto: str, back_days: int = 90) -> list[str]: """交易日历：取一只长期存在的票在 gp_day_data 里的日期序列（表有一千四百万行， 全表 DISTINCT 太慢；按 symbol 走索引）。返回升序 ISO 日期串。""" start = (dt.date.fromisoformat(upto) - dt.timedelta(days=back_days)).isoformat() try: df = db.read_mysql( "factor", "SELECT DISTINCT DATE(`timestamp`) d FROM gp_day_data " "WHERE symbol = %s AND `timestamp` >= %s AND `timestamp` <= %s " "ORDER BY d", ("SH600519", start, upto)) return [pd.Timestamp(x).date().isoformat() for x in df["d"].tolist()] except Exception as e: # noqa: BLE001 print(f" （交易日历读取失败，评分日龄按自然日×5/7 近似: {e!r}）") return [] def night_conclusions(codes, ds: str) -> dict[str, dict]: """决策系统昨夜结论（每票最新一行）：信号、支撑压力位、吸筹块与评分日龄。 评分日龄 = 结论行的 trade_date 到数据日 ds 之间的交易日数（含头不含尾）。 该表会被盘中补扫就地改写、没有落库时刻列，日龄以行的 trade_date 为准， 这是已知局限（方案 2.2 第三项）。""" codes = sorted({common.to_prefix(str(c).strip()) for c in codes if c}) if not codes: return {} # 只取不晚于数据日 ds 的结论行：生产上 ds 就是数据日，与"取最新一行"等价； # 复盘重建历史日时这一条防前视（否则历史日会读到今天的吸筹状态）。 # trade_date 是整数 YYYYMMDD（数据源盘点 §1a），直接比大小。 ds_int = int(ds.replace("-", "")) if ds else 99999999 try: marks = ",".join(["%s"] * len(codes)) df = db.read_mysql( "pms", f"SELECT stock_code, trade_date, signal_type, support_level, pressure_level, " f"raw_logic_json FROM strategy_daily_results " f"WHERE stock_code IN ({marks}) AND trade_date <= %s", tuple(codes) + (ds_int,)) except Exception as e: # noqa: BLE001 print(f" （决策系统结论表读取失败，吸筹确认线与坏信号风险整体缺席: {e!r}）") return {} if df.empty: return {} df = df.sort_values("trade_date").drop_duplicates("stock_code", keep="last") cal = trading_days(ds) cal_index = {d: i for i, d in enumerate(cal)} out = {} for r in df.itertuples(): k = common.to_prefix(str(r.stock_code).strip()) tdate = _ymd(r.trade_date) age = _age(tdate, ds, cal_index) ff = {} try: raw = json.loads(r.raw_logic_json) if isinstance(r.raw_logic_json, str) else (r.raw_logic_json or {}) ff = (raw or {}).get("fund_flow") or {} except Exception: # noqa: BLE001 —— 坏 JSON 当无吸筹块 ff = {} out[k] = { "signal": str(r.signal_type or "").strip().upper() or None, "support": _f(r.support_level), "pressure": _f(r.pressure_level), "conclusion_date": tdate, "accum_state": str(ff.get("state") or "") or None, "accum_score": _f(ff.get("score")), "accum_structure": ff.get("structure"), "accum_pos_tag": ff.get("pos_tag"), "accum_age": age, } return out def _ymd(v) -> str | None: """各表的日期列形态不一（整数 YYYYMMDD、date、datetime、ISO 串），统一成 ISO 日期串。 前几种形态不经 pandas 就能认出来，这样离线单测里的假数据不依赖 pandas；认不出的最后才交给 pandas 解析，仍失败返回 None。""" if v is None or (isinstance(v, float) and v != v): return None if isinstance(v, dt.datetime): return v.date().isoformat() if isinstance(v, dt.date): return v.isoformat() s = str(v).strip() if len(s) == 8 and s.isdigit(): return f"{s[:4]}-{s[4:6]}-{s[6:]}" if len(s) >= 10 and s[4] == "-" and s[7] == "-" and s[:4].isdigit(): return s[:10] try: return pd.Timestamp(s).date().isoformat() except Exception: # noqa: BLE001 return None def _age(tdate: str | None, ds: str, cal_index: dict) -> int | None: if not tdate: return None if tdate in cal_index and ds in cal_index: return cal_index[ds] - cal_index[tdate] try: # 日历缺失或日期在日历之外：自然日 × 5/7 近似 nat = (dt.date.fromisoformat(ds) - dt.date.fromisoformat(tdate)).days return max(0, round(nat * 5 / 7)) except ValueError: return None def _f(v): try: x = float(v) except (TypeError, ValueError): return None return None if x != x else x def _records(df) -> list[dict]: """把读函数的返回统一成字典列表：DataFrame 走 to_dict，普通列表原样返回，None 与空表返回空列表。 这样取数函数的计算部分只面对普通 Python 对象，离线单测的假读函数直接返回字典列表即可。""" if df is None: return [] if isinstance(df, list): return [dict(r) for r in df] if hasattr(df, "to_dict"): if getattr(df, "empty", False): return [] return list(df.to_dict("records")) return list(df) def _to_dot(code: str) -> str: """前缀式 SH600000 转成数据基座的点后缀式 600000.SH；已是点后缀式或纯数字则原样返回。""" s = str(code or "").strip().upper() if "." in s or len(s) < 3: return s if s[:2] in ("SH", "SZ", "BJ") and s[2:].isdigit(): return f"{s[2:]}.{s[:2]}" return s # ============================================================================ # 因果论断（数据基座 v_factor_logic，2026-09-03 方案第 3.3 节"候选卡读因果论断"） # ============================================================================ _LOGIC_COLS = ("ts_code", "subject_name", "object_name", "direction", "mechanism", "condition", "horizon", "strength", "tier", "confidence", "disclosure_date", "doc_id", "doc_title", "source_span", "claim_id", "via_segment", # 质量两列（2026-09-03 起取）：disputed 是这条论断还在多空分歧里没结论， # review_flag 是逻辑评析读过之后标的"疑似误抽"。两者都只作提示，不过滤—— # 那是模型的判断不是确定性事实，过滤会让人看不到系统曾经抽到过什么。 "disputed", "review_flag") def logic_claims(codes, ds: str, per_stock: int | None = None, read_pg=None) -> dict[str, list[dict]]: """这批票在数据基座因果论断视图里、披露日不晚于数据日 ds 的论断，每票取最近披露日的最多 per_stock 条（默认 config.LOGIC_CLAIMS_PER_STOCK），按前缀码索引。 每条论断带：方向、机制、条件、时效、强度、层级、置信度、披露日、出处文档标题与编号、 论断编号、经由环节、主体与客体名。只作展示与出处，不进判决；读失败返回空字典并打印一行原因。 read_pg 可注入（离线单测），默认走 db.read_pg。""" n_per = config.LOGIC_CLAIMS_PER_STOCK if per_stock is None else int(per_stock) if n_per <= 0: return {} wanted = sorted({common.to_prefix(str(c).strip()) for c in codes if c}) if not wanted: return {} dots = [_to_dot(c) for c in wanted] reader = read_pg or db.read_pg try: marks = ",".join(["%s"] * len(dots)) rows = _records(reader( f"SELECT {', '.join(_LOGIC_COLS)} FROM v_factor_logic " f"WHERE ts_code IN ({marks}) AND disclosure_date <= %s " f"ORDER BY ts_code, disclosure_date DESC, claim_id", tuple(dots) + (ds,))) except Exception as e: # noqa: BLE001 print(f" （因果论断视图 v_factor_logic 读取失败，候选卡的论断证据线整体缺席: {e!r}）") return {} if not rows: return {} want = set(wanted) by_code: dict[str, list[dict]] = {} seen: set[tuple[str, str]] = set() for r in rows: k = common.to_prefix(str(r.get("ts_code") or "").strip()) if k not in want: # 只留请求的票，视图返回的多余行不带进卡 continue cid = _s(r.get("claim_id")) if cid and (k, cid) in seen: # 视图里客体为环节的论断按成员展开，同票同论断只留一条 continue if cid: seen.add((k, cid)) by_code.setdefault(k, []).append({ "direction": _s(r.get("direction")), "mechanism": _s(r.get("mechanism")), "condition": _s(r.get("condition")), "horizon": _s(r.get("horizon")), "strength": _s(r.get("strength")), "tier": _s(r.get("tier")), "confidence": _f(r.get("confidence")), "disclosure_date": _ymd(r.get("disclosure_date")), "doc_id": _s(r.get("doc_id")), "doc_title": _s(r.get("doc_title")), "source_span": (_s(r.get("source_span")) or "")[:200] or None, "claim_id": _s(r.get("claim_id")), "via_segment": _s(r.get("via_segment")), "subject": _s(r.get("subject_name")), "object": _s(r.get("object_name")), "disputed": bool(r.get("disputed")), "review_flag": _s(r.get("review_flag")), }) out = {} for k, items in by_code.items(): items.sort(key=lambda c: (c["disclosure_date"] or "", c["confidence"] or -1.0), reverse=True) # 质量指标必须在截断之前算：截断后只剩最近几条，"这票一共有几份研报来源"就看不出来了。 # 算完挂在每条论断上（每条都带同一份，卡上取第一条即可），不另开一个返回值， # 免得上层要同时接两个字典、少接一个就静默丢掉质量提示。 quality = _logic_quality(items, ds) kept = items[:n_per] for c in kept: c["quality"] = quality out[k] = kept return out def _logic_quality(items: list, ds: str) -> dict: """一票全部论断的质量画像，在截断之前算。四项： age_days 最新一条论断距数据日多少天。同一张卡上吸筹评分超过三十个交易日就判陈旧， 而论断此前永不过期，这是双标。这里给出天数，阈值判定在 card.py。 n_docs 论断来自几份不同研报。实测近三成带多条论断的票，全部论断出自同一份研报—— 那份研报一旦过时或本身有偏，这票的整条研究证据一起失效，而卡上看不出来。 n_flagged 被逻辑评析标为疑似误抽的条数；n_disputed 还在多空分歧里没结论的条数。 n_via_segment 客体是环节、按成员展开挂上来的条数。那类不是直接讲这家公司， 与直接讲这家公司的论断分量不同，卡上要分得开。 """ days = None dates = [c["disclosure_date"] for c in items if c.get("disclosure_date")] if dates and ds: try: days = (dt.date.fromisoformat(ds) - dt.date.fromisoformat(max(dates))).days except ValueError: days = None return { "n": len(items), "age_days": days, "latest_date": max(dates) if dates else None, "n_docs": len({c["doc_id"] for c in items if c.get("doc_id")}), "n_flagged": sum(1 for c in items if c.get("review_flag")), "n_disputed": sum(1 for c in items if c.get("disputed")), "n_via_segment": sum(1 for c in items if c.get("via_segment")), } def _s(v) -> str | None: if v is None or (isinstance(v, float) and v != v): return None s = str(v).strip() return s or None def _i(v) -> int | None: x = _f(v) return None if x is None else int(x) def _b(v) -> bool | None: """自我校验结果这类三值列：真、假、还没有结果。空值保持为空，不当成假。""" if v is None or (isinstance(v, float) and v != v): return None if isinstance(v, bool): return v s = str(v).strip().lower() if s in ("true", "t", "1", "yes", "y"): return True if s in ("false", "f", "0", "no", "n"): return False return None def _n_items(v) -> int | None: """JSON 数组列只留条数：列表直接数，字符串先按 JSON 解析，认不出返回空。""" if v is None or (isinstance(v, float) and v != v): return None if isinstance(v, (list, tuple)): return len(v) if isinstance(v, str): try: parsed = json.loads(v) except ValueError: return None return len(parsed) if isinstance(parsed, list) else None return None # ============================================================================ # 产业研判与环节评析（数据基座 v_factor_judgement） # 用处见 judgement.py：选股系统每个计划日自留一份快照，攒版本史。 # ============================================================================ _JUDGEMENT_COLS = ("scope", "subject_name", "segment_name", "cluster_key", "leaning", "n_bull", "n_bear", "n_flags", "verified", "verify_problems", "n_materials", "model", "rounds", "review_date", "reviewed_at", "input_version", "review_id") def judgement_rows(scopes=None, read_pg=None) -> list[dict]: """数据基座研判结论视图里 scope 落在指定几类的全部行，一行一个评析簇。 scopes 不传时取 config.JUDGEMENT_SCOPES（默认产业研判 industry 与环节评析 segment）。 每行带：主题名或环节名、采信倾向、多头论点条数、空头论点条数、旗标条数、自我校验结果与 校验问题条数、材料条数、模型与轮次、生成日期与生成时刻、材料指纹、评析编号、簇键。 这张视图按簇键唯一、重评时整行覆盖（视图注释里的已知局限 a），所以它永远只有最新一版； 版本史由读的一方自己攒，见 judgement.py。读失败返回空列表并打印一行原因，不阻断上层。 read_pg 可注入（离线单测），默认走 db.read_pg。""" want = tuple(scopes) if scopes else tuple(sorted(config.JUDGEMENT_SCOPES)) if not want: return [] reader = read_pg or db.read_pg try: marks = ",".join(["%s"] * len(want)) rows = _records(reader( f"SELECT {', '.join(_JUDGEMENT_COLS)} FROM v_factor_judgement " f"WHERE scope IN ({marks})", want)) except Exception as e: # noqa: BLE001 print(f" （研判结论视图 v_factor_judgement 读取失败，本计划日没有行业观点行可写: {e!r}）") return [] out = [] for r in rows: key = _s(r.get("cluster_key")) subject = _s(r.get("subject_name")) or _s(r.get("segment_name")) if not key or not subject: continue # 簇键或主题名缺一样，这一行没法参与逐版本比对，跳过 out.append({ "scope": _s(r.get("scope")), "subject_name": subject, "segment_name": _s(r.get("segment_name")), "cluster_key": key, "leaning": _s(r.get("leaning")), "n_bull": _i(r.get("n_bull")), "n_bear": _i(r.get("n_bear")), "n_flags": _i(r.get("n_flags")), "verified": _b(r.get("verified")), "n_verify_problems": _n_items(r.get("verify_problems")), "n_materials": _i(r.get("n_materials")), "model": _s(r.get("model")), "rounds": _i(r.get("rounds")), "review_date": _ymd(r.get("review_date")), "reviewed_at": _dt_str(r.get("reviewed_at")), "input_version": _s(r.get("input_version")), "review_id": _s(r.get("review_id")), }) return out def _dt_str(v) -> str | None: """时刻列统一成 MySQL 认的 'YYYY-MM-DD HH:MM:SS'；只有日期的补零点；认不出返回空。""" if v is None or (isinstance(v, float) and v != v): return None if isinstance(v, dt.datetime): return v.strftime("%Y-%m-%d %H:%M:%S") if isinstance(v, dt.date): return f"{v.isoformat()} 00:00:00" s = str(v).strip().replace("T", " ") if len(s) >= 19 and s[4] == "-" and s[7] == "-": return s[:19] day = _ymd(s) return f"{day} 00:00:00" if day else None # ============================================================================ # 计划环境段的市场四项（2026-09-03 方案第 1.4 节清单里"有"与"可自算"的项） # ============================================================================ def market_context(ds: str, read_pg=None, read_mysql=None) -> dict: """数据日 ds 的市场环境四项，全部只展示与复盘分组，不拦任何票。 turnover 两市成交额：指数日线表 zs_day_data 里上证与深成当日 amount 之和，以及相对前五个 交易日均值的比值（原表单位，未换算） breadth 市场广度：基座个股日行情视图当日行自算——上涨、下跌、平盘家数，涨幅达 9.8% 的家数 （涨停近似），涨幅中位数 margin 融资：eastmoney_rzrq_data 最新一日的 financing_balance 与 change_percent_5d fear_greed 恐贪指数：fear_greed_index 最新一日的 index_value 与日期 每一项读失败为 None 并把原因记进 errors，不阻断出计划。read_pg / read_mysql 可注入（离线单测）。 融资与恐贪两张表按"最新一日"取，不按 ds 过滤：它们是 T+1 更新的情绪读数，计划日早晨看到的 就是最新一行；行里带日期，读者自己判断新鲜度。""" rpg = read_pg or db.read_pg rmy = read_mysql or db.read_mysql src = config.MARKET_MYSQL_SOURCE out = {"date": ds, "turnover": None, "breadth": None, "margin": None, "fear_greed": None, "errors": {}, "fetched_at": dt.datetime.now().isoformat(timespec="seconds")} # 一、两市成交额。不按日期列过滤（列的类型未在本仓库内核实：DATE 与整数 YYYYMMDD 的比较 # 语义不同），改为取每个指数最近的几十行，在 Python 里按归一化日期筛不晚于 ds 的行。 try: marks = ",".join(["%s"] * len(MARKET_INDEX_CODES)) rows = _records(rmy( src, f"SELECT symbol, `timestamp` AS d, amount FROM zs_day_data " f"WHERE symbol IN ({marks}) ORDER BY `timestamp` DESC LIMIT 80", tuple(MARKET_INDEX_CODES))) out["turnover"] = _turnover(rows, ds) if out["turnover"] is None: out["errors"]["turnover"] = "zs_day_data 最近 40 个交易日内没有不晚于计划日、且两市齐全的行" except Exception as e: # noqa: BLE001 out["errors"]["turnover"] = repr(e) print(f" （两市成交额读取失败，环境段该项为空: {e!r}）") # 二、市场广度：基座个股日行情视图当日全部行自算。 try: rows = _records(rpg( "SELECT pct_change FROM v_factor_stock_daily WHERE trade_date = %s", (ds,))) out["breadth"] = _breadth(rows) if out["breadth"] is None: out["errors"]["breadth"] = "v_factor_stock_daily 当日无行" except Exception as e: # noqa: BLE001 out["errors"]["breadth"] = repr(e) print(f" （市场广度自算失败，环境段该项为空: {e!r}）") # 三、融资余额与五日变化；四、恐贪指数。两张表都取最新一日一行。 for key, table, cols, label in ( ("margin", "eastmoney_rzrq_data", ("financing_balance", "change_percent_5d"), "融资余额"), ("fear_greed", "fear_greed_index", ("index_value",), "恐贪指数")): try: row, date_col = _latest_row(rmy, src, table) if row is None: out["errors"][key] = f"{table} 为空表" continue item = {"date": _ymd(row.get(date_col)) if date_col else None, "date_col": date_col, "source": table} for c in cols: item[c] = _f(row.get(c)) if all(item[c] is None for c in cols): out["errors"][key] = f"{table} 最新行缺列 {cols}（实际列: {sorted(row)[:12]}）" continue out[key] = item except Exception as e: # noqa: BLE001 out["errors"][key] = repr(e) print(f" （{label}读取失败，环境段该项为空: {e!r}）") return out def _turnover(rows: list[dict], ds: str) -> dict | None: """两市成交额：按日期把两个指数的 amount 相加，只认两市齐全的日子；当日取不晚于 ds 的最近一日， 前五日均值取它之前的五个交易日（不足五个按实际个数）。""" by_day: dict[str, dict] = {} for r in rows: d = _ymd(r.get("d")) a = _f(r.get("amount")) if not d or a is None or d > ds: continue by_day.setdefault(d, {})[str(r.get("symbol") or "").strip()] = a full = sorted((d for d, m in by_day.items() if all(c in m for c in MARKET_INDEX_CODES)), reverse=True) if not full: return None day0 = full[0] amt0 = sum(by_day[day0].values()) prev = [sum(by_day[d].values()) for d in full[1:6]] avg5 = (sum(prev) / len(prev)) if prev else None return {"data_date": day0, "amount": amt0, "prev5_avg": avg5, "ratio_vs_prev5": (amt0 / avg5) if avg5 else None, "prev5_days": len(prev), # 2026-09-03 实测核对：09-02 两市 amount 合计 1,321,831,143，而当日真实成交额约 # 1.32 万亿元，所以这一列的单位是千元，不是元。换算成亿元要除以十万（乘一千再除一亿）。 # 渲染与下游一律用 amount_yi，原始值保留在 amount 里备查。 "unit": "千元", "amount_yi": amt0 / 1e5, "prev5_avg_yi": (avg5 / 1e5) if avg5 else None, "source": "zs_day_data 上证 000001.SH 与深成 399001.SZ 当日 amount 之和"} def _breadth(rows: list[dict]) -> dict | None: pcts = [p for p in (_f(r.get("pct_change")) for r in rows) if p is not None] if not pcts: return None return {"n": len(pcts), "up": sum(1 for p in pcts if p > 0), "down": sum(1 for p in pcts if p < 0), "flat": sum(1 for p in pcts if p == 0), "limit_up_approx": sum(1 for p in pcts if p >= LIMIT_UP_PCT), "pct_median": round(statistics.median(pcts), 3), "limit_up_rule": f"涨幅达 {LIMIT_UP_PCT}% 记为涨停近似", "source": "v_factor_stock_daily 当日行自算"} def _latest_row(rmy, src: str, table: str) -> tuple[dict | None, str | None]: """取一张表按日期列排序的最新一行。日期列名先用一行样本探出（候选名见 _DATE_COL_CANDIDATES）， 探不到就按第一列倒序（通常是自增主键）并把 date_col 记为 None。""" sample = _records(rmy(src, f"SELECT * FROM {table} LIMIT 1")) if not sample: return None, None cols = list(sample[0].keys()) date_col = next((c for c in _DATE_COL_CANDIDATES if c in cols), None) order = f"`{date_col}`" if date_col else "1" rows = _records(rmy(src, f"SELECT * FROM {table} ORDER BY {order} DESC LIMIT 1")) return (rows[0] if rows else None), date_col # 丙路（券商行动）两个窗口各自的长度，自然日。等长是硬要求：窗口不等长会让八成的票 # 假显示覆盖收缩——实测前 135 天对近 45 天时有 907 只票误报。等长本身也是抗抖动的低通。 BROKER_WINDOW_DAYS = 45 def broker_reports(codes, ds: str, *, days: int = BROKER_WINDOW_DAYS * 2, read_mysql=None) -> list: """券商研报明细表里这些票近 days 个自然日的每股收益与市盈率预测——原始行，按机构分箱之前。 券商行动（两个等长窗口）与安全边际三情景（整段窗口）共用这一次取数，两路看的是同一批研报。 每行归一成 {k 前缀码, date 报告日, quarter 预测期, org 机构, eps, pe}；没有每股收益或 预测期的行不要。读失败返回空列表并打印原因，两路都按缺席处理，计划不断产。 """ reader = read_mysql or db.read_mysql end = dt.date.fromisoformat(ds) start = end - dt.timedelta(days=int(days)) dotted = sorted({_to_dot(c) for c in codes if c}) if not dotted: return [] marks = ",".join(["%s"] * len(dotted)) try: df = reader( "factor", f"SELECT ts_code, report_date, quarter, org_name, eps, pe FROM gp_report_rc " f"WHERE ts_code IN ({marks}) AND report_date > %s AND report_date <= %s " f"AND eps IS NOT NULL AND quarter IS NOT NULL", tuple(dotted) + (start.isoformat(), end.isoformat())) except Exception as e: # noqa: BLE001 print(f" （券商研报明细表读取失败，券商行动与安全边际两路整体缺席: {e!r}）") return [] out = [] for r in _records(df): d = _ymd(r.get("report_date")) q = str(r.get("quarter") or "").strip() eps = _f(r.get("eps")) if not d or not q or eps is None: continue out.append({"k": common.to_prefix(str(r.get("ts_code") or "").strip()), "date": d, "quarter": q, "org": str(r.get("org_name") or "").strip() or "未署名", "eps": eps, "pe": _f(r.get("pe"))}) return out def _latest_by_org(rows) -> dict: """同一家机构在窗口里可能发多篇，只留最近一篇：{预测期: {机构: 行}}。""" box: dict = defaultdict(dict) for r in rows: slot = box[r["quarter"]] if r["org"] not in slot or r["date"] > slot[r["org"]]["date"]: slot[r["org"]] = r return box def broker_actions(codes, ds: str, *, window_days: int = BROKER_WINDOW_DAYS, read_mysql=None, rows=None) -> dict: """券商用行动说话这一路：同一财年同一预测期的每股收益预测中位数与覆盖机构数， 比较最近两个等长窗口。返回前缀码到 logic_state.signal 的字典（算不出的票不进字典）。 三条口径必须照做，否则读数是错的： 一，两个窗口等长（见上面那条常量的说明）。 二，同一财年才可比，按预测期字段精确匹配，跨财年比较没有意义。 三，同一家机构在窗口里可能发多篇，先按机构取最近一篇再算中位数， 否则发得勤的机构会被重复计入。 看的是券商的行动不是言辞——券商极少明说不看好某个行业，所以等不到它开口， 只能看预测在不在下修、覆盖在不在收缩。 rows 可传 broker_reports 的返回（计划装配一次取数两路共用）；不传就自己取。 """ import logic_state as ls if rows is None: rows = broker_reports(codes, ds, days=int(window_days) * 2, read_mysql=read_mysql) end = dt.date.fromisoformat(ds) mid = (end - dt.timedelta(days=int(window_days))).isoformat() start = (end - dt.timedelta(days=int(window_days) * 2)).isoformat() by_code: dict = defaultdict(list) for r in rows: if start < r["date"] <= ds: # rows 可能来自更长的窗口，这里再裁一次 by_code[r["k"]].append(r) out = {} for k, rs in by_code.items(): now_box = _latest_by_org([r for r in rs if r["date"] > mid]) prev_box = _latest_by_org([r for r in rs if r["date"] <= mid]) usable = [(q, now_box[q], prev_box[q]) for q in now_box if q in prev_box] if not usable: continue q, n, p = max(usable, key=lambda t: len(t[1]) + len(t[2])) now = {"eps": statistics.median([r["eps"] for r in n.values()]), "firms": len(n)} prev = {"eps": statistics.median([r["eps"] for r in p.values()]), "firms": len(p)} sig = ls.from_broker(now, prev, as_of=ds) if sig["refs"]: sig["refs"][0]["quarter"] = q out[k] = sig return out # ============================================================================ # 安全边际三情景（2026-09-07 下一阶段方案第四件）：只展示，不进判决 # ============================================================================ # # 回答的是价值投资的经典问题：买入价相对内在价值的折扣有多少，最坏情况下现价还有多少 # 下跌空间。这是事前概念，与 PMS 里那个也叫"垫"的字段（建仓后的浮盈）不是一回事。 # # 口径：同一预测期的每股收益预测与市盈率预测，按机构去重后各取最小、中位、最大—— # 悲观 = 最低每股收益 × 最低市盈率，中性 = 两项中位数，乐观 = 两项最高。 # 隐含市盈率 = 现价 ÷ 中位每股收益；赔率 = 中性上行 ÷ 悲观下行。 # 用盈利预测不用目标价：目标价字段覆盖不到三成且不去重不加权，出现过 +181% 的读数； # 每股收益预测覆盖 98%、市盈率预测 91%。恩捷股份 09-02 手算（方案四之三）：悲观 42.64、 # 中性 56.75、乐观 84.96，赔率 0.87 比 1——单测用它对表。 # 机构分歧"极大"的线：每股收益或市盈率预测的最高对最低超过这个倍数就标注。取 3 倍——正常的 # 分歧在一倍多到两倍之间（恩捷 1.3 与 1.5 倍），三倍以上多半是有机构的口径不同或数据有误。 # 一次定死，只影响标注，不影响算法。 WIDE_SPREAD = 3.0 def _pick_period(box: dict): """挑哪个预测期：优先年度（预测期以 Q4 结尾），其中覆盖机构最多的；同数取更近的年份。""" cands = [(q, len(orgs)) for q, orgs in box.items() if orgs] if not cands: return None annual = [c for c in cands if c[0].upper().endswith("Q4")] pool = annual or cands return sorted(pool, key=lambda c: (-c[1], c[0]))[0][0] def scenarios(eps_vals, pe_vals, price, *, quarter=None, as_of=None, min_firms: int = 2) -> dict: """纯函数：三情景估值。算不出时 na 写明原因（四种不适用各一句人话），算得出时 na 为 None。""" eps_vals = [float(x) for x in (eps_vals or []) if x is not None] pe_vals = [float(x) for x in (pe_vals or []) if x is not None and float(x) > 0] base = {"quarter": quarter, "firms": len(eps_vals), "as_of": as_of, "price": None if price is None else float(price)} if len(eps_vals) < int(min_firms): return {**base, "na": f"覆盖机构只有 {len(eps_vals)} 家，不足 {min_firms} 家"} if price is None or float(price) <= 0: return {**base, "na": "现价取不到"} if min(eps_vals) <= 0: return {**base, "na": "每股收益预测有负值或零，市盈率口径不适用"} if len(pe_vals) < int(min_firms): return {**base, "na": f"市盈率预测只有 {len(pe_vals)} 家给了，不足 {min_firms} 家"} px = float(price) e = {"min": min(eps_vals), "med": statistics.median(eps_vals), "max": max(eps_vals)} p = {"min": min(pe_vals), "med": statistics.median(pe_vals), "max": max(pe_vals)} pess, neut, opt = e["min"] * p["min"], e["med"] * p["med"], e["max"] * p["max"] down, up_n, up_o = pess / px - 1, neut / px - 1, opt / px - 1 odds = None note = None if down >= 0: note = "悲观情景仍高于现价，没有下行空间可比" elif up_n <= 0: note = "中性情景低于现价，赔率不成立" else: odds = up_n / (-down) # 机构分歧的量：最高对最低的倍数。悲观取最低乘最低、乐观取最高乘最高，分歧一大两头就会被 # 放大到离谱（实测有票悲观 -89%、乐观 +1054%）。超过阈值只标注"分歧极大"，不改算法—— # 这本身就是一条信息：券商对这家公司的盈利路径没有共识。 spread = {"eps": round(e["max"] / e["min"], 2), "pe": round(p["max"] / p["min"], 2)} wide = spread["eps"] > WIDE_SPREAD or spread["pe"] > WIDE_SPREAD return {**base, "na": None, "eps": {k: round(v, 4) for k, v in e.items()}, "pe": {k: round(v, 2) for k, v in p.items()}, "pess": round(pess, 2), "neut": round(neut, 2), "opt": round(opt, 2), "down": round(down, 4), "up_neut": round(up_n, 4), "up_opt": round(up_o, 4), "implied_pe": round(px / e["med"], 1), "odds": None if odds is None else round(odds, 2), "note": note, "spread": spread, "wide": wide} def valuation_scenarios(codes, ds: str, prices: dict, *, rows=None, window_days: int = BROKER_WINDOW_DAYS * 2, min_firms: int = 2, read_mysql=None) -> dict: """每只票的三情景估值，按前缀码索引；没有任何研报行的票给 None（卡上写"没有券商预测"）。""" if rows is None: rows = broker_reports(codes, ds, days=int(window_days), read_mysql=read_mysql) by_code: dict = defaultdict(list) for r in rows: by_code[r["k"]].append(r) out = {} for k in {str(c) for c in (codes or []) if c}: rs = by_code.get(k) or [] if not rs: out[k] = None continue box = _latest_by_org(rs) q = _pick_period(box) firms = box.get(q, {}) if q else {} out[k] = scenarios([r["eps"] for r in firms.values()], [r["pe"] for r in firms.values()], (prices or {}).get(k), quarter=q, as_of=max((r["date"] for r in firms.values()), default=None), min_firms=min_firms) return out def _prefix_any(s: str) -> str: """行情表的代码列可能是 600000.SH，也可能是裸 6 位码；统一成前缀式。""" s = str(s or "").strip().upper() if "." in s: return common.to_prefix(s) if len(s) == 6 and s.isdigit(): return ("SH" if s[0] == "6" else ("BJ" if s[0] in "48" else "SZ")) + s return s def close_prices(ds: str, read_mysql=None, code_col: str | None = None) -> dict: """数据日的收盘价（前复权行情表，与预期空间同一来源），按前缀码索引；读不到返回空字典。 区间写成 [ds, ds+1) 而不是 = ds：时间列若带时分秒，等号会一行都对不上，而这样写两种 形态都对、也走得了索引。code_col 可注入（离线单测），不传就沿用预期空间那一路的探列。 """ reader = read_mysql or db.read_mysql try: if code_col is None: import factors code_col = factors._price_code_col() # noqa: SLF001 —— 同仓自用 nxt = (dt.date.fromisoformat(ds) + dt.timedelta(days=1)).isoformat() df = reader("price", f"SELECT `{code_col}` AS ts_code, close FROM gp_day_data " f"WHERE `timestamp` >= %s AND `timestamp` < %s", (ds, nxt)) except Exception as e: # noqa: BLE001 print(f" （收盘价读取失败，安全边际这一行整体缺席: {e!r}）") return {} out = {} for r in _records(df): k = _prefix_any(r.get("ts_code")) px = _f(r.get("close")) if k and px and px > 0: out[k] = px return out # ============================================================================ # 催化事件与事件日字段（2026-09-08《量价研判链吸收方案》3.4）：只展示，不进判决 # ============================================================================ # # 研报的起点是四类券商正向事件："间隔一年后深度覆盖推荐买入"、"主动上调盈利预测"、"研报标题含 # 业绩超预期"、"两者兼有"。这四类全部能从券商研报明细表 gp_report_rc 复现（报告类型、标题、评级、 # 每股收益历史都在），不用动数据基座。阈值一次定死（台账 045）： # 深度覆盖 报告类型是"深度"，评级是买入类，且该票在这篇之前 365 天内没有任何研报 # 上调预测 同一家机构对同一预测期，180 天内上一篇每股收益为正且这一篇高出五成以上 # 超预期 标题含"超预期" # 事件窗口 数据日往前 60 个自然日；同一天多篇合并成一个事件（研报的"同日复合"） # 它是论点卡"催化剂"一栏的第一个数据源（此前标无数据源），也是送研判的事件上下文。 EVENT_WINDOW_DAYS = 60 EVENT_COVER_GAP_DAYS = 365 EVENT_UPGRADE_LOOKBACK_DAYS = 180 EVENT_UPGRADE_RATIO = 1.5 EVENT_KEEP = 5 # 每票最多带几条事件到卡上 BUY_RATINGS = {"买入", "增持", "推荐", "强烈推荐", "强推", "跑赢行业", "优于大市", "买入-A", "买入-B", "推荐-A", "增持-A", "审慎增持", "谨慎增持", "优于大市评级"} EV_DEEP, EV_UPGRADE, EV_BEAT = "深度覆盖", "上调盈利预测", "业绩超预期" def analyst_reports(codes, ds: str, *, days: int = EVENT_WINDOW_DAYS + EVENT_COVER_GAP_DAYS, read_mysql=None) -> list: """券商研报明细表的原始行：报告日、类型、标题、评级、机构、预测期、每股收益。窗口要盖住 事件窗口加"前 365 天有没有覆盖"的回看，所以默认取 425 天。读失败返回空列表并打印原因。""" reader = read_mysql or db.read_mysql end = dt.date.fromisoformat(ds) start = end - dt.timedelta(days=int(days)) dotted = sorted({_to_dot(c) for c in codes if c}) if not dotted: return [] marks = ",".join(["%s"] * len(dotted)) try: df = reader( "factor", f"SELECT ts_code, report_date, report_type, report_title, rating, org_name, quarter, eps " f"FROM gp_report_rc WHERE ts_code IN ({marks}) AND report_date > %s AND report_date <= %s", tuple(dotted) + (start.isoformat(), end.isoformat())) except Exception as e: # noqa: BLE001 print(f" （券商研报明细表读取失败，催化事件这一行整体缺席: {e!r}）") return [] out = [] for r in _records(df): d = _ymd(r.get("report_date")) if not d: continue out.append({"k": common.to_prefix(str(r.get("ts_code") or "").strip()), "date": d, "type": str(r.get("report_type") or "").strip(), "title": str(r.get("report_title") or "").strip(), "rating": str(r.get("rating") or "").strip(), "org": str(r.get("org_name") or "").strip() or "未署名", "quarter": str(r.get("quarter") or "").strip(), "eps": _f(r.get("eps"))}) return out def analyst_events(codes, ds: str, *, rows=None, window_days: int = EVENT_WINDOW_DAYS, read_mysql=None) -> dict: """每只票近 window_days 天的券商正向事件，按前缀码索引；没有事件的票不在字典里。 返回 {k: {"latest": 最近事件日, "count": 事件天数, "events": [{date, types, orgs, title, n_reports, compound}, ...] 最新在前}}。types 是这一天命中的事件类型列表；compound 为真表示同一篇研报 同时命中上调预测与超预期（研报的第四类），或同一天多篇研报命中不同类型。""" if rows is None: rows = analyst_reports(codes, ds, read_mysql=read_mysql) by_code: dict = defaultdict(list) for r in rows: by_code[r["k"]].append(r) try: cut = (dt.date.fromisoformat(ds) - dt.timedelta(days=int(window_days))).isoformat() except ValueError: return {} out = {} for k, rs in by_code.items(): rs = sorted(rs, key=lambda r: r["date"]) dates = [r["date"] for r in rs] by_org_q: dict = defaultdict(list) for r in rs: if r["eps"] is not None: by_org_q[(r["org"], r["quarter"])].append(r) days_hit: dict = {} for r in rs: if r["date"] <= cut or r["date"] > ds: continue types = [] if "超预期" in r["title"]: types.append(EV_BEAT) if r["type"] == "深度" and r["rating"] in BUY_RATINGS: gap_start = (dt.date.fromisoformat(r["date"]) - dt.timedelta(days=EVENT_COVER_GAP_DAYS)).isoformat() # 这篇之前 365 天内有没有任何研报（不含同一天） if not any(gap_start < d < r["date"] for d in dates): types.append(EV_DEEP) if r["eps"] is not None and r["eps"] > 0: look = (dt.date.fromisoformat(r["date"]) - dt.timedelta(days=EVENT_UPGRADE_LOOKBACK_DAYS)).isoformat() prev = [p for p in by_org_q[(r["org"], r["quarter"])] if look < p["date"] < r["date"] and p["eps"] and p["eps"] > 0] if prev and r["eps"] >= EVENT_UPGRADE_RATIO * prev[-1]["eps"]: types.append(EV_UPGRADE) if not types: continue ev = days_hit.setdefault(r["date"], {"date": r["date"], "types": [], "orgs": [], "title": r["title"][:60], "n_reports": 0, "compound": False}) for t in types: if t not in ev["types"]: ev["types"].append(t) if r["org"] not in ev["orgs"]: ev["orgs"].append(r["org"]) ev["n_reports"] += 1 if EV_UPGRADE in types and EV_BEAT in types: ev["compound"] = True if not days_hit: continue events = sorted(days_hit.values(), key=lambda e: e["date"], reverse=True) for e in events: if len(e["types"]) > 1: e["compound"] = True out[k] = {"latest": events[0]["date"], "count": len(events), "events": events[:EVENT_KEEP]} return out # 事件日字段：研报"输入事实"的第三块。事件前 5 日与 20 日涨幅判有没有抢跑，事件日跳空、日内收益、 # 收盘位置、量比、涨停判市场有没有确认。行情表 gp_day_data 是前复权（用户 09-08 确认），跳空与 # 涨幅直接算。没有事件的票按数据日算同一组数（那时它回答的是"今天这根 K 线的样子"）。 PRICE_HISTORY_DAYS = 100 # 有事件的票取多少天行情：事件最远 60 天，前面还要 20 个交易日算量与涨幅 PRICE_HISTORY_DAYS_NO_EVENT = 35 # 没有事件的票按数据日算，只要 21 个交易日的历史 def _limit_up_threshold(k: str) -> float: """涨停线（百分数）：创业板与科创板两成，北交所三成，其余一成。用 9.8 而不是 10 是给四舍五入留余地。""" s = str(k or "") num = s[2:] if len(s) == 8 else s if num.startswith(("30", "68")): return 19.8 if num.startswith(("4", "8")): return 29.8 return 9.8 def price_history(codes, ds: str, *, days: int = PRICE_HISTORY_DAYS, read_mysql=None, code_col: str | None = None) -> dict: """每只票近 days 个自然日的前复权日线，按前缀码索引、按日升序。行情表的代码列是前缀式。""" reader = read_mysql or db.read_mysql codes = sorted({_prefix_any(c) for c in codes if c}) if not codes: return {} try: if code_col is None: import factors code_col = factors._price_code_col() # noqa: SLF001 —— 同仓自用 end = dt.date.fromisoformat(ds) start = (end - dt.timedelta(days=int(days))).isoformat() nxt = (end + dt.timedelta(days=1)).isoformat() marks = ",".join(["%s"] * len(codes)) df = reader("price", f"SELECT `{code_col}` AS ts_code, `timestamp` AS d, open, high, low, close, " f"pre_close, percent, volume FROM gp_day_data " f"WHERE `timestamp` >= %s AND `timestamp` < %s AND `{code_col}` IN ({marks})", (start, nxt) + tuple(codes)) except Exception as e: # noqa: BLE001 print(f" （行情表读取失败，事件日字段与定价状态整体缺席: {e!r}）") return {} out: dict = defaultdict(list) for r in _records(df): k = _prefix_any(r.get("ts_code")) d = _ymd(r.get("d")) if not k or not d: continue out[k].append({"date": d, "open": _f(r.get("open")), "high": _f(r.get("high")), "low": _f(r.get("low")), "close": _f(r.get("close")), "pre_close": _f(r.get("pre_close")), "pct": _f(r.get("percent")), "volume": _f(r.get("volume"))}) return {k: sorted(v, key=lambda r: r["date"]) for k, v in out.items()} def event_day_fields(codes, ds: str, events: dict, *, hist=None, read_mysql=None, code_col: str | None = None) -> dict: """每只票的事件日字段，按前缀码索引。events 是 analyst_events 的返回；没有事件的票按数据日算。 六个数：事件前 5 日与 20 日涨幅、事件日跳空幅度（开盘对前收）、日内收益（收盘对开盘）、 收盘位置（收盘在当日高低区间里的位置，0 到 1）、事件日量比（对此前 20 个交易日均量）， 外加当日涨幅与是否涨停。行情不够时相应字段为 None，不硬算。""" if hist is None: # 两段取：没有事件的票按数据日算，只要 21 个交易日的历史（35 个自然日够）；有事件的票 # 事件最远 60 天，前面再要 20 个交易日，取 100 天。一段取 100 天要拉十二万行、十秒多， # 计划的实时重算撑不起（PMS 拉计划的超时是 60 秒），分两段行数少六成。 hist = price_history(codes, ds, days=PRICE_HISTORY_DAYS_NO_EVENT, read_mysql=read_mysql, code_col=code_col) with_event = [k for k in (events or {}) if k in {str(c).strip() for c in (codes or []) if c}] if with_event: hist.update(price_history(with_event, ds, days=PRICE_HISTORY_DAYS, read_mysql=read_mysql, code_col=code_col)) out = {} for k in {str(c).strip() for c in (codes or []) if c}: rows = hist.get(k) or [] if not rows: continue ev = (events or {}).get(k) target = ev["latest"] if ev else ds idx = None for i, r in enumerate(rows): if r["date"] <= target: idx = i if idx is None: continue cur = rows[idx] closes = [r["close"] for r in rows] def _ret(back: int): if idx - back < 0 or not closes[idx - 1] or not closes[idx - back - 1]: return None return closes[idx - 1] / closes[idx - back - 1] - 1 prev_close = cur["pre_close"] or (closes[idx - 1] if idx >= 1 else None) gap = (cur["open"] / prev_close - 1) if cur["open"] and prev_close else None intraday = (cur["close"] / cur["open"] - 1) if cur["close"] and cur["open"] else None rng = (cur["high"] - cur["low"]) if cur["high"] is not None and cur["low"] is not None else None close_pos = ((cur["close"] - cur["low"]) / rng) if rng and cur["close"] is not None else None vols = [r["volume"] for r in rows[max(0, idx - 20):idx] if r["volume"]] vol_ratio = (cur["volume"] / (sum(vols) / len(vols))) if cur["volume"] and len(vols) >= 5 else None pct = cur["pct"] out[k] = {"event_date": cur["date"], "has_event": bool(ev), "pre5": _r4(_ret(5)), "pre20": _r4(_ret(20)), "gap": _r4(gap), "intraday": _r4(intraday), "close_pos": _r4(close_pos), "vol_ratio": None if vol_ratio is None else round(vol_ratio, 2), "day_pct": None if pct is None else round(pct / 100.0, 4), "limit_up": None if pct is None else bool(pct >= _limit_up_threshold(k)), "history_days": idx} return out 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 # --------------------------------------------------------------------------- # 行业催化（2026-09-08，台账 051）：数据基座按环节评析的材料另标出来的"行业级催化事件"， # 挂在环节上不挂个股（表 segment_catalysts，基座 PG）。候选卡按票所在的被指向环节取， # 只展示、只分组复盘，不进判决、不进逻辑状态。表没建或没材料就整体缺席，不断产。 # --------------------------------------------------------------------------- CATALYST_WINDOW_DAYS = 180 CATALYST_MAX_PER_SEGMENT = 3 _MAG_ORDER = {"大": 0, "中": 1, "小": 2} def segment_catalysts(segments, ds: str, *, days: int = CATALYST_WINDOW_DAYS, read_pg=None, limit_per_segment: int = CATALYST_MAX_PER_SEGMENT) -> dict: """{环节名: [{segment, event_date, title, direction, magnitude, horizon, confidence, mechanism}, ...]}， 每个环节按量级大→小、事件日新→旧排，最多 limit_per_segment 条。""" rp = read_pg or db.read_pg want = sorted({str(x) for x in (segments or []) if x}) if not want: return {} try: since = (dt.date.fromisoformat(ds) - dt.timedelta(days=int(days))).isoformat() except ValueError: return {} try: df = rp("SELECT segment_name, event_date, first_seen, title, direction, magnitude, horizon, " "confidence, mechanism FROM segment_catalysts " "WHERE segment_name = ANY(%s) AND coalesce(event_date, first_seen) >= %s", (want, since)) except Exception as e: # noqa: BLE001 —— 表没建、连接失败：整体缺席 logging.getLogger("bridge.catalyst").warning("segment_catalysts 读不到: %r", e) return {} rows = [] for r in df.to_dict("records"): ev = r.get("event_date") or r.get("first_seen") rows.append({"segment": str(r.get("segment_name")), "event_date": str(ev)[:10] if ev else None, "title": str(r.get("title") or "")[:80], "direction": r.get("direction") or "利好", "magnitude": r.get("magnitude") or "小", "horizon": r.get("horizon") or "短期", "confidence": r.get("confidence") or "低", "mechanism": (str(r.get("mechanism") or "")[:120] or None)}) rows.sort(key=lambda x: (_MAG_ORDER.get(x["magnitude"], 9), x["event_date"] or ""), reverse=False) rows.sort(key=lambda x: (_MAG_ORDER.get(x["magnitude"], 9), -(int((x["event_date"] or "0000-00-00").replace("-", "") or 0)))) out: dict = {} for x in rows: slot = out.setdefault(x["segment"], []) if len(slot) < limit_per_segment: slot.append(x) return out def catalysts_for_codes(seg_of: dict, cat_by_seg: dict, *, limit: int = 3) -> dict: """按票汇总所在环节的行业催化：同标题去重，量级大→小，最多 limit 条。没有的票不在字典里。""" out: dict = {} for k, segs in (seg_of or {}).items(): seen: set = set() lst = [] for seg in segs or []: for x in cat_by_seg.get(seg) or []: if x["title"] in seen: continue seen.add(x["title"]) lst.append(x) if lst: lst.sort(key=lambda x: (_MAG_ORDER.get(x["magnitude"], 9), -(int((x["event_date"] or "0000-00-00").replace("-", "") or 0)))) 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 {} if age is not None and age < 0: age = 0 # 报告运行日晚于数据日（今早刚跑）：算零天，不算负数 out[k] = {"period": _clean_str(r.get("period")), "review_version": _clean_str(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": _clean_str(r.get("brain_status")) or _clean_str(brain.get("status")), "brain_used": bool(r.get("brain_used")) if r.get("brain_used") not in (None, "") and r.get("brain_used") == r.get("brain_used") else False, "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 _clean_str(v): """空列经 pandas 会变成 NaN（浮点），JSON 序列化直接失败；统一收成 None。""" if v is None: return None if isinstance(v, float) and v != v: return None s = str(v).strip() return s or None 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