图谱双轨版

tracks.py

@@ -1,12 +1,13 @@
 """赛道映射（硬门槛 C 的实现载体，设计 §3.2）。
 
-config/frontier_tracks.yml 是唯一事实源：每个赛道列出 kg_themes
+config/frontier_tracks.yml 是唯一事实源，三条映射路径并行：
-（industry_pools 主题名）。本模块把它解析成 ts_code 级成员表，并落
+  kg_themes   主题名 → industry_pools（弱锚 fallback，source_rule=pool_theme）
-data/track_members_<日期>.csv 版本化快照——可 git diff、可审计：
+  kg_chains   链名   → 图谱 IN_SEGMENT 边的产业链修饰符（强锚，graph_chain）
-每只股票能追溯到因哪个赛道、哪个主题入选。
+  kg_segments 环节名 → 图谱环节投影（强锚，graph_segment）
-
+后两条读基座第五插槽视图 v_factor_segment_members（akg.project_segments
-当前唯一的映射路径是主题名（kg_segments / kg_concepts 要等基座的环节
+每晨 06:10 刷新）。同一只股票多路命中时**图谱路径优先记账**——这就是设计
-投影表建成，即三批-4），source_rule 统一记 'pool_theme'。
+文档"两级 source_rule：graph 强 / fallback 弱"的落地。成员表快照落
+data/track_members_<日期>.csv，每行带 source_rule，可审计到入选依据。
 """
 import datetime as dt
 import json
@@ -37,15 +38,8 @@ def load_yml(path: str | None = None) -> dict:
         return yaml.safe_load(f)
 
 
-def resolve_members(only_confirmed: bool = True, path: str | None = None,
+def _pool_theme_members() -> dict:
-                    dedup: bool = True):
+    """industry_pools 当前态：主题名 → 成员列表（dict，含 ts_code/name）。"""
-    """yml + industry_pools 当前态 → (成员表, 未命中主题清单)。
-
-    成员表列：ts_code, name, track, status, theme。同股同赛道多主题只留一行，
-    同股跨赛道保留多行。未命中 = yml 里写了、industry_pools 里查无此主题
-    （通常是主题改名或池尚未涌现，体检时重点看）。
-    """
-    d = load_yml(path)
     pools = db.read_pg("SELECT theme, members FROM industry_pools")
     by_theme = {}
     for _, r in pools.iterrows():
@@ -53,58 +47,122 @@ def resolve_members(only_confirmed: bool = True, path: str | None = None,
         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
-        for theme in tr.get("kg_themes") or []:
+        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((tr["name"], t))
+                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"), tr["name"],
+                    rows.append((ts, m.get("name"), tname, status, t, "pool_theme"))
-                                 tr.get("status"), t))
+    df = pd.DataFrame(rows, columns=["ts_code", "name", "track", "status",
-    df = pd.DataFrame(rows, columns=["ts_code", "name", "track",
+                                     "theme", "source_rule"])
-                                     "status", "theme"])
     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，审计列）。"""
+    """成员表落 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="", source_rule="pool_theme",
+    (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:
-    """赛道覆盖体检（G1 清单的首件事）：逐赛道成员数与主题命中率，含 candidate。"""
+    """赛道覆盖体检：逐赛道给出成员数（含图谱锚数）、各路径命中率，含 candidate。"""
-    # dedup=False：主题命中率要在去重前数——成员完全被同赛道更早主题收进来的
+    # dedup=False：命中率要在去重前数——成员完全被同赛道更早键覆盖的键，
-    # 主题（如卫星导航之于卫星），去重后一行不剩，会被误计成"没命中"（07-30 实测）。
+    # 去重后一行不剩，会被误计为"没命中"（07-30 实测教训）。
     df, missing = resolve_members(only_confirmed=False, path=path, dedup=False)
     d = load_yml(path)
-    print("赛道覆盖体检（industry_pools 当前态；成员数已去重，主题命中按去重前算）：")
+    print("赛道覆盖体检（主题=industry_pools 当前态；图谱=环节投影；"
+          "成员数已去重，命中率按去重前算）：")
     for tr in d.get("tracks") or []:
         sub = df[df["track"] == tr["name"]]
-        n_theme = len(tr.get("kg_themes") or [])
+        graph_sub = sub[sub["source_rule"] != "pool_theme"]
-        miss = [t for name, t in missing if name == tr["name"]]
+        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}: 成员 {sub['ts_code'].nunique()} 只"
+        line = (f"  {tr['name']}{tag}: 成员 {n_all} 只（图谱锚 {n_graph}）"
-                f" | 主题命中 {sub['theme'].nunique()}/{n_theme}")
+                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"]
-    print(f"  —— confirmed 合计（去重）: {conf['ts_code'].nunique()} 只")
+    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