diff --git a/research/cause_effect_panic_alpha.py b/research/cause_effect_panic_alpha.py new file mode 100644 index 00000000..2b462a97 --- /dev/null +++ b/research/cause_effect_panic_alpha.py @@ -0,0 +1,76 @@ +# -*- 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() diff --git a/research/s2_panic_v2_gen.py b/research/s2_panic_v2_gen.py new file mode 100644 index 00000000..0bd14709 --- /dev/null +++ b/research/s2_panic_v2_gen.py @@ -0,0 +1,77 @@ +# -*- coding: utf-8 -*- +"""s2_panic_v2 修正：每日 top-N 截断（每恐慌日只选 score 最高的 N 个） +这是关键：恐慌日信号 1200/天 是结构问题，必须每日截断让信号/成交比可控 +""" +import sys, json +sys.path.insert(0, "/home/hmo/MoFin") +import pandas as pd +import numpy as np + +def alpha_score(mcap_q, rsi, sec_ret20, news3): + sc = 0 + if mcap_q is not None and not np.isnan(mcap_q): + sc += 40 if mcap_q < 0.2 else 32 if mcap_q < 0.4 else 24 if mcap_q < 0.6 else 16 if mcap_q < 0.8 else 8 + if rsi is not None and not np.isnan(rsi): + sc += 30 if rsi >= 45 else 22 if rsi >= 35 else 12 if rsi >= 25 else 6 + if sec_ret20 is not None and not np.isnan(sec_ret20): + sc += 20 if sec_ret20 >= 0 else 16 if sec_ret20 >= -10 else 8 if sec_ret20 >= -20 else 3 + if news3 is not None and not np.isnan(news3): + sc += 10 if news3 >= 2 else 7 if news3 >= 1 else 2 + return sc + +def gen_trades(start="2016-01-01", end="2026-07-01", top_n=8): + panel = pd.read_pickle("/tmp/panel_12d.pkl") + panel = panel.sort_values(["code", "date"]).reset_index(drop=True) + g = panel.groupby("code", group_keys=False) + def fwd_max(s, w): return s[::-1].rolling(w, min_periods=1).max()[::-1] + def fwd_min(s, w): return s[::-1].rolling(w, min_periods=1).min()[::-1] + panel["fwd_max60"] = g["close"].transform(lambda x: fwd_max(x, 60)) + panel["fwd_min60"] = g["close"].transform(lambda x: fwd_min(x, 60)) + + panic = panel[(panel["mkt_rsi"] < 25) & (panel["date"] >= start) & (panel["date"] <= end)].copy() + sig = panic[(panic["mcap_q"] < 0.4) & (panic["rsi"] >= 35) & (panic["sec_ret20"] >= -10)].copy() + sig["score"] = sig.apply(lambda r: alpha_score(r["mcap_q"], r["rsi"], r["sec_ret20"], r["news3"]), axis=1) + + # 每日 top-N 截断（score 降序） + sig = sig.sort_values(["date", "score"], ascending=[True, False]).groupby("date").head(top_n) + print(f"每日top{top_n}截断后: {len(sig)} 信号 (原6932)") + + trades = [] + for _, s in sig.iterrows(): + ep = s["close"] + if ep <= 0: + continue + fmax = s["fwd_max60"]; fmin = s["fwd_min60"] + hit_tp = fmax >= ep * 1.30 + hit_sl = fmin <= ep * 0.88 + if hit_tp: pnl, reason = 30.0, "target" + elif hit_sl: pnl, reason = -12.0, "stop" + else: pnl, reason = (fmax / ep - 1) * 100 if not pd.isna(fmax) else 0, "time" + trades.append({"code": s["code"], "entry_date": s["date"], "entry_price": round(ep, 2), + "profit_pct": round(pnl, 2), "exit_reason": reason, + "hold_days": 60, "score": int(s["score"]), "name": str(s["code"]), "boost": 1.0}) + return trades + +if __name__ == "__main__": + from strategy_lab import portfolio_sim + import copy, random + for top_n in [5, 8, 10]: + trades = gen_trades(top_n=top_n) + print(f"\n=== top{top_n} ===") + sim0 = portfolio_sim(trades, 1000000, max_positions=10) + n_sig = len(trades) + n_pos = sim0.get("positions_taken") + print(f"信号{n_sig} 成交{n_pos} 比{n_sig/max(n_pos,1):.1f} 总收益{sim0.get('total_return_pct')}% 年化{sim0.get('cagr_pct')}% 回撤{sim0.get('portfolio_max_dd_pct')}%") + # 稳健性 + rets = [] + for seed in range(5): + t2 = copy.deepcopy(trades); rng = random.Random(seed); rng.shuffle(t2) + rets.append(portfolio_sim(t2, 1000000, max_positions=10).get("total_return_pct")) + print(f" 洗牌5次: 收益{rets} 差{max(rets)-min(rets):.1f}pp") + # 信号/成交分布 + from collections import Counter + dc = Counter(t["entry_date"] for t in trades) + print(f" 每日信号分布: {dict(sorted(dc.items()))}") + if top_n == 8: + with open("/home/hmo/MoFin/data/s2_panic_v2_trades.json", "w") as f: + json.dump(trades, f, ensure_ascii=False) diff --git a/research/s2_panic_v2_score_verify.py b/research/s2_panic_v2_score_verify.py new file mode 100644 index 00000000..e87cd09d --- /dev/null +++ b/research/s2_panic_v2_score_verify.py @@ -0,0 +1,75 @@ +# -*- coding: utf-8 -*- +"""验证：top-N(score择优) vs 随机抽样 的收益差异""" +import sys, random +sys.path.insert(0, "/home/hmo/MoFin") +import pandas as pd +import numpy as np + +def alpha_score(mcap_q, rsi, sec_ret20, news3): + sc = 0 + if mcap_q is not None and not np.isnan(mcap_q): + sc += 40 if mcap_q < 0.2 else 32 if mcap_q < 0.4 else 24 if mcap_q < 0.6 else 16 if mcap_q < 0.8 else 8 + if rsi is not None and not np.isnan(rsi): + sc += 30 if rsi >= 45 else 22 if rsi >= 35 else 12 if rsi >= 25 else 6 + if sec_ret20 is not None and not np.isnan(sec_ret20): + sc += 20 if sec_ret20 >= 0 else 16 if sec_ret20 >= -10 else 8 if sec_ret20 >= -20 else 3 + if news3 is not None and not np.isnan(news3): + sc += 10 if news3 >= 2 else 7 if news3 >= 1 else 2 + return sc + +panel = pd.read_pickle("/tmp/panel_12d.pkl") +panel = panel.sort_values(["code", "date"]).reset_index(drop=True) +g = panel.groupby("code", group_keys=False) +def fwd_max(s, w): return s[::-1].rolling(w, min_periods=1).max()[::-1] +def fwd_min(s, w): return s[::-1].rolling(w, min_periods=1).min()[::-1] +panel["fwd_max60"] = g["close"].transform(lambda x: fwd_max(x, 60)) +panel["fwd_min60"] = g["close"].transform(lambda x: fwd_min(x, 60)) + +panic = panel[panel["mkt_rsi"] < 25].copy() +sig = panic[(panic["mcap_q"] < 0.4) & (panic["rsi"] >= 35) & (panic["sec_ret20"] >= -10)].copy() +sig["score"] = sig.apply(lambda r: alpha_score(r["mcap_q"], r["rsi"], r["sec_ret20"], r["news3"]), axis=1) + +def pnl_of(s): + ep = s["close"] + fmax, fmin = s["fwd_max60"], s["fwd_min60"] + if fmax >= ep * 1.30: return 30.0 + if fmin <= ep * 0.88: return -12.0 + return (fmax / ep - 1) * 100 if not pd.isna(fmax) else 0 + +days = {dt: grp for dt, grp in sig.groupby("date")} + +# top5 择优 +top5 = sig.sort_values(["date", "score"], ascending=[True, False]).groupby("date").head(5) +tp = top5.apply(pnl_of, axis=1) +print(f"top5(score择优): n={len(top5)} 平均pnl={tp.mean():.2f}% 胜率={(tp>0).mean()*100:.1f}%") + +# 随机抽5 +rng = random.Random(42) +allr = [] +for trial in range(20): + chosen = [] + for dt, grp in days.items(): + n = min(5, len(grp)) + chosen.extend(grp.sample(n=n, random_state=1000*trial+7).index.tolist()) + rp = sig.loc[chosen].apply(pnl_of, axis=1) + allr.append(rp.mean()) +print(f"随机抽5×20次: 平均={np.mean(allr):.2f}% 范围{min(allr):.2f}~{max(allr):.2f}%") + +# top10 +top10 = sig.sort_values(["date", "score"], ascending=[True, False]).groupby("date").head(10) +tp10 = top10.apply(pnl_of, axis=1) +print(f"top10(score择优): n={len(top10)} 平均pnl={tp10.mean():.2f}% 胜率={(tp10>0).mean()*100:.1f}%") +allr10 = [] +for trial in range(20): + chosen = [] + for dt, grp in days.items(): + n = min(10, len(grp)) + chosen.extend(grp.sample(n=n, random_state=2000*trial+13).index.tolist()) + rp = sig.loc[chosen].apply(pnl_of, axis=1) + allr10.append(rp.mean()) +print(f"随机抽10×20次: 平均={np.mean(allr10):.2f}% 范围{min(allr10):.2f}~{max(allr10):.2f}%") + +# bottom5（最低分对照组） +bot5 = sig.sort_values(["date", "score"], ascending=[True, True]).groupby("date").head(5) +bp = bot5.apply(pnl_of, axis=1) +print(f"bottom5(最低分): n={len(bot5)} 平均pnl={bp.mean():.2f}% 胜率={(bp>0).mean()*100:.1f}%")