QuantumEdge/trading_system 2.0.py

import numpy as np
import pandas as pd
from datetime import datetime, timedelta
import logging
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
import json
import os

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('logs/trading_system.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

@dataclass
class Trade:
    symbol: str
    direction: str
    entry_price: float
    stop_loss: float
    take_profit: float
    position_size: float
    entry_time: datetime
    exit_price: Optional[float] = None
    exit_time: Optional[datetime] = None
    pnl: Optional[float] = None
    status: str = 'open'

class TradingSystem:
    def __init__(self, initial_capital: float = 50.0):
        self.initial_capital = initial_capital
        self.current_capital = initial_capital
        self.performance_history: List[Dict] = []
        self.active_trades: Dict[str, Trade] = {}
        self.trade_history: List[Trade] = []
        self.strategy_parameters = self._get_initial_parameters()
        self.risk_metrics = self._initialize_risk_metrics()
        self.last_recalibration = datetime.now()
        
        # Load configuration
        self._load_config()
        
    def _load_config(self):
        """Load configuration from config file"""
        try:
            with open('config/config.json', 'r') as f:
                self.config = json.load(f)
        except FileNotFoundError:
            logger.warning("Config file not found. Using default parameters.")
            self.config = self._get_default_config()
    
    def _get_default_config(self) -> Dict:
        """Get default configuration parameters"""
        return {
            'risk_per_trade': 0.01,  # 1% risk per trade
            'max_positions': 2,
            'min_win_rate': 0.4,
            'recalibration_window': 20,
            'max_drawdown': 0.1,  # 10% maximum drawdown
            'leverage': 1,  # No leverage initially
            'position_sizing': {
                'method': 'fixed_fractional',
                'fraction': 0.01  # 1% of capital per trade
            }
        }
    
    def _get_initial_parameters(self) -> Dict:
        """Get initial strategy parameters"""
        return {
            'rsi_period': 14,
            'rsi_overbought': 70,
            'rsi_oversold': 30,
            'bb_period': 20,
            'bb_std': 2,
            'macd_fast': 12,
            'macd_slow': 26,
            'macd_signal': 9,
            'atr_period': 14,
            'atr_multiplier': 2
        }
    
    def _initialize_risk_metrics(self) -> Dict:
        """Initialize risk metrics tracking"""
        return {
            'total_trades': 0,
            'winning_trades': 0,
            'losing_trades': 0,
            'win_rate': 0.0,
            'profit_factor': 0.0,
            'max_drawdown': 0.0,
            'current_drawdown': 0.0,
            'avg_trade': 0.0,
            'largest_win': 0.0,
            'largest_loss': 0.0
        }
    
    def monitor_performance(self, window_size: int = 20) -> bool:
        """
        Monitor recent performance and determine if recalibration is needed
        Returns True if recalibration is needed
        """
        if len(self.performance_history) < window_size:
            return False
            
        recent_performance = self.performance_history[-window_size:]
        win_rate = sum(1 for trade in recent_performance if trade['pnl'] > 0) / window_size
        
        # Check various performance metrics
        needs_recalibration = False
        
        # Win rate check
        if win_rate < self.config['min_win_rate']:
            logger.warning(f"Win rate {win_rate:.2%} below threshold {self.config['min_win_rate']:.2%}")
            needs_recalibration = True
        
        # Drawdown check
        current_drawdown = self._calculate_drawdown()
        if current_drawdown > self.config['max_drawdown']:
            logger.warning(f"Current drawdown {current_drawdown:.2%} exceeds maximum {self.config['max_drawdown']:.2%}")
            needs_recalibration = True
        
        # Profit factor check
        profit_factor = self._calculate_profit_factor(recent_performance)
        if profit_factor < 1.0:
            logger.warning(f"Profit factor {profit_factor:.2f} below 1.0")
            needs_recalibration = True
        
        return needs_recalibration
    
    def recalibrate_strategy(self, market_data: pd.DataFrame):
        """
        Adjust strategy parameters based on recent market conditions
        """
        logger.info("Starting strategy recalibration")
        
        # Analyze market conditions
        volatility = self._calculate_volatility(market_data)
        trend_strength = self._calculate_trend_strength(market_data)
        
        # Adjust parameters based on market conditions
        new_parameters = self.strategy_parameters.copy()
        
        # Adjust RSI levels based on volatility
        if volatility > 0.02:  # High volatility
            new_parameters['rsi_overbought'] = 75
            new_parameters['rsi_oversold'] = 25
        else:  # Low volatility
            new_parameters['rsi_overbought'] = 70
            new_parameters['rsi_oversold'] = 30
        
        # Adjust ATR multiplier based on trend strength
        if trend_strength > 0.7:  # Strong trend
            new_parameters['atr_multiplier'] = 2.5
        else:  # Weak trend
            new_parameters['atr_multiplier'] = 2.0
        
        # Update parameters
        self.strategy_parameters = new_parameters
        self.last_recalibration = datetime.now()
        
        logger.info("Strategy recalibration completed")
        logger.info(f"New parameters: {new_parameters}")
    
    def calculate_position_size(self, entry_price: float, stop_loss: float) -> float:
        """
        Calculate position size based on risk management rules
        """
        risk_amount = self.current_capital * self.config['risk_per_trade']
        risk_per_unit = abs(entry_price - stop_loss)
        
        if risk_per_unit == 0:
            logger.warning("Risk per unit is zero. Cannot calculate position size.")
            return 0
        
        position_size = risk_amount / risk_per_unit
        
        # Apply leverage if configured
        if self.config['leverage'] > 1:
            position_size *= self.config['leverage']
        
        # Ensure position size doesn't exceed maximum allowed
        max_position = self.current_capital * self.config['position_sizing']['fraction']
        position_size = min(position_size, max_position)
        
        return position_size
    
    def _calculate_volatility(self, data: pd.DataFrame) -> float:
        """Calculate market volatility"""
        returns = data['close'].pct_change()
        return returns.std()
    
    def _calculate_trend_strength(self, data: pd.DataFrame) -> float:
        """Calculate trend strength using ADX"""
        # Implementation would go here
        return 0.5  # Placeholder
    
    def _calculate_drawdown(self) -> float:
        """Calculate current drawdown"""
        if not self.performance_history:
            return 0.0
        
        peak = max(self.performance_history, key=lambda x: x['equity'])['equity']
        current = self.performance_history[-1]['equity']
        return (peak - current) / peak
    
    def _calculate_profit_factor(self, trades: List[Dict]) -> float:
        """Calculate profit factor from recent trades"""
        gross_profit = sum(t['pnl'] for t in trades if t['pnl'] > 0)
        gross_loss = abs(sum(t['pnl'] for t in trades if t['pnl'] < 0))
        
        if gross_loss == 0:
            return float('inf')
        
        return gross_profit / gross_loss
    
    def update_risk_metrics(self, trade: Trade):
        """Update risk metrics after a trade"""
        self.risk_metrics['total_trades'] += 1
        
        if trade.pnl and trade.pnl > 0:
            self.risk_metrics['winning_trades'] += 1
            self.risk_metrics['largest_win'] = max(
                self.risk_metrics['largest_win'],
                trade.pnl
            )
        elif trade.pnl and trade.pnl < 0:
            self.risk_metrics['losing_trades'] += 1
            self.risk_metrics['largest_loss'] = min(
                self.risk_metrics['largest_loss'],
                trade.pnl
            )
        
        # Update win rate
        if self.risk_metrics['total_trades'] > 0:
            self.risk_metrics['win_rate'] = (
                self.risk_metrics['winning_trades'] / 
                self.risk_metrics['total_trades']
            )
        
        # Update average trade
        if trade.pnl:
            self.risk_metrics['avg_trade'] = (
                (self.risk_metrics['avg_trade'] * (self.risk_metrics['total_trades'] - 1) + 
                 trade.pnl) / self.risk_metrics['total_trades']
            )
    
    def save_state(self):
        """Save current system state"""
        state = {
            'current_capital': self.current_capital,
            'strategy_parameters': self.strategy_parameters,
            'risk_metrics': self.risk_metrics,
            'last_recalibration': self.last_recalibration.isoformat(),
            'active_trades': {
                symbol: {
                    'direction': trade.direction,
                    'entry_price': trade.entry_price,
                    'stop_loss': trade.stop_loss,
                    'take_profit': trade.take_profit,
                    'position_size': trade.position_size,
                    'entry_time': trade.entry_time.isoformat()
                }
                for symbol, trade in self.active_trades.items()
            }
        }
        
        try:
            with open('data/system_state.json', 'w') as f:
                json.dump(state, f, indent=4)
        except Exception as e:
            logger.error(f"Error saving system state: {str(e)}")
    
    def load_state(self):
        """Load system state from file"""
        try:
            with open('data/system_state.json', 'r') as f:
                state = json.load(f)
                
            self.current_capital = state['current_capital']
            self.strategy_parameters = state['strategy_parameters']
            self.risk_metrics = state['risk_metrics']
            self.last_recalibration = datetime.fromisoformat(state['last_recalibration'])
            
            # Reconstruct active trades
            self.active_trades = {}
            for symbol, trade_data in state['active_trades'].items():
                self.active_trades[symbol] = Trade(
                    symbol=symbol,
                    direction=trade_data['direction'],
                    entry_price=trade_data['entry_price'],
                    stop_loss=trade_data['stop_loss'],
                    take_profit=trade_data['take_profit'],
                    position_size=trade_data['position_size'],
                    entry_time=datetime.fromisoformat(trade_data['entry_time'])
                )
        except FileNotFoundError:
            logger.info("No saved state found. Starting fresh.")
        except Exception as e:
            logger.error(f"Error loading system state: {str(e)}")
    
    def get_system_status(self) -> Dict:
        """Get current system status"""
        return {
            'current_capital': self.current_capital,
            'total_trades': self.risk_metrics['total_trades'],
            'win_rate': self.risk_metrics['win_rate'],
            'profit_factor': self.risk_metrics['profit_factor'],
            'current_drawdown': self.risk_metrics['current_drawdown'],
            'active_trades': len(self.active_trades),
            'last_recalibration': self.last_recalibration.isoformat(),
            'strategy_parameters': self.strategy_parameters
        }