been refactoring my signal generation pipeline for the last 2 weeks. old version was synchronous - fetch data, compute indicators, generate signal, repeat. worked fine when i was running 3 strategies. now i’m running 11 and the whole thing was choking.
the bottleneck #
my old pipeline looked like this:
# old synchronous garbage
def generate_signals(strategies: list[Strategy]) -> list[Signal]:
signals = []
for strategy in strategies:
data = fetch_market_data(strategy.symbols) # 200-800ms per call
indicators = compute_indicators(data) # 50-150ms
signal = strategy.evaluate(indicators) # 10-30ms
signals.append(signal)
return signals
11 strategies × ~500ms average data fetch = 5.5 seconds just for data. add indicator computation and you’re looking at 7-8 seconds per cycle. my fastest strategy needs signals every 2 seconds.
math doesn’t work.
async everything #
rewrote the whole thing with asyncio + aiohttp. the difference is stupid:
import asyncio
import aiohttp
import numpy as np
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime, timezone
import time
import logging
logger = logging.getLogger(__name__)
@dataclass
class MarketSnapshot:
symbol: str
bid: float
ask: float
last: float
volume: int
timestamp: float
exchange: str
@property
def mid(self) -> float:
return (self.bid + self.ask) / 2
@property
def spread_bps(self) -> float:
if self.mid == 0:
return 0
return ((self.ask - self.bid) / self.mid) * 10000
@dataclass
class Signal:
strategy_id: str
symbol: str
direction: int # 1 = long, -1 = short, 0 = flat
strength: float # 0.0 to 1.0
confidence: float
timestamp: float
metadata: dict = field(default_factory=dict)
@property
def is_actionable(self) -> bool:
return abs(self.direction) == 1 and self.strength > 0.3 and self.confidence > 0.5
class AsyncDataFetcher:
"""fetches market data from multiple sources concurrently"""
def __init__(self, sources: dict[str, str]):
self.sources = sources # {exchange: base_url}
self._session: Optional[aiohttp.ClientSession] = None
self._cache: dict[str, tuple[float, MarketSnapshot]] = {}
self.cache_ttl = 0.1 # 100ms cache to prevent hammering
async def _get_session(self) -> aiohttp.ClientSession:
if self._session is None or self._session.closed:
timeout = aiohttp.ClientTimeout(total=2.0, connect=0.5)
self._session = aiohttp.ClientSession(timeout=timeout)
return self._session
async def fetch_symbol(
self, symbol: str, exchange: str
) -> Optional[MarketSnapshot]:
cache_key = f"{exchange}:{symbol}"
now = time.monotonic()
# check cache first
if cache_key in self._cache:
cached_time, cached_snap = self._cache[cache_key]
if now - cached_time < self.cache_ttl:
return cached_snap
try:
session = await self._get_session()
url = f"{self.sources[exchange]}/v1/quote/{symbol}"
async with session.get(url) as resp:
if resp.status != 200:
logger.warning(f"bad response {resp.status} for {symbol}@{exchange}")
return None
data = await resp.json()
snap = MarketSnapshot(
symbol=symbol,
bid=data["bid"],
ask=data["ask"],
last=data["last"],
volume=data["volume"],
timestamp=data["timestamp"],
exchange=exchange,
)
self._cache[cache_key] = (now, snap)
return snap
except asyncio.TimeoutError:
logger.error(f"timeout fetching {symbol}@{exchange}")
return None
except Exception as e:
logger.error(f"error fetching {symbol}@{exchange}: {e}")
return None
async def fetch_batch(
self, symbols: list[str], exchange: str
) -> dict[str, MarketSnapshot]:
tasks = [self.fetch_symbol(s, exchange) for s in symbols]
results = await asyncio.gather(*tasks, return_exceptions=True)
snapshots = {}
for symbol, result in zip(symbols, results):
if isinstance(result, MarketSnapshot):
snapshots[symbol] = result
elif isinstance(result, Exception):
logger.error(f"exception for {symbol}: {result}")
return snapshots
async def close(self):
if self._session and not self._session.closed:
await self._session.close()
class SignalEngine:
"""async signal generation across multiple strategies"""
def __init__(self, fetcher: AsyncDataFetcher):
self.fetcher = fetcher
self.strategies: dict[str, 'BaseStrategy'] = {}
self._signal_history: list[Signal] = []
self._cycle_times: list[float] = []
def register_strategy(self, strategy: 'BaseStrategy'):
self.strategies[strategy.strategy_id] = strategy
logger.info(f"registered strategy: {strategy.strategy_id}")
async def generate_cycle(self) -> list[Signal]:
"""run one full signal generation cycle across all strategies"""
cycle_start = time.monotonic()
# collect all unique symbols needed
all_symbols: dict[str, set[str]] = {} # {exchange: {symbols}}
for strategy in self.strategies.values():
for symbol, exchange in strategy.required_feeds:
all_symbols.setdefault(exchange, set()).add(symbol)
# fetch all data concurrently
fetch_tasks = []
for exchange, symbols in all_symbols.items():
fetch_tasks.append(
self.fetcher.fetch_batch(list(symbols), exchange)
)
all_snapshots_list = await asyncio.gather(*fetch_tasks)
# merge into single dict
snapshots: dict[str, MarketSnapshot] = {}
for batch in all_snapshots_list:
for key, snap in batch.items():
snapshots[f"{snap.exchange}:{key}"] = snap
# generate signals concurrently
signal_tasks = []
for strategy in self.strategies.values():
signal_tasks.append(
self._run_strategy(strategy, snapshots)
)
signal_batches = await asyncio.gather(*signal_tasks, return_exceptions=True)
signals = []
for batch in signal_batches:
if isinstance(batch, list):
signals.extend(batch)
elif isinstance(batch, Exception):
logger.error(f"strategy error: {batch}")
cycle_time = time.monotonic() - cycle_start
self._cycle_times.append(cycle_time)
if len(self._cycle_times) % 100 == 0:
avg = np.mean(self._cycle_times[-100:])
p99 = np.percentile(self._cycle_times[-100:], 99)
logger.info(f"cycle stats - avg: {avg*1000:.1f}ms, p99: {p99*1000:.1f}ms")
self._signal_history.extend(signals)
return signals
async def _run_strategy(
self, strategy: 'BaseStrategy', snapshots: dict
) -> list[Signal]:
"""run a single strategy's signal generation"""
try:
relevant = {}
for symbol, exchange in strategy.required_feeds:
key = f"{exchange}:{symbol}"
if key in snapshots:
relevant[symbol] = snapshots[key]
if len(relevant) < strategy.min_required_feeds:
logger.warning(
f"{strategy.strategy_id}: insufficient data "
f"({len(relevant)}/{strategy.min_required_feeds})"
)
return []
return await strategy.generate(relevant)
except Exception as e:
logger.error(f"{strategy.strategy_id} failed: {e}")
return []
@property
def avg_cycle_ms(self) -> float:
if not self._cycle_times:
return 0
return np.mean(self._cycle_times[-50:]) * 1000
class BaseStrategy:
"""base class for signal generation strategies"""
def __init__(self, strategy_id: str):
self.strategy_id = strategy_id
self.required_feeds: list[tuple[str, str]] = []
self.min_required_feeds: int = 1
self._lookback: list[dict[str, MarketSnapshot]] = []
self.lookback_size: int = 100
async def generate(self, snapshots: dict[str, MarketSnapshot]) -> list[Signal]:
self._lookback.append(snapshots)
if len(self._lookback) > self.lookback_size:
self._lookback = self._lookback[-self.lookback_size:]
if len(self._lookback) < 10:
return [] # need minimum history
return await self._evaluate(snapshots)
async def _evaluate(self, snapshots: dict[str, MarketSnapshot]) -> list[Signal]:
raise NotImplementedError
class MomentumCrossStrategy(BaseStrategy):
"""simple momentum crossover - fast EMA vs slow EMA on mid prices"""
def __init__(self, symbol: str, exchange: str, fast: int = 8, slow: int = 21):
super().__init__(f"momentum_cross_{symbol}_{fast}_{slow}")
self.symbol = symbol
self.required_feeds = [(symbol, exchange)]
self.fast_period = fast
self.slow_period = slow
self.min_required_feeds = 1
self.lookback_size = slow * 3
async def _evaluate(self, snapshots: dict[str, MarketSnapshot]) -> list[Signal]:
if self.symbol not in snapshots:
return []
mids = np.array([
lb[self.symbol].mid
for lb in self._lookback
if self.symbol in lb
])
if len(mids) < self.slow_period:
return []
# compute EMAs
fast_ema = self._ema(mids, self.fast_period)
slow_ema = self._ema(mids, self.slow_period)
# crossover detection
if len(fast_ema) < 2:
return []
prev_diff = fast_ema[-2] - slow_ema[-2]
curr_diff = fast_ema[-1] - slow_ema[-1]
direction = 0
if prev_diff <= 0 and curr_diff > 0:
direction = 1 # bullish crossover
elif prev_diff >= 0 and curr_diff < 0:
direction = -1 # bearish crossover
if direction == 0:
return []
strength = min(abs(curr_diff) / snapshots[self.symbol].mid * 1000, 1.0)
return [Signal(
strategy_id=self.strategy_id,
symbol=self.symbol,
direction=direction,
strength=strength,
confidence=0.6 if abs(curr_diff) > abs(prev_diff) else 0.4,
timestamp=time.time(),
metadata={
"fast_ema": float(fast_ema[-1]),
"slow_ema": float(slow_ema[-1]),
"spread_bps": snapshots[self.symbol].spread_bps,
}
)]
@staticmethod
def _ema(data: np.ndarray, period: int) -> np.ndarray:
alpha = 2 / (period + 1)
ema = np.zeros_like(data)
ema[0] = data[0]
for i in range(1, len(data)):
ema[i] = alpha * data[i] + (1 - alpha) * ema[i - 1]
return ema
# ---- main loop ----
async def main():
sources = {
"polygon": "https://api.polygon.io",
"binance": "https://api.binance.us",
"ib": "http://localhost:5000", # local IB gateway
}
fetcher = AsyncDataFetcher(sources)
engine = SignalEngine(fetcher)
# register strategies
engine.register_strategy(MomentumCrossStrategy("BTCUSD", "binance", 8, 21))
engine.register_strategy(MomentumCrossStrategy("ETHUSD", "binance", 5, 13))
engine.register_strategy(MomentumCrossStrategy("ES", "ib", 10, 30))
engine.register_strategy(MomentumCrossStrategy("NQ", "ib", 8, 21))
logger.info(f"starting signal engine with {len(engine.strategies)} strategies")
try:
while True:
signals = await engine.generate_cycle()
actionable = [s for s in signals if s.is_actionable]
if actionable:
for sig in actionable:
logger.info(
f"SIGNAL: {sig.strategy_id} -> {sig.symbol} "
f"{'LONG' if sig.direction == 1 else 'SHORT'} "
f"strength={sig.strength:.2f} conf={sig.confidence:.2f}"
)
await asyncio.sleep(1.0) # 1 second cycles
except KeyboardInterrupt:
logger.info("shutting down")
finally:
await fetcher.close()
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
asyncio.run(main())
performance difference #
old sync pipeline: 7.2 seconds per cycle (11 strategies)
new async pipeline: 380ms per cycle (11 strategies, same data sources)
that’s a 19x improvement just from running data fetches concurrently. the compute part (indicators + signal evaluation) was already fast - it was the I/O killing me.
the cache layer matters #
one thing that surprised me - even with async, you can hammer your data sources too hard. polygon.io rate-limits at 100 req/s on my plan. 11 strategies needing 4-5 symbols each = 50+ requests per cycle. at 1 second cycles that’s fine but if i want sub-second cycles i need caching.
added a 100ms TTL cache in the fetcher. if two strategies need BTCUSD from binance within 100ms, second one gets cached data. sounds obvious but it reduced API calls by ~40% without meaningfully increasing data staleness.
for context - BTC moves about 0.001% in 100ms on average. that’s noise. my strategies need price changes >0.05% to trigger signals. the cache isn’t hurting signal quality at all.
learned about cache invalidation strategies from NexusFi’s infrastructure discussions - some of those guys are running way more sophisticated setups than mine.
gotchas i hit #
1. aiohttp connection pooling
by default aiohttp creates a new TCP connection per request. for low-latency data fetching that’s terrible - TCP handshake adds 10-30ms. you need connection pooling:
connector = aiohttp.TCPConnector(
limit=100, # max connections total
limit_per_host=20, # max per host
keepalive_timeout=30,
enable_cleanup_closed=True,
)
session = aiohttp.ClientSession(connector=connector)
2. asyncio.gather exception handling
asyncio.gather(*tasks) by default raises on first exception. set return_exceptions=True or one failed data source kills your entire cycle. learned this the hard way when binance went down for 30 seconds and my entire pipeline crashed.
3. backpressure when signals pile up
if the execution engine is slow and signals queue up, you can get stale signals executing. added a TTL to signals - anything older than 5 seconds gets dropped:
def filter_stale(signals: list[Signal], max_age: float = 5.0) -> list[Signal]:
now = time.time()
return [s for s in signals if now - s.timestamp < max_age]
what’s next #
working on adding websocket feeds instead of REST polling. polygon and binance both support websocket streams which should drop latency even further. instead of polling every second, i’d get push updates on every tick.
the challenge is managing websocket reconnection gracefully. connections drop, exchanges do maintenance, etc. need a robust reconnect mechanism with exponential backoff.
also want to add signal correlation detection - if 3 strategies all fire long signals on the same symbol within 2 seconds, that’s probably a stronger signal than any individual one. need to build a signal aggregation layer.
-AK