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}
|