77 lines
3.4 KiB
Python
77 lines
3.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""由果及因（四）：全部恐慌日的 alpha 特征挖掘
|
||
方法：合并所有 mkt_rsi<25 的恐慌日信号（全市场），按特征分桶看 fwd_ret20/60 差异
|
||
目标：找到能区分"反弹强弱"的特征 → 作为信号 score 依据
|
||
关键：所有恐慌日合并（不只2025-04-09），保证样本量和统计意义
|
||
"""
|
||
import sys
|
||
sys.path.insert(0, "/home/hmo/MoFin")
|
||
import pandas as pd
|
||
import numpy as np
|
||
|
||
panel = pd.read_pickle("/tmp/panel_12d.pkl")
|
||
panel = panel.sort_values(["code", "date"]).reset_index(drop=True)
|
||
for w in [10, 20, 60]:
|
||
panel[f"fwd_ret{w}"] = panel.groupby("code")["close"].transform(lambda x, ww=w: x.shift(-ww) / x - 1) * 100
|
||
|
||
# 全部恐慌日（mkt_rsi<25）
|
||
panic = panel[panel["mkt_rsi"] < 25].dropna(subset=["fwd_ret20"]).copy()
|
||
print(f"恐慌日总样本: {len(panic)} (涉及 {panic['date'].nunique()} 天, {panic['code'].nunique()} 只)")
|
||
print(f"恐慌日日期: {sorted(panic['date'].unique())}")
|
||
base = panic["fwd_ret20"].mean()
|
||
base_wr = (panic["fwd_ret20"] > 0).mean() * 100
|
||
print(f"恐慌日基线: 20d均值 {base:.2f}% 胜率 {base_wr:.1f}%")
|
||
print()
|
||
|
||
# 特征分桶（重点看哪些特征桶间差异大）
|
||
features = {
|
||
"ret5(5日跌幅)": ("ret5", [-40, -25, -18, -12, -8, -4, 0]),
|
||
"bias60(60日偏离)": ("bias60", [-45, -30, -20, -10, 0, 10, 30]),
|
||
"bias20(20日偏离)": ("bias20", [-30, -20, -10, 0, 10, 25]),
|
||
"dist_lo20(离低点)": ("dist_lo20", [0, 5, 10, 15, 25, 50]),
|
||
"rsi(个股)": ("rsi", [0, 15, 25, 35, 45, 60]),
|
||
"ret20(20日跌)": ("ret20", [-40, -25, -15, -5, 0, 10]),
|
||
"mcap_q(市值分位)": ("mcap_q", [0, 0.2, 0.4, 0.6, 0.8, 1]),
|
||
"pe_q(估值)": ("pe_q", [0, 0.2, 0.4, 0.6, 0.8, 1]),
|
||
"pb_q(市净率)": ("pb_q", [0, 0.2, 0.4, 0.6, 0.8, 1]),
|
||
"news3(新闻)": ("news3", [0, 1, 2, 4, 8]),
|
||
"vol_ratio(量比)": ("vol_ratio", [0, 0.8, 1.2, 1.8, 3, 6]),
|
||
"flow5(5日资金)": ("flow5", [-50, -20, 0, 20, 50]),
|
||
"mkt_adx(大盘ADX)": ("mkt_adx", [0, 20, 30, 40, 60]),
|
||
"mkt_ret20(大盘20d)": ("mkt_ret20", [-30, -20, -10, 0, 10]),
|
||
"sec_ret20(行业20d)": ("sec_ret20", [-30, -20, -10, 0, 10]),
|
||
"hi20_new(20日新高)": ("hi20_new", [0, 1]),
|
||
}
|
||
|
||
print("=" * 78)
|
||
print(f"{'特征':<16} {'桶':<14} {'n':>6} {'20d均值':>8} {'胜率':>7} {'超额':>7}")
|
||
print("=" * 78)
|
||
for label, (feat, edges) in features.items():
|
||
if feat not in panic.columns:
|
||
continue
|
||
s = panic.dropna(subset=[feat])
|
||
if len(s) < 200:
|
||
continue
|
||
results = []
|
||
for i in range(len(edges) - 1):
|
||
lo, hi = edges[i], edges[i + 1]
|
||
m = s[(s[feat] >= lo) & (s[feat] < hi)]
|
||
if len(m) < 50:
|
||
continue
|
||
wr = (m["fwd_ret20"] > 0).mean() * 100
|
||
results.append((f"{lo}~{hi}", len(m), m["fwd_ret20"].mean(), wr))
|
||
m = s[s[feat] >= edges[-1]]
|
||
if len(m) >= 50:
|
||
wr = (m["fwd_ret20"] > 0).mean() * 100
|
||
results.append((f">={edges[-1]}", len(m), m["fwd_ret20"].mean(), wr))
|
||
# 打印桶间差异（max-min 均值差）
|
||
if len(results) >= 2:
|
||
means = [r[2] for r in results]
|
||
spread = max(means) - min(means)
|
||
mark = " ★" if spread > 15 else (" ◈" if spread > 8 else "")
|
||
print(f"{label:<16} (桶间差 {spread:.1f}pp){mark}")
|
||
for name, n, mean, wr in results:
|
||
bar = "█" * int(abs(mean) / 2)
|
||
print(f"{'':<16} {name:<14} {n:>6} {mean:>7.2f}% {wr:>6.1f}% {mean-base:>+6.2f} {bar}")
|
||
print()
|