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()
|