"""赛道映射（硬门槛 C 的实现载体，设计 §3.2）。 config/frontier_tracks.yml 是唯一事实源，三条映射路径并行： kg_themes 主题名 → industry_pools（弱锚 fallback，source_rule=pool_theme） kg_chains 链名 → 图谱 IN_SEGMENT 边的产业链修饰符（强锚，graph_chain） kg_segments 环节名 → 图谱环节投影（强锚，graph_segment） 后两条读基座第五插槽视图 v_factor_segment_members（akg.project_segments 每晨 06:10 刷新）。同一只股票多路命中时**图谱路径优先记账**——这就是设计 文档"两级 source_rule：graph 强 / fallback 弱"的落地。成员表快照落 data/track_members_<日期>.csv，每行带 source_rule，可审计到入选依据。 """ import datetime as dt import json import os import pandas as pd import config import db try: import yaml except ImportError: # 镜像未装 PyYAML 时给出可执行的修复指令，而不是裸崩 yaml = None def _need_yaml(): if yaml is None: raise SystemExit( "缺 PyYAML：requirements.txt 已加，请在桥机重建镜像——\n" " docker compose build akg-factor-bridge && " "docker compose up -d akg-factor-bridge") def load_yml(path: str | None = None) -> dict: _need_yaml() with open(path or config.TRACKS_YML, "r", encoding="utf-8") as f: return yaml.safe_load(f) def _pool_theme_members() -> dict: """industry_pools 当前态：主题名 → 成员列表（dict，含 ts_code/name）。""" pools = db.read_pg("SELECT theme, members FROM industry_pools") by_theme = {} for _, r in pools.iterrows(): ms = r["members"] if isinstance(ms, str): ms = json.loads(ms) by_theme[str(r["theme"]).strip()] = ms or [] return by_theme def _graph_members(): """环节投影（第五插槽）→ 两个索引：环节名→成员、链名→成员（仅上市成员）。 视图不可用（投影表未建/基座不可达）时返回空索引并提示——图谱路径跳过， 主题路径照常工作，不挡任何调用方。空串链名（抽取未写修饰的边）在此忽略。 """ try: df = db.read_pg( "SELECT segment_name, ts_code, member_name, chain " "FROM v_factor_segment_members WHERE ts_code IS NOT NULL") except Exception as e: # noqa: BLE001 print(f" (环节投影视图不可用，图谱映射路径跳过: {e!r})") return {}, {} by_seg, by_chain = {}, {} for r in df.itertuples(): m = (str(r.ts_code), None if pd.isna(r.member_name) else str(r.member_name)) by_seg.setdefault(str(r.segment_name).strip(), []).append(m) if r.chain is not None and not pd.isna(r.chain) and str(r.chain).strip(): by_chain.setdefault(str(r.chain).strip(), []).append(m) return by_seg, by_chain def resolve_members(only_confirmed: bool = True, path: str | None = None, dedup: bool = True): """yml + 三路数据源 → (成员表, 未命中键清单)。 成员表列：ts_code, name, track, status, theme, source_rule——theme 列放 命中它的那个键（主题名/链名/环节名）。图谱路径的行排在主题路径之前， 去重时同股同赛道保留最先出现的行 ⇒ 图谱锚优先记账。 未命中 = yml 里写了、数据源里查无此键，元素为 (赛道, 键类型, 键名)。 """ d = load_yml(path) by_theme = _pool_theme_members() by_seg, by_chain = _graph_members() excl = {str(x).strip() for x in (d.get("exclude_themes") or [])} rows, missing = [], [] for tr in d.get("tracks") or []: if only_confirmed and tr.get("status") != "confirmed": continue tname, status = tr["name"], tr.get("status") for seg in tr.get("kg_segments") or []: # 图谱路径（强锚）在前 s = str(seg).strip() if s not in by_seg: missing.append((tname, "环节", s)) continue for ts, nm in by_seg[s]: rows.append((ts, nm, tname, status, s, "graph_segment")) for ch in tr.get("kg_chains") or []: c = str(ch).strip() if c not in by_chain: missing.append((tname, "链", c)) continue for ts, nm in by_chain[c]: rows.append((ts, nm, tname, status, c, "graph_chain")) for theme in tr.get("kg_themes") or []: # 主题路径（fallback）在后 t = str(theme).strip() if t in excl: continue if t not in by_theme: missing.append((tname, "主题", t)) continue for m in by_theme[t]: ts = (m or {}).get("ts_code") if ts: rows.append((ts, m.get("name"), tname, status, t, "pool_theme")) df = pd.DataFrame(rows, columns=["ts_code", "name", "track", "status", "theme", "source_rule"]) if dedup: df = df.drop_duplicates(["ts_code", "track"]) return df, missing def snapshot(only_confirmed: bool = True, path: str | None = None): """成员表落 data/ 版本化快照（每行带 source_rule 与 updated_at，审计列）。""" df, missing = resolve_members(only_confirmed, path) os.makedirs("data", exist_ok=True) out = f"data/track_members_{dt.date.today().isoformat()}.csv" (df.assign(layer="", updated_at=dt.datetime.now().isoformat(timespec="seconds")) .to_csv(out, index=False)) return out, df, missing def coverage_report(path: str | None = None) -> pd.DataFrame: """赛道覆盖体检：逐赛道给出成员数（含图谱锚数）、各路径命中率，含 candidate。""" # dedup=False：命中率要在去重前数——成员完全被同赛道更早键覆盖的键， # 去重后一行不剩，会被误计为"没命中"（07-30 实测教训）。 df, missing = resolve_members(only_confirmed=False, path=path, dedup=False) d = load_yml(path) print("赛道覆盖体检（主题=industry_pools 当前态；图谱=环节投影；" "成员数已去重，命中率按去重前算）：") for tr in d.get("tracks") or []: sub = df[df["track"] == tr["name"]] graph_sub = sub[sub["source_rule"] != "pool_theme"] n_all = sub["ts_code"].nunique() n_graph = graph_sub["ts_code"].nunique() n_tkeys = len(tr.get("kg_themes") or []) hit_t = sub[sub["source_rule"] == "pool_theme"]["theme"].nunique() n_gkeys = (len(tr.get("kg_segments") or []) + len(tr.get("kg_chains") or [])) hit_g = graph_sub["theme"].nunique() miss = [f"{kind}:{key}" for name, kind, key in missing if name == tr["name"]] tag = "" if tr.get("status") == "confirmed" else "（candidate）" line = (f" {tr['name']}{tag}: 成员 {n_all} 只（图谱锚 {n_graph}）" f" | 主题命中 {hit_t}/{n_tkeys}") if n_gkeys: line += f" | 图谱键命中 {hit_g}/{n_gkeys}" if miss: line += f" | 未命中: {'、'.join(miss)}" print(line) conf = df[df["status"] == "confirmed"] n_conf = conf["ts_code"].nunique() n_conf_graph = conf[conf["source_rule"] != "pool_theme"]["ts_code"].nunique() print(f" —— confirmed 合计（去重）: {n_conf} 只，其中带图谱锚 {n_conf_graph} 只") return df def gate_simulation(path: str | None = None) -> None: """赛道闸演算（只读）：若 ENABLE_TRACK_GATE 置 1，主榜/观察档变成什么规模。 口径与开闸后的 factors._gate_of 一致： 主榜' = 现主榜 ∩ 赛道成员； 观察档' = 现观察档 ∪（无券商预期 ∧ 在赛道 ∧ 现为不采纳 ∧ 非风险股）。 读最新档位 / upside 因子表与 confirmed 成员表，不写任何东西。""" import common d = db.read_mysql("factor", "SELECT MAX(trade_date) d FROM t_factor_akg_gate") v = None if d.empty else d.iloc[0, 0] if v is None or pd.isna(v): print("（尚无档位表，跳过赛道闸演算）") return ds = pd.Timestamp(v).date().isoformat() g = db.read_mysql("factor", "SELECT stock_code, factor_value " "FROM t_factor_akg_gate WHERE trade_date = %s", (ds,)) up = db.read_mysql("factor", "SELECT stock_code FROM t_factor_akg_upside " "WHERE trade_date = %s", (ds,)) gate = g.set_index("stock_code")["factor_value"].astype(float) covered = set(up["stock_code"]) if not up.empty else set() df, _missing = resolve_members(only_confirmed=True, path=path) if df.empty: print("（赛道成员表为空，跳过演算）") return by_track = {t: {common.to_prefix(x) for x in sub["ts_code"]} for t, sub in df.groupby("track")} tset = set().union(*by_track.values()) try: import factors risk = factors._risk_set() # noqa: SLF001 —— 同仓自用 except Exception: # noqa: BLE001 —— 风险名单拿不到时演算照做 risk = set() main_now = set(gate[gate == 2.0].index) obs_now = set(gate[gate == 1.0].index) main_after = main_now & tset obs_extra = {k for k in tset if k not in covered and gate.get(k) == 0.0 and k not in risk} obs_after = obs_now | obs_extra print(f"\n赛道闸演算（截面 {ds}，假想 ENABLE_TRACK_GATE=1）：") print(f" 主榜: {len(main_now)} → {len(main_after)}" f"（保留 {len(main_after) / max(1, len(main_now)):.0%}）") print(f" 观察档: {len(obs_now)} → {len(obs_after)}" f"（并入赛道内无覆盖股 +{len(obs_extra)}）") parts = [f"{t} {len(main_now & s)}" for t, s in sorted(by_track.items(), key=lambda kv: -len(main_now & kv[1]))] print(f" 逐赛道主榜贡献（跨赛道有重叠）: {' | '.join(parts)}")