111 lines
4.5 KiB
Python
111 lines
4.5 KiB
Python
"""赛道映射（硬门槛 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
|