76 lines
3.2 KiB
Python
76 lines
3.2 KiB
Python
# -*- 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}%")
|