Build Real-Time TCA for Gold Algorithms in 45 Minutes

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from collections import deque

class SpreadMonitor:
    def __init__(self, window_size=300):
        # Personal note: 300 samples = 30 seconds at 100ms sampling
        self.spreads = deque(maxlen=window_size)
        self.timestamps = deque(maxlen=window_size)
        
    def update(self, bid, ask):
        """Update with new quote"""
        spread_bps = ((ask - bid) / bid) * 10000  # basis points
        self.spreads.append(spread_bps)
        self.timestamps.append(datetime.now())
        
    def get_current_stats(self):
        """Calculate real-time spread metrics"""
        if len(self.spreads) < 10:
            return None
            
        spreads_array = np.array(self.spreads)
        
        return {
            'current_spread': spreads_array[-1],
            'avg_30s': np.mean(spreads_array),
            'std_30s': np.std(spreads_array),
            'max_30s': np.max(spreads_array),
            'percentile_95': np.percentile(spreads_array, 95),
            'is_elevated': spreads_array[-1] > np.mean(spreads_array) + 2 * np.std(spreads_array)
        }
        
    # Watch out: Don't use regular lists - deque is 10x faster for rolling windows

# Initialize monitor
spread_monitor = SpreadMonitor(window_size=300)

# Simulate incoming quotes (in production, connect to your broker API)
for _ in range(50):
    bid, ask = 2045.30, 2045.80  # Example gold prices
    spread_monitor.update(bid, ask)
    
stats = spread_monitor.get_current_stats()
print(f"Current spread: {stats['current_spread']:.2f} bps")
print(f"30s average: {stats['avg_30s']:.2f} bps")
print(f"Alert: {'ELEVATED COSTS' if stats['is_elevated'] else 'Normal'}")

class SlippageEstimator:
    def __init__(self, base_slippage_bps=0.5):
        self.base_slippage = base_slippage_bps
        self.recent_fills = deque(maxlen=100)
        
    def estimate_slippage(self, order_size_oz, current_volatility, depth_imbalance):
        """
        Calculate expected slippage in basis points
        
        Parameters:
        - order_size_oz: Size in troy ounces
        - current_volatility: ATR in dollars (14-period)
        - depth_imbalance: (bid_volume - ask_volume) / total_volume
        """
        
        # Size impact (nonlinear - learned this from real fills)
        size_multiplier = 1 + (order_size_oz / 10) ** 1.3
        
        # Volatility impact (higher vol = wider spreads = more slippage)
        vol_multiplier = 1 + (current_volatility / 10)
        
        # Liquidity impact (buying into thin asks hurts)
        if order_size_oz > 0:  # Buy order
            liquidity_penalty = max(0, -depth_imbalance * 2)
        else:  # Sell order
            liquidity_penalty = max(0, depth_imbalance * 2)
            
        estimated_slippage = (
            self.base_slippage * 
            size_multiplier * 
            vol_multiplier * 
            (1 + liquidity_penalty)
        )
        
        return estimated_slippage
    
    def record_actual_fill(self, expected_price, fill_price, order_size_oz):
        """Track actual slippage to improve estimates"""
        actual_slippage_bps = abs((fill_price - expected_price) / expected_price) * 10000
        self.recent_fills.append({
            'expected': expected_price,
            'actual': fill_price,
            'slippage_bps': actual_slippage_bps,
            'size': abs(order_size_oz)
        })
        
        # Adjust base slippage based on recent performance
        if len(self.recent_fills) >= 20:
            avg_actual = np.mean([f['slippage_bps'] for f in self.recent_fills])
            self.base_slippage = 0.7 * self.base_slippage + 0.3 * avg_actual
            
    # Watch out: Don't estimate slippage once per day - it changes every minute

# Example usage
slippage_est = SlippageEstimator(base_slippage_bps=0.5)

# Scenario: Want to buy 15 oz gold during elevated volatility
order_size = 15  # troy ounces
current_atr = 12.50  # dollars
depth_imbalance = -0.15  # More offers than bids (bad for buying)

estimated_slip = slippage_est.estimate_slippage(order_size, current_atr, depth_imbalance)
print(f"Estimated slippage: {estimated_slip:.2f} bps")
print(f"On 15 oz @ $2045: ${15 * 2045 * estimated_slip / 10000:.2f} cost")

from dataclasses import dataclass
from typing import Optional

@dataclass
class TCAResult:
    """Complete transaction cost breakdown"""
    spread_cost_bps: float
    slippage_cost_bps: float
    timing_cost_bps: float
    total_cost_bps: float
    total_cost_usd: float
    recommendation: str
    confidence: float

class TCAEngine:
    def __init__(self, spread_monitor: SpreadMonitor, 
                 slippage_estimator: SlippageEstimator):
        self.spread_monitor = spread_monitor
        self.slippage_estimator = slippage_estimator
        
    def analyze_trade(self, 
                     order_size_oz: float,
                     target_price: float,
                     current_bid: float,
                     current_ask: float,
                     current_volatility: float,
                     depth_imbalance: float,
                     urgency: str = 'normal') -> TCAResult:
        """
        Complete pre-trade cost analysis
        
        urgency: 'low' (patient), 'normal', 'high' (urgent)
        """
        
        # 1. Spread cost (guaranteed)
        spread_stats = self.spread_monitor.get_current_stats()
        if order_size_oz > 0:  # Buy order
            spread_cost_bps = ((current_ask - target_price) / target_price) * 10000
        else:  # Sell order
            spread_cost_bps = ((target_price - current_bid) / target_price) * 10000
            
        # 2. Slippage cost (estimated)
        slippage_cost_bps = self.slippage_estimator.estimate_slippage(
            order_size_oz, current_volatility, depth_imbalance
        )
        
        # 3. Timing cost (opportunity cost of waiting)
        timing_cost_bps = self._calculate_timing_cost(urgency, current_volatility)
        
        # Total cost
        total_cost_bps = spread_cost_bps + slippage_cost_bps + timing_cost_bps
        total_cost_usd = abs(order_size_oz) * target_price * total_cost_bps / 10000
        
        # Generate recommendation
        recommendation = self._generate_recommendation(
            total_cost_bps, spread_stats, urgency
        )
        
        # Confidence based on data quality
        confidence = self._calculate_confidence(spread_stats)
        
        return TCAResult(
            spread_cost_bps=spread_cost_bps,
            slippage_cost_bps=slippage_cost_bps,
            timing_cost_bps=timing_cost_bps,
            total_cost_bps=total_cost_bps,
            total_cost_usd=total_cost_usd,
            recommendation=recommendation,
            confidence=confidence
        )
    
    def _calculate_timing_cost(self, urgency: str, volatility: float) -> float:
        """Opportunity cost of waiting for better execution"""
        if urgency == 'high':
            return 0  # Execute immediately
        elif urgency == 'normal':
            return volatility * 0.1  # Small cost for standard orders
        else:  # low urgency
            return volatility * 0.3  # Higher cost for patient orders
            
    def _generate_recommendation(self, total_cost_bps: float, 
                                 spread_stats: dict, urgency: str) -> str:
        """Actionable trading recommendation"""
        
        # Learned thresholds from 6 months of live trading
        if total_cost_bps < 2.0:
            return "EXECUTE - Low cost environment"
        elif total_cost_bps < 4.0:
            if urgency == 'high':
                return "EXECUTE - Acceptable given urgency"
            else:
                return "WAIT - Costs elevated, consider delaying"
        else:
            if spread_stats and spread_stats['is_elevated']:
                return "HOLD - Spreads abnormally wide, wait for normalization"
            else:
                return "REDUCE SIZE - High costs, consider smaller order"
                
    def _calculate_confidence(self, spread_stats: Optional[dict]) -> float:
        """Confidence in cost estimates (0-1)"""
        if not spread_stats or len(self.spread_monitor.spreads) < 50:
            return 0.5  # Limited data
        
        # Higher confidence when spreads are stable
        stability = 1 - (spread_stats['std_30s'] / spread_stats['avg_30s'])
        return max(0.5, min(0.95, stability))

# Example: Analyze a trade before execution
tca_engine = TCAEngine(spread_monitor, slippage_est)

result = tca_engine.analyze_trade(
    order_size_oz=20,
    target_price=2045.50,
    current_bid=2045.30,
    current_ask=2045.80,
    current_volatility=12.50,
    depth_imbalance=-0.15,
    urgency='normal'
)

print(f"\n=== TCA Analysis ===")
print(f"Spread cost:  {result.spread_cost_bps:.2f} bps")
print(f"Slippage:     {result.slippage_cost_bps:.2f} bps")
print(f"Timing cost:  {result.timing_cost_bps:.2f} bps")
print(f"Total cost:   {result.total_cost_bps:.2f} bps (${result.total_cost_usd:.2f})")
print(f"\nRecommendation: {result.recommendation}")
print(f"Confidence: {result.confidence:.0%}")

class GoldTradingAlgorithm:
    def __init__(self, tca_engine: TCAEngine, max_cost_bps: float = 5.0):
        self.tca_engine = tca_engine
        self.max_cost_bps = max_cost_bps
        self.pending_orders = []
        
    def generate_signal(self, market_data: dict) -> Optional[dict]:
        """Your existing alpha generation logic"""
        # ... your strategy code ...
        # Returns: {'side': 'buy'/'sell', 'size': oz, 'urgency': str}
        pass
    
    def execute_with_tca(self, signal: dict, market_data: dict):
        """TCA-aware execution logic"""
        
        # Run TCA analysis
        tca_result = self.tca_engine.analyze_trade(
            order_size_oz=signal['size'] if signal['side'] == 'buy' else -signal['size'],
            target_price=market_data['mid_price'],
            current_bid=market_data['bid'],
            current_ask=market_data['ask'],
            current_volatility=market_data['atr'],
            depth_imbalance=market_data['depth_imbalance'],
            urgency=signal['urgency']
        )
        
        # Decision logic based on TCA
        if "EXECUTE" in tca_result.recommendation:
            # Go ahead with order
            self._send_order(signal, market_data)
            print(f"âœ" Executed {signal['side']} {signal['size']} oz (cost: {tca_result.total_cost_bps:.2f} bps)")
            
        elif "WAIT" in tca_result.recommendation:
            # Queue for better conditions
            self.pending_orders.append({
                'signal': signal,
                'queued_at': datetime.now(),
                'max_wait_minutes': 5
            })
            print(f"⏸ Queued {signal['side']} order - costs elevated")
            
        elif "REDUCE SIZE" in tca_result.recommendation:
            # Split into smaller chunks
            reduced_size = signal['size'] * 0.5
            signal['size'] = reduced_size
            self._send_order(signal, market_data)
            print(f"âš  Reduced size to {reduced_size} oz due to high costs")
            
        elif "HOLD" in tca_result.recommendation:
            # Abort - conditions too poor
            print(f"✗ Skipped trade - spreads abnormal ({tca_result.spread_cost_bps:.2f} bps)")
            
        return tca_result
    
    def check_pending_orders(self, market_data: dict):
        """Retry queued orders when costs improve"""
        for order in self.pending_orders[:]:
            # Check if costs have improved
            tca_result = self.tca_engine.analyze_trade(
                order_size_oz=order['signal']['size'],
                target_price=market_data['mid_price'],
                current_bid=market_data['bid'],
                current_ask=market_data['ask'],
                current_volatility=market_data['atr'],
                depth_imbalance=market_data['depth_imbalance'],
                urgency='low'  # Patient on queued orders
            )
            
            if tca_result.total_cost_bps < self.max_cost_bps:
                self._send_order(order['signal'], market_data)
                self.pending_orders.remove(order)
                print(f"âœ" Executed pending order (improved cost: {tca_result.total_cost_bps:.2f} bps)")
                
            # Timeout check
            elif (datetime.now() - order['queued_at']).seconds > order['max_wait_minutes'] * 60:
                self.pending_orders.remove(order)
                print(f"⌛ Expired pending order after {order['max_wait_minutes']} minutes")
    
    def _send_order(self, signal: dict, market_data: dict):
        """Your broker integration here"""
        # ... actual order execution ...
        pass

# Watch out: Always run check_pending_orders() in your main loop

# Example integration
algo = GoldTradingAlgorithm(tca_engine, max_cost_bps=5.0)

# In your main trading loop
signal = {'side': 'buy', 'size': 25, 'urgency': 'normal'}
market_data = {
    'bid': 2045.30,
    'ask': 2045.80,
    'mid_price': 2045.55,
    'atr': 12.50,
    'depth_imbalance': -0.15
}

result = algo.execute_with_tca(signal, market_data)