对账工具：观察档下界修复+锚强度排序+锚质量审计
This commit is contained in:
zlt
parent
f7ea196545
commit
d076c96a02
|
|
@ -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):
|
||||
|
|
|
|||
Loading…
Reference in New Issue