Spaces:

chen470
/

drl-trading-bot-dev2

Paused

drl-trading-bot-dev2 / backtest_strategy.py

DRL Trading Bot

Feature: HTF Agent integration — live trading, API endpoints, UI tab

fc115d5 8 months ago

27.7 kB

#!/usr/bin/env python3
"""
Strategy Backtester
Replays the EXACT live trading pipeline on historical data to validate
model performance before deployment.
Usage:
    python backtest_strategy.py                     # Backtest all assets
    python backtest_strategy.py --asset BTCUSDT     # Single asset
    python backtest_strategy.py --days 180          # 6 months
"""

import sys
import os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

import numpy as np
import pandas as pd
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from datetime import datetime, timedelta
import json
import logging
import argparse

from stable_baselines3 import PPO
from stable_baselines3.common.vec_env import DummyVecEnv, VecNormalize

from src.features.ultimate_features import UltimateFeatureEngine
from src.features.correlation_engine import SimulatedDominanceEngine
from src.features.regime_detector import MarketRegimeDetector
from src.features.risk_manager import AdaptiveRiskManager
from src.backtest.data_loader import DataLoader

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(name)s] %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class StrategyBacktester:
    """
    Replays the exact live trading pipeline on historical data.
    
    Mirrors:
    - Feature computation (UltimateFeatureEngine + SimulatedDominanceEngine)
    - VecNormalize observation scaling
    - PPO model inference
    - Regime-adaptive SL/TP
    - 60% trailing stops
    - Min hold time + cooldown
    """
    
    def __init__(
        self,
        model_path: str = './data/models/ultimate_agent.zip',
        vec_norm_path: str = './data/models/ultimate_agent_vec_normalize.pkl',
        initial_balance: float = 5000.0,
        position_size: float = 0.25,
        min_hold_bars: int = 4,      # 4 hours at 1h timeframe
        cooldown_bars: int = 2,      # 2 hours cooldown after SL
    ):
        self.initial_balance = initial_balance
        self.position_size = position_size
        self.min_hold_bars = min_hold_bars
        self.cooldown_bars = cooldown_bars
        
        # Load model
        self.model = None
        if Path(model_path).exists():
            try:
                self.model = PPO.load(model_path)
                logger.info(f"✅ Model loaded from {model_path}")
            except Exception as e:
                logger.error(f"Failed to load model: {e}")
        
        # Load VecNormalize
        self.vec_normalize = None
        if Path(vec_norm_path).exists():
            try:
                import gymnasium as gym
                dummy_env = DummyVecEnv([lambda: gym.make('CartPole-v1')])
                self.vec_normalize = VecNormalize.load(vec_norm_path, dummy_env)
                self.vec_normalize.training = False
                self.vec_normalize.norm_reward = False
                logger.info(f"✅ VecNormalize loaded from {vec_norm_path}")
            except Exception as e:
                logger.warning(f"⚠️ VecNormalize load failed: {e}")
        
        # Feature engines
        self.feature_engine = UltimateFeatureEngine()
        self.dominance_engine = SimulatedDominanceEngine()
        self.regime_detector = MarketRegimeDetector()
        self.risk_manager = AdaptiveRiskManager()
    
    def _compute_observation(
        self, df: pd.DataFrame, position: int, 
        position_price: float, balance: float
    ) -> Optional[np.ndarray]:
        """Compute observation vector matching live pipeline."""
        if len(df) < 200:
            return None
        
        try:
            # Features (same as live)
            all_features = self.feature_engine.get_all_features(df)
            dominance_features = self.dominance_engine.compute_simulated_dominance(df)
            all_features.update(dominance_features)
            
            features_df = pd.DataFrame(all_features)
            features_df = features_df.fillna(0).replace([np.inf, -np.inf], 0)
            
            for col in features_df.columns:
                if features_df[col].dtype in [np.float64, np.float32]:
                    features_df[col] = features_df[col].clip(-10, 10)
            
            features = features_df.values.astype(np.float32)
            if len(features) < 1:
                return None
            
            last_features = features[-1].copy()
            
            # Position info
            current_price = df.iloc[-1]['close']
            if position != 0 and position_price > 0:
                if position == 1:
                    unrealized_pnl = (current_price - position_price) / position_price
                else:
                    unrealized_pnl = (position_price - current_price) / position_price
            else:
                unrealized_pnl = 0.0
            
            balance_ratio = (balance - self.initial_balance) / self.initial_balance
            
            position_info = np.array([
                float(position),
                np.clip(unrealized_pnl, -0.5, 0.5),
                np.clip(balance_ratio, -0.5, 0.5),
            ], dtype=np.float32)
            
            observation = np.concatenate([last_features, position_info]).astype(np.float32)
            
            # Apply VecNormalize
            if self.vec_normalize is not None:
                try:
                    obs_2d = observation.reshape(1, -1)
                    observation = self.vec_normalize.normalize_obs(obs_2d).flatten().astype(np.float32)
                except:
                    pass
            
            return observation
        except Exception as e:
            logger.warning(f"Feature computation failed: {e}")
            return None
    
    def _get_regime_adjustments(self, df: pd.DataFrame, direction: str) -> Tuple[float, float]:
        """Get regime-adaptive SL/TP multipliers."""
        sl_mult, tp_mult = 1.0, 1.0
        try:
            regime_info = self.regime_detector.detect_regime(df)
            regime_name = regime_info.regime.value
            if regime_name == 'high_volatility':
                sl_mult, tp_mult = 1.5, 1.5
            elif regime_name in ('trending_up', 'trending_down'):
                tp_mult = 1.5
            elif regime_name == 'ranging':
                sl_mult, tp_mult = 0.8, 0.8
        except:
            pass
        return sl_mult, tp_mult
    
    def run(self, symbol: str = 'BTC/USDT', days: int = 90) -> Dict:
        """
        Run backtest on historical data.
        
        Returns detailed performance report.
        """
        logger.info(f"{'='*60}")
        logger.info(f"BACKTESTING {symbol} — last {days} days")
        logger.info(f"{'='*60}")
        
        # Load data
        loader = DataLoader()
        end_date = datetime.now()
        start_date = end_date - timedelta(days=days)
        
        df = loader.load(
            symbol=symbol,
            timeframe='1h',
            start_date=start_date.strftime('%Y-%m-%d'),
            end_date=end_date.strftime('%Y-%m-%d'),
        )
        
        if df.empty or len(df) < 250:
            logger.error(f"Insufficient data for {symbol}: {len(df)} candles")
            return {"error": f"Insufficient data: {len(df)} candles"}
        
        logger.info(f"Loaded {len(df)} candles ({df.index[0]} to {df.index[-1]})")
        
        # Trading state
        balance = self.initial_balance
        position = 0          # 0=flat, 1=long, -1=short
        position_price = 0.0
        position_units = 0.0
        sl_price = 0.0
        tp_price = 0.0
        highest_price = 0.0
        lowest_price = 0.0
        entry_bar = 0
        last_loss_bar = -100
        
        # Tracking
        trades = []
        equity_curve = []
        max_equity = balance
        
        # Lookback window for features (need 200+ bars)
        start_idx = 200
        
        for i in range(start_idx, len(df)):
            # Current data slice (everything up to current bar)
            current_df = df.iloc[:i+1].copy()
            current_price = float(current_df.iloc[-1]['close'])
            bars_in_position = i - entry_bar if position != 0 else 0
            
            # ─── Check SL/TP ───────────────────────────────────────
            if position != 0:
                exit_trade = False
                exit_reason = ""
                exit_price = current_price
                
                if position == 1:  # LONG
                    highest_price = max(highest_price, current_price)
                    
                    # Enhanced trailing stop (60% of gains)
                    if highest_price > position_price:
                        gain = highest_price - position_price
                        new_sl = position_price + gain * 0.6
                        if new_sl > sl_price:
                            sl_price = new_sl
                    
                    if current_price <= sl_price and sl_price > 0:
                        exit_trade = True
                        exit_reason = "STOP_LOSS" if sl_price <= position_price else "TRAILING_STOP"
                    elif current_price >= tp_price and tp_price > 0:
                        exit_trade = True
                        exit_reason = "TAKE_PROFIT"
                        
                elif position == -1:  # SHORT
                    lowest_price = min(lowest_price, current_price) if lowest_price > 0 else current_price
                    
                    # Enhanced trailing stop (60% of gains)
                    if lowest_price < position_price and lowest_price > 0:
                        gain = position_price - lowest_price
                        new_sl = position_price - gain * 0.6
                        if new_sl < sl_price:
                            sl_price = new_sl
                    
                    if current_price >= sl_price and sl_price > 0:
                        exit_trade = True
                        exit_reason = "STOP_LOSS" if sl_price >= position_price else "TRAILING_STOP"
                    elif current_price <= tp_price and tp_price > 0:
                        exit_trade = True
                        exit_reason = "TAKE_PROFIT"
                
                if exit_trade:
                    # Calculate P&L
                    if position == 1:
                        pnl = (current_price - position_price) * position_units
                    else:
                        pnl = (position_price - current_price) * position_units
                    
                    balance += position_price * position_units + pnl
                    
                    trades.append({
                        'entry_bar': entry_bar,
                        'exit_bar': i,
                        'direction': 'LONG' if position == 1 else 'SHORT',
                        'entry_price': position_price,
                        'exit_price': current_price,
                        'pnl': pnl,
                        'reason': exit_reason,
                        'bars_held': bars_in_position,
                        'timestamp': str(current_df.index[-1]) if hasattr(current_df.index[-1], 'strftime') else str(i),
                    })
                    
                    if "STOP_LOSS" in exit_reason:
                        last_loss_bar = i
                    
                    position = 0
                    position_units = 0
                    position_price = 0
                    sl_price = 0
                    tp_price = 0
                    highest_price = 0
                    lowest_price = 0
                    continue
            
            # ─── Get Model Action ─────────────────────────────────
            if self.model is None:
                equity_curve.append(balance)
                continue
            
            obs = self._compute_observation(current_df, position, position_price, balance)
            if obs is None:
                equity_curve.append(balance)
                continue
            
            try:
                action, _ = self.model.predict(obs, deterministic=True)
                action = int(action.item() if hasattr(action, 'item') else action)
            except Exception as e:
                action = 0
            
            # ─── Apply Guards ─────────────────────────────────────
            # Cooldown after loss
            if action != 0 and position == 0 and (i - last_loss_bar) < self.cooldown_bars:
                action = 0
            
            # Min hold time
            if action != 0 and position != 0 and bars_in_position < self.min_hold_bars:
                action = 0
            
            # ─── Execute Trade ────────────────────────────────────
            if action == 1 and position <= 0:  # BUY
                # Close short first
                if position == -1:
                    pnl = (position_price - current_price) * position_units
                    balance += position_price * position_units + pnl
                    trades.append({
                        'entry_bar': entry_bar,
                        'exit_bar': i,
                        'direction': 'SHORT',
                        'entry_price': position_price,
                        'exit_price': current_price,
                        'pnl': pnl,
                        'reason': 'SIGNAL_EXIT',
                        'bars_held': bars_in_position,
                        'timestamp': str(current_df.index[-1]) if hasattr(current_df.index[-1], 'strftime') else str(i),
                    })
                    position = 0
                
                # Open long
                if position == 0:
                    trade_value = balance * self.position_size
                    if trade_value > 10:
                        position_units = trade_value / current_price
                        position_price = current_price
                        position = 1
                        balance -= trade_value
                        entry_bar = i
                        highest_price = current_price
                        
                        # Calculate SL/TP (regime-adaptive)
                        try:
                            sl_pct, tp_pct = self.risk_manager.get_adaptive_sl_tp(current_df, "long")
                            sl_mult, tp_mult = self._get_regime_adjustments(current_df, "long")
                            sl_pct *= sl_mult
                            tp_pct *= tp_mult
                        except:
                            sl_pct, tp_pct = 0.025, 0.05
                        
                        sl_price = current_price * (1 - sl_pct)
                        tp_price = current_price * (1 + tp_pct)
            
            elif action == 2 and position >= 0:  # SELL
                # Close long first
                if position == 1:
                    pnl = (current_price - position_price) * position_units
                    balance += position_price * position_units + pnl
                    trades.append({
                        'entry_bar': entry_bar,
                        'exit_bar': i,
                        'direction': 'LONG',
                        'entry_price': position_price,
                        'exit_price': current_price,
                        'pnl': pnl,
                        'reason': 'SIGNAL_EXIT',
                        'bars_held': bars_in_position,
                        'timestamp': str(current_df.index[-1]) if hasattr(current_df.index[-1], 'strftime') else str(i),
                    })
                    position = 0
                
                # Open short
                if position == 0:
                    trade_value = balance * self.position_size
                    if trade_value > 10:
                        position_units = trade_value / current_price
                        position_price = current_price
                        position = -1
                        balance -= trade_value
                        entry_bar = i
                        lowest_price = current_price
                        
                        # Calculate SL/TP (regime-adaptive)
                        try:
                            sl_pct, tp_pct = self.risk_manager.get_adaptive_sl_tp(current_df, "short")
                            sl_mult, tp_mult = self._get_regime_adjustments(current_df, "short")
                            sl_pct *= sl_mult
                            tp_pct *= tp_mult
                        except:
                            sl_pct, tp_pct = 0.025, 0.05
                        
                        sl_price = current_price * (1 + sl_pct)
                        tp_price = current_price * (1 - tp_pct)
            
            # Track equity
            unrealized = 0
            if position == 1:
                unrealized = (current_price - position_price) * position_units
            elif position == -1:
                unrealized = (position_price - current_price) * position_units
            
            equity = balance + (position_price * position_units if position != 0 else 0) + unrealized
            equity_curve.append(equity)
            max_equity = max(max_equity, equity)
        
        # ─── Close any remaining position ─────────────────────
        if position != 0:
            final_price = float(df.iloc[-1]['close'])
            if position == 1:
                pnl = (final_price - position_price) * position_units
            else:
                pnl = (position_price - final_price) * position_units
            balance += position_price * position_units + pnl
            trades.append({
                'entry_bar': entry_bar,
                'exit_bar': len(df) - 1,
                'direction': 'LONG' if position == 1 else 'SHORT',
                'entry_price': position_price,
                'exit_price': final_price,
                'pnl': pnl,
                'reason': 'END_OF_DATA',
                'bars_held': len(df) - 1 - entry_bar,
                'timestamp': str(df.index[-1]) if hasattr(df.index[-1], 'strftime') else 'end',
            })
        
        # ─── Calculate Metrics ────────────────────────────────
        report = self._calculate_metrics(trades, equity_curve, symbol)
        return report
    
    def _calculate_metrics(self, trades: List[Dict], equity_curve: List[float], symbol: str) -> Dict:
        """Calculate comprehensive performance metrics."""
        if not trades:
            return {
                'symbol': symbol,
                'total_trades': 0,
                'status': 'NO_TRADES',
                'message': 'Model generated no trades',
            }
        
        # Basic metrics
        pnls = [t['pnl'] for t in trades]
        wins = [p for p in pnls if p > 0]
        losses = [p for p in pnls if p <= 0]
        
        total_return = (sum(pnls) / self.initial_balance) * 100
        win_rate = len(wins) / len(pnls) * 100 if pnls else 0
        avg_win = np.mean(wins) if wins else 0
        avg_loss = abs(np.mean(losses)) if losses else 0
        profit_factor = sum(wins) / abs(sum(losses)) if losses and sum(losses) != 0 else float('inf')
        
        # Sharpe Ratio (annualized from hourly)
        if len(equity_curve) > 1:
            returns = pd.Series(equity_curve).pct_change().dropna()
            if returns.std() > 0:
                sharpe = (returns.mean() / returns.std()) * np.sqrt(8760)  # Annualize from hourly
            else:
                sharpe = 0
        else:
            sharpe = 0
        
        # Max Drawdown
        if equity_curve:
            peak = equity_curve[0]
            max_dd = 0
            for eq in equity_curve:
                peak = max(peak, eq)
                dd = (peak - eq) / peak if peak > 0 else 0
                max_dd = max(max_dd, dd)
        else:
            max_dd = 0
        
        # Win/Loss by reason
        reason_stats = {}
        for t in trades:
            reason = t['reason']
            if reason not in reason_stats:
                reason_stats[reason] = {'count': 0, 'total_pnl': 0, 'wins': 0}
            reason_stats[reason]['count'] += 1
            reason_stats[reason]['total_pnl'] += t['pnl']
            if t['pnl'] > 0:
                reason_stats[reason]['wins'] += 1
        
        # Direction analysis
        long_trades = [t for t in trades if t['direction'] == 'LONG']
        short_trades = [t for t in trades if t['direction'] == 'SHORT']
        long_pnl = sum(t['pnl'] for t in long_trades)
        short_pnl = sum(t['pnl'] for t in short_trades)
        long_wins = sum(1 for t in long_trades if t['pnl'] > 0)
        short_wins = sum(1 for t in short_trades if t['pnl'] > 0)
        
        # Average hold time
        avg_hold = np.mean([t['bars_held'] for t in trades]) if trades else 0
        
        # Validation
        passed = sharpe > 0.5 and win_rate > 50 and max_dd < 0.10
        
        report = {
            'symbol': symbol,
            'status': 'PASSED ✅' if passed else 'FAILED ❌',
            'total_trades': len(trades),
            'total_return_pct': round(total_return, 2),
            'total_pnl': round(sum(pnls), 2),
            'win_rate_pct': round(win_rate, 1),
            'wins': len(wins),
            'losses': len(losses),
            'avg_win': round(avg_win, 2),
            'avg_loss': round(avg_loss, 2),
            'profit_factor': round(profit_factor, 3),
            'sharpe_ratio': round(sharpe, 3),
            'max_drawdown_pct': round(max_dd * 100, 2),
            'avg_hold_bars': round(avg_hold, 1),
            'long_trades': len(long_trades),
            'long_pnl': round(long_pnl, 2),
            'long_win_rate': round(long_wins / len(long_trades) * 100, 1) if long_trades else 0,
            'short_trades': len(short_trades),
            'short_pnl': round(short_pnl, 2),
            'short_win_rate': round(short_wins / len(short_trades) * 100, 1) if short_trades else 0,
            'exit_reasons': reason_stats,
            'trades': trades[-20:],  # Last 20 trades for inspection
        }
        
        # Print summary
        logger.info(f"\n{'='*60}")
        logger.info(f"BACKTEST RESULTS: {symbol}")
        logger.info(f"{'='*60}")
        logger.info(f"  Status:         {report['status']}")
        logger.info(f"  Total Trades:   {report['total_trades']}")
        logger.info(f"  Total Return:   {report['total_return_pct']}%")
        logger.info(f"  Total P&L:      ${report['total_pnl']}")
        logger.info(f"  Win Rate:       {report['win_rate_pct']}%")
        logger.info(f"  Avg Win:        ${report['avg_win']}")
        logger.info(f"  Avg Loss:       ${report['avg_loss']}")
        logger.info(f"  Profit Factor:  {report['profit_factor']}")
        logger.info(f"  Sharpe Ratio:   {report['sharpe_ratio']}")
        logger.info(f"  Max Drawdown:   {report['max_drawdown_pct']}%")
        logger.info(f"  Avg Hold Time:  {report['avg_hold_bars']} bars ({report['avg_hold_bars']:.0f}h)")
        logger.info(f"  LONG:  {report['long_trades']} trades, ${report['long_pnl']} P&L, {report['long_win_rate']}% win")
        logger.info(f"  SHORT: {report['short_trades']} trades, ${report['short_pnl']} P&L, {report['short_win_rate']}% win")
        logger.info(f"{'='*60}")
        
        for reason, stats in reason_stats.items():
            wr = stats['wins'] / stats['count'] * 100 if stats['count'] > 0 else 0
            logger.info(f"  {reason}: {stats['count']} exits, ${stats['total_pnl']:.2f} P&L, {wr:.0f}% win")
        
        return report
    
    def run_all_assets(self, days: int = 90) -> Dict:
        """Run backtest on all supported assets."""
        assets = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'XRP/USDT']
        results = {}
        
        for asset in assets:
            try:
                results[asset] = self.run(asset, days)
            except Exception as e:
                logger.error(f"Backtest failed for {asset}: {e}")
                results[asset] = {'error': str(e), 'status': 'ERROR'}
        
        # Aggregate results
        total_trades = sum(r.get('total_trades', 0) for r in results.values() if 'total_trades' in r)
        total_pnl = sum(r.get('total_pnl', 0) for r in results.values() if 'total_pnl' in r)
        total_wins = sum(r.get('wins', 0) for r in results.values() if 'wins' in r)
        total_losses = sum(r.get('losses', 0) for r in results.values() if 'losses' in r)
        
        agg_win_rate = total_wins / (total_wins + total_losses) * 100 if (total_wins + total_losses) > 0 else 0
        avg_sharpe = np.mean([r.get('sharpe_ratio', 0) for r in results.values() if 'sharpe_ratio' in r])
        max_dd = max([r.get('max_drawdown_pct', 0) for r in results.values() if 'max_drawdown_pct' in r], default=0)
        
        passed = avg_sharpe > 0.5 and agg_win_rate > 50 and max_dd < 10
        
        logger.info(f"\n{'='*60}")
        logger.info(f"AGGREGATE BACKTEST RESULTS")
        logger.info(f"{'='*60}")
        logger.info(f"  Overall Status: {'PASSED ✅' if passed else 'FAILED ❌'}")
        logger.info(f"  Total Trades:   {total_trades}")
        logger.info(f"  Total P&L:      ${total_pnl:.2f}")
        logger.info(f"  Win Rate:       {agg_win_rate:.1f}%")
        logger.info(f"  Avg Sharpe:     {avg_sharpe:.3f}")
        logger.info(f"  Max Drawdown:   {max_dd:.2f}%")
        logger.info(f"{'='*60}")
        
        results['_aggregate'] = {
            'status': 'PASSED ✅' if passed else 'FAILED ❌',
            'total_trades': total_trades,
            'total_pnl': round(total_pnl, 2),
            'win_rate_pct': round(agg_win_rate, 1),
            'avg_sharpe': round(avg_sharpe, 3),
            'max_drawdown_pct': round(max_dd, 2),
        }
        
        # Save report
        report_path = Path('./data/backtest_report.json')
        report_path.parent.mkdir(parents=True, exist_ok=True)
        
        # Make JSON serializable
        serializable = {}
        for k, v in results.items():
            if isinstance(v, dict):
                serializable[k] = {
                    sk: sv for sk, sv in v.items() 
                    if not isinstance(sv, (np.floating, np.integer))
                }
            else:
                serializable[k] = v
        
        with open(report_path, 'w') as f:
            json.dump(serializable, f, indent=2, default=str)
        logger.info(f"\n📄 Report saved to {report_path}")
        
        return results


def main():
    parser = argparse.ArgumentParser(description='Strategy Backtester')
    parser.add_argument('--asset', type=str, default=None, help='Single asset to backtest (e.g., BTCUSDT)')
    parser.add_argument('--days', type=int, default=90, help='Days of history to backtest')
    parser.add_argument('--balance', type=float, default=5000.0, help='Initial balance')
    args = parser.parse_args()
    
    backtester = StrategyBacktester(initial_balance=args.balance)
    
    if args.asset:
        symbol = args.asset.replace('USDT', '/USDT')
        report = backtester.run(symbol, args.days)
    else:
        report = backtester.run_all_assets(args.days)
    
    return report


if __name__ == '__main__':
    main()

	#!/usr/bin/env python3
	"""
	Strategy Backtester

	Replays the EXACT live trading pipeline on historical data to validate
	model performance before deployment.

	Usage:
	python backtest_strategy.py # Backtest all assets
	python backtest_strategy.py --asset BTCUSDT # Single asset
	python backtest_strategy.py --days 180 # 6 months
	"""

	import sys
	import os
	sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

	import numpy as np
	import pandas as pd
	from pathlib import Path
	from typing import Dict, List, Optional, Tuple
	from datetime import datetime, timedelta
	import json
	import logging
	import argparse

	from stable_baselines3 import PPO
	from stable_baselines3.common.vec_env import DummyVecEnv, VecNormalize

	from src.features.ultimate_features import UltimateFeatureEngine
	from src.features.correlation_engine import SimulatedDominanceEngine
	from src.features.regime_detector import MarketRegimeDetector
	from src.features.risk_manager import AdaptiveRiskManager
	from src.backtest.data_loader import DataLoader

	logging.basicConfig(
	level=logging.INFO,
	format='%(asctime)s [%(name)s] %(levelname)s - %(message)s'
	)
	logger = logging.getLogger(__name__)


	class StrategyBacktester:
	"""
	Replays the exact live trading pipeline on historical data.

	Mirrors:
	- Feature computation (UltimateFeatureEngine + SimulatedDominanceEngine)
	- VecNormalize observation scaling
	- PPO model inference
	- Regime-adaptive SL/TP
	- 60% trailing stops
	- Min hold time + cooldown
	"""

	def __init__(
	self,
	model_path: str = './data/models/ultimate_agent.zip',
	vec_norm_path: str = './data/models/ultimate_agent_vec_normalize.pkl',
	initial_balance: float = 5000.0,
	position_size: float = 0.25,
	min_hold_bars: int = 4, # 4 hours at 1h timeframe
	cooldown_bars: int = 2, # 2 hours cooldown after SL
	):
	self.initial_balance = initial_balance
	self.position_size = position_size
	self.min_hold_bars = min_hold_bars
	self.cooldown_bars = cooldown_bars

	# Load model
	self.model = None
	if Path(model_path).exists():
	try:
	self.model = PPO.load(model_path)
	logger.info(f"✅ Model loaded from {model_path}")
	except Exception as e:
	logger.error(f"Failed to load model: {e}")

	# Load VecNormalize
	self.vec_normalize = None
	if Path(vec_norm_path).exists():
	try:
	import gymnasium as gym
	dummy_env = DummyVecEnv([lambda: gym.make('CartPole-v1')])
	self.vec_normalize = VecNormalize.load(vec_norm_path, dummy_env)
	self.vec_normalize.training = False
	self.vec_normalize.norm_reward = False
	logger.info(f"✅ VecNormalize loaded from {vec_norm_path}")
	except Exception as e:
	logger.warning(f"⚠️ VecNormalize load failed: {e}")

	# Feature engines
	self.feature_engine = UltimateFeatureEngine()
	self.dominance_engine = SimulatedDominanceEngine()
	self.regime_detector = MarketRegimeDetector()
	self.risk_manager = AdaptiveRiskManager()

	def _compute_observation(
	self, df: pd.DataFrame, position: int,
	position_price: float, balance: float
	) -> Optional[np.ndarray]:
	"""Compute observation vector matching live pipeline."""
	if len(df) < 200:
	return None

	try:
	# Features (same as live)
	all_features = self.feature_engine.get_all_features(df)
	dominance_features = self.dominance_engine.compute_simulated_dominance(df)
	all_features.update(dominance_features)

	features_df = pd.DataFrame(all_features)
	features_df = features_df.fillna(0).replace([np.inf, -np.inf], 0)

	for col in features_df.columns:
	if features_df[col].dtype in [np.float64, np.float32]:
	features_df[col] = features_df[col].clip(-10, 10)

	features = features_df.values.astype(np.float32)
	if len(features) < 1:
	return None

	last_features = features[-1].copy()

	# Position info
	current_price = df.iloc[-1]['close']
	if position != 0 and position_price > 0:
	if position == 1:
	unrealized_pnl = (current_price - position_price) / position_price
	else:
	unrealized_pnl = (position_price - current_price) / position_price
	else:
	unrealized_pnl = 0.0

	balance_ratio = (balance - self.initial_balance) / self.initial_balance

	position_info = np.array([
	float(position),
	np.clip(unrealized_pnl, -0.5, 0.5),
	np.clip(balance_ratio, -0.5, 0.5),
	], dtype=np.float32)

	observation = np.concatenate([last_features, position_info]).astype(np.float32)

	# Apply VecNormalize
	if self.vec_normalize is not None:
	try:
	obs_2d = observation.reshape(1, -1)
	observation = self.vec_normalize.normalize_obs(obs_2d).flatten().astype(np.float32)
	except:
	pass

	return observation
	except Exception as e:
	logger.warning(f"Feature computation failed: {e}")
	return None

	def _get_regime_adjustments(self, df: pd.DataFrame, direction: str) -> Tuple[float, float]:
	"""Get regime-adaptive SL/TP multipliers."""
	sl_mult, tp_mult = 1.0, 1.0
	try:
	regime_info = self.regime_detector.detect_regime(df)
	regime_name = regime_info.regime.value
	if regime_name == 'high_volatility':
	sl_mult, tp_mult = 1.5, 1.5
	elif regime_name in ('trending_up', 'trending_down'):
	tp_mult = 1.5
	elif regime_name == 'ranging':
	sl_mult, tp_mult = 0.8, 0.8
	except:
	pass
	return sl_mult, tp_mult

	def run(self, symbol: str = 'BTC/USDT', days: int = 90) -> Dict:
	"""
	Run backtest on historical data.

	Returns detailed performance report.
	"""
	logger.info(f"{'='*60}")
	logger.info(f"BACKTESTING {symbol} — last {days} days")
	logger.info(f"{'='*60}")

	# Load data
	loader = DataLoader()
	end_date = datetime.now()
	start_date = end_date - timedelta(days=days)

	df = loader.load(
	symbol=symbol,
	timeframe='1h',
	start_date=start_date.strftime('%Y-%m-%d'),
	end_date=end_date.strftime('%Y-%m-%d'),
	)

	if df.empty or len(df) < 250:
	logger.error(f"Insufficient data for {symbol}: {len(df)} candles")
	return {"error": f"Insufficient data: {len(df)} candles"}

	logger.info(f"Loaded {len(df)} candles ({df.index[0]} to {df.index[-1]})")

	# Trading state
	balance = self.initial_balance
	position = 0 # 0=flat, 1=long, -1=short
	position_price = 0.0
	position_units = 0.0
	sl_price = 0.0
	tp_price = 0.0
	highest_price = 0.0
	lowest_price = 0.0
	entry_bar = 0
	last_loss_bar = -100

	# Tracking
	trades = []
	equity_curve = []
	max_equity = balance

	# Lookback window for features (need 200+ bars)
	start_idx = 200

	for i in range(start_idx, len(df)):
	# Current data slice (everything up to current bar)
	current_df = df.iloc[:i+1].copy()
	current_price = float(current_df.iloc[-1]['close'])
	bars_in_position = i - entry_bar if position != 0 else 0

	# ─── Check SL/TP ───────────────────────────────────────
	if position != 0:
	exit_trade = False
	exit_reason = ""
	exit_price = current_price

	if position == 1: # LONG
	highest_price = max(highest_price, current_price)

	# Enhanced trailing stop (60% of gains)
	if highest_price > position_price:
	gain = highest_price - position_price
	new_sl = position_price + gain * 0.6
	if new_sl > sl_price:
	sl_price = new_sl

	if current_price <= sl_price and sl_price > 0:
	exit_trade = True
	exit_reason = "STOP_LOSS" if sl_price <= position_price else "TRAILING_STOP"
	elif current_price >= tp_price and tp_price > 0:
	exit_trade = True
	exit_reason = "TAKE_PROFIT"

	elif position == -1: # SHORT
	lowest_price = min(lowest_price, current_price) if lowest_price > 0 else current_price

	# Enhanced trailing stop (60% of gains)
	if lowest_price < position_price and lowest_price > 0:
	gain = position_price - lowest_price
	new_sl = position_price - gain * 0.6
	if new_sl < sl_price:
	sl_price = new_sl

	if current_price >= sl_price and sl_price > 0:
	exit_trade = True
	exit_reason = "STOP_LOSS" if sl_price >= position_price else "TRAILING_STOP"
	elif current_price <= tp_price and tp_price > 0:
	exit_trade = True
	exit_reason = "TAKE_PROFIT"

	if exit_trade:
	# Calculate P&L
	if position == 1:
	pnl = (current_price - position_price) * position_units
	else:
	pnl = (position_price - current_price) * position_units

	balance += position_price * position_units + pnl

	trades.append({
	'entry_bar': entry_bar,
	'exit_bar': i,
	'direction': 'LONG' if position == 1 else 'SHORT',
	'entry_price': position_price,
	'exit_price': current_price,
	'pnl': pnl,
	'reason': exit_reason,
	'bars_held': bars_in_position,
	'timestamp': str(current_df.index[-1]) if hasattr(current_df.index[-1], 'strftime') else str(i),
	})

	if "STOP_LOSS" in exit_reason:
	last_loss_bar = i

	position = 0
	position_units = 0
	position_price = 0
	sl_price = 0
	tp_price = 0
	highest_price = 0
	lowest_price = 0
	continue

	# ─── Get Model Action ─────────────────────────────────
	if self.model is None:
	equity_curve.append(balance)
	continue

	obs = self._compute_observation(current_df, position, position_price, balance)
	if obs is None:
	equity_curve.append(balance)
	continue

	try:
	action, _ = self.model.predict(obs, deterministic=True)
	action = int(action.item() if hasattr(action, 'item') else action)
	except Exception as e:
	action = 0

	# ─── Apply Guards ─────────────────────────────────────
	# Cooldown after loss
	if action != 0 and position == 0 and (i - last_loss_bar) < self.cooldown_bars:
	action = 0

	# Min hold time
	if action != 0 and position != 0 and bars_in_position < self.min_hold_bars:
	action = 0

	# ─── Execute Trade ────────────────────────────────────
	if action == 1 and position <= 0: # BUY
	# Close short first
	if position == -1:
	pnl = (position_price - current_price) * position_units
	balance += position_price * position_units + pnl
	trades.append({
	'entry_bar': entry_bar,
	'exit_bar': i,
	'direction': 'SHORT',
	'entry_price': position_price,
	'exit_price': current_price,
	'pnl': pnl,
	'reason': 'SIGNAL_EXIT',
	'bars_held': bars_in_position,
	'timestamp': str(current_df.index[-1]) if hasattr(current_df.index[-1], 'strftime') else str(i),
	})
	position = 0

	# Open long
	if position == 0:
	trade_value = balance * self.position_size
	if trade_value > 10:
	position_units = trade_value / current_price
	position_price = current_price
	position = 1
	balance -= trade_value
	entry_bar = i
	highest_price = current_price

	# Calculate SL/TP (regime-adaptive)
	try:
	sl_pct, tp_pct = self.risk_manager.get_adaptive_sl_tp(current_df, "long")
	sl_mult, tp_mult = self._get_regime_adjustments(current_df, "long")
	sl_pct *= sl_mult
	tp_pct *= tp_mult
	except:
	sl_pct, tp_pct = 0.025, 0.05

	sl_price = current_price * (1 - sl_pct)
	tp_price = current_price * (1 + tp_pct)

	elif action == 2 and position >= 0: # SELL
	# Close long first
	if position == 1:
	pnl = (current_price - position_price) * position_units
	balance += position_price * position_units + pnl
	trades.append({
	'entry_bar': entry_bar,
	'exit_bar': i,
	'direction': 'LONG',
	'entry_price': position_price,
	'exit_price': current_price,
	'pnl': pnl,
	'reason': 'SIGNAL_EXIT',
	'bars_held': bars_in_position,
	'timestamp': str(current_df.index[-1]) if hasattr(current_df.index[-1], 'strftime') else str(i),
	})
	position = 0

	# Open short
	if position == 0:
	trade_value = balance * self.position_size
	if trade_value > 10:
	position_units = trade_value / current_price
	position_price = current_price
	position = -1
	balance -= trade_value
	entry_bar = i
	lowest_price = current_price

	# Calculate SL/TP (regime-adaptive)
	try:
	sl_pct, tp_pct = self.risk_manager.get_adaptive_sl_tp(current_df, "short")
	sl_mult, tp_mult = self._get_regime_adjustments(current_df, "short")
	sl_pct *= sl_mult
	tp_pct *= tp_mult
	except:
	sl_pct, tp_pct = 0.025, 0.05

	sl_price = current_price * (1 + sl_pct)
	tp_price = current_price * (1 - tp_pct)

	# Track equity
	unrealized = 0
	if position == 1:
	unrealized = (current_price - position_price) * position_units
	elif position == -1:
	unrealized = (position_price - current_price) * position_units

	equity = balance + (position_price * position_units if position != 0 else 0) + unrealized
	equity_curve.append(equity)
	max_equity = max(max_equity, equity)

	# ─── Close any remaining position ─────────────────────
	if position != 0:
	final_price = float(df.iloc[-1]['close'])
	if position == 1:
	pnl = (final_price - position_price) * position_units
	else:
	pnl = (position_price - final_price) * position_units
	balance += position_price * position_units + pnl
	trades.append({
	'entry_bar': entry_bar,
	'exit_bar': len(df) - 1,
	'direction': 'LONG' if position == 1 else 'SHORT',
	'entry_price': position_price,
	'exit_price': final_price,
	'pnl': pnl,
	'reason': 'END_OF_DATA',
	'bars_held': len(df) - 1 - entry_bar,
	'timestamp': str(df.index[-1]) if hasattr(df.index[-1], 'strftime') else 'end',
	})

	# ─── Calculate Metrics ────────────────────────────────
	report = self._calculate_metrics(trades, equity_curve, symbol)
	return report

	def _calculate_metrics(self, trades: List[Dict], equity_curve: List[float], symbol: str) -> Dict:
	"""Calculate comprehensive performance metrics."""
	if not trades:
	return {
	'symbol': symbol,
	'total_trades': 0,
	'status': 'NO_TRADES',
	'message': 'Model generated no trades',
	}

	# Basic metrics
	pnls = [t['pnl'] for t in trades]
	wins = [p for p in pnls if p > 0]
	losses = [p for p in pnls if p <= 0]

	total_return = (sum(pnls) / self.initial_balance) * 100
	win_rate = len(wins) / len(pnls) * 100 if pnls else 0
	avg_win = np.mean(wins) if wins else 0
	avg_loss = abs(np.mean(losses)) if losses else 0
	profit_factor = sum(wins) / abs(sum(losses)) if losses and sum(losses) != 0 else float('inf')

	# Sharpe Ratio (annualized from hourly)
	if len(equity_curve) > 1:
	returns = pd.Series(equity_curve).pct_change().dropna()
	if returns.std() > 0:
	sharpe = (returns.mean() / returns.std()) * np.sqrt(8760) # Annualize from hourly
	else:
	sharpe = 0
	else:
	sharpe = 0

	# Max Drawdown
	if equity_curve:
	peak = equity_curve[0]
	max_dd = 0
	for eq in equity_curve:
	peak = max(peak, eq)
	dd = (peak - eq) / peak if peak > 0 else 0
	max_dd = max(max_dd, dd)
	else:
	max_dd = 0

	# Win/Loss by reason
	reason_stats = {}
	for t in trades:
	reason = t['reason']
	if reason not in reason_stats:
	reason_stats[reason] = {'count': 0, 'total_pnl': 0, 'wins': 0}
	reason_stats[reason]['count'] += 1
	reason_stats[reason]['total_pnl'] += t['pnl']
	if t['pnl'] > 0:
	reason_stats[reason]['wins'] += 1

	# Direction analysis
	long_trades = [t for t in trades if t['direction'] == 'LONG']
	short_trades = [t for t in trades if t['direction'] == 'SHORT']
	long_pnl = sum(t['pnl'] for t in long_trades)
	short_pnl = sum(t['pnl'] for t in short_trades)
	long_wins = sum(1 for t in long_trades if t['pnl'] > 0)
	short_wins = sum(1 for t in short_trades if t['pnl'] > 0)

	# Average hold time
	avg_hold = np.mean([t['bars_held'] for t in trades]) if trades else 0

	# Validation
	passed = sharpe > 0.5 and win_rate > 50 and max_dd < 0.10

	report = {
	'symbol': symbol,
	'status': 'PASSED ✅' if passed else 'FAILED ❌',
	'total_trades': len(trades),
	'total_return_pct': round(total_return, 2),
	'total_pnl': round(sum(pnls), 2),
	'win_rate_pct': round(win_rate, 1),
	'wins': len(wins),
	'losses': len(losses),
	'avg_win': round(avg_win, 2),
	'avg_loss': round(avg_loss, 2),
	'profit_factor': round(profit_factor, 3),
	'sharpe_ratio': round(sharpe, 3),
	'max_drawdown_pct': round(max_dd * 100, 2),
	'avg_hold_bars': round(avg_hold, 1),
	'long_trades': len(long_trades),
	'long_pnl': round(long_pnl, 2),
	'long_win_rate': round(long_wins / len(long_trades) * 100, 1) if long_trades else 0,
	'short_trades': len(short_trades),
	'short_pnl': round(short_pnl, 2),
	'short_win_rate': round(short_wins / len(short_trades) * 100, 1) if short_trades else 0,
	'exit_reasons': reason_stats,
	'trades': trades[-20:], # Last 20 trades for inspection
	}

	# Print summary
	logger.info(f"\n{'='*60}")
	logger.info(f"BACKTEST RESULTS: {symbol}")
	logger.info(f"{'='*60}")
	logger.info(f" Status: {report['status']}")
	logger.info(f" Total Trades: {report['total_trades']}")
	logger.info(f" Total Return: {report['total_return_pct']}%")
	logger.info(f" Total P&L: ${report['total_pnl']}")
	logger.info(f" Win Rate: {report['win_rate_pct']}%")
	logger.info(f" Avg Win: ${report['avg_win']}")
	logger.info(f" Avg Loss: ${report['avg_loss']}")
	logger.info(f" Profit Factor: {report['profit_factor']}")
	logger.info(f" Sharpe Ratio: {report['sharpe_ratio']}")
	logger.info(f" Max Drawdown: {report['max_drawdown_pct']}%")
	logger.info(f" Avg Hold Time: {report['avg_hold_bars']} bars ({report['avg_hold_bars']:.0f}h)")
	logger.info(f" LONG: {report['long_trades']} trades, ${report['long_pnl']} P&L, {report['long_win_rate']}% win")
	logger.info(f" SHORT: {report['short_trades']} trades, ${report['short_pnl']} P&L, {report['short_win_rate']}% win")
	logger.info(f"{'='*60}")

	for reason, stats in reason_stats.items():
	wr = stats['wins'] / stats['count'] * 100 if stats['count'] > 0 else 0
	logger.info(f" {reason}: {stats['count']} exits, ${stats['total_pnl']:.2f} P&L, {wr:.0f}% win")

	return report

	def run_all_assets(self, days: int = 90) -> Dict:
	"""Run backtest on all supported assets."""
	assets = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'XRP/USDT']
	results = {}

	for asset in assets:
	try:
	results[asset] = self.run(asset, days)
	except Exception as e:
	logger.error(f"Backtest failed for {asset}: {e}")
	results[asset] = {'error': str(e), 'status': 'ERROR'}

	# Aggregate results
	total_trades = sum(r.get('total_trades', 0) for r in results.values() if 'total_trades' in r)
	total_pnl = sum(r.get('total_pnl', 0) for r in results.values() if 'total_pnl' in r)
	total_wins = sum(r.get('wins', 0) for r in results.values() if 'wins' in r)
	total_losses = sum(r.get('losses', 0) for r in results.values() if 'losses' in r)

	agg_win_rate = total_wins / (total_wins + total_losses) * 100 if (total_wins + total_losses) > 0 else 0
	avg_sharpe = np.mean([r.get('sharpe_ratio', 0) for r in results.values() if 'sharpe_ratio' in r])
	max_dd = max([r.get('max_drawdown_pct', 0) for r in results.values() if 'max_drawdown_pct' in r], default=0)

	passed = avg_sharpe > 0.5 and agg_win_rate > 50 and max_dd < 10

	logger.info(f"\n{'='*60}")
	logger.info(f"AGGREGATE BACKTEST RESULTS")
	logger.info(f"{'='*60}")
	logger.info(f" Overall Status: {'PASSED ✅' if passed else 'FAILED ❌'}")
	logger.info(f" Total Trades: {total_trades}")
	logger.info(f" Total P&L: ${total_pnl:.2f}")
	logger.info(f" Win Rate: {agg_win_rate:.1f}%")
	logger.info(f" Avg Sharpe: {avg_sharpe:.3f}")
	logger.info(f" Max Drawdown: {max_dd:.2f}%")
	logger.info(f"{'='*60}")

	results['_aggregate'] = {
	'status': 'PASSED ✅' if passed else 'FAILED ❌',
	'total_trades': total_trades,
	'total_pnl': round(total_pnl, 2),
	'win_rate_pct': round(agg_win_rate, 1),
	'avg_sharpe': round(avg_sharpe, 3),
	'max_drawdown_pct': round(max_dd, 2),
	}

	# Save report
	report_path = Path('./data/backtest_report.json')
	report_path.parent.mkdir(parents=True, exist_ok=True)

	# Make JSON serializable
	serializable = {}
	for k, v in results.items():
	if isinstance(v, dict):
	serializable[k] = {
	sk: sv for sk, sv in v.items()
	if not isinstance(sv, (np.floating, np.integer))
	}
	else:
	serializable[k] = v

	with open(report_path, 'w') as f:
	json.dump(serializable, f, indent=2, default=str)
	logger.info(f"\n📄 Report saved to {report_path}")

	return results


	def main():
	parser = argparse.ArgumentParser(description='Strategy Backtester')
	parser.add_argument('--asset', type=str, default=None, help='Single asset to backtest (e.g., BTCUSDT)')
	parser.add_argument('--days', type=int, default=90, help='Days of history to backtest')
	parser.add_argument('--balance', type=float, default=5000.0, help='Initial balance')
	args = parser.parse_args()

	backtester = StrategyBacktester(initial_balance=args.balance)

	if args.asset:
	symbol = args.asset.replace('USDT', '/USDT')
	report = backtester.run(symbol, args.days)
	else:
	report = backtester.run_all_assets(args.days)

	return report


	if __name__ == '__main__':
	main()