hmo
b9c68a83a7
- 新增 strategy_research_methodology.md（由果及因/12维/铁律/支撑压力规范） - 新增 predictive_oversold_strategy.md（v5定稿，年化18.57%） - 新增 deployment-plan-predictive-oversold.md（整合部署计划） - 归档 docs/research/（63份研究过程文档）+ scripts/research/（19个研究脚本） - 更新 docs/README.md 文档中心（策略研究章节）
64 lines
3.0 KiB
Python
64 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
||
"""step28_predictive_signal.py — 组合预测因子为信号（数据定阈值，全量验证）
|
||
step27 扫描出的预测因子（事前可计算）：
|
||
- 大盘RSI<41（弱市大涨率4.97% vs 基线3.52%）
|
||
- 小市值<0.2分位（4.60%）
|
||
- 低PE<0.2分位（4.66%）
|
||
- 新闻≥1条（4.23-4.52%）
|
||
组合这些因子，验证叠加后的预测力（全量不抽样）
|
||
"""
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
print("=== 加载 ===", flush=True)
|
||
panel = pd.read_pickle("/tmp/panel_12d.pkl")
|
||
panel = panel.sort_values(["code", "date"]).reset_index(drop=True)
|
||
panel["fwd_ret60"] = panel.groupby("code")["close"].transform(lambda x: x.shift(-60)/x - 1) * 100
|
||
panel["is_big"] = (panel["fwd_ret60"] >= 50).astype(int)
|
||
print("面板:", len(panel), flush=True)
|
||
|
||
base = panel.dropna(subset=["fwd_ret60"])
|
||
print("基线: 大涨率={:.2f}% avg60={:.2f}%".format(
|
||
base["is_big"].mean()*100, base["fwd_ret60"].mean()), flush=True)
|
||
|
||
def validate(cond, label, min_n=100):
|
||
s = base[cond]
|
||
if len(s) < min_n:
|
||
print("{}: n={} 样本不足".format(label, len(s)), flush=True)
|
||
return None
|
||
rate = s["is_big"].mean()*100
|
||
avg = s["fwd_ret60"].mean()
|
||
wr = (s["fwd_ret60"]>0).mean()*100
|
||
# 去重统计信号量（同股30日）
|
||
cand = s[["code", "date"]].sort_values(["code", "date"])
|
||
cand["prev"] = cand.groupby("code")["date"].shift(1)
|
||
cand["gap"] = (pd.to_datetime(cand["date"]) - pd.to_datetime(cand["prev"])).dt.days
|
||
cand = cand[(cand["prev"].isna()) | (cand["gap"] > 30)]
|
||
cand["year"] = cand["date"].str[:4]
|
||
n_dedup = len(cand)
|
||
monthly = n_dedup / max(len(cand["year"].unique()),1) / 12
|
||
print("{}: n={} 大涨率={:.2f}% avg60={:.2f}% wr={:.1f}% | 去重{} 月均{:.1f}".format(
|
||
label, len(s), rate, avg, wr, n_dedup, monthly), flush=True)
|
||
print(" 分年:", {str(y): int(c) for y, c in cand["year"].value_counts().sort_index().items()}, flush=True)
|
||
return {"n": n_dedup, "monthly": monthly, "rate": rate, "avg": avg, "wr": wr}
|
||
|
||
print("\n=== 单因子 ===", flush=True)
|
||
validate(panel["mkt_rsi"] < 41, "大盘RSI<41")
|
||
validate(panel["mcap_q"] < 0.2, "小市值<0.2")
|
||
validate(panel["pe_q"] < 0.2, "低PE<0.2")
|
||
validate(panel["news3"] >= 1, "新闻>=1")
|
||
|
||
print("\n=== 两因子组合 ===", flush=True)
|
||
validate((panel["mkt_rsi"] < 41) & (panel["mcap_q"] < 0.2), "弱市+小市值")
|
||
validate((panel["mkt_rsi"] < 41) & (panel["pe_q"] < 0.2), "弱市+低PE")
|
||
validate((panel["mcap_q"] < 0.2) & (panel["pe_q"] < 0.2), "小市值+低PE")
|
||
|
||
print("\n=== 三因子组合 ===", flush=True)
|
||
validate((panel["mkt_rsi"] < 41) & (panel["mcap_q"] < 0.2) & (panel["pe_q"] < 0.2), "弱市+小市值+低PE")
|
||
validate((panel["mkt_rsi"] < 41) & (panel["mcap_q"] < 0.2) & (panel["news3"] >= 1), "弱市+小市值+新闻")
|
||
|
||
print("\n=== 四因子组合 ===", flush=True)
|
||
validate((panel["mkt_rsi"] < 41) & (panel["mcap_q"] < 0.2) & (panel["pe_q"] < 0.2) & (panel["news3"] >= 1), "弱市+小市值+低PE+新闻")
|
||
|
||
print("\n=== 完成 ===", flush=True)
|