Gahow Wang
251014d3b2
SNDK's October-2026 expiry month (1000 contracts, 3.5 months of daily bars) timed out on every retry, so the backfill skipped it twice. A 504 on a request for 50+ symbols now restarts that chunk as two halves. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
693 lines
36 KiB
Python
693 lines
36 KiB
Python
"""
|
|
Option data for the S&P 500, for options-implied stock factors:
|
|
- IV skew (Xing, Zhang & Zhao 2010)
|
|
- the call-put IV spread (Cremers & Weinbaum 2010)
|
|
- option/stock volume (Johnson & So 2012)
|
|
- implied-vol level and changes
|
|
|
|
uv run python -m research.lab.fetch_options snapshot # after the close (systemd timer)
|
|
uv run python -m research.lab.fetch_options snapshot --limit 5 --force --out /tmp/x # smoke test
|
|
uv run python -m research.lab.fetch_options backfill # OPRA daily bars since 2024-02 (resumable)
|
|
uv run python -m research.lab.fetch_options volume-panel # daily option volume per underlying
|
|
uv run python -m research.lab.fetch_options status # runs, backfill progress, quote-quality verdict
|
|
|
|
Sources: Alpaca. Keys are ALPACA_API_KEY / ALPACA_API_SECRET from .env and
|
|
are never logged. The account has the free Basic plan.
|
|
- Option quotes come from /v1beta1/options/snapshots, feed=indicative,
|
|
with Alpaca's IV and greeks. Alpaca's docs say indicative quotes "are not
|
|
actual OPRA quotes". The OPRA BBO needs Algo Trader Plus ($99/month)
|
|
plus a signed OPRA agreement. Quotes have no free history, so they are
|
|
collected forward from 2026-10-12.
|
|
- Option bars come from /v1beta1/options/bars. They are computed from
|
|
OPRA trades, so they are real prints. History starts in 2024-02, and
|
|
Basic cannot read the last 15 minutes. Volume factors use these bars,
|
|
backfilled.
|
|
- Open interest comes from /v2/options/contracts (OCC, prior day); a
|
|
null means none was reported. Stock prices are SIP daily bars,
|
|
unadjusted, so they match the strikes.
|
|
|
|
snapshot (16:25 America/New_York, Mon-Fri, deploy/systemd/quant-trading-options.timer)
|
|
The Alpaca calendar decides the session: the latest one that closed at
|
|
least 10 minutes ago. The run is skipped if the next session has opened
|
|
or the session's chain file exists, unless --force.
|
|
universe data/universe_sp500.json + SPY, QQQ, IWM
|
|
quotes expiries 5-100 days, strikes 0.7-1.3 x close
|
|
bars + OI every contract expiring within 0-100 days, all strikes
|
|
quality for each underlying, the ATM call, the ATM put and the
|
|
0.95-moneyness put of the expiry nearest 30 days (14-45 days)
|
|
are compared with OPRA prints. The print is the last 1-minute
|
|
bar from 15:45-16:00, and the reference is the indicative bid,
|
|
ask and mid. Measured: share of prints inside the quote,
|
|
position in the spread, IV(print) - IV(mid).
|
|
Pre-set rule: once >= 20 sessions are in, indicative quotes are
|
|
adequate if >= 90% of prints are inside the quote and the
|
|
median |IV(print) - IV(mid)| <= 1 vol point. Otherwise
|
|
consider the OPRA plan. `status` prints the verdict.
|
|
backfill
|
|
- Universe: every name that was in the S&P 500 at some point since
|
|
2024-02-01 (data/sp500_history.json), plus the three ETFs, so names that
|
|
left the index are kept.
|
|
- Contracts: expired and active, expiring from 2024-02-01 to end+100 days.
|
|
- Bars: daily, from 100 days before each expiry to its expiry, for every
|
|
strike.
|
|
- One file per underlying, so the run resumes. It pauses 16:10-16:55 New
|
|
York on weekdays so it does not compete with snapshot for the
|
|
200/minute limit.
|
|
|
|
Files (data/options/us/):
|
|
chains/YYYY/chain_<session>.csv.gz quotes, OPRA bar, OI, Alpaca IV/greeks, close,
|
|
r, implied forward, own IVs (iv_fwd, iv_spot)
|
|
summary/summary_<session>.csv per underlying: iv_atm30, skew_xzz, cp_spread,
|
|
option volume (OPRA), os_volume, pc ratios
|
|
bars/YYYY/bars_<session>.csv.gz OPRA daily bars, 0-100 day expiries
|
|
bars_hist/<UNDERLYING>.csv.gz the backfill, same columns
|
|
stock/stock_<session>.csv, stock_raw.csv.gz unadjusted stock close/volume
|
|
quality/quality_<session>.csv the print-vs-quote sample
|
|
volume_panel.csv.gz date x underlying option volume, built from bars
|
|
manifest.jsonl, backfill_manifest.jsonl
|
|
IVs are European. iv_fwd is Black-76 on the forward implied by put-call
|
|
parity at the three strikes nearest the close, so dividends drop out.
|
|
iv_spot is Black-Scholes on the close with no dividends; cp_spread uses it
|
|
(as Cremers-Weinbaum do), so dividend payers read negative. American early
|
|
exercise is ignored, so the summaries use near-the-money strikes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import datetime as dt
|
|
import glob
|
|
import json
|
|
import os
|
|
import re
|
|
import time
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import requests
|
|
from scipy.special import ndtr
|
|
|
|
from research.lab.fetch_alpaca import _keys
|
|
|
|
NY = ZoneInfo("America/New_York")
|
|
DATA = "https://data.alpaca.markets"
|
|
TRADE = "https://paper-api.alpaca.markets"
|
|
OUT = os.path.join("data", "options", "us")
|
|
UNIVERSE = os.path.join("data", "universe_sp500.json")
|
|
HISTORY = os.path.join("data", "sp500_history.json")
|
|
RF_PATH = os.path.join("data", "research", "ext", "fred_dtb3.csv")
|
|
EXTRA = ["SPY", "QQQ", "IWM"]
|
|
DTE_MIN, DTE_MAX = 5, 100
|
|
MNY_LO, MNY_HI = 0.7, 1.3
|
|
HIST_START = "2024-02-01"
|
|
QUALITY_MIN_SESSIONS, QUALITY_INSIDE, QUALITY_IV = 20, 0.90, 0.01
|
|
OCC = re.compile(r"^[A-Z]{1,5}\d{6,7}[CP]\d{8}$") # what the bars endpoint accepts; adjusted roots like 1AAPL fail
|
|
BAR_COLS = ["date", "underlying", "symbol", "type", "expiry", "strike", "o", "h", "l", "c", "v", "n", "vw"]
|
|
|
|
|
|
class BackendTimeout(RuntimeError):
|
|
"""HTTP 504 on a request for many symbols: the caller splits the batch."""
|
|
|
|
|
|
class Api:
|
|
"""Paced GET with retries. ``yield_to_snapshot`` sleeps through 16:10-16:55
|
|
New York on weekdays so a backfill leaves the rate limit to the daily run."""
|
|
|
|
def __init__(self, min_interval: float = 0.32, yield_to_snapshot: bool = False) -> None:
|
|
self.s = requests.Session()
|
|
self.s.headers.update(_keys())
|
|
self.min_interval = min_interval
|
|
self.yield_to_snapshot = yield_to_snapshot
|
|
self.next = 0.0
|
|
self.n = 0
|
|
|
|
def _quiet_window(self) -> None:
|
|
now = dt.datetime.now(NY)
|
|
start = now.replace(hour=16, minute=10, second=0, microsecond=0)
|
|
end = now.replace(hour=16, minute=55, second=0, microsecond=0)
|
|
if now.weekday() < 5 and start <= now < end:
|
|
print(f" pausing until 16:55 New York for the daily snapshot", flush=True)
|
|
time.sleep((end - now).total_seconds())
|
|
|
|
def get(self, url: str, params: dict) -> dict:
|
|
for attempt in range(7):
|
|
if self.yield_to_snapshot:
|
|
self._quiet_window()
|
|
wait = self.next - time.time()
|
|
if wait > 0:
|
|
time.sleep(wait)
|
|
self.next = time.time() + self.min_interval
|
|
try:
|
|
r = self.s.get(url, params=params, timeout=90)
|
|
except requests.RequestException:
|
|
time.sleep(2 ** attempt)
|
|
continue
|
|
self.n += 1
|
|
if r.status_code == 504 and params.get("symbols", "").count(",") >= 50:
|
|
raise BackendTimeout(r.text[:200]) # the same batch times out again (SNDK, 1000 contracts)
|
|
if r.status_code == 429 or r.status_code >= 500:
|
|
time.sleep(2 ** attempt)
|
|
continue
|
|
if r.status_code != 200:
|
|
raise RuntimeError(f"HTTP {r.status_code} {r.text[:200]}")
|
|
return r.json()
|
|
raise RuntimeError("retries exhausted")
|
|
|
|
def pages(self, url: str, params: dict, key: str):
|
|
params = dict(params)
|
|
while True:
|
|
j = self.get(url, params)
|
|
yield j.get(key) or ({} if key in ("snapshots", "bars") else [])
|
|
tok = j.get("next_page_token")
|
|
if not tok:
|
|
return
|
|
params["page_token"] = tok
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# inputs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def pick_session(api: Api, now: dt.datetime, force: bool) -> dt.date | None:
|
|
today = now.astimezone(NY).date()
|
|
cal = api.get(f"{TRADE}/v2/calendar", {"start": (today - dt.timedelta(days=10)).isoformat(),
|
|
"end": (today + dt.timedelta(days=10)).isoformat()})
|
|
days = []
|
|
for c in cal:
|
|
d = dt.date.fromisoformat(c["date"])
|
|
at = lambda hm: dt.datetime.combine(d, dt.time(int(hm[:2]), int(hm[3:5])), NY) # noqa: E731
|
|
days.append((d, at(c["open"]), at(c["close"])))
|
|
done = [x for x in days if x[2] + dt.timedelta(minutes=10) <= now]
|
|
if not done:
|
|
print("skip: no session in the calendar window has closed")
|
|
return None
|
|
d, _, _ = done[-1]
|
|
nxt = [x for x in days if x[0] > d]
|
|
if nxt and now >= nxt[0][1] and not force:
|
|
print(f"skip: session {d} is over and {nxt[0][0]} has opened; quotes would be from another day")
|
|
return None
|
|
return d
|
|
|
|
|
|
def universe() -> list[str]:
|
|
with open(UNIVERSE) as fh:
|
|
ticks = json.load(fh)["tickers"]
|
|
return sorted({t.replace("-", ".") for t in ticks} | set(EXTRA))
|
|
|
|
|
|
def pit_universe(start: str, end: str) -> list[str]:
|
|
"""Every name in the S&P 500 at some point in [start, end], plus EXTRA."""
|
|
with open(HISTORY) as fh:
|
|
iv = json.load(fh)["intervals"]
|
|
keep = {t for t, spans in iv.items()
|
|
if any((a or "0000") <= end and (b is None or b >= start) for a, b in spans)}
|
|
return sorted({t.replace("-", ".") for t in keep} | set(universe()))
|
|
|
|
|
|
def stock_bars(api: Api, syms: list[str], start: str, end: dt.datetime) -> pd.DataFrame:
|
|
rows = []
|
|
for i in range(0, len(syms), 100):
|
|
params = {"symbols": ",".join(syms[i:i + 100]), "timeframe": "1Day", "start": start,
|
|
"end": end.isoformat(), "feed": "sip", "adjustment": "raw", "limit": 10000}
|
|
for bars in api.pages(f"{DATA}/v2/stocks/bars", params, "bars"):
|
|
for sym, bl in bars.items():
|
|
rows += [{"date": b["t"][:10], "symbol": sym, "close": b["c"], "volume": b["v"]} for b in bl]
|
|
return pd.DataFrame(rows, columns=["date", "symbol", "close", "volume"])
|
|
|
|
|
|
def list_contracts(api: Api, syms: list[str], gte: str, lte: str, status: str = "active") -> list[dict]:
|
|
out = []
|
|
for i in range(0, len(syms), 10):
|
|
params = {"underlying_symbols": ",".join(syms[i:i + 10]), "status": status,
|
|
"expiration_date_gte": gte, "expiration_date_lte": lte, "limit": 10000}
|
|
try:
|
|
for page in api.pages(f"{TRADE}/v2/options/contracts", params, "option_contracts"):
|
|
for c in page:
|
|
out.append({"symbol": c["symbol"], "underlying": c["underlying_symbol"],
|
|
"type": "C" if c["type"] == "call" else "P", "expiry": c["expiration_date"],
|
|
"strike": float(c["strike_price"]),
|
|
"oi": float(c["open_interest"]) if c.get("open_interest") else np.nan,
|
|
"oi_date": c.get("open_interest_date")})
|
|
except RuntimeError as exc:
|
|
print(f"contracts {syms[i]}..: {exc}", flush=True)
|
|
return out
|
|
|
|
|
|
def option_bars(api: Api, symbols: list[str], start: str, end: str, timeframe: str = "1Day") -> list[dict]:
|
|
"""OPRA-derived bars. Contracts adjusted for corporate actions (non-standard
|
|
roots) are skipped: the endpoint rejects their symbols."""
|
|
symbols = [x for x in symbols if OCC.match(x)]
|
|
# Basic cannot read the last 15 minutes, and a date-only end of today counts
|
|
# as all of today (HTTP 403 "OPRA agreement is not signed"): send a timestamp
|
|
# no later than 16 minutes ago.
|
|
e = (dt.datetime.combine(dt.date.fromisoformat(end), dt.time(0), NY) if len(end) == 10
|
|
else dt.datetime.fromisoformat(end))
|
|
end = min(e, dt.datetime.now(NY) - dt.timedelta(minutes=16)).isoformat()
|
|
|
|
def fetch(chunk: list[str]) -> list[dict]:
|
|
params = {"symbols": ",".join(chunk), "timeframe": timeframe, "start": start, "end": end, "limit": 10000}
|
|
got = []
|
|
try:
|
|
for bars in api.pages(f"{DATA}/v1beta1/options/bars", params, "bars"):
|
|
for sym, bl in bars.items():
|
|
got += [{"symbol": sym, "t": b["t"], "o": b["o"], "h": b["h"], "l": b["l"], "c": b["c"],
|
|
"v": b["v"], "n": b.get("n"), "vw": b.get("vw")} for b in bl]
|
|
except BackendTimeout: # restart the chunk as two halves
|
|
h = len(chunk) // 2
|
|
return fetch(chunk[:h]) + fetch(chunk[h:])
|
|
return got
|
|
|
|
rows = []
|
|
for i in range(0, len(symbols), 1000):
|
|
rows += fetch(symbols[i:i + 1000])
|
|
return rows
|
|
|
|
|
|
def chain(api: Api, sym: str, close: float, session: dt.date) -> list[dict]:
|
|
params = {"feed": "indicative", "limit": 1000,
|
|
"expiration_date_gte": (session + dt.timedelta(days=DTE_MIN)).isoformat(),
|
|
"expiration_date_lte": (session + dt.timedelta(days=DTE_MAX)).isoformat(),
|
|
"strike_price_gte": round(MNY_LO * close, 2), "strike_price_lte": round(MNY_HI * close, 2)}
|
|
rows = []
|
|
for snaps in api.pages(f"{DATA}/v1beta1/options/snapshots/{sym}", params, "snapshots"):
|
|
for occ, s in snaps.items():
|
|
q, t, b, g = s.get("latestQuote") or {}, s.get("latestTrade") or {}, s.get("dailyBar") or {}, s.get("greeks") or {}
|
|
today = b.get("t", "")[:10] == session.isoformat()
|
|
rows.append({"underlying": sym, "symbol": occ, "type": occ[-9],
|
|
"expiry": f"20{occ[-15:-13]}-{occ[-13:-11]}-{occ[-11:-9]}",
|
|
"strike": int(occ[-8:]) / 1000, "bid": q.get("bp"), "ask": q.get("ap"),
|
|
"bid_size": q.get("bs"), "ask_size": q.get("as"), "quote_time": q.get("t"),
|
|
"ind_last": t.get("p"), "ind_last_time": t.get("t"),
|
|
"ind_day_volume": b.get("v", 0) if today else 0,
|
|
"alp_iv": s.get("impliedVolatility"), "delta": g.get("delta"), "gamma": g.get("gamma"),
|
|
"theta": g.get("theta"), "vega": g.get("vega")})
|
|
return rows
|
|
|
|
|
|
def rate(session: dt.date) -> tuple[float, str]:
|
|
try:
|
|
rf = pd.read_csv(RF_PATH, index_col=0, parse_dates=True).iloc[:, 0]
|
|
rf = pd.to_numeric(rf, errors="coerce").dropna().loc[:pd.Timestamp(session)]
|
|
return float(rf.iloc[-1]) / 100, rf.index[-1].date().isoformat()
|
|
except (OSError, IndexError):
|
|
return 0.04, "default"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# implied volatility and per-underlying summary
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def black76(F, K, T, sig, call, disc):
|
|
sd = sig * np.sqrt(T)
|
|
d1 = (np.log(F / K) + 0.5 * sd ** 2) / sd
|
|
d2 = d1 - sd
|
|
return np.where(call, disc * (F * ndtr(d1) - K * ndtr(d2)), disc * (K * ndtr(-d2) - F * ndtr(-d1)))
|
|
|
|
|
|
def implied_vol(price, F, K, T, call, disc, iters: int = 80) -> np.ndarray:
|
|
price, F, K, T, call, disc = (np.asarray(x, dtype=float if i != 4 else bool)
|
|
for i, x in enumerate((price, F, K, T, call, disc)))
|
|
lo, hi = np.full(len(price), 1e-4), np.full(len(price), 5.0)
|
|
intrinsic = disc * np.where(call, np.maximum(F - K, 0), np.maximum(K - F, 0))
|
|
ok = np.isfinite(price) & (price > intrinsic + 1e-6) & (price < black76(F, K, T, hi, call, disc))
|
|
for _ in range(iters):
|
|
mid = (lo + hi) / 2
|
|
above = black76(F, K, T, mid, call, disc) > price
|
|
hi, lo = np.where(above, mid, hi), np.where(above, lo, mid)
|
|
return np.where(ok, (lo + hi) / 2, np.nan)
|
|
|
|
|
|
def add_iv(df: pd.DataFrame, r: float) -> pd.DataFrame:
|
|
df["mid"] = (df["bid"] + df["ask"]) / 2
|
|
good = (df["bid"] > 0) & (df["ask"] > df["bid"])
|
|
df.loc[~good, "mid"] = np.nan
|
|
df["T"] = df["dte"] / 365.0
|
|
disc = np.exp(-r * df["T"])
|
|
df["fwd"] = np.nan
|
|
for _, g in df.groupby(["underlying", "expiry"]):
|
|
piv = g.pivot_table(index="strike", columns="type", values="mid")
|
|
if {"C", "P"} <= set(piv.columns):
|
|
piv = piv.dropna()
|
|
if len(piv):
|
|
S = g["stock_close"].iloc[0]
|
|
near = piv.iloc[np.argsort(np.abs(piv.index.to_numpy() - S))[:3]]
|
|
f = near.index.to_numpy() + np.exp(r * g["T"].iloc[0]) * (near["C"] - near["P"]).to_numpy()
|
|
df.loc[g.index, "fwd"] = float(np.median(f))
|
|
nofwd = df["fwd"].isna()
|
|
df.loc[nofwd, "fwd"] = df.loc[nofwd, "stock_close"] * np.exp(r * df.loc[nofwd, "T"])
|
|
call = (df["type"] == "C").to_numpy()
|
|
args = (df["strike"].to_numpy(), df["T"].to_numpy(), call, disc.to_numpy())
|
|
df["iv_fwd"] = implied_vol(df["mid"].to_numpy(), df["fwd"].to_numpy(), *args)
|
|
df["iv_spot"] = implied_vol(df["mid"].to_numpy(), (df["stock_close"] * np.exp(r * df["T"])).to_numpy(), *args)
|
|
return df
|
|
|
|
|
|
def summarize(g: pd.DataFrame, vol: dict) -> dict:
|
|
S = g["stock_close"].iloc[0]
|
|
v = g[g["iv_fwd"].notna()]
|
|
out = {"underlying": g["underlying"].iloc[0], "stock_close": S, "stock_volume": g["stock_volume"].iloc[0],
|
|
"n_contracts": len(g), "n_valid": len(v),
|
|
"med_rel_spread": float(((v["ask"] - v["bid"]) / v["mid"]).median()) if len(v) else np.nan}
|
|
atm = {}
|
|
for _, x in v.groupby("expiry"):
|
|
F = x["fwd"].iloc[0]
|
|
k = x["strike"].iloc[np.argmin(np.abs(x["strike"].to_numpy() - F))]
|
|
atm[int(x["dte"].iloc[0])] = float(x.loc[x["strike"] == k, "iv_fwd"].mean())
|
|
ds = sorted(d for d in atm if 7 <= d <= DTE_MAX)
|
|
if ds:
|
|
lo = [d for d in ds if d <= 30]
|
|
hi = [d for d in ds if d >= 30]
|
|
if lo and hi and lo[-1] != hi[0]: # total variance, linear in time
|
|
a, b = lo[-1], hi[0]
|
|
w = atm[a] ** 2 * a + (atm[b] ** 2 * b - atm[a] ** 2 * a) * (30 - a) / (b - a)
|
|
out["iv_atm30"] = float(np.sqrt(w / 30)) if w > 0 else np.nan
|
|
else: # 30 days exactly, or only one side: nearest
|
|
out["iv_atm30"] = atm[hi[0]] if hi else atm[lo[-1]]
|
|
mid_exp = v[(v["dte"] >= 10) & (v["dte"] <= 60)]
|
|
if len(mid_exp):
|
|
e = mid_exp.loc[(mid_exp["dte"] - 30).abs().idxmin(), "expiry"]
|
|
x = mid_exp[mid_exp["expiry"] == e]
|
|
m = x["strike"] / S
|
|
puts = x[(x["type"] == "P") & (m >= 0.80) & (m <= 0.95)]
|
|
calls = x[(x["type"] == "C") & (m >= 0.95) & (m <= 1.05)]
|
|
if len(puts) and len(calls):
|
|
p = puts.loc[(puts["strike"] / S - 0.95).abs().idxmin(), "iv_fwd"]
|
|
c = calls.loc[(calls["strike"] / S - 1.0).abs().idxmin(), "iv_fwd"]
|
|
out["skew_xzz"] = float(p - c)
|
|
pairs = mid_exp[(mid_exp["strike"] / S).between(0.8, 1.2) & mid_exp["iv_spot"].notna()]
|
|
pv = pairs.pivot_table(index=["expiry", "strike"], columns="type", values=["iv_spot", "oi"])
|
|
if ("iv_spot", "C") in pv.columns and ("iv_spot", "P") in pv.columns:
|
|
pv = pv.dropna(subset=[("iv_spot", "C"), ("iv_spot", "P")])
|
|
if len(pv):
|
|
w = pv[("oi", "C")].fillna(0) + pv[("oi", "P")].fillna(0) if ("oi", "C") in pv and ("oi", "P") in pv \
|
|
else pd.Series(1.0, index=pv.index)
|
|
w = w if w.sum() > 0 else pd.Series(1.0, index=pv.index)
|
|
out["cp_spread"] = float(((pv[("iv_spot", "C")] - pv[("iv_spot", "P")]) * w).sum() / w.sum())
|
|
cv, pv_ = vol.get("C", np.nan), vol.get("P", np.nan)
|
|
co, po = g.loc[g["type"] == "C", "oi"].sum(), g.loc[g["type"] == "P", "oi"].sum()
|
|
out.update({"call_volume": cv, "put_volume": pv_, "call_oi": co, "put_oi": po,
|
|
"pc_volume": pv_ / cv if cv else np.nan, "pc_oi": po / co if co else np.nan,
|
|
"os_volume": (cv + pv_) * 100 / out["stock_volume"] if out["stock_volume"] else np.nan})
|
|
return out
|
|
|
|
|
|
def quality(api: Api, df: pd.DataFrame, session: dt.date) -> tuple[pd.DataFrame, dict]:
|
|
"""Last OPRA print in 15:45-16:00 vs the indicative quote, near the money."""
|
|
x = df[df["dte"].between(14, 45) & df["mid"].notna() & df["iv_fwd"].notna()].copy()
|
|
if x.empty:
|
|
return x, {}
|
|
x["m_f"] = (x["strike"] / x["fwd"] - 1).abs()
|
|
x["m_95"] = (x["strike"] / x["stock_close"] - 0.95).abs()
|
|
pick = []
|
|
for _, g in x.groupby("underlying"):
|
|
e = g.loc[(g["dte"] - 30).abs().idxmin(), "expiry"]
|
|
g = g[g["expiry"] == e]
|
|
for typ, col in (("C", "m_f"), ("P", "m_f"), ("P", "m_95")):
|
|
h = g[g["type"] == typ]
|
|
if len(h):
|
|
pick.append(h[col].idxmin())
|
|
s = x.loc[sorted(set(pick))].copy()
|
|
t0 = dt.datetime.combine(session, dt.time(15, 30), NY).isoformat()
|
|
t1 = dt.datetime.combine(session, dt.time(16, 0), NY).isoformat()
|
|
bars = pd.DataFrame(option_bars(api, s["symbol"].tolist(), t0, t1, "1Min"))
|
|
if bars.empty:
|
|
return s.iloc[:0], {}
|
|
bars["t"] = pd.to_datetime(bars["t"], utc=True).dt.tz_convert(NY)
|
|
last = bars.sort_values("t").groupby("symbol").tail(1).set_index("symbol")
|
|
s = s.join(last[["c", "t"]].rename(columns={"c": "print", "t": "print_time"}), on="symbol")
|
|
cut = pd.Timestamp(dt.datetime.combine(session, dt.time(15, 45), NY))
|
|
s = s[s["print_time"].notna() & (s["print_time"] >= cut)].copy()
|
|
if s.empty:
|
|
return s, {}
|
|
s["inside"] = (s["print"] >= s["bid"] - 0.005) & (s["print"] <= s["ask"] + 0.005)
|
|
s["pos"] = (s["print"] - s["mid"]) / (s["ask"] - s["bid"])
|
|
s["iv_print"] = implied_vol(s["print"], s["fwd"], s["strike"], s["T"], s["type"] == "C", np.exp(-s["r"] * s["T"]))
|
|
s["iv_err"] = s["iv_print"] - s["iv_fwd"]
|
|
s["rel_spread"] = (s["ask"] - s["bid"]) / s["mid"]
|
|
summ = {"q_n": int(len(s)), "q_inside": float(s["inside"].mean()), "q_abs_pos": float(s["pos"].abs().median()),
|
|
"q_abs_iv_err": float(s["iv_err"].abs().median()), "q_rel_spread": float(s["rel_spread"].median())}
|
|
cols = ["session", "underlying", "symbol", "type", "expiry", "dte", "strike", "stock_close", "fwd", "bid", "ask",
|
|
"quote_time", "print", "print_time", "inside", "pos", "iv_fwd", "iv_print", "iv_err", "rel_spread"]
|
|
return s[cols], summ
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# commands
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _atomic_csv(df: pd.DataFrame, path: str) -> None:
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
df.to_csv(path + ".tmp", index=False, compression="gzip" if path.endswith(".gz") else None)
|
|
os.replace(path + ".tmp", path)
|
|
|
|
|
|
def _bars_frame(rows: list[dict], meta: pd.DataFrame) -> pd.DataFrame:
|
|
if not rows:
|
|
return pd.DataFrame(columns=BAR_COLS)
|
|
b = pd.DataFrame(rows)
|
|
b["date"] = b.pop("t").str[:10]
|
|
b = b.join(meta.set_index("symbol")[["underlying", "type", "expiry", "strike"]], on="symbol")
|
|
return b[BAR_COLS]
|
|
|
|
|
|
def snapshot(args) -> None:
|
|
out = args.out or OUT
|
|
now = dt.datetime.now(NY)
|
|
api = Api()
|
|
session = dt.date.fromisoformat(args.session) if args.session else pick_session(api, now, args.force)
|
|
if session is None:
|
|
return
|
|
path = os.path.join(out, "chains", str(session.year), f"chain_{session}.csv.gz")
|
|
if os.path.exists(path) and not args.force:
|
|
print(f"{path} exists; nothing to do")
|
|
return
|
|
syms = universe()[: args.limit] if args.limit else universe()
|
|
t0 = time.time()
|
|
end = min(now - dt.timedelta(minutes=16), dt.datetime.combine(session + dt.timedelta(days=1), dt.time(0), NY))
|
|
st = stock_bars(api, syms, session.isoformat(), end)
|
|
st = st[st["date"] == session.isoformat()]
|
|
_atomic_csv(st.assign(session=session.isoformat()), os.path.join(out, "stock", f"stock_{session}.csv"))
|
|
px = st.set_index("symbol")[["close", "volume"]].to_dict("index")
|
|
have = [s for s in syms if s in px]
|
|
# contracts expiring on the session itself may already be inactive at 16:25,
|
|
# and 0DTE carries much of the day's volume: list both, as the backfill does
|
|
meta = pd.DataFrame(list_contracts(api, have, session.isoformat(), (session + dt.timedelta(days=DTE_MAX)).isoformat())
|
|
+ list_contracts(api, have, session.isoformat(), session.isoformat(), "inactive"))
|
|
meta = meta.drop_duplicates("symbol")
|
|
rows, failed = [], []
|
|
for k, sym in enumerate(syms):
|
|
if sym not in px:
|
|
failed.append([sym, "no stock bar"])
|
|
continue
|
|
try:
|
|
rs = chain(api, sym, px[sym]["close"], session)
|
|
except RuntimeError as exc:
|
|
failed.append([sym, str(exc)[:120]])
|
|
continue
|
|
for r in rs:
|
|
r.update(stock_close=px[sym]["close"], stock_volume=px[sym]["volume"])
|
|
rows += rs
|
|
if (k + 1) % 100 == 0:
|
|
print(f" {k + 1}/{len(syms)} underlyings, {len(rows)} contracts, {api.n} requests "
|
|
f"[{time.time() - t0:.0f}s]", flush=True)
|
|
if not rows:
|
|
raise SystemExit("no contracts collected")
|
|
bars_err = None
|
|
try:
|
|
bars = _bars_frame(option_bars(api, meta["symbol"].tolist(), session.isoformat(),
|
|
(session + dt.timedelta(days=1)).isoformat()), meta)
|
|
bars = bars[bars["date"] == session.isoformat()]
|
|
except RuntimeError as exc:
|
|
bars, bars_err = pd.DataFrame(columns=BAR_COLS), str(exc)[:200]
|
|
_atomic_csv(bars, os.path.join(out, "bars", str(session.year), f"bars_{session}.csv.gz"))
|
|
df = pd.DataFrame(rows)
|
|
df = df.join(meta.set_index("symbol")[["oi", "oi_date"]], on="symbol")
|
|
df = df.join(bars.set_index("symbol")[["c", "v", "vw", "n"]].rename(
|
|
columns={"c": "opra_close", "v": "opra_volume", "vw": "opra_vwap", "n": "opra_trades"}), on="symbol")
|
|
df.insert(0, "session", session.isoformat())
|
|
df["dte"] = (pd.to_datetime(df["expiry"]) - pd.Timestamp(session)).dt.days
|
|
r, r_date = rate(session)
|
|
df["r"] = r
|
|
df = add_iv(df, r)
|
|
try:
|
|
qdf, qsum = quality(api, df, session)
|
|
_atomic_csv(qdf, os.path.join(out, "quality", f"quality_{session}.csv"))
|
|
except RuntimeError as exc:
|
|
qsum = {"q_error": str(exc)[:200]}
|
|
_atomic_csv(df.drop(columns=["mid", "T"]), path)
|
|
vol = bars.groupby(["underlying", "type"])["v"].sum()
|
|
summ = pd.DataFrame([summarize(g, vol.get(u, pd.Series(dtype=float)).to_dict())
|
|
for u, g in df.groupby("underlying")])
|
|
summ.insert(0, "session", session.isoformat())
|
|
_atomic_csv(summ, os.path.join(out, "summary", f"summary_{session}.csv"))
|
|
rec = {"session": session.isoformat(), "started": dt.datetime.fromtimestamp(t0, NY).isoformat(timespec="seconds"),
|
|
"finished": dt.datetime.now(NY).isoformat(timespec="seconds"), "underlyings": len(syms),
|
|
"with_chain": int(df["underlying"].nunique()), "contracts": len(df),
|
|
"valid_iv": int(df["iv_fwd"].notna().sum()), "bars": len(bars), "bars_error": bars_err,
|
|
"requests": api.n, "r": r, "r_date": r_date, "feed": "indicative", **qsum,
|
|
"failed": failed[:50], "n_failed": len(failed), "forced": bool(args.force)}
|
|
with open(os.path.join(out, "manifest.jsonl"), "a") as fh:
|
|
fh.write(json.dumps(rec) + "\n")
|
|
print(f"session {session}: {rec['with_chain']}/{len(syms)} underlyings, {len(df)} quoted contracts "
|
|
f"({rec['valid_iv']} with IV), {len(bars)} OPRA bars, quality {qsum}, {api.n} requests, "
|
|
f"{time.time() - t0:.0f}s, failed {len(failed)}")
|
|
|
|
|
|
def backfill(args) -> None:
|
|
out = args.out or OUT
|
|
api = Api(min_interval=0.4, yield_to_snapshot=True) # ~150 requests a minute
|
|
end = args.end or (dt.datetime.now(NY).date() - dt.timedelta(days=1)).isoformat()
|
|
names = args.only or pit_universe(args.start, end)
|
|
hist = os.path.join(out, "bars_hist")
|
|
os.makedirs(hist, exist_ok=True)
|
|
raw = os.path.join(out, "stock_raw.csv.gz")
|
|
if not os.path.exists(raw) or args.only:
|
|
st = stock_bars(api, names, args.start, dt.datetime.combine(dt.date.fromisoformat(end), dt.time(23, 59), NY))
|
|
if not args.only:
|
|
_atomic_csv(st, raw)
|
|
print(f"stock bars: {st['symbol'].nunique()} symbols, {len(st)} rows", flush=True)
|
|
todo = [u for u in names if not os.path.exists(os.path.join(hist, f"{u}.csv.gz"))]
|
|
print(f"backfill {args.start}..{end}: {len(names)} underlyings, {len(todo)} to do", flush=True)
|
|
lte_active = (dt.date.fromisoformat(end) + dt.timedelta(days=DTE_MAX)).isoformat()
|
|
for k, u in enumerate(todo):
|
|
try:
|
|
backfill_one(api, u, args.start, end, lte_active, hist, out, k, len(todo))
|
|
except RuntimeError as exc: # no file is written, so the next run retries it
|
|
print(f" [{k + 1}/{len(todo)}] {u}: failed, will retry on the next run: {exc}", flush=True)
|
|
if not args.only and not args.no_panel:
|
|
volume_panel(args)
|
|
|
|
|
|
def backfill_one(api: Api, u: str, start: str, end: str, lte_active: str, hist: str, out: str, k: int, n: int) -> None:
|
|
t0, n0 = time.time(), api.n
|
|
meta = pd.DataFrame(list_contracts(api, [u], start, end, "inactive")
|
|
+ list_contracts(api, [u], start, lte_active, "active"))
|
|
frames = []
|
|
if len(meta):
|
|
meta = meta.drop_duplicates("symbol")
|
|
meta["month"] = meta["expiry"].str[:7]
|
|
for _, g in meta.groupby("month"):
|
|
first, last = g["expiry"].min(), g["expiry"].max()
|
|
a = max(start, (dt.date.fromisoformat(first) - dt.timedelta(days=DTE_MAX)).isoformat())
|
|
b = min(end, last)
|
|
if a > b:
|
|
continue
|
|
f = _bars_frame(option_bars(api, g["symbol"].tolist(), a, (dt.date.fromisoformat(b)
|
|
+ dt.timedelta(days=1)).isoformat()), g)
|
|
lo = (pd.to_datetime(f["expiry"]) - pd.Timedelta(days=DTE_MAX)).dt.strftime("%Y-%m-%d")
|
|
frames.append(f[(f["date"] >= lo) & (f["date"] <= f["expiry"]) & (f["date"] <= end)])
|
|
frames = [f for f in frames if len(f)]
|
|
bars = pd.concat(frames) if frames else pd.DataFrame(columns=BAR_COLS)
|
|
_atomic_csv(bars, os.path.join(hist, f"{u}.csv.gz"))
|
|
rec = {"underlying": u, "contracts": int(len(meta)), "bars": int(len(bars)), "requests": api.n - n0,
|
|
"seconds": round(time.time() - t0, 1), "start": start, "end": end,
|
|
"finished": dt.datetime.now(NY).isoformat(timespec="seconds")}
|
|
with open(os.path.join(out, "backfill_manifest.jsonl"), "a") as fh:
|
|
fh.write(json.dumps(rec) + "\n")
|
|
print(f" [{k + 1}/{n}] {u}: {rec['contracts']} contracts, {rec['bars']} bars, "
|
|
f"{rec['requests']} requests, {rec['seconds']}s (total {api.n})", flush=True)
|
|
|
|
|
|
def volume_panel(args) -> None:
|
|
"""Daily option volume per underlying from the OPRA bars (backfill + daily
|
|
files): all strikes with 0-100 days to expiry, and the band the quote
|
|
snapshot covers (5-100 days, 0.7-1.3 x the unadjusted close)."""
|
|
out = args.out or OUT
|
|
stock = [pd.read_csv(os.path.join(out, "stock_raw.csv.gz"))] if os.path.exists(os.path.join(out, "stock_raw.csv.gz")) else []
|
|
stock += [pd.read_csv(f).drop(columns=["session"], errors="ignore") for f in glob.glob(os.path.join(out, "stock", "*.csv"))]
|
|
stock = pd.concat(stock).drop_duplicates(["date", "symbol"], keep="last").rename(columns={"symbol": "underlying"})
|
|
files = sorted(glob.glob(os.path.join(out, "bars_hist", "*.csv.gz")))
|
|
daily = sorted(glob.glob(os.path.join(out, "bars", "*", "*.csv.gz")))
|
|
bman = os.path.join(out, "backfill_manifest.jsonl")
|
|
hist_end = max((json.loads(x)["end"] for x in open(bman) if x.strip()), default="") if os.path.exists(bman) else ""
|
|
parts = []
|
|
for f in files + daily:
|
|
b = pd.read_csv(f)
|
|
if b.empty:
|
|
continue
|
|
if f in daily:
|
|
b = b[b["date"] > hist_end] # the backfill wins where both exist
|
|
b = b.merge(stock[["date", "underlying", "close"]], on=["date", "underlying"], how="left")
|
|
dte = (pd.to_datetime(b["expiry"]) - pd.to_datetime(b["date"])).dt.days
|
|
band = dte.between(DTE_MIN, DTE_MAX) & (b["strike"] / b["close"]).between(MNY_LO, MNY_HI)
|
|
b["v_band"] = b["v"].where(band, 0)
|
|
g = b.groupby(["date", "underlying", "type"])[["v", "v_band"]].sum().unstack("type").fillna(0)
|
|
g.columns = [f"{a}_{t}" for a, t in g.columns]
|
|
parts.append(g.reset_index())
|
|
if not parts:
|
|
print("no bars yet")
|
|
return
|
|
P = pd.concat(parts).groupby(["date", "underlying"]).sum().reset_index()
|
|
for c in ("v_C", "v_P", "v_band_C", "v_band_P"):
|
|
if c not in P:
|
|
P[c] = 0.0
|
|
P = P.merge(stock[["date", "underlying", "close", "volume"]], on=["date", "underlying"], how="left")
|
|
P["os_volume"] = (P["v_C"] + P["v_P"]) * 100 / P["volume"]
|
|
P["os_volume_band"] = (P["v_band_C"] + P["v_band_P"]) * 100 / P["volume"]
|
|
P["pc_volume"] = P["v_P"] / P["v_C"].replace(0, np.nan)
|
|
_atomic_csv(P.sort_values(["date", "underlying"]), os.path.join(out, "volume_panel.csv.gz"))
|
|
print(f"volume panel: {P['underlying'].nunique()} underlyings, {P['date'].nunique()} days "
|
|
f"({P['date'].min()}..{P['date'].max()}), {len(P)} rows")
|
|
|
|
|
|
def status(args) -> None:
|
|
out = args.out or OUT
|
|
man = os.path.join(out, "manifest.jsonl")
|
|
if os.path.exists(man):
|
|
recs = [json.loads(x) for x in open(man) if x.strip()]
|
|
print(f"snapshots: {len(recs)} runs, {len({r['session'] for r in recs})} sessions, "
|
|
f"{recs[0]['session']}..{recs[-1]['session']}")
|
|
for r in recs[-5:]:
|
|
print(f" {r['session']} {r['with_chain']}/{r['underlyings']} underlyings, {r['contracts']} contracts, "
|
|
f"{r.get('bars', '-')} OPRA bars, failed {r['n_failed']}, {r['started']} -> {r['finished']}")
|
|
q = [r for r in recs if r.get("q_n")]
|
|
q = list({r["session"]: r for r in q}.values())
|
|
if q:
|
|
ins = float(np.average([r["q_inside"] for r in q], weights=[r["q_n"] for r in q]))
|
|
err = float(np.median([r["q_abs_iv_err"] for r in q]))
|
|
print(f"quote quality over {len(q)} sessions: {ins:.1%} of prints inside the indicative quote, "
|
|
f"median |IV(print) - IV(mid)| {err * 100:.2f} vol points")
|
|
if len(q) >= QUALITY_MIN_SESSIONS:
|
|
ok = ins >= QUALITY_INSIDE and err <= QUALITY_IV
|
|
print(" verdict: indicative quotes adequate" if ok else " verdict: consider the OPRA plan")
|
|
else:
|
|
print(f" verdict after {QUALITY_MIN_SESSIONS} sessions")
|
|
else:
|
|
print("snapshots: none yet")
|
|
bman = os.path.join(out, "backfill_manifest.jsonl")
|
|
if os.path.exists(bman):
|
|
b = [json.loads(x) for x in open(bman) if x.strip()]
|
|
print(f"backfill: {len(b)} underlyings done, {sum(x['bars'] for x in b):,} bars, last {b[-1]['underlying']} "
|
|
f"at {b[-1]['finished']}")
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser()
|
|
sub = ap.add_subparsers(dest="cmd", required=True)
|
|
s = sub.add_parser("snapshot")
|
|
s.add_argument("--limit", type=int, help="first N underlyings only (testing)")
|
|
s.add_argument("--force", action="store_true", help="record even if late or already recorded")
|
|
s.add_argument("--session", help="label the snapshot with this session date (testing)")
|
|
s.add_argument("--out", help="output directory (default data/options/us)")
|
|
s.set_defaults(fn=snapshot)
|
|
b = sub.add_parser("backfill")
|
|
b.add_argument("--start", default=HIST_START)
|
|
b.add_argument("--end", help="last date (default yesterday)")
|
|
b.add_argument("--only", nargs="*", help="these underlyings only (testing; no panel)")
|
|
b.add_argument("--no-panel", action="store_true")
|
|
b.add_argument("--out")
|
|
b.set_defaults(fn=backfill)
|
|
v = sub.add_parser("volume-panel")
|
|
v.add_argument("--out")
|
|
v.set_defaults(fn=volume_panel)
|
|
t = sub.add_parser("status")
|
|
t.add_argument("--out")
|
|
t.set_defaults(fn=status)
|
|
args = ap.parse_args()
|
|
args.fn(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|