2:30 AM friday. been at this since 9 PM.
promised myself two weeks ago, right in the middle of the tariff chaos, that i’d actually fix the IV rank staleness issue. the signal quality scoring work was the band-aid — a composite gate that tells the system “this signal isn’t reliable right now.” it worked. it’s in production. but the underlying problem was unchanged: during the spike, my IV rank was being computed from options data that was 10-14 minutes old. the signal wasn’t wrong, technically. it was just answering a question about a market that no longer existed.
you can gate on signal quality all day. eventually you have to fix the actual data.
the polling problem at speed #
the IV rank calculation runs every 15 minutes. it pulls a snapshot of the SPX options chain from ThetaData via REST, computes the 30-day IV percentile rank against a trailing 252-day window, and publishes the result to Redis. straightforward, deterministic, easy to debug.
during normal conditions — VIX in the 16-22 range, IV rank drifting around 20-35%, nothing dramatic happening — this is fine. IV rank moves slowly. a 12-minute-old reading and a current one are usually within 2-3 percentage points of each other. that difference doesn’t change any signal decision.
tariff week was not normal conditions.
on monday april 7th, IV rank was at 82% in pre-market and moving roughly 3-4 percentage points every 15 minutes. by the time my system ran a fresh calculation, the market had shifted enough that the previous snapshot was already describing a different environment. i was making decisions with a 12-minute-old map of terrain that was changing by the minute.
the SQS freshness component flagged this correctly — data freshness scores were in the 20-30 range during the worst periods, which suppressed new position entries. so the damage was contained. but suppressing signal confidence is different from actually having good signal data. ideally you want both.
so: websockets.
why streaming for options greeks specifically #
most data APIs are pull-based. you poll on a schedule. fine for many use cases, but it means your data is always at least as stale as your polling interval.
ThetaData has a websocket interface for real-time streaming of options quotes — bid/ask IV, mid IV, delta, gamma, theta, vega per contract, pushing updates as market data changes. instead of a scheduled pull every 15 minutes, the stream delivers updates reactively as the exchange publishes them.
the practical difference during a vol event:
- REST polling (15-min): during the tariff monday open (9:30–10:45 AM ET), IV rank moved ~22 percentage points. my system saw that in 2 data points. resolution: 11pp per update.
- websocket streaming: same 75-minute window would deliver ~900+ quote updates across the SPX chain. resolution: 0.025pp per update.
most of those 900 updates don’t change IV rank meaningfully — the signal is a smoothed 252-day percentile rank, not a tick-by-tick reactive value. but the system knows that within 2 seconds now, instead of 14 minutes later.
architecture: streaming primary, polling fallback #
before writing any code, thought through the failure modes. websockets drop. ThetaData has maintenance windows. the connection can go stale without a clean disconnect signal.
the call: websocket as primary, REST polling as fallback. if the connection is healthy and delivering data, IV rank updates stream in. if the connection drops or goes silent for more than 30 seconds during market hours, the system reverts to the existing 15-minute polling cycle. not ideal, but better than no IV rank.
health tracking via Redis: a heartbeat key that gets updated on every valid quote received. the IV rank consumer checks the key age — if it’s been more than 30 seconds since last update during market hours, it raises a staleness flag and routes computation through the SQS staleness penalty path (same path as before, just now it’s a rare fallback instead of the default state).
here’s the implementation:
import asyncio
import json
import websockets
import redis.asyncio as aioredis
from datetime import datetime, timezone
from typing import Optional
import logging
logger = logging.getLogger(__name__)
THETADATA_WS_URL = "wss://stream.thetadata.us/v1"
REDIS_URL = "redis://localhost:6379"
HEARTBEAT_KEY = "options:ws:last_update"
IV_RANK_KEY = "signal:spx:iv_rank"
IV_RANK_CHANNEL = "signals:iv_rank"
STALENESS_THRESHOLD_S = 30
class OptionsStreamHandler:
"""
Handles real-time SPX options quote streaming from ThetaData.
Maintains rolling IV history and publishes IV rank to Redis pub/sub.
"""
def __init__(self, redis_client: aioredis.Redis, api_key: str):
self.redis = redis_client
self.api_key = api_key
self.iv_history: list[tuple[int, float]] = [] # (ts_ms, mid_iv)
self.current_iv_rank: Optional[float] = None
self._ws: Optional[websockets.WebSocketClientProtocol] = None
self._connected = False
async def connect(self) -> None:
headers = {"Authorization": f"Bearer {self.api_key}"}
self._ws = await websockets.connect(
THETADATA_WS_URL,
extra_headers=headers,
ping_interval=20,
ping_timeout=10,
close_timeout=5,
)
self._connected = True
logger.info("ThetaData websocket connected")
subscribe_msg = {
"msg_type": "SUBSCRIBE",
"sec_type": "OPTION",
"root": "SPX",
"exp": "front_2", # front 2 expirations
"fields": ["BID_IV", "ASK_IV", "MID_IV", "DELTA", "GAMMA", "VEGA"],
"interval_ms": 500, # max update rate: every 500ms per contract
}
await self._ws.send(json.dumps(subscribe_msg))
logger.info("Subscribed to SPX chain")
def _compute_iv_rank(self, current_iv: float) -> float:
"""
IV rank: (current_iv - 252d_low) / (252d_high - 252d_low) * 100
Standard percentile definition.
"""
if len(self.iv_history) < 30:
return 50.0 # insufficient history
ivs = [iv for _, iv in self.iv_history]
iv_low = min(ivs)
iv_high = max(ivs)
if iv_high == iv_low:
return 50.0
rank = (current_iv - iv_low) / (iv_high - iv_low) * 100.0
return max(0.0, min(100.0, rank))
async def handle_quote(self, msg: dict) -> None:
mid_iv = msg.get("mid_iv")
if mid_iv is None or mid_iv <= 0:
return
now_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
# append and prune to 252 trading days (~rolling year)
self.iv_history.append((now_ms, mid_iv))
cutoff_ms = now_ms - (252 * 86_400 * 1000)
self.iv_history = [(ts, iv) for ts, iv in self.iv_history if ts > cutoff_ms]
new_rank = self._compute_iv_rank(mid_iv)
# only publish when rank moves by >0.5pp — filters noise without hiding real moves
if self.current_iv_rank is None or abs(new_rank - self.current_iv_rank) > 0.5:
self.current_iv_rank = new_rank
payload = json.dumps({
"value": round(new_rank, 2),
"mid_iv": round(mid_iv, 4),
"updated_ms": now_ms,
"history_length": len(self.iv_history),
})
pipe = self.redis.pipeline()
pipe.set(IV_RANK_KEY, payload, ex=120) # expire if stream goes dark
pipe.publish(IV_RANK_CHANNEL, payload) # notify downstream consumers
pipe.set(HEARTBEAT_KEY, str(now_ms), ex=300) # health check key
await pipe.execute()
async def stream_loop(self) -> None:
while True:
try:
if not self._connected:
await self.connect()
async for raw_msg in self._ws:
msg = json.loads(raw_msg)
if msg.get("msg_type") == "QUOTE":
await self.handle_quote(msg)
elif msg.get("msg_type") == "ERROR":
logger.error(f"ThetaData error: {msg}")
except websockets.ConnectionClosed as e:
logger.warning(f"WS closed ({e}), reconnecting in 5s")
self._connected = False
await asyncio.sleep(5)
except Exception as e:
logger.error(f"Stream loop error: {e}, reconnecting in 10s")
self._connected = False
await asyncio.sleep(10)
async def run_options_stream(api_key: str) -> None:
redis = await aioredis.from_url(REDIS_URL, decode_responses=True)
handler = OptionsStreamHandler(redis_client=redis, api_key=api_key)
await handler.stream_loop()
if __name__ == "__main__":
import os
asyncio.run(run_options_stream(api_key=os.environ["THETADATA_API_KEY"]))
runs as its own process under runit, alongside the rest of the stack. the main signal pipeline subscribes to signals:iv_rank via Redis pub/sub and reacts to updates reactively instead of polling on a schedule.
latency comparison: three days shadow mode #
ran streaming alongside the old 15-minute cycle in shadow mode monday through wednesday this week. measured end-to-end latency from ThetaData publishing a quote to my Redis IV rank key being updated.
log scale on the y-axis because the difference is literally two orders of magnitude. streaming p99 is 4.3 seconds. polling p50 is 7.5 minutes. they’re not even playing the same game.
the shadow mode numbers confirmed what i expected: during calm conditions monday-tuesday, the two methods produced IV rank values within 0.3-0.8pp of each other almost all the time. when a brief VIX move happened wednesday afternoon (19→23 over about 40 minutes), streaming saw the peak IV rank 11.7 minutes before the polling cycle caught up. during tariff week that gap was 8-12pp on IV rank. during wednesday’s smaller move it was 4.2pp at peak. both matter.
redis pub/sub: downstream consumers #
three services subscribe to signals:iv_rank:
SPX premium selling strategy: re-evaluates entry conditions when IV rank crosses its configured tier thresholds (currently: 25%, 50%, 75% trigger different position sizing multipliers). with polling it was checking stale values. now it reacts within 2 seconds of a threshold crossing.
SQS service: recalculates the data freshness component on every pub/sub event. that freshness component — which was scoring 60-70 during market hours even in calm conditions because polling always has some staleness — now shows 92-98 consistently during market hours. the SQS score went up on average just from this change.
Grafana: live IV rank gauge on the monitoring dashboard, now updates reactively. visual improvement, not operationally critical, but i like knowing the number i’m looking at is accurate.
the 0.5pp publish threshold is a practical filter. raw streaming produces 600-900 IV updates per hour during active markets. publishing all of them would create downstream noise and unnecessary processing. the filter means most of the micro-fluctuations get absorbed, but any sustained directional move shows up within a few updates.
what’s still left on the list #
adaptive subscription depth: currently subscribe to front 2 expirations regardless of conditions. during vol spikes, the most relevant data concentrates in 0DTE and next-week options — that’s where the market is pricing panic. want to dynamically shift subscription weight toward shorter expirations when VIX is elevated. requires the streaming process to receive regime signals from SQS (currently it’s one-way: stream → SQS). small architectural change, two-way communication between two services.
IV history persistence across restarts: when the streaming process restarts, the 252-day IV history built up in memory is lost. currently falls back to a TimescaleDB snapshot written every 6 hours. need to write that snapshot every 30 minutes so a restart costs at most 30 minutes of history rather than up to 6 hours. bookkeeping task, not urgent, just annoying.
went live on thursday. running clean so far. SQS data freshness scores are holding 92-98 during market hours. week overall ended up around +$3,800. quiet compared to the chaos of last week but that’s fine — the goal was infrastructure, not performance.
the NexusFi diversified option selling thread has had some good discussion lately on how options traders handle IV data during extreme moves. the consensus there is basically “your risk controls can’t outrun your data quality,” which matches exactly what tariff week taught me.
something my dad once said about his engineering work: “the systems that fail are the ones that assume the data is good.” he was talking about biotech assays, not trading algos, but it’s the same principle. it took me two years and a tariff panic to really understand what he meant.
alright. actually going to sleep before 4 AM for once.
-AK