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 文档中心（策略研究章节）
203 lines
7.9 KiB
Python
203 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
||
"""step43_negative_factor.py — 大回撤案例负面因子提取
|
||
方法：对每个信号模拟交易（信号日买→止损/止盈/40日），
|
||
分组：亏损组(ret<0) vs 盈利组(ret>0)，对比信号日12维特征+市场气氛，
|
||
提取亏损交易共有的负面因子。
|
||
"""
|
||
import numpy as np
|
||
import pandas as pd
|
||
import sqlite3
|
||
|
||
print("=== 加载 ===", flush=True)
|
||
panel = pd.read_pickle("/tmp/panel_12d.pkl")
|
||
panel = panel.sort_values(["code", "date"]).reset_index(drop=True)
|
||
panel["_key"] = panel["code"] + "_" + panel["date"]
|
||
pos_map = {k: i for i, k in enumerate(panel["_key"])}
|
||
print("面板:", len(panel), flush=True)
|
||
|
||
# 大盘/气氛指标
|
||
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["mkt_ret5"] = idx_df["close"].pct_change(5) * 100
|
||
idx_df["hi60"] = idx_df["high"].rolling(60).max()
|
||
idx_df["mkt_dd60"] = (idx_df["close"] / idx_df["hi60"] - 1) * 100
|
||
close_arr = idx_df["close"].values
|
||
down_days, streak, prev = [], 0, None
|
||
for c in close_arr:
|
||
streak = streak + 1 if (prev is not None and c < prev) else 0
|
||
down_days.append(streak)
|
||
prev = c
|
||
idx_df["mkt_down_days"] = down_days
|
||
mkt_map = {}
|
||
for r in idx_df.itertuples():
|
||
mkt_map[r.date] = (r.mkt_ret5, r.mkt_dd60, r.mkt_down_days)
|
||
|
||
news_cnt = pd.read_sql("SELECT substr(date,1,10) d, COUNT(*) c FROM stock_news GROUP BY d", conn)
|
||
news_cnt["d"] = news_cnt["d"].astype(str)
|
||
news_cnt["mkt_news5"] = news_cnt["c"].rolling(5, min_periods=1).mean()
|
||
news_map = dict(zip(news_cnt["d"], news_cnt["mkt_news5"]))
|
||
news_med = news_cnt["mkt_news5"].median()
|
||
print("新闻中位:", news_med, flush=True)
|
||
|
||
# 信号（step37 同款）
|
||
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)
|
||
)
|
||
panel["_dd60"] = panel["date"].map(lambda d: mkt_map[d][1] if d in mkt_map else np.nan)
|
||
sig_cond = sig_cond & (panel["_dd60"] <= -5)
|
||
cand = panel[sig_cond][["code", "date"]].copy()
|
||
cand = cand.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)]
|
||
print("信号:", len(cand), flush=True)
|
||
|
||
# K线
|
||
codes = cand["code"].unique().tolist()
|
||
ph = ",".join("?" * len(codes))
|
||
df = pd.read_sql("SELECT code, date, close, high, low FROM stock_daily WHERE code IN ({}) ORDER BY code, date".format(ph), conn, params=codes)
|
||
df["date"] = df["date"].astype(str)
|
||
df["code"] = df["code"].astype(str).str.zfill(6)
|
||
df = df.sort_values(["code", "date"]).reset_index(drop=True)
|
||
df["_key"] = df["code"] + "_" + df["date"]
|
||
dpos = {k: i for i, k in enumerate(df["_key"])}
|
||
closes = df["close"].values
|
||
highs = df["high"].values
|
||
lows = df["low"].values
|
||
cand["kidx"] = (cand["code"] + "_" + cand["date"]).map(dpos)
|
||
cand = cand.dropna(subset=["kidx"]).copy()
|
||
cand["kidx"] = cand["kidx"].astype(int)
|
||
print("可定位:", len(cand), flush=True)
|
||
|
||
# 支撑压力（用于止损止盈）
|
||
import sys
|
||
sys.path.insert(0, "/tmp")
|
||
from sr_calculator import SRCalculator
|
||
sr = SRCalculator()
|
||
print("=== 计算支撑压力 + 模拟交易 ===", flush=True)
|
||
STOP_BUF = 0.05
|
||
HOLD = 40
|
||
rows = []
|
||
for i, r in enumerate(cand.itertuples()):
|
||
if i % 500 == 0:
|
||
print(" {} / {}".format(i, len(cand)), flush=True)
|
||
bars_code = sr.get_bars(r.code)
|
||
d_idx = bars_code.index[bars_code["date"] == r.date]
|
||
if len(d_idx) == 0:
|
||
continue
|
||
sr_full = sr.sr_full(r.code, d_idx[0])
|
||
pv = sr_full["pivot"]
|
||
chip = sr_full["chip"]
|
||
if not pv:
|
||
continue
|
||
sig_close = closes[r.kidx]
|
||
support = pv["s2"]
|
||
if chip and chip["chip_ss"] < sig_close:
|
||
support = max(pv["s2"], chip["chip_ss"])
|
||
resist = pv["r2"]
|
||
if chip and chip["chip_sr"] > sig_close:
|
||
resist = min(pv["r2"], chip["chip_sr"])
|
||
if support >= sig_close or resist <= sig_close:
|
||
continue
|
||
# 模拟：信号日买，支撑下5%止损/压力止盈/40日
|
||
stop = support * (1 - STOP_BUF)
|
||
tp = resist
|
||
ret = None
|
||
reason = None
|
||
for j in range(r.kidx+1, min(r.kidx+HOLD+1, len(closes))):
|
||
hi, lo = highs[j], lows[j]
|
||
if hi >= tp:
|
||
ret = (tp / sig_close - 1) * 100
|
||
reason = "tp"
|
||
break
|
||
if lo <= stop:
|
||
ret = (lo / sig_close - 1) * 100
|
||
reason = "stop"
|
||
break
|
||
if ret is None:
|
||
ret = (closes[min(r.kidx+HOLD, len(closes)-1)] / sig_close - 1) * 100
|
||
reason = "hold40"
|
||
# 信号日特征（12维 + 气氛）
|
||
row = {"code": r.code, "date": r.date, "ret": ret, "reason": reason,
|
||
"win": 1 if ret > 0 else 0, "ret_abs": abs(ret)}
|
||
feat = panel.iloc[pos_map[r.code + "_" + r.date]]
|
||
for col in ["mkt_rsi", "mkt_adx", "sec_ret20", "sec_above", "rsi", "bias60",
|
||
"bias20", "dist_lo20", "ret1", "ret5", "ret20", "vol_ratio",
|
||
"limit_up", "hi20_new", "news3", "flow1", "flow5",
|
||
"mcap_q", "pe_q", "pb_q"]:
|
||
row["sig_" + col] = feat[col]
|
||
# 气氛
|
||
mk = mkt_map.get(r.date, (np.nan, np.nan, np.nan))
|
||
row["mkt_ret5"] = mk[0]
|
||
row["mkt_dd60"] = mk[1]
|
||
row["mkt_down_days"] = mk[2]
|
||
row["mkt_news5"] = news_map.get(r.date, np.nan)
|
||
rows.append(row)
|
||
|
||
tr = pd.DataFrame(rows)
|
||
print("\n交易:", len(tr), flush=True)
|
||
tr.to_csv("/tmp/step43_trades.csv", index=False)
|
||
|
||
# ── 负面因子提取：亏损 vs 盈利 ──
|
||
print("\n=== 亏损 vs 盈利 特征对比 ===", flush=True)
|
||
win = tr[tr["win"] == 1]
|
||
lose = tr[tr["win"] == 0]
|
||
print("盈利: {} ({:.1f}%) 亏损: {} ({:.1f}%)".format(
|
||
len(win), len(win)/len(tr)*100, len(lose), len(lose)/len(tr)*100), flush=True)
|
||
print("盈利 avg={:.2f}% 亏损 avg={:.2f}%".format(
|
||
win["ret"].mean(), lose["ret"].mean()), flush=True)
|
||
|
||
out = ["# 大回撤负面因子提取\n",
|
||
"- 盈利 {} 笔 avg{:.2f}% / 亏损 {} 笔 avg{:.2f}%".format(
|
||
len(win), win["ret"].mean(), len(lose), lose["ret"].mean()),
|
||
"- 对比信号日特征：亏损组显著偏离盈利组的 = 负面因子\n",
|
||
"| 特征 | 亏损组 | 盈利组 | 差异 | 方向 |",
|
||
"|---|---:|---:|---:|---|"]
|
||
|
||
feat_cols = [c for c in tr.columns if c.startswith("sig_")] + \
|
||
["mkt_ret5", "mkt_dd60", "mkt_down_days", "mkt_news5"]
|
||
for col in feat_cols:
|
||
w = win[col].dropna()
|
||
l = lose[col].dropna()
|
||
if len(w) < 50 or len(l) < 50:
|
||
continue
|
||
wm, lm = w.mean(), l.mean()
|
||
diff = lm - wm
|
||
rel = abs(diff) / max(abs(wm), 1e-9)
|
||
if rel > 0.05:
|
||
out.append("| {} | {:.3f} | {:.3f} | {:.3f} | {} |".format(
|
||
col, lm, wm, diff, "↑亏高" if diff > 0 else "↓亏低"))
|
||
|
||
# 大亏损（<-10%）特征
|
||
out.append("\n## 大亏损（ret<-10%）vs 盈利 特征\n")
|
||
big_lose = tr[tr["ret"] < -10]
|
||
out.append("- 大亏损 {} 笔 ({}%)".format(len(big_lose), len(big_lose)/len(tr)*100))
|
||
out.append("| 特征 | 大亏损组 | 盈利组 | 差异 |")
|
||
out.append("|---|---:|---:|---:|")
|
||
for col in feat_cols:
|
||
b = big_lose[col].dropna()
|
||
w = win[col].dropna()
|
||
if len(b) < 20 or len(w) < 50:
|
||
continue
|
||
bm, wm = b.mean(), w.mean()
|
||
diff = bm - wm
|
||
rel = abs(diff) / max(abs(wm), 1e-9)
|
||
if rel > 0.05:
|
||
out.append("| {} | {:.3f} | {:.3f} | {:.3f} |".format(col, bm, wm, diff))
|
||
|
||
# 卖出原因分布
|
||
out.append("\n## 卖出原因 × 盈亏\n")
|
||
out.append("| 原因 | 笔数 | 平均收益 | 胜率 |")
|
||
out.append("|---|---:|---:|---:|")
|
||
for rs in tr["reason"].unique():
|
||
sub = tr[tr["reason"] == rs]
|
||
out.append("| {} | {} | {:.2f}% | {:.1f}% |".format(
|
||
rs, len(sub), sub["ret"].mean(), (sub["ret"]>0).mean()*100))
|
||
|
||
with open("/tmp/step43_negative_report.md", "w", encoding="utf-8") as f:
|
||
f.write("\n".join(out))
|
||
print("\n报告已写 /tmp/step43_negative_report.md", flush=True)
|
||
print("=== 完成 ===", flush=True)
|