moved all my real-time market data from postgres to redis about 8 weeks ago. latency dropped from ~15ms to sub-millisecond. should’ve done this way earlier.
the postgres problem #
when i first built my algo infrastructure i stored everything in postgres because that’s what i knew. tick data, order book snapshots, greeks, everything went into timescaledb (postgres extension for time-series).
worked fine for backtesting and analysis. reading historical data? postgres crushes it.
but real-time execution? different story.
my algos were querying postgres every 500ms for:
- latest BTC/ETH prices across 3 exchanges
- SPX option greeks for 50+ contracts
- ES/NQ futures prices
- current portfolio positions
postgres was averaging 12-18ms per query. sounds fast but when you’re making trading decisions in real-time, 15ms of stale data can fuck you.
example: algo sees BTC at $68,450 (from postgres), decides to buy. by the time order hits binance, BTC is at $68,510. slippage just ate my edge.
learned a ton from NexusFi infrastructure threads about how pro traders handle real-time data. answer: in-memory everything.
enter redis #
redis is an in-memory key-value store. no disk I/O for reads (unless you force persistence). everything lives in RAM.
query latency? <1ms. usually 0.2-0.5ms.
that 15ms vs 0.3ms difference doesn’t sound like much but when your algo is checking prices hundreds of times per second, it adds up to actual edge.
architecture #
here’s what i built:
import redis
import json
import time
from typing import Dict, Optional
from dataclasses import dataclass, asdict
from datetime import datetime
@dataclass
class MarketData:
"""Standardized market data structure"""
symbol: str
price: float
bid: float
ask: float
volume_24h: float
timestamp: float
exchange: str
class RedisMarketData:
"""
Real-time market data cache using Redis
Design:
- Each symbol gets a hash key: "market:{symbol}"
- TTL on all keys: 5 seconds (auto-expire stale data)
- Pub/Sub for live updates to subscribed algos
- Sorted sets for fast top-N queries
"""
def __init__(
self,
host: str = 'localhost',
port: int = 6379,
db: int = 0
):
self.redis = redis.Redis(
host=host,
port=port,
db=db,
decode_responses=True
)
self.pubsub = self.redis.pubsub()
def update_market_data(
self,
data: MarketData,
ttl: int = 5
) -> None:
"""
Store latest market data with TTL
TTL ensures stale data expires automatically.
If feed dies, data disappears after 5 seconds.
"""
key = f"market:{data.symbol}"
# Store as hash (faster than JSON for partial updates)
pipeline = self.redis.pipeline()
pipeline.hset(key, mapping=asdict(data))
pipeline.expire(key, ttl)
# Update sorted set for fast ranking queries
# Score = timestamp (allows time-based sorting)
pipeline.zadd(
"market:timestamps",
{data.symbol: data.timestamp}
)
pipeline.execute()
# Publish update to subscribers
self.redis.publish(
f"updates:{data.symbol}",
json.dumps(asdict(data))
)
def get_market_data(
self,
symbol: str
) -> Optional[MarketData]:
"""
Retrieve latest market data (sub-ms latency)
Returns None if data expired (feed died)
"""
key = f"market:{data.symbol}"
data = self.redis.hgetall(key)
if not data:
return None
# Convert back to dataclass
return MarketData(
symbol=data['symbol'],
price=float(data['price']),
bid=float(data['bid']),
ask=float(data['ask']),
volume_24h=float(data['volume_24h']),
timestamp=float(data['timestamp']),
exchange=data['exchange']
)
def get_spread_bps(self, symbol: str) -> Optional[float]:
"""Calculate current bid-ask spread in basis points"""
data = self.get_market_data(symbol)
if not data:
return None
mid = (data.bid + data.ask) / 2
spread = data.ask - data.bid
return (spread / mid) * 10000 # Convert to bps
def get_stale_symbols(
self,
max_age_seconds: float = 2.0
) -> list[str]:
"""
Find symbols with data older than threshold
Useful for detecting dead feeds
"""
now = time.time()
cutoff = now - max_age_seconds
# Query sorted set for timestamps < cutoff
stale = self.redis.zrangebyscore(
"market:timestamps",
"-inf",
cutoff
)
return stale
def subscribe_to_updates(
self,
symbols: list[str],
callback
) -> None:
"""
Subscribe to real-time updates via pub/sub
Callback receives MarketData object on each update
"""
channels = [f"updates:{s}" for s in symbols]
self.pubsub.subscribe(*channels)
for message in self.pubsub.listen():
if message['type'] == 'message':
data = json.loads(message['data'])
callback(MarketData(**data))
class MarketDataAggregator:
"""
Aggregate market data across multiple exchanges
Uses Redis to cache best bid/ask across venues
"""
def __init__(self, redis_client: RedisMarketData):
self.redis = redis_client
def update_aggregated_book(
self,
symbol: str,
exchange_data: list[MarketData]
) -> None:
"""
Find best bid/ask across all exchanges
Stores in Redis for instant lookup
"""
if not exchange_data:
return
# Find best bid (highest)
best_bid = max(exchange_data, key=lambda x: x.bid)
# Find best ask (lowest)
best_ask = min(exchange_data, key=lambda x: x.ask)
# Create aggregated view
agg = MarketData(
symbol=symbol,
price=(best_bid.bid + best_ask.ask) / 2, # Mid price
bid=best_bid.bid,
ask=best_ask.ask,
volume_24h=sum(d.volume_24h for d in exchange_data),
timestamp=time.time(),
exchange='aggregated'
)
# Store with "agg:" prefix
key = f"market:agg:{symbol}"
self.redis.redis.hset(key, mapping=asdict(agg))
self.redis.redis.expire(key, 5)
def get_best_execution_venue(
self,
symbol: str,
side: str, # 'buy' or 'sell'
size_usd: float
) -> Optional[str]:
"""
Find best exchange for execution based on current book
Accounts for size vs available depth
"""
# Get data from all exchanges
exchanges = ['binance', 'kraken', 'coinbase']
venue_data = []
for exchange in exchanges:
key = f"market:{symbol}:{exchange}"
data = self.redis.redis.hgetall(key)
if data:
venue_data.append((exchange, data))
if not venue_data:
return None
# Simple: pick best price
# TODO: factor in depth, fees, latency
if side == 'buy':
# Want lowest ask
best = min(venue_data, key=lambda x: float(x[1]['ask']))
else:
# Want highest bid
best = max(venue_data, key=lambda x: float(x[1]['bid']))
return best[0]
# Usage in trading algo
def algo_execution_example():
"""
Example: Algo checks Redis for latest prices
"""
redis_data = RedisMarketData()
# Algo loop
while True:
# Get latest BTC price across exchanges
btc_binance = redis_data.get_market_data("BTCUSDT:binance")
btc_kraken = redis_data.get_market_data("XBTUSD:kraken")
btc_coinbase = redis_data.get_market_data("BTC-USD:coinbase")
if not all([btc_binance, btc_kraken, btc_coinbase]):
print("Stale data detected, skipping cycle")
time.sleep(0.1)
continue
# Calculate cross-exchange spread
prices = [btc_binance.price, btc_kraken.price, btc_coinbase.price]
spread = (max(prices) - min(prices)) / min(prices) * 10000 # bps
if spread > 10: # 10 bps spread
print(f"Arb opportunity: {spread:.2f} bps")
# Execute arb strategy...
time.sleep(0.1) # 100ms cycle time
# Feed updater (runs separately)
async def feed_updater():
"""
Background process: pulls from exchange APIs, updates Redis
This runs 24/7, constantly updating Redis cache
"""
import ccxt
redis_data = RedisMarketData()
binance = ccxt.binance()
kraken = ccxt.kraken()
symbols = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT']
while True:
for symbol in symbols:
try:
# Fetch from Binance
ticker = binance.fetch_ticker(symbol)
data = MarketData(
symbol=f"{symbol}:binance",
price=ticker['last'],
bid=ticker['bid'],
ask=ticker['ask'],
volume_24h=ticker['quoteVolume'],
timestamp=time.time(),
exchange='binance'
)
# Update Redis (0.3ms operation)
redis_data.update_market_data(data)
except Exception as e:
print(f"Error updating {symbol}: {e}")
await asyncio.sleep(0.5) # Update every 500ms
results #
before redis (postgres):
- avg query latency: 15ms
- p99 query latency: 45ms
- max concurrent queries: ~50/sec before DB chokes
after redis:
- avg query latency: 0.3ms
- p99 query latency: 1.2ms
- max concurrent queries: 10,000+/sec (not even breaking a sweat)
that 50x latency improvement translated to tighter fills and less slippage. measuring ~0.8 bps improvement on avg fill quality for crypto trades.
doesn’t sound like much but on $2M+ daily volume it’s like $1600/day or $400k/year. pays for the entire server rack.
postgres still has a role #
didn’t completely ditch postgres. still use it for:
- historical data storage - all tick data goes to timescaledb after the fact for backtesting
- EOD analytics - calculate daily stats, store in postgres
- audit trail - every trade logged to postgres for compliance
- slow queries - anything that doesn’t need <10ms latency
redis is ephemeral (in-memory). if server crashes, data gone. so postgres is the permanent record. redis is the speed layer.
lessons #
-
know your query patterns - if you’re doing tons of small reads, in-memory wins. big analytical queries? postgres.
-
latency matters exponentially - 15ms doesn’t sound bad until you realize your competitor is at 0.5ms and they’re front-running you.
-
TTLs prevent stale data bugs - auto-expiring keys in redis means if a feed dies, your algo knows immediately. no more trading on 5-minute-old prices.
-
redis isn’t a database replacement - it’s a cache. treat it as such. postgres is still the source of truth.
-
pub/sub is underrated - instead of polling redis every 100ms, use pub/sub to get pushed updates. even lower latency.
still tuning the setup. thinking about adding redis cluster for HA (currently single instance = single point of failure). also want to experiment with valkey (redis fork) to see if performance is any different.
but for now this setup is crushing it. latency down 50x, fills tighter, edge preserved.
-AK