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 文档中心（策略研究章节）
96 lines
4.0 KiB
Python
96 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""step27_predictive_scan.py — 预测性指标扫描（真正的由果及因）
|
|
核心：不是事后分类，而是事前预测。
|
|
对全市场所有股票-日，每个12维因子，扫描各取值区间的"未来大涨概率"
|
|
预测目标：fwd_ret60 >= 50%（未来60日累计涨50% = 大涨段）
|
|
找事前有预测力的因子组合 → 形成信号
|
|
"""
|
|
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["flow5_q"] = panel.groupby("date")["flow5"].transform(lambda x: x.rank(pct=True))
|
|
print("面板:", len(panel), flush=True)
|
|
|
|
# 预测目标：未来大涨
|
|
panel["is_big_future"] = (panel["fwd_ret60"] >= 50).astype(int)
|
|
base = panel.dropna(subset=["fwd_ret60"])
|
|
base_rate = base["is_big_future"].mean() * 100
|
|
base_avg = base["fwd_ret60"].mean()
|
|
print("全市场基线: 大涨率={:.2f}% avg60={:.2f}%".format(base_rate, base_avg), flush=True)
|
|
print("样本:", len(base), flush=True)
|
|
|
|
out = ["# 预测性指标扫描（事前预测，非事后分类）\n",
|
|
"- 预测目标: 未来60日累计收益 >= 50%（大涨段）",
|
|
"- 基线: 全市场大涨率 {:.2f}%\n".format(base_rate),
|
|
"## 各因子分位扫描（大涨率 vs 基线）\n"]
|
|
|
|
def scan_factor(series, label, bins=None, is_rank=True):
|
|
"""扫描因子分位/取值区间的大涨率"""
|
|
print("\n=== {} ===".format(label), flush=True)
|
|
print("| 区间 | n | 大涨率 | avg60 | 超额 |", flush=True)
|
|
print("|---|---:|---:|---:|---:|", flush=True)
|
|
s = base[series].dropna()
|
|
if len(s) < 1000:
|
|
print("样本不足", flush=True)
|
|
return
|
|
if is_rank:
|
|
# 分位区间
|
|
qs = [0, 0.2, 0.4, 0.6, 0.8, 1.0]
|
|
edges = s.quantile(qs).values
|
|
for i in range(len(qs)-1):
|
|
lo, hi = edges[i], edges[i+1]
|
|
m = base[(base[series] >= lo) & (base[series] <= hi if i == len(qs)-2 else base[series] < hi)]
|
|
if len(m) < 100:
|
|
continue
|
|
rate = m["is_big_future"].mean() * 100
|
|
avg = m["fwd_ret60"].mean()
|
|
out.append("| [{:.2f},{:.2f}) | {} | {:.2f}% | {:.2f}% | {:+.2f}pp |".format(
|
|
lo, hi, len(m), rate, avg, rate - base_rate))
|
|
print("| [{:.2f},{:.2f}) | {} | {:.2f}% | {:.2f}% | {:+.2f}pp |".format(
|
|
lo, hi, len(m), rate, avg, rate - base_rate), flush=True)
|
|
else:
|
|
# 绝对值区间
|
|
for lo, hi in bins:
|
|
m = base[(base[series] >= lo) & (base[series] < hi)]
|
|
if len(m) < 100:
|
|
continue
|
|
rate = m["is_big_future"].mean() * 100
|
|
avg = m["fwd_ret60"].mean()
|
|
out.append("| [{},{}) | {} | {:.2f}% | {:.2f}% | {:+.2f}pp |".format(
|
|
lo, hi, len(m), rate, avg, rate - base_rate))
|
|
print("| [{},{}) | {} | {:.2f}% | {:.2f}% | {:+.2f}pp |".format(
|
|
lo, hi, len(m), rate, avg, rate - base_rate), flush=True)
|
|
|
|
# 技术面
|
|
scan_factor("rsi", "个股RSI")
|
|
scan_factor("bias60", "个股bias60")
|
|
scan_factor("bias20", "个股bias20")
|
|
scan_factor("dist_lo20", "距20日低点")
|
|
scan_factor("ret1", "当日涨幅")
|
|
scan_factor("ret5", "5日涨幅")
|
|
scan_factor("ret20", "20日涨幅")
|
|
scan_factor("vol_ratio", "量能比v5/v20")
|
|
# 大盘
|
|
scan_factor("mkt_rsi", "大盘RSI")
|
|
scan_factor("mkt_adx", "大盘ADX")
|
|
scan_factor("mkt_ret20", "大盘20日收益")
|
|
# 行业
|
|
scan_factor("sec_ret20", "行业20日动量")
|
|
# 资金
|
|
scan_factor("flow5_q", "资金流5日分位")
|
|
# 基本面
|
|
scan_factor("mcap_q", "市值分位")
|
|
scan_factor("pe_q", "PE分位")
|
|
scan_factor("pb_q", "PB分位")
|
|
# 消息
|
|
scan_factor("news3", "3日新闻数", is_rank=False, bins=[(0,1),(1,2),(2,4),(4,8),(8,100)])
|
|
|
|
with open("/tmp/step27_predictive_report.md", "w", encoding="utf-8") as f:
|
|
f.write("\n".join(out))
|
|
print("\n报告已写 /tmp/step27_predictive_report.md", flush=True)
|
|
print("=== 完成 ===", flush=True)
|