78 lines
3.8 KiB
Python
78 lines
3.8 KiB
Python
# -*- 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)
|