gavindiaz/XauBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
XauBot/backtests/backtest_16_quasimodo.py
T
GifariKemal e8355b3f62 feat: add 5 dashboard features — dark mode, trade history, backtests, model insights, alerts 
- 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>
2026-02-09 05:46:54 +07:00

1120 lines
46 KiB
Python
Raw Permalink Blame History

"""
Backtest #16 — SMC + RTM Quasimodo (QM) Pattern
==================================================================
Base: SMC-Only v4 (#1 Baseline)
Added: Quasimodo pattern detection from RTM methodology
QM = 5-point reversal pattern:
Bearish: H(A) → L(B) → HH(C) → LL(D) → entry at QML (A level)
Bullish: L(A) → H(B) → LL(C) → HH(D) → entry at QML (A level)
Two improvements:
1. QM-enhanced SL: When SMC signal + QM align → use tighter SL from QM head
2. QM-only entries: When price retraces to QML zone without SMC signal
Usage:
python backtests/backtest_16_quasimodo.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"
entry_source: str = "SMC" # "SMC", "QM+SMC", "QM-only"
@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
# QM stats
qm_enhanced: int = 0 # SMC + QM aligned → QM SL used
qm_only: int = 0 # QM-only entries (no SMC signal)
standard_smc: int = 0 # Standard SMC entries (no QM)
qm_patterns_found: int = 0
# ─── QM + SMC Backtest ──────────────────────────
class QuasimodoBacktest:
"""SMC-Only v4 + RTM Quasimodo pattern detection."""
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,
# QM params
qm_lookback: int = 60, # Bars to scan for QM patterns
qm_max_age: int = 30, # Max bars since D-point for valid QM
qm_zone_tolerance_pct: float = 0.004, # 0.4% = ~$11 at $2800
qm_rr_ratio: float = 2.0, # RR for QM entries
):
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
self.qm_lookback = qm_lookback
self.qm_max_age = qm_max_age
self.qm_zone_tolerance_pct = qm_zone_tolerance_pct
self.qm_rr_ratio = qm_rr_ratio
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")
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 = 2000000
# ═══ QUASIMODO PATTERN DETECTION ═══
def _detect_qm_patterns(self, df_slice: pl.DataFrame) -> List[dict]:
"""
Detect Quasimodo patterns from swing points.
Bearish QM: H(A) → L(B) → HH(C) → LL(D)
- C > A (higher high = head)
- D < B (lower low = CHoCH / structure break)
- QML = A level (entry zone for SELL)
- SL = above C (head)
Bullish QM: L(A) → H(B) → LL(C) → HH(D)
- C < A (lower low = head)
- D > B (higher high = CHoCH / structure break)
- QML = A level (entry zone for BUY)
- SL = below C (head)
"""
n = len(df_slice)
if n < self.qm_lookback:
return []
sh_col = df_slice["swing_high"].to_list()
sl_col = df_slice["swing_low"].to_list()
# Get swing high/low levels (actual prices at swing point bars)
sh_level = df_slice["swing_high_level"].to_list()
sl_level = df_slice["swing_low_level"].to_list()
# Collect recent swing points as (index, price, type)
swings = []
scan_start = max(0, n - self.qm_lookback)
for i in range(scan_start, n):
if sh_col[i] == 1 and sh_level[i] is not None:
swings.append((i, float(sh_level[i]), "H"))
if sl_col[i] == -1 and sl_level[i] is not None:
swings.append((i, float(sl_level[i]), "L"))
# Sort by index (should already be, but ensure)
swings.sort(key=lambda x: x[0])
if len(swings) < 4:
return []
patterns = []
# Scan for QM patterns in consecutive swing points
for i in range(len(swings) - 3):
a = swings[i]
b = swings[i + 1]
c = swings[i + 2]
d = swings[i + 3]
# Bearish QM: H(A) - L(B) - H(C) - L(D)
# where C > A (higher high) and D < B (lower low)
if a[2] == "H" and b[2] == "L" and c[2] == "H" and d[2] == "L":
if c[1] > a[1] and d[1] < b[1]:
# D must be recent enough
if n - d[0] <= self.qm_max_age:
sl_price = c[1] + 2.0 # $2 above head
patterns.append({
"direction": "SELL",
"qml_level": a[1],
"head_level": c[1],
"sl_price": sl_price,
"d_idx": d[0],
"freshness": n - d[0],
})
# Bullish QM: L(A) - H(B) - L(C) - H(D)
# where C < A (lower low) and D > B (higher high)
if a[2] == "L" and b[2] == "H" and c[2] == "L" and d[2] == "H":
if c[1] < a[1] and d[1] > b[1]:
if n - d[0] <= self.qm_max_age:
sl_price = c[1] - 2.0 # $2 below head
patterns.append({
"direction": "BUY",
"qml_level": a[1],
"head_level": c[1],
"sl_price": sl_price,
"d_idx": d[0],
"freshness": n - d[0],
})
return patterns
def _find_matching_qm(self, patterns: List[dict], current_price: float, direction: str = None) -> Optional[dict]:
"""Find QM pattern where current price is near QML level."""
tolerance = current_price * self.qm_zone_tolerance_pct
best = None
for qm in patterns:
if direction and qm["direction"] != direction:
continue
dist = abs(current_price - qm["qml_level"])
if dist <= tolerance:
# For SELL: price should be AT or ABOVE QML
# For BUY: price should be AT or BELOW QML
if qm["direction"] == "SELL" and current_price >= qm["qml_level"] - tolerance:
if best is None or qm["freshness"] < best["freshness"]:
best = qm
elif qm["direction"] == "BUY" and current_price <= qm["qml_level"] + tolerance:
if best is None or qm["freshness"] < best["freshness"]:
best = qm
return best
# ── Session filter (synced) ──
def _get_session_from_time(self, dt):
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):
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):
if dt.tzinfo is None:
dt = dt.replace(tzinfo=ZoneInfo("UTC"))
wib = dt.astimezone(WIB)
return wib.weekday() == 5 and wib.hour >= 4 and wib.minute >= 30
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 (all 3 systems — same as 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_s = df.head(i + 1)
ml_pred = self.ml_model.predict(df_s, 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
momentum = max(-100, min(100, ((recent[-1] - recent[0]) / 10) * 50))
profit_growing = momentum > 0
# A) TP hit
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
# Trailing/breakeven SL hit
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
# Breakeven move
if pip_profit_from_entry >= self.breakeven_pips and not breakeven_moved:
current_sl = entry_price + 2 if direction == "BUY" else entry_price - 2
breakeven_moved = True
# Trailing SL
if pip_profit_from_entry >= self.trail_start_pips:
trail_dist = self.trail_step_pips * 0.1
if direction == "BUY":
new_sl = close - trail_dist
if new_sl > current_sl: current_sl = new_sl
else:
new_sl = close + trail_dist
if current_sl == 0 or new_sl < current_sl: current_sl = new_sl
# Peak protect
if peak_profit > self.min_profit_to_protect:
dd_pct = ((peak_profit - current_profit) / peak_profit) * 100 if peak_profit > 0 else 0
if dd_pct > self.max_drawdown_from_peak:
return current_profit, current_pips, ExitReason.PEAK_PROTECT, i, close
# Market analysis
if bars_since_entry % 5 == 0 and bars_since_entry >= 5 and 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") or \
(direction == "SELL" and cached_ml_signal == "BUY"):
should_exit = True; urgency += 2
if rsi_val:
if (rsi_val > 75 and direction == "BUY") or (rsi_val < 25 and direction == "SELL"):
should_exit = True; urgency += 2
if (direction == "BUY" and trend == "BEARISH" and mom_dir == "BEARISH") or \
(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
# Weekend
if self._is_near_weekend_close(current_time):
if 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
tp_prob = min(40, max(0, progress * 0.4)) + ((momentum + 100) / 200) * 30 + 10 - min(10, bars_since_entry / 4 * 2)
if tp_prob < 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_rev = (direction == "BUY" and cached_ml_signal == "SELL") or \
(direction == "SELL" and cached_ml_signal == "BUY")
if is_rev:
return current_profit, current_pips, ExitReason.EARLY_EXIT, i, close
if current_profit < 0:
loss_pct = abs(current_profit) / self.max_loss_per_trade * 100
if momentum < -30 and loss_pct >= 30:
return current_profit, current_pips, ExitReason.EARLY_CUT, i, close
is_ml_rev = False
if (direction == "BUY" and cached_ml_signal == "SELL" and cached_ml_confidence >= self.trend_reversal_threshold) or \
(direction == "SELL" and cached_ml_signal == "BUY" and cached_ml_confidence >= self.trend_reversal_threshold):
is_ml_rev = True
reversal_warnings += 1
if is_ml_rev and current_profit < -8 and abs(current_profit) > self.max_loss_per_trade * 0.4:
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 not (htg <= 1 and htg > 0 and momentum > -40):
return current_profit, current_pips, ExitReason.MAX_LOSS, i, close
if len(profit_history) >= 10:
r = max(profit_history[-10:]) - min(profit_history[-10:])
if r < 3 and current_profit < -15:
stall_count += 1
if stall_count >= 5:
return current_profit, current_pips, ExitReason.STALL, i, close
pot_daily = daily_loss_so_far + abs(min(0, current_profit))
if pot_daily >= self.max_daily_loss_usd:
return current_profit, current_pips, ExitReason.DAILY_LIMIT, i, close
# C) Time exit
if bars_since_entry >= 16:
if current_profit < 5 and not profit_growing:
if 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:
rc = closes[i-5:i+1]
mom_val = rc[-1] - rc[0]
if direction == "BUY" and mom_val < -reversal_momentum_threshold and current_profit < -min_loss_for_reversal_exit:
return current_profit, current_pips, ExitReason.TREND_REVERSAL, i, close
elif direction == "SELL" and mom_val > reversal_momentum_threshold and 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)
fp = closes[final_idx]
pips = (fp - entry_price) / 0.1 if direction == "BUY" else (entry_price - fp) / 0.1
return pips * 10 * lot_size, pips, ExitReason.TIMEOUT, final_idx, fp
# ── 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
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()
closes = df["close"].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 + Quasimodo backtest...")
print(f" QM: lookback={self.qm_lookback}, max_age={self.qm_max_age}, tolerance={self.qm_zone_tolerance_pct*100:.1f}%")
print(f" QM RR: 1:{self.qm_rr_ratio}")
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]
current_price = closes[i]
# Daily reset
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
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 check
regime = "normal"
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
# Dynamic confidence AVOID
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
# ═══ SMC Signal ═══
smc_signal = None
try:
smc_signal = self.smc.generate_signal(df_slice)
except Exception:
pass
# ═══ QM Pattern Detection ═══
qm_patterns = self._detect_qm_patterns(df_slice)
if qm_patterns:
stats.qm_patterns_found += len(qm_patterns)
# ═══ DETERMINE ENTRY ═══
entry_source = None
direction = None
entry_price = None
stop_loss = None
take_profit = None
confidence = 0.5
signal_reason = ""
if smc_signal:
# Check if any QM pattern aligns with SMC direction
qm_match = self._find_matching_qm(qm_patterns, current_price, smc_signal.signal_type)
if qm_match:
# QM + SMC aligned: use QM's tighter SL
entry_source = "QM+SMC"
direction = smc_signal.signal_type
entry_price = smc_signal.entry_price
# Use QM head as SL (typically tighter than ATR-based)
qm_sl = qm_match["sl_price"]
smc_sl = smc_signal.stop_loss
# Choose tighter SL (closer to entry) but minimum $5 distance
if direction == "BUY":
qm_risk = entry_price - qm_sl
smc_risk = entry_price - smc_sl
if qm_risk > 5 and qm_risk < smc_risk:
stop_loss = qm_sl
else:
stop_loss = smc_sl
else:
qm_risk = qm_sl - entry_price
smc_risk = smc_sl - entry_price
if qm_risk > 5 and qm_risk < smc_risk:
stop_loss = qm_sl
else:
stop_loss = smc_sl
# Recalculate TP with QM RR ratio
risk = abs(entry_price - stop_loss)
if direction == "BUY":
take_profit = entry_price + risk * self.qm_rr_ratio
else:
take_profit = entry_price - risk * self.qm_rr_ratio
confidence = max(smc_signal.confidence, 0.65)
signal_reason = f"QM+SMC: {smc_signal.reason} | QML={qm_match['qml_level']:.2f}"
stats.qm_enhanced += 1
else:
# Standard SMC entry (no QM)
entry_source = "SMC"
direction = smc_signal.signal_type
entry_price = smc_signal.entry_price
stop_loss = smc_signal.stop_loss
take_profit = smc_signal.take_profit
confidence = smc_signal.confidence
signal_reason = smc_signal.reason
stats.standard_smc += 1
elif qm_patterns:
# No SMC signal, but check for QM-only entry
qm_match = self._find_matching_qm(qm_patterns, current_price)
if qm_match:
entry_source = "QM-only"
direction = qm_match["direction"]
entry_price = current_price
stop_loss = qm_match["sl_price"]
risk = abs(entry_price - stop_loss)
# Minimum risk $5
if risk < 5:
continue
if direction == "BUY":
take_profit = entry_price + risk * self.qm_rr_ratio
else:
take_profit = entry_price - risk * self.qm_rr_ratio
confidence = 0.58 # Moderate confidence for QM-only
signal_reason = f"QM-only: QML={qm_match['qml_level']:.2f}, head={qm_match['head_level']:.2f}"
stats.qm_only += 1
if entry_source is None:
continue
# SMC details (for tracking)
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 and atr_val > 0: atr_at_entry = atr_val
# ML confidence adjustment
ml_agrees = (direction == "BUY" and ml_signal == "BUY") or \
(direction == "SELL" and ml_signal == "SELL")
if ml_agrees and entry_source != "QM-only":
confidence = (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
risk = abs(entry_price - stop_loss)
rr = abs(take_profit - 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=direction,
entry_price=entry_price, take_profit=take_profit,
stop_loss=stop_loss, 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=direction, entry_price=entry_price,
exit_price=exit_price, stop_loss=stop_loss,
take_profit=take_profit, 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=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, entry_source=entry_source,
)
stats.trades.append(trade)
stats.total_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
dd_pct = (peak_capital - capital) / peak_capital * 100
dd_usd = peak_capital - capital
if dd_pct > stats.max_drawdown:
stats.max_drawdown = dd_pct
stats.max_drawdown_usd = dd_usd
stats.equity_curve.append(capital)
last_trade_idx = exit_idx
if stats.total_trades % 100 == 0:
print(f" {stats.total_trades} trades processed...")
# Final stats
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")
wp = stats.wins / stats.total_trades
lp = stats.losses / stats.total_trades
stats.expectancy = (wp * stats.avg_win) - (lp * stats.avg_loss)
rets = [t.profit_usd for t in stats.trades]
if len(rets) > 1:
stats.sharpe_ratio = (np.mean(rets) / np.std(rets)) * np.sqrt(252) if np.std(rets) > 0 else 0
return stats
# ─── Main ──────────────────────────────────────────────────────
def main():
print("=" * 70)
print("XAUBOT AI — #16 SMC + RTM Quasimodo Pattern")
print("Base: SMC-Only v4 | Added: QM pattern for entry + SL improvement")
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 data...")
df = mt5.get_market_data(symbol="XAUUSD", timeframe="M15", count=50000)
if len(df) == 0:
print("ERROR: No data")
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 = QuasimodoBacktest(
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,
qm_lookback=60,
qm_max_age=30,
qm_zone_tolerance_pct=0.004,
qm_rr_ratio=2.0,
)
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("#16 SMC + QUASIMODO — RESULTS")
print("=" * 70)
print(f"\n QM Pattern Stats:")
print(f" QM patterns found: {stats.qm_patterns_found}")
print(f" QM+SMC entries: {stats.qm_enhanced} (QM SL used)")
print(f" QM-only entries: {stats.qm_only}")
print(f" Standard SMC: {stats.standard_smc}")
# Per-source breakdown
for src in ["SMC", "QM+SMC", "QM-only"]:
src_trades = [t for t in stats.trades if t.entry_source == src]
if src_trades:
sw = sum(1 for t in src_trades if t.result == TradeResult.WIN)
sp = sum(t.profit_usd for t in src_trades)
swr = sw / len(src_trades) * 100
print(f" {src:10s}: {len(src_trades):3d} trades, {swr:.1f}% WR, ${sp:,.2f}")
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 (#1): ${net_pnl - 1449.86:+,.2f}")
print(f"\n Exit Reasons:")
exit_counts = {}
for t in stats.trades:
r = t.exit_reason.value
exit_counts[r] = exit_counts.get(r, 0) + 1
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
print(f" {reason:20s}: {count} ({pct:.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}")
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "16_quasimodo_results")
os.makedirs(output_dir, exist_ok=True)
# Log
log_path = os.path.join(output_dir, f"quasimodo_{timestamp}.log")
lines = []
lines.append("=" * 80)
lines.append("#16 SMC + RTM Quasimodo — Backtest Log")
lines.append("=" * 80)
lines.append(f"Period: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}")
lines.append(f"QM params: lookback=60, max_age=30, tolerance=0.4%, RR=1:2")
lines.append(f"")
lines.append(f"QM patterns: {stats.qm_patterns_found} | QM+SMC: {stats.qm_enhanced} | QM-only: {stats.qm_only} | Standard: {stats.standard_smc}")
lines.append(f"Trades: {stats.total_trades} | WR: {stats.win_rate:.1f}% | Net: ${net_pnl:,.2f}")
lines.append(f"PF: {stats.profit_factor:.2f} | DD: {stats.max_drawdown:.1f}% | Sharpe: {stats.sharpe_ratio:.2f}")
lines.append(f"vs Baseline: ${net_pnl - 1449.86:+,.2f}")
lines.append("")
lines.append("--- TRADE LOG ---")
lines.append(f"{'#':>4} {'Entry Time':>16} {'Dir':>4} {'Entry':>10} {'Exit':>10} {'P/L($)':>8} {'Result':>6} {'Exit Reason':>18} {'Source':>10}")
lines.append("-" * 110)
for idx, t in enumerate(stats.trades, 1):
lines.append(
f"{idx:4d} {t.entry_time.strftime('%Y-%m-%d %H:%M'):>16} {t.direction:>4} "
f"{t.entry_price:>10.2f} {t.exit_price:>10.2f} {t.profit_usd:>8.2f} "
f"{t.result.value:>6} {t.exit_reason.value:>18} {t.entry_source:>10}"
)
with open(log_path, "w", encoding="utf-8") as f:
f.write("\n".join(lines))
print(f" Log saved: {log_path}")
# XLSX
xlsx_path = os.path.join(output_dir, f"quasimodo_{timestamp}.xlsx")
wb = Workbook()
hf = Font(name="Calibri", bold=True, size=12, color="FFFFFF")
hfl = PatternFill(start_color="1F4E79", end_color="1F4E79", fill_type="solid")
wf = PatternFill(start_color="C6EFCE", end_color="C6EFCE", fill_type="solid")
lf = PatternFill(start_color="FFC7CE", end_color="FFC7CE", fill_type="solid")
bd = Border(left=Side(style="thin"), right=Side(style="thin"), top=Side(style="thin"), bottom=Side(style="thin"))
ws = wb.active
ws.title = "Summary"
ws["A1"] = "#16 SMC + Quasimodo Report"
ws["A1"].font = Font(bold=True, size=16, color="1F4E79")
r = 3
for lbl, val in [
("Trades", stats.total_trades), ("WR", f"{stats.win_rate:.1f}%"),
("Net PnL", f"${net_pnl:,.2f}"), ("PF", f"{stats.profit_factor:.2f}"),
("Max DD", f"{stats.max_drawdown:.1f}%"), ("Sharpe", f"{stats.sharpe_ratio:.2f}"),
("Avg Win", f"${stats.avg_win:,.2f}"), ("Avg Loss", f"${stats.avg_loss:,.2f}"),
("QM+SMC", stats.qm_enhanced), ("QM-only", stats.qm_only), ("Standard SMC", stats.standard_smc),
]:
ws.cell(row=r, column=1, value=lbl)
ws.cell(row=r, column=2, value=val)
r += 1
ws2 = wb.create_sheet("Trade Log")
headers = ["#", "Entry Time", "Dir", "Entry", "Exit", "SL", "TP", "Lot", "P/L", "Result", "Exit", "Source", "RR"]
for c, h in enumerate(headers, 1):
ws2.cell(row=1, column=c, value=h).font = hf
ws2.cell(row=1, column=c).fill = hfl
for ri, t in enumerate(stats.trades, 2):
vals = [ri-1, t.entry_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), t.result.value, t.exit_reason.value,
t.entry_source, round(t.rr_ratio, 2)]
for ci, v in enumerate(vals, 1):
cell = ws2.cell(row=ri, column=ci, value=v)
cell.border = bd
if ci == 9 and isinstance(v, (int, float)):
cell.fill = wf if v > 0 else (lf if v < 0 else PatternFill())
wb.save(xlsx_path)
print(f" Report saved: {xlsx_path}")
mt5.disconnect()
print("\n" + "=" * 70)
print(f"Output: {output_dir}")
print("=" * 70)
print("Backtest complete!")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink