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 文档中心（策略研究章节）
88 lines
3.8 KiB
Python
88 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""step47_drawdown_trades.py — 大回撤期开仓交易 vs 盈利交易 事前对比
|
|
关键问题：造成大回撤的那批交易，信号日(事前)指标能否与其他交易区分？
|
|
方法：
|
|
1. 从净值找大回撤期(dd<-20%)
|
|
2. 找出回撤期【开仓】的交易（信号日在回撤期内）
|
|
3. 对比 回撤期开仓 vs 盈利交易 的信号日特征
|
|
4. 检验事前可区分性（均值差异 + 分布重叠）
|
|
"""
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
print("=== 加载 ===", flush=True)
|
|
tr = pd.read_csv("/tmp/step43_trades.csv") # 含信号日特征
|
|
nav = pd.read_csv("/tmp/step37_nav.csv")
|
|
|
|
# 1. 找大回撤期
|
|
nav["cummax"] = nav["nav"].cummax()
|
|
nav["dd"] = (nav["nav"] / nav["cummax"] - 1) * 100
|
|
dd_periods = nav[nav["dd"] < -20]
|
|
print("回撤>20%天数:", len(dd_periods), flush=True)
|
|
print("回撤期分布:", dd_periods["date"].str[:4].value_counts().to_dict(), flush=True)
|
|
|
|
# 回撤期的日期集合（前后各含5天，覆盖开仓时点）
|
|
dd_dates = set(dd_periods["date"].tolist())
|
|
# 扩大窗口：回撤期开仓 = 信号日接近回撤期（前10天~后10天）
|
|
import datetime
|
|
dd_window = set()
|
|
for d in dd_dates:
|
|
dt = pd.to_datetime(d)
|
|
for i in range(-10, 11):
|
|
dd_window.add((dt + pd.Timedelta(days=i)).strftime("%Y-%m-%d"))
|
|
print("回撤窗口日期数:", len(dd_window), flush=True)
|
|
|
|
# 2. 标记回撤期开仓的交易
|
|
tr["in_dd"] = tr["date"].isin(dd_window)
|
|
dd_trades = tr[tr["in_dd"]]
|
|
normal_trades = tr[~tr["in_dd"]]
|
|
print("\n回撤期开仓交易:", len(dd_trades), "| 其他交易:", len(normal_trades), flush=True)
|
|
|
|
# 3. 对比信号日特征
|
|
win_trades = tr[tr["win"] == 1] # 盈利交易
|
|
print("\n=== 回撤期开仓 vs 盈利交易：信号日特征对比 ===", flush=True)
|
|
out = ["# 大回撤期开仓交易 事前区分性检验\n",
|
|
"- 回撤期开仓: {} 笔 avg{:.2f}% / 盈利: {} 笔 avg{:.2f}%".format(
|
|
len(dd_trades), dd_trades["ret"].mean(), len(win_trades), win_trades["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:
|
|
ddv = dd_trades[col].dropna()
|
|
wv = win_trades[col].dropna()
|
|
if len(ddv) < 10 or len(wv) < 30:
|
|
continue
|
|
ddm, wm = ddv.mean(), wv.mean()
|
|
diff = ddm - wm
|
|
rel = abs(diff) / max(abs(wm), 1e-9)
|
|
# 重叠度：两组分布的重叠程度（用标准差近似）
|
|
dd_std = ddv.std()
|
|
w_std = wv.std()
|
|
overlap = min(ddm, wm) + max(ddm, wm) # 简化
|
|
# 用 Cohen's d 衡量区分度
|
|
pooled_std = np.sqrt((ddv.std()**2 + wv.std()**2) / 2) if ddv.std() > 0 and wv.std() > 0 else 1
|
|
cohens_d = abs(diff) / pooled_std if pooled_std > 0 else 0
|
|
if rel > 0.03 or cohens_d > 0.2:
|
|
out.append("| {} | {:.3f} | {:.3f} | {:.3f} | d={:.2f} |".format(
|
|
col, ddm, wm, diff, cohens_d))
|
|
|
|
out.append("\n## 解读")
|
|
out.append("- cohen's d > 0.5 = 中等以上区分度（事前可识别）")
|
|
out.append("- d < 0.2 = 基本无法事前区分（回撤不可避免）")
|
|
|
|
with open("/tmp/step47_dd_report.md", "w", encoding="utf-8") as f:
|
|
f.write("\n".join(out))
|
|
print("报告已写 /tmp/step47_dd_report.md", flush=True)
|
|
|
|
# 4. 关键：回撤期交易本身能否被"事前规则"识别？
|
|
print("\n=== 回撤期交易的共同事前特征（聚类）===", flush=True)
|
|
print("回撤期交易信号日:")
|
|
for col in ["sig_mkt_adx", "mkt_news5", "mkt_down_days", "sig_flow1", "sig_mkt_rsi", "mkt_dd60"]:
|
|
v = dd_trades[col].dropna()
|
|
print(" {}: mean={:.2f} median={:.2f}".format(col, v.mean(), v.median()), flush=True)
|
|
|
|
print("\n=== 完成 ===", flush=True)
|