forked from animatedread/Warrior_EA
5a4b36bb61
- Implemented AFML part A for testing the dip-z book against search artifacts, including PBO, DSR, and CPCV metrics. - Developed AFML part B to generate time and tick bars from M1 broker data, including return distribution statistics. - Created AFML part C to build a pipeline for dip-z primary analysis, incorporating features and a random forest model for classification. - Added HCC history decoder to read and process broker M1 `.hcc` files, ensuring proper handling of data structure and integrity.
248 lines
10 KiB
Python
248 lines
10 KiB
Python
"""
|
|
AFML part A (see AFML_PLAN.md): is the dip-z book a product of the search?
|
|
|
|
Three tests on the 1,200-config neighbourhood the book was picked from:
|
|
1. PBO by CSCV (Bailey/Borwein/Lopez de Prado/Zhu 2015)
|
|
2. Deflated Sharpe Ratio (Bailey & Lopez de Prado 2014) of the chosen config
|
|
3. CPCV backtest paths (AFML ch.12) -> distribution of path Sharpe and maxDD
|
|
|
|
Equity is exit-based at 0.25% risk (applies each trade's R at exit), which
|
|
understates mark-to-market DD by ~0.6pp on this book - read maxDD as a floor.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import itertools
|
|
import sys
|
|
from concurrent.futures import ProcessPoolExecutor
|
|
|
|
import numpy as np
|
|
from scipy.stats import norm, skew, kurtosis
|
|
|
|
sys.path.insert(0, __file__.rsplit("\\", 1)[0] if "\\" in __file__ else ".")
|
|
|
|
import backtest as bt # noqa: E402
|
|
from vol_filter_test import vol_pctile # noqa: E402
|
|
|
|
IDX = ["SP500", "NAS100", "US30", "DAX40"]
|
|
PERIOD = "PERIOD_H4"
|
|
RISK = 0.0025
|
|
Z_TH = [-1.0, -1.25, -1.5, -1.75, -2.0]
|
|
Z_N = [10, 15, 20, 30, 40]
|
|
MAX_BARS = [5, 10, 15, 20]
|
|
GATE = [None, 0.30, 0.50, 0.70]
|
|
STOP = [2.0, 3.0, 4.0]
|
|
BOOK = (-1.5, 20, 10, 0.50, 3.0)
|
|
CONFIGS = list(itertools.product(Z_TH, Z_N, MAX_BARS, GATE, STOP))
|
|
|
|
_D, _P = {}, {}
|
|
|
|
|
|
def _init():
|
|
for s in IDX:
|
|
_D[s] = bt.load(s, PERIOD)
|
|
_P[s] = vol_pctile(_D[s])
|
|
|
|
|
|
def run_config(cfg):
|
|
zt, zn, mb, g, st = cfg
|
|
out = []
|
|
for s in IDX:
|
|
d = _D[s]
|
|
c = d["c"]
|
|
m = bt.sma(c, zn)
|
|
sd = bt.rolling_std(c, zn)
|
|
z = (c - m) / np.where(sd > 0, sd, np.nan)
|
|
e = np.nan_to_num(z <= zt, nan=0).astype(bool)
|
|
if g is not None:
|
|
e &= np.nan_to_num(_P[s] >= g, nan=0).astype(bool)
|
|
tr = bt.simulate(d, e, side=1, exit_ma=m, max_bars=mb, stop_atr=st)
|
|
tr = bt.add_r(tr, d, stop_atr=st)
|
|
for t in tr:
|
|
out.append((d["ts"][t["exit_i"]], t["r"], s))
|
|
out.sort(key=lambda x: x[0])
|
|
return out
|
|
|
|
|
|
def weekly_matrix(all_trades, weeks):
|
|
"""T x N matrix of weekly additive returns at RISK."""
|
|
M = np.zeros((len(weeks), len(all_trades)))
|
|
w0 = weeks[0]
|
|
for j, tr in enumerate(all_trades):
|
|
for t, r, _ in tr:
|
|
k = int((t - w0) / np.timedelta64(7, "D"))
|
|
if 0 <= k < len(weeks):
|
|
M[k, j] += RISK * r
|
|
return M
|
|
|
|
|
|
def curve_stats(rs):
|
|
if len(rs) == 0:
|
|
return 0.0, 0.0, np.nan
|
|
eq = np.concatenate([[1.0], np.cumprod(1.0 + RISK * np.asarray(rs))])
|
|
peak = np.maximum.accumulate(eq)
|
|
dd = ((peak - eq) / peak).max()
|
|
tot = eq[-1] - 1.0
|
|
return tot, dd, (tot / dd if dd > 1e-12 else np.nan)
|
|
|
|
|
|
def sharpe(x, axis=0):
|
|
sd = x.std(axis=axis, ddof=1)
|
|
return np.where(sd > 0, x.mean(axis=axis) / np.where(sd > 0, sd, 1), 0.0)
|
|
|
|
|
|
# ------------------------------------------------------------------ CSCV / PBO
|
|
def pbo_cscv(M, S=16):
|
|
T, N = M.shape
|
|
edges = np.linspace(0, T, S + 1).astype(int)
|
|
bs = np.array([M[edges[i]:edges[i + 1]].sum(0) for i in range(S)])
|
|
bq = np.array([(M[edges[i]:edges[i + 1]] ** 2).sum(0) for i in range(S)])
|
|
bn = np.diff(edges)
|
|
logits, deg = [], []
|
|
for combo in itertools.combinations(range(S), S // 2):
|
|
isb = np.zeros(S, bool)
|
|
isb[list(combo)] = True
|
|
|
|
def sr(mask):
|
|
n = bn[mask].sum()
|
|
mu = bs[mask].sum(0) / n
|
|
var = bq[mask].sum(0) / n - mu * mu
|
|
return mu / np.sqrt(np.maximum(var, 1e-18))
|
|
s_is, s_oos = sr(isb), sr(~isb)
|
|
best = int(np.argmax(s_is))
|
|
rank = (s_oos < s_oos[best]).sum() + 0.5 * ((s_oos == s_oos[best]).sum() - 1)
|
|
w = (rank + 0.5) / N # relative rank in (0,1)
|
|
logits.append(np.log(w / (1 - w)))
|
|
deg.append((s_is[best], s_oos[best]))
|
|
logits = np.array(logits)
|
|
deg = np.array(deg)
|
|
return (logits <= 0).mean(), logits, deg
|
|
|
|
|
|
# ------------------------------------------------------------------ DSR
|
|
def expected_max_sr(var_sr, N):
|
|
g = 0.5772156649
|
|
return np.sqrt(var_sr) * ((1 - g) * norm.ppf(1 - 1.0 / N) + g * norm.ppf(1 - 1.0 / (N * np.e)))
|
|
|
|
|
|
def dsr(x, sr0):
|
|
T = len(x)
|
|
sr = x.mean() / x.std(ddof=1)
|
|
g3, g4 = skew(x), kurtosis(x, fisher=False)
|
|
z = (sr - sr0) * np.sqrt(T - 1) / np.sqrt(1 - g3 * sr + (g4 - 1) / 4 * sr * sr)
|
|
return norm.cdf(z), sr, g3, g4
|
|
|
|
|
|
# ------------------------------------------------------------------ CPCV
|
|
def cpcv(all_trades, t0, t1, n_groups=10, k=2, purge_days=5, embargo_frac=0.01):
|
|
edges = [t0 + (t1 - t0) * i / n_groups for i in range(n_groups + 1)]
|
|
emb = (t1 - t0) * embargo_frac
|
|
purge = np.timedelta64(purge_days, "D")
|
|
combos = list(itertools.combinations(range(n_groups), k))
|
|
pre = [] # per config: r, group, purge zones
|
|
for tr in all_trades:
|
|
ts = np.array([t for t, _, _ in tr], dtype="datetime64[s]")
|
|
r = np.array([x for _, x, _ in tr])
|
|
g = np.minimum(((ts - t0) / (t1 - t0) * n_groups).astype(int), n_groups - 1)
|
|
zone = np.array([(ts >= edges[tg] - purge) & (ts < edges[tg + 1] + purge + emb)
|
|
for tg in range(n_groups)]) if len(ts) else np.zeros((n_groups, 0), bool)
|
|
pre.append((r, g, zone))
|
|
chosen = {} # (combo) -> config index
|
|
for cb in combos:
|
|
best, best_score = None, -np.inf
|
|
for j, (r, g, zone) in enumerate(pre):
|
|
# train = outside the test groups, minus purge before / purge+embargo after
|
|
keep = ~zone[list(cb)].any(0) if len(r) else np.zeros(0, bool)
|
|
tot, dd, rdd = curve_stats(r[keep])
|
|
score = rdd if np.isfinite(rdd) else -np.inf
|
|
if score > best_score:
|
|
best, best_score = j, score
|
|
chosen[cb] = best
|
|
# assemble paths: each group appears in (n_groups-1 choose k-1) combos
|
|
n_paths = len(combos) * k // n_groups
|
|
use = {g: [cb for cb in combos if g in cb] for g in range(n_groups)}
|
|
paths = []
|
|
for p in range(n_paths):
|
|
rs, cfgs = [], []
|
|
for g in range(n_groups):
|
|
cb = use[g][p]
|
|
j = chosen[cb]
|
|
cfgs.append(j)
|
|
rs += [r for t, r, _ in all_trades[j]
|
|
if min(int((t - t0) / (t1 - t0) * n_groups), n_groups - 1) == g]
|
|
paths.append((rs, cfgs))
|
|
return chosen, paths
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print(f"running {len(CONFIGS)} configs x {len(IDX)} indices ...", flush=True)
|
|
with ProcessPoolExecutor(max_workers=10, initializer=_init) as ex:
|
|
all_trades = list(ex.map(run_config, CONFIGS, chunksize=20))
|
|
book_j = CONFIGS.index(BOOK)
|
|
t0 = min(tr[0][0] for tr in all_trades if tr)
|
|
t1 = max(tr[-1][0] for tr in all_trades if tr)
|
|
weeks = np.arange(t0.astype("datetime64[D]"), t1.astype("datetime64[D]") + 7, 7)
|
|
M = weekly_matrix(all_trades, weeks)
|
|
months = (t1 - t0) / np.timedelta64(30, "D")
|
|
|
|
# --- family overview
|
|
stats = [curve_stats([r for _, r, _ in tr]) for tr in all_trades]
|
|
rdd = np.array([s[2] for s in stats])
|
|
dd = np.array([s[1] for s in stats])
|
|
permo = np.array([len(tr) / months for tr in all_trades])
|
|
passes = (permo >= 2) & (dd <= 0.05) & (rdd >= 2)
|
|
srw = sharpe(M)
|
|
print(f"\nweeks T={M.shape[0]} configs N={M.shape[1]}")
|
|
print(f"prop screen passes: {passes.sum()} / {len(CONFIGS)} "
|
|
f"median ret/DD {np.nanmedian(rdd):.2f} book ret/DD {rdd[book_j]:.2f} "
|
|
f"(rank {int((rdd > rdd[book_j]).sum()) + 1} of {len(CONFIGS)})")
|
|
print(f"weekly Sharpe: family median {np.median(srw):.3f}, sd {srw.std(ddof=1):.3f}, "
|
|
f"max {srw.max():.3f}; book {srw[book_j]:.3f} "
|
|
f"(ann {srw[book_j]*np.sqrt(52):.2f})")
|
|
print(f"configs with positive total: {(np.array([s[0] for s in stats]) > 0).mean():.1%}")
|
|
|
|
# --- 1. PBO
|
|
pbo, logits, deg = pbo_cscv(M, 16)
|
|
slope = np.polyfit(deg[:, 0], deg[:, 1], 1)[0]
|
|
print(f"\n[1] PBO (CSCV, S=16, {len(logits)} splits): {pbo:.3f} "
|
|
f"-> {'PASS' if pbo <= 0.20 else 'FAIL'} (<= 0.20)")
|
|
print(f" OOS Sharpe of the IS winner: median {np.median(deg[:,1]):.3f}, "
|
|
f"P(OOS SR<0) {(deg[:,1] < 0).mean():.3f}; IS->OOS slope {slope:.2f}")
|
|
|
|
# --- 2. DSR
|
|
x = M[:, book_j]
|
|
var_sr = srw.var(ddof=1)
|
|
for N in (1200, 5000):
|
|
sr0 = expected_max_sr(var_sr, N)
|
|
p, sr, g3, g4 = dsr(x, sr0)
|
|
print(f"[2] DSR N={N:>5}: SR_book {sr:.4f}/wk vs SR0 {sr0:.4f} skew {g3:.2f} "
|
|
f"kurt {g4:.2f} DSR {p:.3f} -> {'PASS' if p >= 0.95 else 'FAIL'}")
|
|
p1, _, _, _ = dsr(x, 0.0)
|
|
print(f" PSR vs 0 (no deflation): {p1:.4f}")
|
|
# the family's best, deflated - what a naive search would have shipped
|
|
jb = int(np.argmax(srw))
|
|
pbest, _, _, _ = dsr(M[:, jb], expected_max_sr(var_sr, 1200))
|
|
print(f" family-best config {CONFIGS[jb]}: SR {srw[jb]:.4f}, DSR(1200) {pbest:.3f}")
|
|
|
|
# --- 3. CPCV
|
|
print("\n[3] CPCV (10 groups, 2 test, purge 5d, embargo 1%) ... ", flush=True)
|
|
chosen, paths = cpcv(all_trades, t0, t1)
|
|
picks = {}
|
|
for j in chosen.values():
|
|
picks[CONFIGS[j]] = picks.get(CONFIGS[j], 0) + 1
|
|
print(" configs chosen on train:", sorted(picks.items(), key=lambda kv: -kv[1])[:6])
|
|
res = []
|
|
for rs, cfgs in paths:
|
|
tot, pdd, prdd = curve_stats(rs)
|
|
res.append((tot, pdd, prdd, len(rs) / months))
|
|
print(f" path: n {len(rs):>4} ({len(rs)/months:4.1f}/mo) total {tot:6.1%} "
|
|
f"maxDD {pdd:5.2%} ret/DD {prdd:5.2f}")
|
|
res = np.array(res)
|
|
ok = (res[:, 1] <= 0.05).all() and np.median(res[:, 2]) >= 2
|
|
print(f" maxDD range {res[:,1].min():.2%}..{res[:,1].max():.2%} median ret/DD "
|
|
f"{np.median(res[:,2]):.2f} -> {'PASS' if ok else 'FAIL'}")
|
|
bt_ = curve_stats([r for _, r, _ in all_trades[book_j]])
|
|
print(f" book single path: total {bt_[0]:.1%} maxDD {bt_[1]:.2%} ret/DD {bt_[2]:.2f}")
|
|
np.savez(r"C:\Users\admin\AppData\Local\Temp\2\claude\c--Users-admin-Documents-Workspaces-Warrior-EA"
|
|
r"\e314e8b8-08b3-4745-9acc-2063a9c30b23\scratchpad\afml_M.npz",
|
|
M=M, rdd=rdd, dd=dd, permo=permo, srw=srw)
|