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 文档中心（策略研究章节）
82 lines
4.0 KiB
Python
82 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""step36_mkt_phase.py — 大盘下跌阶段 vs 信号质量（控制回撤）
|
|
假设：超跌信号在"下跌末期"(恐慌见底)质量远好于"下跌初期"(刚破位)
|
|
区分大盘阶段：
|
|
- 用大盘 MA20 斜率 + 大盘距 60 日高点回撤 判断阶段
|
|
- 下跌初期: 刚跌破MA20 且 大盘距高点回撤浅(<10%)
|
|
- 下跌末期: 大盘深回撤(>15%) 且 RSI 极低
|
|
扫描各阶段信号的大涨率/avg60，找最优阶段过滤
|
|
"""
|
|
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)
|
|
|
|
# 大盘阶段指标：从大盘序列算
|
|
idx = panel[panel["code"] == "SH000001" if "SH000001" in panel["code"].values else panel["code"].str.upper().eq("SH000001")].copy() if False else None
|
|
# 用 panel 里的大盘状态列（mkt_rsi/mkt_ret20/mkt_adx 已有多日）
|
|
print("面板:", len(panel), flush=True)
|
|
|
|
# 大盘指标：距60日高点回撤需要大盘序列，从 stock_daily 加载 sh000001
|
|
import sqlite3
|
|
conn = sqlite3.connect("file:/home/hmo/MoFin/data/mofin.db?mode=ro", uri=True)
|
|
idx_df = pd.read_sql("SELECT date, close, high FROM stock_daily WHERE code='sh000001' ORDER BY date", conn)
|
|
idx_df["date"] = idx_df["date"].astype(str)
|
|
idx_df["hi60"] = idx_df["high"].rolling(60).max()
|
|
idx_df["mkt_dd60"] = (idx_df["close"] / idx_df["hi60"] - 1) * 100
|
|
idx_df["ma20"] = idx_df["close"].rolling(20).mean()
|
|
idx_df["ma20_slope"] = idx_df["ma20"].pct_change(5) * 100
|
|
panel = panel.merge(idx_df[["date", "mkt_dd60", "ma20_slope"]], on="date", how="left")
|
|
print("合并大盘阶段:", len(panel), flush=True)
|
|
|
|
# 信号
|
|
sig_cond = (
|
|
(panel["mkt_rsi"] < 50) & (panel["mcap_q"] < 0.2) & (panel["pe_q"] < 0.2) &
|
|
(panel["news3"] >= 1) & (panel["sec_ret20"] < 0) & (panel["bias60"] < -20)
|
|
)
|
|
sig = panel[sig_cond].dropna(subset=["fwd_ret60", "mkt_dd60"]).copy()
|
|
print("信号:", len(sig), flush=True)
|
|
base_rate = sig["is_big"].mean() * 100
|
|
print("信号基线: 大涨率={:.2f}% avg60={:.2f}%".format(base_rate, sig["fwd_ret60"].mean()), flush=True)
|
|
|
|
def scan(cond, label, min_n=50):
|
|
m = sig[cond]
|
|
if len(m) < min_n:
|
|
print("{}: n={} 不足".format(label, len(m)), flush=True)
|
|
return
|
|
print("{}: n={} 大涨率={:.2f}% avg60={:.2f}% wr={:.1f}% 超额={:+.2f}pp".format(
|
|
label, len(m), m["is_big"].mean()*100, m["fwd_ret60"].mean(),
|
|
(m["fwd_ret60"]>0).mean()*100, m["is_big"].mean()*100 - base_rate), flush=True)
|
|
|
|
print("\n=== 大盘距60日高点回撤（下跌深度）===", flush=True)
|
|
scan(sig["mkt_dd60"] > -5, "大盘浅回撤>-5%")
|
|
scan(sig["mkt_dd60"] <= -5, "大盘回撤-5~")
|
|
scan((sig["mkt_dd60"] > -10) & (sig["mkt_dd60"] <= -5), "回撤-5~-10%")
|
|
scan((sig["mkt_dd60"] > -15) & (sig["mkt_dd60"] <= -10), "回撤-10~-15%")
|
|
scan((sig["mkt_dd60"] > -20) & (sig["mkt_dd60"] <= -15), "回撤-15~-20%")
|
|
scan(sig["mkt_dd60"] <= -20, "深回撤<-20%")
|
|
|
|
print("\n=== 大盘MA20斜率（下跌动能）===", flush=True)
|
|
scan(sig["ma20_slope"] > 0, "MA20上升")
|
|
scan(sig["ma20_slope"] <= 0, "MA20下降")
|
|
scan((sig["ma20_slope"] > -1) & (sig["ma20_slope"] <= 0), "MA20缓降-1~0")
|
|
scan(sig["ma20_slope"] <= -1, "MA20快降<-1")
|
|
|
|
print("\n=== 大盘RSI细化 ===", flush=True)
|
|
scan((sig["mkt_rsi"] >= 20) & (sig["mkt_rsi"] < 25), "大盘RSI 20-25")
|
|
scan((sig["mkt_rsi"] >= 25) & (sig["mkt_rsi"] < 30), "大盘RSI 25-30")
|
|
scan((sig["mkt_rsi"] >= 30) & (sig["mkt_rsi"] < 35), "大盘RSI 30-35")
|
|
scan((sig["mkt_rsi"] >= 35) & (sig["mkt_rsi"] < 40), "大盘RSI 35-40")
|
|
scan((sig["mkt_rsi"] >= 40) & (sig["mkt_rsi"] < 50), "大盘RSI 40-50")
|
|
|
|
print("\n=== 组合：阶段过滤 ===", flush=True)
|
|
scan((sig["mkt_dd60"] <= -10) & (sig["ma20_slope"] <= 0), "深回撤+MA20下")
|
|
scan((sig["mkt_dd60"] <= -10), "仅深回撤<-10%")
|
|
scan((sig["mkt_rsi"] < 35) & (sig["mkt_dd60"] <= -10), "RSI<35+深回撤")
|
|
|
|
print("\n=== 完成 ===", flush=True)
|