jackyu66gitandCursor
7f393b93ed
删除根目录旧 Chan 模块、策略、配置、文档及 wyckoff 相关代码；更新缠论 pipeline 与笔中枢计算；补充 research 研究与 web 测试。 Co-authored-by: Cursor <cursoragent@cursor.com>
118 lines
4.7 KiB
Python
118 lines
4.7 KiB
Python
"""Step 14：对 Step 13 结果的尽职调查。
|
|
|
|
Step 13 报表里的 +533966% 是每笔全仓连续复利的产物，且不同品种交易在时间上重叠，
|
|
不可当真。本步只问四个问题：
|
|
|
|
1. 收益是不是被少数极端盈利撑起来的（去极值后还剩多少）
|
|
2. 中位数收益是正是负
|
|
3. 按固定风险仓位、按时间排序的真实权益曲线长什么样
|
|
4. 成本敏感性 —— 手续费滑点翻倍还活不活
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import warnings
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
warnings.filterwarnings("ignore")
|
|
pd.set_option("display.width", 260)
|
|
|
|
TRADES = Path(__file__).parent / "out" / "step13_all_trades.csv"
|
|
FEE = 0.0008
|
|
|
|
|
|
def main() -> None:
|
|
t = pd.read_csv(TRADES, parse_dates=["date"]).sort_values("date").reset_index(drop=True)
|
|
r = t["ret"].to_numpy()
|
|
print(f"[样本] {len(t)} 笔 {t['date'].min()} -> {t['date'].max()}\n")
|
|
|
|
print("########## 1. 收益分布 ##########")
|
|
qs = [0.01, 0.05, 0.25, 0.5, 0.75, 0.95, 0.99]
|
|
d = pd.DataFrame({
|
|
"分位": [f"P{int(q * 100)}" for q in qs],
|
|
"收益": [f"{np.quantile(r, q) * 100:+.2f}%" for q in qs],
|
|
})
|
|
print(d.to_string(index=False))
|
|
print(f"\n 均值 {r.mean() * 100:+.3f}% 中位数 {np.median(r) * 100:+.3f}% "
|
|
f"标准差 {r.std(ddof=1) * 100:.2f}% 偏度 {pd.Series(r).skew():.2f}")
|
|
|
|
print("\n########## 2. 去极值稳健性 ##########")
|
|
rows = []
|
|
for k in (0, 1, 2, 5, 10):
|
|
if k == 0:
|
|
v = r
|
|
else:
|
|
hi = np.quantile(r, 1 - k / 100)
|
|
v = r[r <= hi]
|
|
win, loss = v[v > 0], v[v <= 0]
|
|
pf = win.sum() / abs(loss.sum()) if len(loss) else np.inf
|
|
sd = v.std(ddof=1)
|
|
rows.append({
|
|
"剔除最赚的": f"{k}%", "剩余笔数": len(v),
|
|
"均收益": f"{v.mean() * 100:+.3f}%",
|
|
"PF": f"{pf:.2f}",
|
|
"t值": f"{v.mean() / (sd / np.sqrt(len(v))):+.2f}",
|
|
})
|
|
print(pd.DataFrame(rows).to_string(index=False))
|
|
print(" 剔掉最赚的 5% 后仍为正，才说明不是靠几笔暴利撑着。")
|
|
|
|
print("\n########## 3. 真实权益曲线（固定 1% 风险，按时间顺序，不重叠加仓）##########")
|
|
# 每笔风险敞口固定为账户 1%，收益按 ret/单笔风险幅度折算
|
|
risk_per_trade = 0.01
|
|
# 单笔风险幅度近似为止损距离，用 |最差收益| 的稳健估计代替
|
|
stop_pct = np.abs(np.quantile(r, 0.05))
|
|
scaled = r / max(stop_pct, 1e-6) * risk_per_trade
|
|
scaled = np.clip(scaled, -0.1, 0.5) # 防止单笔异常主导
|
|
eq = np.cumprod(1 + scaled)
|
|
dd = 1 - eq / np.maximum.accumulate(eq)
|
|
yrs = (t["date"].max() - t["date"].min()).days / 365.25
|
|
cagr = eq[-1] ** (1 / yrs) - 1 if yrs > 0 else np.nan
|
|
sharpe = scaled.mean() / scaled.std(ddof=1) * np.sqrt(len(scaled) / yrs) if yrs > 0 else np.nan
|
|
print(f" 总收益 {(eq[-1] - 1) * 100:+.1f}% 年化 {cagr * 100:+.1f}% "
|
|
f"最大回撤 {dd.max() * 100:.1f}% Sharpe {sharpe:.2f}")
|
|
print(f" 交易频率 {len(t) / yrs:.0f} 笔/年（9 个品种周期合计）")
|
|
|
|
print("\n########## 4. 成本敏感性 ##########")
|
|
rows = []
|
|
for mult, name in [(0, "零成本"), (1, "0.08%基准"), (2, "0.16%"), (3, "0.24%"), (5, "0.40%")]:
|
|
v = r + FEE - mult * FEE # 还原毛收益再扣不同成本
|
|
win, loss = v[v > 0], v[v <= 0]
|
|
pf = win.sum() / abs(loss.sum()) if len(loss) else np.inf
|
|
sd = v.std(ddof=1)
|
|
rows.append({
|
|
"成本": name, "均收益": f"{v.mean() * 100:+.3f}%", "PF": f"{pf:.2f}",
|
|
"t值": f"{v.mean() / (sd / np.sqrt(len(v))):+.2f}",
|
|
})
|
|
print(pd.DataFrame(rows).to_string(index=False))
|
|
|
|
print("\n########## 5. 分品种周期的稳定性（剔除极值后）##########")
|
|
rows = []
|
|
for (sym, tf), g in t.groupby(["symbol", "tf"]):
|
|
if len(g) < 25:
|
|
continue
|
|
v = g["ret"].to_numpy()
|
|
hi = np.quantile(v, 0.95)
|
|
vt = v[v <= hi]
|
|
win, loss = vt[vt > 0], vt[vt <= 0]
|
|
rows.append({
|
|
"品种周期": f"{sym} {tf}", "笔数": len(v),
|
|
"原PF": f"{v[v > 0].sum() / abs(v[v <= 0].sum()):.2f}",
|
|
"去极值PF": f"{win.sum() / abs(loss.sum()):.2f}" if len(loss) else "—",
|
|
"中位收益": f"{np.median(v) * 100:+.3f}%",
|
|
})
|
|
print(pd.DataFrame(rows).to_string(index=False))
|
|
|
|
print("\n########## 6. 出场原因分布 ##########")
|
|
print(t.groupby("reason").agg(
|
|
笔数=("ret", "size"),
|
|
均收益=("ret", lambda x: f"{x.mean() * 100:+.3f}%"),
|
|
占比=("ret", lambda x: f"{len(x) / len(t) * 100:.0f}%"),
|
|
).to_string())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|