tika/trader-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: aligner les candles du chart avec la période du backtest

Browse Source 
Bug: les markers BUY/SELL du chart utilisaient les timestamps des trades
du backtest mais les candles étaient fetchées séparément (500 candles récentes),
causant un désalignement visuel.

- Backend: /backtest retourne désormais les candles exactes du DataFrame analysé
- Frontend: Backtest.tsx utilise result.candles directement (suppression du
  fetchCandles séparé)
- Ajout: sérialisation reason/OB/LL sur les trades, overlays OB/LL bornés
  dans le temps, trade reasons expandables, filtre HTF, badge tendance HTF

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
tika
2026-03-01 01:23:42 +01:00
parent 4df8d53b1a
commit adbc41102e
10 changed files with 698 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
backend/app/core/backtester.py
View File

@@ -31,6 +31,9 @@ class BacktestTrade:
pnl_pct: Optional[float] = None
status: str = "open" # "win" | "loss" | "open"
signal_type: str = ""
reason: Optional[dict] = None
order_block: Optional[dict] = None
liquidity_level: Optional[dict] = None
@dataclass
@@ -52,6 +55,8 @@ class BacktestMetrics:
profit_factor: float
equity_curve: list[dict] = field(default_factory=list)
trades: list[BacktestTrade] = field(default_factory=list)
all_order_blocks: list[dict] = field(default_factory=list)
all_liquidity_levels: list[dict] = field(default_factory=list)
class Backtester:
@@ -62,12 +67,14 @@ class Backtester:
risk_percent: float = 1.0,
spread_pips: float = 1.5,
pip_value: float = 0.0001,
filters: Optional[list] = None,
) -> None:
self.strategy = strategy
self.initial_balance = initial_balance
self.risk_percent = risk_percent # % du capital risqué par trade
self.spread_pips = spread_pips
self.pip_value = pip_value
self._filters = filters or []
def run(
self,
@@ -87,6 +94,10 @@ class Backtester:
last_signal_time: Optional[datetime] = None
used_signals: set[tuple[str, int]] = set() # (direction, entry_price_rounded)
# Accumuler toutes les zones détectées (dédupliquées par id)
all_obs: dict[str, dict] = {}
all_lls: dict[str, dict] = {}
for i in range(window, len(df)):
candle = df.iloc[i]
@@ -111,10 +122,24 @@ class Backtester:
slice_df = df.iloc[i - window:i + 1]
result = self.strategy.analyze(slice_df)
if result.signals:
# Accumuler les zones détectées
for ob in result.order_blocks:
if ob.id not in all_obs:
all_obs[ob.id] = self._serialize_ob(ob)
for ll in result.liquidity_levels:
if ll.id not in all_lls:
all_lls[ll.id] = self._serialize_ll(ll)
# Appliquer les filtres sur les signaux
signals = result.signals
for f in self._filters:
if hasattr(f, "filter_signals"):
signals = f.filter_signals(signals)
if signals:
# Filtrer les signaux déjà exploités (éviter doublons)
new_signals = [
s for s in result.signals
s for s in signals
if last_signal_time is None or s.time > last_signal_time
]
if not new_signals:
@@ -135,6 +160,9 @@ class Backtester:
take_profit=tp,
entry_time=candle["time"],
signal_type=signal.signal_type,
reason=self._serialize_reason(signal.reason) if signal.reason else None,
order_block=self._serialize_ob(signal.order_block) if signal.order_block else None,
liquidity_level=self._serialize_ll(signal.liquidity_level) if signal.liquidity_level else None,
)
# Fermer la position ouverte en fin de période
@@ -151,7 +179,46 @@ class Backtester:
trades.append(open_trade)
equity_curve.append({"time": str(df.iloc[-1]["time"]), "balance": balance})
return self._compute_metrics(trades, balance, equity_curve)
metrics = self._compute_metrics(trades, balance, equity_curve)
metrics.all_order_blocks = list(all_obs.values())
metrics.all_liquidity_levels = list(all_lls.values())
return metrics
# ─── Serialization ─────────────────────────────────────────────────────────
@staticmethod
def _serialize_reason(reason) -> dict:
return {
"summary": reason.summary,
"swept_level_price": reason.swept_level_price,
"swept_level_direction": reason.swept_level_direction,
"ob_direction": reason.ob_direction,
"ob_top": reason.ob_top,
"ob_bottom": reason.ob_bottom,
"ob_origin_time": reason.ob_origin_time,
"filters_applied": reason.filters_applied,
}
@staticmethod
def _serialize_ob(ob) -> dict:
return {
"id": ob.id,
"direction": ob.direction,
"top": ob.top,
"bottom": ob.bottom,
"origin_time": str(ob.origin_time),
"mitigated": ob.mitigated,
}
@staticmethod
def _serialize_ll(ll) -> dict:
return {
"id": ll.id,
"direction": ll.direction,
"price": ll.price,
"origin_time": str(ll.origin_time),
"swept": ll.swept,
}
# ─── Helpers ──────────────────────────────────────────────────────────────

14
backend/app/core/strategy/base.py
View File

@@ -26,6 +26,19 @@ class LiquidityLevel:
swept: bool = False # True si le prix a dépassé ce niveau
@dataclass
class TradeReason:
"""Explication structurée de pourquoi un signal a été généré."""
summary: str # "EQH sweep at 1.08542 → Bullish OB [1.082 - 1.0835]"
swept_level_price: Optional[float] = None
swept_level_direction: Optional[str] = None # "high" | "low"
ob_direction: Optional[str] = None # "bullish" | "bearish"
ob_top: Optional[float] = None
ob_bottom: Optional[float] = None
ob_origin_time: Optional[str] = None
filters_applied: list[str] = field(default_factory=list)
@dataclass
class TradeSignal:
"""Signal de trading généré par la stratégie."""
@@ -37,6 +50,7 @@ class TradeSignal:
time: pd.Timestamp
order_block: Optional[OrderBlockZone] = None
liquidity_level: Optional[LiquidityLevel] = None
reason: Optional[TradeReason] = None
@dataclass

90
backend/app/core/strategy/htf_trend_filter.py Normal file
View File

@@ -0,0 +1,90 @@
"""
Filtre tendance Higher Timeframe (HTF).
Filtre les signaux de trading en fonction de la tendance sur le timeframe supérieur :
- Si la tendance HTF est haussière → seuls les signaux BUY sont autorisés
- Si la tendance HTF est baissière → seuls les signaux SELL sont autorisés
La tendance est déterminée par un croisement EMA (fast vs slow) sur les données HTF.
"""
from dataclasses import dataclass
from typing import Optional
import pandas as pd
from app.core.strategy.base import TradeSignal
# Mapping : timeframe courante → timeframe supérieure
HTF_MAP: dict[str, str] = {
"M1": "M5",
"M5": "M15",
"M15": "H1",
"M30": "H1",
"H1": "H4",
"H4": "D",
"D": "D",
}
@dataclass
class HTFTrendFilterParams:
ema_fast: int = 20
ema_slow: int = 50
class HTFTrendFilter:
"""Filtre les signaux selon la tendance du timeframe supérieur (EMA crossover)."""
def __init__(self, params: Optional[HTFTrendFilterParams] = None) -> None:
self.params = params or HTFTrendFilterParams()
self._trend: Optional[str] = None # "bullish" | "bearish" | "neutral"
@staticmethod
def get_htf_granularity(current_granularity: str) -> str:
return HTF_MAP.get(current_granularity, current_granularity)
def determine_trend(self, htf_df: pd.DataFrame) -> str:
"""Détermine la tendance via EMA crossover. Retourne 'bullish', 'bearish' ou 'neutral'."""
if len(htf_df) < self.params.ema_slow + 5:
self._trend = "neutral"
return self._trend
close = htf_df["close"]
ema_fast = close.ewm(span=self.params.ema_fast, adjust=False).mean()
ema_slow = close.ewm(span=self.params.ema_slow, adjust=False).mean()
if ema_fast.iloc[-1] > ema_slow.iloc[-1]:
self._trend = "bullish"
else:
self._trend = "bearish"
return self._trend
def filter_signals(self, signals: list[TradeSignal]) -> list[TradeSignal]:
"""Supprime les signaux qui vont contre la tendance HTF."""
if self._trend is None or self._trend == "neutral":
return signals
filtered = []
for signal in signals:
allowed = (
(self._trend == "bullish" and signal.direction == "buy")
or (self._trend == "bearish" and signal.direction == "sell")
)
if allowed:
if signal.reason:
signal.reason.filters_applied.append(
f"HTF {self._trend} → {signal.direction.upper()} autorise"
)
filtered.append(signal)
return filtered
def get_params(self) -> dict:
return {
"ema_fast": self.params.ema_fast,
"ema_slow": self.params.ema_slow,
"detected_trend": self._trend,
}

22
backend/app/core/strategy/order_block_sweep.py
View File

@@ -26,6 +26,7 @@ from app.core.strategy.base import (
AnalysisResult,
LiquidityLevel,
OrderBlockZone,
TradeReason,
TradeSignal,
)
@@ -120,7 +121,7 @@ class OrderBlockSweepStrategy(AbstractStrategy):
# Vérifier si déjà sweepé dans les données actuelles
swept = self._is_level_swept(df, level_price, "high", sh_times[j])
levels.append(LiquidityLevel(
id=f"EQH_{i}_{j}",
id=f"EQH_{sh_times[i]}_{sh_times[j]}",
direction="high",
price=level_price,
origin_time=pd.Timestamp(sh_times[j]),
@@ -136,7 +137,7 @@ class OrderBlockSweepStrategy(AbstractStrategy):
level_price = (sl_prices[i] + sl_prices[j]) / 2
swept = self._is_level_swept(df, level_price, "low", sl_times[j])
levels.append(LiquidityLevel(
id=f"EQL_{i}_{j}",
id=f"EQL_{sl_times[i]}_{sl_times[j]}",
direction="low",
price=level_price,
origin_time=pd.Timestamp(sl_times[j]),
@@ -183,7 +184,7 @@ class OrderBlockSweepStrategy(AbstractStrategy):
if df.loc[k, "close"] < df.loc[k, "open"]: # bearish
mitigated = self._is_ob_mitigated(df, k, "bullish", i + min_imp)
blocks.append(OrderBlockZone(
id=f"BullishOB_{k}",
id=f"BullishOB_{df.loc[k, 'time']}",
direction="bullish",
top=df.loc[k, "open"], # top = open de la bougie bearish
bottom=df.loc[k, "low"],
@@ -199,7 +200,7 @@ class OrderBlockSweepStrategy(AbstractStrategy):
if df.loc[k, "close"] > df.loc[k, "open"]: # bullish
mitigated = self._is_ob_mitigated(df, k, "bearish", i + min_imp)
blocks.append(OrderBlockZone(
id=f"BearishOB_{k}",
id=f"BearishOB_{df.loc[k, 'time']}",
direction="bearish",
top=df.loc[k, "high"],
bottom=df.loc[k, "open"], # bottom = open de la bougie bullish
@@ -369,6 +370,18 @@ class OrderBlockSweepStrategy(AbstractStrategy):
sl = ob.top + 2 * self.params.pip_value
tp = entry - (sl - entry) * self.params.rr_ratio
level_label = "EQH" if swept_level.direction == "high" else "EQL"
ob_label = "Bullish" if ob.direction == "bullish" else "Bearish"
reason = TradeReason(
summary=f"{level_label} sweep at {swept_level.price:.5f} → {ob_label} OB [{ob.bottom:.5f} - {ob.top:.5f}]",
swept_level_price=swept_level.price,
swept_level_direction=swept_level.direction,
ob_direction=ob.direction,
ob_top=ob.top,
ob_bottom=ob.bottom,
ob_origin_time=str(ob.origin_time),
)
return TradeSignal(
direction=direction,
entry_price=entry,
@@ -378,4 +391,5 @@ class OrderBlockSweepStrategy(AbstractStrategy):
time=sweep_time,
order_block=ob,
liquidity_level=swept_level,
reason=reason,
)