62 lines
3.1 KiB
Python
62 lines
3.1 KiB
Python
 | """Usage:
| ||
python -m backtest.run --strategy regime_rsi --symbol XAUUSD --csv data/XAUUSD_M15.csv --walkforward
| |||
python -m backtest.run --strategy ema_atr --symbol XAUUSD --mt5 --bars 40000
| |||
python -m backtest.run --strategy ema_atr --symbol XAUUSD --demo (données fictives, test du moteur)
| |||
"""
| |||
import argparse, os, time, yaml
| |||
from backtest import data
| |||
from backtest.engine import run_backtest
| |||
from backtest.metrics import compute_metrics, fmt
| |||
from backtest.montecarlo import monte_carlo
| |||
from backtest.walkforward import walk_forward
| |||
from backtest.report import render_report
| |||
| |||
def main():
| |||
ap = argparse.ArgumentParser()
| |||
ap.add_argument("--strategy", required=True)
| |||
ap.add_argument("--symbol", default="XAUUSD")
| |||
ap.add_argument("--tf", default=None)
| |||
ap.add_argument("--csv"); ap.add_argument("--mt5", action="store_true"); ap.add_argument("--demo", action="store_true")
| |||
ap.add_argument("--bars", type=int, default=30000)
| |||
ap.add_argument("--walkforward", action="store_true")
| |||
ap.add_argument("--out", default=None)
| |||
ap.add_argument("--config", default="config.yaml")
| |||
a = ap.parse_args()
| |||
| |||
cfg = yaml.safe_load(open(a.config, encoding="utf-8"))
| |||
bt = dict(cfg["backtest"]); bt["_filters"] = cfg["filters"]
| |||
spec = bt["symbols"][a.symbol]
| |||
entry = next(s for s in cfg["strategies"] if s["name"] == a.strategy)
| |||
tf, params = a.tf or entry["timeframe"], entry["params"]
| |||
| |||
if a.csv: df = data.load_csv(a.csv)
| |||
elif a.mt5: df = data.from_mt5(a.symbol, tf, a.bars, cfg["account"])
| |||
elif a.demo: df = data.synthetic(a.bars, start_price=2000 if a.symbol == "XAUUSD" else 1.1)
| |||
else: raise SystemExit("Choisis une source: --csv, --mt5 ou --demo")
| |||
print(f"{len(df)} barres {tf} | {df.time.iloc[0]} → {df.time.iloc[-1]}")
| |||
| |||
t0 = time.time()
| |||
r = run_backtest(df, a.strategy, params, tf, a.symbol, spec, cfg["risk"], bt)
| |||
m = compute_metrics(r["trades"], r["equity"], r["initial"])
| |||
print("BACKTEST |", fmt(m), f"| {time.time()-t0:.0f}s")
| |||
if r["halted"]: print("⚠️ Arrêt par limite de drawdown pendant le test")
| |||
| |||
mcc = cfg["backtest"]["montecarlo"]
| |||
mc = monte_carlo(r["trades"], r["initial"], mcc["runs"], mcc["ruin_dd_pct"])
| |||
if mc: print(f"MONTE CARLO | DD médian {mc['dd_p50']:.1f}% | DD 95e pct {mc['dd_p95']:.1f}% | proba perte {mc['prob_loss']:.0f}% | proba DD>=10% {mc['prob_dd10']:.0f}%")
| |||
| |||
wf = None
| |||
if a.walkforward:
| |||
grid = cfg["backtest"]["optimize"].get(a.strategy)
| |||
if not grid: raise SystemExit("Pas de grille 'optimize' pour cette stratégie")
| |||
wf = walk_forward(df, a.strategy, params, grid, tf, a.symbol, spec, cfg["risk"], bt, cfg["backtest"]["walkforward"])
| |||
if wf["metrics"]:
| |||
print("WALK-FORWARD (OOS) |", fmt(wf["metrics"]), "| WFE", f"{wf['wfe']:.0f}%" if wf["wfe"] is not None else "n/a")
| |||
| |||
os.makedirs("reports", exist_ok=True)
| |||
out = a.out or f"reports/{a.strategy}_{a.symbol}.html"
| |||
render_report(out, f"{a.strategy} — {a.symbol} {tf}", m, r["equity"], r["trades"], wf, mc, synthetic=a.demo)
| |||
print("Rapport:", out)
| |||
| |||
if __name__ == "__main__":
| |||
main()
|