akg-factor-bridge/tracks.py

"""赛道映射（硬门槛 C 的实现载体，设计 §3.2）。

config/frontier_tracks.yml 是唯一事实源：每个赛道列出 kg_themes
（industry_pools 主题名）。本模块把它解析成 ts_code 级成员表，并落
data/track_members_<日期>.csv 版本化快照——可 git diff、可审计：
每只股票能追溯到因哪个赛道、哪个主题入选。

当前唯一的映射路径是主题名（kg_segments / kg_concepts 要等基座的环节
投影表建成，即三批-4），source_rule 统一记 'pool_theme'。
"""
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 resolve_members(only_confirmed: bool = True, path: str | None = None,
                    dedup: bool = True):
    """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():
        ms = r["members"]
        if isinstance(ms, str):
            ms = json.loads(ms)
        by_theme[str(r["theme"]).strip()] = ms or []
    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 []:
            t = str(theme).strip()
            if t in excl:
                continue
            if t not in by_theme:
                missing.append((tr["name"], t))
                continue
            for m in by_theme[t]:
                ts = (m or {}).get("ts_code")
                if ts:
                    rows.append((ts, m.get("name"), tr["name"],
                                 tr.get("status"), t))
    df = pd.DataFrame(rows, columns=["ts_code", "name", "track",
                                     "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，审计列）。"""
    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",
               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。"""
    # 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"]]
        n_theme = len(tr.get("kg_themes") or [])
        miss = [t for name, t in missing if name == tr["name"]]
        tag = "" if tr.get("status") == "confirmed" else "（candidate）"
        line = (f"  {tr['name']}{tag}: 成员 {sub['ts_code'].nunique()} 只"
                f" | 主题命中 {sub['theme'].nunique()}/{n_theme}")
        if miss:
            line += f" | 未命中: {'、'.join(miss)}"
        print(line)
    conf = df[df["status"] == "confirmed"]
    print(f"  —— confirmed 合计（去重）: {conf['ts_code'].nunique()} 只")
    return df