274 lines
14 KiB
Python
274 lines
14 KiB
Python
"""G1 体检（只读探查，不写任何库）。设计依据：量化因子导出与合成设计.md §8-G1
|
||
+ 2026-07-26 交叉评审。
|
||
|
||
python run.py probe # 四节全跑
|
||
python run.py probe --section corr # 单跑：pools | price | upside | corr
|
||
|
||
四节：
|
||
[pools ] industry_pools 列结构 + 逐池清单 + 成员元素键采样
|
||
—— frontier_tracks.yml 的映射要对着真实主题名/字段起草，不猜 KG 命名
|
||
[price ] gp_day_data 按年 distinct 天数 —— 复核 §6.2 起始日疑点（5584 天 vs 理论 ~7000）
|
||
[upside] 覆盖池 upside 截面分布 + q 档位表 —— 定 §9-3 估值门槛分位 q
|
||
[corr ] 池内 (z_T, z_H, z_V) 相关矩阵 + 权重体检 —— 定 §9-8 权重结构（评审 §2）
|
||
|
||
注意：pools 节直读基座 industry_pools 表——这是一次性诊断的例外（§4 的运行时路径
|
||
只许走只读视图）；看清 schema 后，G2 把池清单固化成第五个插槽视图再供 tracks.py 用。
|
||
若连接账号只被授了视图权限，此节会报 permission denied——换基座 owner 账号跑一次即可。
|
||
任一节失败不影响其余节。
|
||
"""
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
import common
|
||
import db
|
||
import factors
|
||
|
||
|
||
def _sec(t):
|
||
print(f"\n{'=' * 64}\n[{t}]\n{'=' * 64}")
|
||
|
||
|
||
# ---------------------------------------------------------------- pools
|
||
def probe_pools():
|
||
_sec("pools · industry_pools 结构与主题清单")
|
||
cols = db.read_pg(
|
||
"SELECT column_name, data_type FROM information_schema.columns "
|
||
"WHERE table_name = 'industry_pools' "
|
||
" AND table_schema NOT IN ('pg_catalog','information_schema') "
|
||
"ORDER BY ordinal_position")
|
||
if cols.empty:
|
||
print("❌ 取不到 industry_pools 列信息（表不存在或无权限）")
|
||
return
|
||
print("列结构：")
|
||
for c, t in zip(cols["column_name"], cols["data_type"]):
|
||
print(f" {c}: {t}")
|
||
|
||
scalar = [c for c, t in zip(cols["column_name"], cols["data_type"])
|
||
if t not in ("jsonb", "json", "ARRAY")]
|
||
sel = ", ".join(f'"{c}"' for c in scalar) if scalar else "'(无标量列)' AS note"
|
||
df = db.read_pg(
|
||
f"SELECT {sel}, jsonb_array_length(COALESCE(members, '[]'::jsonb)) AS n_members "
|
||
f"FROM industry_pools ORDER BY n_members DESC")
|
||
print(f"\n逐池清单（共 {len(df)} 个池；标量列全给，赛道映射草案对着这个起）：")
|
||
with pd.option_context("display.max_rows", None, "display.max_columns", None,
|
||
"display.width", 220, "display.max_colwidth", 80):
|
||
print(df.to_string(index=False))
|
||
|
||
keys = db.read_pg(
|
||
"SELECT k AS member_key, count(*) AS n FROM industry_pools p "
|
||
"CROSS JOIN LATERAL jsonb_array_elements(COALESCE(p.members, '[]'::jsonb)) m "
|
||
"CROSS JOIN LATERAL jsonb_object_keys(m) k GROUP BY k ORDER BY n DESC")
|
||
print("\n成员元素键频次（看成员级带不带环节/层级/概念字段）：")
|
||
print(keys.to_string(index=False))
|
||
print(" 注（评审 §3）：实测成员级只有 ts_code/name/tier/confidence/supporting —— "
|
||
"**没有 segment 也没有 layer**，赛道门槛 C 的 kg_segments 映射在四视图上无源，"
|
||
"需第五/六个插槽视图或 claims 救急口径。")
|
||
|
||
smp = db.read_pg(
|
||
"SELECT m::text AS s FROM industry_pools p "
|
||
"CROSS JOIN LATERAL jsonb_array_elements(COALESCE(p.members, '[]'::jsonb)) m "
|
||
"LIMIT 3")
|
||
print("\n成员元素原样采样：")
|
||
for s in smp["s"]:
|
||
print(f" {s}")
|
||
|
||
|
||
# ---------------------------------------------------------------- price
|
||
def probe_price():
|
||
_sec("price · gp_day_data 按年 distinct 天数（§6.2 起始日复核）")
|
||
print("（全表聚合，视服务器规格可能要 1~3 分钟……）")
|
||
df = db.read_mysql("price",
|
||
"SELECT YEAR(`timestamp`) AS y, COUNT(DISTINCT `timestamp`) AS days, "
|
||
"COUNT(*) AS n_rows FROM gp_day_data GROUP BY YEAR(`timestamp`) ORDER BY y")
|
||
print(df.to_string(index=False))
|
||
full = df[df["days"] >= 230]
|
||
if full.empty:
|
||
print("\n⚠️ 没有任何年份 ≥230 天——行情比预想稀疏，§7 upside 回填深度需重议。")
|
||
else:
|
||
print(f"\n首个 ≥230 天（≈全年连续）的年份：{int(full['y'].iloc[0])} "
|
||
f"—— §7 upside 理论回填上限以此为准；之前年份视为零星记录。")
|
||
|
||
|
||
# ---------------------------------------------------------------- upside
|
||
def probe_upside():
|
||
_sec("upside · 覆盖池截面分布与 q 档位表（§9-3 / §9-4）")
|
||
d = db.read_mysql("price", "SELECT MAX(`timestamp`) AS d FROM gp_day_data").iloc[0, 0]
|
||
if d is None or pd.isna(d):
|
||
print("❌ gp_day_data 为空——先跑 views 看连通性。")
|
||
return
|
||
ds = pd.Timestamp(d).date().isoformat()
|
||
print(f"截面日 = 最新行情日 {ds}（as-of 口径与 build_upside 完全一致）")
|
||
df = factors.build_upside(ds, ds)
|
||
n_uni = len(common.load_universe())
|
||
if df is None or df.empty:
|
||
print("❌ 当日无可算 upside（consensus/行情缺）——先跑 views 看数据前沿。")
|
||
return
|
||
v = pd.to_numeric(df["factor_value"], errors="coerce").dropna()
|
||
print(f"覆盖池 {n_uni} 只：当日有 upside {len(v)} 只，"
|
||
f"无券商覆盖 {n_uni - len(v)} 只（按 §3.3 全被门槛②挡掉——§9-4 张力的量级）")
|
||
print("\n分位数：")
|
||
for q in (0.05, 0.10, 0.25, 0.50, 0.75, 0.90, 0.95):
|
||
print(f" P{int(q * 100):02d}: {float(v.quantile(q)):+.4f}")
|
||
print(f" mean : {v.mean():+.4f}（均值显著为正 = 券商乐观偏差的直读，§3.3）")
|
||
print(f" <0 占比: {(v < 0).mean():.1%}（θ_v=0 单靠绝对下限能挡掉的比例）")
|
||
print("\nq 档位表（θ_v = max(0, 池内 q 分位) → 门槛②通过只数）：")
|
||
for q in (0.00, 0.10, 0.20, 0.25, 0.30, 0.40, 0.50):
|
||
theta = max(0.0, float(v.quantile(q)))
|
||
print(f" q={q:.2f} → θ_v={theta:+.4f} → 通过 {int((v >= theta).sum())} 只")
|
||
print("\n注：本表基于覆盖池 U 全体；G2 赛道成员表落地后按 U∩C 重切一遍再定稿 q。")
|
||
|
||
|
||
# ---------------------------------------------------------------- corr
|
||
def _robust_z(v: pd.Series) -> pd.Series:
|
||
"""稳健标准化（中位数 / MAD），对齐平台 MathKernel.winsorize_mad 的处置口径。"""
|
||
med = v.median()
|
||
mad = (v - med).abs().median()
|
||
if not mad or np.isnan(mad) or mad <= 0:
|
||
return pd.Series(0.0, index=v.index)
|
||
return (v - med) / (1.4826 * mad)
|
||
|
||
|
||
def _latest_heat_day(on_or_before: str) -> str | None:
|
||
"""热度是 T+1 到达的，故 akg_score(T) 实际用 T−1 热度（设计 §3.4 末）。
|
||
这里如实复现该口径：取 <= 给定日的最新热度日。"""
|
||
df = db.read_mysql(
|
||
"heat", "SELECT MAX(trade_date) AS d FROM stock_fund_heat_scores "
|
||
"WHERE trade_date <= %s", (on_or_before,))
|
||
d = df.iloc[0, 0]
|
||
return None if d is None or pd.isna(d) else pd.Timestamp(d).date().isoformat()
|
||
|
||
|
||
def probe_corr():
|
||
"""池内三项相关矩阵 + 权重体检。
|
||
|
||
回答两个问题（评审 §2）：
|
||
① corr(z_H, z_V) 有多高？若 > 0.5，「三项加权」实际是两项，§9-8 要重开；
|
||
② 0.5/0.3/0.2 到底是「加权混合」还是「传导优先」？——直接在真实截面上数
|
||
top-K 里有几只是传导票。这比任何模拟都有说服力。
|
||
"""
|
||
_sec("corr · 池内 (z_T, z_H, z_V) 相关矩阵与权重体检（评审 §2）")
|
||
d = db.read_pg("SELECT MAX(scan_date) AS d FROM v_factor_transmission").iloc[0, 0]
|
||
if d is None or pd.isna(d):
|
||
print("❌ 无传导台账——先确认 transmission_scan 已跑。")
|
||
return
|
||
ds = pd.Timestamp(d).date().isoformat()
|
||
hd = _latest_heat_day(ds)
|
||
print(f"截面日 = 最新传导日 {ds}；热度取 {hd}（T+1 到达 → 实际用 T−1，设计 §3.4 末）")
|
||
if hd is None:
|
||
print("❌ 无可用热度日。")
|
||
return
|
||
|
||
up = factors.build_upside(ds, ds)
|
||
ht = factors.build_heat(hd, hd)
|
||
tr = factors.build_transmission(ds, ds)
|
||
if up is None or up.empty:
|
||
print("❌ 当日无 upside，无法构造候选池。")
|
||
return
|
||
|
||
def _s(df, name):
|
||
if df is None or df.empty:
|
||
return pd.Series(dtype=float, name=name)
|
||
x = df.copy()
|
||
x["k"] = x["stock_code"].map(common.to_prefix)
|
||
return (x.groupby("k")["factor_value"].max()
|
||
.astype(float).rename(name))
|
||
|
||
panel = pd.concat([_s(up, "upside"), _s(ht, "heat"), _s(tr, "transmission")],
|
||
axis=1)
|
||
n_uni = len(common.load_universe())
|
||
print(f"\n覆盖池 U = {n_uni} 只；当日面板 {len(panel)} 只"
|
||
f"（upside {panel['upside'].notna().sum()} / "
|
||
f"heat {panel['heat'].notna().sum()} / "
|
||
f"transmission {panel['transmission'].notna().sum()}）")
|
||
|
||
# ---- 候选池 P：S1 只有门槛② 可算（门槛① 赛道 C 待 G2 的成员表）----
|
||
P = panel[panel["upside"].notna() & (panel["upside"] >= 0.0)].copy()
|
||
print(f"候选池 P（仅门槛② upside>=0，赛道门槛 C 待 G2）= {len(P)} 只")
|
||
if len(P) < 10:
|
||
print("⚠️ 候选池过小，以下统计量意义有限。")
|
||
if P.empty:
|
||
return
|
||
|
||
# ---- 缺失处置（对齐设计 §3.4）----
|
||
P["transmission"] = P["transmission"].fillna(0.0) # 传导缺 = 0（取基准值）
|
||
P["heat"] = P["heat"].fillna(P["heat"].median()) # 热度缺 = 池内中位数
|
||
|
||
n_tr = int((P["transmission"] > 0).sum())
|
||
print(f"P 内有传导的股票 = {n_tr} 只（{n_tr / len(P):.1%}）"
|
||
f" ← 这个数才是真正决定榜首的量级（评审 §7）")
|
||
|
||
# ---- 三项标准化（传导不能用 MAD：池内多数为 0 → MAD=0 除零，见 §3.4）----
|
||
lt = np.log1p(P["transmission"])
|
||
sd = lt.std()
|
||
P["z_T"] = (lt - lt.mean()) / sd if sd and sd > 0 else 0.0
|
||
P["z_H"] = -_robust_z(P["heat"])
|
||
P["z_V"] = _robust_z(P["upside"])
|
||
|
||
print("\n① 相关矩阵（Spearman，池内）：")
|
||
# 整表 DataFrame.corr(spearman) 是 pandas 内置实现；单对 Series.corr(spearman)
|
||
# 却会 import scipy（容器未装，07-30 首跑即崩）——复用整表结果，零新依赖。
|
||
cm = P[["z_T", "z_H", "z_V"]].corr(method="spearman")
|
||
print(cm.round(3).to_string())
|
||
c_hv = float(cm.loc["z_H", "z_V"])
|
||
if abs(c_hv) > 0.5:
|
||
print(f"\n ⚠️ corr(z_H, z_V) = {c_hv:+.3f} —— 「还没热」与「便宜」高度共线，"
|
||
f"\n 「三项加权」实际是两项，§9-8 的权重讨论应重开"
|
||
f"（且门槛② 与 z_V 本就是同一变量进两次）。")
|
||
else:
|
||
print(f"\n corr(z_H, z_V) = {c_hv:+.3f} —— 共线性可接受，三项各自有独立信息。")
|
||
|
||
# ---- ② 权重体检：0.5/0.3/0.2 是混合还是词典序？----
|
||
print("\n② 权重体检：不同 w_T 下 top-K 里有几只是传导票")
|
||
print(f" （P 内共 {n_tr} 只传导票 / {len(P)} 只候选）")
|
||
if n_tr == 0:
|
||
print(" 当日无传导票，本节跳过。")
|
||
else:
|
||
hdr = " w_T " + "".join(f" top{k:<3}" for k in (10, 20, 30, 50))
|
||
print(hdr)
|
||
for wT in (0.50, 0.30, 0.20, 0.15, 0.10):
|
||
wrest = 1.0 - wT
|
||
s = wT * P["z_T"] + wrest * (0.6 * P["z_H"] + 0.4 * P["z_V"])
|
||
line = f" {wT:.2f} "
|
||
for k in (10, 20, 30, 50):
|
||
top = s.nlargest(min(k, len(P))).index
|
||
hit = int((P.loc[top, "transmission"] > 0).sum())
|
||
line += f" {hit:>2}/{min(k, n_tr):<3}"
|
||
print(line + (" ← 当前设计" if abs(wT - 0.50) < 1e-9 else ""))
|
||
print("\n 读法：若 w_T=0.50 那行的命中数≈min(K, 传导票数)，说明传导票"
|
||
"\n 几乎必然占满前列 —— 0.3/0.2 只在组内排序、跨组不起作用，"
|
||
"\n 即当前参数事实上是「传导优先」而非「加权混合」（评审 §2）。"
|
||
"\n 若确实如此，建议改结构（两段式）而不是改数值。")
|
||
|
||
# 两段式对照：primary=传导档位, tiebreak=0.6z_H+0.4z_V
|
||
tier = pd.Series(0.0, index=P.index)
|
||
hit_m = P["transmission"] > 0
|
||
if hit_m.any():
|
||
med = lt[hit_m].median()
|
||
tier[hit_m] = np.where(lt[hit_m] >= med, 2.0, 1.0)
|
||
s2 = tier * 10.0 + (0.6 * P["z_H"] + 0.4 * P["z_V"])
|
||
agree = len(set(s2.nlargest(min(20, len(P))).index)
|
||
& set((0.5 * P["z_T"] + 0.3 * P["z_H"] + 0.2 * P["z_V"])
|
||
.nlargest(min(20, len(P))).index))
|
||
print(f"\n 两段式(档位+组内)与当前公式的 top20 重合度：{agree}/20"
|
||
f" —— 越接近 20 越说明两者本就等价，改结构零代价。")
|
||
|
||
print("\n③ 门槛的量级（供 §9-3 定 q / 判断 C 的必要性）：")
|
||
print(f" U={n_uni} → 有 upside {int(panel['upside'].notna().sum())} "
|
||
f"→ 过门槛② {len(P)} → 其中有传导 {n_tr}")
|
||
print(" 若最后一个数长期是 0~3，组合层面就是 1~3 只票，且来自被 12×12 截断的"
|
||
"\n 候选列表——集中度需要在文档里明写（评审 §7）。")
|
||
|
||
|
||
SECTIONS = {"pools": probe_pools, "price": probe_price,
|
||
"upside": probe_upside, "corr": probe_corr}
|
||
|
||
|
||
def run(section="all"):
|
||
picked = list(SECTIONS) if section == "all" else [section]
|
||
print("G1 体检（只读）· 将跑节：" + ", ".join(picked))
|
||
for name in picked:
|
||
try:
|
||
SECTIONS[name]()
|
||
except Exception as e: # noqa: BLE001 —— 单节失败不拖累其余（体检尽量多产出）
|
||
print(f"\n❌ [{name}] 失败: {e!r}")
|