diff --git a/plan_reconcile.py b/plan_reconcile.py index 0fc735a..633a118 100644 --- a/plan_reconcile.py +++ b/plan_reconcile.py @@ -60,6 +60,13 @@ def _load(d: str): for r in tdf.itertuples(): hits.setdefault(common.to_prefix(str(r.ts_code)), []).append( (r.track, r.theme, r.source_rule)) + # 按锚强度排序(环节锚 > 链锚 > 主题弱锚)。resolve_members 是按 yml 里 + # 赛道的先后顺序产行的,不排序的话 hits[0] 取到的是"yml 中最靠前的赛道", + # 而不是"最强的锚"——08-05 实测宁德时代显示成算力、国轩高科显示成通信, + # 就是这么来的:真凭据在后面,前面那条是宽主题捞进来的。勿回退。 + _RULE_RANK = {"graph_segment": 0, "graph_chain": 1, "pool_theme": 2} + for k in hits: + hits[k].sort(key=lambda h: _RULE_RANK.get(h[2], 9)) # 传导指向:scan_date=d 的候选里谁的 quiet 含这只票 try: @@ -107,10 +114,14 @@ def _load(d: str): except Exception as e: # noqa: BLE001 print(f"⚠️ 简称加载失败: {e!r}") - # 榜位名次:主榜=score>=150 降序;观察档=100<=score<150 降序 + # 榜位名次:150 是主榜与观察档的分界,**下方不设界**—— + # 观察档分 = 100 + clip(组内分, ±9.9),区间是 [90.1, 109.9], + # 首版写 sc>=100 把 [90.1,100) 那半截整段漏掉(08-05 实测:md 说 675、 + # 工具只数出 232,差 443 只)。score 表本就只有 gate>0 的票出行, + # 低于 150 的一律是观察档,不需要下界。勿回退。 sc = pd.Series(score) main = sc[sc >= 150].sort_values(ascending=False) - obs = sc[(sc >= 100) & (sc < 150)].sort_values(ascending=False) + obs = sc[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, @@ -178,8 +189,57 @@ def explain(k: str, L: dict) -> None: + ("(可能不在覆盖池 universe,或当日未出行)" if not hits else "——在赛道却无档位行,把本行发我核")) +def audit(L: dict) -> None: + """锚质量审计(--audit):主榜成员里有多少是靠宽主题弱锚进来的。 + + 主题弱锚(pool_theme)是 industry_pools 的概念池成员,池宽则噪声大—— + 08-05 实测:宁德时代记成算力、国轩高科记成通信、东方日升记成商业航空航天。 + 档位判定本身不看是哪路锚(在赛道就算数),所以噪声不改变对错、只改变 + "凭什么";但宽主题同时也在**放人进主榜**,这才是要拍板的地方。 + 本审计回答两问:逐赛道的主榜成员有多少带可审计的图谱锚; + 哪些主题键在独家放人(该股在所有赛道都没有图谱锚)——它们就是下一批 + "降级或剔除"的候选。链锚扩充落地后重跑本审计,看图谱锚占比抬升多少。""" + main_set = {k for k, g in L["gate"].items() if g == 2.0} + by_track: dict[str, dict[str, set]] = {} + theme_only_admits: dict[str, set] = {} + n_graph = 0 + for k in main_set: + hs = L["hits"].get(k, []) + if not hs: + continue + has_graph_any = any(r != "pool_theme" for _, _, r in hs) + if has_graph_any: + n_graph += 1 + for t, key, rule in hs: + d = by_track.setdefault(t, {"all": set(), "graph": set()}) + d["all"].add(k) + if rule != "pool_theme": + d["graph"].add(k) + if rule == "pool_theme" and not has_graph_any: + theme_only_admits.setdefault(key, set()).add(k) + + n_in_track = sum(1 for k in main_set if L["hits"].get(k)) + print(f"\n—— 锚质量审计:主榜 {len(main_set)} 只,其中 {n_in_track} 只在赛道内 ——") + print(f" 带图谱锚(环节/链,可审计到图谱边): {n_graph} 只" + f"({n_graph / max(1, n_in_track):.0%});" + f"仅靠主题弱锚: {n_in_track - n_graph} 只") + print("\n 逐赛道(成员数 / 其中带图谱锚):") + for t, d in sorted(by_track.items(), key=lambda kv: -len(kv[1]["all"])): + print(f" {t:<8} {len(d['all']):>4} / {len(d['graph']):>4}" + f"({len(d['graph']) / max(1, len(d['all'])):.0%})") + print("\n 独家放人的主题键前 20(这些股在所有赛道都没有图谱锚," + "只靠该主题进的主榜——降级候选):") + for key, ks in sorted(theme_only_admits.items(), + key=lambda kv: -len(kv[1]))[:20]: + sample = "、".join(L["names"].get(x, x) for x in sorted(ks)[:3]) + print(f" {key:<16} {len(ks):>4} 只 例:{sample}") + + def main() -> int: args = [a for a in sys.argv[1:]] + do_audit = "--audit" in args + if do_audit: + args.remove("--audit") d = None if "--date" in args: i = args.index("--date") @@ -196,17 +256,24 @@ def main() -> int: print(f"选股计划对账 @ 档位日 {d}{L['stale']}") print(f"主榜 {len(L['main'])} 只 / 观察档 {len(L['obs'])} 只(150 分界;主榜默认展示前 20)") + if do_audit: + audit(L) if args: for a in args: explain(_norm_code(a), L) - else: - print("\n—— 主榜前 20(传导档优先,与计划 API 同序)——") + elif not do_audit: + print("\n—— 主榜前 20(传导档优先,与计划 API 同序;本清单不设每主题限额," + "计划 API 默认 theme_cap=5 故名单会略有出入)——") for i, (k, s) in enumerate(L["main"].head(20).items(), 1): hits = L["hits"].get(k, []) + anchor = "?" + if hits: + t, key, rule = hits[0] + anchor = f"{t}←{key}" + ("(弱)" if rule == "pool_theme" else "(图谱)") 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"赛道={anchor}" + (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):