signal decay and execution latency - my hidden edge killer

import asyncio
import time
import redis
import logging
import statistics
from dataclasses import dataclass, field
from typing import Optional, List, Dict

logger = logging.getLogger(__name__)


@dataclass
class SignalEvent:
    signal_id: str
    symbol: str
    strategy: str
    signal_type: str        # 'entry' | 'exit'
    expected_price: float
    expected_alpha: float   # expected return at signal time
    generated_at: float     # unix timestamp ms precision


@dataclass
class FillEvent:
    signal_id: str
    fill_price: float
    fill_qty: float
    filled_at: float        # unix timestamp ms precision
    venue: str              # 'home_san_diego' | 'colo_chicago'


@dataclass
class LatencyRecord:
    signal: SignalEvent
    fill: FillEvent
    signal_to_fill_ms: float
    price_slippage_bps: float
    alpha_captured: float   # fraction of expected alpha actually captured
    edge_lost_pct: float


class ExecutionLatencyTracker:
    """
    tracks signal-to-fill latency and measures alpha decay
    in real time across all execution venues.

    core insight: every ms of latency costs alpha.
    this class quantifies exactly how much edge is being left on the table.
    """

    # strategy-specific signal half-lives in ms
    # measured empirically: latency at which 50% alpha is gone
    SIGNAL_HALF_LIVES = {
        'crypto_momentum':    85,    # fast decay - competitive market
        'crypto_mean_rev':   450,    # slower - reversion persists longer
        'es_momentum':       120,    # moderate - futures HFT competition
        'nq_momentum':       130,
        'options_momentum':  350,    # slower - options flow is less reactive
        'options_vol_sell':  900,    # very slow - structural edge persists
    }

    def __init__(self, redis_client: redis.Redis, strategy_name: str):
        self.redis = redis_client
        self.strategy = strategy_name
        self.records: List[LatencyRecord] = []
        self.half_life_ms = self.SIGNAL_HALF_LIVES.get(strategy_name, 200)

    def record_signal(self, signal: SignalEvent) -> None:
        """log signal with microsecond timestamp for latency measurement"""
        key = f"signal:{signal.signal_id}"
        payload = {k: str(v) for k, v in {
            'symbol': signal.symbol,
            'strategy': signal.strategy,
            'signal_type': signal.signal_type,
            'expected_price': signal.expected_price,
            'expected_alpha': signal.expected_alpha,
            'generated_at': signal.generated_at
        }.items()}

        self.redis.hset(key, mapping=payload)
        self.redis.expire(key, 60)  # signals expire after 60s - stale if not filled

    async def record_fill(self, fill: FillEvent) -> Optional[LatencyRecord]:
        """match fill to originating signal, compute latency and alpha decay"""
        key = f"signal:{fill.signal_id}"
        raw = self.redis.hgetall(key)

        if not raw:
            logger.warning(f"signal {fill.signal_id} expired or missing - fill latency >60s")
            return None

        signal = SignalEvent(
            signal_id=fill.signal_id,
            symbol=raw[b'symbol'].decode(),
            strategy=raw[b'strategy'].decode(),
            signal_type=raw[b'signal_type'].decode(),
            expected_price=float(raw[b'expected_price']),
            expected_alpha=float(raw[b'expected_alpha']),
            generated_at=float(raw[b'generated_at'])
        )

        latency_ms = fill.filled_at - signal.generated_at
        price_slippage_bps = (
            abs(fill.fill_price - signal.expected_price) / signal.expected_price * 10000
        )

        # exponential decay model: alpha(t) = alpha_0 * 0.5^(t / half_life)
        alpha_captured = signal.expected_alpha * (0.5 ** (latency_ms / self.half_life_ms))
        edge_lost_pct = max(0, 1 - (alpha_captured / signal.expected_alpha))

        record = LatencyRecord(
            signal=signal,
            fill=fill,
            signal_to_fill_ms=latency_ms,
            price_slippage_bps=price_slippage_bps,
            alpha_captured=alpha_captured,
            edge_lost_pct=edge_lost_pct
        )

        self.records.append(record)
        await self._push_to_grafana(record)
        return record

    async def _push_to_grafana(self, record: LatencyRecord) -> None:
        """update redis timeseries for grafana dashboard"""
        ts_key = f"latency_ts:{self.strategy}"
        pipe = self.redis.pipeline()
        pipe.lpush(f"{ts_key}:latency_ms", round(record.signal_to_fill_ms, 2))
        pipe.lpush(f"{ts_key}:alpha_captured", round(record.alpha_captured, 6))
        pipe.lpush(f"{ts_key}:edge_lost", round(record.edge_lost_pct, 4))
        pipe.ltrim(f"{ts_key}:latency_ms", 0, 2999)
        pipe.ltrim(f"{ts_key}:alpha_captured", 0, 2999)
        pipe.ltrim(f"{ts_key}:edge_lost", 0, 2999)
        await pipe.execute()

    def get_stats(self) -> Dict:
        """current latency and alpha capture statistics"""
        if not self.records:
            return {'error': 'no records yet'}

        latencies = sorted(r.signal_to_fill_ms for r in self.records)
        alphas = [r.alpha_captured for r in self.records]
        total_expected = sum(r.signal.expected_alpha for r in self.records)
        total_captured = sum(r.alpha_captured for r in self.records)

        return {
            'count': len(latencies),
            'median_ms': statistics.median(latencies),
            'p95_ms': latencies[int(0.95 * len(latencies))],
            'p99_ms': latencies[int(0.99 * len(latencies))],
            'mean_alpha_captured': statistics.mean(alphas),
            'total_edge_lost_pct': round((1 - total_captured / total_expected) * 100, 1),
            'half_life_ms': self.half_life_ms
        }

    def get_venue_comparison(self) -> Dict[str, Dict]:
        """latency breakdown by execution venue"""
        venues: Dict[str, list] = {}
        for r in self.records:
            v = r.fill.venue
            if v not in venues:
                venues[v] = []
            venues[v].append(r.signal_to_fill_ms)

        result = {}
        for venue, lats in venues.items():
            s = sorted(lats)
            result[venue] = {
                'count': len(s),
                'median_ms': statistics.median(s),
                'p95_ms': s[int(0.95 * len(s))],
                'mean_alpha_captured': statistics.mean(
                    r.alpha_captured for r in self.records if r.fill.venue == venue
                )
            }
        return result

    def identify_routing_errors(self) -> List[LatencyRecord]:
        """
        find fills that should have gone to colo but didn't.
        any fill >50ms on a strategy with half_life <100ms is a routing error.
        """
        if self.half_life_ms >= 300:
            return []  # slow strategy, home is fine
        return [
            r for r in self.records
            if r.signal_to_fill_ms > 50 and r.fill.venue == 'home_san_diego'
        ]

ROUTING_RULES = {
    'crypto_momentum':  'colo_chicago',   # half_life=85ms, colo mandatory
    'crypto_mean_rev':  'colo_chicago',   # half_life=450ms, colo preferred
    'es_momentum':      'colo_chicago',   # half_life=120ms, colo mandatory
    'nq_momentum':      'colo_chicago',   # half_life=130ms, colo mandatory
    'options_momentum': 'home_san_diego', # half_life=350ms, home acceptable
    'options_vol_sell': 'home_san_diego', # half_life=900ms, home fine
}