Crypto Quantitative Trading Part 3: Optimization, Validation, and Production Deployment

  1import pandas as pd
  2import numpy as np
  3from datetime import datetime, timedelta
  4from typing import List, Dict, Tuple
  5import itertools
  6
  7class WalkForwardAnalysis:
  8    """
  9    Walk-forward analysis framework
 10    """
 11
 12    def __init__(self, train_period_days=180, test_period_days=60,
 13                 step_days=30):
 14        """
 15        train_period_days: Training window size
 16        test_period_days: Testing (out-of-sample) window size
 17        step_days: How much to move forward each iteration
 18        """
 19        self.train_period = timedelta(days=train_period_days)
 20        self.test_period = timedelta(days=test_period_days)
 21        self.step = timedelta(days=step_days)
 22
 23    def generate_windows(self, start_date: datetime,
 24                        end_date: datetime) -> List[Dict]:
 25        """
 26        Generate train/test windows
 27
 28        Returns: List of {'train_start', 'train_end', 'test_start', 'test_end'}
 29        """
 30        windows = []
 31        current_start = start_date
 32
 33        while True:
 34            train_start = current_start
 35            train_end = train_start + self.train_period
 36            test_start = train_end
 37            test_end = test_start + self.test_period
 38
 39            if test_end > end_date:
 40                break
 41
 42            windows.append({
 43                'train_start': train_start,
 44                'train_end': train_end,
 45                'test_start': test_start,
 46                'test_end': test_end,
 47            })
 48
 49            current_start += self.step
 50
 51        return windows
 52
 53    def optimize_parameters(self, df: pd.DataFrame, strategy_class,
 54                           param_grid: Dict) -> Tuple[Dict, float]:
 55        """
 56        Optimize strategy parameters on training data
 57
 58        param_grid: Dict of parameter names to list of values
 59        Example: {'fast_period': [10, 20, 30], 'slow_period': [40, 50, 60]}
 60
 61        Returns: (best_params, best_score)
 62        """
 63        from crypto_quantitative_trading_part2_strategies_backtesting import (
 64            backtest_with_risk_management, PerformanceAnalyzer
 65        )
 66
 67        best_params = None
 68        best_sharpe = -np.inf
 69
 70        # Generate all parameter combinations
 71        param_names = list(param_grid.keys())
 72        param_values = list(param_grid.values())
 73        combinations = list(itertools.product(*param_values))
 74
 75        print(f"Testing {len(combinations)} parameter combinations...")
 76
 77        for i, combination in enumerate(combinations):
 78            params = dict(zip(param_names, combination))
 79
 80            try:
 81                # Create strategy with these parameters
 82                strategy = strategy_class(**params)
 83
 84                # Backtest
 85                engine = backtest_with_risk_management(df, strategy)
 86
 87                # Calculate Sharpe ratio
 88                equity = engine.get_equity_curve()
 89                trades = engine.get_trades_df()
 90
 91                if len(trades) < 10:  # Minimum trades requirement
 92                    continue
 93
 94                metrics = PerformanceAnalyzer.calculate_metrics(
 95                    equity, trades, engine.initial_capital
 96                )
 97
 98                sharpe = metrics.get('sharpe_ratio', -np.inf)
 99
100                if sharpe > best_sharpe:
101                    best_sharpe = sharpe
102                    best_params = params
103
104                if (i + 1) % 10 == 0:
105                    print(f"Tested {i+1}/{len(combinations)} combinations, "
106                          f"best Sharpe: {best_sharpe:.2f}")
107
108            except Exception as e:
109                print(f"Error testing params {params}: {e}")
110                continue
111
112        return best_params, best_sharpe
113
114    def run_walk_forward(self, df: pd.DataFrame, strategy_class,
115                        param_grid: Dict, initial_capital: float = 100000):
116        """
117        Run complete walk-forward analysis
118
119        Returns: Dict with results for each window
120        """
121        from crypto_quantitative_trading_part2_strategies_backtesting import (
122            backtest_with_risk_management, PerformanceAnalyzer
123        )
124
125        # Generate windows
126        start_date = df.index[0]
127        end_date = df.index[-1]
128        windows = self.generate_windows(start_date, end_date)
129
130        print(f"\nRunning walk-forward analysis with {len(windows)} windows")
131        print(f"Train period: {self.train_period.days} days")
132        print(f"Test period: {self.test_period.days} days")
133        print(f"Step: {self.step.days} days\n")
134
135        results = []
136
137        for i, window in enumerate(windows):
138            print(f"\n{'='*60}")
139            print(f"Window {i+1}/{len(windows)}")
140            print(f"Train: {window['train_start']} to {window['train_end']}")
141            print(f"Test: {window['test_start']} to {window['test_end']}")
142            print(f"{'='*60}")
143
144            # Split data
145            train_df = df[window['train_start']:window['train_end']]
146            test_df = df[window['test_start']:window['test_end']]
147
148            if len(train_df) < 100 or len(test_df) < 20:
149                print("Insufficient data in window, skipping...")
150                continue
151
152            # Optimize on training data
153            print("\nOptimizing parameters on training data...")
154            best_params, train_sharpe = self.optimize_parameters(
155                train_df, strategy_class, param_grid
156            )
157
158            if best_params is None:
159                print("No valid parameters found, skipping...")
160                continue
161
162            print(f"\nBest parameters: {best_params}")
163            print(f"Training Sharpe: {train_sharpe:.2f}")
164
165            # Test on out-of-sample data
166            print("\nTesting on out-of-sample data...")
167            strategy = strategy_class(**best_params)
168            engine = backtest_with_risk_management(
169                test_df, strategy, initial_capital
170            )
171
172            equity = engine.get_equity_curve()
173            trades = engine.get_trades_df()
174
175            if len(trades) < 5:
176                print("Insufficient trades in test period, skipping...")
177                continue
178
179            metrics = PerformanceAnalyzer.calculate_metrics(
180                equity, trades, initial_capital
181            )
182
183            print(f"\nOut-of-sample performance:")
184            print(f"  Total Return: {metrics['total_return']:.2f}%")
185            print(f"  Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
186            print(f"  Max Drawdown: {metrics['max_drawdown']:.2f}%")
187            print(f"  Win Rate: {metrics['win_rate']:.2f}%")
188            print(f"  Total Trades: {metrics['total_trades']}")
189
190            results.append({
191                'window': i + 1,
192                'train_start': window['train_start'],
193                'train_end': window['train_end'],
194                'test_start': window['test_start'],
195                'test_end': window['test_end'],
196                'best_params': best_params,
197                'train_sharpe': train_sharpe,
198                'test_metrics': metrics,
199            })
200
201        return results
202
203    def analyze_walk_forward_results(self, results: List[Dict]):
204        """
205        Analyze walk-forward results
206
207        Check for:
208        1. Consistency of out-of-sample performance
209        2. Parameter stability
210        3. Performance degradation
211        """
212        print("\n" + "="*80)
213        print("WALK-FORWARD ANALYSIS SUMMARY")
214        print("="*80)
215
216        # Aggregate out-of-sample performance
217        oos_returns = [r['test_metrics']['total_return'] for r in results]
218        oos_sharpes = [r['test_metrics']['sharpe_ratio'] for r in results]
219        oos_drawdowns = [r['test_metrics']['max_drawdown'] for r in results]
220
221        print(f"\nOut-of-Sample Performance Across {len(results)} Windows:")
222        print(f"  Average Return: {np.mean(oos_returns):.2f}% (±{np.std(oos_returns):.2f}%)")
223        print(f"  Average Sharpe: {np.mean(oos_sharpes):.2f} (±{np.std(oos_sharpes):.2f})")
224        print(f"  Average Max DD: {np.mean(oos_drawdowns):.2f}% (±{np.std(oos_drawdowns):.2f}%)")
225        print(f"  Positive Periods: {sum(1 for r in oos_returns if r > 0)}/{len(oos_returns)}")
226
227        # Check parameter stability
228        print(f"\nParameter Stability:")
229        all_params = {}
230        for result in results:
231            for param, value in result['best_params'].items():
232                if param not in all_params:
233                    all_params[param] = []
234                all_params[param].append(value)
235
236        for param, values in all_params.items():
237            unique_values = len(set(values))
238            print(f"  {param}: {unique_values} unique values (range: {min(values)}-{max(values)})")
239
240        # Check for performance degradation over time
241        print(f"\nPerformance Degradation Analysis:")
242        first_half = oos_sharpes[:len(oos_sharpes)//2]
243        second_half = oos_sharpes[len(oos_sharpes)//2:]
244
245        print(f"  First half avg Sharpe: {np.mean(first_half):.2f}")
246        print(f"  Second half avg Sharpe: {np.mean(second_half):.2f}")
247        print(f"  Degradation: {np.mean(second_half) - np.mean(first_half):.2f}")
248
249        # Overall verdict
250        print(f"\n{'='*80}")
251        print("VERDICT:")
252
253        avg_sharpe = np.mean(oos_sharpes)
254        consistency = sum(1 for s in oos_sharpes if s > 0) / len(oos_sharpes)
255
256        if avg_sharpe > 1.0 and consistency > 0.7:
257            print("✅ STRATEGY PASSES: Consistent positive performance across windows")
258        elif avg_sharpe > 0.5 and consistency > 0.6:
259            print("⚠️  STRATEGY MARGINAL: Moderate performance, use with caution")
260        else:
261            print("❌ STRATEGY FAILS: Inconsistent or negative performance")
262
263        print("="*80)
264
265        return {
266            'avg_return': np.mean(oos_returns),
267            'avg_sharpe': np.mean(oos_sharpes),
268            'avg_drawdown': np.mean(oos_drawdowns),
269            'consistency': consistency,
270        }
271
272# Example usage
273from crypto_quantitative_trading_part1_fundamentals import CryptoDataCollector
274from crypto_quantitative_trading_part2_strategies_backtesting import MovingAverageCrossover
275
276# Fetch 2 years of data
277collector = CryptoDataCollector('binance')
278df = collector.fetch_historical_data('BTC/USDT', '1h', '2022-01-01', '2024-01-01')
279
280# Run walk-forward analysis
281wfa = WalkForwardAnalysis(
282    train_period_days=180,  # 6 months training
283    test_period_days=60,    # 2 months testing
284    step_days=30            # Move forward 1 month
285)
286
287param_grid = {
288    'fast_period': [10, 15, 20, 25, 30],
289    'slow_period': [40, 50, 60, 70, 80],
290}
291
292results = wfa.run_walk_forward(df, MovingAverageCrossover, param_grid)
293summary = wfa.analyze_walk_forward_results(results)

 1class MonteCarloSimulation:
 2    """
 3    Monte Carlo simulation for strategy validation
 4    """
 5
 6    def __init__(self, n_simulations=1000):
 7        self.n_simulations = n_simulations
 8
 9    def simulate_trades(self, trades_df: pd.DataFrame) -> List[pd.DataFrame]:
10        """
11        Generate simulated equity curves by randomizing trade sequence
12
13        This tests if performance is due to skill or luck
14        """
15        simulations = []
16
17        for i in range(self.n_simulations):
18            # Randomly shuffle trades
19            shuffled = trades_df.sample(frac=1).reset_index(drop=True)
20
21            # Calculate cumulative P&L
22            shuffled['cumulative_pnl'] = shuffled['pnl'].cumsum()
23
24            simulations.append(shuffled)
25
26        return simulations
27
28    def analyze_simulations(self, original_pnl: float,
29                           simulations: List[pd.DataFrame]):
30        """
31        Analyze Monte Carlo results
32
33        Determine if original performance is statistically significant
34        """
35        final_pnls = [sim['cumulative_pnl'].iloc[-1] for sim in simulations]
36
37        # Calculate percentile of original performance
38        percentile = sum(1 for pnl in final_pnls if pnl < original_pnl) / len(final_pnls)
39
40        print("\n" + "="*60)
41        print("MONTE CARLO ANALYSIS")
42        print("="*60)
43
44        print(f"\nOriginal Total P&L: ${original_pnl:,.2f}")
45        print(f"Simulated P&L Distribution:")
46        print(f"  Mean: ${np.mean(final_pnls):,.2f}")
47        print(f"  Median: ${np.median(final_pnls):,.2f}")
48        print(f"  Std Dev: ${np.std(final_pnls):,.2f}")
49        print(f"  Min: ${np.min(final_pnls):,.2f}")
50        print(f"  Max: ${np.max(final_pnls):,.2f}")
51
52        print(f"\nOriginal performance percentile: {percentile:.1%}")
53
54        if percentile > 0.95:
55            print("✅ Statistically significant (top 5%)")
56        elif percentile > 0.75:
57            print("⚠️  Above average but not exceptional")
58        else:
59            print("❌ Performance likely due to luck")
60
61        print("="*60)
62
63        return {
64            'percentile': percentile,
65            'mean_pnl': np.mean(final_pnls),
66            'std_pnl': np.std(final_pnls),
67        }
68
69# Example usage
70# After backtesting
71trades_df = result.get_trades_df()
72original_pnl = trades_df['pnl'].sum()
73
74mc = MonteCarloSimulation(n_simulations=1000)
75simulations = mc.simulate_trades(trades_df)
76mc_results = mc.analyze_simulations(original_pnl, simulations)

  1import ccxt
  2from typing import Optional, Dict
  3import time
  4from datetime import datetime
  5
  6class LiveTradingEngine:
  7    """
  8    Live trading engine with exchange integration
  9    """
 10
 11    def __init__(self, exchange_name: str = 'binance',
 12                 api_key: str = None, api_secret: str = None,
 13                 testnet: bool = True):
 14        """
 15        Initialize live trading engine
 16
 17        testnet: Use testnet for paper trading (recommended!)
 18        """
 19        # Initialize exchange
 20        exchange_class = getattr(ccxt, exchange_name)
 21
 22        config = {
 23            'enableRateLimit': True,
 24            'options': {'defaultType': 'spot'},
 25        }
 26
 27        if api_key and api_secret:
 28            config['apiKey'] = api_key
 29            config['secret'] = api_secret
 30
 31        if testnet:
 32            config['options']['defaultType'] = 'future'
 33            if exchange_name == 'binance':
 34                config['urls'] = {
 35                    'api': {
 36                        'public': 'https://testnet.binance.vision/api/v3',
 37                        'private': 'https://testnet.binance.vision/api/v3',
 38                    }
 39                }
 40
 41        self.exchange = exchange_class(config)
 42        self.testnet = testnet
 43
 44        # State tracking
 45        self.active_orders = {}
 46        self.positions = {}
 47
 48        print(f"Initialized {exchange_name} {'testnet' if testnet else 'live'} trading")
 49
 50    def get_account_balance(self) -> Dict:
 51        """Get account balance"""
 52        try:
 53            balance = self.exchange.fetch_balance()
 54            return {
 55                'total': balance['total'],
 56                'free': balance['free'],
 57                'used': balance['used'],
 58            }
 59        except Exception as e:
 60            print(f"Error fetching balance: {e}")
 61            return {}
 62
 63    def get_current_price(self, symbol: str) -> Optional[float]:
 64        """Get current market price"""
 65        try:
 66            ticker = self.exchange.fetch_ticker(symbol)
 67            return ticker['last']
 68        except Exception as e:
 69            print(f"Error fetching price for {symbol}: {e}")
 70            return None
 71
 72    def place_market_order(self, symbol: str, side: str,
 73                          amount: float) -> Optional[Dict]:
 74        """
 75        Place market order
 76
 77        side: 'buy' or 'sell'
 78        amount: Quantity in base currency
 79        """
 80        try:
 81            order = self.exchange.create_market_order(
 82                symbol=symbol,
 83                side=side,
 84                amount=amount
 85            )
 86
 87            print(f"Order placed: {side} {amount} {symbol}")
 88            print(f"Order ID: {order['id']}")
 89
 90            self.active_orders[order['id']] = order
 91
 92            return order
 93
 94        except Exception as e:
 95            print(f"Error placing order: {e}")
 96            return None
 97
 98    def place_limit_order(self, symbol: str, side: str,
 99                         amount: float, price: float) -> Optional[Dict]:
100        """
101        Place limit order
102
103        side: 'buy' or 'sell'
104        amount: Quantity
105        price: Limit price
106        """
107        try:
108            order = self.exchange.create_limit_order(
109                symbol=symbol,
110                side=side,
111                amount=amount,
112                price=price
113            )
114
115            print(f"Limit order placed: {side} {amount} {symbol} @ ${price}")
116            print(f"Order ID: {order['id']}")
117
118            self.active_orders[order['id']] = order
119
120            return order
121
122        except Exception as e:
123            print(f"Error placing limit order: {e}")
124            return None
125
126    def cancel_order(self, order_id: str, symbol: str) -> bool:
127        """Cancel an order"""
128        try:
129            self.exchange.cancel_order(order_id, symbol)
130            print(f"Order {order_id} cancelled")
131
132            if order_id in self.active_orders:
133                del self.active_orders[order_id]
134
135            return True
136
137        except Exception as e:
138            print(f"Error cancelling order: {e}")
139            return False
140
141    def get_order_status(self, order_id: str, symbol: str) -> Optional[Dict]:
142        """Check order status"""
143        try:
144            order = self.exchange.fetch_order(order_id, symbol)
145            return {
146                'id': order['id'],
147                'status': order['status'],  # open, closed, canceled
148                'filled': order['filled'],
149                'remaining': order['remaining'],
150                'average': order['average'],  # Average fill price
151            }
152        except Exception as e:
153            print(f"Error fetching order: {e}")
154            return None
155
156    def get_position(self, symbol: str) -> Optional[Dict]:
157        """Get current position for a symbol"""
158        try:
159            balance = self.exchange.fetch_balance()
160
161            # Extract base currency from symbol (e.g., BTC from BTC/USDT)
162            base_currency = symbol.split('/')[0]
163
164            if base_currency in balance['total']:
165                quantity = balance['total'][base_currency]
166
167                if quantity > 0:
168                    return {
169                        'symbol': symbol,
170                        'quantity': quantity,
171                        'value': quantity * self.get_current_price(symbol),
172                    }
173
174            return None
175
176        except Exception as e:
177            print(f"Error getting position: {e}")
178            return None
179
180# Safe wrapper for live trading
181class SafeLiveTradingEngine:
182    """
183    Wrapper with safety checks and risk controls
184    """
185
186    def __init__(self, engine: LiveTradingEngine,
187                 max_position_size: float = 0.1,
188                 max_daily_trades: int = 10,
189                 max_daily_loss: float = 0.02):
190        """
191        max_position_size: Max fraction of capital per position
192        max_daily_trades: Maximum trades per day
193        max_daily_loss: Maximum daily loss (fraction)
194        """
195        self.engine = engine
196        self.max_position_size = max_position_size
197        self.max_daily_trades = max_daily_trades
198        self.max_daily_loss = max_daily_loss
199
200        self.daily_trades = 0
201        self.daily_pnl = 0.0
202        self.start_of_day_balance = 0.0
203        self.last_reset_date = None
204
205        self._reset_daily_counters()
206
207    def _reset_daily_counters(self):
208        """Reset daily tracking"""
209        current_date = datetime.now().date()
210
211        if self.last_reset_date != current_date:
212            self.daily_trades = 0
213            self.daily_pnl = 0.0
214
215            balance = self.engine.get_account_balance()
216            self.start_of_day_balance = balance['total'].get('USDT', 0)
217
218            self.last_reset_date = current_date
219
220    def can_trade(self) -> Tuple[bool, str]:
221        """Check if trading is allowed"""
222        self._reset_daily_counters()
223
224        # Check daily trade limit
225        if self.daily_trades >= self.max_daily_trades:
226            return False, f"Daily trade limit reached ({self.max_daily_trades})"
227
228        # Check daily loss limit
229        balance = self.engine.get_account_balance()
230        current_balance = balance['total'].get('USDT', 0)
231        daily_loss = (current_balance - self.start_of_day_balance) / self.start_of_day_balance
232
233        if daily_loss <= -self.max_daily_loss:
234            return False, f"Daily loss limit reached ({self.max_daily_loss:.1%})"
235
236        return True, "OK"
237
238    def place_order(self, symbol: str, side: str, amount: float,
239                   order_type: str = 'market', price: float = None):
240        """
241        Place order with safety checks
242
243        order_type: 'market' or 'limit'
244        """
245        # Check if trading is allowed
246        can_trade, reason = self.can_trade()
247        if not can_trade:
248            print(f"❌ Trade rejected: {reason}")
249            return None
250
251        # Validate position size
252        balance = self.engine.get_account_balance()
253        total_usdt = balance['total'].get('USDT', 0)
254        current_price = price or self.engine.get_current_price(symbol)
255        position_value = amount * current_price
256
257        if position_value > total_usdt * self.max_position_size:
258            print(f"❌ Trade rejected: Position size exceeds limit")
259            return None
260
261        # Place order
262        if order_type == 'market':
263            order = self.engine.place_market_order(symbol, side, amount)
264        elif order_type == 'limit':
265            if price is None:
266                print(f"❌ Trade rejected: Price required for limit order")
267                return None
268            order = self.engine.place_limit_order(symbol, side, amount, price)
269        else:
270            print(f"❌ Trade rejected: Invalid order type")
271            return None
272
273        if order:
274            self.daily_trades += 1
275            print(f"✅ Trade executed ({self.daily_trades}/{self.max_daily_trades} today)")
276
277        return order
278
279# Example usage (TESTNET ONLY - NEVER START WITH REAL MONEY!)
280live_engine = LiveTradingEngine(
281    exchange_name='binance',
282    api_key='your_testnet_api_key',
283    api_secret='your_testnet_api_secret',
284    testnet=True  # ALWAYS start with testnet!
285)
286
287safe_engine = SafeLiveTradingEngine(
288    live_engine,
289    max_position_size=0.1,
290    max_daily_trades=5,
291    max_daily_loss=0.02
292)
293
294# Check balance
295balance = live_engine.get_account_balance()
296print(f"Account balance: {balance}")
297
298# Get current price
299btc_price = live_engine.get_current_price('BTC/USDT')
300print(f"BTC Price: ${btc_price:,.2f}")
301
302# Place order (with safety checks)
303order = safe_engine.place_order('BTC/USDT', 'buy', 0.001)

  1import logging
  2from datetime import datetime
  3from typing import Dict, List
  4import json
  5import smtplib
  6from email.mime.text import MIMEText
  7from email.mime.multipart import MIMEMultipart
  8
  9class TradingMonitor:
 10    """
 11    Comprehensive monitoring system for live trading
 12    """
 13
 14    def __init__(self, log_file: str = 'trading.log',
 15                 alert_email: str = None, alert_smtp: Dict = None):
 16        """
 17        Initialize monitoring system
 18
 19        alert_smtp: {'host': 'smtp.gmail.com', 'port': 587,
 20                    'user': 'you@gmail.com', 'password': 'password'}
 21        """
 22        # Setup logging
 23        logging.basicConfig(
 24            level=logging.INFO,
 25            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
 26            handlers=[
 27                logging.FileHandler(log_file),
 28                logging.StreamHandler()
 29            ]
 30        )
 31        self.logger = logging.getLogger('TradingMonitor')
 32
 33        # Alert configuration
 34        self.alert_email = alert_email
 35        self.alert_smtp = alert_smtp
 36
 37        # Metrics tracking
 38        self.metrics = {
 39            'trades_today': 0,
 40            'pnl_today': 0.0,
 41            'max_drawdown_today': 0.0,
 42            'errors_today': 0,
 43            'last_trade_time': None,
 44            'last_error_time': None,
 45        }
 46
 47    def log_trade(self, order: Dict, pnl: float = None):
 48        """Log trade execution"""
 49        self.logger.info(f"Trade executed: {order['side']} {order['amount']} "
 50                        f"{order['symbol']} @ ${order.get('price', 'market')}")
 51
 52        self.metrics['trades_today'] += 1
 53        self.metrics['last_trade_time'] = datetime.now()
 54
 55        if pnl is not None:
 56            self.metrics['pnl_today'] += pnl
 57            self.logger.info(f"Trade P&L: ${pnl:,.2f} | Daily P&L: ${self.metrics['pnl_today']:,.2f}")
 58
 59    def log_error(self, error: Exception, context: str = ''):
 60        """Log error"""
 61        self.logger.error(f"Error in {context}: {str(error)}", exc_info=True)
 62
 63        self.metrics['errors_today'] += 1
 64        self.metrics['last_error_time'] = datetime.now()
 65
 66        # Send alert for critical errors
 67        if self.metrics['errors_today'] >= 3:
 68            self.send_alert(
 69                subject="⚠️ Multiple Trading Errors",
 70                body=f"Detected {self.metrics['errors_today']} errors today. Latest: {str(error)}"
 71            )
 72
 73    def log_position_update(self, symbol: str, position: Dict):
 74        """Log position change"""
 75        self.logger.info(f"Position update: {symbol} - "
 76                        f"Quantity: {position['quantity']}, "
 77                        f"Value: ${position['value']:,.2f}, "
 78                        f"P&L: ${position.get('pnl', 0):,.2f}")
 79
 80    def check_health(self, engine: LiveTradingEngine) -> Dict:
 81        """
 82        Perform health check
 83
 84        Returns: Dict with health status
 85        """
 86        health = {
 87            'timestamp': datetime.now(),
 88            'status': 'healthy',
 89            'issues': [],
 90        }
 91
 92        try:
 93            # Check exchange connectivity
 94            balance = engine.get_account_balance()
 95            if not balance:
 96                health['status'] = 'warning'
 97                health['issues'].append('Cannot fetch balance')
 98
 99            # Check if stuck (no trades in 24h)
100            if self.metrics['last_trade_time']:
101                hours_since_trade = (
102                    datetime.now() - self.metrics['last_trade_time']
103                ).total_seconds() / 3600
104
105                if hours_since_trade > 24:
106                    health['status'] = 'warning'
107                    health['issues'].append(f'No trades in {hours_since_trade:.1f} hours')
108
109            # Check error rate
110            if self.metrics['errors_today'] > 5:
111                health['status'] = 'critical'
112                health['issues'].append(f"High error rate: {self.metrics['errors_today']} errors today")
113
114            # Log health status
115            if health['status'] != 'healthy':
116                self.logger.warning(f"Health check: {health['status']} - {health['issues']}")
117            else:
118                self.logger.info("Health check: healthy")
119
120        except Exception as e:
121            health['status'] = 'critical'
122            health['issues'].append(f"Health check failed: {str(e)}")
123            self.logger.error(f"Health check failed: {e}")
124
125        return health
126
127    def send_alert(self, subject: str, body: str):
128        """Send email alert"""
129        if not self.alert_email or not self.alert_smtp:
130            self.logger.warning(f"Alert (email not configured): {subject}")
131            return
132
133        try:
134            msg = MIMEMultipart()
135            msg['From'] = self.alert_smtp['user']
136            msg['To'] = self.alert_email
137            msg['Subject'] = subject
138
139            msg.attach(MIMEText(body, 'plain'))
140
141            server = smtplib.SMTP(self.alert_smtp['host'], self.alert_smtp['port'])
142            server.starttls()
143            server.login(self.alert_smtp['user'], self.alert_smtp['password'])
144            server.send_message(msg)
145            server.quit()
146
147            self.logger.info(f"Alert sent: {subject}")
148
149        except Exception as e:
150            self.logger.error(f"Failed to send alert: {e}")
151
152    def generate_daily_report(self) -> str:
153        """Generate daily performance report"""
154        report = f"""
155Daily Trading Report - {datetime.now().strftime('%Y-%m-%d')}
156{'='*60}
157
158Performance:
159  Trades: {self.metrics['trades_today']}
160  P&L: ${self.metrics['pnl_today']:,.2f}
161  Max Drawdown: {self.metrics['max_drawdown_today']:.2%}
162
163System Health:
164  Errors: {self.metrics['errors_today']}
165  Last Trade: {self.metrics['last_trade_time']}
166  Last Error: {self.metrics['last_error_time']}
167
168{'='*60}
169        """
170
171        self.logger.info("Daily report generated")
172        return report
173
174# Example usage
175monitor = TradingMonitor(
176    log_file='trading.log',
177    alert_email='your-email@example.com',
178    alert_smtp={
179        'host': 'smtp.gmail.com',
180        'port': 587,
181        'user': 'alerts@yourbot.com',
182        'password': 'your-app-password'
183    }
184)
185
186# Log trade
187monitor.log_trade({
188    'side': 'buy',
189    'amount': 0.1,
190    'symbol': 'BTC/USDT',
191    'price': 43000
192}, pnl=150.50)
193
194# Health check
195health = monitor.check_health(live_engine)
196
197# Daily report
198report = monitor.generate_daily_report()
199print(report)

  1import pandas as pd
  2import numpy as np
  3from sklearn.ensemble import RandomForestClassifier
  4from sklearn.model_selection import TimeSeriesSplit
  5from sklearn.metrics import classification_report, accuracy_score
  6
  7class MLStrategyEnhancement:
  8    """
  9    Machine learning enhancement for trading strategies
 10    """
 11
 12    def __init__(self):
 13        self.model = RandomForestClassifier(
 14            n_estimators=100,
 15            max_depth=10,
 16            random_state=42
 17        )
 18        self.feature_names = []
 19
 20    def engineer_features(self, df: pd.DataFrame) -> pd.DataFrame:
 21        """
 22        Create ML features from price data
 23        """
 24        from crypto_quantitative_trading_part1_fundamentals import (
 25            TrendIndicators, MomentumIndicators, VolatilityIndicators
 26        )
 27
 28        features = df.copy()
 29
 30        # Price features
 31        features['returns'] = features['close'].pct_change()
 32        features['log_returns'] = np.log(features['close'] / features['close'].shift())
 33
 34        # Trend features
 35        for period in [5, 10, 20, 50]:
 36            features[f'sma_{period}'] = TrendIndicators.sma(features['close'], period)
 37            features[f'ema_{period}'] = TrendIndicators.ema(features['close'], period)
 38            features[f'price_to_sma_{period}'] = features['close'] / features[f'sma_{period}']
 39
 40        # Momentum features
 41        features['rsi'] = MomentumIndicators.rsi(features['close'])
 42        features['roc_12'] = MomentumIndicators.roc(features['close'], 12)
 43        features['roc_24'] = MomentumIndicators.roc(features['close'], 24)
 44
 45        # Volatility features
 46        features['atr'] = VolatilityIndicators.atr(
 47            features['high'], features['low'], features['close']
 48        )
 49        bb_mid, bb_upper, bb_lower = VolatilityIndicators.bollinger_bands(features['close'])
 50        features['bb_position'] = (features['close'] - bb_lower) / (bb_upper - bb_lower)
 51
 52        # Volume features
 53        features['volume_sma_20'] = features['volume'].rolling(20).mean()
 54        features['volume_ratio'] = features['volume'] / features['volume_sma_20']
 55
 56        # Lag features
 57        for lag in [1, 2, 3, 5, 10]:
 58            features[f'returns_lag_{lag}'] = features['returns'].shift(lag)
 59            features[f'volume_lag_{lag}'] = features['volume'].shift(lag)
 60
 61        # Rolling statistics
 62        for window in [5, 10, 20]:
 63            features[f'returns_mean_{window}'] = features['returns'].rolling(window).mean()
 64            features[f'returns_std_{window}'] = features['returns'].rolling(window).std()
 65            features[f'volume_mean_{window}'] = features['volume'].rolling(window).mean()
 66
 67        # Target: Next hour return direction
 68        features['target'] = np.where(
 69            features['returns'].shift(-1) > 0, 1, 0
 70        )
 71
 72        return features
 73
 74    def train_model(self, df: pd.DataFrame, train_size: float = 0.8):
 75        """
 76        Train ML model on historical data
 77        """
 78        # Engineer features
 79        df_features = self.engineer_features(df)
 80        df_features = df_features.dropna()
 81
 82        # Select feature columns (exclude target and non-features)
 83        exclude_cols = ['open', 'high', 'low', 'close', 'volume', 'target',
 84                       'returns', 'log_returns']
 85        feature_cols = [col for col in df_features.columns if col not in exclude_cols]
 86
 87        X = df_features[feature_cols]
 88        y = df_features['target']
 89
 90        self.feature_names = feature_cols
 91
 92        # Split data (time series split)
 93        split_idx = int(len(df_features) * train_size)
 94        X_train, X_test = X[:split_idx], X[split_idx:]
 95        y_train, y_test = y[:split_idx], y[split_idx:]
 96
 97        print(f"Training ML model on {len(X_train)} samples...")
 98
 99        # Train model
100        self.model.fit(X_train, y_train)
101
102        # Evaluate
103        y_pred = self.model.predict(X_test)
104
105        accuracy = accuracy_score(y_test, y_pred)
106        print(f"\nModel Accuracy: {accuracy:.2%}")
107        print(f"\nClassification Report:")
108        print(classification_report(y_test, y_pred))
109
110        # Feature importance
111        feature_importance = pd.DataFrame({
112            'feature': feature_cols,
113            'importance': self.model.feature_importances_
114        }).sort_values('importance', ascending=False)
115
116        print(f"\nTop 10 Important Features:")
117        print(feature_importance.head(10))
118
119        return accuracy
120
121    def predict(self, df: pd.DataFrame) -> np.ndarray:
122        """
123        Make predictions on new data
124
125        Returns: Array of predictions (0 or 1)
126        """
127        df_features = self.engineer_features(df)
128        df_features = df_features.dropna()
129
130        X = df_features[self.feature_names]
131        predictions = self.model.predict(X)
132
133        return predictions
134
135    def predict_proba(self, df: pd.DataFrame) -> np.ndarray:
136        """
137        Predict probabilities
138
139        Returns: Array of probabilities for class 1
140        """
141        df_features = self.engineer_features(df)
142        df_features = df_features.dropna()
143
144        X = df_features[self.feature_names]
145        probabilities = self.model.predict_proba(X)[:, 1]  # Probability of class 1
146
147        return probabilities
148
149# Example usage
150ml_enhancer = MLStrategyEnhancement()
151
152# Train on historical data
153df = collector.fetch_historical_data('BTC/USDT', '1h', '2023-01-01', '2024-01-01')
154accuracy = ml_enhancer.train_model(df, train_size=0.8)
155
156# Make predictions on new data
157predictions = ml_enhancer.predict(df.tail(100))
158probabilities = ml_enhancer.predict_proba(df.tail(100))
159
160print(f"\nRecent predictions: {predictions[-10:]}")
161print(f"Recent probabilities: {probabilities[-10:]}")