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