gavindiaz/XauBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
XauBot/backtests/backtest_39_h1_hmm.py
T
GifariKemal 4626f2a295 chore: add backtest #39 and HMM investigation artifacts 
Added research artifacts from HMM investigation:
- backtest_39_h1_hmm.py — H1 vs M15 HMM comparison attempt
- 39_h1_hmm_results/ — Partial backtest results
- analyze_*.py — ML model and H1 feature analysis scripts
- *_output.txt — Analysis outputs showing HMM degeneracy

Updated:
- data/risk_state.txt — Latest risk state (daily_loss: 18.77, daily_profit: 22.49)

Note: Backtest #39 had import compatibility issues but led to critical
discovery of alternating HMM pattern bug (fixed in c02c2e9).

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-09 10:51:49 +07:00

488 lines
18 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""
Backtest #39: H1 HMM Regime Detector
Test moving HMM from M15 to H1 timeframe for more stable regime detection.
Hypothesis:
- H1 HMM will have 4-8x longer regime duration
- Fewer regime transitions = more stable risk management
- Better regime classification due to less noise
Expected Impact: +15-20% Sharpe improvement
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
import polars as pl
import numpy as np
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Optional
from loguru import logger
from src.mt5_connector import MT5Connector
from src.config import TradingConfig, get_config
from src.feature_eng import FeatureEngineer
from src.smc_polars import SMCAnalyzer
from src.regime_detector import MarketRegimeDetector
from backtests.ml_v2.ml_v2_model import TradingModelV2
from src.smart_risk_manager import SmartRiskManager
from src.session_filter import SessionFilter
from src.dynamic_confidence import DynamicConfidenceManager
# Import V2 features
from backtests.ml_v2.ml_v2_feature_eng import MLV2FeatureEngineer
@dataclass
class BacktestConfig:
"""Backtest configuration"""
symbol: str = "XAUUSD"
m15_bars: int = 5000
h1_bars: int = 1500
initial_balance: float = 500.0
results_dir: str = "backtests/39_h1_hmm_results"
class H1HMMBacktest:
"""Backtest with H1-based HMM regime detector"""
def __init__(self, config: BacktestConfig, use_h1_hmm: bool = True):
self.config = config
self.use_h1_hmm = use_h1_hmm
self.balance = config.initial_balance
self.equity = config.initial_balance
self.trades = []
self.equity_curve = []
# Trading config
self.trading_config = TradingConfig()
# Components
self.mt5 = None
self.features = FeatureEngineer()
self.smc = SMCAnalyzer()
self.ml_model = TradingModelV2()
self.fe_v2 = MLV2FeatureEngineer()
# Skip risk manager and session filter for backtest simplicity
self.risk_manager = None
self.session_filter = None
self.dynamic_conf = None
# Regime detectors - pass individual params (API changed)
self.regime_m15 = MarketRegimeDetector(
n_regimes=3,
lookback_periods=500,
retrain_frequency=20
)
self.regime_h1 = MarketRegimeDetector(
n_regimes=3,
lookback_periods=500,
retrain_frequency=20
)
# Stats
self.regime_changes_m15 = []
self.regime_changes_h1 = []
def connect_mt5(self):
"""Connect to MT5"""
logger.info("Connecting to MT5...")
config = get_config()
self.mt5 = MT5Connector(
login=config.mt5_login,
password=config.mt5_password,
server=config.mt5_server,
path=config.mt5_path
)
if not self.mt5.connect():
raise RuntimeError("Failed to connect to MT5")
logger.info("✓ MT5 connected")
def fetch_data(self):
"""Fetch M15 and H1 data"""
logger.info(f"Fetching {self.config.m15_bars} M15 bars...")
df_m15 = self.mt5.get_market_data(
self.config.symbol, "M15", self.config.m15_bars
)
logger.info(f"Fetching {self.config.h1_bars} H1 bars...")
df_h1 = self.mt5.get_market_data(
self.config.symbol, "H1", self.config.h1_bars
)
logger.info(f"✓ M15: {len(df_m15)} bars, H1: {len(df_h1)} bars")
return df_m15, df_h1
def prepare_data(self, df_m15: pl.DataFrame, df_h1: pl.DataFrame):
"""Calculate all features"""
logger.info("Calculating M15 features...")
df_m15 = self.features.calculate_all(df_m15, include_ml_features=True)
df_m15 = self.smc.calculate_all(df_m15)
logger.info("Calculating H1 features...")
df_h1 = self.features.calculate_all(df_h1, include_ml_features=True)
df_h1 = self.smc.calculate_all(df_h1)
logger.info("Adding V2 features...")
df_m15 = self.fe_v2.add_all_v2_features(df_m15, df_h1)
return df_m15, df_h1
def fit_regimes(self, df_m15: pl.DataFrame, df_h1: pl.DataFrame):
"""Train both M15 and H1 HMM models"""
logger.info("Training M15 HMM (baseline)...")
self.regime_m15.fit(df_m15.slice(0, 500))
logger.info("Training H1 HMM (test)...")
self.regime_h1.fit(df_h1.slice(0, 500))
logger.info("✓ Both HMM models fited")
def detect_regime(self, df_m15: pl.DataFrame, df_h1: pl.DataFrame, m15_idx: int):
"""Detect regime using M15 or H1 and return updated df with regime columns"""
if self.use_h1_hmm:
# Use H1 regime (map M15 index to H1)
h1_idx = m15_idx // 4 # 4 M15 bars = 1 H1 bar
if h1_idx >= len(df_h1):
h1_idx = len(df_h1) - 1
df_h1_slice = df_h1.slice(max(0, h1_idx - 100), h1_idx + 1)
df_h1_pred = self.regime_h1.predict(df_h1_slice)
regime_state = self.regime_h1.get_current_state(df_h1_pred)
# Track H1 regime changes
if hasattr(self, '_last_h1_regime') and self._last_h1_regime != regime_state.regime.value:
self.regime_changes_h1.append({
'm15_idx': m15_idx,
'old': self._last_h1_regime,
'new': regime_state.regime.value
})
self._last_h1_regime = regime_state.regime.value
else:
# Use M15 regime (baseline)
df_m15_slice = df_m15.slice(max(0, m15_idx - 100), m15_idx + 1)
df_m15_pred = self.regime_m15.predict(df_m15_slice)
regime_state = self.regime_m15.get_current_state(df_m15_pred)
# Track M15 regime changes
if hasattr(self, '_last_m15_regime') and self._last_m15_regime != regime_state.regime.value:
self.regime_changes_m15.append({
'm15_idx': m15_idx,
'old': self._last_m15_regime,
'new': regime_state.regime.value
})
self._last_m15_regime = regime_state.regime.value
return regime_state, df_m15_pred if not self.use_h1_hmm else df_h1_pred
def run_backtest(self, df_m15: pl.DataFrame, df_h1: pl.DataFrame):
"""Run backtest loop"""
logger.info(f"Running backtest ({'H1 HMM' if self.use_h1_hmm else 'M15 HMM'})...")
# Load ML model
self.ml_model.load("models/xgboost_model_v2d.pkl")
# Start from bar 600 (after fiting window)
for idx in range(600, len(df_m15)):
# Get current bar
row = df_m15.row(idx, named=True)
timestamp = row['time']
price = row['close']
# Detect regime
regime_state, df_regime_pred = self.detect_regime(df_m15, df_h1, idx)
# Add regime columns to current df slice for ML prediction
if 'regime' not in df_m15.columns:
# Initialize regime columns in df_m15
df_m15 = df_m15.with_columns([
pl.lit(0).alias('regime'),
pl.lit(0.0).alias('regime_confidence')
])
# Check if regime blocks trading
if regime_state.recommendation == "SLEEP":
continue
# Get ML prediction (use model's stored feature names)
df_slice = df_m15.slice(0, idx + 1)
if hasattr(self.ml_model, 'feature_names') and self.ml_model.feature_names:
feature_cols = self.ml_model.feature_names
else:
# Fallback: filter out OHLC and intermediate columns
exclude_cols = {'time', 'open', 'high', 'low', 'close', 'spread', 'real_volume', 'volume',
'swing_high_level', 'swing_low_level', 'last_swing_high', 'last_swing_low',
'fvg_top', 'fvg_bottom', 'fvg_mid', 'ob_top', 'ob_bottom'}
feature_cols = [c for c in df_slice.columns if c not in exclude_cols]
ml_pred = self.ml_model.predict(df_slice, feature_cols)
if ml_pred.signal == "HOLD":
continue
# Skip session check for simplicity
# session_ok, _, session_mult = self.session_filter.can_trade(timestamp)
# if not session_ok:
# continue
# Entry filters (simplified)
if ml_pred.confidence < 0.50:
continue
# SMC signal
smc_signal = row.get('ob', 0)
if smc_signal == 0:
continue
if (ml_pred.signal == "BUY" and smc_signal < 0) or \
(ml_pred.signal == "SELL" and smc_signal > 0):
continue
# Execute trade (simplified)
direction = ml_pred.signal
entry_price = price
# Calculate SL/TP (simplified)
atr = row.get('atr', 12.0)
if direction == "BUY":
sl = entry_price - (2.0 * atr)
tp = entry_price + (3.0 * atr)
else:
sl = entry_price + (2.0 * atr)
tp = entry_price - (3.0 * atr)
# Simulate trade (find exit)
exit_price = None
exit_reason = None
exit_idx = None
for future_idx in range(idx + 1, min(idx + 100, len(df_m15))):
future_row = df_m15.row(future_idx, named=True)
future_high = future_row['high']
future_low = future_row['low']
if direction == "BUY":
if future_low <= sl:
exit_price = sl
exit_reason = "SL"
exit_idx = future_idx
break
elif future_high >= tp:
exit_price = tp
exit_reason = "TP"
exit_idx = future_idx
break
else: # SELL
if future_high >= sl:
exit_price = sl
exit_reason = "SL"
exit_idx = future_idx
break
elif future_low <= tp:
exit_price = tp
exit_reason = "TP"
exit_idx = future_idx
break
if exit_price is None:
# No exit found, close at last bar
exit_price = df_m15.row(min(idx + 99, len(df_m15) - 1), named=True)['close']
exit_reason = "EOD"
exit_idx = min(idx + 99, len(df_m15) - 1)
# Calculate P&L
if direction == "BUY":
profit = exit_price - entry_price
else:
profit = entry_price - exit_price
profit_usd = profit * 0.01 # 0.01 lot
# Update balance
self.balance += profit_usd
self.equity = self.balance
# Log trade
self.trades.append({
'entry_time': timestamp,
'exit_time': df_m15.row(exit_idx, named=True)['time'],
'direction': direction,
'entry_price': entry_price,
'exit_price': exit_price,
'profit_usd': profit_usd,
'exit_reason': exit_reason,
'regime': regime_state.regime.value,
'ml_conf': ml_pred.confidence,
})
# Record equity
self.equity_curve.append({
'time': df_m15.row(exit_idx, named=True)['time'],
'equity': self.equity
})
logger.info(f"✓ Backtest complete: {len(self.trades)} trades")
def calculate_metrics(self):
"""Calculate performance metrics"""
if not self.trades:
return {}
df_trades = pl.DataFrame(self.trades)
total_trades = len(self.trades)
wins = df_trades.filter(pl.col('profit_usd') > 0).shape[0]
losses = df_trades.filter(pl.col('profit_usd') < 0).shape[0]
win_rate = wins / total_trades * 100 if total_trades > 0 else 0
net_profit = df_trades['profit_usd'].sum()
gross_profit = df_trades.filter(pl.col('profit_usd') > 0)['profit_usd'].sum()
gross_loss = abs(df_trades.filter(pl.col('profit_usd') < 0)['profit_usd'].sum())
profit_factor = gross_profit / gross_loss if gross_loss > 0 else 0
# Drawdown
equity_curve = [self.config.initial_balance] + [e['equity'] for e in self.equity_curve]
running_max = np.maximum.accumulate(equity_curve)
drawdown = running_max - equity_curve
max_dd = np.max(drawdown)
max_dd_pct = max_dd / self.config.initial_balance * 100
# Sharpe ratio (annualized)
returns = df_trades['profit_usd'].to_numpy()
if len(returns) > 1 and np.std(returns) > 0:
avg_return = np.mean(returns)
std_return = np.std(returns)
# Assume ~5 trades/day, 252 trading days/year
sharpe = (avg_return / std_return) * np.sqrt(5 * 252)
else:
sharpe = 0
# Regime changes
regime_changes_m15 = len(self.regime_changes_m15)
regime_changes_h1 = len(self.regime_changes_h1)
return {
'total_trades': total_trades,
'wins': wins,
'losses': losses,
'win_rate': win_rate,
'net_profit': net_profit,
'gross_profit': gross_profit,
'gross_loss': gross_loss,
'profit_factor': profit_factor,
'max_dd': max_dd,
'max_dd_pct': max_dd_pct,
'sharpe': sharpe,
'final_balance': self.balance,
'regime_changes_m15': regime_changes_m15,
'regime_changes_h1': regime_changes_h1,
}
def save_results(self, variant_name: str, metrics: dict):
"""Save results to file"""
Path(self.config.results_dir).mkdir(parents=True, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
log_file = Path(self.config.results_dir) / f"h1_hmm_{timestamp}.log"
with open(log_file, 'w') as f:
f.write(f"#39 H1 HMM Results\n")
f.write(f"Generated: {datetime.now()}\n")
f.write(f"Variant: {variant_name}\n\n")
f.write("--- PERFORMANCE SUMMARY ---\n")
f.write(f" Total Trades: {metrics['total_trades']}\n")
f.write(f" Win Rate: {metrics['win_rate']:.1f}%\n")
f.write(f" Net PnL: ${metrics['net_profit']:.2f}\n")
f.write(f" Profit Factor: {metrics['profit_factor']:.2f}\n")
f.write(f" Max Drawdown: ${metrics['max_dd']:.2f} ({metrics['max_dd_pct']:.1f}%)\n")
f.write(f" Sharpe Ratio: {metrics['sharpe']:.2f}\n")
f.write(f" Final Balance: ${metrics['final_balance']:.2f}\n\n")
f.write("--- REGIME STABILITY ---\n")
f.write(f" M15 Regime Changes: {metrics['regime_changes_m15']}\n")
f.write(f" H1 Regime Changes: {metrics['regime_changes_h1']}\n")
if metrics['regime_changes_h1'] > 0:
improvement = (metrics['regime_changes_m15'] - metrics['regime_changes_h1']) / metrics['regime_changes_m15'] * 100
f.write(f" Improvement: {improvement:.1f}% fewer changes\n\n")
f.write("--- TRADE LOG ---\n")
f.write(f"{'#':>4} {'Entry Time':>19} {'Dir':>4} {'Entry':>8} {'Exit':>8} {'P/L':>7} {'Result':>6} {'Exit_Reason':>12} {'Regime':>15} {'ML_Conf':>7}\n")
f.write("-" * 120 + "\n")
for i, trade in enumerate(self.trades, 1):
result = "WIN" if trade['profit_usd'] > 0 else "LOSS"
f.write(f"{i:4d} {trade['entry_time']:%Y-%m-%d %H:%M:%S} "
f"{trade['direction']:>4} {trade['entry_price']:8.2f} {trade['exit_price']:8.2f} "
f"{trade['profit_usd']:7.2f} {result:>6} {trade['exit_reason']:>12} "
f"{trade['regime']:>15} {trade['ml_conf']:7.3f}\n")
logger.info(f"✓ Results saved to {log_file}")
def main():
"""Run both M15 and H1 HMM backtests"""
config = BacktestConfig()
# Baseline: M15 HMM
logger.info("=" * 80)
logger.info("BASELINE: M15 HMM")
logger.info("=" * 80)
bt_m15 = H1HMMBacktest(config, use_h1_hmm=False)
bt_m15.connect_mt5()
df_m15, df_h1 = bt_m15.fetch_data()
df_m15, df_h1 = bt_m15.prepare_data(df_m15, df_h1)
bt_m15.fit_regimes(df_m15, df_h1)
bt_m15.run_backtest(df_m15, df_h1)
metrics_m15 = bt_m15.calculate_metrics()
bt_m15.save_results("M15_HMM", metrics_m15)
logger.info(f"\nBASELINE Results:")
logger.info(f" Trades: {metrics_m15['total_trades']}, WR: {metrics_m15['win_rate']:.1f}%, "
f"PnL: ${metrics_m15['net_profit']:.2f}, Sharpe: {metrics_m15['sharpe']:.2f}")
logger.info(f" M15 Regime Changes: {metrics_m15['regime_changes_m15']}")
# Test: H1 HMM
logger.info("\n" + "=" * 80)
logger.info("TEST: H1 HMM")
logger.info("=" * 80)
bt_h1 = H1HMMBacktest(config, use_h1_hmm=True)
bt_h1.connect_mt5()
# Reuse same data
bt_h1.fit_regimes(df_m15, df_h1)
bt_h1.run_backtest(df_m15, df_h1)
metrics_h1 = bt_h1.calculate_metrics()
bt_h1.save_results("H1_HMM", metrics_h1)
logger.info(f"\nH1 HMM Results:")
logger.info(f" Trades: {metrics_h1['total_trades']}, WR: {metrics_h1['win_rate']:.1f}%, "
f"PnL: ${metrics_h1['net_profit']:.2f}, Sharpe: {metrics_h1['sharpe']:.2f}")
logger.info(f" H1 Regime Changes: {metrics_h1['regime_changes_h1']}")
# Comparison
logger.info("\n" + "=" * 80)
logger.info("COMPARISON")
logger.info("=" * 80)
pnl_diff = metrics_h1['net_profit'] - metrics_m15['net_profit']
sharpe_diff = metrics_h1['sharpe'] - metrics_m15['sharpe']
regime_reduction = (metrics_m15['regime_changes_m15'] - metrics_h1['regime_changes_h1']) / metrics_m15['regime_changes_m15'] * 100 if metrics_m15['regime_changes_m15'] > 0 else 0
logger.info(f"PnL Difference: ${pnl_diff:+.2f} ({pnl_diff/metrics_m15['net_profit']*100:+.1f}%)")
logger.info(f"Sharpe Difference: {sharpe_diff:+.2f} ({sharpe_diff/metrics_m15['sharpe']*100:+.1f}%)")
logger.info(f"Regime Stability: {regime_reduction:.1f}% fewer regime changes")
bt_m15.mt5.disconnect()
bt_h1.mt5.disconnect()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink