gavindiaz/XauBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
XauBot/backtests/backtest_34_ml_v2d.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

1115 lines
46 KiB
Python
Raw Permalink Blame History

"""
Backtest #34 ML-V2D -- Time Filter + ML V2 Model D (76 features)
=================================================================
Clone of backtest_34_time_filter.py but using model_d.pkl from
backtests/36_ml_v2_results/ instead of the V1 xgboost_model.pkl.
Model D: 76 features (53 base + 8 H1 + 7 continuous SMC + 4 regime + 4 PA)
Test AUC: 0.7339 (+5.5% vs live model)
Usage:
python backtests/backtest_34_ml_v2d.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, Set
from dataclasses import dataclass, field
from enum import Enum
from collections import defaultdict
import sys
import os
from zoneinfo import ZoneInfo
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.config import get_config
from src.dynamic_confidence import DynamicConfidenceManager, create_dynamic_confidence, MarketQuality
from backtests.ml_v2.ml_v2_model import TradingModelV2
from backtests.ml_v2.ml_v2_feature_eng import MLV2FeatureEngineer
from loguru import logger
logger.remove()
logger.add(sys.stderr, level="WARNING")
WIB = ZoneInfo("Asia/Jakarta")
DAY_NAMES = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
# Path to model_d.pkl
MODEL_D_PATH = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"36_ml_v2_results", "model_d.pkl"
)
# --- 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"
h1_trend: str = "NEUTRAL"
wib_hour: int = 0
weekday: int = 0
@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
session_blocked: int = 0
h1_filtered: int = 0
time_filtered: int = 0
# --- Time Filter Backtest with ML V2 Model D ---
class TimeFilterBacktestV2D:
"""#31B base + time-of-hour/day-of-week filtering + ML V2 Model D."""
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,
max_concurrent_positions: int = 2,
min_profit_to_protect: float = 5.0,
max_drawdown_from_peak: float = 50.0,
trade_cooldown_bars: int = 10,
# #24B base
skip_tokyo_london: bool = True,
early_cut_momentum: float = -50.0,
early_cut_loss_pct: float = 30.0,
be_mult: float = 2.0,
trail_start_mult: float = 4.0,
trail_step_mult: float = 3.0,
# #28B: Smart breakeven
be_profit_lock_atr_mult: float = 0.5,
# === #34 TIME FILTER PARAMS ===
skip_wib_hours: Set[int] = None, # Set of WIB hours to skip
skip_weekdays: Set[int] = None, # Set of weekdays to skip (0=Mon, 4=Fri)
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.max_concurrent_positions = max_concurrent_positions
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.skip_tokyo_london = skip_tokyo_london
self.early_cut_momentum = early_cut_momentum
self.early_cut_loss_pct = early_cut_loss_pct
self.be_mult = be_mult
self.trail_start_mult = trail_start_mult
self.trail_step_mult = trail_step_mult
self.be_profit_lock_atr_mult = be_profit_lock_atr_mult
# #34 params
self.skip_wib_hours = skip_wib_hours or set()
self.skip_weekdays = skip_weekdays or set()
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()
# === ML V2 Model D (instead of V1) ===
self.ml_model = TradingModelV2(model_path=MODEL_D_PATH)
try:
self.ml_model.load()
print(f" ML V2 Model D loaded: {len(self.ml_model.feature_names)} features")
except Exception as e:
print(f" [WARN] ML V2 Model D load failed: {e}")
self.regime_detector = MarketRegimeDetector(model_path="models/hmm_regime.pkl")
try:
self.regime_detector.load()
except Exception:
pass
self._ticket_counter = 2340000
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:
if self.skip_tokyo_london:
return "Tokyo-London Overlap", False, 0.0
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 _get_wib_hour(self, dt):
if dt.tzinfo is None:
dt = dt.replace(tzinfo=ZoneInfo("UTC"))
return dt.astimezone(WIB).hour
def _get_wib_weekday(self, dt):
if dt.tzinfo is None:
dt = dt.replace(tzinfo=ZoneInfo("UTC"))
return dt.astimezone(WIB).weekday()
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)
def _calc_ema(self, data, period):
if len(data) < period:
return data[-1] if data else 0
multiplier = 2 / (period + 1)
ema = np.mean(data[:period])
for val in data[period:]:
ema = (val - ema) * multiplier + ema
return ema
def _get_h1_trend(self, df_h1_slice):
if df_h1_slice is None or len(df_h1_slice) < 20:
return "NEUTRAL"
closes = df_h1_slice["close"].to_list()
ema20 = self._calc_ema(closes, 20)
current_price = closes[-1]
if current_price > ema20 * 1.001:
return "BULLISH"
elif current_price < ema20 * 0.999:
return "BEARISH"
return "NEUTRAL"
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]
adaptive_breakeven_pips = atr * self.be_mult
adaptive_trail_start_pips = atr * self.trail_start_mult
adaptive_trail_step_pips = atr * self.trail_step_mult
reversal_momentum_threshold = atr * self.trend_reversal_mult
min_loss_for_reversal_exit = atr * 0.8
if self.be_profit_lock_atr_mult > 0:
be_lock_distance = atr * self.be_profit_lock_atr_mult
else:
be_lock_distance = 2.0
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.0 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
# A.0b Trailing 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 >= adaptive_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 >= adaptive_trail_start_pips else ExitReason.BREAKEVEN_EXIT
return pips * pip_value * lot_size, pips, reason, i, current_sl
# A.1 Breakeven (#28B: Smart)
if pip_profit_from_entry >= adaptive_breakeven_pips and not breakeven_moved:
if direction == "BUY":
current_sl = entry_price + be_lock_distance
else:
current_sl = entry_price - be_lock_distance
breakeven_moved = True
# A.2 Trailing SL
if pip_profit_from_entry >= adaptive_trail_start_pips:
trail_distance = adaptive_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
# A.3 Peak protect
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
# A.4 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 = "BULLISH" if ma_fast > ma_slow * 1.001 else ("BEARISH" if ma_fast < ma_slow * 0.999 else "NEUTRAL")
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
# A.5 Weekend close
if self._is_near_weekend_close(current_time):
if current_profit > 0 or current_profit > -10:
return current_profit, current_pips, ExitReason.WEEKEND_CLOSE, i, close
# B.1 Smart TP
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
# B.2 Smart Early Exit
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
# B.3 Early cut
if current_profit < 0:
loss_percent_of_max = abs(current_profit) / self.max_loss_per_trade * 100
if momentum < self.early_cut_momentum and loss_percent_of_max >= self.early_cut_loss_pct:
return current_profit, current_pips, ExitReason.EARLY_CUT, i, close
# B.4 Trend Reversal
is_ml_reversal = False
if (direction == "BUY" and cached_ml_signal == "SELL" and cached_ml_confidence >= 0.75) or \
(direction == "SELL" and cached_ml_signal == "BUY" and cached_ml_confidence >= 0.75):
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
# B.5 Max loss
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
# B.6 Stall
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
# B.7 Daily loss limit
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
if bars_since_entry >= 16 and current_profit < 5 and not profit_growing:
if current_profit >= 0 or current_profit > -15:
return current_profit, current_pips, ExitReason.TIMEOUT, i, close
if bars_since_entry >= 24 and (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
# C.2 ATR trend reversal
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) or \
(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) if direction == "BUY" else (entry_price - final_price)) / 0.1
return pips * pip_value * lot_size, pips, ExitReason.TIMEOUT, final_idx, final_price
def run(self, df_m15, df_h1, 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_m15.columns]
missing = [f for f in self.ml_model.feature_names if f not in df_m15.columns]
if missing:
print(f" [WARN] Missing {len(missing)} features: {missing[:5]}...")
times_m15 = df_m15["time"].to_list()
times_h1 = df_h1["time"].to_list() if df_h1 is not None else []
start_idx = next((i for i, t in enumerate(times_m15) if t >= start_date), 100) if start_date else 100
end_idx = next((i for i, t in enumerate(times_m15) if t > end_date), len(df_m15) - 100) if end_date else len(df_m15) - 100
last_trade_idx = -self.trade_cooldown_bars * 2
skip_hours_str = ",".join(str(h) for h in sorted(self.skip_wib_hours)) if self.skip_wib_hours else "none"
skip_days_str = ",".join(DAY_NAMES[d] for d in sorted(self.skip_weekdays)) if self.skip_weekdays else "none"
print(f" #34 ML-V2D skip hours(WIB): [{skip_hours_str}], skip days: [{skip_days_str}]")
print(f" ML features available: {len(feature_cols)}/{len(self.ml_model.feature_names) if self.ml_model.fitted else 0}")
print(f" Date range: {times_m15[start_idx]} to {times_m15[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_m15[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:
if session_name == "Tokyo-London Overlap":
stats.session_blocked += 1
continue
if hasattr(current_time, 'weekday') and current_time.weekday() >= 5:
continue
# #34: Time-of-hour filter
wib_hour = self._get_wib_hour(current_time)
if wib_hour in self.skip_wib_hours:
stats.time_filtered += 1
continue
# #34: Day-of-week filter
wib_weekday = self._get_wib_weekday(current_time)
if wib_weekday in self.skip_weekdays:
stats.time_filtered += 1
continue
df_slice = df_m15.head(i + 1)
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
try:
ml_signal = ""
ml_confidence = 0.5
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
# #31B: H1 Price vs EMA20 filter
h1_trend = "NEUTRAL"
if df_h1 is not None and len(times_h1) > 0:
h1_idx = 0
for j, t in enumerate(times_h1):
if t <= current_time:
h1_idx = j
else:
break
if h1_idx > 20:
df_h1_slice = df_h1.head(h1_idx + 1)
h1_trend = self._get_h1_trend(df_h1_slice)
if smc_signal.signal_type == "BUY" and h1_trend != "BULLISH":
stats.h1_filtered += 1
continue
if smc_signal.signal_type == "SELL" and h1_trend != "BEARISH":
stats.h1_filtered += 1
continue
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_m15, 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_m15[exit_idx] if exit_idx < len(times_m15) else times_m15[-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,
h1_trend=h1_trend,
wib_hour=wib_hour,
weekday=wib_weekday,
)
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
def analyze_hourly_daily(stats):
"""Analyze trade performance by WIB hour and weekday."""
hour_stats = defaultdict(lambda: {"trades": 0, "wins": 0, "pnl": 0.0})
day_stats = defaultdict(lambda: {"trades": 0, "wins": 0, "pnl": 0.0})
for t in stats.trades:
h = t.wib_hour
hour_stats[h]["trades"] += 1
hour_stats[h]["pnl"] += t.profit_usd
if t.result == TradeResult.WIN:
hour_stats[h]["wins"] += 1
d = t.weekday
day_stats[d]["trades"] += 1
day_stats[d]["pnl"] += t.profit_usd
if t.result == TradeResult.WIN:
day_stats[d]["wins"] += 1
return hour_stats, day_stats
# --- Main ---
def main():
print("=" * 70)
print("XAUBOT AI -- #34 ML-V2D: Time Filter + ML V2 Model D (76 features)")
print("Base: #34 Time Filter | ML: model_d.pkl (Test AUC 0.7339)")
print("=" * 70)
config = get_config()
mt5_conn = MT5Connector(
login=config.mt5_login, password=config.mt5_password,
server=config.mt5_server, path=config.mt5_path,
)
mt5_conn.connect()
print(f"\nConnected to MT5")
print("Fetching XAUUSD M15 historical data...")
df_m15 = mt5_conn.get_market_data(symbol="XAUUSD", timeframe="M15", count=50000)
print(f" M15: {len(df_m15)} bars")
print("Fetching XAUUSD H1 historical data...")
df_h1 = mt5_conn.get_market_data(symbol="XAUUSD", timeframe="H1", count=15000)
print(f" H1: {len(df_h1)} bars")
times = df_m15["time"].to_list()
print(f" M15 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')}")
# === Calculate base indicators ===
print("\nCalculating M15 indicators...")
features = FeatureEngineer()
smc = SMCAnalyzer(swing_length=config.smc.swing_length, ob_lookback=config.smc.ob_lookback)
df_m15 = features.calculate_all(df_m15, include_ml_features=True)
df_m15 = smc.calculate_all(df_m15)
regime_detector = MarketRegimeDetector(model_path="models/hmm_regime.pkl")
try:
regime_detector.load()
df_m15 = regime_detector.predict(df_m15)
print(" HMM regime loaded")
except Exception:
print(" [WARN] HMM not available")
df_m15 = df_m15.with_columns([
pl.lit(1).alias("regime"),
pl.lit("medium_volatility").alias("regime_name"),
])
print("Calculating H1 indicators...")
df_h1 = features.calculate_all(df_h1, include_ml_features=False)
# H1 also needs SMC for V2 H1 features (ob_top, fvg_top, bos, etc.)
df_h1 = smc.calculate_all(df_h1)
print(" H1 base + SMC indicators calculated")
# === Add V2 features (23 new features for model_d) ===
print("\nAdding ML V2 features (23 new features)...")
fe_v2 = MLV2FeatureEngineer()
df_m15 = fe_v2.add_all_v2_features(df_m15, df_h1)
v2_cols = fe_v2.get_v2_feature_columns()
available_v2 = [c for c in v2_cols if c in df_m15.columns]
print(f" V2 features available: {len(available_v2)}/{len(v2_cols)}")
print(f" Total M15 columns: {len(df_m15.columns)}")
baseline_34_pnl = 2806.56 # #31B baseline for comparison
# ===============================================================
# PHASE 1: Run baseline to analyze per-hour and per-day performance
# ===============================================================
print(f"\n{'=' * 60}")
print(" PHASE 1: Baseline analysis (no time filter, ML V2 Model D)")
bt_base = TimeFilterBacktestV2D()
stats_base = bt_base.run(df_m15=df_m15, df_h1=df_h1, start_date=start_date, end_date=end_date)
net_base = stats_base.total_profit - stats_base.total_loss
hour_stats, day_stats = analyze_hourly_daily(stats_base)
print(f"\n Baseline (V2D): {stats_base.total_trades} trades, {stats_base.win_rate:.1f}% WR, ${net_base:,.2f}")
# Print hourly analysis
print(f"\n === HOURLY ANALYSIS (WIB) ===")
print(f" {'Hour':>4} {'Trades':>7} {'Wins':>5} {'WR':>7} {'PnL':>10} {'Avg':>8}")
print(f" {'-' * 45}")
hour_ranking = []
for h in sorted(hour_stats.keys()):
s = hour_stats[h]
wr = s["wins"] / s["trades"] * 100 if s["trades"] > 0 else 0
avg = s["pnl"] / s["trades"] if s["trades"] > 0 else 0
marker = " <-- WORST" if s["trades"] >= 5 and (wr < 75 or s["pnl"] < 0) else ""
print(f" {h:>4} {s['trades']:>7} {s['wins']:>5} {wr:>6.1f}% ${s['pnl']:>9,.2f} ${avg:>7,.2f}{marker}")
if s["trades"] >= 5:
hour_ranking.append((h, wr, s["pnl"], s["trades"]))
# Sort by PnL (worst first)
hour_ranking.sort(key=lambda x: x[2])
worst_2_hours = set(h[0] for h in hour_ranking[:2])
worst_3_hours = set(h[0] for h in hour_ranking[:3])
print(f"\n Worst 2 hours (by PnL): {sorted(worst_2_hours)}")
print(f" Worst 3 hours (by PnL): {sorted(worst_3_hours)}")
# Print daily analysis
print(f"\n === DAY-OF-WEEK ANALYSIS ===")
print(f" {'Day':>4} {'Trades':>7} {'Wins':>5} {'WR':>7} {'PnL':>10} {'Avg':>8}")
print(f" {'-' * 45}")
day_ranking = []
for d in sorted(day_stats.keys()):
s = day_stats[d]
wr = s["wins"] / s["trades"] * 100 if s["trades"] > 0 else 0
avg = s["pnl"] / s["trades"] if s["trades"] > 0 else 0
marker = " <-- WORST" if s["trades"] >= 10 and (wr < 78 or s["pnl"] < 0) else ""
print(f" {DAY_NAMES[d]:>4} {s['trades']:>7} {s['wins']:>5} {wr:>6.1f}% ${s['pnl']:>9,.2f} ${avg:>7,.2f}{marker}")
if s["trades"] >= 10:
day_ranking.append((d, wr, s["pnl"], s["trades"]))
day_ranking.sort(key=lambda x: x[2])
worst_day = {day_ranking[0][0]} if day_ranking else set()
print(f"\n Worst day: {[DAY_NAMES[d] for d in sorted(worst_day)]}")
# ===============================================================
# PHASE 2: Run filtered configs based on Phase 1 analysis
# ===============================================================
print(f"\n{'=' * 60}")
print(" PHASE 2: Testing filtered configurations (ML V2 Model D)")
configs = [
("A: Skip worst 2 hours", {
"skip_wib_hours": worst_2_hours,
}),
("B: Skip worst 3 hours", {
"skip_wib_hours": worst_3_hours,
}),
("C: Skip worst day", {
"skip_weekdays": worst_day,
}),
("D: Worst 2h + worst day", {
"skip_wib_hours": worst_2_hours,
"skip_weekdays": worst_day,
}),
("E: Worst 3h + worst day", {
"skip_wib_hours": worst_3_hours,
"skip_weekdays": worst_day,
}),
]
all_results = []
for cfg_name, cfg_params in configs:
print(f"\n{'=' * 60}")
print(f" Config: {cfg_name}")
bt = TimeFilterBacktestV2D(**cfg_params)
stats = bt.run(df_m15=df_m15, df_h1=df_h1, start_date=start_date, end_date=end_date, initial_capital=5000.0)
net_pnl = stats.total_profit - stats.total_loss
diff = net_pnl - baseline_34_pnl
buy_trades = [t for t in stats.trades if t.direction == "BUY"]
sell_trades = [t for t in stats.trades if t.direction == "SELL"]
buy_wins = sum(1 for t in buy_trades if t.result == TradeResult.WIN)
sell_wins = sum(1 for t in sell_trades if t.result == TradeResult.WIN)
buy_wr = buy_wins / len(buy_trades) * 100 if buy_trades else 0
sell_wr = sell_wins / len(sell_trades) * 100 if sell_trades else 0
buy_pnl = sum(t.profit_usd for t in buy_trades)
sell_pnl = sum(t.profit_usd for t in sell_trades)
print(f"\n [{cfg_name}] Results:")
print(f" Trades: {stats.total_trades} | WR: {stats.win_rate:.1f}%")
print(f" Net PnL: ${net_pnl:,.2f} | PF: {stats.profit_factor:.2f}")
print(f" Max DD: {stats.max_drawdown:.1f}% | Sharpe: {stats.sharpe_ratio:.2f}")
print(f" BUY: {len(buy_trades)}, {buy_wr:.1f}% WR, ${buy_pnl:,.2f}")
print(f" SELL: {len(sell_trades)}, {sell_wr:.1f}% WR, ${sell_pnl:,.2f}")
print(f" Time-filtered: {stats.time_filtered} signals blocked")
print(f" vs #31B: ${diff:+,.2f}")
all_results.append((cfg_name, stats, net_pnl, diff))
# === FINAL SUMMARY ===
print(f"\n{'=' * 70}")
print("#34 ML-V2D: TIME FILTER + ML V2 MODEL D -- ALL CONFIGURATIONS")
print("=" * 70)
print(f"\n {'Config':<25} {'Trades':>6} {'WR':>6} {'Net PnL':>10} {'DD':>6} {'Sharpe':>7} {'PF':>5} {'Blocked':>8} {'vs #31B':>10}")
print(f" {'-' * 90}")
print(f" {'#31B (V1 model)':<25} {'625':>6} {'81.8%':>6} {'$2,807':>10} {'2.5%':>6} {'3.97':>7} {'2.19':>5} {'--':>8} {'--':>10}")
print(f" {'V2D Baseline (no filt)':<25} {stats_base.total_trades:>6} {stats_base.win_rate:>5.1f}% ${net_base:>9,.2f} {stats_base.max_drawdown:>5.1f}% {stats_base.sharpe_ratio:>7.2f} {stats_base.profit_factor:>5.2f} {'--':>8} ${net_base - baseline_34_pnl:>+9,.2f}")
for cfg_name, stats, net_pnl, diff in all_results:
blocked = stats.time_filtered
print(f" {cfg_name:<25} {stats.total_trades:>6} {stats.win_rate:>5.1f}% ${net_pnl:>9,.2f} {stats.max_drawdown:>5.1f}% {stats.sharpe_ratio:>7.2f} {stats.profit_factor:>5.2f} {blocked:>8} ${diff:>+9,.2f}")
best_pnl = -999999
best_name = ""
best_stats = None
for entry in all_results:
if entry[2] > best_pnl:
best_pnl = entry[2]
best_name = entry[0]
best_stats = entry[1]
# Also compare baseline (no filter)
if net_base > best_pnl:
best_pnl = net_base
best_name = "Baseline (no filter)"
best_stats = stats_base
print(f"\n Best config: {best_name}")
# Exit reasons
print(f"\n Exit Reasons (best config):")
exit_counts = {}
for t in best_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 / best_stats.total_trades * 100 if best_stats.total_trades > 0 else 0
print(f" {reason:20s}: {count} ({pct:.1f}%)")
# Save
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "34_ml_v2d_results")
os.makedirs(output_dir, exist_ok=True)
log_path = os.path.join(output_dir, f"ml_v2d_time_filter_{timestamp}.log")
with open(log_path, "w") as f:
f.write(f"#34 ML-V2D: Time Filter + ML V2 Model D Results\n")
f.write(f"Generated: {datetime.now()}\n")
f.write(f"ML Model: model_d.pkl (76 features, Test AUC 0.7339)\n")
f.write(f"Base comparison: #31B (625 trades, 81.8% WR, $2,807)\n\n")
f.write(f"=== BASELINE (V2D, no time filter) ===\n")
f.write(f" Trades: {stats_base.total_trades}, WR: {stats_base.win_rate:.1f}%, "
f"PnL: ${net_base:,.2f}, DD: {stats_base.max_drawdown:.1f}%, "
f"Sharpe: {stats_base.sharpe_ratio:.2f}, PF: {stats_base.profit_factor:.2f}\n\n")
f.write(f"=== HOURLY ANALYSIS (WIB) ===\n")
for h in sorted(hour_stats.keys()):
s = hour_stats[h]
wr = s["wins"] / s["trades"] * 100 if s["trades"] > 0 else 0
avg = s["pnl"] / s["trades"] if s["trades"] > 0 else 0
f.write(f" {h:>2}:00 WIB {s['trades']:>4} trades {wr:>5.1f}% WR ${s['pnl']:>8,.2f} avg ${avg:>6,.2f}\n")
f.write(f"\nWorst 2 hours: {sorted(worst_2_hours)}\n")
f.write(f"Worst 3 hours: {sorted(worst_3_hours)}\n")
f.write(f"\n=== DAY-OF-WEEK ANALYSIS ===\n")
for d in sorted(day_stats.keys()):
s = day_stats[d]
wr = s["wins"] / s["trades"] * 100 if s["trades"] > 0 else 0
avg = s["pnl"] / s["trades"] if s["trades"] > 0 else 0
f.write(f" {DAY_NAMES[d]:>3} {s['trades']:>4} trades {wr:>5.1f}% WR ${s['pnl']:>8,.2f} avg ${avg:>6,.2f}\n")
f.write(f"\nWorst day: {[DAY_NAMES[d] for d in sorted(worst_day)]}\n")
f.write(f"\n=== FILTERED RESULTS ===\n")
for cfg_name, stats, net_pnl, diff in all_results:
f.write(f" {cfg_name}: {stats.total_trades} trades, {stats.win_rate:.1f}% WR, "
f"${net_pnl:,.2f}, DD: {stats.max_drawdown:.1f}%, "
f"Sharpe: {stats.sharpe_ratio:.2f}, PF: {stats.profit_factor:.2f}, "
f"Blocked: {stats.time_filtered}, vs #31B: ${diff:+,.2f}\n")
f.write(f"\nBest: {best_name}\n")
print(f" Log saved: {log_path}")
try:
from backtests.backtest_01_smc_only import generate_xlsx_report as gen_xlsx
xlsx_path = os.path.join(output_dir, f"ml_v2d_time_filter_{timestamp}.xlsx")
gen_xlsx(best_stats, xlsx_path, start_date, end_date)
print(f"\n Report saved: {xlsx_path}")
except Exception as e:
print(f" [WARN] XLSX: {e}")
mt5_conn.disconnect()
print(f"\n{'=' * 70}")
print(f"Output: {output_dir}")
print(f" Log: {os.path.basename(log_path)}")
print("=" * 70)
print("Backtest complete!")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink