mirror of
https://github.com/B-Wear/QuantumEdge.git
synced 2026-07-27 15:37:46 +00:00
B-Wear
318 lines
12 KiB
Python
318 lines
12 KiB
Python
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
|
|
} |