redis timeseries - cutting latency from 45ms to 8ms

redis-cli SLOWLOG GET 10

import redis
from redis.commands.timeseries import TimeSeries
from typing import List, Tuple, Optional
import struct
import time

class OptimizedMarketDataStore:
    def __init__(self, host: str = 'localhost', port: int = 6379):
        self.redis = redis.Redis(host=host, port=port, decode_responses=False)
        self.ts = self.redis.ts()

    def create_series(self, symbol: str, retention_ms: int = 86400000):
        """Create timeseries for symbol with 24h retention"""
        key = f"tick:{symbol}"
        try:
            self.ts.create(
                key,
                retention_msecs=retention_ms,
                labels={'symbol': symbol, 'type': 'tick'}
            )
        except redis.ResponseError as e:
            if 'already exists' not in str(e):
                raise

    def store_tick(self, symbol: str, price: float, volume: int,
                   timestamp_ms: Optional[int] = None) -> int:
        """
        Store tick with minimal overhead
        Returns timestamp of stored tick
        """
        key = f"tick:{symbol}"
        ts = timestamp_ms or int(time.time() * 1000)

        # Pack price and volume into single value using bit manipulation
        # Price as integer cents (multiply by 100)
        # Volume in lower 32 bits, price in upper 32 bits
        price_cents = int(price * 100)
        packed_value = (price_cents << 32) | (volume & 0xFFFFFFFF)

        return self.ts.add(key, ts, packed_value)

    def get_ticks(self, symbol: str,
                  from_ms: int, to_ms: int) -> List[Tuple[int, float, int]]:
        """
        Retrieve ticks efficiently
        Returns list of (timestamp, price, volume)
        """
        key = f"tick:{symbol}"
        raw_data = self.ts.range(key, from_ms, to_ms)

        result = []
        for ts, packed_value in raw_data:
            packed = int(packed_value)
            price_cents = packed >> 32
            volume = packed & 0xFFFFFFFF
            result.append((ts, price_cents / 100.0, volume))

        return result

    def get_latest(self, symbol: str) -> Optional[Tuple[int, float, int]]:
        """Get most recent tick"""
        key = f"tick:{symbol}"
        try:
            result = self.ts.get(key)
            if result:
                ts, packed = result
                packed = int(packed)
                price_cents = packed >> 32
                volume = packed & 0xFFFFFFFF
                return (ts, price_cents / 100.0, volume)
        except:
            pass
        return None

    def get_ohlcv(self, symbol: str, from_ms: int, to_ms: int,
                   bucket_size_ms: int = 60000) -> List[dict]:
        """
        Aggregate ticks into OHLCV bars
        Uses redis timeseries aggregation (server-side)
        """
        key = f"tick:{symbol}"

        # Get first, last, min, max for each bucket
        aggregations = ['first', 'last', 'min', 'max', 'count']
        results = {}

        for agg in aggregations:
            data = self.ts.range(
                key, from_ms, to_ms,
                aggregation_type=agg,
                bucket_size_msec=bucket_size_ms
            )
            results[agg] = {ts: val for ts, val in data}

        # Combine into OHLCV bars
        bars = []
        for ts in results['first'].keys():
            if ts in results['last']:
                # Unpack prices
                open_packed = int(results['first'][ts])
                close_packed = int(results['last'][ts])
                high_packed = int(results['max'][ts])
                low_packed = int(results['min'][ts])

                bars.append({
                    'timestamp': ts,
                    'open': (open_packed >> 32) / 100.0,
                    'high': (high_packed >> 32) / 100.0,
                    'low': (low_packed >> 32) / 100.0,
                    'close': (close_packed >> 32) / 100.0,
                    'volume': int(results['count'].get(ts, 0))
                })

        return bars


class MarketDataPipeline:
    def __init__(self, store: OptimizedMarketDataStore):
        self.store = store
        self.buffer = {}
        self.buffer_size = 100
        self.last_flush = time.time()

    async def process_tick(self, symbol: str, price: float,
                            volume: int, timestamp_ms: int):
        """
        Buffer ticks and batch write to redis
        Reduces network round-trips
        """
        if symbol not in self.buffer:
            self.buffer[symbol] = []

        self.buffer[symbol].append((timestamp_ms, price, volume))

        # Flush conditions: buffer full OR 100ms elapsed
        should_flush = (
            len(self.buffer[symbol]) >= self.buffer_size or
            time.time() - self.last_flush > 0.1
        )

        if should_flush:
            await self.flush_buffer(symbol)

    async def flush_buffer(self, symbol: str):
        """Pipeline write all buffered ticks"""
        if symbol not in self.buffer or not self.buffer[symbol]:
            return

        pipe = self.store.redis.pipeline()

        for ts, price, volume in self.buffer[symbol]:
            price_cents = int(price * 100)
            packed_value = (price_cents << 32) | (volume & 0xFFFFFFFF)
            pipe.ts().add(f"tick:{symbol}", ts, packed_value)

        pipe.execute()
        self.buffer[symbol] = []
        self.last_flush = time.time()