53 lines
2.9 KiB
Python
53 lines
2.9 KiB
Python
 | import itertools, random
| ||
import pandas as pd
| |||
from backtest.engine import run_backtest, WINDOW
| |||
from backtest.metrics import compute_metrics
| |||
| |||
def _score(m, min_trades):
| |||
if m["trades"] < min_trades:
| |||
return float("-inf")
| |||
return m["return_pct"] / max(m["max_dd_pct"], 1.0) # rendement / drawdown
| |||
| |||
def walk_forward(df, strat_name, base_params, grid, tf, symbol, spec, risk_cfg, bt, wf, log=print):
| |||
keys = list(grid)
| |||
combos = [dict(base_params, **dict(zip(keys, v))) for v in itertools.product(*grid.values())]
| |||
if len(combos) > wf["max_combos"]:
| |||
combos = random.Random(1).sample(combos, wf["max_combos"])
| |||
n, tr_n, te_n = len(df), wf["train_bars"], wf["test_bars"]
| |||
windows, oos_trades, oos_eq = [], [], []
| |||
balance, start, k = bt["initial_balance"], 0, 0
| |||
while start + tr_n + te_n <= n:
| |||
k += 1
| |||
a, b, e = start, start + tr_n, start + tr_n + te_n
| |||
train = df.iloc[a:b].reset_index(drop=True)
| |||
best, best_s, best_m = None, float("-inf"), None
| |||
for p in combos:
| |||
r = run_backtest(train, strat_name, p, tf, symbol, spec, risk_cfg, bt)
| |||
m = compute_metrics(r["trades"], r["equity"], r["initial"])
| |||
s = _score(m, wf["min_trades"])
| |||
if s > best_s:
| |||
best, best_s, best_m = p, s, m
| |||
if best is None:
| |||
log(f"Fenêtre {k}: aucune combinaison valide (trop peu de trades), ignorée")
| |||
start += te_n
| |||
continue
| |||
test = df.iloc[b - WINDOW:e].reset_index(drop=True) # warm-up + test
| |||
r = run_backtest(test, strat_name, best, tf, symbol, spec, risk_cfg, bt, initial=balance, start_idx=WINDOW)
| |||
m = compute_metrics(r["trades"], r["equity"], r["initial"])
| |||
balance = m["final"]
| |||
oos_trades.append(r["trades"]); oos_eq.append(r["equity"])
| |||
windows.append({"k": k, "train_from": str(df.time.iloc[a])[:10], "test_from": str(df.time.iloc[b])[:10],
| |||
"test_to": str(df.time.iloc[e - 1])[:10], "params": {x: best[x] for x in keys},
| |||
"is_return": best_m["return_pct"], "is_dd": best_m["max_dd_pct"],
| |||
"oos_return": m["return_pct"], "oos_dd": m["max_dd_pct"], "oos_trades": m["trades"]})
| |||
log(f"Fenêtre {k}: IS {best_m['return_pct']:+.1f}% -> OOS {m['return_pct']:+.1f}% | {windows[-1]['params']}")
| |||
start += te_n
| |||
trades = pd.concat(oos_trades, ignore_index=True) if oos_trades else pd.DataFrame()
| |||
eq = pd.concat(oos_eq) if oos_eq else pd.Series(dtype=float)
| |||
metrics = compute_metrics(trades, eq, bt["initial_balance"]) if len(eq) else None
| |||
wfe = None
| |||
if windows:
| |||
is_avg = sum(w["is_return"] for w in windows) / len(windows)
| |||
oos_avg = sum(w["oos_return"] for w in windows) / len(windows)
| |||
wfe = (oos_avg / is_avg * 100) if is_avg > 0 else None # >50% = plutôt robuste
| |||
return {"windows": windows, "trades": trades, "equity": eq, "metrics": metrics, "wfe": wfe}
|