it’s 1:30am. A. passed out on the couch around midnight waiting for me to come to bed. carried her there, went back to the desk. couldn’t sleep anyway.
march has been annoying. january was solid (+1.9%). february was decent (+2.3%). march so far: -0.5% with two weeks left. not a disaster but it stings after a clean Q4. the algos aren’t broken, the signals are just getting weaker in this chop. SPX has been range-bound for three weeks and my momentum strategies are getting whipsawed.
so i did what i always do when i’m frustrated: i build something new.
tonight’s project: an options flow scanner. the hypothesis is simple - unusual options activity often precedes big moves, and smart money leaves footprints. large OTM option buys before earnings surprises. massive put sweeps before sector selloffs. coordinated call buying that starts before a catalyst lands publicly.
i’ve been thinking about building this for months. tonight was the night.
the theory: options as an information leak #
the equity options market trades about $1-3 trillion in notional daily. most of it is hedging, retail speculation, and systematic strategies. but buried in that flow is something interesting: informed trading.
when someone with real edge wants to express a directional view, they often go to options. leverage. asymmetric payoff. they can size up without moving the stock much. and when a single trader (or coordinated group) puts $2M into OTM calls on a stock that normally sees $50k/day in premium… that’s a signal.
what we’re hunting:
- volume/OI ratio anomalies: vol many times higher than open interest
- unusual premium sweeps: large single orders buying at the ask (aggressor behavior)
- strike/expiry clustering: multiple strikes at same expiry (directional conviction)
- implied vol divergence: IV moving opposite to what the VIX suggests
none of this is guaranteed alpha. but it’s information. and information asymmetry is how you make money.
data architecture #
my setup uses polygon.io for options data. their REST API covers options chains, Greeks, historical trades, and real-time snapshots. i’ve been using them for equity data for a while but only recently started pulling options.
for a scanner like this you need:
- options chain snapshots (all strikes/expiries for a ticker)
- real-time trade-level data (individual option trades as they happen)
- historical baseline (30/60/90 day average volume to detect anomalies)
infra:
- polygon.io python client for API access
- timescaledb for storing historical options volume baselines
- redis for real-time caching of the scanner state
- asyncio for concurrent scanning across multiple tickers
the chicago colo handles the latency-sensitive execution side. the san diego setup (home rack) is where i do the scanning and signal generation since it’s less latency-critical.
the scanner #
here’s the core of what i built tonight. still rough in places but it works.
#!/usr/bin/env python3
"""
Options Flow Scanner - Unusual Activity Detection
Uses Polygon.io options chain data to identify anomalous flow
"""
import asyncio
import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta, date
from typing import Optional
import aiohttp
import pandas as pd
import numpy as np
from polygon import RESTClient
logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger(__name__)
@dataclass
class FlowSignal:
ticker: str
strike: float
expiry: str
option_type: str # 'call' or 'put'
volume: int
open_interest: int
voi_ratio: float # volume / open_interest
premium_spent: float # approx notional ($)
iv: float # implied volatility
delta: float
signal_score: float # composite anomaly score
detected_at: datetime = field(default_factory=datetime.now)
@property
def is_call(self) -> bool:
return self.option_type.lower() == 'call'
@property
def is_high_confidence(self) -> bool:
return self.signal_score >= 0.75
class OptionsFlowScanner:
"""
Scans options markets for unusual activity using Polygon.io data.
Detects anomalous volume-to-open-interest ratios, premium sweeps,
and IV divergence as potential alpha signals.
"""
def __init__(self, api_key: str, tickers: list[str],
voi_threshold: float = 3.0,
min_premium: float = 100_000,
scan_interval: int = 300):
self.client = RESTClient(api_key=api_key)
self.tickers = tickers
self.voi_threshold = voi_threshold # flag if vol > X * avg_OI
self.min_premium = min_premium # minimum notional to flag
self.scan_interval = scan_interval # seconds between full scans
self.signals: list[FlowSignal] = []
self._baseline_cache: dict[str, pd.DataFrame] = {}
self._session: Optional[aiohttp.ClientSession] = None
async def _get_options_chain(self, ticker: str) -> pd.DataFrame:
"""Pull full options chain snapshot for ticker."""
try:
contracts = []
# Polygon snapshot endpoint covers all active contracts
for contract in self.client.list_snapshot_options_chain(
ticker,
params={
"expiration_date.gte": date.today().isoformat(),
"expiration_date.lte": (date.today() + timedelta(days=60)).isoformat(),
"strike_price.gte": 0
}
):
try:
contracts.append({
"ticker": ticker,
"contract": contract.details.ticker,
"strike": contract.details.strike_price,
"expiry": contract.details.expiration_date,
"type": contract.details.contract_type,
"volume": contract.day.volume if contract.day else 0,
"open_interest": contract.open_interest or 1,
"last_price": contract.last_quote.ask if contract.last_quote else 0,
"iv": contract.implied_volatility or 0,
"delta": contract.greeks.delta if contract.greeks else 0,
"gamma": contract.greeks.gamma if contract.greeks else 0,
"vega": contract.greeks.vega if contract.greeks else 0,
})
except AttributeError:
continue
if not contracts:
return pd.DataFrame()
df = pd.DataFrame(contracts)
df["voi_ratio"] = df["volume"] / df["open_interest"].clip(lower=1)
df["premium_spent"] = df["volume"] * df["last_price"] * 100 # per contract = 100 shares
return df
except Exception as e:
logger.error(f"Failed to fetch chain for {ticker}: {e}")
return pd.DataFrame()
def _compute_baseline(self, ticker: str, df: pd.DataFrame) -> dict:
"""
Compute baseline stats for anomaly detection.
In production this hits TimescaleDB for historical averages.
Here we derive rough baseline from the current chain.
"""
if df.empty:
return {"avg_voi": 1.0, "avg_premium": 0, "vol_percentiles": {}}
avg_voi = df["voi_ratio"].median()
avg_premium = df["premium_spent"].median()
# percentile thresholds for ranking
vol_percentiles = {
"p75": df["volume"].quantile(0.75),
"p90": df["volume"].quantile(0.90),
"p95": df["volume"].quantile(0.95),
"p99": df["volume"].quantile(0.99),
}
return {
"avg_voi": avg_voi,
"avg_premium": avg_premium,
"vol_percentiles": vol_percentiles
}
def _score_contract(self, row: pd.Series, baseline: dict) -> float:
"""
Score a contract on 0-1 scale for anomaly likelihood.
Higher = more unusual.
"""
score = 0.0
weights = {
"voi": 0.35,
"premium": 0.30,
"vol_percentile": 0.20,
"iv_component": 0.15
}
# VOI component: how many times baseline VOI is exceeded
voi_baseline = max(baseline["avg_voi"], 0.5)
voi_excess = row["voi_ratio"] / voi_baseline
voi_score = min(voi_excess / self.voi_threshold, 1.0)
score += weights["voi"] * voi_score
# Premium component: is this unusual spending?
pct_tiles = baseline["vol_percentiles"]
if row["volume"] >= pct_tiles.get("p99", float("inf")):
score += weights["vol_percentile"] * 1.0
elif row["volume"] >= pct_tiles.get("p95", float("inf")):
score += weights["vol_percentile"] * 0.75
elif row["volume"] >= pct_tiles.get("p90", float("inf")):
score += weights["vol_percentile"] * 0.50
elif row["volume"] >= pct_tiles.get("p75", float("inf")):
score += weights["vol_percentile"] * 0.25
# Premium size component
if row["premium_spent"] >= self.min_premium * 5:
score += weights["premium"] * 1.0
elif row["premium_spent"] >= self.min_premium * 2:
score += weights["premium"] * 0.7
elif row["premium_spent"] >= self.min_premium:
score += weights["premium"] * 0.4
# IV component: OTM options with high IV can indicate conviction
if abs(row.get("delta", 0.5)) < 0.35 and row["iv"] > 0.40:
score += weights["iv_component"] * 0.8
elif row["iv"] > 0.30:
score += weights["iv_component"] * 0.4
return round(min(score, 1.0), 3)
async def scan_ticker(self, ticker: str) -> list[FlowSignal]:
"""Full scan pipeline for a single ticker."""
df = await self._get_options_chain(ticker)
if df.empty:
return []
baseline = self._compute_baseline(ticker, df)
# compute scores
df["signal_score"] = df.apply(
lambda row: self._score_contract(row, baseline), axis=1
)
# filter candidates
candidates = df[
(df["volume"] > 0) &
(df["voi_ratio"] >= self.voi_threshold) &
(df["premium_spent"] >= self.min_premium) &
(df["signal_score"] >= 0.45)
].sort_values("signal_score", ascending=False)
signals = []
for _, row in candidates.head(10).iterrows():
sig = FlowSignal(
ticker=ticker,
strike=row["strike"],
expiry=str(row["expiry"]),
option_type=row["type"],
volume=int(row["volume"]),
open_interest=int(row["open_interest"]),
voi_ratio=round(row["voi_ratio"], 2),
premium_spent=round(row["premium_spent"], 2),
iv=round(row["iv"], 4),
delta=round(row.get("delta", 0), 4),
signal_score=row["signal_score"]
)
signals.append(sig)
return signals
async def run_continuous(self):
"""Run scanner in a loop, emitting signals."""
logger.info(f"Starting flow scanner for {len(self.tickers)} tickers")
while True:
scan_start = asyncio.get_event_loop().time()
tasks = [self.scan_ticker(t) for t in self.tickers]
results = await asyncio.gather(*tasks, return_exceptions=True)
new_signals = []
for result in results:
if isinstance(result, Exception):
logger.error(f"Scan error: {result}")
continue
new_signals.extend(result)
# rank all signals and emit top hits
new_signals.sort(key=lambda s: s.signal_score, reverse=True)
for sig in new_signals:
if sig.is_high_confidence:
logger.info(
f"HIGH CONFIDENCE: {sig.ticker} {sig.option_type.upper()} "
f"${sig.strike:.0f} exp {sig.expiry} | "
f"Vol: {sig.volume:,} | VOI: {sig.voi_ratio:.1f}x | "
f"Premium: ${sig.premium_spent:,.0f} | Score: {sig.signal_score}"
)
self.signals.extend(new_signals)
# trim to last 1000 signals in memory
self.signals = self.signals[-1000:]
elapsed = asyncio.get_event_loop().time() - scan_start
sleep_time = max(0, self.scan_interval - elapsed)
logger.info(f"Scan complete in {elapsed:.1f}s. Next scan in {sleep_time:.0f}s")
await asyncio.sleep(sleep_time)
if __name__ == "__main__":
import os
WATCHLIST = [
# mega caps - high options volume baseline
"SPY", "QQQ", "IWM", "AAPL", "MSFT", "NVDA", "META", "AMZN",
# sector ETFs
"XLF", "XLE", "XLK", "XBI", "SMH",
# high-beta individual names
"TSLA", "COIN", "MSTR",
]
scanner = OptionsFlowScanner(
api_key=os.getenv("POLYGON_API_KEY"),
tickers=WATCHLIST,
voi_threshold=3.0, # flag if vol is 3x+ open interest
min_premium=150_000, # minimum $150k in premium moved
scan_interval=300 # scan every 5 minutes
)
asyncio.run(scanner.run_continuous())
what it looks like in practice #
i ran this against last tuesday’s data to validate. here’s what the VOI ratio distribution looks like across SPY/QQQ option chains, with anomalies highlighted:
the tail on the right is where it gets interesting. most contracts sit under 3x. anything above that gets flagged by the scanner.
signal ranking: not all anomalies are created equal #
the raw VOI filter catches a lot of noise - weekly expirations with low OI get inflated ratios from even moderate volume. so i layered in the multi-factor scoring:
def rank_signals(signals: list[FlowSignal],
min_score: float = 0.60) -> list[FlowSignal]:
"""
Apply additional ranking criteria to filter noise:
- prefer near-term expiries (7-45 days) over LEAPS
- prefer OTM strikes (delta 0.15-0.40 for directional bets)
- downweight high-IV environments (reduces significance of vol anomalies)
- boost confidence on same-day sweeps (multiple fills at ask, same strike)
"""
ranked = []
for sig in signals:
if sig.signal_score < min_score:
continue
adjusted_score = sig.signal_score
# expiry preference: sweet spot is 7-45 DTE
try:
dte = (datetime.strptime(sig.expiry, "%Y-%m-%d") - datetime.now()).days
if 7 <= dte <= 45:
adjusted_score *= 1.15
elif dte < 7:
adjusted_score *= 0.70 # weeklies have too much noise
elif dte > 90:
adjusted_score *= 0.85 # LEAPS move slowly
except ValueError:
pass
# delta preference: OTM directional bets are most informative
abs_delta = abs(sig.delta)
if 0.15 <= abs_delta <= 0.35:
adjusted_score *= 1.10 # OTM - classic informed bet range
elif abs_delta > 0.60:
adjusted_score *= 0.80 # ITM - less informative, could be hedge
# cap at 1.0
sig.signal_score = min(round(adjusted_score, 3), 1.0)
ranked.append(sig)
return sorted(ranked, key=lambda s: s.signal_score, reverse=True)
def format_signal_alert(sig: FlowSignal) -> str:
direction = "BULLISH" if sig.is_call else "BEARISH"
confidence = "HIGH" if sig.is_high_confidence else "MODERATE"
return (
f"[{confidence}] {sig.ticker} {direction} flow detected\n"
f" Contract: {sig.option_type.upper()} ${sig.strike:.0f} exp {sig.expiry}\n"
f" Volume: {sig.volume:,} | OI: {sig.open_interest:,} | VOI: {sig.voi_ratio:.1f}x\n"
f" Premium: ${sig.premium_spent:,.0f} | IV: {sig.iv:.1%} | Delta: {sig.delta:.2f}\n"
f" Score: {sig.signal_score:.3f}"
)
i also started tracking signal performance retrospectively. look at directional accuracy by signal score bucket - this is off 8 weeks of paper trading the signals (not live yet, still validating):
the n is small - i know. 12 signals in the 0.85+ bucket over 8 weeks isn’t enough to declare victory. but the directional trend is what i’m looking for. below 0.55 you’re basically coin-flipping. above 0.75 it starts looking interesting. above 0.85 is where i might actually start risking real money.
that sample size problem is the main issue with options flow signals - the highest quality ones are rare. so you’re always fighting between statistical significance and signal quality. classic tradeoff.
why this matters for my current setup #
my three main strategies right now are:
- SPX premium selling (theta decay plays) - currently struggling in this vol environment
- ES/NQ momentum (intraday futures) - getting chopped up in the range-bound market
- crypto momentum (BTC/ETH via binance) - holding up okay, +0.8% this month
the flow scanner isn’t replacing any of these. it’s additive. when the scanner flags an unusual sweep on a name, i might:
- hedge an existing position that’s on the wrong side
- add a small directional position (capped at 2% portfolio risk per signal)
- use it as a regime indicator (lots of put sweeps = market nervousness, reduce exposure)
the information is useful even when i don’t trade on it directly. context is alpha.
i’ve been hanging around the algo trading community on NexusFi since early 2023 - there’s a solid thread there on turning algo journals into live system design. the discussion on incorporating signal layers without overcomplicating execution is actually relevant here. worth reading if you’re building anything similar.
what’s next #
a few things i still need to build:
- sweep detection: identifying when a single large order gets filled across multiple exchanges (aggressor behavior). polygon’s trade-level data can do this but i need to build the aggregation logic
- same-strike clustering: when multiple trades hit the same strike/expiry within 10 minutes, that’s more interesting than a single big trade
- historical backtesting: the 8-week paper period is useful but i need to backfill against polygon historical options data (they have it back to 2004) and see how this would have performed
the scanner took about 5 hours to build tonight. it’ll probably take another 20 hours to get to a point where i trust it enough to trade real size against it.
march continues to be annoying. -0.5% on the month. the premium selling portfolio is getting killed by realized vol spiking without a corresponding move in the underlying - worst of both worlds. been thinking about what my dad used to say whenever markets got weird: “when you can’t figure out what’s happening, watch what the smart money is doing, not what they’re saying.”
he was talking about wall street analysts vs actual fund flows. but the advice applies here too. the flow scanner is basically that principle turned into code.
gonna try to sleep before A. wakes up for her 6am run.
-AK