diff --git a/plan_reconcile.py b/plan_reconcile.py new file mode 100644 index 0000000..0fc735a --- /dev/null +++ b/plan_reconcile.py @@ -0,0 +1,222 @@ +"""选股计划逐票对账（只读）——"这只票为什么在/不在计划里"。 + +背景：机会线索页（基座前端，环节级雷达，跑全部主题池）与每日选股计划 +（个股级，券商覆盖 ∩ upside≥0 ∩ 十条赛道 ∩ 非风险，传导档优先排序） +答的不是同一个问题，读数经常"看起来大相径庭"。本工具把任意一只票的 +全部判定摊开：赛道命中（哪路）、覆盖与 upside、热度、传导（谁指向它）、 +gate、score、榜位名次，落榜给出确切原因——审计承诺（下游对接文档 +"每只候选可回溯"）的落地件。 + +跑法【桥机 factorevaluation-UTC · ~/akg-factor-bridge】： + docker compose exec -T akg-factor-bridge python plan_reconcile.py 300750 SH600438 002074.SZ + docker compose exec -T akg-factor-bridge python plan_reconcile.py # 不带票=打印当日主榜前20/观察档前10 + docker compose exec -T akg-factor-bridge python plan_reconcile.py --date 2026-08-04 300750 +代码三种形态都收（600000.SH / SH600000 / 600000，600/688 归 SH，其余归 SZ）。 +纯只读：因子表 / 基座视图 / yml 全部 SELECT 与本地解析，不写任何东西。 +""" +from __future__ import annotations + +import re +import sys + +import pandas as pd + +import common +import db +import tracks + +_RISK_RE = re.compile(r"^(\*?S?ST|退市)") + + +def _norm_code(s: str) -> str: + """任意形态 -> 前缀式（SH600000）。裸 6 位码按交易所惯例补前缀。""" + s = s.strip().upper() + if re.fullmatch(r"\d{6}\.(SH|SZ|BJ)", s): + return common.to_prefix(s) + if re.fullmatch(r"(SH|SZ|BJ)\d{6}", s): + return s + if re.fullmatch(r"\d{6}", s): + exch = "SH" if s[0] == "6" else ("BJ" if s[0] in "48" else "SZ") + return f"{exch}{s}" + raise SystemExit(f"认不出的代码形态: {s!r}（收 600000.SH / SH600000 / 600000）") + + +def _factor(table: str, d: str) -> dict[str, float]: + df = db.read_mysql("factor", f"SELECT stock_code, factor_value FROM {table} " + f"WHERE trade_date=%s", (d,)) + return {str(r.stock_code).strip(): float(r.factor_value) for r in df.itertuples()} + + +def _load(d: str): + gate = _factor("t_factor_akg_gate", d) + score = _factor("t_factor_akg_score", d) + upside = _factor("t_factor_akg_upside", d) + heat = _factor("t_factor_akg_heat", d) + tr = _factor("t_factor_akg_transmission", d) + + # 赛道命中：前缀码 -> [(赛道, 命中键, 路)]，图谱路在前（resolve 已排好） + tdf, _missing = tracks.resolve_members(only_confirmed=True, dedup=False) + hits: dict[str, list[tuple[str, str, str]]] = {} + for r in tdf.itertuples(): + hits.setdefault(common.to_prefix(str(r.ts_code)), []).append( + (r.track, r.theme, r.source_rule)) + + # 传导指向：scan_date=d 的候选里谁的 quiet 含这只票 + try: + tv = db.read_pg( + "SELECT ts_code, target, n_sources, moved_ratio, mkt_trade_date " + "FROM v_factor_transmission WHERE scan_date=%s", (d,)) + except Exception as e: # noqa: BLE001 —— 视图不可用只影响传导明细 + print(f"⚠️ v_factor_transmission 读取失败（传导明细缺席）: {e!r}") + tv = pd.DataFrame(columns=["ts_code", "target", "n_sources", + "moved_ratio", "mkt_trade_date"]) + tmap: dict[str, list[str]] = {} + for r in tv.itertuples(): + tmap.setdefault(common.to_prefix(str(r.ts_code)), []).append( + f"{r.target}(源{int(r.n_sources)}×空间{1 - float(r.moved_ratio):.0%})") + stale = "" + if len(tv) and tv["mkt_trade_date"].notna().any(): + md = str(tv["mkt_trade_date"].dropna().iloc[0])[:10] + if md != d: + stale = f"（⚠️ 传导快照日 {md} ≠ 档位日 {d}，该日传导降级采信）" + + # 图谱证据 + 简称 + ev: dict[str, dict] = {} + try: + mem = db.read_pg("SELECT segment_name, ts_code, chain " + "FROM v_factor_segment_members WHERE ts_code IS NOT NULL") + for r in mem.itertuples(): + k = common.to_prefix(str(r.ts_code).strip()) + e = ev.setdefault(k, {"segs": set(), "chains": set()}) + e["segs"].add(str(r.segment_name)) + if r.chain and str(r.chain).strip(): + e["chains"].add(str(r.chain).strip()) + except Exception as e: # noqa: BLE001 + print(f"⚠️ 环节投影读取失败（图谱证据缺席）: {e!r}") + names: dict[str, str] = {} + try: + import json as _json + pools = db.read_pg("SELECT members FROM industry_pools") + for _, row in pools.iterrows(): + ms = row["members"] + if isinstance(ms, str): + ms = _json.loads(ms) + for m in ms or []: + if (m or {}).get("ts_code"): + names.setdefault(common.to_prefix(m["ts_code"]), m.get("name") or "") + except Exception as e: # noqa: BLE001 + print(f"⚠️ 简称加载失败: {e!r}") + + # 榜位名次：主榜=score>=150 降序；观察档=100<=score<150 降序 + sc = pd.Series(score) + main = sc[sc >= 150].sort_values(ascending=False) + obs = sc[(sc >= 100) & (sc < 150)].sort_values(ascending=False) + rank_main = {k: i + 1 for i, k in enumerate(main.index)} + rank_obs = {k: i + 1 for i, k in enumerate(obs.index)} + return dict(gate=gate, score=score, upside=upside, heat=heat, tr=tr, + hits=hits, tmap=tmap, stale=stale, ev=ev, names=names, + main=main, obs=obs, rank_main=rank_main, rank_obs=rank_obs) + + +def _fmt(v, pat="{:.2f}"): + return "—" if v is None else pat.format(v) + + +def explain(k: str, L: dict) -> None: + nm = L["names"].get(k, "") + g = L["gate"].get(k) + up = L["upside"].get(k) + ht = L["heat"].get(k) + trv = L["tr"].get(k) + sc = L["score"].get(k) + hits = L["hits"].get(k, []) + paths = L["tmap"].get(k, []) + e = L["ev"].get(k, {"segs": set(), "chains": set()}) + risk = bool(_RISK_RE.match(nm.replace(" ", ""))) if nm else False + + print(f"\n◆ {k} {nm}") + if hits: + shown = "；".join(f"{t}←{key}({rule})" for t, key, rule in hits[:4]) + print(f" 赛道命中 {len(hits)} 路：{shown}" + ("…" if len(hits) > 4 else "")) + else: + print(" 赛道命中：无（十条赛道三路映射都够不着）") + print(f" 覆盖/upside：{'有覆盖，upside=' + _fmt(up, '{:+.1%}') if up is not None else '无券商覆盖'}" + f" 热度：{_fmt(ht)} 传导分：{_fmt(trv)}") + if paths: + print(f" 传导指向：{'；'.join(paths[:4])}" + ("…" if len(paths) > 4 else "")) + if e["segs"]: + print(f" 图谱证据：环节 {len(e['segs'])} 个（{'、'.join(sorted(e['segs'])[:4])}" + + ("…" if len(e["segs"]) > 4 else "") + "）" + + (f"，链名 {'、'.join(sorted(e['chains'])[:4])}" if e["chains"] else "，边上无链名")) + + if g == 2.0: + n = L["rank_main"].get(k) + # 档界：score=200+档×20+组内分(±9.9) → 强≥230.1、弱≥210.1、无≤209.9 + tier = "强传导" if sc and sc >= 230 else ("弱传导" if sc and sc >= 210 else "无传导") + print(f" ▶ 判决：主榜第 {n}/{len(L['main'])} 名（score={sc:.1f}，{tier}档）" + + ("——计划默认只显示前 20，名次靠后不等于不在计划里" if n and n > 20 else "")) + elif g == 1.0: + print(f" ▶ 判决：观察档第 {L['rank_obs'].get(k)}/{len(L['obs'])} 名（score={sc:.1f}，" + f"无券商覆盖、低置信）") + elif g == 0.0: + if risk: + reason = "重大风险闸（ST/退市族），07-31 拍板默认杜绝" + elif up is not None and up < 0: + reason = f"有覆盖但 upside={up:+.1%} < 0——贵了不买是绝对下限" + elif up is not None and hits: + reason = "覆盖、upside、赛道三者都过——按机制不该是 0，把本行发我核（疑似口径错位）" + elif up is not None: + reason = ("有覆盖、upside≥0，但不在十条赛道（赛道闸拦下）" + + ("——图上有环节证据，属映射够不着的错杀池，看首批锚能否接住" if e["segs"] else "")) + elif not hits and (trv or 0) <= 0: + reason = "无覆盖、无传导、不在赛道——两锚皆无" + else: + reason = "无覆盖且当日不在传导链上（观察档条件差最后一步）" + print(f" ▶ 判决：不采纳（gate=0）。原因：{reason}") + else: + print(" ▶ 判决：当日档位表无此票" + + ("（可能不在覆盖池 universe，或当日未出行）" if not hits else "——在赛道却无档位行，把本行发我核")) + + +def main() -> int: + args = [a for a in sys.argv[1:]] + d = None + if "--date" in args: + i = args.index("--date") + d = args[i + 1] + del args[i:i + 2] + if d is None: + row = db.read_mysql("factor", "SELECT MAX(trade_date) d FROM t_factor_akg_gate") + v = None if row.empty else row.iloc[0, 0] + if v is None or pd.isna(v): + print("档位表为空——先跑当日构建。") + return 2 + d = pd.Timestamp(v).date().isoformat() + L = _load(d) + print(f"选股计划对账 @ 档位日 {d}{L['stale']}") + print(f"主榜 {len(L['main'])} 只 / 观察档 {len(L['obs'])} 只（150 分界；主榜默认展示前 20）") + + if args: + for a in args: + explain(_norm_code(a), L) + else: + print("\n—— 主榜前 20（传导档优先，与计划 API 同序）——") + for i, (k, s) in enumerate(L["main"].head(20).items(), 1): + hits = L["hits"].get(k, []) + print(f" {i:>2}. {k} {L['names'].get(k, ''):　<6} score={s:.1f} " + f"upside={_fmt(L['upside'].get(k), '{:+.1%}')} " + f"热度={_fmt(L['heat'].get(k))} " + f"赛道={hits[0][0] if hits else '?'}" + + (f" 传导指向={L['tmap'][k][0]}" if L["tmap"].get(k) else "")) + print("\n—— 观察档前 10 ——") + for i, (k, s) in enumerate(L["obs"].head(10).items(), 1): + print(f" {i:>2}. {k} {L['names'].get(k, ''):　<6} score={s:.1f} " + f"热度={_fmt(L['heat'].get(k))}" + + (f" 传导指向={L['tmap'][k][0]}" if L["tmap"].get(k) else "")) + print("\n用法：把机会线索卡片上的代表公司码传进来逐票对账，" + "例如 python plan_reconcile.py 300750 002074 688041") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())