Fix Latency Arbitrage in Gold Execution Systems Before Losing $50K

# audit_latency.py
# Personal note: Run this during market hours for realistic data
import time
from datetime import datetime
import redis
import statistics

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def measure_quote_staleness():
    """Check how old quotes are when executed"""
    staleness_samples = []
    
    # Sample 1000 recent executions
    executions = r.lrange('executions:gold', 0, 999)
    
    for exec_data in executions:
        parts = exec_data.split('|')
        quote_time = float(parts[1])  # Unix timestamp
        exec_time = float(parts[2])
        
        staleness_ms = (exec_time - quote_time) * 1000
        staleness_samples.append(staleness_ms)
    
    print(f"Median staleness: {statistics.median(staleness_samples):.1f}ms")
    print(f"95th percentile: {sorted(staleness_samples)[949]:.1f}ms")
    print(f"Max staleness: {max(staleness_samples):.1f}ms")
    
    # Watch out: Values over 500ms are red flags
    vulnerabilities = [s for s in staleness_samples if s > 500]
    print(f"\nVulnerable executions: {len(vulnerabilities)} ({len(vulnerabilities)/10:.1f}%)")
    
    return staleness_samples

if __name__ == "__main__":
    samples = measure_quote_staleness()

# quote_validator.py
# Personal note: I set 50ms after testing - adjust for your latency profile
import time
import redis
from typing import Optional, Tuple

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Max age in seconds (50ms = 0.050s)
MAX_QUOTE_AGE = 0.050

class QuoteValidator:
    def validate_execution(self, quote_id: str, client_id: str) -> Tuple[bool, str]:
        """
        Validates quote freshness before execution
        Returns: (is_valid, reason)
        """
        current_time = time.time()
        
        # Fetch quote with timestamp
        quote_data = r.get(f"quote:{quote_id}")
        
        if not quote_data:
            return False, "QUOTE_NOT_FOUND"
        
        # Parse quote: price|timestamp|instrument
        parts = quote_data.split('|')
        quote_timestamp = float(parts[1])
        
        age_seconds = current_time - quote_timestamp
        age_ms = age_seconds * 1000
        
        if age_seconds > MAX_QUOTE_AGE:
            # Log for audit trail
            r.lpush('rejected_stale', 
                   f"{client_id}|{quote_id}|{age_ms:.2f}|{current_time}")
            return False, f"QUOTE_STALE_{age_ms:.1f}ms"
        
        # Watch out: Check if quote already used (replay attack)
        if r.sismember('used_quotes', quote_id):
            return False, "QUOTE_ALREADY_USED"
        
        # Mark quote as used (expires in 60s)
        r.sadd('used_quotes', quote_id)
        r.expire('used_quotes', 60)
        
        return True, f"VALID_{age_ms:.2f}ms"

# Usage in execution handler
def execute_gold_order(quote_id: str, client_id: str, quantity: float):
    validator = QuoteValidator()
    is_valid, reason = validator.validate_execution(quote_id, client_id)
    
    if not is_valid:
        print(f"✗ Execution rejected: {reason}")
        return {"status": "REJECTED", "reason": reason}
    
    # Proceed with execution
    print(f"âœ" Execution validated: {reason}")
    # ... actual execution logic here ...
    
    return {"status": "FILLED", "validation_latency": reason.split('_')[1]}

# rollout_config.py
# Personal note: Started at 10% after testing in staging for 48 hours
import random
import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

class RolloutController:
    def __init__(self):
        # Start conservative
        self.rollout_percentage = self._get_current_percentage()
    
    def _get_current_percentage(self) -> int:
        """Fetch current rollout % from Redis"""
        pct = r.get('validation_rollout_pct')
        return int(pct) if pct else 0
    
    def should_validate(self, client_id: str) -> bool:
        """Deterministic decision based on client_id hash"""
        # Use consistent hashing so same client gets same behavior
        client_hash = hash(client_id) % 100
        return client_hash < self.rollout_percentage
    
    def increase_rollout(self, new_percentage: int):
        """Gradually increase validation coverage"""
        if new_percentage > 100:
            new_percentage = 100
        
        r.set('validation_rollout_pct', new_percentage)
        self.rollout_percentage = new_percentage
        
        print(f"Rollout increased to {new_percentage}%")
        print("Monitor for 30min before next increase")

# Integration with execution handler
def execute_with_rollout(quote_id: str, client_id: str, quantity: float):
    rollout = RolloutController()
    
    if rollout.should_validate(client_id):
        # New validation path
        validator = QuoteValidator()
        is_valid, reason = validator.validate_execution(quote_id, client_id)
        
        if not is_valid:
            # Log but don't block during rollout
            print(f"[ROLLOUT] Would reject: {reason}")
            r.hincrby('rollout_stats', 'would_reject', 1)
        else:
            r.hincrby('rollout_stats', 'would_accept', 1)
    
    # Existing execution logic (unchanged)
    # ... execute order ...
    
    return {"status": "FILLED"}

# Gradual increase schedule
# Day 1: 10% -> 25% -> 50%
# Day 2: 50% -> 75% -> 100%

# arbitrage_detector.py
# Personal note: Catches sophisticated arbitrageurs who work within 50ms window
import time
import redis
from collections import defaultdict
from dataclasses import dataclass
from typing import List

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

@dataclass
class ExecutionPattern:
    client_id: str
    executions: List[float]  # timestamps
    avg_interval_ms: float
    is_suspicious: bool

class ArbitrageDetector:
    # Thresholds from analyzing known HFT patterns
    RAPID_FIRE_THRESHOLD = 5  # executions
    TIME_WINDOW = 1.0  # within 1 second
    MIN_INTERVAL_MS = 15  # faster than human possible
    
    def analyze_execution_pattern(self, client_id: str) -> ExecutionPattern:
        """Check if client exhibits HFT characteristics"""
        # Get recent executions (last 60 seconds)
        lookback = time.time() - 60
        
        exec_times = []
        executions = r.lrange(f'client_execs:{client_id}', 0, -1)
        
        for exec_data in executions:
            timestamp = float(exec_data.split('|')[0])
            if timestamp > lookback:
                exec_times.append(timestamp)
        
        if len(exec_times) < self.RAPID_FIRE_THRESHOLD:
            return ExecutionPattern(client_id, exec_times, 0, False)
        
        # Calculate intervals
        intervals = []
        for i in range(1, len(exec_times)):
            interval_ms = (exec_times[i] - exec_times[i-1]) * 1000
            intervals.append(interval_ms)
        
        avg_interval = sum(intervals) / len(intervals)
        
        # Check for rapid-fire pattern
        rapid_count = sum(1 for iv in intervals if iv < self.MIN_INTERVAL_MS)
        
        is_suspicious = (
            rapid_count >= 3 or  # Multiple sub-15ms intervals
            (len(exec_times) >= 10 and avg_interval < 50)  # Sustained HFT speed
        )
        
        if is_suspicious:
            r.sadd('flagged_clients', client_id)
            r.lpush('suspicious_patterns', 
                   f"{client_id}|{time.time()}|{avg_interval:.2f}ms|{rapid_count}")
            print(f"âš ï¸ Flagged {client_id}: {rapid_count} rapid executions, "
                  f"avg interval {avg_interval:.1f}ms")
        
        return ExecutionPattern(client_id, exec_times, avg_interval, is_suspicious)

# Integration with execution flow
def execute_with_detection(quote_id: str, client_id: str, quantity: float):
    # Run validation first
    validator = QuoteValidator()
    is_valid, reason = validator.validate_execution(quote_id, client_id)
    
    if not is_valid:
        return {"status": "REJECTED", "reason": reason}
    
    # Log execution for pattern analysis
    r.lpush(f'client_execs:{client_id}', f"{time.time()}|{quote_id}|{quantity}")
    r.expire(f'client_execs:{client_id}', 300)  # Keep 5min history
    
    # Analyze pattern (async in production)
    detector = ArbitrageDetector()
    pattern = detector.analyze_execution_pattern(client_id)
    
    if pattern.is_suspicious:
        # Flag but still execute (prevents false positives)
        r.hincrby('execution_flags', client_id, 1)
    
    # ... proceed with execution ...
    
    return {"status": "FILLED", "pattern_analysis": pattern}