133 lines
6.1 KiB
Python
133 lines
6.1 KiB
Python
 | from dataclasses import dataclass
| ||
from types import SimpleNamespace
| |||
import numpy as np
| |||
import pandas as pd
| |||
from strategies import REGISTRY
| |||
from risk.manager import RiskManager
| |||
from utils.indicators import atr
| |||
| |||
WINDOW = 320 # barres passées données à la stratégie (assez pour EMA200)
| |||
| |||
@dataclass
| |||
class Pos:
| |||
type: int # 0 buy, 1 sell (même convention que MT5/RiskManager)
| |||
price_open: float
| |||
sl: float
| |||
tp: float
| |||
lots: float
| |||
open_time: object
| |||
equity_before: float
| |||
risk_money: float
| |||
| |||
def run_backtest(df, strat_name, params, tf, symbol, spec, risk_cfg, bt, initial=None, start_idx=None):
| |||
"""Simulation barre par barre. Signal à la clôture de la barre i -> exécution à l'ouverture de i+1.
| |||
Bid = prix du graphique; Ask = bid + spread. Si SL et TP sont touchés dans la même barre, le SL gagne."""
| |||
strat = REGISTRY[strat_name](tf, dict(params))
| |||
rm = RiskManager(risk_cfg)
| |||
n = len(df)
| |||
o, h, l, c = (df[k].to_numpy(float) for k in ("open", "high", "low", "close"))
| |||
times = pd.DatetimeIndex(df["time"])
| |||
a_s = atr(df, 14)
| |||
atr_arr = a_s.to_numpy()
| |||
atr_mean = a_s.rolling(100).mean().to_numpy()
| |||
point = spec["point"]
| |||
spread = spec["spread_points"] * point
| |||
slip = bt.get("slippage_points", 0) * point
| |||
comm = bt.get("commission_per_lot", 0.0)
| |||
flt = bt.get("apply_filters", True)
| |||
fcfg = bt.get("_filters", {})
| |||
| |||
def vpu(price): # USD par 1.0 de variation de prix et par lot
| |||
return spec["contract_size"] if spec["usd_quote"] else spec["contract_size"] / price
| |||
| |||
balance = float(initial if initial is not None else bt["initial_balance"])
| |||
init_bal = balance
| |||
start = max(start_idx if start_idx is not None else WINDOW, WINDOW)
| |||
eq_vals, eq_times, trades = [], [], []
| |||
pos, pending, dead = None, None, False
| |||
cur_day, day_start, day_halt, peak = None, balance, False, balance
| |||
| |||
def close_pos(i, exit_price, reason):
| |||
nonlocal balance, pos
| |||
mv = (exit_price - pos.price_open) if pos.type == 0 else (pos.price_open - exit_price)
| |||
pnl = mv * pos.lots * vpu(exit_price) - comm * pos.lots
| |||
balance += pnl
| |||
trades.append({"entry_time": pos.open_time, "exit_time": times[i], "side": "buy" if pos.type == 0 else "sell",
| |||
"lots": pos.lots, "entry": pos.price_open, "exit": exit_price, "pnl": pnl,
| |||
"r": pnl / pos.risk_money if pos.risk_money else 0.0,
| |||
"equity_before": pos.equity_before, "reason": reason})
| |||
pos = None
| |||
| |||
for i in range(start, n):
| |||
# 1) exécution du signal en attente à l'ouverture de la barre
| |||
if pending is not None and pos is None and not dead and not day_halt:
| |||
buy = pending.side == "buy"
| |||
entry = o[i] + spread + slip if buy else o[i] - slip
| |||
info = SimpleNamespace(trade_tick_size=1.0, trade_tick_value=vpu(entry), volume_step=0.01,
| |||
volume_min=0.01, volume_max=50.0)
| |||
lots = rm.lot_size(balance, pending.sl_distance, info)
| |||
if lots > 0:
| |||
sl = entry - pending.sl_distance if buy else entry + pending.sl_distance
| |||
tp = entry + pending.tp_distance if buy else entry - pending.tp_distance
| |||
pos = Pos(0 if buy else 1, entry, sl, tp, lots, times[i], balance,
| |||
balance * risk_cfg["risk_per_trade_pct"] / 100)
| |||
pending = None
| |||
| |||
# 2) SL / TP dans la barre
| |||
if pos is not None:
| |||
if pos.type == 0:
| |||
hit_sl, hit_tp = l[i] <= pos.sl, h[i] >= pos.tp
| |||
if hit_sl: close_pos(i, min(pos.sl, o[i]), "SL")
| |||
elif hit_tp: close_pos(i, pos.tp, "TP")
| |||
else:
| |||
hit_sl, hit_tp = h[i] + spread >= pos.sl, l[i] + spread <= pos.tp
| |||
if hit_sl: close_pos(i, max(pos.sl, o[i] + spread), "SL")
| |||
elif hit_tp: close_pos(i, pos.tp, "TP")
| |||
| |||
# 3) equity à la clôture
| |||
floating = 0.0
| |||
if pos is not None:
| |||
px = c[i] if pos.type == 0 else c[i] + spread
| |||
mv = (px - pos.price_open) if pos.type == 0 else (pos.price_open - px)
| |||
floating = mv * pos.lots * vpu(px)
| |||
equity = balance + floating
| |||
eq_vals.append(equity); eq_times.append(times[i])
| |||
peak = max(peak, equity)
| |||
| |||
# 4) break-even / trailing (même logique que le bot live)
| |||
if pos is not None and not np.isnan(atr_arr[i]):
| |||
px = c[i] if pos.type == 0 else c[i] + spread
| |||
new = rm.new_sl(pos, atr_arr[i], px)
| |||
if new is not None:
| |||
pos.sl = new
| |||
| |||
# 5) limites de risque
| |||
d = times[i].date()
| |||
if d != cur_day:
| |||
cur_day, day_start, day_halt = d, equity, False
| |||
daily_loss = (day_start - equity) / day_start * 100
| |||
ref = rm.initial if (rm.prop and rm.initial) else peak
| |||
dd = (ref - equity) / ref * 100
| |||
if pos is not None and (daily_loss >= rm.daily_pct or dd >= rm.dd_pct):
| |||
close_pos(i, c[i] if pos.type == 0 else c[i] + spread, "RISK")
| |||
if daily_loss >= rm.daily_pct: day_halt = True
| |||
if dd >= rm.dd_pct: dead = True
| |||
| |||
# 6) nouveau signal (exécuté à la barre suivante)
| |||
if pos is None and not day_halt and not dead and i < n - 1:
| |||
if flt and fcfg:
| |||
hr = times[i].hour
| |||
if not any(a <= hr < b for a, b in fcfg.get("sessions_utc", [[0, 24]])):
| |||
continue
| |||
rg = fcfg.get("regime", {})
| |||
if rg.get("enabled") and not np.isnan(atr_mean[i]) and atr_arr[i] / atr_mean[i] < rg["min_atr_ratio"]:
| |||
continue
| |||
pending = strat.on_bar(symbol, df.iloc[i - WINDOW + 1:i + 1])
| |||
| |||
if pos is not None:
| |||
close_pos(n - 1, c[-1] if pos.type == 0 else c[-1] + spread, "EOD")
| |||
eq_vals[-1] = balance
| |||
tr = pd.DataFrame(trades)
| |||
if tr.empty:
| |||
tr = pd.DataFrame(columns=["entry_time", "exit_time", "side", "lots", "entry", "exit", "pnl", "r", "equity_before", "reason"])
| |||
return {"trades": tr, "equity": pd.Series(eq_vals, index=pd.DatetimeIndex(eq_times)), "initial": init_bal, "halted": dead}
|