Machine Learning Trading Strategies: From Research to Production

1class MLTradingSystem:
2    """Production-grade ML trading system architecture"""
3    
4    def __init__(self):
5        self.data_pipeline = DataPipeline()
6        self.feature_store = FeatureStore()
7        self.model_registry = ModelRegistry()
8        self.risk_manager = RiskManager()
9        self.execution_engine = ExecutionEngine()
10        self.monitoring = MonitoringSystem()
11    
12    async def process_market_data(self, tick: MarketTick):
13        # 1. Clean and validate incoming data
14        validated_tick = await self.data_pipeline.validate(tick)
15        
16        # 2. Extract features (pre-computed where possible)
17        features = await self.feature_store.get_features(
18            tick.symbol, 
19            tick.timestamp
20        )
21        
22        # 3. Generate prediction with latency budget
23        with self.monitoring.track_latency("inference"):
24            prediction = await self.model_registry.predict(
25                features,
26                timeout_ms=50  # Hard latency constraint
27            )
28        
29        # 4. Risk checks before execution
30        signal = await self.risk_manager.evaluate(
31            prediction,
32            current_positions=self.get_positions()
33        )
34        
35        # 5. Execute if signal passes all checks
36        if signal.is_valid():
37            await self.execution_engine.submit(signal)
38        
39        # 6. Log everything for later analysis
40        await self.monitoring.log_decision(tick, features, prediction, signal)
41

1import numpy as np
2import pandas as pd
3from numba import jit
4
5@jit(nopython=True)
6def compute_microstructure_features(prices, volumes, timestamps):
7    """
8    Compute market microstructure features efficiently.
9    These capture order flow and price dynamics.
10    """
11    n = len(prices)
12    features = np.zeros((n, 10))
13    
14    for i in range(10, n):
15        window = slice(i-10, i)
16        
17        # Price momentum at multiple scales
18        features[i, 0] = (prices[i] - prices[i-1]) / prices[i-1]
19        features[i, 1] = (prices[i] - prices[i-5]) / prices[i-5]
20        features[i, 2] = (prices[i] - prices[i-10]) / prices[i-10]
21        
22        # Volume-weighted features
23        vwap = np.sum(prices[window] * volumes[window]) / np.sum(volumes[window])
24        features[i, 3] = (prices[i] - vwap) / vwap
25        
26        # Volatility (realized from high-frequency data)
27        returns = np.diff(prices[window]) / prices[window][:-1]
28        features[i, 4] = np.std(returns)
29        
30        # Order flow imbalance
31        price_changes = np.diff(prices[window])
32        buy_volume = np.sum(volumes[window][1:][price_changes > 0])
33        sell_volume = np.sum(volumes[window][1:][price_changes < 0])
34        features[i, 5] = (buy_volume - sell_volume) / (buy_volume + sell_volume + 1e-10)
35        
36        # Tick rule (Lee-Ready algorithm approximation)
37        features[i, 6] = np.sign(prices[i] - prices[i-1])
38        
39        # Time since last trade (important for illiquid markets)
40        features[i, 7] = timestamps[i] - timestamps[i-1]
41        
42        # Spread proxy
43        high_low_spread = (np.max(prices[window]) - np.min(prices[window])) / prices[i]
44        features[i, 8] = high_low_spread
45        
46        # Volume acceleration
47        vol_change = volumes[i] - np.mean(volumes[window])
48        features[i, 9] = vol_change / (np.std(volumes[window]) + 1e-10)
49    
50    return features
51
52class FeatureStore:
53    """
54    Centralized feature storage with pre-computation.
55    Critical for low-latency prediction.
56    """
57    
58    def __init__(self, redis_client):
59        self.cache = redis_client
60        self.feature_definitions = self._load_feature_definitions()
61    
62    async def get_features(self, symbol: str, timestamp: int):
63        """
64        Get features with fallback to computation if cache miss.
65        """
66        cache_key = f"features:{symbol}:{timestamp}"
67        
68        # Try cache first (sub-millisecond)
69        cached = await self.cache.get(cache_key)
70        if cached:
71            return self._deserialize(cached)
72        
73        # Compute if not cached (only for cold start)
74        features = await self._compute_features(symbol, timestamp)
75        
76        # Cache with TTL
77        await self.cache.setex(
78            cache_key, 
79            ttl=300,  # 5 minutes
80            value=self._serialize(features)
81        )
82        
83        return features
84    
85    async def _compute_features(self, symbol: str, timestamp: int):
86        """Compute features from raw data"""
87        # Get historical data window
88        window = await self._get_data_window(symbol, timestamp, lookback=100)
89        
90        # Extract arrays for fast computation
91        prices = window['price'].values
92        volumes = window['volume'].values
93        timestamps = window['timestamp'].values
94        
95        # Compute microstructure features
96        micro_features = compute_microstructure_features(prices, volumes, timestamps)
97        
98        # Add fundamental features (cached separately, updated slowly)
99        fundamental_features = await self._get_fundamental_features(symbol)
100        
101        # Combine feature sets
102        return np.concatenate([
103            micro_features[-1],  # Latest observation
104            fundamental_features
105        ])
106

1class WalkForwardValidator:
2    """
3    Walk-forward validation with proper train/validation/test splits.
4    Prevents lookahead bias and tests adaptability.
5    """
6    
7    def __init__(
8        self, 
9        train_period_days: int = 252,
10        validation_period_days: int = 63,
11        test_period_days: int = 21,
12        step_days: int = 21
13    ):
14        self.train_period = train_period_days
15        self.validation_period = validation_period_days
16        self.test_period = test_period_days
17        self.step = step_days
18    
19    def split_data(self, df: pd.DataFrame):
20        """
21        Generate train/val/test splits for walk-forward analysis.
22        """
23        splits = []
24        
25        start_date = df.index.min()
26        end_date = df.index.max()
27        current_date = start_date
28        
29        while current_date + pd.Timedelta(days=self.train_period + 
30                                          self.validation_period + 
31                                          self.test_period) <= end_date:
32            
33            train_end = current_date + pd.Timedelta(days=self.train_period)
34            val_end = train_end + pd.Timedelta(days=self.validation_period)
35            test_end = val_end + pd.Timedelta(days=self.test_period)
36            
37            train_data = df[current_date:train_end]
38            val_data = df[train_end:val_end]
39            test_data = df[val_end:test_end]
40            
41            splits.append({
42                'train': train_data,
43                'validation': val_data,
44                'test': test_data,
45                'period': (current_date, test_end)
46            })
47            
48            current_date += pd.Timedelta(days=self.step)
49        
50        return splits
51    
52    def evaluate_strategy(self, model_class, data: pd.DataFrame, features: list):
53        """
54        Evaluate strategy across all walk-forward periods.
55        """
56        splits = self.split_data(data)
57        results = []
58        
59        for i, split in enumerate(splits):
60            print(f"Period {i+1}/{len(splits)}: {split['period']}")
61            
62            # Train model
63            model = model_class()
64            model.fit(
65                split['train'][features],
66                split['train']['target']
67            )
68            
69            # Validate and tune hyperparameters
70            val_predictions = model.predict(split['validation'][features])
71            threshold = self._optimize_threshold(
72                val_predictions,
73                split['validation']['target'],
74                split['validation']['returns']
75            )
76            
77            # Test on out-of-sample period
78            test_predictions = model.predict(split['test'][features])
79            test_signals = (test_predictions > threshold).astype(int)
80            
81            # Calculate performance metrics
82            metrics = self._calculate_metrics(
83                test_signals,
84                split['test']['returns'],
85                split['test']
86            )
87            
88            results.append({
89                'period': split['period'],
90                'metrics': metrics,
91                'model': model,
92                'threshold': threshold
93            })
94        
95        return self._aggregate_results(results)
96    
97    def _calculate_metrics(self, signals, returns, data):
98        """Calculate comprehensive performance metrics"""
99        strategy_returns = signals * returns
100        
101        return {
102            'sharpe_ratio': self._sharpe_ratio(strategy_returns),
103            'max_drawdown': self._max_drawdown(strategy_returns),
104            'win_rate': np.mean(strategy_returns > 0),
105            'profit_factor': self._profit_factor(strategy_returns),
106            'total_return': np.sum(strategy_returns),
107            'num_trades': np.sum(np.abs(np.diff(signals))),
108            'avg_trade_return': np.mean(strategy_returns[strategy_returns != 0]),
109            # Transaction cost impact
110            'return_after_costs': self._apply_transaction_costs(
111                strategy_returns, 
112                signals
113            )
114        }
115

1class MLRiskManager:
2    """
3    Multi-layer risk management for ML trading strategies.
4    """
5    
6    def __init__(self, config: dict):
7        self.max_position_size = config['max_position_size']
8        self.max_portfolio_var = config['max_portfolio_var']
9        self.max_correlation = config['max_correlation']
10        self.model_confidence_threshold = config['confidence_threshold']
11        self.regime_detector = RegimeDetector()
12    
13    async def evaluate(self, prediction, current_positions) -> Signal:
14        """
15        Multi-stage risk evaluation pipeline.
16        """
17        signal = Signal(prediction=prediction)
18        
19        # Stage 1: Model confidence
20        if prediction.confidence < self.model_confidence_threshold:
21            signal.reject("Low model confidence")
22            return signal
23        
24        # Stage 2: Position limits
25        proposed_size = self._calculate_position_size(prediction)
26        if proposed_size > self.max_position_size:
27            signal.reject(f"Position size {proposed_size} exceeds limit")
28            return signal
29        
30        # Stage 3: Portfolio-level risk
31        portfolio_var = self._calculate_portfolio_var(
32            current_positions,
33            signal
34        )
35        if portfolio_var > self.max_portfolio_var:
36            signal.reject(f"Portfolio VaR {portfolio_var} exceeds limit")
37            return signal
38        
39        # Stage 4: Correlation checks (avoid crowding)
40        correlation = self._check_correlation(current_positions, signal)
41        if correlation > self.max_correlation:
42            signal.reject(f"High correlation {correlation} with existing positions")
43            return signal
44        
45        # Stage 5: Regime detection
46        current_regime = await self.regime_detector.get_current_regime()
47        if not self._is_regime_suitable(current_regime, signal):
48            signal.reject(f"Unsuitable market regime: {current_regime}")
49            return signal
50        
51        # Stage 6: Model performance monitoring
52        if await self._is_model_degraded(prediction.model_id):
53            signal.reject("Model performance degradation detected")
54            return signal
55        
56        signal.approve(proposed_size)
57        return signal
58    
59    async def _is_model_degraded(self, model_id: str) -> bool:
60        """
61        Check if model performance has degraded.
62        Compare recent performance to validation period.
63        """
64        recent_performance = await self._get_recent_performance(model_id, days=5)
65        expected_performance = await self._get_expected_performance(model_id)
66        
67        # Check if Sharpe ratio has dropped significantly
68        if recent_performance['sharpe'] < expected_performance['sharpe'] * 0.5:
69            return True
70        
71        # Check if win rate has dropped
72        if recent_performance['win_rate'] < expected_performance['win_rate'] * 0.7:
73            return True
74        
75        # Check for unusual loss streaks
76        if recent_performance['consecutive_losses'] > 10:
77            return True
78        
79        return False
80
81class RegimeDetector:
82    """
83    Detect market regime changes that may invalidate model assumptions.
84    """
85    
86    def __init__(self):
87        self.hmm_model = self._load_hmm_model()
88        self.volatility_threshold = 2.0  # Standard deviations
89    
90    async def get_current_regime(self) -> str:
91        """
92        Classify current market regime.
93        """
94        # Get recent market data
95        market_data = await self._get_recent_market_data(days=20)
96        
97        # Calculate regime features
98        volatility = self._calculate_realized_volatility(market_data)
99        trend_strength = self._calculate_trend_strength(market_data)
100        correlation = self._calculate_cross_asset_correlation(market_data)
101        
102        # Historical volatility comparison
103        vol_zscore = self._calculate_volatility_zscore(volatility)
104        
105        # Classify regime
106        if vol_zscore > self.volatility_threshold:
107            return "high_volatility"
108        elif trend_strength > 0.7:
109            return "trending"
110        elif correlation > 0.8:
111            return "risk_off"
112        else:
113            return "normal"
114

1class MLTradingMonitor:
2    """
3    Comprehensive monitoring for ML trading systems.
4    """
5    
6    def __init__(self, prometheus_client, logger):
7        self.metrics = prometheus_client
8        self.logger = logger
9        self.alert_manager = AlertManager()
10    
11    async def log_decision(self, tick, features, prediction, signal):
12        """
13        Log every decision for later analysis and debugging.
14        """
15        decision_log = {
16            'timestamp': tick.timestamp,
17            'symbol': tick.symbol,
18            'features': features.tolist(),
19            'prediction': {
20                'value': prediction.value,
21                'confidence': prediction.confidence,
22                'model_id': prediction.model_id
23            },
24            'signal': {
25                'action': signal.action,
26                'size': signal.size,
27                'reason': signal.reason
28            },
29            'market_state': {
30                'price': tick.price,
31                'volume': tick.volume,
32                'spread': tick.spread
33            }
34        }
35        
36        # Store in time-series database
37        await self.logger.write(decision_log)
38        
39        # Update Prometheus metrics
40        self.metrics.predictions_total.inc()
41        self.metrics.prediction_confidence.observe(prediction.confidence)
42        
43        if signal.is_approved():
44            self.metrics.signals_generated.inc()
45    
46    async def check_model_health(self):
47        """
48        Periodic health checks on model performance.
49        """
50        models = await self._get_active_models()
51        
52        for model in models:
53            # Check prediction latency
54            latency = await self._get_avg_latency(model.id, minutes=5)
55            if latency > model.sla_latency_ms:
56                await self.alert_manager.send_alert(
57                    f"Model {model.id} latency {latency}ms exceeds SLA"
58                )
59            
60            # Check prediction accuracy drift
61            recent_accuracy = await self._get_recent_accuracy(model.id, hours=24)
62            expected_accuracy = model.validation_accuracy
63            
64            if recent_accuracy < expected_accuracy * 0.8:
65                await self.alert_manager.send_alert(
66                    f"Model {model.id} accuracy drift: {recent_accuracy:.2%} vs {expected_accuracy:.2%}"
67                )
68            
69            # Check for data quality issues
70            feature_stats = await self._get_feature_statistics(model.id, hours=1)
71            if self._detect_distribution_shift(feature_stats, model.training_stats):
72                await self.alert_manager.send_alert(
73                    f"Model {model.id} feature distribution shift detected"
74                )
75