"""Data acquisition + caching from Cboe delayed quotes.""" import gzip import json import logging import os import re import time from datetime import date, datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Tuple import requests from .config import GexConfig logger = logging.getLogger(__name__) OPTION_RE = re.compile( r"^(?P[A-Z]+)(?P\d{2})(?P\d{2})(?P
\d{2})" r"(?P[CP])(?P\d{8})$" ) def parse_option_symbol(sym: str) -> Optional[Dict[str, Any]]: """Parse an OSI-ish option symbol into components.""" m = OPTION_RE.match(sym) if not m: return None return { "root": m.group("root"), "expiry": date(2000 + int(m.group("yy")), int(m.group("mm")), int(m.group("dd"))), "cp": m.group("cp"), "strike": int(m.group("strike")) / 1000.0, } def _fetch_url(url: str, cfg: GexConfig) -> Optional[Dict]: """Fetch a single URL with retries and exponential backoff.""" for attempt in range(cfg.retries): try: resp = requests.get( url, headers={"User-Agent": cfg.user_agent}, timeout=30, ) if resp.status_code == 200: return resp.json() logger.warning("HTTP %d for %s (attempt %d)", resp.status_code, url, attempt + 1) if resp.status_code in (403, 404): return None # signal to try URL B except requests.RequestException as exc: logger.warning("Request error for %s: %s (attempt %d)", url, exc, attempt + 1) if attempt < cfg.retries - 1: delay = cfg.backoff_base * (2 ** attempt) logger.info("Backing off %.1fs before retry", delay) time.sleep(delay) return None def fetch_chain(symbol: str, cfg: GexConfig, from_cache: bool = False, cache_id: Optional[str] = None) -> Tuple[Dict, str, str]: """ Fetch the full options chain for *symbol*. Returns (parsed_json, snapshot_id, endpoint_variant). If from_cache, loads the most recent (or specified) cached snapshot. FIX 34: caches the winning endpoint variant to data/endpoint_map.json so subsequent runs skip the failed attempt. endpoint_variant is "plain" or "underscore". """ cache_dir = Path(cfg.cache_dir) cache_dir.mkdir(parents=True, exist_ok=True) if from_cache: data, snap_id = _load_cache(symbol, cfg, cache_id) # read cached endpoint variant emap = _load_endpoint_map(cfg) variant = emap.get(symbol, "plain") # FIX 75: recover the frozen capture time so source-timestamp age is a # property of the snapshot, not of when it happens to be re-rendered. captured_utc = _read_capture_time(symbol, cfg, snap_id) data["_captured_at_utc"] = captured_utc return data, snap_id, variant # FIX 34: check endpoint cache first emap = _load_endpoint_map(cfg) cached_variant = emap.get(symbol) if cached_variant == "underscore": url = cfg.cboe_url_b.format(symbol=symbol) data = _fetch_url(url, cfg) variant = "underscore" elif cached_variant == "plain": url = cfg.cboe_url_a.format(symbol=symbol) data = _fetch_url(url, cfg) variant = "plain" else: # Try URL A first, then URL B url = cfg.cboe_url_a.format(symbol=symbol) data = _fetch_url(url, cfg) variant = "plain" if data is None: url = cfg.cboe_url_b.format(symbol=symbol) data = _fetch_url(url, cfg) variant = "underscore" if data is None: raise RuntimeError(f"Cboe returned no data for {symbol} after retries on both URLs") # Cache the winning endpoint variant emap[symbol] = variant _save_endpoint_map(cfg, emap) # Build snapshot id from API timestamp ts_str = data.get("timestamp", "") snapshot_id = ts_str.replace(" ", "_").replace(":", "") if ts_str else datetime.utcnow().strftime("%Y%m%d_%H%M%S") # Cache raw JSON gzipped cache_path = cache_dir / f"{symbol}_{snapshot_id}.json.gz" with gzip.open(cache_path, "wt", encoding="utf-8") as f: json.dump(data, f) logger.info("Cached raw JSON -> %s (endpoint: %s)", cache_path, variant) # FIX 75: freeze the capture time NOW (live fetch). Any later --from-cache # re-render reads this back so the reported source-timestamp age is the age at # capture, not at render. Without this, re-rendering an old snapshot reports it # as hours stale even though the data is unchanged. _write_capture_time(symbol, cfg, snapshot_id) data["_captured_at_utc"] = datetime.now(timezone.utc).isoformat() return data, snapshot_id, variant def _load_endpoint_map(cfg: GexConfig) -> Dict: """FIX 34: load the endpoint variant cache.""" p = Path(cfg.endpoint_map_path) if p.exists(): try: with open(p) as f: return json.load(f) except Exception: pass return {} def _save_endpoint_map(cfg: GexConfig, emap: Dict): """FIX 34: persist the endpoint variant cache.""" p = Path(cfg.endpoint_map_path) p.parent.mkdir(parents=True, exist_ok=True) try: with open(p, "w") as f: json.dump(emap, f, indent=2) except Exception: pass def _load_cache(symbol: str, cfg: GexConfig, cache_id: Optional[str] = None) -> Tuple[Dict, str]: cache_dir = Path(cfg.cache_dir) pattern = f"{symbol}_*.json.gz" files = sorted(cache_dir.glob(pattern)) if not files: raise FileNotFoundError(f"No cached snapshots for {symbol} in {cache_dir}") if cache_id: target = cache_dir / f"{symbol}_{cache_id}.json.gz" if not target.exists(): raise FileNotFoundError(f"Cache file not found: {target}") files = [target] latest = files[-1] snapshot_id = latest.stem.replace(f"{symbol}_", "").replace(".json", "") with gzip.open(latest, "rt", encoding="utf-8") as f: data = json.load(f) logger.info("Loaded cache: %s", latest) return data, snapshot_id def _capture_time_path(symbol: str, cfg: GexConfig, snapshot_id: str) -> Path: """FIX 75: sidecar file holding the frozen capture time for a cached snapshot.""" return Path(cfg.cache_dir) / f"{symbol}_{snapshot_id}.captured.json" def _write_capture_time(symbol: str, cfg: GexConfig, snapshot_id: str) -> None: """FIX 75: persist the capture time (UTC ISO-8601) next to the cached chain.""" p = _capture_time_path(symbol, cfg, snapshot_id) try: p.parent.mkdir(parents=True, exist_ok=True) with open(p, "w") as f: json.dump({"captured_at_utc": datetime.now(timezone.utc).isoformat()}, f) except Exception as exc: # never fail the fetch over a metadata sidecar logger.warning("FIX 75: could not persist capture time for %s: %s", symbol, exc) def _read_capture_time(symbol: str, cfg: GexConfig, snapshot_id: str) -> Optional[str]: """FIX 75: read back the frozen capture time. Falls back to the cached chain file's mtime (the moment it was written) when no sidecar exists — e.g. chains cached before v1.7.5. Returns an ISO-8601 UTC string, or None if unknowable.""" p = _capture_time_path(symbol, cfg, snapshot_id) if p.exists(): try: with open(p) as f: return json.load(f).get("captured_at_utc") except Exception: pass # fallback: the cache file's mtime is the best available capture proxy cache_path = Path(cfg.cache_dir) / f"{symbol}_{snapshot_id}.json.gz" if cache_path.exists(): try: return datetime.fromtimestamp(cache_path.stat().st_mtime, tz=timezone.utc).isoformat() except Exception: pass return None def parse_chain(data: Dict, symbol: str) -> Tuple[List[Dict], float, str]: """ Parse raw Cboe JSON into a list of contract dicts + spot price + timestamp. Each contract dict: {strike, expiry, cp, iv, oi, volume, delta, gamma, vega, theta, theo, bid, ask} """ d = data.get("data", {}) spot = d.get("current_price", 0) or d.get("close", 0) ts = data.get("timestamp", "") contracts = [] for opt in d.get("options", []): parsed = parse_option_symbol(opt.get("option", "")) if parsed is None: continue contracts.append({ "strike": parsed["strike"], "expiry": parsed["expiry"], "cp": parsed["cp"], "iv": float(opt.get("iv", 0) or 0), "oi": int(opt.get("open_interest", 0) or 0), "volume": int(opt.get("volume", 0) or 0), "delta": float(opt.get("delta", 0) or 0), "gamma": abs(float(opt.get("gamma", 0) or 0)), # Cboe reports positive for both "vega": float(opt.get("vega", 0) or 0), "theta": float(opt.get("theta", 0) or 0), "theo": float(opt.get("theo", 0) or 0), "bid": float(opt.get("bid", 0) or 0), "ask": float(opt.get("ask", 0) or 0), }) return contracts, float(spot), ts