""" Memory-conscious loader and point-in-time query utilities for iVolatility-style end-of-day SPY (or other) options CSV dumps. Typical workflow ---------------- **Single CSV** 1. ``IVolatilityLoader.convert_to_parquet("huge.csv", "cache/options.parquet")`` **Several shards** (e.g. ``data_download.csv`` + ``data_download 2.csv`` …) 1. ``paths = IVolatilityLoader.sorted_ivolatility_csv_paths("SPY-Option-Data")`` 2. ``IVolatilityLoader.combine_csvs_to_parquet(paths, "cache/options.parquet")`` **Query** 3. ``loader = IVolatilityLoader("cache/options.parquet")`` 4. ``chain = loader.get_chain_for_date(pd.Timestamp("2013-02-15"))`` 5. ``leg = chain.find_target_leg(target_dte=30, target_delta=-0.10, option_type="P")`` Dependencies: pandas, pyarrow (for Parquet I/O and predicate pushdown). """ from __future__ import annotations import math from dataclasses import dataclass, field from pathlib import Path from typing import Any, ClassVar, Iterator, Sequence import pandas as pd try: import pyarrow as pa import pyarrow.parquet as pq except ImportError as e: # pragma: no cover - import guard for optional dep pa = None # type: ignore[assignment] pq = None # type: ignore[assignment] _PYARROW_IMPORT_ERROR = e else: _PYARROW_IMPORT_ERROR = None # --------------------------------------------------------------------------- # 1. Data structures # --------------------------------------------------------------------------- @dataclass class OptionContract: """ One EOD option quote + greeks on a single ``date`` (as-of / trade date). ``option_type`` is normalized to 'C' (call) or 'P' (put). """ date: pd.Timestamp expiration: pd.Timestamp strike: float option_type: str bid: float ask: float mid: float iv: float delta: float gamma: float theta: float vega: float @dataclass class OptionChain: """ All contracts observed on one trading day (same ``as_of``). ``as_of`` is the chain date used for DTE = (expiration - as_of).days in ``find_target_leg``. """ as_of: pd.Timestamp contracts: list[OptionContract] = field(default_factory=list) def find_target_leg( self, target_dte: int, target_delta: float, option_type: str, ) -> OptionContract: """ Pick one contract: closest expiration by DTE, then closest |delta - target|. Parameters ---------- target_dte Desired days-to-expiration (calendar days between chain date and expiry). target_delta Model delta to match (e.g. -0.10 for a ~10-delta put; calls are usually positive). option_type 'C', 'P', 'Call', 'Put', etc. (normalized to single-letter). """ if not self.contracts: raise ValueError("OptionChain is empty; cannot find target leg.") want = _normalize_option_type(option_type) typed = [c for c in self.contracts if c.option_type == want] if not typed: raise ValueError(f"No contracts with option_type={want!r} on {self.as_of.date()}.") as_of = pd.Timestamp(self.as_of).normalize() # Unique expirations with their DTE vs chain date expiries: dict[pd.Timestamp, int] = {} for c in typed: exp = pd.Timestamp(c.expiration).normalize() dte = int((exp - as_of).days) if dte < 0: continue # expired before as_of; skip expiries[exp] = dte if not expiries: raise ValueError("No non-expired expirations for the requested option type.") # Closest expiration bucket to target_dte (min absolute error) best_exp, best_dte = min(expiries.items(), key=lambda kv: abs(kv[1] - target_dte)) bucket = [ c for c in typed if pd.Timestamp(c.expiration).normalize() == best_exp and math.isfinite(c.delta) ] if not bucket: raise ValueError( f"No contracts with finite delta at expiration {best_exp.date()} " f"(DTE≈{best_dte}, target DTE={target_dte})." ) # Closest delta in absolute terms return min(bucket, key=lambda c: abs(float(c.delta) - float(target_delta))) # --------------------------------------------------------------------------- # 2. Loader + Parquet conversion # --------------------------------------------------------------------------- class IVolatilityLoader: """ Load iVolatility CSV (via Parquet cache) and query single-day chains. ``COLUMN_MAP`` maps **raw CSV header** -> **internal column name** used in Parquet and in memory. Adjust keys to match your export (e.g. some files use ``implied_volatility`` instead of ``iv``). """ # Raw iVolatility header -> canonical name (edit keys to match your file). COLUMN_MAP: ClassVar[dict[str, str]] = { "date": "date", "expiration": "expiration", "strike": "strike", "call/put": "option_type", # Alternate spellings you may see in other downloads: "call_put": "option_type", "Call/Put": "option_type", "implied_volatility": "iv", "IV": "iv", "bid": "bid", "ask": "ask", "price": "price", # often used as last/mid; see _standardize_chunk "mean price": "price", # alternate iVolatility export (e.g. some split files) "delta": "delta", "gamma": "gamma", "theta": "theta", "vega": "vega", "rho": "rho", # ingested if present; not stored on OptionContract } # Canonical columns written to Parquet (rho optional drop later) _PARQUET_COLS: ClassVar[list[str]] = [ "date", "expiration", "strike", "option_type", "bid", "ask", "mid", "iv", "delta", "gamma", "theta", "vega", ] def __init__(self, file_path: str | Path) -> None: """ Parameters ---------- file_path Path to a **Parquet** file produced by ``convert_to_parquet``. Using Parquet keeps random access and row-group filters fast and RAM low. """ self.file_path = Path(file_path).expanduser() if self.file_path.suffix.lower() not in {".parquet", ".pq"}: # Soft warning only: some tools use odd extensions pass @staticmethod def convert_to_parquet( csv_path: str | Path, parquet_path: str | Path, *, chunksize: int = 200_000, date_columns: tuple[str, ...] = ("date", "expiration"), float32_cols: tuple[str, ...] | None = None, ) -> Path: """ Stream-read a large CSV in chunks, normalize columns, downcast floats, write Parquet. * Dates are parsed with ``pd.to_datetime`` (UTC-naive midnight normalized). * Greeks, IV, bid/ask/mid/strike use ``float32`` to halve RAM vs float64. * ``mid`` is ``(bid+ask)/2`` when not otherwise supplied from ``price``. Parameters ---------- chunksize Rows per read from CSV; tune based on available RAM and column count. """ if pq is None or pa is None: raise ImportError( "convert_to_parquet requires pyarrow. Install with: pip install pyarrow" ) from _PYARROW_IMPORT_ERROR csv_path = Path(csv_path).expanduser() parquet_path = Path(parquet_path).expanduser() parquet_path.parent.mkdir(parents=True, exist_ok=True) if float32_cols is None: float32_cols = ( "strike", "bid", "ask", "mid", "iv", "delta", "gamma", "theta", "vega", ) reader = pd.read_csv(csv_path, chunksize=chunksize) writer: pq.ParquetWriter | None = None try: for raw_chunk in reader: std = IVolatilityLoader._standardize_chunk( raw_chunk, column_map=IVolatilityLoader.COLUMN_MAP, date_columns=date_columns, float32_cols=float32_cols, ) if std.empty: continue table = pa.Table.from_pandas(std, preserve_index=False) if writer is None: writer = pq.ParquetWriter(str(parquet_path), table.schema) writer.write_table(table) finally: if writer is not None: writer.close() return parquet_path @staticmethod def sorted_ivolatility_csv_paths(directory: str | Path) -> list[Path]: """ Return iVolatility split exports in a stable order: ``data_download.csv`` first, then ``data_download 2.csv`` … ``data_download 7.csv`` by numeric suffix. Use this when combining multiple shards from the same vendor export. """ d = Path(directory).expanduser() paths = sorted(d.glob("data_download*.csv"), key=_ivolatility_csv_sort_key) return paths @staticmethod def combine_csvs_to_parquet( csv_paths: Sequence[str | Path], parquet_path: str | Path, *, chunksize: int = 200_000, date_columns: tuple[str, ...] = ("date", "expiration"), float32_cols: tuple[str, ...] | None = None, ) -> Path: """ Merge several iVolatility CSV shards into **one** Parquet file (streaming). Each file is read in ``chunksize`` rows so memory stays bounded. Schema is unified via ``COLUMN_MAP`` (e.g. ``mean price`` → ``price`` for files that omit ``is_settlement``). """ if pq is None or pa is None: raise ImportError( "combine_csvs_to_parquet requires pyarrow. Install with: pip install pyarrow" ) from _PYARROW_IMPORT_ERROR parquet_path = Path(parquet_path).expanduser() parquet_path.parent.mkdir(parents=True, exist_ok=True) if float32_cols is None: float32_cols = ( "strike", "bid", "ask", "mid", "iv", "delta", "gamma", "theta", "vega", ) writer: pq.ParquetWriter | None = None wrote_any = False try: for csv_path in csv_paths: path = Path(csv_path).expanduser() if not path.is_file(): raise FileNotFoundError(f"CSV not found: {path}") reader = pd.read_csv(path, chunksize=chunksize) for raw_chunk in reader: std = IVolatilityLoader._standardize_chunk( raw_chunk, column_map=IVolatilityLoader.COLUMN_MAP, date_columns=date_columns, float32_cols=float32_cols, ) if std.empty: continue table = pa.Table.from_pandas(std, preserve_index=False) if writer is None: writer = pq.ParquetWriter(str(parquet_path), table.schema) writer.write_table(table) wrote_any = True finally: if writer is not None: writer.close() if not wrote_any: raise ValueError("No rows written from csv_paths (empty files or schema mismatch).") return parquet_path @staticmethod def _standardize_chunk( df: pd.DataFrame, *, column_map: dict[str, str], date_columns: tuple[str, ...], float32_cols: tuple[str, ...], ) -> pd.DataFrame: """Apply renames, parse dates, compute mid, coerce dtypes, keep Parquet schema.""" chunk = df.rename(columns=dict(column_map)) for col in date_columns: if col in chunk.columns: chunk[col] = pd.to_datetime(chunk[col], errors="coerce").dt.normalize() if "option_type" in chunk.columns: chunk["option_type"] = chunk["option_type"].map(_normalize_option_type) # Mid: prefer explicit mid; else exchange price column; else average of bid/ask if "mid" not in chunk.columns: if "price" in chunk.columns: chunk["mid"] = pd.to_numeric(chunk["price"], errors="coerce") else: chunk["mid"] = (pd.to_numeric(chunk.get("bid"), errors="coerce") + pd.to_numeric(chunk.get("ask"), errors="coerce")) / 2.0 else: chunk["mid"] = pd.to_numeric(chunk["mid"], errors="coerce") for c in ("bid", "ask", "strike", "iv", "delta", "gamma", "theta", "vega"): if c in chunk.columns: chunk[c] = pd.to_numeric(chunk[c], errors="coerce") # Drop rows missing critical fields need = ["date", "expiration", "strike", "option_type", "bid", "ask"] chunk = chunk.dropna(subset=[c for c in need if c in chunk.columns]) if chunk.empty: return chunk for c in float32_cols: if c in chunk.columns: chunk[c] = chunk[c].astype("float32") # Ensure all Parquet columns exist (fill greeks with NaN if missing) for c in IVolatilityLoader._PARQUET_COLS: if c not in chunk.columns: chunk[c] = pd.NA if c in ("date", "expiration", "option_type") else math.nan out = chunk[IVolatilityLoader._PARQUET_COLS].copy() return out def iter_chain_dates(self) -> Iterator[pd.Timestamp]: """Yield unique ``date`` values present in the Parquet file (sorted).""" if pq is None: raise ImportError("pyarrow is required") from _PYARROW_IMPORT_ERROR pf = pq.ParquetFile(self.file_path) dates: set[pd.Timestamp] = set() for i in range(pf.num_row_groups): col = pf.read_row_group(i, columns=["date"]).column(0) # Convert to pandas for unique extraction (per row group keeps peak RAM low) s = pd.to_datetime(col.to_pandas(), errors="coerce").dt.normalize() for u in s.dropna().unique(): dates.add(pd.Timestamp(u)) return iter(sorted(dates)) def get_chain_for_date(self, target_date: pd.Timestamp) -> OptionChain: """ Load **only** rows for ``target_date`` using Parquet predicate pushdown. Uses the PyArrow engine so filters prune row groups where possible. If your ``date`` column was written as plain ``date32``/``timestamp`` without timezone, equality against a normalized ``pd.Timestamp`` is typically recognized by the scanner. We do **not** fall back to a full-file read (that would defeat the purpose on multi-GB datasets). """ if pq is None: raise ImportError("get_chain_for_date requires pyarrow") from _PYARROW_IMPORT_ERROR ts = pd.Timestamp(target_date).normalize() # Try a few literal types iVolatility Parquet exports commonly use. filter_variants: list[list[tuple[str, str, Any]]] = [ [("date", "==", ts)], [("date", "==", ts.to_pydatetime())], [("date", "==", ts.date())], ] last_err: Exception | None = None df = pd.DataFrame() for filters in filter_variants: try: df = pd.read_parquet( self.file_path, columns=self._PARQUET_COLS, filters=filters, engine="pyarrow", ) last_err = None break except Exception as e: # noqa: BLE001 — try alternate filter literals last_err = e continue if last_err is not None and df.empty: raise RuntimeError( "Could not apply Parquet row filter on 'date'. " "Re-run convert_to_parquet so 'date' is timezone-naive datetime64, " "or pass a filter literal matching the on-disk type." ) from last_err if df.empty: return OptionChain(as_of=ts, contracts=[]) df["date"] = pd.to_datetime(df["date"], errors="coerce").dt.normalize() df = df[df["date"] == ts] if df.empty: return OptionChain(as_of=ts, contracts=[]) contracts = [_row_to_contract(r) for _, r in df.iterrows()] return OptionChain(as_of=ts, contracts=contracts) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _ivolatility_csv_sort_key(p: Path) -> tuple[int, int, str]: """``data_download.csv`` first, then ``data_download 2.csv`` … by numeric suffix.""" stem = p.stem if stem == "data_download": return (0, 0, "") prefix = "data_download " if stem.startswith(prefix): suf = stem[len(prefix) :].strip() try: return (0, int(suf), "") except ValueError: return (1, 0, stem) return (1, 0, stem) def _normalize_option_type(x: Any) -> str: if x is None or (isinstance(x, float) and math.isnan(x)): return "" s = str(x).strip().upper() if s.startswith("C"): return "C" if s.startswith("P"): return "P" if s in {"CALL", "PUT"}: return "C" if s == "CALL" else "P" return s[:1] if s else "" def _row_to_contract(r: pd.Series) -> OptionContract: return OptionContract( date=pd.Timestamp(r["date"]).normalize(), expiration=pd.Timestamp(r["expiration"]).normalize(), strike=float(r["strike"]), option_type=str(r["option_type"]), bid=float(r["bid"]), ask=float(r["ask"]), mid=float(r["mid"]), iv=float(r["iv"]), delta=float(r["delta"]), gamma=float(r["gamma"]), theta=float(r["theta"]), vega=float(r["vega"]), ) # --------------------------------------------------------------------------- # Demo # --------------------------------------------------------------------------- if __name__ == "__main__": # Synthetic chain: no Parquet file required — shows API for targeting a put wing. day = pd.Timestamp("2024-06-15") def mk( exp: str, strike: float, delta: float, opt: str = "P", ) -> OptionContract: exp_ts = pd.Timestamp(exp) return OptionContract( date=day, expiration=exp_ts, strike=strike, option_type=opt, bid=1.0, ask=1.1, mid=1.05, iv=0.25, delta=delta, gamma=0.01, theta=-0.02, vega=0.03, ) # Two expiries: ~25 DTE and ~32 DTE — target_dte=30 should pick ~32 (|32-30| < |25-30| if only these) # Actually 30-25=5, 32-30=2 -> pick 32 DTE expiry chain = OptionChain( as_of=day, contracts=[ mk("2024-07-10", 500.0, -0.05), # ~25 DTE, wrong delta mk("2024-07-17", 498.0, -0.09), # ~32 DTE, close to -0.10 delta mk("2024-07-17", 495.0, -0.35), # same expiry, worse delta match mk("2024-07-17", 502.0, -0.11), # best delta match for 10-delta put mk("2024-08-15", 490.0, -0.10), # farther DTE ], ) leg = chain.find_target_leg(target_dte=30, target_delta=-0.10, option_type="P") print("Demo: 30 DTE (closest bucket), ~10-delta put") print(f" Selected: exp={leg.expiration.date()} strike={leg.strike} delta={leg.delta:.3f}") print(f" DTE={(leg.expiration.normalize() - day.normalize()).days}") # Combine split iVolatility exports, then query: # paths = IVolatilityLoader.sorted_ivolatility_csv_paths("SPY-Option-Data") # out = IVolatilityLoader.combine_csvs_to_parquet(paths, "SPY-Option-Data/spy_options_eod_combined.parquet") # loader = IVolatilityLoader(out) # ch = loader.get_chain_for_date(pd.Timestamp("2013-02-15")) # print(len(ch.contracts), "contracts loaded")