Algorithmic Trading Strategies: From Research to Production

1import numpy as np
2import pandas as pd
3from dataclasses import dataclass
4from typing import Optional
5
6@dataclass
7class Position:
8    symbol: str
9    quantity: int
10    entry_price: float
11    entry_time: pd.Timestamp
12
13class MeanReversionStrategy:
14    def __init__(
15        self,
16        lookback_period: int = 20,
17        entry_threshold: float = 2.0,  # Standard deviations
18        exit_threshold: float = 0.5,
19        max_holding_period_hours: int = 24
20    ):
21        self.lookback_period = lookback_period
22        self.entry_threshold = entry_threshold
23        self.exit_threshold = exit_threshold
24        self.max_holding_period_hours = max_holding_period_hours
25        
26        self.position: Optional[Position] = None
27        self.price_history = []
28    
29    def calculate_zscore(self, price: float) -> float:
30        """Calculate z-score of current price vs historical mean."""
31        if len(self.price_history) < self.lookback_period:
32            return 0.0
33        
34        recent_prices = self.price_history[-self.lookback_period:]
35        mean_price = np.mean(recent_prices)
36        std_price = np.std(recent_prices)
37        
38        if std_price == 0:
39            return 0.0
40        
41        return (price - mean_price) / std_price
42    
43    def generate_signal(
44        self,
45        symbol: str,
46        price: float,
47        timestamp: pd.Timestamp
48    ) -> Optional[str]:
49        """
50        Generate trading signal.
51        
52        Returns:
53            'BUY' - price significantly below mean
54            'SELL' - price significantly above mean
55            'CLOSE' - revert to mean, close position
56            None - no action
57        """
58        self.price_history.append(price)
59        
60        # Limit history size
61        if len(self.price_history) > self.lookback_period * 2:
62            self.price_history = self.price_history[-self.lookback_period * 2:]
63        
64        zscore = self.calculate_zscore(price)
65        
66        # Check if should close existing position
67        if self.position:
68            holding_hours = (timestamp - self.position.entry_time).total_seconds() / 3600
69            
70            # Exit conditions
71            if holding_hours > self.max_holding_period_hours:
72                return 'CLOSE'
73            
74            if self.position.quantity > 0:  # Long position
75                # Exit if reverted to mean or profitable
76                if zscore > -self.exit_threshold or price > self.position.entry_price * 1.01:
77                    return 'CLOSE'
78            
79            elif self.position.quantity < 0:  # Short position
80                if zscore < self.exit_threshold or price < self.position.entry_price * 0.99:
81                    return 'CLOSE'
82        
83        # Entry signals (only if no position)
84        if not self.position:
85            if zscore < -self.entry_threshold:
86                # Price significantly below mean -> BUY
87                return 'BUY'
88            elif zscore > self.entry_threshold:
89                # Price significantly above mean -> SELL (short)
90                return 'SELL'
91        
92        return None
93    
94    def execute_signal(
95        self,
96        signal: str,
97        symbol: str,
98        price: float,
99        timestamp: pd.Timestamp,
100        position_size: int = 100
101    ):
102        """Execute trading signal."""
103        if signal == 'BUY':
104            self.position = Position(
105                symbol=symbol,
106                quantity=position_size,
107                entry_price=price,
108                entry_time=timestamp
109            )
110            print(f"[{timestamp}] BUY {position_size} {symbol} @ {price:.2f}")
111        
112        elif signal == 'SELL':
113            self.position = Position(
114                symbol=symbol,
115                quantity=-position_size,  # Negative for short
116                entry_price=price,
117                entry_time=timestamp
118            )
119            print(f"[{timestamp}] SELL {position_size} {symbol} @ {price:.2f}")
120        
121        elif signal == 'CLOSE' and self.position:
122            pnl = (price - self.position.entry_price) * self.position.quantity
123            holding_time = timestamp - self.position.entry_time
124            
125            print(f"[{timestamp}] CLOSE {abs(self.position.quantity)} {symbol} @ {price:.2f}, "
126                  f"P&L: ${pnl:.2f}, holding: {holding_time}")
127            
128            self.position = None
129
130# Example usage
131strategy = MeanReversionStrategy(
132    lookback_period=20,
133    entry_threshold=2.0,
134    exit_threshold=0.5,
135    max_holding_period_hours=24
136)
137
138# Simulate price data
139np.random.seed(42)
140prices = 100 + np.cumsum(np.random.randn(100) * 0.5)  # Random walk
141
142for i, price in enumerate(prices):
143    timestamp = pd.Timestamp('2024-01-01') + pd.Timedelta(hours=i)
144    
145    signal = strategy.generate_signal('AAPL', price, timestamp)
146    if signal:
147        strategy.execute_signal(signal, 'AAPL', price, timestamp, position_size=100)
148

1class MomentumStrategy:
2    def __init__(
3        self,
4        fast_period: int = 10,
5        slow_period: int = 30,
6        atr_period: int = 14,
7        atr_multiplier: float = 2.0
8    ):
9        self.fast_period = fast_period
10        self.slow_period = slow_period
11        self.atr_period = atr_period
12        self.atr_multiplier = atr_multiplier
13        
14        self.price_history = []
15        self.high_history = []
16        self.low_history = []
17        self.position: Optional[Position] = None
18    
19    def calculate_ema(self, prices: list, period: int) -> float:
20        """Calculate Exponential Moving Average."""
21        if len(prices) < period:
22            return np.mean(prices) if prices else 0.0
23        
24        recent = prices[-period:]
25        multiplier = 2 / (period + 1)
26        ema = recent[0]
27        
28        for price in recent[1:]:
29            ema = (price - ema) * multiplier + ema
30        
31        return ema
32    
33    def calculate_atr(self) -> float:
34        """Calculate Average True Range (volatility)."""
35        if len(self.high_history) < self.atr_period:
36            return 0.0
37        
38        true_ranges = []
39        for i in range(1, len(self.high_history)):
40            high = self.high_history[i]
41            low = self.low_history[i]
42            prev_close = self.price_history[i - 1]
43            
44            tr = max(
45                high - low,
46                abs(high - prev_close),
47                abs(low - prev_close)
48            )
49            true_ranges.append(tr)
50        
51        return np.mean(true_ranges[-self.atr_period:])
52    
53    def generate_signal(
54        self,
55        symbol: str,
56        price: float,
57        high: float,
58        low: float,
59        timestamp: pd.Timestamp
60    ) -> Optional[str]:
61        """Generate momentum signal based on EMA crossover."""
62        self.price_history.append(price)
63        self.high_history.append(high)
64        self.low_history.append(low)
65        
66        # Need enough data
67        if len(self.price_history) < self.slow_period:
68            return None
69        
70        fast_ema = self.calculate_ema(self.price_history, self.fast_period)
71        slow_ema = self.calculate_ema(self.price_history, self.slow_period)
72        atr = self.calculate_atr()
73        
74        # Previous EMAs for crossover detection
75        prev_fast_ema = self.calculate_ema(
76            self.price_history[:-1], self.fast_period
77        )
78        prev_slow_ema = self.calculate_ema(
79            self.price_history[:-1], self.slow_period
80        )
81        
82        # Check for position management
83        if self.position:
84            # Trailing stop using ATR
85            if self.position.quantity > 0:  # Long
86                stop_price = price - (atr * self.atr_multiplier)
87                if price < stop_price:
88                    return 'CLOSE'
89                
90                # Exit on trend reversal
91                if fast_ema < slow_ema:
92                    return 'CLOSE'
93            
94            elif self.position.quantity < 0:  # Short
95                stop_price = price + (atr * self.atr_multiplier)
96                if price > stop_price:
97                    return 'CLOSE'
98                
99                if fast_ema > slow_ema:
100                    return 'CLOSE'
101        
102        # Entry signals (only if no position)
103        if not self.position:
104            # Bullish crossover
105            if prev_fast_ema <= prev_slow_ema and fast_ema > slow_ema:
106                return 'BUY'
107            
108            # Bearish crossover
109            elif prev_fast_ema >= prev_slow_ema and fast_ema < slow_ema:
110                return 'SELL'
111        
112        return None
113    
114    def calculate_position_size(self, price: float, account_value: float) -> int:
115        """
116        Calculate position size using ATR-based risk management.
117        
118        Risk 1% of account per trade.
119        """
120        atr = self.calculate_atr()
121        if atr == 0:
122            return 100  # Default size
123        
124        # Risk amount
125        risk_amount = account_value * 0.01
126        
127        # Stop distance = ATR * multiplier
128        stop_distance = atr * self.atr_multiplier
129        
130        # Shares = risk_amount / stop_distance
131        shares = int(risk_amount / stop_distance)
132        
133        return max(shares, 100)  # Minimum 100 shares
134
135# Usage
136strategy = MomentumStrategy(
137    fast_period=10,
138    slow_period=30,
139    atr_period=14,
140    atr_multiplier=2.0
141)
142
143# Simulate trending price data
144trend = np.linspace(0, 20, 100)
145noise = np.random.randn(100) * 0.5
146prices = 100 + trend + noise
147
148for i in range(len(prices)):
149    price = prices[i]
150    high = price + abs(np.random.randn() * 0.3)
151    low = price - abs(np.random.randn() * 0.3)
152    timestamp = pd.Timestamp('2024-01-01') + pd.Timedelta(hours=i)
153    
154    signal = strategy.generate_signal('AAPL', price, high, low, timestamp)
155    if signal:
156        position_size = strategy.calculate_position_size(price, account_value=100000)
157        strategy.position = Position('AAPL', position_size if signal == 'BUY' else -position_size, price, timestamp)
158        print(f"[{timestamp}] {signal} {position_size} @ {price:.2f}")
159

1from scipy import stats
2
3class PairsTrading:
4    """
5    Pairs trading: go long undervalued security, short overvalued.
6    """
7    
8    def __init__(
9        self,
10        lookback_period: int = 60,
11        entry_threshold: float = 2.0,
12        exit_threshold: float = 0.5
13    ):
14        self.lookback_period = lookback_period
15        self.entry_threshold = entry_threshold
16        self.exit_threshold = exit_threshold
17        
18        self.stock1_prices = []
19        self.stock2_prices = []
20        
21        self.position_stock1 = 0
22        self.position_stock2 = 0
23        self.entry_spread = None
24    
25    def calculate_hedge_ratio(self) -> float:
26        """Calculate hedge ratio using linear regression."""
27        if len(self.stock1_prices) < self.lookback_period:
28            return 1.0
29        
30        recent1 = self.stock1_prices[-self.lookback_period:]
31        recent2 = self.stock2_prices[-self.lookback_period:]
32        
33        # Regression: stock2 = beta * stock1 + alpha
34        slope, intercept, r_value, p_value, std_err = stats.linregress(
35            recent1, recent2
36        )
37        
38        return slope
39    
40    def calculate_spread(self, price1: float, price2: float) -> float:
41        """Calculate spread between stock prices."""
42        hedge_ratio = self.calculate_hedge_ratio()
43        
44        # Spread = stock2 - (hedge_ratio * stock1)
45        return price2 - (hedge_ratio * price1)
46    
47    def calculate_spread_zscore(self, current_spread: float) -> float:
48        """Calculate z-score of current spread."""
49        if len(self.stock1_prices) < self.lookback_period:
50            return 0.0
51        
52        # Calculate historical spreads
53        spreads = []
54        for i in range(self.lookback_period):
55            idx = len(self.stock1_prices) - self.lookback_period + i
56            if idx >= 0:
57                p1 = self.stock1_prices[idx]
58                p2 = self.stock2_prices[idx]
59                spreads.append(self.calculate_spread(p1, p2))
60        
61        mean_spread = np.mean(spreads)
62        std_spread = np.std(spreads)
63        
64        if std_spread == 0:
65            return 0.0
66        
67        return (current_spread - mean_spread) / std_spread
68    
69    def generate_signal(
70        self,
71        stock1_symbol: str,
72        stock2_symbol: str,
73        stock1_price: float,
74        stock2_price: float,
75        timestamp: pd.Timestamp
76    ) -> Optional[dict]:
77        """Generate pairs trading signal."""
78        self.stock1_prices.append(stock1_price)
79        self.stock2_prices.append(stock2_price)
80        
81        # Limit history
82        if len(self.stock1_prices) > self.lookback_period * 2:
83            self.stock1_prices = self.stock1_prices[-self.lookback_period * 2:]
84            self.stock2_prices = self.stock2_prices[-self.lookback_period * 2:]
85        
86        current_spread = self.calculate_spread(stock1_price, stock2_price)
87        zscore = self.calculate_spread_zscore(current_spread)
88        
89        # Check existing position
90        if self.position_stock1 != 0:
91            # Exit if spread reverts to mean
92            if abs(zscore) < self.exit_threshold:
93                signal = {
94                    'action': 'CLOSE',
95                    'stock1': stock1_symbol,
96                    'stock2': stock2_symbol,
97                    'zscore': zscore
98                }
99                
100                self.position_stock1 = 0
101                self.position_stock2 = 0
102                self.entry_spread = None
103                
104                return signal
105        
106        # Entry signals
107        else:
108            hedge_ratio = self.calculate_hedge_ratio()
109            
110            if zscore > self.entry_threshold:
111                # Spread too high -> short spread
112                # Short stock2, long stock1
113                signal = {
114                    'action': 'ENTER_SHORT_SPREAD',
115                    'stock1': stock1_symbol,
116                    'stock2': stock2_symbol,
117                    'stock1_action': 'BUY',
118                    'stock2_action': 'SELL',
119                    'hedge_ratio': hedge_ratio,
120                    'zscore': zscore
121                }
122                
123                self.position_stock1 = 100
124                self.position_stock2 = -int(100 * hedge_ratio)
125                self.entry_spread = current_spread
126                
127                return signal
128            
129            elif zscore < -self.entry_threshold:
130                # Spread too low -> long spread
131                # Long stock2, short stock1
132                signal = {
133                    'action': 'ENTER_LONG_SPREAD',
134                    'stock1': stock1_symbol,
135                    'stock2': stock2_symbol,
136                    'stock1_action': 'SELL',
137                    'stock2_action': 'BUY',
138                    'hedge_ratio': hedge_ratio,
139                    'zscore': zscore
140                }
141                
142                self.position_stock1 = -100
143                self.position_stock2 = int(100 * hedge_ratio)
144                self.entry_spread = current_spread
145                
146                return signal
147        
148        return None
149
150# Usage - correlated stocks
151pairs = PairsTrading(lookback_period=60, entry_threshold=2.0, exit_threshold=0.5)
152
153# Simulate correlated prices
154base = np.cumsum(np.random.randn(200) * 0.5) + 100
155stock1_prices = base + np.random.randn(200) * 0.2
156stock2_prices = base * 1.5 + np.random.randn(200) * 0.3
157
158for i in range(len(stock1_prices)):
159    timestamp = pd.Timestamp('2024-01-01') + pd.Timedelta(hours=i)
160    
161    signal = pairs.generate_signal(
162        'KO', 'PEP',
163        stock1_prices[i],
164        stock2_prices[i],
165        timestamp
166    )
167    
168    if signal:
169        print(f"[{timestamp}] {signal['action']}, z-score: {signal['zscore']:.2f}")
170        if 'hedge_ratio' in signal:
171            print(f"  Hedge ratio: {signal['hedge_ratio']:.3f}")
172

1class CapacityAnalyzer:
2    """Analyze strategy capacity and market impact."""
3    
4    def __init__(self, strategy_name: str):
5        self.strategy_name = strategy_name
6        self.historical_trades = []
7    
8    def add_trade(
9        self,
10        symbol: str,
11        quantity: int,
12        price: float,
13        slippage_bps: float
14    ):
15        """Record trade with slippage."""
16        self.historical_trades.append({
17            'symbol': symbol,
18            'quantity': quantity,
19            'price': price,
20            'slippage_bps': slippage_bps
21        })
22    
23    def estimate_market_impact(
24        self,
25        quantity: int,
26        avg_daily_volume: int
27    ) -> float:
28        """
29        Estimate market impact.
30        
31        Impact = k * (quantity / ADV) ^ 0.5
32        where k ~ 0.1 for liquid stocks
33        """
34        participation_rate = quantity / avg_daily_volume
35        
36        # Square root model
37        impact_bps = 10 * np.sqrt(participation_rate)
38        
39        return impact_bps
40    
41    def estimate_capacity(
42        self,
43        target_symbols: list,
44        avg_daily_volumes: dict,
45        max_slippage_bps: float = 5.0,
46        max_participation_rate: float = 0.05
47    ) -> float:
48        """
49        Estimate strategy capacity.
50        
51        Args:
52            target_symbols: List of tradable symbols
53            avg_daily_volumes: Dict mapping symbol to ADV
54            max_slippage_bps: Max acceptable slippage
55            max_participation_rate: Max % of daily volume
56        
57        Returns:
58            Estimated capacity in dollars
59        """
60        capacity_per_symbol = {}
61        
62        for symbol in target_symbols:
63            adv = avg_daily_volumes.get(symbol, 0)
64            if adv == 0:
65                continue
66            
67            # Max shares based on participation rate
68            max_shares_participation = int(adv * max_participation_rate)
69            
70            # Max shares based on slippage
71            # Solve: max_slippage = 10 * sqrt(shares / adv)
72            max_shares_slippage = int(
73                (max_slippage_bps / 10) ** 2 * adv
74            )
75            
76            # Use more conservative
77            max_shares = min(max_shares_participation, max_shares_slippage)
78            
79            # Assume average price $100 for capacity estimate
80            capacity_per_symbol[symbol] = max_shares * 100
81        
82        # Total capacity
83        total_capacity = sum(capacity_per_symbol.values())
84        
85        print(f"Strategy Capacity Analysis: {self.strategy_name}")
86        print(f"Number of symbols: {len(capacity_per_symbol)}")
87        print(f"Total capacity: ${total_capacity:,.0f}")
88        print(f"\nTop 5 symbols by capacity:")
89        
90        top_symbols = sorted(
91            capacity_per_symbol.items(),
92            key=lambda x: x[1],
93            reverse=True
94        )[:5]
95        
96        for symbol, capacity in top_symbols:
97            print(f"  {symbol}: ${capacity:,.0f}")
98        
99        return total_capacity
100
101# Usage
102analyzer = CapacityAnalyzer("Mean Reversion Strategy")
103
104# Example: S&P 500 stocks
105symbols = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'TSLA']
106avg_daily_volumes = {
107    'AAPL': 50_000_000,
108    'MSFT': 25_000_000,
109    'GOOGL': 20_000_000,
110    'AMZN': 30_000_000,
111    'TSLA': 100_000_000,
112}
113
114capacity = analyzer.estimate_capacity(
115    symbols,
116    avg_daily_volumes,
117    max_slippage_bps=5.0,
118    max_participation_rate=0.05
119)
120

1import asyncio
2from typing import Callable
3from enum import Enum
4
5class OrderStatus(Enum):
6    PENDING = "PENDING"
7    SUBMITTED = "SUBMITTED"
8    FILLED = "FILLED"
9    PARTIAL_FILL = "PARTIAL_FILL"
10    CANCELLED = "CANCELLED"
11    REJECTED = "REJECTED"
12
13@dataclass
14class Order:
15    order_id: str
16    symbol: str
17    side: str
18    quantity: int
19    price: Optional[float]
20    status: OrderStatus
21    filled_quantity: int = 0
22    avg_fill_price: float = 0.0
23
24class TradingEngine:
25    """Production trading engine."""
26    
27    def __init__(
28        self,
29        strategy: Callable,
30        risk_limits: dict,
31        execution_venue: str = "SIMULATION"
32    ):
33        self.strategy = strategy
34        self.risk_limits = risk_limits
35        self.execution_venue = execution_venue
36        
37        self.orders: dict[str, Order] = {}
38        self.positions: dict[str, int] = {}
39        self.cash = risk_limits.get('max_capital', 1_000_000)
40        
41        self.order_counter = 0
42        
43        logging.basicConfig(level=logging.INFO)
44        self.logger = logging.getLogger(__name__)
45    
46    def check_risk_limits(
47        self,
48        symbol: str,
49        side: str,
50        quantity: int,
51        price: float
52    ) -> tuple[bool, str]:
53        """Check if order passes risk limits."""
54        # Position limit
55        current_position = self.positions.get(symbol, 0)
56        new_position = current_position + (quantity if side == 'BUY' else -quantity)
57        
58        max_position = self.risk_limits.get('max_position_size', 10000)
59        if abs(new_position) > max_position:
60            return False, f"Position limit exceeded: {abs(new_position)} > {max_position}"
61        
62        # Capital requirement
63        order_value = quantity * price
64        if side == 'BUY' and order_value > self.cash:
65            return False, f"Insufficient capital: need ${order_value:.2f}, have ${self.cash:.2f}"
66        
67        # Single order size limit
68        max_order_value = self.risk_limits.get('max_order_value', 100_000)
69        if order_value > max_order_value:
70            return False, f"Order value ${order_value:.2f} exceeds limit ${max_order_value:.2f}"
71        
72        return True, "OK"
73    
74    def submit_order(
75        self,
76        symbol: str,
77        side: str,
78        quantity: int,
79        price: Optional[float] = None
80    ) -> Optional[str]:
81        """Submit order to execution venue."""
82        # Risk check
83        check_price = price if price else 100.0  # Use reasonable estimate if market order
84        passed, message = self.check_risk_limits(symbol, side, quantity, check_price)
85        
86        if not passed:
87            self.logger.warning(f"Order rejected: {message}")
88            return None
89        
90        # Generate order ID
91        self.order_counter += 1
92        order_id = f"ORD{self.order_counter:06d}"
93        
94        # Create order
95        order = Order(
96            order_id=order_id,
97            symbol=symbol,
98            side=side,
99            quantity=quantity,
100            price=price,
101            status=OrderStatus.SUBMITTED
102        )
103        
104        self.orders[order_id] = order
105        
106        self.logger.info(
107            f"Order submitted: {order_id} {side} {quantity} {symbol} @ "
108            f"{price if price else 'MARKET'}"
109        )
110        
111        return order_id
112    
113    def handle_fill(
114        self,
115        order_id: str,
116        fill_quantity: int,
117        fill_price: float
118    ):
119        """Handle order fill."""
120        if order_id not in self.orders:
121            self.logger.error(f"Unknown order ID: {order_id}")
122            return
123        
124        order = self.orders[order_id]
125        
126        # Update order
127        order.filled_quantity += fill_quantity
128        order.avg_fill_price = (
129            (order.avg_fill_price * (order.filled_quantity - fill_quantity) +
130             fill_price * fill_quantity) / order.filled_quantity
131        )
132        
133        if order.filled_quantity >= order.quantity:
134            order.status = OrderStatus.FILLED
135        else:
136            order.status = OrderStatus.PARTIAL_FILL
137        
138        # Update position
139        current_position = self.positions.get(order.symbol, 0)
140        if order.side == 'BUY':
141            self.positions[order.symbol] = current_position + fill_quantity
142            self.cash -= fill_quantity * fill_price
143        else:
144            self.positions[order.symbol] = current_position - fill_quantity
145            self.cash += fill_quantity * fill_price
146        
147        self.logger.info(
148            f"Fill: {order_id} {fill_quantity} @ {fill_price:.2f}, "
149            f"position: {self.positions[order.symbol]}, cash: ${self.cash:.2f}"
150        )
151    
152    async def run(self, market_data_stream):
153        """Run trading engine."""
154        self.logger.info(f"Starting trading engine with ${self.cash:,.0f} capital")
155        
156        async for market_data in market_data_stream:
157            # Generate signal from strategy
158            signal = self.strategy.generate_signal(
159                symbol=market_data['symbol'],
160                price=market_data['price'],
161                timestamp=market_data['timestamp']
162            )
163            
164            if signal:
165                # Execute signal
166                if signal == 'BUY':
167                    order_id = self.submit_order(
168                        market_data['symbol'],
169                        'BUY',
170                        100,
171                        market_data['price']
172                    )
173                    
174                    if order_id:
175                        # Simulate fill (in production, wait for exchange)
176                        self.handle_fill(order_id, 100, market_data['price'])
177                
178                elif signal == 'SELL':
179                    order_id = self.submit_order(
180                        market_data['symbol'],
181                        'SELL',
182                        100,
183                        market_data['price']
184                    )
185                    
186                    if order_id:
187                        self.handle_fill(order_id, 100, market_data['price'])
188            
189            # Small delay
190            await asyncio.sleep(0.1)
191

1class PerformanceAttribution:
2    """Attribute strategy returns to different factors."""
3    
4    def __init__(self):
5        self.trades = []
6    
7    def add_trade(
8        self,
9        symbol: str,
10        entry_price: float,
11        exit_price: float,
12        quantity: int,
13        hold_time_hours: float,
14        strategy_signal: str
15    ):
16        """Record completed trade."""
17        pnl = (exit_price - entry_price) * quantity
18        return_pct = (exit_price / entry_price - 1) * 100
19        
20        self.trades.append({
21            'symbol': symbol,
22            'entry_price': entry_price,
23            'exit_price': exit_price,
24            'quantity': quantity,
25            'pnl': pnl,
26            'return_pct': return_pct,
27            'hold_time_hours': hold_time_hours,
28            'strategy_signal': strategy_signal
29        })
30    
31    def analyze(self) -> dict:
32        """Perform attribution analysis."""
33        if not self.trades:
34            return {}
35        
36        df = pd.DataFrame(self.trades)
37        
38        # Overall metrics
39        total_pnl = df['pnl'].sum()
40        win_rate = (df['pnl'] > 0).mean()
41        avg_win = df[df['pnl'] > 0]['pnl'].mean() if any(df['pnl'] > 0) else 0
42        avg_loss = df[df['pnl'] < 0]['pnl'].mean() if any(df['pnl'] < 0) else 0
43        
44        # Attribution by strategy signal
45        signal_attribution = df.groupby('strategy_signal').agg({
46            'pnl': ['sum', 'count', 'mean'],
47            'return_pct': 'mean'
48        }).round(2)
49        
50        # Attribution by holding period
51        df['hold_bucket'] = pd.cut(
52            df['hold_time_hours'],
53            bins=[0, 1, 4, 24, float('inf')],
54            labels=['<1h', '1-4h', '4-24h', '>24h']
55        )
56        
57        time_attribution = df.groupby('hold_bucket').agg({
58            'pnl': ['sum', 'count'],
59            'return_pct': 'mean'
60        }).round(2)
61        
62        # Win/loss attribution
63        df['outcome'] = df['pnl'].apply(lambda x: 'Win' if x > 0 else 'Loss')
64        outcome_attribution = df.groupby('outcome').agg({
65            'pnl': ['sum', 'count', 'mean']
66        }).round(2)
67        
68        print("=== Performance Attribution ===\n")
69        print(f"Total P&L: ${total_pnl:,.2f}")
70        print(f"Win Rate: {win_rate:.1%}")
71        print(f"Avg Win: ${avg_win:,.2f}")
72        print(f"Avg Loss: ${avg_loss:,.2f}")
73        print(f"Profit Factor: {abs(avg_win/avg_loss) if avg_loss != 0 else 0:.2f}\n")
74        
75        print("Attribution by Signal:")
76        print(signal_attribution)
77        print("\nAttribution by Hold Time:")
78        print(time_attribution)
79        print("\nAttribution by Outcome:")
80        print(outcome_attribution)
81        
82        return {
83            'total_pnl': total_pnl,
84            'win_rate': win_rate,
85            'avg_win': avg_win,
86            'avg_loss': avg_loss,
87            'signal_attribution': signal_attribution,
88            'time_attribution': time_attribution
89        }
90
91# Usage
92attribution = PerformanceAttribution()
93
94# Add example trades
95attribution.add_trade('AAPL', 150.0, 152.0, 100, 2.5, 'MOMENTUM_BUY')
96attribution.add_trade('MSFT', 300.0, 298.0, 50, 1.2, 'MEAN_REVERT_BUY')
97attribution.add_trade('GOOGL', 140.0, 143.0, 75, 4.8, 'MOMENTUM_BUY')
98attribution.add_trade('AMZN', 180.0, 179.0, 60, 0.8, 'MEAN_REVERT_SELL')
99
100results = attribution.analyze()
101

1Overall Portfolio:
2- AUM: $450M
3- Strategies: 8 active
4- Sharpe Ratio: 2.1
5- Max Drawdown: -8.3%
6- Win Rate: 62%
7- Avg Trade Duration: 2.3 hours
8- Annual Return: 18.4%
9
10Strategy Breakdown:
111. Mean Reversion (35% allocation):
12   - Sharpe: 2.4
13   - Win rate: 68%
14   - Avg hold: 4.2 hours
15   - Capacity: $180M
16
172. Momentum (25% allocation):
18   - Sharpe: 1.9
19   - Win rate: 58%
20   - Avg hold: 8.5 hours
21   - Capacity: $120M
22
233. Statistical Arbitrage (20% allocation):
24   - Sharpe: 2.7
25   - Win rate: 71%
26   - Avg hold: 1.8 hours
27   - Capacity: $90M
28
294. Others (20% allocation):
30   - Various alpha strategies
31   - Sharpe: 1.6
32   - Capacity: $60M
33

1Order Execution:
2- Avg fill rate: 96.2%
3- Avg slippage: 0.8 bps
4- Avg latency: 2.3ms (order to fill)
5- Rejected orders: 0.3%
6
7Risk Management:
8- Position limit breaches: 0
9- Margin calls: 0
10- Max intraday drawdown: -2.1%
11- Avg daily VaR: $320k (99% confidence)
12