↓ Skip to main content

data pipeline - real-time market data with python and redis

·1087 words
Infrastructure Python Infrastructure Data-Feeds Redis
Table of Contents

real-time data = critical for algo trading.

redis = in-memory cache for speed.

python pipeline implementation.

the latency problem
#

pulling data every request:

query Polygon API → 50-100ms latency.

multiply by 10 strategies → 500ms-1s total.

too slow for real-time trading.

learned this architecture from NexusFi infrastructure discussions on building low-latency data pipelines for automated trading.

redis solution
#

architecture:

  1. background process streams data → redis
  2. strategies read from redis (< 1ms)
  3. single API connection, multiple consumers

speed improvement:

before: 50-100ms per strategy after: <1ms per strategy

50-100x faster.

python implementation
#

basic redis data pipeline:

import redis
import json
import asyncio
from polygon import WebSocketClient
from typing import Dict, List
import time

class MarketDataPipeline:
    def __init__(self, polygon_api_key: str, redis_host: str = 'localhost', redis_port: int = 6379):
        """
        Initialize market data pipeline

        Args:
            polygon_api_key: Polygon.io API key
            redis_host: Redis server host
            redis_port: Redis server port
        """
        self.polygon_key = polygon_api_key
        self.redis_client = redis.Redis(
            host=redis_host,
            port=redis_port,
            db=0,
            decode_responses=True
        )
        self.ws_client = None
        self.subscriptions = set()

    async def start_stream(self, symbols: List[str]):
        """Start WebSocket stream for symbols"""
        self.subscriptions = set(symbols)

        # Polygon WebSocket client
        self.ws_client = WebSocketClient(
            api_key=self.polygon_key,
            feed='stocks',  # or 'crypto', 'forex'
            market='stocks'
        )

        # Subscribe to quote updates
        for symbol in symbols:
            self.ws_client.subscribe_quotes(symbol, self._handle_quote)

        # Run WebSocket client
        self.ws_client.run_async()

    def _handle_quote(self, msg: Dict):
        """
        Handle incoming quote update

        Stores in Redis with key pattern: quote:{symbol}
        """
        try:
            symbol = msg.get('symbol')
            if not symbol:
                return

            # Build quote data
            quote_data = {
                'symbol': symbol,
                'bid': msg.get('bid_price'),
                'ask': msg.get('ask_price'),
                'bid_size': msg.get('bid_size'),
                'ask_size': msg.get('ask_size'),
                'timestamp': msg.get('timestamp'),
                'exchange': msg.get('exchange'),
                'received_at': time.time() * 1000  # Local timestamp
            }

            # Store in Redis
            key = f"quote:{symbol}"
            self.redis_client.setex(
                key,
                60,  # Expire after 60 seconds
                json.dumps(quote_data)
            )

            # Update last price for quick access
            self.redis_client.setex(
                f"price:{symbol}",
                60,
                quote_data['ask']  # or midpoint
            )

        except Exception as e:
            print(f"Error handling quote: {e}")

    def get_quote(self, symbol: str) -> Dict:
        """
        Get latest quote from Redis

        Args:
            symbol: Stock symbol

        Returns:
            Quote data dict or None
        """
        key = f"quote:{symbol}"
        data = self.redis_client.get(key)

        if data:
            return json.loads(data)
        return None

    def get_price(self, symbol: str) -> float:
        """
        Get latest price quickly

        Args:
            symbol: Stock symbol

        Returns:
            Latest price or None
        """
        key = f"price:{symbol}"
        price = self.redis_client.get(key)

        return float(price) if price else None

    def get_multiple_prices(self, symbols: List[str]) -> Dict[str, float]:
        """
        Get multiple prices in single Redis call

        Args:
            symbols: List of symbols

        Returns:
            Dict mapping symbol to price
        """
        # Use Redis pipeline for batch get
        pipe = self.redis_client.pipeline()

        for symbol in symbols:
            pipe.get(f"price:{symbol}")

        results = pipe.execute()

        return {
            symbol: float(price) if price else None
            for symbol, price in zip(symbols, results)
        }

# Example usage
if __name__ == "__main__":
    # Initialize pipeline
    pipeline = MarketDataPipeline(polygon_api_key='YOUR_API_KEY')

    # Start streaming (async)
    symbols = ['SPY', 'QQQ', 'AAPL', 'MSFT']
    asyncio.run(pipeline.start_stream(symbols))

    # In separate process/thread, strategies read from Redis:
    quote = pipeline.get_quote('SPY')
    print(f"SPY Quote: Bid {quote['bid']} | Ask {quote['ask']}")

    # Or get just price for speed
    price = pipeline.get_price('SPY')
    print(f"SPY Price: ${price:.2f}")

    # Batch get multiple prices
    prices = pipeline.get_multiple_prices(['SPY', 'QQQ', 'AAPL'])
    print(f"Prices: {prices}")

advanced: historical candles caching
#

cache 1-minute candles for backtesting:

class CandleCache:
    def __init__(self, redis_client: redis.Redis):
        """Initialize candle cache"""
        self.redis = redis_client

    def store_candle(self, symbol: str, timeframe: str, candle: Dict):
        """
        Store candle in Redis sorted set

        Args:
            symbol: Stock symbol
            timeframe: '1m', '5m', '1h', etc.
            candle: Dict with open, high, low, close, volume, timestamp
        """
        key = f"candles:{symbol}:{timeframe}"

        # Use timestamp as score for sorting
        score = candle['timestamp']

        # Store candle as JSON
        member = json.dumps(candle)

        # Add to sorted set
        self.redis.zadd(key, {member: score})

        # Keep only last 1000 candles
        self.redis.zremrangebyrank(key, 0, -1001)

        # Set expiration (7 days)
        self.redis.expire(key, 60 * 60 * 24 * 7)

    def get_candles(
        self,
        symbol: str,
        timeframe: str,
        start_time: int = None,
        end_time: int = None,
        limit: int = 100
    ) -> List[Dict]:
        """
        Retrieve candles from Redis

        Args:
            symbol: Stock symbol
            timeframe: Timeframe
            start_time: Start timestamp (ms)
            end_time: End timestamp (ms)
            limit: Max candles to return

        Returns:
            List of candle dicts
        """
        key = f"candles:{symbol}:{timeframe}"

        # Get from sorted set
        if start_time and end_time:
            # Range query
            results = self.redis.zrangebyscore(
                key,
                start_time,
                end_time,
                start=0,
                num=limit
            )
        else:
            # Get latest N candles
            results = self.redis.zrevrange(key, 0, limit - 1)

        # Parse JSON
        candles = [json.loads(r) for r in results]

        return candles

# Example usage
if __name__ == "__main__":
    redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
    cache = CandleCache(redis_client)

    # Store candle
    candle = {
        'timestamp': 1709049600000,  # ms
        'open': 450.20,
        'high': 451.80,
        'low': 449.90,
        'close': 451.20,
        'volume': 125000
    }
    cache.store_candle('SPY', '1m', candle)

    # Retrieve last 100 candles
    candles = cache.get_candles('SPY', '1m', limit=100)
    print(f"Retrieved {len(candles)} candles")

my current setup
#

infrastructure:

redis server on san diego rack (same machine as strategies).

polygon websocket → redis pipeline.

10 strategies reading from redis simultaneously.

symbols tracked:

SPY, QQQ, IWM (index ETFs)

ES, NQ (futures)

BTC, ETH (crypto)

total: 7 symbols, <1MB memory usage

latency:

polygon → redis: ~15ms

strategies → redis: <1ms

total roundtrip: ~16ms

vs previous 50-100ms per strategy.

monitoring redis
#

simple monitoring script:

def monitor_redis_pipeline():
    """Monitor Redis data pipeline health"""
    r = redis.Redis(host='localhost', port=6379, decode_responses=True)

    # Check symbol freshness
    symbols = ['SPY', 'QQQ', 'ES', 'NQ', 'BTC', 'ETH']

    for symbol in symbols:
        quote = r.get(f"quote:{symbol}")
        if quote:
            data = json.loads(quote)
            age = (time.time() * 1000) - data['received_at']
            print(f"{symbol}: {age:.0f}ms old")

            if age > 5000:  # 5 seconds
                print(f"  ⚠️ STALE DATA for {symbol}")
        else:
            print(f"{symbol}: NO DATA")

    # Check Redis memory usage
    info = r.info('memory')
    mem_used = info['used_memory_human']
    print(f"\nRedis memory: {mem_used}")

# Run every 60 seconds
while True:
    monitor_redis_pipeline()
    time.sleep(60)

why redis over database
#

postgres/mysql:

disk-based storage.

query latency 10-50ms.

redis:

in-memory storage.

query latency <1ms.

for real-time trading:

redis 10-50x faster.

critical for millisecond-sensitive strategies.

cost
#

redis cloud: $0

self-hosted on my server rack.

alternative (redis cloud):

$10-30/month for 1GB.

worth it if don’t have own server.

resources
#

redis docs:

https://redis.io/docs/

thorough examples.

polygon websocket:

https://polygon.io/docs/websockets

real-time market data streams.

NexusFi infrastructure discussions:

learned this architecture from experienced algo traders.

tonight (february 26, 2:38am)
#

redis = 50-100x faster than direct API calls.

architecture:

polygon websocket → redis → strategies.

latency:

before: 50-100ms per strategy.

after: <1ms per strategy.

my setup:

7 symbols tracked.

10 strategies reading simultaneously.

self-hosted redis on san diego rack.

this is production algo trading infrastructure.

2:38am wednesday. real-time market data pipeline with python and redis. polygon websocket streams to redis cache, strategies read <1ms latency vs 50-100ms direct API. architecture: single websocket connection, multiple strategy consumers. 50-100x speedup. my setup: 7 symbols (SPY/QQQ/ES/NQ/BTC/ETH/IWM), 10 strategies, self-hosted redis. learned from NexusFi infrastructure discussions. production-grade low-latency data pipeline.

-AK

Related

polygon.io vs alpha vantage - which data feed for algo trading
·1165 words
Infrastructure Data-Feeds Infrastructure Python Api
data feeds = foundation of algo trading. garbage data = garbage trades. i’ve used both polygon.io and alpha vantage extensively. spent months researching data feeds when i started trading. NexusFi community helped narrow down options to these two.
added redis caching - cut market data latency by 60%
·910 words
Infrastructure Python Infrastructure Optimization Redis
been noticing market data latency creeping up. average fetch time: 180ms from polygon API. slowing down entry execution. the problem # every time algo needs current price:
slippage correlation volume deep dive - data analysis
·891 words
Trading Infrastructure Trading Infrastructure Python Data-Analysis
slippage been great november. wanted to understand why. data analysis time. november slippage performance # november avg (through nov 18): 1.9 ticks
chicago colocation - 67ms to 12ms latency improvement, worth the cost
·956 words
Infrastructure Infrastructure Colocation Latency Python
moved execution server to chicago colo march 2024. 3 months data in. latency dropped 67ms → 12ms average. why chicago # CME exchange location: chicago
refactored data pipeline to async - 3x faster market data processing
·907 words
Infrastructure Python Async Optimization Infrastructure
been running synchronous data fetching since january. works but slow during market open. refactored to async this week. 3x speed improvement. the problem with sync code # # Old synchronous approach def fetch_market_data(symbols): results = [] for symbol in symbols: data = fetch_from_api(symbol) # Blocks here results.append(data) return results # With 10 symbols, takes 10 * 180ms = 1,800ms total each API call blocks until complete.
rebuilt backtesting pipeline - 10x faster parameter optimization
·881 words
Algo Development Python Backtesting Optimization Infrastructure
spent last 3 days rebuilding backtest optimization pipeline. went from 6 hours to 35 minutes for full parameter sweep. the problem # old approach: sequential parameter testing.