buckybonez
d93d790428
- Dark mode: class-based theme toggle with localStorage persistence and flash prevention - Trade History (/trades): paginated table, stats cards, equity curve chart with DB API endpoints - Backtest Viewer (/backtests): log parser for 35 backtest results, sidebar + detail + comparison tabs - Model Insights: dashboard card + dialog showing feature importance, regime distribution, training history - Alert/Signal Log (/alerts): signal stats, filterable table with execution tracking - API: 8 new endpoints with psycopg2 DB connection pool - Dark mode sweep across books page, about dialog, and all dashboard components - Architecture docs rewritten with Mermaid diagrams (23 docs) - README and FEATURES.md rewritten bilingual (Indonesian + English) - main_live.py: write model_metrics.json on startup and retrain Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1247 lines
55 KiB
Python
1247 lines
55 KiB
Python
"""
|
|
Backtest SMC-Only + Pullback Filter ONLY
|
|
=========================================
|
|
Base: backtest_smc_only.py (100% synced with main_live.py v4)
|
|
|
|
IMPROVEMENT #3 — Pullback Filter (re-enable from main_live.py):
|
|
- Block entry when short-term momentum opposes signal direction
|
|
- Uses 3 confirmations: 3-candle momentum, MACD histogram, price vs EMA9
|
|
- ATR-based dynamic thresholds (not hardcoded)
|
|
- NO sell filter — pure pullback filter test
|
|
|
|
Usage:
|
|
python backtests/backtest_pullback_only.py
|
|
"""
|
|
|
|
import polars as pl
|
|
import pandas as pd
|
|
import numpy as np
|
|
from datetime import datetime, timedelta, date
|
|
from typing import Dict, List, Tuple, Optional
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
import sys
|
|
import os
|
|
from zoneinfo import ZoneInfo
|
|
from openpyxl import Workbook
|
|
from openpyxl.styles import Font, Alignment, PatternFill, Border, Side
|
|
from openpyxl.chart import LineChart, Reference
|
|
from openpyxl.utils import get_column_letter
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from src.mt5_connector import MT5Connector
|
|
from src.smc_polars import SMCAnalyzer, SMCSignal
|
|
from src.feature_eng import FeatureEngineer
|
|
from src.regime_detector import MarketRegimeDetector, MarketRegime
|
|
from src.ml_model import TradingModel
|
|
from src.config import get_config
|
|
from src.dynamic_confidence import DynamicConfidenceManager, create_dynamic_confidence, MarketQuality
|
|
from loguru import logger
|
|
|
|
logger.remove()
|
|
logger.add(sys.stderr, level="WARNING")
|
|
|
|
WIB = ZoneInfo("Asia/Jakarta")
|
|
|
|
|
|
# ─── Enums & Dataclasses ──────────────────────────────────────
|
|
|
|
class TradeResult(Enum):
|
|
WIN = "WIN"
|
|
LOSS = "LOSS"
|
|
BREAKEVEN = "BREAKEVEN"
|
|
|
|
|
|
class ExitReason(Enum):
|
|
TAKE_PROFIT = "take_profit"
|
|
SMART_TP = "smart_tp"
|
|
PEAK_PROTECT = "peak_protect"
|
|
EARLY_EXIT = "early_exit"
|
|
EARLY_CUT = "early_cut"
|
|
MAX_LOSS = "max_loss"
|
|
STALL = "stall"
|
|
TREND_REVERSAL = "trend_reversal"
|
|
TIMEOUT = "timeout"
|
|
WEEKEND_CLOSE = "weekend_close"
|
|
TRAILING_SL = "trailing_sl"
|
|
BREAKEVEN_EXIT = "breakeven_exit"
|
|
DAILY_LIMIT = "daily_limit"
|
|
REGIME_DANGER = "regime_danger"
|
|
MARKET_SIGNAL = "market_signal"
|
|
|
|
|
|
class TradingMode(Enum):
|
|
NORMAL = "normal"
|
|
RECOVERY = "recovery"
|
|
PROTECTED = "protected"
|
|
STOPPED = "stopped"
|
|
|
|
|
|
@dataclass
|
|
class SimulatedTrade:
|
|
ticket: int
|
|
entry_time: datetime
|
|
exit_time: datetime
|
|
direction: str
|
|
entry_price: float
|
|
exit_price: float
|
|
stop_loss: float
|
|
take_profit: float
|
|
lot_size: float
|
|
profit_usd: float
|
|
profit_pips: float
|
|
result: TradeResult
|
|
exit_reason: ExitReason
|
|
smc_confidence: float
|
|
regime: str
|
|
session: str
|
|
signal_reason: str
|
|
has_bos: bool = False
|
|
has_choch: bool = False
|
|
has_fvg: bool = False
|
|
has_ob: bool = False
|
|
atr_at_entry: float = 0.0
|
|
rr_ratio: float = 0.0
|
|
trading_mode: str = "normal"
|
|
|
|
|
|
@dataclass
|
|
class BacktestStats:
|
|
total_trades: int = 0
|
|
wins: int = 0
|
|
losses: int = 0
|
|
total_profit: float = 0.0
|
|
total_loss: float = 0.0
|
|
max_drawdown: float = 0.0
|
|
max_drawdown_usd: float = 0.0
|
|
win_rate: float = 0.0
|
|
profit_factor: float = 0.0
|
|
avg_win: float = 0.0
|
|
avg_loss: float = 0.0
|
|
avg_trade: float = 0.0
|
|
expectancy: float = 0.0
|
|
sharpe_ratio: float = 0.0
|
|
trades: List[SimulatedTrade] = field(default_factory=list)
|
|
equity_curve: List[float] = field(default_factory=list)
|
|
avoided_signals: int = 0
|
|
daily_limit_stops: int = 0
|
|
recovery_mode_trades: int = 0
|
|
pullback_filtered: int = 0
|
|
pullback_filtered_buy: int = 0
|
|
pullback_filtered_sell: int = 0
|
|
|
|
|
|
# ─── SMC-Only + Pullback Filter Only ────────────────────────
|
|
|
|
class SMCOnlyPullbackOnly:
|
|
"""
|
|
Base: 100% synced with main_live.py Signal Logic v4 + all exit systems.
|
|
ADDED: Pullback Filter only (no sell filter).
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
capital: float = 5000.0,
|
|
max_daily_loss_percent: float = 5.0,
|
|
max_loss_per_trade_percent: float = 1.0,
|
|
base_lot_size: float = 0.01,
|
|
max_lot_size: float = 0.02,
|
|
recovery_lot_size: float = 0.01,
|
|
trend_reversal_threshold: float = 0.75,
|
|
max_concurrent_positions: int = 2,
|
|
breakeven_pips: float = 30.0,
|
|
trail_start_pips: float = 50.0,
|
|
trail_step_pips: float = 30.0,
|
|
min_profit_to_protect: float = 5.0,
|
|
max_drawdown_from_peak: float = 50.0,
|
|
trade_cooldown_bars: int = 10,
|
|
trend_reversal_mult: float = 0.6,
|
|
):
|
|
self.capital = capital
|
|
self.max_daily_loss_usd = capital * (max_daily_loss_percent / 100)
|
|
self.max_loss_per_trade = capital * (max_loss_per_trade_percent / 100)
|
|
self.base_lot_size = base_lot_size
|
|
self.max_lot_size = max_lot_size
|
|
self.recovery_lot_size = recovery_lot_size
|
|
self.trend_reversal_threshold = trend_reversal_threshold
|
|
self.max_concurrent_positions = max_concurrent_positions
|
|
self.breakeven_pips = breakeven_pips
|
|
self.trail_start_pips = trail_start_pips
|
|
self.trail_step_pips = trail_step_pips
|
|
self.min_profit_to_protect = min_profit_to_protect
|
|
self.max_drawdown_from_peak = max_drawdown_from_peak
|
|
self.trade_cooldown_bars = trade_cooldown_bars
|
|
self.trend_reversal_mult = trend_reversal_mult
|
|
|
|
config = get_config()
|
|
self.smc = SMCAnalyzer(
|
|
swing_length=config.smc.swing_length,
|
|
ob_lookback=config.smc.ob_lookback,
|
|
)
|
|
self.features = FeatureEngineer()
|
|
self.dynamic_confidence = create_dynamic_confidence()
|
|
|
|
self.ml_model = TradingModel(model_path="models/xgboost_model.pkl")
|
|
try:
|
|
self.ml_model.load()
|
|
print(" ML model loaded (for exit evaluation)")
|
|
except Exception:
|
|
print(" [WARN] ML model not loaded — exit ML checks disabled")
|
|
|
|
self.regime_detector = MarketRegimeDetector(model_path="models/hmm_regime.pkl")
|
|
try:
|
|
self.regime_detector.load()
|
|
except Exception:
|
|
print(" [WARN] HMM model not loaded")
|
|
|
|
self._ticket_counter = 5000000
|
|
|
|
# ── Session filter (synced) ──
|
|
|
|
def _get_session_from_time(self, dt: datetime) -> Tuple[str, bool, float]:
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=ZoneInfo("UTC"))
|
|
wib_time = dt.astimezone(WIB)
|
|
hour = wib_time.hour
|
|
if 6 <= hour < 15:
|
|
return "Sydney-Tokyo", True, 0.5
|
|
elif 15 <= hour < 16:
|
|
return "Tokyo-London Overlap", True, 0.75
|
|
elif 16 <= hour < 19:
|
|
return "London Early", True, 0.8
|
|
elif 19 <= hour < 24:
|
|
return "London-NY Overlap (Golden)", True, 1.0
|
|
elif 0 <= hour < 4:
|
|
return "NY Session", True, 0.9
|
|
else:
|
|
return "Off Hours", False, 0.0
|
|
|
|
def _hours_to_golden(self, dt: datetime) -> float:
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=ZoneInfo("UTC"))
|
|
wib = dt.astimezone(WIB)
|
|
if 19 <= wib.hour < 24:
|
|
return 0
|
|
target = wib.replace(hour=19, minute=0, second=0, microsecond=0)
|
|
if wib.hour >= 19:
|
|
target += timedelta(days=1)
|
|
return max(0, (target - wib).total_seconds() / 3600)
|
|
|
|
def _is_near_weekend_close(self, dt: datetime) -> bool:
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=ZoneInfo("UTC"))
|
|
wib = dt.astimezone(WIB)
|
|
if wib.weekday() == 5 and wib.hour >= 4 and wib.minute >= 30:
|
|
return True
|
|
return False
|
|
|
|
# ══════════════════════════════════════════════════════════════
|
|
# Pullback Filter (100% synced from main_live.py)
|
|
# ══════════════════════════════════════════════════════════════
|
|
|
|
def _check_pullback_filter(
|
|
self,
|
|
df_slice: pl.DataFrame,
|
|
signal_direction: str,
|
|
current_price: float,
|
|
) -> Tuple[bool, str]:
|
|
"""100% synced with main_live.py._check_pullback_filter()"""
|
|
recent = df_slice.tail(10)
|
|
|
|
if len(recent) < 5:
|
|
return True, "Not enough data"
|
|
|
|
atr = 12.0
|
|
if "atr" in df_slice.columns:
|
|
atr_val = recent["atr"].to_list()[-1]
|
|
if atr_val is not None and atr_val > 0:
|
|
atr = atr_val
|
|
|
|
bounce_threshold = atr * 0.15
|
|
consolidation_threshold = atr * 0.10
|
|
|
|
closes = recent["close"].to_list()
|
|
last_3_closes = closes[-3:]
|
|
short_momentum = last_3_closes[-1] - last_3_closes[0]
|
|
momentum_direction = "UP" if short_momentum > 0 else "DOWN"
|
|
|
|
macd_hist_direction = "NEUTRAL"
|
|
if "macd_histogram" in df_slice.columns:
|
|
macd_hist = recent["macd_histogram"].to_list()
|
|
last_hist = macd_hist[-1] if macd_hist[-1] is not None else 0
|
|
prev_hist = macd_hist[-2] if macd_hist[-2] is not None else 0
|
|
if last_hist > prev_hist:
|
|
macd_hist_direction = "RISING"
|
|
else:
|
|
macd_hist_direction = "FALLING"
|
|
|
|
price_vs_ema = "NEUTRAL"
|
|
if "ema_9" in df_slice.columns:
|
|
ema_9 = recent["ema_9"].to_list()[-1]
|
|
if ema_9 is not None:
|
|
if current_price > ema_9 * 1.001:
|
|
price_vs_ema = "ABOVE"
|
|
elif current_price < ema_9 * 0.999:
|
|
price_vs_ema = "BELOW"
|
|
|
|
if signal_direction == "SELL":
|
|
if momentum_direction == "UP" and short_momentum > bounce_threshold:
|
|
return False, f"SELL blocked: Price bouncing UP (+${short_momentum:.2f})"
|
|
if macd_hist_direction == "RISING" and momentum_direction == "UP":
|
|
return False, "SELL blocked: MACD bullish + price rising"
|
|
if price_vs_ema == "ABOVE" and momentum_direction == "UP":
|
|
return False, "SELL blocked: Price above EMA9 and rising"
|
|
if momentum_direction == "DOWN":
|
|
return True, "SELL OK: Momentum aligned"
|
|
if abs(short_momentum) < consolidation_threshold:
|
|
return True, "SELL OK: Consolidation"
|
|
|
|
elif signal_direction == "BUY":
|
|
if momentum_direction == "DOWN" and short_momentum < -bounce_threshold:
|
|
return False, f"BUY blocked: Price falling DOWN (${short_momentum:.2f})"
|
|
if macd_hist_direction == "FALLING" and momentum_direction == "DOWN":
|
|
return False, "BUY blocked: MACD bearish + price falling"
|
|
if price_vs_ema == "BELOW" and momentum_direction == "DOWN":
|
|
return False, "BUY blocked: Price below EMA9 and falling"
|
|
if momentum_direction == "UP":
|
|
return True, "BUY OK: Momentum aligned"
|
|
if abs(short_momentum) < consolidation_threshold:
|
|
return True, "BUY OK: Consolidation"
|
|
|
|
return True, "Pullback check passed"
|
|
|
|
# ── Lot sizing (synced) ──
|
|
|
|
def _calculate_lot_size(self, confidence, regime, trading_mode, session_mult):
|
|
if trading_mode == TradingMode.STOPPED:
|
|
return 0
|
|
lot = self.base_lot_size
|
|
if trading_mode in (TradingMode.RECOVERY, TradingMode.PROTECTED):
|
|
lot = self.recovery_lot_size
|
|
else:
|
|
if confidence >= 0.65:
|
|
lot = self.max_lot_size
|
|
elif confidence >= 0.55:
|
|
lot = self.base_lot_size
|
|
else:
|
|
lot = self.recovery_lot_size
|
|
if regime.lower() in ["high_volatility", "crisis"]:
|
|
lot = self.recovery_lot_size
|
|
lot = max(0.01, lot * session_mult)
|
|
return round(lot, 2)
|
|
|
|
# ── Full exit simulation (UNCHANGED from baseline) ──
|
|
|
|
def _simulate_trade_exit(self, df, entry_idx, direction, entry_price, take_profit,
|
|
stop_loss, lot_size, daily_loss_so_far, feature_cols, max_bars=100):
|
|
pip_value = 10
|
|
highs = df["high"].to_list()
|
|
lows = df["low"].to_list()
|
|
closes = df["close"].to_list()
|
|
times = df["time"].to_list()
|
|
|
|
atr = 12.0
|
|
if "atr" in df.columns:
|
|
atr_list = df["atr"].to_list()
|
|
if entry_idx < len(atr_list) and atr_list[entry_idx] is not None:
|
|
atr = atr_list[entry_idx]
|
|
|
|
reversal_momentum_threshold = atr * self.trend_reversal_mult
|
|
min_loss_for_reversal_exit = atr * 0.8
|
|
|
|
profit_history = []
|
|
peak_profit = 0.0
|
|
stall_count = 0
|
|
reversal_warnings = 0
|
|
current_sl = stop_loss
|
|
breakeven_moved = False
|
|
|
|
if direction == "BUY":
|
|
target_tp_profit = (take_profit - entry_price) / 0.1 * pip_value * lot_size
|
|
else:
|
|
target_tp_profit = (entry_price - take_profit) / 0.1 * pip_value * lot_size
|
|
|
|
cached_ml_signal = ""
|
|
cached_ml_confidence = 0.5
|
|
|
|
for i in range(entry_idx + 1, min(entry_idx + max_bars, len(df))):
|
|
high = highs[i]
|
|
low = lows[i]
|
|
close = closes[i]
|
|
current_time = times[i]
|
|
|
|
if direction == "BUY":
|
|
current_pips = (close - entry_price) / 0.1
|
|
pip_profit_from_entry = current_pips
|
|
else:
|
|
current_pips = (entry_price - close) / 0.1
|
|
pip_profit_from_entry = current_pips
|
|
current_profit = current_pips * pip_value * lot_size
|
|
|
|
profit_history.append(current_profit)
|
|
if current_profit > peak_profit:
|
|
peak_profit = current_profit
|
|
|
|
bars_since_entry = i - entry_idx
|
|
|
|
if bars_since_entry % 4 == 0 and self.ml_model.fitted:
|
|
try:
|
|
df_slice = df.head(i + 1)
|
|
ml_pred = self.ml_model.predict(df_slice, feature_cols)
|
|
cached_ml_signal = ml_pred.signal
|
|
cached_ml_confidence = ml_pred.confidence
|
|
except Exception:
|
|
pass
|
|
|
|
momentum = 0.0
|
|
if len(profit_history) >= 3:
|
|
recent = profit_history[-5:] if len(profit_history) >= 5 else profit_history
|
|
profit_change = recent[-1] - recent[0]
|
|
momentum = max(-100, min(100, (profit_change / 10) * 50))
|
|
|
|
profit_growing = momentum > 0
|
|
|
|
# A) SmartPositionManager
|
|
if direction == "BUY" and high >= take_profit:
|
|
pips = (take_profit - entry_price) / 0.1
|
|
return pips * pip_value * lot_size, pips, ExitReason.TAKE_PROFIT, i, take_profit
|
|
elif direction == "SELL" and low <= take_profit:
|
|
pips = (entry_price - take_profit) / 0.1
|
|
return pips * pip_value * lot_size, pips, ExitReason.TAKE_PROFIT, i, take_profit
|
|
|
|
if breakeven_moved and current_sl > 0:
|
|
if direction == "BUY" and low <= current_sl:
|
|
pips = (current_sl - entry_price) / 0.1
|
|
reason = ExitReason.TRAILING_SL if pip_profit_from_entry >= self.trail_start_pips else ExitReason.BREAKEVEN_EXIT
|
|
return pips * pip_value * lot_size, pips, reason, i, current_sl
|
|
elif direction == "SELL" and high >= current_sl:
|
|
pips = (entry_price - current_sl) / 0.1
|
|
reason = ExitReason.TRAILING_SL if pip_profit_from_entry >= self.trail_start_pips else ExitReason.BREAKEVEN_EXIT
|
|
return pips * pip_value * lot_size, pips, reason, i, current_sl
|
|
|
|
if pip_profit_from_entry >= self.breakeven_pips and not breakeven_moved:
|
|
if direction == "BUY":
|
|
current_sl = entry_price + 2
|
|
else:
|
|
current_sl = entry_price - 2
|
|
breakeven_moved = True
|
|
|
|
if pip_profit_from_entry >= self.trail_start_pips:
|
|
trail_distance = self.trail_step_pips * 0.1
|
|
if direction == "BUY":
|
|
new_trail_sl = close - trail_distance
|
|
if new_trail_sl > current_sl:
|
|
current_sl = new_trail_sl
|
|
else:
|
|
new_trail_sl = close + trail_distance
|
|
if current_sl == 0 or new_trail_sl < current_sl:
|
|
current_sl = new_trail_sl
|
|
|
|
if peak_profit > self.min_profit_to_protect:
|
|
drawdown_pct = ((peak_profit - current_profit) / peak_profit) * 100 if peak_profit > 0 else 0
|
|
if drawdown_pct > self.max_drawdown_from_peak:
|
|
return current_profit, current_pips, ExitReason.PEAK_PROTECT, i, close
|
|
|
|
if bars_since_entry % 5 == 0 and bars_since_entry >= 5:
|
|
if i >= 20:
|
|
ma_fast = np.mean(closes[i-4:i+1])
|
|
ma_slow = np.mean(closes[i-19:i+1])
|
|
trend = "NEUTRAL"
|
|
if ma_fast > ma_slow * 1.001:
|
|
trend = "BULLISH"
|
|
elif ma_fast < ma_slow * 0.999:
|
|
trend = "BEARISH"
|
|
|
|
roc = (closes[i] / closes[max(0,i-4)] - 1) * 100
|
|
mom_dir = "BULLISH" if roc > 0.3 else ("BEARISH" if roc < -0.3 else "NEUTRAL")
|
|
|
|
rsi_val = None
|
|
if "rsi" in df.columns:
|
|
rsi_list = df["rsi"].to_list()
|
|
if i < len(rsi_list):
|
|
rsi_val = rsi_list[i]
|
|
|
|
urgency = 0
|
|
should_exit = False
|
|
|
|
if cached_ml_confidence > 0.75:
|
|
if direction == "BUY" and cached_ml_signal == "SELL":
|
|
should_exit = True; urgency += 2
|
|
elif direction == "SELL" and cached_ml_signal == "BUY":
|
|
should_exit = True; urgency += 2
|
|
|
|
if rsi_val:
|
|
if rsi_val > 75 and direction == "BUY":
|
|
should_exit = True; urgency += 2
|
|
elif rsi_val < 25 and direction == "SELL":
|
|
should_exit = True; urgency += 2
|
|
|
|
if direction == "BUY" and trend == "BEARISH" and mom_dir == "BEARISH":
|
|
should_exit = True; urgency += 3
|
|
elif direction == "SELL" and trend == "BULLISH" and mom_dir == "BULLISH":
|
|
should_exit = True; urgency += 3
|
|
|
|
if should_exit and current_profit > self.min_profit_to_protect / 2:
|
|
return current_profit, current_pips, ExitReason.MARKET_SIGNAL, i, close
|
|
if urgency >= 7 and current_profit > 0:
|
|
return current_profit, current_pips, ExitReason.MARKET_SIGNAL, i, close
|
|
|
|
if self._is_near_weekend_close(current_time):
|
|
if current_profit > 0:
|
|
return current_profit, current_pips, ExitReason.WEEKEND_CLOSE, i, close
|
|
elif current_profit > -10:
|
|
return current_profit, current_pips, ExitReason.WEEKEND_CLOSE, i, close
|
|
|
|
# B) SmartRiskManager
|
|
if current_profit >= 15:
|
|
if current_profit >= 40:
|
|
return current_profit, current_pips, ExitReason.SMART_TP, i, close
|
|
if current_profit >= 25 and momentum < -30:
|
|
return current_profit, current_pips, ExitReason.SMART_TP, i, close
|
|
if peak_profit > 30 and current_profit < peak_profit * 0.6:
|
|
return current_profit, current_pips, ExitReason.PEAK_PROTECT, i, close
|
|
if current_profit >= 20:
|
|
progress = (current_profit / target_tp_profit) * 100 if target_tp_profit > 0 else 0
|
|
progress_score = min(40, max(0, progress * 0.4))
|
|
momentum_score = ((momentum + 100) / 200) * 30
|
|
time_penalty = min(10, bars_since_entry / 4 * 2)
|
|
tp_probability = progress_score + momentum_score + 10 - time_penalty
|
|
if tp_probability < 25:
|
|
return current_profit, current_pips, ExitReason.SMART_TP, i, close
|
|
|
|
if 5 <= current_profit < 15:
|
|
if momentum < -50 and cached_ml_confidence >= 0.65:
|
|
is_reversal = (
|
|
(direction == "BUY" and cached_ml_signal == "SELL") or
|
|
(direction == "SELL" and cached_ml_signal == "BUY")
|
|
)
|
|
if is_reversal:
|
|
return current_profit, current_pips, ExitReason.EARLY_EXIT, i, close
|
|
|
|
if current_profit < 0:
|
|
loss_percent_of_max = abs(current_profit) / self.max_loss_per_trade * 100
|
|
if momentum < -30 and loss_percent_of_max >= 30:
|
|
return current_profit, current_pips, ExitReason.EARLY_CUT, i, close
|
|
|
|
is_ml_reversal = False
|
|
if direction == "BUY" and cached_ml_signal == "SELL" and cached_ml_confidence >= self.trend_reversal_threshold:
|
|
is_ml_reversal = True; reversal_warnings += 1
|
|
elif direction == "SELL" and cached_ml_signal == "BUY" and cached_ml_confidence >= self.trend_reversal_threshold:
|
|
is_ml_reversal = True; reversal_warnings += 1
|
|
|
|
loss_moderate = abs(current_profit) > (self.max_loss_per_trade * 0.4)
|
|
if is_ml_reversal and current_profit < -8 and loss_moderate:
|
|
return current_profit, current_pips, ExitReason.TREND_REVERSAL, i, close
|
|
if reversal_warnings >= 3 and current_profit < -10:
|
|
return current_profit, current_pips, ExitReason.TREND_REVERSAL, i, close
|
|
|
|
if current_profit <= -(self.max_loss_per_trade * 0.50):
|
|
htg = self._hours_to_golden(current_time)
|
|
if htg <= 1 and htg > 0 and momentum > -40:
|
|
pass
|
|
else:
|
|
return current_profit, current_pips, ExitReason.MAX_LOSS, i, close
|
|
|
|
if len(profit_history) >= 10:
|
|
recent_range = max(profit_history[-10:]) - min(profit_history[-10:])
|
|
if recent_range < 3 and current_profit < -15:
|
|
stall_count += 1
|
|
if stall_count >= 5:
|
|
return current_profit, current_pips, ExitReason.STALL, i, close
|
|
|
|
potential_daily_loss = daily_loss_so_far + abs(min(0, current_profit))
|
|
if potential_daily_loss >= self.max_daily_loss_usd:
|
|
return current_profit, current_pips, ExitReason.DAILY_LIMIT, i, close
|
|
|
|
# C) Time-based exit
|
|
if bars_since_entry >= 16:
|
|
if current_profit < 5 and not profit_growing:
|
|
if current_profit >= 0:
|
|
return current_profit, current_pips, ExitReason.TIMEOUT, i, close
|
|
elif current_profit > -15:
|
|
return current_profit, current_pips, ExitReason.TIMEOUT, i, close
|
|
|
|
if bars_since_entry >= 24:
|
|
if current_profit < 10 or not profit_growing:
|
|
return current_profit, current_pips, ExitReason.TIMEOUT, i, close
|
|
|
|
if bars_since_entry >= 32:
|
|
return current_profit, current_pips, ExitReason.TIMEOUT, i, close
|
|
|
|
if bars_since_entry > 10:
|
|
recent_closes = closes[i-5:i+1]
|
|
mom = recent_closes[-1] - recent_closes[0]
|
|
if direction == "BUY" and mom < -reversal_momentum_threshold:
|
|
if current_profit < -min_loss_for_reversal_exit:
|
|
return current_profit, current_pips, ExitReason.TREND_REVERSAL, i, close
|
|
elif direction == "SELL" and mom > reversal_momentum_threshold:
|
|
if current_profit < -min_loss_for_reversal_exit:
|
|
return current_profit, current_pips, ExitReason.TREND_REVERSAL, i, close
|
|
|
|
final_idx = min(entry_idx + max_bars - 1, len(df) - 1)
|
|
final_price = closes[final_idx]
|
|
pips = (final_price - entry_price) / 0.1 if direction == "BUY" else (entry_price - final_price) / 0.1
|
|
return pips * pip_value * lot_size, pips, ExitReason.TIMEOUT, final_idx, final_price
|
|
|
|
# ── Main backtest run ──
|
|
|
|
def run(self, df, start_date=None, end_date=None, initial_capital=5000.0):
|
|
stats = BacktestStats()
|
|
capital = initial_capital
|
|
peak_capital = initial_capital
|
|
stats.equity_curve.append(capital)
|
|
|
|
daily_loss = 0.0
|
|
daily_profit = 0.0
|
|
daily_trades = 0
|
|
consecutive_losses = 0
|
|
trading_mode = TradingMode.NORMAL
|
|
current_date = None
|
|
|
|
feature_cols = []
|
|
if self.ml_model.fitted and self.ml_model.feature_names:
|
|
feature_cols = [f for f in self.ml_model.feature_names if f in df.columns]
|
|
|
|
times = df["time"].to_list()
|
|
start_idx = next((i for i, t in enumerate(times) if t >= start_date), 100) if start_date else 100
|
|
end_idx = next((i for i, t in enumerate(times) if t > end_date), len(df) - 100) if end_date else len(df) - 100
|
|
|
|
last_trade_idx = -self.trade_cooldown_bars * 2
|
|
|
|
print(f"\n Running SMC-Only + Pullback Filter ONLY backtest...")
|
|
print(f" Date range: {times[start_idx]} to {times[end_idx - 1]}")
|
|
print(f" Total bars: {end_idx - start_idx}")
|
|
|
|
for i in range(start_idx, end_idx):
|
|
if i - last_trade_idx < self.trade_cooldown_bars:
|
|
continue
|
|
|
|
current_time = times[i]
|
|
|
|
trade_date = current_time.date() if hasattr(current_time, 'date') else current_time
|
|
if current_date is None or trade_date != current_date:
|
|
daily_loss = 0.0
|
|
daily_profit = 0.0
|
|
daily_trades = 0
|
|
current_date = trade_date
|
|
if consecutive_losses < 2:
|
|
trading_mode = TradingMode.NORMAL
|
|
|
|
if trading_mode == TradingMode.STOPPED:
|
|
continue
|
|
|
|
session_name, can_trade, lot_mult = self._get_session_from_time(current_time)
|
|
if not can_trade:
|
|
continue
|
|
|
|
if hasattr(current_time, 'weekday') and current_time.weekday() >= 5:
|
|
continue
|
|
|
|
df_slice = df.head(i + 1)
|
|
|
|
regime = "normal"
|
|
regime_state = None
|
|
try:
|
|
if self.regime_detector.fitted:
|
|
regime_state = self.regime_detector.get_current_state(df_slice)
|
|
if regime_state:
|
|
regime = regime_state.regime.value
|
|
if regime_state.regime == MarketRegime.CRISIS:
|
|
continue
|
|
if regime_state.recommendation == "SLEEP":
|
|
continue
|
|
except Exception:
|
|
pass
|
|
|
|
ml_signal = ""
|
|
ml_confidence = 0.5
|
|
try:
|
|
if self.ml_model.fitted and feature_cols:
|
|
ml_pred = self.ml_model.predict(df_slice, feature_cols)
|
|
ml_signal = ml_pred.signal
|
|
ml_confidence = ml_pred.confidence
|
|
|
|
market_analysis = self.dynamic_confidence.analyze_market(
|
|
session=session_name, regime=regime, volatility="medium",
|
|
trend_direction=regime, has_smc_signal=True,
|
|
ml_signal=ml_signal, ml_confidence=ml_confidence,
|
|
)
|
|
if market_analysis.quality == MarketQuality.AVOID:
|
|
stats.avoided_signals += 1
|
|
continue
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
smc_signal = self.smc.generate_signal(df_slice)
|
|
except Exception:
|
|
continue
|
|
|
|
if smc_signal is None:
|
|
continue
|
|
|
|
# ══════════════════════════════════════════════════════════
|
|
# PULLBACK FILTER (the only improvement — no sell filter)
|
|
# ══════════════════════════════════════════════════════════
|
|
current_price = df_slice.tail(1)["close"].item()
|
|
can_trade_pb, pb_reason = self._check_pullback_filter(
|
|
df_slice, smc_signal.signal_type, current_price
|
|
)
|
|
if not can_trade_pb:
|
|
stats.pullback_filtered += 1
|
|
if smc_signal.signal_type == "BUY":
|
|
stats.pullback_filtered_buy += 1
|
|
else:
|
|
stats.pullback_filtered_sell += 1
|
|
continue
|
|
|
|
# SMC details
|
|
recent_df = df_slice.tail(10)
|
|
recent_bos = recent_df["bos"].to_list() if "bos" in df_slice.columns else []
|
|
recent_choch = recent_df["choch"].to_list() if "choch" in df_slice.columns else []
|
|
recent_fvg_bull = recent_df["is_fvg_bull"].to_list() if "is_fvg_bull" in df_slice.columns else []
|
|
recent_fvg_bear = recent_df["is_fvg_bear"].to_list() if "is_fvg_bear" in df_slice.columns else []
|
|
recent_obs = recent_df["ob"].to_list() if "ob" in df_slice.columns else []
|
|
|
|
has_bos = 1 in recent_bos or -1 in recent_bos
|
|
has_choch = 1 in recent_choch or -1 in recent_choch
|
|
has_fvg = any(recent_fvg_bull) or any(recent_fvg_bear)
|
|
has_ob = 1 in recent_obs or -1 in recent_obs
|
|
|
|
atr_at_entry = 12.0
|
|
if "atr" in df_slice.columns:
|
|
atr_val = df_slice.tail(1)["atr"].item()
|
|
if atr_val is not None and atr_val > 0:
|
|
atr_at_entry = atr_val
|
|
|
|
confidence = smc_signal.confidence
|
|
ml_agrees = (
|
|
(smc_signal.signal_type == "BUY" and ml_signal == "BUY") or
|
|
(smc_signal.signal_type == "SELL" and ml_signal == "SELL")
|
|
)
|
|
if ml_agrees:
|
|
confidence = (smc_signal.confidence + ml_confidence) / 2
|
|
if regime == "high_volatility":
|
|
confidence *= 0.9
|
|
|
|
lot_size = self._calculate_lot_size(confidence, regime, trading_mode, lot_mult)
|
|
if lot_size <= 0:
|
|
continue
|
|
|
|
if trading_mode == TradingMode.RECOVERY:
|
|
stats.recovery_mode_trades += 1
|
|
|
|
entry_price = smc_signal.entry_price
|
|
take_profit_price = smc_signal.take_profit
|
|
stop_loss_price = smc_signal.stop_loss
|
|
risk = abs(entry_price - stop_loss_price)
|
|
rr = abs(take_profit_price - entry_price) / risk if risk > 0 else 0
|
|
|
|
profit, pips, exit_reason, exit_idx, exit_price = self._simulate_trade_exit(
|
|
df=df, entry_idx=i, direction=smc_signal.signal_type,
|
|
entry_price=entry_price, take_profit=take_profit_price,
|
|
stop_loss=stop_loss_price, lot_size=lot_size,
|
|
daily_loss_so_far=daily_loss, feature_cols=feature_cols,
|
|
)
|
|
|
|
self._ticket_counter += 1
|
|
result = TradeResult.WIN if profit > 0 else (TradeResult.LOSS if profit < 0 else TradeResult.BREAKEVEN)
|
|
|
|
trade = SimulatedTrade(
|
|
ticket=self._ticket_counter, entry_time=current_time,
|
|
exit_time=times[exit_idx] if exit_idx < len(times) else times[-1],
|
|
direction=smc_signal.signal_type, entry_price=entry_price,
|
|
exit_price=exit_price, stop_loss=stop_loss_price,
|
|
take_profit=take_profit_price, lot_size=lot_size,
|
|
profit_usd=profit, profit_pips=pips, result=result,
|
|
exit_reason=exit_reason, smc_confidence=confidence,
|
|
regime=regime, session=session_name,
|
|
signal_reason=smc_signal.reason, has_bos=has_bos,
|
|
has_choch=has_choch, has_fvg=has_fvg, has_ob=has_ob,
|
|
atr_at_entry=atr_at_entry, rr_ratio=rr,
|
|
trading_mode=trading_mode.value,
|
|
)
|
|
stats.trades.append(trade)
|
|
|
|
stats.total_trades += 1
|
|
daily_trades += 1
|
|
capital += profit
|
|
|
|
if profit > 0:
|
|
stats.wins += 1
|
|
stats.total_profit += profit
|
|
daily_profit += profit
|
|
consecutive_losses = 0
|
|
if trading_mode == TradingMode.RECOVERY:
|
|
trading_mode = TradingMode.NORMAL
|
|
else:
|
|
stats.losses += 1
|
|
stats.total_loss += abs(profit)
|
|
daily_loss += abs(profit)
|
|
consecutive_losses += 1
|
|
|
|
if daily_loss >= self.max_daily_loss_usd:
|
|
trading_mode = TradingMode.STOPPED
|
|
stats.daily_limit_stops += 1
|
|
elif consecutive_losses >= 3 or daily_loss >= self.max_daily_loss_usd * 0.6:
|
|
trading_mode = TradingMode.PROTECTED
|
|
elif consecutive_losses >= 2:
|
|
trading_mode = TradingMode.RECOVERY
|
|
|
|
if capital > peak_capital:
|
|
peak_capital = capital
|
|
drawdown_pct = (peak_capital - capital) / peak_capital * 100
|
|
drawdown_usd = peak_capital - capital
|
|
if drawdown_pct > stats.max_drawdown:
|
|
stats.max_drawdown = drawdown_pct
|
|
stats.max_drawdown_usd = drawdown_usd
|
|
|
|
stats.equity_curve.append(capital)
|
|
last_trade_idx = exit_idx
|
|
|
|
if stats.total_trades % 100 == 0:
|
|
print(f" {stats.total_trades} trades processed...")
|
|
|
|
if stats.total_trades > 0:
|
|
stats.win_rate = stats.wins / stats.total_trades * 100
|
|
stats.avg_win = stats.total_profit / stats.wins if stats.wins > 0 else 0
|
|
stats.avg_loss = stats.total_loss / stats.losses if stats.losses > 0 else 0
|
|
stats.avg_trade = (stats.total_profit - stats.total_loss) / stats.total_trades
|
|
stats.profit_factor = stats.total_profit / stats.total_loss if stats.total_loss > 0 else float("inf")
|
|
win_prob = stats.wins / stats.total_trades
|
|
loss_prob = stats.losses / stats.total_trades
|
|
stats.expectancy = (win_prob * stats.avg_win) - (loss_prob * stats.avg_loss)
|
|
returns = [t.profit_usd for t in stats.trades]
|
|
if len(returns) > 1:
|
|
avg_return = np.mean(returns)
|
|
std_return = np.std(returns)
|
|
stats.sharpe_ratio = (avg_return / std_return) * np.sqrt(252) if std_return > 0 else 0
|
|
|
|
return stats
|
|
|
|
|
|
# ─── XLSX Report ───────────────────────────────────────────────
|
|
|
|
def generate_xlsx_report(stats, filepath, start_date, end_date):
|
|
wb = Workbook()
|
|
hf = Font(name="Calibri", bold=True, size=12, color="FFFFFF")
|
|
hfill = PatternFill(start_color="1F4E79", end_color="1F4E79", fill_type="solid")
|
|
sf = Font(name="Calibri", bold=True, size=10)
|
|
sfill = PatternFill(start_color="D6E4F0", end_color="D6E4F0", fill_type="solid")
|
|
wfill = PatternFill(start_color="C6EFCE", end_color="C6EFCE", fill_type="solid")
|
|
lfill = PatternFill(start_color="FFC7CE", end_color="FFC7CE", fill_type="solid")
|
|
bdr = Border(left=Side(style="thin"), right=Side(style="thin"), top=Side(style="thin"), bottom=Side(style="thin"))
|
|
net_pnl = stats.total_profit - stats.total_loss
|
|
|
|
ws = wb.active
|
|
ws.title = "Summary"
|
|
ws.sheet_properties.tabColor = "1F4E79"
|
|
ws.merge_cells("A1:F1")
|
|
ws["A1"] = "XAUBot AI — Pullback Filter ONLY Backtest"
|
|
ws["A1"].font = Font(name="Calibri", bold=True, size=16, color="1F4E79")
|
|
ws["A2"] = f"Period: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}"
|
|
ws["A2"].font = Font(name="Calibri", size=10, italic=True)
|
|
ws["A3"] = f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
|
|
ws["A3"].font = Font(name="Calibri", size=10, italic=True)
|
|
ws["A4"] = "Change: Pullback Filter (momentum/MACD/EMA9) — NO sell filter"
|
|
ws["A4"].font = Font(name="Calibri", size=10, italic=True, color="CC6600")
|
|
|
|
data = [
|
|
("Performance Metrics", "", True),
|
|
("Total Trades", stats.total_trades, False),
|
|
("Wins", stats.wins, False), ("Losses", stats.losses, False),
|
|
("Win Rate", f"{stats.win_rate:.1f}%", False),
|
|
("Pullback Filtered", stats.pullback_filtered, False),
|
|
(" - BUY blocked", stats.pullback_filtered_buy, False),
|
|
(" - SELL blocked", stats.pullback_filtered_sell, False),
|
|
("Avoided (AVOID)", stats.avoided_signals, False),
|
|
("Recovery Trades", stats.recovery_mode_trades, False),
|
|
("", "", False),
|
|
("Profit - Loss", "", True),
|
|
("Total Profit", f"${stats.total_profit:,.2f}", False),
|
|
("Total Loss", f"${stats.total_loss:,.2f}", False),
|
|
("Net PnL", f"${net_pnl:,.2f}", False),
|
|
("Profit Factor", f"{stats.profit_factor:.2f}", False),
|
|
("", "", False),
|
|
("Risk Metrics", "", True),
|
|
("Max Drawdown", f"{stats.max_drawdown:.1f}%", False),
|
|
("Max Drawdown ($)", f"${stats.max_drawdown_usd:,.2f}", False),
|
|
("Avg Win", f"${stats.avg_win:,.2f}", False),
|
|
("Avg Loss", f"${stats.avg_loss:,.2f}", False),
|
|
("Expectancy", f"${stats.expectancy:,.2f}", False),
|
|
("Sharpe Ratio", f"{stats.sharpe_ratio:.2f}", False),
|
|
]
|
|
row = 6
|
|
for label, value, is_header in data:
|
|
ws.cell(row=row, column=1, value=label)
|
|
ws.cell(row=row, column=2, value=value)
|
|
if is_header:
|
|
ws.cell(row=row, column=1).font = sf
|
|
ws.cell(row=row, column=1).fill = sfill
|
|
ws.cell(row=row, column=2).fill = sfill
|
|
if label == "Net PnL":
|
|
ws.cell(row=row, column=2).font = Font(bold=True, color="006100" if net_pnl > 0 else "9C0006")
|
|
row += 1
|
|
ws.column_dimensions["A"].width = 24
|
|
ws.column_dimensions["B"].width = 18
|
|
|
|
# Exit reasons
|
|
exit_counts = {}
|
|
for t in stats.trades:
|
|
exit_counts[t.exit_reason.value] = exit_counts.get(t.exit_reason.value, 0) + 1
|
|
ws.cell(row=6, column=4, value="Exit Reasons").font = sf
|
|
ws.cell(row=6, column=4).fill = sfill
|
|
ws.cell(row=6, column=5).fill = sfill
|
|
ws.cell(row=6, column=6).fill = sfill
|
|
row = 7
|
|
for reason, count in sorted(exit_counts.items(), key=lambda x: -x[1]):
|
|
pct = count / stats.total_trades * 100 if stats.total_trades > 0 else 0
|
|
ws.cell(row=row, column=4, value=reason)
|
|
ws.cell(row=row, column=5, value=count)
|
|
ws.cell(row=row, column=6, value=f"{pct:.1f}%")
|
|
row += 1
|
|
|
|
# Session
|
|
row += 1
|
|
ws.cell(row=row, column=4, value="Session Performance").font = sf
|
|
ws.cell(row=row, column=4).fill = sfill
|
|
for c in range(5, 8): ws.cell(row=row, column=c).fill = sfill
|
|
row += 1
|
|
for lbl, col in [("Session", 4), ("Trades", 5), ("WR", 6), ("PnL", 7)]:
|
|
ws.cell(row=row, column=col, value=lbl).font = Font(bold=True)
|
|
row += 1
|
|
ss = {}
|
|
for t in stats.trades:
|
|
if t.session not in ss: ss[t.session] = {"w": 0, "l": 0, "p": 0.0}
|
|
if t.result == TradeResult.WIN: ss[t.session]["w"] += 1
|
|
else: ss[t.session]["l"] += 1
|
|
ss[t.session]["p"] += t.profit_usd
|
|
for s, d in sorted(ss.items(), key=lambda x: -x[1]["p"]):
|
|
total = d["w"] + d["l"]
|
|
wr = d["w"] / total * 100 if total > 0 else 0
|
|
ws.cell(row=row, column=4, value=s)
|
|
ws.cell(row=row, column=5, value=total)
|
|
ws.cell(row=row, column=6, value=f"{wr:.1f}%")
|
|
ws.cell(row=row, column=7, value=f"${d['p']:,.2f}")
|
|
ws.cell(row=row, column=7).font = Font(color="006100" if d["p"] >= 0 else "9C0006")
|
|
row += 1
|
|
|
|
for c, w in {4: 28, 5: 10, 6: 12, 7: 14}.items():
|
|
ws.column_dimensions[get_column_letter(c)].width = w
|
|
|
|
# Trade Log
|
|
ws2 = wb.create_sheet("Trade Log")
|
|
ws2.sheet_properties.tabColor = "2E75B6"
|
|
headers = ["Ticket","Entry Time","Exit Time","Dir","Entry","Exit","SL","TP","Lot","Profit ($)","Pips","Result","Exit Reason","SMC Conf","Regime","Session","Signal","BOS","CHoCH","FVG","OB","ATR","RR","Mode"]
|
|
for col, h in enumerate(headers, 1):
|
|
cell = ws2.cell(row=1, column=col, value=h)
|
|
cell.font = hf; cell.fill = hfill; cell.alignment = Alignment(horizontal="center")
|
|
for ri, t in enumerate(stats.trades, 2):
|
|
vals = [t.ticket, t.entry_time.strftime("%Y-%m-%d %H:%M"), t.exit_time.strftime("%Y-%m-%d %H:%M"), t.direction, t.entry_price, t.exit_price, t.stop_loss, t.take_profit, t.lot_size, round(t.profit_usd,2), round(t.profit_pips,1), t.result.value, t.exit_reason.value, round(t.smc_confidence,2), t.regime, t.session, t.signal_reason, "Y" if t.has_bos else "", "Y" if t.has_choch else "", "Y" if t.has_fvg else "", "Y" if t.has_ob else "", round(t.atr_at_entry,2), round(t.rr_ratio,2), t.trading_mode]
|
|
for ci, v in enumerate(vals, 1):
|
|
cell = ws2.cell(row=ri, column=ci, value=v)
|
|
cell.border = bdr
|
|
if ci == 10 and isinstance(v, (int,float)): cell.fill = wfill if v > 0 else (lfill if v < 0 else PatternFill())
|
|
if ci == 12: cell.fill = wfill if v == "WIN" else (lfill if v == "LOSS" else PatternFill())
|
|
for col in range(1, len(headers)+1):
|
|
ws2.column_dimensions[get_column_letter(col)].width = max(11, len(headers[col-1])+3)
|
|
|
|
# Equity
|
|
ws3 = wb.create_sheet("Equity Curve")
|
|
ws3.sheet_properties.tabColor = "548235"
|
|
for c, h in enumerate(["Trade #","Equity","Drawdown ($)"], 1):
|
|
ws3.cell(row=1, column=c, value=h).font = hf
|
|
ws3.cell(row=1, column=c).fill = hfill
|
|
pk = stats.equity_curve[0] if stats.equity_curve else 5000
|
|
for idx, eq in enumerate(stats.equity_curve):
|
|
if eq > pk: pk = eq
|
|
ws3.cell(row=idx+2, column=1, value=idx)
|
|
ws3.cell(row=idx+2, column=2, value=round(eq,2))
|
|
ws3.cell(row=idx+2, column=3, value=round(pk-eq,2))
|
|
if len(stats.equity_curve) > 1:
|
|
chart = LineChart()
|
|
chart.title = "Equity Curve (Pullback Only)"
|
|
chart.style = 10; chart.y_axis.title = "Equity ($)"; chart.x_axis.title = "Trade #"
|
|
chart.width = 30; chart.height = 15
|
|
d = Reference(ws3, min_col=2, min_row=1, max_row=len(stats.equity_curve)+1)
|
|
chart.add_data(d, titles_from_data=True)
|
|
chart.series[0].graphicalProperties.line.width = 20000
|
|
ws3.add_chart(chart, "E2")
|
|
|
|
# Daily PnL
|
|
ws4 = wb.create_sheet("Daily PnL")
|
|
ws4.sheet_properties.tabColor = "BF8F00"
|
|
dpnl = {}
|
|
for t in stats.trades:
|
|
day = t.entry_time.strftime("%Y-%m-%d")
|
|
if day not in dpnl: dpnl[day] = {"trades":0,"wins":0,"profit":0.0}
|
|
dpnl[day]["trades"] += 1
|
|
if t.result == TradeResult.WIN: dpnl[day]["wins"] += 1
|
|
dpnl[day]["profit"] += t.profit_usd
|
|
for c, h in enumerate(["Date","Trades","Wins","WR","Net PnL","Cumulative"], 1):
|
|
ws4.cell(row=1, column=c, value=h).font = hf
|
|
ws4.cell(row=1, column=c).fill = hfill
|
|
cum = 0.0
|
|
for ri, (day, d) in enumerate(sorted(dpnl.items()), 2):
|
|
wr = d["wins"]/d["trades"]*100 if d["trades"]>0 else 0
|
|
cum += d["profit"]
|
|
ws4.cell(row=ri, column=1, value=day)
|
|
ws4.cell(row=ri, column=2, value=d["trades"])
|
|
ws4.cell(row=ri, column=3, value=d["wins"])
|
|
ws4.cell(row=ri, column=4, value=f"{wr:.0f}%")
|
|
ws4.cell(row=ri, column=5, value=round(d["profit"],2))
|
|
ws4.cell(row=ri, column=6, value=round(cum,2))
|
|
ws4.cell(row=ri, column=5).fill = wfill if d["profit"]>=0 else lfill
|
|
for c in range(1,7): ws4.column_dimensions[get_column_letter(c)].width = 16
|
|
|
|
# vs Baseline
|
|
ws5 = wb.create_sheet("vs Baseline")
|
|
ws5.sheet_properties.tabColor = "FF6600"
|
|
ws5["A1"] = "Comparison: Pullback Only vs Baseline"
|
|
ws5["A1"].font = Font(name="Calibri", bold=True, size=14, color="1F4E79")
|
|
ws5["A3"] = "Metric"; ws5["B3"] = "Baseline"; ws5["C3"] = "Pullback Only"; ws5["D3"] = "Delta"
|
|
for c in range(1,5):
|
|
ws5.cell(row=3, column=c).font = sf
|
|
ws5.cell(row=3, column=c).fill = sfill
|
|
baseline = {"Total Trades":686,"Wins":495,"Losses":191,"Win Rate":72.2,"Net PnL":1449.86,"Profit Factor":1.42,"Max Drawdown %":5.4,"Avg Win":9.92,"Avg Loss":18.11,"Expectancy":2.11,"Sharpe Ratio":1.98,"Early Cut":92}
|
|
improved = {"Total Trades":stats.total_trades,"Wins":stats.wins,"Losses":stats.losses,"Win Rate":stats.win_rate,"Net PnL":net_pnl,"Profit Factor":stats.profit_factor,"Max Drawdown %":stats.max_drawdown,"Avg Win":stats.avg_win,"Avg Loss":stats.avg_loss,"Expectancy":stats.expectancy,"Sharpe Ratio":stats.sharpe_ratio,"Early Cut":sum(1 for t in stats.trades if t.exit_reason==ExitReason.EARLY_CUT)}
|
|
row = 4
|
|
for m in baseline:
|
|
ws5.cell(row=row, column=1, value=m)
|
|
ws5.cell(row=row, column=2, value=baseline[m])
|
|
ws5.cell(row=row, column=3, value=improved[m])
|
|
if isinstance(baseline[m], (int,float)) and isinstance(improved[m], (int,float)):
|
|
delta = improved[m] - baseline[m]
|
|
ws5.cell(row=row, column=4, value=round(delta,2))
|
|
better = delta > 0
|
|
if m in ["Losses","Max Drawdown %","Avg Loss","Early Cut"]: better = delta < 0
|
|
ws5.cell(row=row, column=4).font = Font(bold=True, color="006100" if better else "9C0006")
|
|
row += 1
|
|
for c in range(1,5): ws5.column_dimensions[get_column_letter(c)].width = 22
|
|
|
|
wb.save(filepath)
|
|
print(f"\n Report saved: {filepath}")
|
|
|
|
|
|
# ─── Log Generator ─────────────────────────────────────────────
|
|
|
|
def generate_log(stats, filepath, start_date, end_date):
|
|
net_pnl = stats.total_profit - stats.total_loss
|
|
L = []
|
|
L.append("=" * 80)
|
|
L.append("XAUBOT AI — Pullback Filter ONLY Backtest Log")
|
|
L.append("=" * 80)
|
|
L.append(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
L.append(f"Period: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}")
|
|
L.append(f"Strategy: SMC-Only v4 + Pullback Filter (no sell filter)")
|
|
L.append("")
|
|
L.append("--- FILTER STATS ---")
|
|
L.append(f" Pullback signals blocked: {stats.pullback_filtered}")
|
|
L.append(f" BUY blocked: {stats.pullback_filtered_buy}")
|
|
L.append(f" SELL blocked: {stats.pullback_filtered_sell}")
|
|
L.append("")
|
|
L.append("--- PERFORMANCE SUMMARY ---")
|
|
L.append(f" Total Trades: {stats.total_trades}")
|
|
L.append(f" Wins: {stats.wins}")
|
|
L.append(f" Losses: {stats.losses}")
|
|
L.append(f" Win Rate: {stats.win_rate:.1f}%")
|
|
L.append(f" Total Profit: ${stats.total_profit:,.2f}")
|
|
L.append(f" Total Loss: ${stats.total_loss:,.2f}")
|
|
L.append(f" Net PnL: ${net_pnl:,.2f}")
|
|
L.append(f" Profit Factor: {stats.profit_factor:.2f}")
|
|
L.append(f" Max Drawdown: {stats.max_drawdown:.1f}% (${stats.max_drawdown_usd:,.2f})")
|
|
L.append(f" Avg Win: ${stats.avg_win:,.2f}")
|
|
L.append(f" Avg Loss: ${stats.avg_loss:,.2f}")
|
|
L.append(f" Expectancy: ${stats.expectancy:,.2f}")
|
|
L.append(f" Sharpe Ratio: {stats.sharpe_ratio:.2f}")
|
|
L.append(f" Avoided (AVOID): {stats.avoided_signals}")
|
|
L.append(f" Recovery Trades: {stats.recovery_mode_trades}")
|
|
L.append(f" Daily Stops: {stats.daily_limit_stops}")
|
|
L.append("")
|
|
L.append("--- COMPARISON vs BASELINE ---")
|
|
L.append(f" Baseline Net PnL: $1,449.86 | Improved: ${net_pnl:,.2f} | Delta: ${net_pnl-1449.86:+,.2f}")
|
|
L.append(f" Baseline WR: 72.2% | Improved: {stats.win_rate:.1f}%")
|
|
L.append(f" Baseline Trades: 686 | Improved: {stats.total_trades}")
|
|
L.append("")
|
|
|
|
L.append("--- EXIT REASON BREAKDOWN ---")
|
|
ec = {}
|
|
for t in stats.trades: ec[t.exit_reason.value] = ec.get(t.exit_reason.value, 0) + 1
|
|
for r, c in sorted(ec.items(), key=lambda x: -x[1]):
|
|
L.append(f" {r:20s}: {c:4d} ({c/stats.total_trades*100:5.1f}%)")
|
|
L.append("")
|
|
|
|
L.append("--- DIRECTION BREAKDOWN ---")
|
|
for d in ["BUY","SELL"]:
|
|
dt = [t for t in stats.trades if t.direction == d]
|
|
dw = sum(1 for t in dt if t.result == TradeResult.WIN)
|
|
dp = sum(t.profit_usd for t in dt)
|
|
dwr = dw/len(dt)*100 if dt else 0
|
|
L.append(f" {d}: {len(dt)} trades, {dwr:.1f}% WR, ${dp:,.2f}")
|
|
L.append("")
|
|
|
|
L.append("--- SESSION BREAKDOWN ---")
|
|
ss = {}
|
|
for t in stats.trades:
|
|
if t.session not in ss: ss[t.session] = {"w":0,"l":0,"p":0.0}
|
|
if t.result == TradeResult.WIN: ss[t.session]["w"] += 1
|
|
else: ss[t.session]["l"] += 1
|
|
ss[t.session]["p"] += t.profit_usd
|
|
for s, d in sorted(ss.items(), key=lambda x: -x[1]["p"]):
|
|
total = d["w"]+d["l"]
|
|
wr = d["w"]/total*100 if total > 0 else 0
|
|
L.append(f" {s:30s}: {total:3d} trades, {wr:5.1f}% WR, ${d['p']:>8,.2f}")
|
|
L.append("")
|
|
|
|
L.append("--- SMC COMPONENT ANALYSIS ---")
|
|
for cn, attr in [("BOS","has_bos"),("CHoCH","has_choch"),("FVG","has_fvg"),("OB","has_ob")]:
|
|
ct = [t for t in stats.trades if getattr(t, attr)]
|
|
cw = sum(1 for t in ct if t.result == TradeResult.WIN)
|
|
cp = sum(t.profit_usd for t in ct)
|
|
cwr = cw/len(ct)*100 if ct else 0
|
|
L.append(f" {cn:6s}: {len(ct):3d} trades, {cwr:5.1f}% WR, ${cp:>8,.2f}")
|
|
L.append("")
|
|
|
|
L.append("--- TRADE LOG ---")
|
|
L.append(f"{'#':>4} {'Entry Time':>16} {'Dir':>4} {'Entry':>10} {'Exit':>10} {'P/L($)':>8} {'Result':>6} {'Exit Reason':>18} {'Conf':>5} {'Mode':>10} {'Session':>20}")
|
|
L.append("-" * 140)
|
|
for idx, t in enumerate(stats.trades, 1):
|
|
L.append(f"{idx:4d} {t.entry_time.strftime('%Y-%m-%d %H:%M'):>16} {t.direction:>4} {t.entry_price:>10.2f} {t.exit_price:>10.2f} {t.profit_usd:>8.2f} {t.result.value:>6} {t.exit_reason.value:>18} {t.smc_confidence:>5.0%} {t.trading_mode:>10} {t.session:>20}")
|
|
L.append("\n" + "=" * 80)
|
|
L.append("END OF REPORT")
|
|
|
|
with open(filepath, "w", encoding="utf-8") as f:
|
|
f.write("\n".join(L))
|
|
print(f" Log saved: {filepath}")
|
|
|
|
|
|
# ─── Main ──────────────────────────────────────────────────────
|
|
|
|
def main():
|
|
print("=" * 70)
|
|
print("XAUBOT AI — Pullback Filter ONLY Backtest")
|
|
print("Base: SMC-Only v4 (100% synced) + Pullback Filter")
|
|
print("NO sell filter applied")
|
|
print("=" * 70)
|
|
|
|
config = get_config()
|
|
mt5 = MT5Connector(login=config.mt5_login, password=config.mt5_password,
|
|
server=config.mt5_server, path=config.mt5_path)
|
|
mt5.connect()
|
|
print(f"\nConnected to MT5")
|
|
|
|
print("Fetching XAUUSD M15 historical data...")
|
|
df = mt5.get_market_data(symbol="XAUUSD", timeframe="M15", count=50000)
|
|
if len(df) == 0:
|
|
print("ERROR: No data received"); mt5.disconnect(); return
|
|
|
|
print(f" Received {len(df)} bars")
|
|
times = df["time"].to_list()
|
|
print(f" Data range: {times[0]} to {times[-1]}")
|
|
|
|
end_date = datetime.now()
|
|
start_date = datetime(2025, 8, 1)
|
|
data_start = times[0]
|
|
if hasattr(data_start, 'replace') and data_start.tzinfo:
|
|
start_date = start_date.replace(tzinfo=data_start.tzinfo)
|
|
end_date = end_date.replace(tzinfo=data_start.tzinfo)
|
|
if data_start > start_date:
|
|
start_date = data_start + timedelta(days=5)
|
|
|
|
print(f"\n Backtest period: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}")
|
|
|
|
print("\nCalculating indicators...")
|
|
features = FeatureEngineer()
|
|
smc = SMCAnalyzer(swing_length=config.smc.swing_length, ob_lookback=config.smc.ob_lookback)
|
|
df = features.calculate_all(df, include_ml_features=True)
|
|
df = smc.calculate_all(df)
|
|
|
|
regime_detector = MarketRegimeDetector(model_path="models/hmm_regime.pkl")
|
|
try:
|
|
regime_detector.load()
|
|
df = regime_detector.predict(df)
|
|
print(" HMM regime loaded")
|
|
except Exception:
|
|
print(" [WARN] HMM not available")
|
|
print(" Indicators calculated")
|
|
|
|
bt = SMCOnlyPullbackOnly(
|
|
capital=5000.0, max_daily_loss_percent=5.0, max_loss_per_trade_percent=1.0,
|
|
base_lot_size=0.01, max_lot_size=0.02, recovery_lot_size=0.01,
|
|
breakeven_pips=30.0, trail_start_pips=50.0, trail_step_pips=30.0,
|
|
min_profit_to_protect=5.0, max_drawdown_from_peak=50.0,
|
|
trade_cooldown_bars=10, trend_reversal_mult=0.6,
|
|
)
|
|
|
|
stats = bt.run(df=df, start_date=start_date, end_date=end_date, initial_capital=5000.0)
|
|
net_pnl = stats.total_profit - stats.total_loss
|
|
|
|
print("\n" + "=" * 70)
|
|
print("PULLBACK FILTER ONLY — RESULTS")
|
|
print("=" * 70)
|
|
|
|
print(f"\n Pullback Filter Stats:")
|
|
print(f" Total blocked: {stats.pullback_filtered}")
|
|
print(f" BUY blocked: {stats.pullback_filtered_buy}")
|
|
print(f" SELL blocked: {stats.pullback_filtered_sell}")
|
|
|
|
print(f"\n Performance:")
|
|
print(f" Total Trades: {stats.total_trades}")
|
|
print(f" Wins: {stats.wins}")
|
|
print(f" Losses: {stats.losses}")
|
|
print(f" Win Rate: {stats.win_rate:.1f}%")
|
|
|
|
print(f"\n Profit/Loss:")
|
|
print(f" Total Profit: ${stats.total_profit:,.2f}")
|
|
print(f" Total Loss: ${stats.total_loss:,.2f}")
|
|
print(f" Net PnL: ${net_pnl:,.2f}")
|
|
print(f" Profit Factor: {stats.profit_factor:.2f}")
|
|
|
|
print(f"\n Risk Metrics:")
|
|
print(f" Max Drawdown: {stats.max_drawdown:.1f}% (${stats.max_drawdown_usd:,.2f})")
|
|
print(f" Avg Win: ${stats.avg_win:,.2f}")
|
|
print(f" Avg Loss: ${stats.avg_loss:,.2f}")
|
|
print(f" Expectancy: ${stats.expectancy:,.2f}")
|
|
print(f" Sharpe Ratio: {stats.sharpe_ratio:.2f}")
|
|
|
|
print(f"\n vs BASELINE:")
|
|
print(f" Baseline Net PnL: $1,449.86")
|
|
print(f" Improved Net PnL: ${net_pnl:,.2f}")
|
|
print(f" Delta: ${net_pnl - 1449.86:+,.2f}")
|
|
|
|
print(f"\n Exit Reasons:")
|
|
ec = {}
|
|
for t in stats.trades: ec[t.exit_reason.value] = ec.get(t.exit_reason.value, 0) + 1
|
|
for r, c in sorted(ec.items(), key=lambda x: -x[1]):
|
|
print(f" {r:20s}: {c} ({c/stats.total_trades*100:.1f}%)")
|
|
|
|
print(f"\n Direction:")
|
|
for d in ["BUY","SELL"]:
|
|
dt = [t for t in stats.trades if t.direction == d]
|
|
dw = sum(1 for t in dt if t.result == TradeResult.WIN)
|
|
dp = sum(t.profit_usd for t in dt)
|
|
dwr = dw/len(dt)*100 if dt else 0
|
|
print(f" {d}: {len(dt)} trades, {dwr:.1f}% WR, ${dp:,.2f}")
|
|
|
|
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
out_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "04_pullback_only_results")
|
|
os.makedirs(out_dir, exist_ok=True)
|
|
log_path = os.path.join(out_dir, f"pullback_only_{ts}.log")
|
|
xlsx_path = os.path.join(out_dir, f"pullback_only_{ts}.xlsx")
|
|
|
|
generate_log(stats, log_path, start_date, end_date)
|
|
generate_xlsx_report(stats, xlsx_path, start_date, end_date)
|
|
|
|
mt5.disconnect()
|
|
print("\n" + "=" * 70)
|
|
print(f"Output: {out_dir}")
|
|
print(f" Log: {os.path.basename(log_path)}")
|
|
print(f" Report: {os.path.basename(xlsx_path)}")
|
|
print("=" * 70)
|
|
print("Backtest complete!")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|