rckrdmrd
75c4d07690
Hierarchical ML Pipeline for trading predictions:
- Level 0: Attention Models (volatility/flow classification)
- Level 1: Base Models (XGBoost per symbol/timeframe)
- Level 2: Metamodels (XGBoost Stacking + Neural Gating)
Key components:
- src/pipelines/hierarchical_pipeline.py - Main prediction pipeline
- src/models/ - All ML model classes
- src/training/ - Training utilities
- src/api/ - FastAPI endpoints
- scripts/ - Training and evaluation scripts
- config/ - YAML configurations
Note: Trained models (*.joblib, *.pt) are gitignored.
Regenerate with training scripts.
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
1225 lines
42 KiB
Python
1225 lines
42 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Multi-Model Strategy Backtester
|
|
===============================
|
|
Combines predictions from multiple ML models across timeframes:
|
|
- Range Predictor (5m and 15m)
|
|
- Movement Magnitude Predictor
|
|
- AMD Phase Detector
|
|
|
|
Strategy:
|
|
1. 5m signals prepare potential entry
|
|
2. 15m confirms direction and provides context
|
|
3. AMD phase filters unsuitable market conditions
|
|
4. Minimum R:R ratio of 2:1 or 3:1
|
|
|
|
Backtest: Full year 2025 with weekly reports
|
|
|
|
Author: ML-Specialist (NEXUS v4.0)
|
|
Date: 2026-01-05
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / 'src'))
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, List, Optional, Tuple, Any
|
|
from datetime import datetime, timedelta
|
|
from enum import Enum
|
|
import joblib
|
|
from loguru import logger
|
|
import json
|
|
|
|
# ML-Engine imports
|
|
from config.reduced_features import generate_reduced_features, get_feature_columns_without_ohlcv
|
|
from data.database import MySQLConnection
|
|
|
|
|
|
# ============================================================
|
|
# Configuration
|
|
# ============================================================
|
|
|
|
@dataclass
|
|
class MultiModelConfig:
|
|
"""Configuration for multi-model strategy"""
|
|
# Capital
|
|
initial_capital: float = 1000.0
|
|
max_risk_per_trade: float = 0.02 # 2%
|
|
max_drawdown: float = 0.15 # 15%
|
|
max_positions: int = 1 # 1 position at a time for simplicity
|
|
|
|
# R:R requirements
|
|
min_rr_ratio: float = 2.0 # Minimum 2:1 R:R
|
|
preferred_rr_ratio: float = 3.0 # Preferred 3:1 R:R
|
|
|
|
# Confidence thresholds
|
|
min_5m_confidence: float = 0.65
|
|
min_15m_confidence: float = 0.70
|
|
min_combined_score: float = 0.60
|
|
|
|
# Timeframe alignment
|
|
require_timeframe_alignment: bool = True # 5m and 15m must agree
|
|
|
|
# AMD filter
|
|
use_amd_filter: bool = True
|
|
avoid_manipulation_phase: bool = True
|
|
|
|
# Technical confirmation
|
|
use_rsi_filter: bool = True
|
|
use_sar_filter: bool = True
|
|
|
|
# Position management
|
|
atr_sl_multiplier: float = 1.5
|
|
trailing_stop_activation: float = 1.0 # Activate after +1R profit
|
|
|
|
|
|
class TradeDirection(Enum):
|
|
LONG = "LONG"
|
|
SHORT = "SHORT"
|
|
HOLD = "HOLD"
|
|
|
|
|
|
class TradeStatus(Enum):
|
|
OPEN = "OPEN"
|
|
CLOSED_TP = "CLOSED_TP"
|
|
CLOSED_SL = "CLOSED_SL"
|
|
CLOSED_TRAILING = "CLOSED_TRAILING"
|
|
CLOSED_TIMEOUT = "CLOSED_TIMEOUT"
|
|
|
|
|
|
@dataclass
|
|
class MultiModelSignal:
|
|
"""Signal combining multiple model predictions"""
|
|
timestamp: datetime
|
|
symbol: str
|
|
|
|
# 5m predictions
|
|
pred_5m_high: float
|
|
pred_5m_low: float
|
|
conf_5m: float
|
|
direction_5m: TradeDirection
|
|
|
|
# 15m predictions
|
|
pred_15m_high: float
|
|
pred_15m_low: float
|
|
conf_15m: float
|
|
direction_15m: TradeDirection
|
|
|
|
# AMD phase
|
|
amd_phase: str = "unknown"
|
|
amd_confidence: float = 0.0
|
|
|
|
# Magnitude predictions
|
|
asymmetry_ratio: float = 1.0
|
|
suggested_rr: float = 1.0
|
|
|
|
# Technical indicators
|
|
rsi: float = 50.0
|
|
sar_signal: str = "neutral"
|
|
cmf: float = 0.0
|
|
|
|
# Combined assessment
|
|
final_direction: TradeDirection = TradeDirection.HOLD
|
|
combined_score: float = 0.0
|
|
suggested_entry: float = 0.0
|
|
suggested_sl: float = 0.0
|
|
suggested_tp: float = 0.0
|
|
actual_rr: float = 0.0
|
|
|
|
def to_dict(self) -> Dict:
|
|
return {
|
|
'timestamp': self.timestamp.isoformat(),
|
|
'symbol': self.symbol,
|
|
'direction_5m': self.direction_5m.value,
|
|
'direction_15m': self.direction_15m.value,
|
|
'final_direction': self.final_direction.value,
|
|
'combined_score': round(self.combined_score, 3),
|
|
'amd_phase': self.amd_phase,
|
|
'actual_rr': round(self.actual_rr, 2),
|
|
'suggested_entry': round(self.suggested_entry, 4),
|
|
'suggested_sl': round(self.suggested_sl, 4),
|
|
'suggested_tp': round(self.suggested_tp, 4)
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Trade:
|
|
"""Trade record"""
|
|
id: str
|
|
symbol: str
|
|
direction: TradeDirection
|
|
entry_price: float
|
|
stop_loss: float
|
|
take_profit: float
|
|
size: float
|
|
entry_time: datetime
|
|
exit_time: Optional[datetime] = None
|
|
exit_price: Optional[float] = None
|
|
pnl: float = 0.0
|
|
status: TradeStatus = TradeStatus.OPEN
|
|
|
|
# Signal info
|
|
signal: Optional[MultiModelSignal] = None
|
|
|
|
# Tracking
|
|
max_favorable: float = 0.0 # Max favorable excursion
|
|
max_adverse: float = 0.0 # Max adverse excursion
|
|
bars_held: int = 0
|
|
|
|
def calculate_pnl(self, exit_price: float):
|
|
"""Calculate P&L based on exit price"""
|
|
if self.direction == TradeDirection.LONG:
|
|
self.pnl = (exit_price - self.entry_price) * self.size
|
|
else: # SHORT
|
|
self.pnl = (self.entry_price - exit_price) * self.size
|
|
|
|
|
|
@dataclass
|
|
class WeeklyReport:
|
|
"""Weekly performance report"""
|
|
week_start: datetime
|
|
week_end: datetime
|
|
week_number: int
|
|
|
|
# Performance
|
|
starting_equity: float
|
|
ending_equity: float
|
|
net_pnl: float
|
|
return_pct: float
|
|
|
|
# Trade stats
|
|
total_trades: int
|
|
winning_trades: int
|
|
losing_trades: int
|
|
win_rate: float
|
|
|
|
# Risk metrics
|
|
max_drawdown: float
|
|
sharpe_ratio: float
|
|
profit_factor: float
|
|
|
|
# Trade details
|
|
trades: List[Trade] = field(default_factory=list)
|
|
avg_winner: float = 0.0
|
|
avg_loser: float = 0.0
|
|
best_trade: float = 0.0
|
|
worst_trade: float = 0.0
|
|
|
|
def to_dict(self) -> Dict:
|
|
return {
|
|
'week_number': self.week_number,
|
|
'week_start': self.week_start.strftime('%Y-%m-%d'),
|
|
'week_end': self.week_end.strftime('%Y-%m-%d'),
|
|
'starting_equity': round(self.starting_equity, 2),
|
|
'ending_equity': round(self.ending_equity, 2),
|
|
'net_pnl': round(self.net_pnl, 2),
|
|
'return_pct': round(self.return_pct, 2),
|
|
'total_trades': self.total_trades,
|
|
'winning_trades': self.winning_trades,
|
|
'losing_trades': self.losing_trades,
|
|
'win_rate': round(self.win_rate, 1),
|
|
'max_drawdown': round(self.max_drawdown, 2),
|
|
'profit_factor': round(self.profit_factor, 2),
|
|
'avg_winner': round(self.avg_winner, 2),
|
|
'avg_loser': round(self.avg_loser, 2),
|
|
'best_trade': round(self.best_trade, 2),
|
|
'worst_trade': round(self.worst_trade, 2)
|
|
}
|
|
|
|
|
|
# ============================================================
|
|
# Multi-Model Signal Generator
|
|
# ============================================================
|
|
|
|
class MultiModelSignalGenerator:
|
|
"""Generates signals by combining multiple model predictions"""
|
|
|
|
def __init__(
|
|
self,
|
|
model_dir: str = 'models/reduced_features_models',
|
|
config: MultiModelConfig = None
|
|
):
|
|
self.model_dir = Path(model_dir)
|
|
self.config = config or MultiModelConfig()
|
|
self.models = {}
|
|
self.load_models()
|
|
|
|
def load_models(self):
|
|
"""Load all available models"""
|
|
if not self.model_dir.exists():
|
|
logger.warning(f"Model directory not found: {self.model_dir}")
|
|
return
|
|
|
|
for model_file in self.model_dir.glob("*.joblib"):
|
|
if model_file.name != 'metadata.joblib':
|
|
key = model_file.stem
|
|
self.models[key] = joblib.load(model_file)
|
|
logger.info(f"Loaded model: {key}")
|
|
|
|
def get_predictions(
|
|
self,
|
|
features_5m: pd.DataFrame,
|
|
features_15m: pd.DataFrame,
|
|
symbol: str,
|
|
idx_5m: int,
|
|
idx_15m: int
|
|
) -> Dict[str, Any]:
|
|
"""Get predictions from all models for both timeframes"""
|
|
|
|
predictions = {}
|
|
feature_cols = get_feature_columns_without_ohlcv()
|
|
|
|
# 5m predictions
|
|
available_5m = [c for c in feature_cols if c in features_5m.columns]
|
|
if available_5m:
|
|
X_5m = features_5m[available_5m].iloc[[idx_5m]].values
|
|
|
|
key_5m_high = f"{symbol}_5m_high_h3"
|
|
key_5m_low = f"{symbol}_5m_low_h3"
|
|
|
|
if key_5m_high in self.models:
|
|
predictions['5m_high'] = self.models[key_5m_high].predict(X_5m)[0]
|
|
if key_5m_low in self.models:
|
|
predictions['5m_low'] = self.models[key_5m_low].predict(X_5m)[0]
|
|
|
|
# 15m predictions
|
|
available_15m = [c for c in feature_cols if c in features_15m.columns]
|
|
if available_15m:
|
|
X_15m = features_15m[available_15m].iloc[[idx_15m]].values
|
|
|
|
key_15m_high = f"{symbol}_15m_high_h3"
|
|
key_15m_low = f"{symbol}_15m_low_h3"
|
|
|
|
if key_15m_high in self.models:
|
|
predictions['15m_high'] = self.models[key_15m_high].predict(X_15m)[0]
|
|
if key_15m_low in self.models:
|
|
predictions['15m_low'] = self.models[key_15m_low].predict(X_15m)[0]
|
|
|
|
return predictions
|
|
|
|
def calculate_direction_and_confidence(
|
|
self,
|
|
pred_high: float,
|
|
pred_low: float,
|
|
current_price: float,
|
|
atr: float
|
|
) -> Tuple[TradeDirection, float]:
|
|
"""Calculate direction and confidence from predictions"""
|
|
|
|
# Normalize predictions to ATR
|
|
high_potential = pred_high / (atr + 1e-10)
|
|
low_potential = pred_low / (atr + 1e-10)
|
|
|
|
# Calculate asymmetry
|
|
if low_potential > 1e-10:
|
|
asymmetry = high_potential / low_potential
|
|
else:
|
|
asymmetry = high_potential * 10 if high_potential > 0 else 1.0
|
|
|
|
# Determine direction
|
|
if asymmetry > 1.3: # Bullish
|
|
direction = TradeDirection.LONG
|
|
confidence = min(asymmetry / 3.0, 1.0)
|
|
elif asymmetry < 0.77: # Bearish
|
|
direction = TradeDirection.SHORT
|
|
confidence = min(1.0 / (asymmetry + 0.1), 1.0)
|
|
else:
|
|
direction = TradeDirection.HOLD
|
|
confidence = 0.0
|
|
|
|
return direction, confidence
|
|
|
|
def generate_signal(
|
|
self,
|
|
df_5m: pd.DataFrame,
|
|
df_15m: pd.DataFrame,
|
|
features_5m: pd.DataFrame,
|
|
features_15m: pd.DataFrame,
|
|
symbol: str,
|
|
idx_5m: int,
|
|
idx_15m: int
|
|
) -> Optional[MultiModelSignal]:
|
|
"""Generate combined signal from multiple models"""
|
|
|
|
# Get current data
|
|
current_price = df_5m['close'].iloc[idx_5m]
|
|
atr_5m = features_5m['ATR'].iloc[idx_5m] if 'ATR' in features_5m.columns else 1.0
|
|
atr_15m = features_15m['ATR'].iloc[idx_15m] if 'ATR' in features_15m.columns else atr_5m * 2
|
|
|
|
# Get predictions
|
|
predictions = self.get_predictions(
|
|
features_5m, features_15m, symbol, idx_5m, idx_15m
|
|
)
|
|
|
|
if not predictions:
|
|
return None
|
|
|
|
# Get 5m direction
|
|
pred_5m_high = predictions.get('5m_high', 0)
|
|
pred_5m_low = predictions.get('5m_low', 0)
|
|
direction_5m, conf_5m = self.calculate_direction_and_confidence(
|
|
pred_5m_high, pred_5m_low, current_price, atr_5m
|
|
)
|
|
|
|
# Get 15m direction
|
|
pred_15m_high = predictions.get('15m_high', 0)
|
|
pred_15m_low = predictions.get('15m_low', 0)
|
|
direction_15m, conf_15m = self.calculate_direction_and_confidence(
|
|
pred_15m_high, pred_15m_low, current_price, atr_15m
|
|
)
|
|
|
|
# Get technical indicators
|
|
rsi = features_5m['RSI'].iloc[idx_5m] if 'RSI' in features_5m.columns else 50
|
|
sar = features_5m['SAR'].iloc[idx_5m] if 'SAR' in features_5m.columns else current_price
|
|
cmf = features_5m['CMF'].iloc[idx_5m] if 'CMF' in features_5m.columns else 0
|
|
|
|
sar_signal = "bullish" if sar < current_price else "bearish"
|
|
|
|
# Create signal object
|
|
signal = MultiModelSignal(
|
|
timestamp=df_5m.index[idx_5m],
|
|
symbol=symbol,
|
|
pred_5m_high=pred_5m_high,
|
|
pred_5m_low=pred_5m_low,
|
|
conf_5m=conf_5m,
|
|
direction_5m=direction_5m,
|
|
pred_15m_high=pred_15m_high,
|
|
pred_15m_low=pred_15m_low,
|
|
conf_15m=conf_15m,
|
|
direction_15m=direction_15m,
|
|
rsi=rsi,
|
|
sar_signal=sar_signal,
|
|
cmf=cmf
|
|
)
|
|
|
|
# Calculate combined assessment
|
|
self._assess_signal(signal, current_price, atr_5m)
|
|
|
|
return signal
|
|
|
|
def _assess_signal(
|
|
self,
|
|
signal: MultiModelSignal,
|
|
current_price: float,
|
|
atr: float
|
|
):
|
|
"""Assess signal quality and calculate entry/exit levels"""
|
|
|
|
config = self.config
|
|
|
|
# Check timeframe alignment
|
|
if config.require_timeframe_alignment:
|
|
if signal.direction_5m != signal.direction_15m:
|
|
signal.final_direction = TradeDirection.HOLD
|
|
signal.combined_score = 0.0
|
|
return
|
|
if signal.direction_5m == TradeDirection.HOLD:
|
|
signal.final_direction = TradeDirection.HOLD
|
|
signal.combined_score = 0.0
|
|
return
|
|
|
|
# Check confidence thresholds
|
|
if signal.conf_5m < config.min_5m_confidence:
|
|
signal.final_direction = TradeDirection.HOLD
|
|
signal.combined_score = 0.0
|
|
return
|
|
|
|
if signal.conf_15m < config.min_15m_confidence:
|
|
signal.final_direction = TradeDirection.HOLD
|
|
signal.combined_score = 0.0
|
|
return
|
|
|
|
# Technical confirmation
|
|
tech_score = 0.0
|
|
max_tech_score = 0.0
|
|
|
|
if config.use_rsi_filter:
|
|
max_tech_score += 1.0
|
|
if signal.direction_5m == TradeDirection.SHORT and signal.rsi > 55:
|
|
tech_score += 1.0
|
|
elif signal.direction_5m == TradeDirection.LONG and signal.rsi < 45:
|
|
tech_score += 1.0
|
|
|
|
if config.use_sar_filter:
|
|
max_tech_score += 1.0
|
|
if signal.direction_5m == TradeDirection.SHORT and signal.sar_signal == "bearish":
|
|
tech_score += 1.0
|
|
elif signal.direction_5m == TradeDirection.LONG and signal.sar_signal == "bullish":
|
|
tech_score += 1.0
|
|
|
|
# CMF confirmation
|
|
max_tech_score += 1.0
|
|
if signal.direction_5m == TradeDirection.SHORT and signal.cmf < 0:
|
|
tech_score += 1.0
|
|
elif signal.direction_5m == TradeDirection.LONG and signal.cmf > 0:
|
|
tech_score += 1.0
|
|
|
|
tech_confirmation = tech_score / max_tech_score if max_tech_score > 0 else 0.5
|
|
|
|
# Combined score
|
|
signal.combined_score = (
|
|
signal.conf_5m * 0.3 +
|
|
signal.conf_15m * 0.4 +
|
|
tech_confirmation * 0.3
|
|
)
|
|
|
|
if signal.combined_score < config.min_combined_score:
|
|
signal.final_direction = TradeDirection.HOLD
|
|
return
|
|
|
|
# Set final direction
|
|
signal.final_direction = signal.direction_5m
|
|
|
|
# Calculate entry/exit levels
|
|
signal.suggested_entry = current_price
|
|
|
|
if signal.final_direction == TradeDirection.SHORT:
|
|
# SHORT: SL above, TP below
|
|
signal.suggested_sl = current_price + atr * config.atr_sl_multiplier
|
|
|
|
# Use 15m prediction for TP (larger move)
|
|
tp_distance = signal.pred_15m_low * 0.8 if signal.pred_15m_low > 0 else atr * 3
|
|
signal.suggested_tp = current_price - tp_distance
|
|
|
|
else: # LONG
|
|
# LONG: SL below, TP above
|
|
signal.suggested_sl = current_price - atr * config.atr_sl_multiplier
|
|
|
|
# Use 15m prediction for TP
|
|
tp_distance = signal.pred_15m_high * 0.8 if signal.pred_15m_high > 0 else atr * 3
|
|
signal.suggested_tp = current_price + tp_distance
|
|
|
|
# Calculate actual R:R
|
|
risk = abs(signal.suggested_entry - signal.suggested_sl)
|
|
reward = abs(signal.suggested_tp - signal.suggested_entry)
|
|
signal.actual_rr = reward / risk if risk > 0 else 0
|
|
|
|
# Check minimum R:R
|
|
if signal.actual_rr < config.min_rr_ratio:
|
|
# Try to adjust TP for minimum R:R
|
|
min_reward = risk * config.min_rr_ratio
|
|
if signal.final_direction == TradeDirection.SHORT:
|
|
signal.suggested_tp = current_price - min_reward
|
|
else:
|
|
signal.suggested_tp = current_price + min_reward
|
|
signal.actual_rr = config.min_rr_ratio
|
|
|
|
|
|
# ============================================================
|
|
# Multi-Timeframe Backtester
|
|
# ============================================================
|
|
|
|
class MultiModelBacktester:
|
|
"""Backtester for multi-model strategy with weekly reports"""
|
|
|
|
def __init__(self, config: MultiModelConfig = None):
|
|
self.config = config or MultiModelConfig()
|
|
self.signal_generator = MultiModelSignalGenerator(config=self.config)
|
|
|
|
# State
|
|
self.trades: List[Trade] = []
|
|
self.weekly_reports: List[WeeklyReport] = []
|
|
self.equity_curve: List[Tuple[datetime, float]] = []
|
|
|
|
# Risk management
|
|
self.current_equity = self.config.initial_capital
|
|
self.peak_equity = self.config.initial_capital
|
|
self.current_drawdown = 0.0
|
|
|
|
def _load_data(
|
|
self,
|
|
symbol: str,
|
|
start_date: str,
|
|
end_date: str,
|
|
timeframe: str = '5m'
|
|
) -> pd.DataFrame:
|
|
"""Load data from PostgreSQL database"""
|
|
try:
|
|
import psycopg2
|
|
|
|
# PostgreSQL connection
|
|
conn = psycopg2.connect(
|
|
host="localhost",
|
|
port=5432,
|
|
dbname="orbiquant_trading",
|
|
user="orbiquant_user",
|
|
password="orbiquant_dev_2025"
|
|
)
|
|
logger.info(f"Connected to PostgreSQL")
|
|
|
|
# Get ticker_id
|
|
with conn.cursor() as cur:
|
|
cur.execute("SELECT id FROM market_data.tickers WHERE symbol = %s", (symbol,))
|
|
result = cur.fetchone()
|
|
if not result:
|
|
logger.warning(f"Symbol not found: {symbol}")
|
|
return pd.DataFrame()
|
|
ticker_id = result[0]
|
|
|
|
# Load data from parent table (covers all partitions)
|
|
table = "market_data.ohlcv_5m"
|
|
|
|
query = f"""
|
|
SELECT
|
|
timestamp as time,
|
|
open, high, low, close, volume
|
|
FROM {table}
|
|
WHERE ticker_id = %s
|
|
AND timestamp >= %s
|
|
AND timestamp <= %s
|
|
ORDER BY timestamp ASC
|
|
"""
|
|
|
|
df = pd.read_sql_query(
|
|
query, conn,
|
|
params=(ticker_id, start_date, end_date),
|
|
parse_dates=['time']
|
|
)
|
|
conn.close()
|
|
|
|
if df.empty:
|
|
logger.warning(f"No data for {symbol}")
|
|
return df
|
|
|
|
df.set_index('time', inplace=True)
|
|
|
|
# Resample if needed
|
|
if timeframe != '5m':
|
|
tf_map = {'15m': '15min', '30m': '30min', '1H': '1H'}
|
|
offset = tf_map.get(timeframe, timeframe)
|
|
|
|
df = df.resample(offset).agg({
|
|
'open': 'first',
|
|
'high': 'max',
|
|
'low': 'min',
|
|
'close': 'last',
|
|
'volume': 'sum'
|
|
}).dropna()
|
|
|
|
return df
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to load data: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return pd.DataFrame()
|
|
|
|
def run_backtest(
|
|
self,
|
|
symbol: str,
|
|
start_date: str = "2025-01-01",
|
|
end_date: str = "2025-12-31"
|
|
) -> Dict[str, Any]:
|
|
"""Run backtest for full year with weekly reports"""
|
|
|
|
logger.info(f"\n{'='*60}")
|
|
logger.info(f"MULTI-MODEL STRATEGY BACKTEST")
|
|
logger.info(f"Symbol: {symbol}")
|
|
logger.info(f"Period: {start_date} to {end_date}")
|
|
logger.info(f"Capital: ${self.config.initial_capital:,.2f}")
|
|
logger.info(f"Min R:R: {self.config.min_rr_ratio}:1")
|
|
logger.info(f"{'='*60}")
|
|
|
|
# Load data for both timeframes
|
|
logger.info(f"Loading {symbol} 5m data...")
|
|
df_5m = self._load_data(symbol, start_date, end_date, "5m")
|
|
if df_5m is None or df_5m.empty:
|
|
logger.error(f"No 5m data for {symbol}")
|
|
return {}
|
|
logger.info(f"Loaded {len(df_5m)} 5m records")
|
|
|
|
logger.info(f"Loading {symbol} 15m data...")
|
|
df_15m = self._load_data(symbol, start_date, end_date, "15m")
|
|
if df_15m is None or df_15m.empty:
|
|
logger.error(f"No 15m data for {symbol}")
|
|
return {}
|
|
logger.info(f"Loaded {len(df_15m)} 15m records")
|
|
|
|
# Generate features
|
|
logger.info("Generating features...")
|
|
features_5m = generate_reduced_features(df_5m)
|
|
features_15m = generate_reduced_features(df_15m)
|
|
|
|
# Reset state
|
|
self.trades = []
|
|
self.weekly_reports = []
|
|
self.equity_curve = []
|
|
self.current_equity = self.config.initial_capital
|
|
self.peak_equity = self.config.initial_capital
|
|
self.current_drawdown = 0.0
|
|
|
|
# Track weekly data
|
|
current_week_trades = []
|
|
week_start_equity = self.current_equity
|
|
current_week_start = None
|
|
|
|
open_trade: Optional[Trade] = None
|
|
trade_id = 0
|
|
|
|
# Main backtest loop (iterate through 5m data)
|
|
warmup = 50 # Skip warmup period for indicators
|
|
|
|
for i in range(warmup, len(df_5m)):
|
|
current_time = df_5m.index[i]
|
|
current_price = df_5m['close'].iloc[i]
|
|
high_price = df_5m['high'].iloc[i]
|
|
low_price = df_5m['low'].iloc[i]
|
|
|
|
# Find corresponding 15m bar
|
|
# 15m bar that contains this 5m timestamp
|
|
idx_15m = self._find_15m_index(df_15m, current_time)
|
|
if idx_15m is None or idx_15m < 10:
|
|
continue
|
|
|
|
# Weekly tracking
|
|
if current_week_start is None:
|
|
current_week_start = current_time
|
|
|
|
# Check for week change (Monday start)
|
|
if current_time.weekday() == 0 and current_time.hour < 1:
|
|
if current_week_start is not None and current_week_trades:
|
|
# Generate weekly report
|
|
report = self._generate_weekly_report(
|
|
current_week_start,
|
|
current_time - timedelta(hours=1),
|
|
week_start_equity,
|
|
self.current_equity,
|
|
current_week_trades
|
|
)
|
|
self.weekly_reports.append(report)
|
|
logger.info(f"Week {report.week_number}: {report.return_pct:+.2f}%, "
|
|
f"{report.total_trades} trades, WR: {report.win_rate:.1f}%")
|
|
|
|
# Reset for new week
|
|
current_week_start = current_time
|
|
week_start_equity = self.current_equity
|
|
current_week_trades = []
|
|
|
|
# Check and manage open trade
|
|
if open_trade is not None:
|
|
closed = self._check_trade_exit(open_trade, high_price, low_price, current_price)
|
|
if closed:
|
|
open_trade.exit_time = current_time
|
|
self.current_equity += open_trade.pnl
|
|
|
|
# Update peak and drawdown
|
|
if self.current_equity > self.peak_equity:
|
|
self.peak_equity = self.current_equity
|
|
self.current_drawdown = (self.peak_equity - self.current_equity) / self.peak_equity
|
|
|
|
self.trades.append(open_trade)
|
|
current_week_trades.append(open_trade)
|
|
open_trade = None
|
|
|
|
# Record equity
|
|
self.equity_curve.append((current_time, self.current_equity))
|
|
|
|
# Check for new signal (only if no open trade)
|
|
if open_trade is None:
|
|
# Check drawdown limit
|
|
if self.current_drawdown >= self.config.max_drawdown:
|
|
continue
|
|
|
|
signal = self.signal_generator.generate_signal(
|
|
df_5m, df_15m,
|
|
features_5m, features_15m,
|
|
symbol, i, idx_15m
|
|
)
|
|
|
|
if signal and signal.final_direction != TradeDirection.HOLD:
|
|
# Check minimum R:R
|
|
if signal.actual_rr >= self.config.min_rr_ratio:
|
|
# Calculate position size
|
|
risk_amount = self.current_equity * self.config.max_risk_per_trade
|
|
risk_per_unit = abs(signal.suggested_entry - signal.suggested_sl)
|
|
|
|
if risk_per_unit > 0:
|
|
position_size = risk_amount / risk_per_unit
|
|
|
|
# Scale for Gold
|
|
if 'XAU' in symbol:
|
|
position_size = round(position_size, 2)
|
|
else:
|
|
position_size = round(position_size * 10000) / 10000
|
|
|
|
if position_size > 0:
|
|
trade_id += 1
|
|
open_trade = Trade(
|
|
id=f"T{trade_id:04d}",
|
|
symbol=symbol,
|
|
direction=signal.final_direction,
|
|
entry_price=signal.suggested_entry,
|
|
stop_loss=signal.suggested_sl,
|
|
take_profit=signal.suggested_tp,
|
|
size=position_size,
|
|
entry_time=current_time,
|
|
signal=signal
|
|
)
|
|
logger.debug(f"Opened trade {open_trade.id}: "
|
|
f"{signal.final_direction.value} @ {current_price:.2f}, "
|
|
f"R:R={signal.actual_rr:.1f}")
|
|
|
|
# Close any remaining trade
|
|
if open_trade is not None:
|
|
open_trade.exit_price = df_5m['close'].iloc[-1]
|
|
open_trade.exit_time = df_5m.index[-1]
|
|
open_trade.status = TradeStatus.CLOSED_TIMEOUT
|
|
open_trade.calculate_pnl(open_trade.exit_price)
|
|
self.current_equity += open_trade.pnl
|
|
self.trades.append(open_trade)
|
|
current_week_trades.append(open_trade)
|
|
|
|
# Generate final week report if needed
|
|
if current_week_trades:
|
|
report = self._generate_weekly_report(
|
|
current_week_start,
|
|
df_5m.index[-1],
|
|
week_start_equity,
|
|
self.current_equity,
|
|
current_week_trades
|
|
)
|
|
self.weekly_reports.append(report)
|
|
|
|
# Calculate final metrics
|
|
return self._calculate_final_metrics(symbol)
|
|
|
|
def _find_15m_index(self, df_15m: pd.DataFrame, timestamp: datetime) -> Optional[int]:
|
|
"""Find the 15m bar index that contains the given 5m timestamp"""
|
|
try:
|
|
# Find the 15m bar that started at or before this time
|
|
mask = df_15m.index <= timestamp
|
|
if mask.any():
|
|
return mask.sum() - 1
|
|
return None
|
|
except:
|
|
return None
|
|
|
|
def _check_trade_exit(
|
|
self,
|
|
trade: Trade,
|
|
high: float,
|
|
low: float,
|
|
close: float
|
|
) -> bool:
|
|
"""Check if trade should be closed"""
|
|
|
|
trade.bars_held += 1
|
|
|
|
if trade.direction == TradeDirection.LONG:
|
|
# Update MFE/MAE
|
|
trade.max_favorable = max(trade.max_favorable, high - trade.entry_price)
|
|
trade.max_adverse = max(trade.max_adverse, trade.entry_price - low)
|
|
|
|
# Check stop loss
|
|
if low <= trade.stop_loss:
|
|
trade.exit_price = trade.stop_loss
|
|
trade.status = TradeStatus.CLOSED_SL
|
|
trade.calculate_pnl(trade.exit_price)
|
|
return True
|
|
|
|
# Check take profit
|
|
if high >= trade.take_profit:
|
|
trade.exit_price = trade.take_profit
|
|
trade.status = TradeStatus.CLOSED_TP
|
|
trade.calculate_pnl(trade.exit_price)
|
|
return True
|
|
|
|
else: # SHORT
|
|
# Update MFE/MAE
|
|
trade.max_favorable = max(trade.max_favorable, trade.entry_price - low)
|
|
trade.max_adverse = max(trade.max_adverse, high - trade.entry_price)
|
|
|
|
# Check stop loss
|
|
if high >= trade.stop_loss:
|
|
trade.exit_price = trade.stop_loss
|
|
trade.status = TradeStatus.CLOSED_SL
|
|
trade.calculate_pnl(trade.exit_price)
|
|
return True
|
|
|
|
# Check take profit
|
|
if low <= trade.take_profit:
|
|
trade.exit_price = trade.take_profit
|
|
trade.status = TradeStatus.CLOSED_TP
|
|
trade.calculate_pnl(trade.exit_price)
|
|
return True
|
|
|
|
# Timeout after 6 hours (72 bars of 5m)
|
|
if trade.bars_held >= 72:
|
|
trade.exit_price = close
|
|
trade.status = TradeStatus.CLOSED_TIMEOUT
|
|
trade.calculate_pnl(trade.exit_price)
|
|
return True
|
|
|
|
return False
|
|
|
|
def _generate_weekly_report(
|
|
self,
|
|
week_start: datetime,
|
|
week_end: datetime,
|
|
start_equity: float,
|
|
end_equity: float,
|
|
trades: List[Trade]
|
|
) -> WeeklyReport:
|
|
"""Generate weekly performance report"""
|
|
|
|
net_pnl = end_equity - start_equity
|
|
return_pct = (net_pnl / start_equity) * 100 if start_equity > 0 else 0
|
|
|
|
winning = [t for t in trades if t.pnl > 0]
|
|
losing = [t for t in trades if t.pnl <= 0]
|
|
|
|
win_rate = len(winning) / len(trades) * 100 if trades else 0
|
|
|
|
avg_winner = np.mean([t.pnl for t in winning]) if winning else 0
|
|
avg_loser = np.mean([t.pnl for t in losing]) if losing else 0
|
|
|
|
gross_profit = sum(t.pnl for t in winning)
|
|
gross_loss = abs(sum(t.pnl for t in losing))
|
|
profit_factor = gross_profit / gross_loss if gross_loss > 0 else float('inf')
|
|
|
|
# Calculate max drawdown for the week
|
|
max_dd = 0
|
|
peak = start_equity
|
|
current = start_equity
|
|
for trade in trades:
|
|
current += trade.pnl
|
|
if current > peak:
|
|
peak = current
|
|
dd = (peak - current) / peak if peak > 0 else 0
|
|
max_dd = max(max_dd, dd)
|
|
|
|
# Week number
|
|
week_number = week_start.isocalendar()[1]
|
|
|
|
return WeeklyReport(
|
|
week_start=week_start,
|
|
week_end=week_end,
|
|
week_number=week_number,
|
|
starting_equity=start_equity,
|
|
ending_equity=end_equity,
|
|
net_pnl=net_pnl,
|
|
return_pct=return_pct,
|
|
total_trades=len(trades),
|
|
winning_trades=len(winning),
|
|
losing_trades=len(losing),
|
|
win_rate=win_rate,
|
|
max_drawdown=max_dd * 100,
|
|
sharpe_ratio=0, # Would need daily returns to calculate
|
|
profit_factor=min(profit_factor, 999),
|
|
trades=trades,
|
|
avg_winner=avg_winner,
|
|
avg_loser=avg_loser,
|
|
best_trade=max(t.pnl for t in trades) if trades else 0,
|
|
worst_trade=min(t.pnl for t in trades) if trades else 0
|
|
)
|
|
|
|
def _calculate_final_metrics(self, symbol: str) -> Dict[str, Any]:
|
|
"""Calculate final backtest metrics"""
|
|
|
|
total_return = (self.current_equity - self.config.initial_capital) / self.config.initial_capital * 100
|
|
|
|
winning = [t for t in self.trades if t.pnl > 0]
|
|
losing = [t for t in self.trades if t.pnl <= 0]
|
|
|
|
win_rate = len(winning) / len(self.trades) * 100 if self.trades else 0
|
|
|
|
avg_winner = np.mean([t.pnl for t in winning]) if winning else 0
|
|
avg_loser = np.mean([t.pnl for t in losing]) if losing else 0
|
|
|
|
gross_profit = sum(t.pnl for t in winning)
|
|
gross_loss = abs(sum(t.pnl for t in losing))
|
|
profit_factor = gross_profit / gross_loss if gross_loss > 0 else float('inf')
|
|
|
|
# Max drawdown from equity curve
|
|
max_dd = 0
|
|
peak = self.config.initial_capital
|
|
for _, equity in self.equity_curve:
|
|
if equity > peak:
|
|
peak = equity
|
|
dd = (peak - equity) / peak if peak > 0 else 0
|
|
max_dd = max(max_dd, dd)
|
|
|
|
# Direction breakdown
|
|
long_trades = [t for t in self.trades if t.direction == TradeDirection.LONG]
|
|
short_trades = [t for t in self.trades if t.direction == TradeDirection.SHORT]
|
|
|
|
long_wins = len([t for t in long_trades if t.pnl > 0])
|
|
short_wins = len([t for t in short_trades if t.pnl > 0])
|
|
|
|
return {
|
|
'symbol': symbol,
|
|
'period': f"{self.equity_curve[0][0].date()} to {self.equity_curve[-1][0].date()}" if self.equity_curve else "N/A",
|
|
'initial_capital': self.config.initial_capital,
|
|
'final_capital': round(self.current_equity, 2),
|
|
'total_return_pct': round(total_return, 2),
|
|
'total_trades': len(self.trades),
|
|
'winning_trades': len(winning),
|
|
'losing_trades': len(losing),
|
|
'win_rate': round(win_rate, 1),
|
|
'profit_factor': round(min(profit_factor, 999), 2),
|
|
'max_drawdown_pct': round(max_dd * 100, 2),
|
|
'avg_winner': round(avg_winner, 2),
|
|
'avg_loser': round(avg_loser, 2),
|
|
'best_trade': round(max(t.pnl for t in self.trades), 2) if self.trades else 0,
|
|
'worst_trade': round(min(t.pnl for t in self.trades), 2) if self.trades else 0,
|
|
'long_trades': len(long_trades),
|
|
'long_wins': long_wins,
|
|
'long_wr': round(long_wins / len(long_trades) * 100, 1) if long_trades else 0,
|
|
'short_trades': len(short_trades),
|
|
'short_wins': short_wins,
|
|
'short_wr': round(short_wins / len(short_trades) * 100, 1) if short_trades else 0,
|
|
'total_weeks': len(self.weekly_reports),
|
|
'profitable_weeks': len([w for w in self.weekly_reports if w.net_pnl > 0])
|
|
}
|
|
|
|
|
|
# ============================================================
|
|
# Report Generator
|
|
# ============================================================
|
|
|
|
class MultiModelReportGenerator:
|
|
"""Generate comprehensive reports"""
|
|
|
|
@staticmethod
|
|
def generate_annual_report(
|
|
backtester: MultiModelBacktester,
|
|
metrics: Dict[str, Any]
|
|
) -> str:
|
|
"""Generate annual summary report"""
|
|
|
|
report = f"""# INFORME ANUAL - ESTRATEGIA MULTI-MODELO
|
|
|
|
**Símbolo:** {metrics['symbol']}
|
|
**Período:** {metrics['period']}
|
|
**Capital Inicial:** ${metrics['initial_capital']:,.2f}
|
|
**Capital Final:** ${metrics['final_capital']:,.2f}
|
|
|
|
---
|
|
|
|
## RESUMEN EJECUTIVO
|
|
|
|
| Métrica | Valor |
|
|
|---------|-------|
|
|
| **Retorno Total** | {metrics['total_return_pct']:+.2f}% |
|
|
| **Total Trades** | {metrics['total_trades']} |
|
|
| **Win Rate** | {metrics['win_rate']:.1f}% |
|
|
| **Profit Factor** | {metrics['profit_factor']:.2f} |
|
|
| **Max Drawdown** | {metrics['max_drawdown_pct']:.2f}% |
|
|
|
|
---
|
|
|
|
## DESGLOSE POR DIRECCIÓN
|
|
|
|
### LONG Trades
|
|
| Métrica | Valor |
|
|
|---------|-------|
|
|
| Total | {metrics['long_trades']} |
|
|
| Ganadores | {metrics['long_wins']} |
|
|
| Win Rate | {metrics['long_wr']:.1f}% |
|
|
|
|
### SHORT Trades
|
|
| Métrica | Valor |
|
|
|---------|-------|
|
|
| Total | {metrics['short_trades']} |
|
|
| Ganadores | {metrics['short_wins']} |
|
|
| Win Rate | {metrics['short_wr']:.1f}% |
|
|
|
|
---
|
|
|
|
## ESTADÍSTICAS DE TRADES
|
|
|
|
| Métrica | Valor |
|
|
|---------|-------|
|
|
| Promedio Ganador | ${metrics['avg_winner']:,.2f} |
|
|
| Promedio Perdedor | ${metrics['avg_loser']:,.2f} |
|
|
| Mejor Trade | ${metrics['best_trade']:,.2f} |
|
|
| Peor Trade | ${metrics['worst_trade']:,.2f} |
|
|
|
|
---
|
|
|
|
## RENDIMIENTO SEMANAL
|
|
|
|
| Semana | Inicio | Fin | P&L | Retorno | Trades | WR | Max DD |
|
|
|--------|--------|-----|-----|---------|--------|-----|--------|
|
|
"""
|
|
|
|
for w in backtester.weekly_reports:
|
|
report += f"| {w.week_number} | {w.week_start.strftime('%m/%d')} | "
|
|
report += f"{w.week_end.strftime('%m/%d')} | ${w.net_pnl:+.2f} | "
|
|
report += f"{w.return_pct:+.2f}% | {w.total_trades} | "
|
|
report += f"{w.win_rate:.0f}% | {w.max_drawdown:.1f}% |\n"
|
|
|
|
report += f"""
|
|
---
|
|
|
|
## SEMANAS RENTABLES
|
|
|
|
- **Total Semanas:** {metrics['total_weeks']}
|
|
- **Semanas Rentables:** {metrics['profitable_weeks']}
|
|
- **% Semanas Positivas:** {metrics['profitable_weeks']/metrics['total_weeks']*100:.1f}%
|
|
|
|
---
|
|
|
|
## CONFIGURACIÓN DE ESTRATEGIA
|
|
|
|
- **R:R Mínimo:** {backtester.config.min_rr_ratio}:1
|
|
- **Riesgo por Trade:** {backtester.config.max_risk_per_trade*100:.0f}%
|
|
- **Max Drawdown Permitido:** {backtester.config.max_drawdown*100:.0f}%
|
|
- **Alineación Timeframes:** {'Sí' if backtester.config.require_timeframe_alignment else 'No'}
|
|
- **Filtro RSI:** {'Sí' if backtester.config.use_rsi_filter else 'No'}
|
|
- **Filtro SAR:** {'Sí' if backtester.config.use_sar_filter else 'No'}
|
|
|
|
---
|
|
|
|
*Generado: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*
|
|
"""
|
|
return report
|
|
|
|
@staticmethod
|
|
def generate_weekly_details(backtester: MultiModelBacktester) -> str:
|
|
"""Generate detailed weekly reports"""
|
|
|
|
report = "# INFORMES SEMANALES DETALLADOS\n\n"
|
|
|
|
for w in backtester.weekly_reports:
|
|
report += f"""
|
|
## Semana {w.week_number} ({w.week_start.strftime('%Y-%m-%d')} - {w.week_end.strftime('%Y-%m-%d')})
|
|
|
|
### Resumen
|
|
| Métrica | Valor |
|
|
|---------|-------|
|
|
| Equity Inicial | ${w.starting_equity:,.2f} |
|
|
| Equity Final | ${w.ending_equity:,.2f} |
|
|
| P&L Neto | ${w.net_pnl:+,.2f} |
|
|
| Retorno | {w.return_pct:+.2f}% |
|
|
| Trades | {w.total_trades} |
|
|
| Win Rate | {w.win_rate:.1f}% |
|
|
| Profit Factor | {w.profit_factor:.2f} |
|
|
| Max Drawdown | {w.max_drawdown:.2f}% |
|
|
|
|
### Trades de la Semana
|
|
| ID | Dirección | Entrada | SL | TP | Salida | P&L | Status |
|
|
|----|-----------|---------|-----|-----|--------|-----|--------|
|
|
"""
|
|
for t in w.trades:
|
|
exit_str = f"{t.exit_price:.2f}" if t.exit_price else "N/A"
|
|
report += f"| {t.id} | {t.direction.value} | {t.entry_price:.2f} | "
|
|
report += f"{t.stop_loss:.2f} | {t.take_profit:.2f} | "
|
|
report += f"{exit_str} | ${t.pnl:+.2f} | {t.status.value} |\n"
|
|
|
|
report += "\n---\n"
|
|
|
|
return report
|
|
|
|
@staticmethod
|
|
def save_reports(
|
|
backtester: MultiModelBacktester,
|
|
metrics: Dict[str, Any],
|
|
output_dir: str = 'reports'
|
|
):
|
|
"""Save all reports to files"""
|
|
|
|
output_path = Path(output_dir)
|
|
output_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
symbol = metrics['symbol']
|
|
|
|
# Annual report
|
|
annual_report = MultiModelReportGenerator.generate_annual_report(backtester, metrics)
|
|
annual_file = output_path / f"annual_report_{symbol}_{timestamp}.md"
|
|
with open(annual_file, 'w') as f:
|
|
f.write(annual_report)
|
|
logger.info(f"Saved annual report to {annual_file}")
|
|
|
|
# Weekly details
|
|
weekly_report = MultiModelReportGenerator.generate_weekly_details(backtester)
|
|
weekly_file = output_path / f"weekly_details_{symbol}_{timestamp}.md"
|
|
with open(weekly_file, 'w') as f:
|
|
f.write(weekly_report)
|
|
logger.info(f"Saved weekly details to {weekly_file}")
|
|
|
|
# JSON metrics
|
|
json_file = output_path / f"backtest_metrics_{symbol}_{timestamp}.json"
|
|
with open(json_file, 'w') as f:
|
|
json.dump(metrics, f, indent=2)
|
|
logger.info(f"Saved JSON metrics to {json_file}")
|
|
|
|
# Weekly summary CSV
|
|
csv_data = []
|
|
for w in backtester.weekly_reports:
|
|
csv_data.append(w.to_dict())
|
|
|
|
if csv_data:
|
|
df = pd.DataFrame(csv_data)
|
|
csv_file = output_path / f"weekly_summary_{symbol}_{timestamp}.csv"
|
|
df.to_csv(csv_file, index=False)
|
|
logger.info(f"Saved weekly CSV to {csv_file}")
|
|
|
|
return annual_file, weekly_file, json_file
|
|
|
|
|
|
# ============================================================
|
|
# Main Execution
|
|
# ============================================================
|
|
|
|
def run_multi_model_backtest():
|
|
"""Run full year backtest with multi-model strategy"""
|
|
|
|
# Configure strategy
|
|
config = MultiModelConfig(
|
|
initial_capital=1000.0,
|
|
max_risk_per_trade=0.02,
|
|
min_rr_ratio=2.0, # Minimum 2:1 R:R
|
|
preferred_rr_ratio=3.0, # Prefer 3:1
|
|
require_timeframe_alignment=True,
|
|
use_rsi_filter=True,
|
|
use_sar_filter=True
|
|
)
|
|
|
|
logger.info("="*60)
|
|
logger.info("MULTI-MODEL STRATEGY BACKTESTER")
|
|
logger.info(f"Min R:R: {config.min_rr_ratio}:1")
|
|
logger.info(f"Timeframe Alignment: {config.require_timeframe_alignment}")
|
|
logger.info("="*60)
|
|
|
|
# Create backtester
|
|
backtester = MultiModelBacktester(config)
|
|
|
|
# Run backtest for XAUUSD (available data: Jan-Mar 2025)
|
|
metrics = backtester.run_backtest(
|
|
symbol="XAUUSD",
|
|
start_date="2025-01-01",
|
|
end_date="2025-03-18" # Data only available until this date
|
|
)
|
|
|
|
if metrics:
|
|
# Print summary
|
|
print("\n" + "="*60)
|
|
print("BACKTEST RESULTS")
|
|
print("="*60)
|
|
print(f"Symbol: {metrics['symbol']}")
|
|
print(f"Period: {metrics['period']}")
|
|
print(f"Initial Capital: ${metrics['initial_capital']:,.2f}")
|
|
print(f"Final Capital: ${metrics['final_capital']:,.2f}")
|
|
print(f"Total Return: {metrics['total_return_pct']:+.2f}%")
|
|
print(f"Total Trades: {metrics['total_trades']}")
|
|
print(f"Win Rate: {metrics['win_rate']:.1f}%")
|
|
print(f"Profit Factor: {metrics['profit_factor']:.2f}")
|
|
print(f"Max Drawdown: {metrics['max_drawdown_pct']:.2f}%")
|
|
print(f"Profitable Weeks: {metrics['profitable_weeks']}/{metrics['total_weeks']}")
|
|
print("="*60)
|
|
|
|
# Save reports
|
|
MultiModelReportGenerator.save_reports(backtester, metrics)
|
|
|
|
return metrics, backtester
|
|
|
|
|
|
if __name__ == "__main__":
|
|
metrics, backtester = run_multi_model_backtest()
|