"""CLI entrypoint + schedule guard.""" import argparse import csv import json import logging import sys import time as _time from datetime import date, datetime, time, timedelta, timezone from pathlib import Path from typing import Optional from zoneinfo import ZoneInfo import numpy as np import pandas as pd from .compute import ( aggregate, atm_expected_move, atm_iv_cross_check, build_gex_by_expiry, build_oi_totals, build_outlier_report, compute_bands, compute_levels, compute_realised_vol, detect_increment, detect_render_spacing, detect_spread_candidates, dex_min_price, dex_profile, filter_contracts_band, filter_contracts_full, find_delta_neutral, gamma_precision, gex_profile, is_third_friday, level_front_expiry_pct, minutes_to_settlement, reconcile, reconcile_by_expiry, reconciliation_floor_unsigned, render_bucket, snap_to_increment, _settlement_time, total_net_gex_from_contracts, ) from .config import GexConfig from .fetch import fetch_chain, parse_chain from .plot import render_chart # FIX 89: provenance stamp — every artifact carries the producing code version. from . import __version__ as _GEX_VERSION, __schema_version__ as _GEX_SCHEMA_VERSION logger = logging.getLogger("gex") ET = ZoneInfo("America/New_York") def _slot_label(now_et: datetime, cfg: GexConfig) -> str: """FIX 86b: explicit asymmetric windows (not center ± tolerance). AM window: slot_am_start–slot_am_end ET (default 09:30–12:00). PM window: slot_pm_start–slot_pm_end ET (default 14:00–16:15). The slot label describes INTENT, not precision. The recorded capture time is authoritative; this gate only decides whether a capture is accepted for a given slot filename. Late-but-same-day captures with correct timestamps are usable data — losing them entirely is the worse failure.""" def _hm(s: str): h, m = map(int, s.split(":")) return now_et.replace(hour=h, minute=m, second=0, microsecond=0) am_start, am_end = _hm(cfg.slot_am_start), _hm(cfg.slot_am_end) pm_start, pm_end = _hm(cfg.slot_pm_start), _hm(cfg.slot_pm_end) if am_start <= now_et <= am_end: return "am" if pm_start <= now_et <= pm_end: return "pm" return "" def _is_trading_day(d: date) -> bool: try: import pandas_market_calendars as mcal nyse = mcal.get_calendar("NYSE") sched = nyse.schedule(start_date=pd.Timestamp(d), end_date=pd.Timestamp(d)) return len(sched) > 0 except Exception: return d.weekday() < 5 # fallback: Mon-Fri # FIX 49: explicit NYSE trading-day check (same logic, clearer name for the guard). _is_nyse_trading_day = _is_trading_day def run_ticker(symbol: str, cfg: GexConfig, slot: str, from_cache: bool = False, replay_unsafe: bool = False, overwrite: bool = False) -> bool: """Process one ticker end-to-end. Returns True on success. FIX 97: the ex-front (0DTE-removed) second chart was removed. There is now a single pipeline and a single "All Expirations" variant per ticker/slot. The drift risk that the old FIX 68 identical-keys test guarded against disappears with the second path. The JSON `variant` field is retained with the constant value "all_expirations" so the canonical key set does not churn. replay_unsafe (FIX 52): bypass ONLY the FIX 49 trading-day/session-window check so an out-of-session cached chain (e.g. the Saturday NDX chain that carries expired contracts) can exercise FIX 48/50 on real data. Output is flagged non-publishable and forced to a scratch outdir by main(). overwrite (FIX 79b): force-overwrite an existing output whose snapshot_id differs. Default False = non-destructive (refuse rather than clobber history). """ try: logger.info("=== %s ===", symbol) data, snap_id, endpoint_variant = fetch_chain(symbol, cfg, from_cache=from_cache) contracts, spot, ts_str = parse_chain(data, symbol) # FIX 75: the frozen capture time (UTC ISO) persisted at fetch; used to report # source-timestamp age as-of capture rather than as-of render. captured_at_utc = data.get("_captured_at_utc") logger.info("Spot=%.2f contracts=%d ts=%s endpoint=%s", spot, len(contracts), ts_str, endpoint_variant) # FIX 34: instrument class inferred from endpoint variant instrument_class = "index" if endpoint_variant == "underscore" else "equity_etf" today = datetime.now(ET).date() # FIX 15: use the SNAPSHOT timestamp (converted to ET) for time-to-expiry, # not wall-clock now, so --from-cache reproduces exactly. try: snap_et = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S").replace( tzinfo=ZoneInfo(cfg.source_timestamp_tz)).astimezone(ET) except Exception: snap_et = datetime.now(ET) # FIX 49 (supersedes FIX 47): validate the SOURCE timestamp (converted to # ET), not wall-clock. Refuse to publish unless it falls on an NYSE trading # day AND within 09:30-16:15 ET. The slot guard checks wall-clock; this # checks the DATA. Both are required. # FIX 52: --replay-unsafe bypasses ONLY this trading-day/session-window check # (so an out-of-session cached chain can exercise FIX 48/50), logging an ERROR # on every run. The age check below stays active in all modes. in_session = (_is_nyse_trading_day(snap_et.date()) and time(9, 30) <= snap_et.time() <= time(16, 15)) if not in_session: if replay_unsafe: logger.error("REPLAY MODE — source timestamp %s -> %s ET is outside " "session, output is not publishable (%s/%s).", ts_str, snap_et.strftime("%Y-%m-%d %H:%M"), symbol, slot) elif not _is_nyse_trading_day(snap_et.date()): logger.error("SOURCE timestamp %s -> %s ET is not an NYSE trading day. " "Refusing snapshot for %s/%s.", ts_str, snap_et.strftime("%Y-%m-%d %H:%M"), symbol, slot) return False else: logger.error("SOURCE timestamp %s -> %s ET is outside 09:30-16:15 " "trading window. Refusing snapshot for %s/%s.", ts_str, snap_et.strftime("%Y-%m-%d %H:%M"), symbol, slot) return False # FIX 49: refuse stale data outright if the source is > 240 min old. # v1.6.2: this age check compares the source timestamp to wall-clock now, # so it must NOT block --from-cache replays (a cached snapshot is always # "old" by wall-clock; blocking it breaks the reproducibility guarantee the # audit page makes for --from-cache). Log INFO instead when replaying from # cache. The two checks above (NYSE trading day + 09:30-16:15 ET window) # validate the DATA itself, not its age, and stay active in all modes. age_min = (datetime.now(ET) - snap_et).total_seconds() / 60.0 if age_min > 240: if from_cache: logger.info("SOURCE timestamp age %.1f min (>240) for %s/%s — " "allowed because --from-cache (reproducibility replay).", age_min, symbol, slot) else: logger.error("SOURCE timestamp age %.1f min (>240) for %s/%s — refusing.", age_min, symbol, slot) return False # FIX 79a: enforce the slot source-window at WRITE time, not just at fetch. # The output filename is {date}_{slot}.json where `date` comes from the Cboe # source timestamp but `slot` is only a label (--slot / auto). Without this # check a 15:55-ET capture run with --slot am writes {date}_am.json and # clobbers the 10:00-ET original. Refuse any capture whose CAPTURE TIME (when # the fetch happened, not the Cboe source timestamp which has a ~15-min feed # delay) falls outside the requested slot's window (FIX 86b: explicit # asymmetric windows, AM 09:30–12:00 / PM 14:00–16:15 ET). Applies in every # mode (including --from-cache) because the filename must always match the # capture's true window; bypassed only under --replay-unsafe (diagnostic # scratch output that is never published). if not replay_unsafe: # Use the capture time (when the fetch happened), not the Cboe source # timestamp. The source timestamp has a ~15-min feed delay, so a fetch # at 10:00 ET gets data timestamped ~10:15+ ET — checking the source # timestamp would reject all legitimate captures. if captured_at_utc: try: capture_et = datetime.fromisoformat(captured_at_utc).astimezone(ET) except Exception: capture_et = snap_et # fallback to source timestamp else: capture_et = snap_et # no capture time available (pre-FIX 75 cache) capture_slot = _slot_label(capture_et, cfg) if capture_slot != slot: if slot == "am": window = "%s–%s" % (cfg.slot_am_start, cfg.slot_am_end) else: window = "%s–%s" % (cfg.slot_pm_start, cfg.slot_pm_end) logger.error("SLOT WINDOW: capture taken at %s ET falls in slot %r, not " "the requested %r (window %s ET). Refusing to write " "%s/%s for %s.", capture_et.strftime("%Y-%m-%d %H:%M"), capture_slot or "(none)", slot, window, symbol, slot, symbol) return False df_full = filter_contracts_full(contracts, spot, cfg, snap_et, instrument_class=instrument_class) if df_full.empty: logger.warning("No contracts survived filtering for %s", symbol) return False # FIX 45: count contracts dropped by rule 4 (OI>0 but iv<=0 or gamma==0) zero_greek = sum(1 for c in contracts if c["oi"] > 0 and (c["iv"] <= 0 or c["gamma"] == 0)) # FIX 97: the per-expiry zero-greek tally was removed with the ex-front # variant — only that second path needed a post-filter count. The single # pipeline reports the full-chain count for both fields. # primary run (all expirations) primary_result = _process_and_render( symbol, cfg, slot, snap_id, ts_str, spot, df_full, suffix="", endpoint_variant=endpoint_variant, instrument_class=instrument_class, zero_greek_contracts=zero_greek, zero_greek_full_chain=zero_greek, replay_unsafe=replay_unsafe, captured_at_utc=captured_at_utc, overwrite=overwrite) if primary_result is None: logger.error("Write refused for %s/%s (snapshot_id conflict). " "Pass --overwrite to force.", symbol, slot) return False # FIX 97: the ex-front (0DTE-removed) second chart was removed. There is a # single pipeline and a single "All Expirations" variant per ticker/slot. return True except Exception as exc: logger.exception("FAILED %s: %s", symbol, exc) return False def _load_recent_closes(symbol: str, cfg: GexConfig) -> list: """FIX 32: load recent daily closes for realised vol. Tries Yahoo Finance chart API first (free, no key), then falls back to cached snapshot spots. Returns a list of floats (oldest first), or empty list on failure. """ closes = [] # Try Yahoo Finance chart API (free, no key) try: import urllib.request url = f"https://query1.finance.yahoo.com/v8/finance/chart/{symbol}?range=1mo&interval=1d" req = urllib.request.Request(url, headers={"User-Agent": cfg.user_agent}) with urllib.request.urlopen(req, timeout=10) as resp: data = json.loads(resp.read().decode()) raw = data["chart"]["result"][0]["indicators"]["quote"][0]["close"] closes = [float(c) for c in raw if c is not None] except Exception: pass # Fallback: cached snapshot spots (from prior JSON files) if len(closes) < 5: try: out_dir = Path(cfg.outdir) / symbol if out_dir.exists(): for jf in sorted(out_dir.glob("*.json")): try: with open(jf) as f: d = json.load(f) if "spot" in d: closes.append(float(d["spot"])) except Exception: pass except Exception: pass return closes def _atm_iv_source_detail(df_band: pd.DataFrame, levels: dict, spot: float) -> list: """FIX 32: publish the two source contracts' bid/ask/OI/IV for full traceability.""" src_strikes = levels.get("atm_iv_source_strikes") expiry = levels.get("atm_iv_expiry") if not src_strikes or not expiry: return [] detail = [] for k in src_strikes: sub = df_band[(df_band["strike"] == k) & (df_band["expiry"].astype(str) == expiry)] for _, row in sub.iterrows(): detail.append({ "strike": float(k), "cp": row["cp"], "oi": int(row["oi"]), "bid": float(row.get("bid", 0)), "ask": float(row.get("ask", 0)), "iv": float(row["iv"]), }) return detail def _pass1_atm_iv(df_full: pd.DataFrame, spot: float, cfg: GexConfig) -> tuple: """FIX 35 PASS 1: coarse ATM IV estimate for band derivation. Filter: OI > 0, IV in (0.01, 3.0), DTE in [5, 60]. FIX 43: delegate to the SAME robust resolver (chain-relative gates + three-expiry widening) that the full pipeline uses, so pass-1 resolves whenever the full run does — otherwise a low-OI chain would resolve a real atm_iv downstream yet still derive its bands from the fallback. FIX 67: also returns the atm_iv_status so the caller can set band_basis="atm_iv_single_strike" when the IV rests on one strike. FIX 82: returns the full result dict (not just atm_iv + status) so the caller can publish bands.atm_iv_source_expiry and bands.atm_iv_source_strikes, making the pass-1 vs pass-2 divergence visible (pass 1 runs on df_coarse with a DTE filter; pass 2 runs on df_band with a strike filter — they can pick different expiries/strikes, so the two atm_iv values can differ). """ if df_full.empty: return None, "rejected: empty", {} # coarse filter mask = ( (df_full["oi"] > 0) & (df_full["iv"] > 0.01) & (df_full["iv"] < 3.0) & (df_full["dte"] >= 5) & (df_full["dte"] <= 60) ) df_coarse = df_full[mask] if df_coarse.empty: return None, "rejected: no eligible expiry", {} result = atm_expected_move(df_coarse, spot) return result.get("atm_iv"), result.get("atm_iv_status", "unknown"), result def _append_history_csv(symbol: str, slot: str, variant: str, levels: dict, cfg: GexConfig): """FIX 40: append one row per ticker/slot/variant to data/history/{SYMBOL}.csv.""" hist_dir = Path(cfg.history_dir) hist_dir.mkdir(parents=True, exist_ok=True) csv_path = hist_dir / f"{symbol}.csv" header = [ "timestamp_et", "slot", "variant", "spot", "call_resistance", "put_support", "hvl", "hvl_distance_pct", "gex_transition", "delta_neutral", "total_net_gex", "net_gex_at_spot", "gamma_condition", "atm_iv", "realised_vol_20d", "iv_hv_ratio", "max_expiry_share", "spread_flagged_share", "gex_profile_max", "dex_profile_max", ] write_header = not csv_path.exists() row = [ levels.get("timestamp_et", ""), slot, variant, levels.get("spot"), levels.get("call_resistance"), levels.get("put_support"), levels.get("hvl"), levels.get("hvl_distance_pct"), levels.get("gex_transition"), levels.get("delta_neutral"), levels.get("total_net_gex"), levels.get("net_gex_at_spot"), levels.get("gamma_condition"), levels.get("atm_iv"), levels.get("realised_vol_20d"), levels.get("iv_hv_ratio"), levels.get("max_expiry_share"), levels.get("spread_flagged_share"), levels.get("gex_profile_max"), levels.get("dex_profile_max"), ] with open(csv_path, "a", newline="") as f: writer = csv.writer(f) if write_header: writer.writerow(header) writer.writerow(row) logger.info("History CSV -> %s", csv_path) def _rolling_profile_limit(symbol: str, column: str, cfg: GexConfig) -> Optional[float]: """FIX 50: cross-snapshot-comparable axis limit = 1.2 * median of this ticker's last `rolling_window` profile maxima (from the FIX 40 history CSV). Returns None when fewer than 5 snapshots exist, so the caller falls back to "data" mode. OI-aware by construction (each ticker's own history).""" csv_path = Path(cfg.history_dir) / f"{symbol}.csv" if not csv_path.exists(): return None try: df = pd.read_csv(csv_path) except Exception: return None if column not in df.columns: return None vals = pd.to_numeric(df[column], errors="coerce").dropna() if len(vals) < 5: return None recent = vals.tail(cfg.rolling_window) return float(1.2 * recent.median()) def build_index_json(cfg: GexConfig): """FIX 38 / FIX 51: scan out/ for available JSON snapshots and write index.json. FIX 51: the index is DERIVED from what is actually on disk, never authored from cfg.tickers. `tickers` is the sorted list of directories that contain >=1 snapshot JSON; `default` is cfg.default_ticker only if it is a member of that list, else the first listed ticker. A configured ticker with zero published snapshots never appears (and is named in a WARNING). This keeps `default` always a member of `tickers`. Format: {"tickers": ["SMH"], "default": "SMH", "snapshots": {"SMH": {"2026-07-24": ["pm"]}, ...}} """ out_dir = Path(cfg.outdir) index = {"tickers": [], "default": None, "snapshots": {}} if not out_dir.exists(): idx_path = out_dir / "index.json" with open(idx_path, "w") as f: json.dump(index, f, indent=2) logger.warning("Index -> %s (outdir missing; empty index)", idx_path) return for ticker_dir in sorted(out_dir.iterdir()): if not ticker_dir.is_dir(): continue sym = ticker_dir.name snaps = {} for jf in sorted(ticker_dir.glob("*.json")): # filename: 2026-07-24_pm.json (single variant since FIX 97) stem = jf.stem # e.g. "2026-07-24_pm" parts = stem.split("_", 1) if len(parts) < 2: continue date_str, slot_variant = parts[0], parts[1] snaps.setdefault(date_str, []) if slot_variant not in snaps[date_str]: snaps[date_str].append(slot_variant) if snaps: index["tickers"].append(sym) index["snapshots"][sym] = snaps # FIX 51: default must be a member of tickers; fall back to the first listed. if index["tickers"]: index["default"] = (cfg.default_ticker if cfg.default_ticker in index["tickers"] else index["tickers"][0]) if cfg.default_ticker not in index["tickers"]: logger.warning("Configured default_ticker %r has no published snapshots; " "index default falls back to %r.", cfg.default_ticker, index["default"]) # FIX 51: name any configured ticker that produced no output. for t in cfg.tickers: if t not in index["tickers"]: logger.warning("Configured ticker %r has no published snapshots; " "omitted from index.", t) # write to out/index.json (deployed as gex_out/index.json) idx_path = out_dir / "index.json" with open(idx_path, "w") as f: json.dump(index, f, indent=2) logger.info("Index -> %s (%d tickers, default=%s)", idx_path, len(index["tickers"]), index["default"]) def _process_and_render(symbol, cfg, slot, snap_id, ts_str, spot, df_full, suffix="", endpoint_variant="plain", instrument_class="equity_etf", zero_greek_contracts=0, zero_greek_full_chain=0, replay_unsafe=False, captured_at_utc=None, overwrite=False): """Shared pipeline: two-pass bands, profiles, levels, chart, JSON, parquet, CSV. `suffix` is appended to the output filenames. FIX 97: the ex-front variant was removed, so this is always "" in practice; the parameter is retained so the JSON `variant` field resolves to the constant "all_expirations" and the canonical key set does not churn. `replay_unsafe` (FIX 52) is stamped into the levels JSON so downstream consumers can tell a non-publishable replay apart from a live snapshot. `captured_at_utc` (FIX 75) is the frozen fetch time (UTC ISO); when present, the published source_timestamp_age_min is computed as-of capture, and the wall-clock delta since capture is published separately as render_lag_min. `zero_greek_contracts` (FIX 78) is the count of rule-4-dropped contracts for the book being charted. `zero_greek_full_chain` is the raw full-chain count. With the ex-front variant removed (FIX 97) the two are always equal. `overwrite` (FIX 79b): when False (default), an existing output JSON whose snapshot_id differs from `snap_id` is NOT clobbered — the write is refused. A re-render of the SAME snapshot_id is always allowed (that is what FIX 75's byte-identity guarantee depends on). Set True via --overwrite to force. """ # --- FIX 43: compute realised vol early so it can back the band fallback --- closes = _load_recent_closes(symbol, cfg) rv_early = compute_realised_vol(closes, cfg.realised_vol_days) # --- FIX 35 PASS 1: coarse ATM IV -> derive bands --- pass1_iv, pass1_iv_status, pass1_result = _pass1_atm_iv(df_full, spot, cfg) bands = compute_bands(pass1_iv, cfg, realised_vol=rv_early, atm_iv_status=pass1_iv_status) # FIX 82: publish the pass-1 ATM IV source so the divergence between # bands.atm_iv_used (pass 1, df_coarse) and the top-level atm_iv (pass 2, # df_band) is visible. They can differ because the two passes run on # different contract sets (DTE filter vs strike filter). bands["atm_iv_source_expiry"] = pass1_result.get("atm_iv_expiry") bands["atm_iv_source_strikes"] = pass1_result.get("atm_iv_source_strikes") if bands["band_basis"] == "fallback_from_rv": logger.warning("ATM IV unresolved — bands derived from realised vol " "(rv=%.4f -> iv=%.4f).", rv_early, bands["atm_iv_used"]) logger.info("Bands (basis=%s): iv=%.4f sigma30d=%.4f strike=%.3f plot=%.3f " "profile=%.3f dex=%.3f", bands["band_basis"], bands["atm_iv_used"], bands["sigma_30d"], bands["strike_band"], bands["plot_band"], bands["profile_band"], bands["dex_band"]) # --- FIX 35 PASS 2: run pipeline with derived bands --- # Override cfg bands for this run cfg_run = GexConfig( **{**vars(cfg), "strike_band": bands["strike_band"], "plot_band": bands["plot_band"], "profile_band": bands["profile_band"], "profile_band_dex": bands["dex_band"], } ) df_band = filter_contracts_band(df_full, spot, cfg_run) logger.info("Filtered contracts%s: full=%d band=%d", suffix or "", len(df_full), len(df_band)) agg = aggregate(df_band, spot, cfg_run) increment = detect_increment(agg.index.to_numpy()) # FIX 71 / FIX 72: ONE render bucket, computed once here from the PLOT-BAND # window (spot × (1 ± plot_band)) — the region where bars actually exist — NOT # the level-extended window (distant HVL/GEX-transition levels used to stretch # it and force a coarser bucket). This bucket drives BOTH the bar aggregation in # the renderer and the published render_spacing/render_bucket fields, so they can # never disagree. detect_render_spacing (FIX 44) is retained only as a diagnostic. _diag_spacing = detect_render_spacing( agg.index.to_numpy(), agg["net_gex"].to_numpy(), increment) _pb_lo = spot * (1 - cfg_run.plot_band) _pb_hi = spot * (1 + cfg_run.plot_band) render_bucket_val = render_bucket(_pb_hi - _pb_lo, increment) logger.info("Strike increment: %s render bucket: %s (plot-band span %.1f; " "diag spacing %s)", increment, render_bucket_val, _pb_hi - _pb_lo, _diag_spacing) # --- FIX 36: fixed-count profile grid (scale-free) --- n_pts = cfg_run.profile_grid_points grid = np.linspace(spot * (1 - cfg_run.profile_band), spot * (1 + cfg_run.profile_band), n_pts) grid_dex = np.linspace(spot * (1 - cfg_run.profile_band_dex), spot * (1 + cfg_run.profile_band_dex), n_pts) logger.info("Profile grid: gex=%d pts (±%.1f%%) dex=%d pts (±%.1f%%)", len(grid), cfg_run.profile_band * 100, len(grid_dex), cfg_run.profile_band_dex * 100) _t0 = _time.perf_counter() gp, gp_outliers = gex_profile(df_full, spot, cfg_run, grid) dp, dp_outliers = dex_profile(df_full, cfg_run, grid_dex, spot) _dt = _time.perf_counter() - _t0 logger.info("Profile compute: %.2fs (gex grid=%d, dex grid=%d)", _dt, len(grid), len(grid_dex)) if _dt > 30: logger.warning("PROFILE TIME: %.1fs exceeds 30s threshold for %s", _dt, symbol) # FIX 48: assert every profile array is finite before plotting; refuse to render if not. if not (np.isfinite(gp).all() and np.isfinite(dp).all()): logger.error("FIX 48: non-finite profile values for %s — refusing to render.", symbol) return None # FIX 22: profile at spot (interpolated on the GEX grid) drives gamma_condition profile_at_spot = float(np.interp(spot, grid, gp)) # FIX 29: HVL = zero crossing nearest spot. If no crossing in the display # grid, widen to ±profile_band_hvl and retry. Still none -> real market state. from .compute import compute_hvl, _hvl_zero_cross hvl_nearest, hvl_crossings = _hvl_zero_cross(grid, gp, spot) if hvl_nearest is None: grid_hvl = np.linspace(spot * (1 - cfg_run.profile_band_hvl), spot * (1 + cfg_run.profile_band_hvl), n_pts) gp_hvl, _ = gex_profile(df_full, spot, cfg_run, grid_hvl) hvl_nearest, hvl_crossings = _hvl_zero_cross(grid_hvl, gp_hvl, spot) if hvl_nearest is not None: logger.info("HVL found on widened ±%.0f%% grid: %.2f", cfg_run.profile_band_hvl * 100, hvl_nearest) else: logger.warning("HVL: no gamma flip within ±%.0f%% of spot — real market state.", cfg_run.profile_band_hvl * 100) levels = compute_levels(agg, grid, gp, spot, increment, cfg_run, profile_at_spot=profile_at_spot) # FIX 44 / FIX 71: publish the true strike grid increment (used for rounding) and # the render bucket (used for bar aggregation + height). render_spacing is kept as # an ALIAS of render_bucket for one version so the two can never disagree. levels["strike_increment"] = float(increment) levels["render_bucket"] = float(render_bucket_val) levels["render_spacing"] = float(render_bucket_val) # FIX 71 alias of render_bucket # FIX 52: stamp replay provenance so consumers can reject non-publishable output. levels["replay_unsafe"] = bool(replay_unsafe) # NOTE (v1.7.0): the redundant top-level `band_basis` convenience copy added in # v1.6.5 is removed. The canonical field is levels["bands"]["band_basis"], which # is what the page reads (d.bands.band_basis). No published consumer used the # top-level copy, so removing it breaks nothing. exp_move = atm_expected_move(df_band, spot) levels.update(exp_move) # FIX 15: reconcile bars vs profile at spot (full chain). total_full = total_net_gex_from_contracts(df_full, spot, cfg_run) total_band = float(agg["net_gex"].sum()) levels["total_net_gex_full"] = total_full levels["total_net_gex_band"] = total_band levels["total_net_gex"] = total_full # FIX 80: guard reconciliation against near-settlement expiries. Near settlement # (T -> 0), gamma ∝ 1/sqrt(T) makes the BS recompute unstable against Cboe's # ~15-min-delayed feed, so the front expiry's reported-vs-recomputed gap is # dominated by feed lag, not model error. Exclude expiries inside # min_minutes_to_settlement from the reconciliation numerator AND denominator # (they are still plotted from reported gamma — bars/levels/HVL use the full # chain). Compute a second profile from the filtered chain for the signed check. try: _snap_et = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S").replace( tzinfo=ZoneInfo(cfg.source_timestamp_tz)).astimezone(ET) except Exception: _snap_et = datetime.now(ET) _min_min = cfg_run.min_minutes_to_settlement _excluded_expiries = [] _recon_expiries = set() for _exp in df_full["expiry"].unique(): _am = (instrument_class == "index" and is_third_friday(_exp)) _mts = minutes_to_settlement(_exp, _snap_et, am_settled=_am, instrument_class=instrument_class) # FIX 84: publish the settlement time so the assumption is auditable. _settle = _settlement_time(_am, instrument_class) _settle_str = _settle.strftime("%H:%M") if _mts < _min_min: _excluded_expiries.append({ "expiry": str(_exp), "minutes_to_settlement": round(_mts, 1), "settlement_time_et": _settle_str, }) else: _recon_expiries.add(_exp) if _excluded_expiries: df_recon = df_full[df_full["expiry"].isin(_recon_expiries)].copy() reconciliation_scope = "excl_near_settlement" logger.info("FIX 80: excluding %d near-settlement expiries from reconciliation " "(< %d min): %s", len(_excluded_expiries), _min_min, ", ".join(e["expiry"] for e in _excluded_expiries)) else: df_recon = df_full reconciliation_scope = "full" # FIX 90 / FIX 94: disclose how much of the book the reconciliation actually # covers. The near-settlement guard excludes the 0DTE expiry on every expiry day; # here that is a material share of GEX validated by nothing. # FIX 94: use the SAME basis as gex_by_expiry.share so the two are comparable. # Numerator: Σ|net GEX per excluded expiry| (net per expiry, then abs, then sum). # Denominator: Σ|net GEX per strike| (same as gex_by_expiry's denom). # Previously mixed a gross numerator (Σ|GEX| per contract) with a gross # denominator, which was inconsistent with gex_by_expiry.share's net basis. _M = cfg_run.contract_multiplier _S2 = spot * spot _sgn_full = np.where(df_full["cp"].to_numpy() == "C", 1.0, -1.0) _signed_gex_full = (_sgn_full * df_full["gamma"].abs().to_numpy() * df_full["oi"].to_numpy() * _M * _S2 * 0.01) _df_tmp = df_full.assign(_gex=_signed_gex_full) _per_strike_net = _df_tmp.groupby("strike")["_gex"].sum() _denom_net = float(_per_strike_net.abs().sum()) _excl_set = {e["expiry"] for e in _excluded_expiries} if _denom_net > 0 and _excl_set: _excl_by_exp = _df_tmp[_df_tmp["expiry"].astype(str).isin(_excl_set)].groupby("expiry")["_gex"].sum() _net_excl = float(_excl_by_exp.abs().sum()) reconciliation_excluded_share = _net_excl / _denom_net else: reconciliation_excluded_share = 0.0 # Signed reconciliation: profile@spot vs total, both from the recon chain. if df_recon.empty: # Degenerate: every expiry is near-settlement. Reconciliation undefined. rec = {"total_net_gex": 0.0, "profile_at_spot": 0.0, "rel_err": None, "rel_err_denominator": 0.0, "pass": True} rec_by_exp = {"reconciliation_worst_expiries": [], "zero_greek_contracts_dropped": 0, "rel_err_unsigned": None, "rel_err_unsigned_denominator": 0.0} else: total_recon = total_net_gex_from_contracts(df_recon, spot, cfg_run) if reconciliation_scope == "excl_near_settlement": # Recompute the profile from the filtered chain for a consistent comparison. gp_recon, _ = gex_profile(df_recon, spot, cfg_run, grid) rec = reconcile(total_recon, grid, gp_recon, spot, threshold=0.05) else: rec = reconcile(total_recon, grid, gp, spot, threshold=0.05) rec_by_exp = reconcile_by_expiry(df_recon, spot, cfg_run) levels["reconciliation"] = rec levels["reconciliation_scope"] = reconciliation_scope levels["reconciliation_excluded_expiries"] = _excluded_expiries # FIX 90: share of Σ|GEX| excluded by the near-settlement guard (0.0 when the # scope is "full"). The pass covers (1 - share) of the book, not 100%. levels["reconciliation_excluded_share"] = reconciliation_excluded_share # FIX 45: per-expiry reconciliation breakdown levels["reconciliation_worst_expiries"] = rec_by_exp["reconciliation_worst_expiries"] # FIX 64: publish the unsigned reconciliation alongside the signed one. levels["rel_err_unsigned"] = rec_by_exp["rel_err_unsigned"] # FIX 73: publish the denominators each error is scored against. With the # ex-front variant removed (FIX 97) there is a single book per ticker/slot, so # these describe the full chain; they remain published for auditability. levels["rel_err_denominator"] = rec["rel_err_denominator"] levels["rel_err_unsigned_denominator"] = rec_by_exp["rel_err_unsigned_denominator"] # FIX 76: reconciliation is scored against the book this chart depicts. Each # chart's curve-derived levels are fit to that book, so the validation that # matters is "does THIS profile reproduce THESE bars". With a single pipeline # (FIX 97) the book is the full chain. Published so a reader can see which # convention produced the flag. levels["reconciliation_denominator_basis"] = "variant" # FIX 73: pass requires BOTH the signed error < 0.05 AND the unsigned error # < 0.10. The signed headline can pass on cancelling per-expiry errors (NDX main: # signed 0.031 but unsigned 0.076); the unsigned figure cannot, so it is a second # gate. reconciliation_pass_basis names which metric bound the decision. # FIX 80: _signed/_unsigned are None when every expiry is near-settlement # (reconciliation undefined); treat that as a pass with basis "excluded". _signed = rec["rel_err"] _unsigned = rec_by_exp["rel_err_unsigned"] # FIX 87: estimate the irreducible unsigned error from PUBLICATION ROUNDING via # the grid_rounding method (round the recomputed truth gamma to the 4dp grid and # measure rounded-vs-unrounded). This isolates pure quantisation noise — the model # error cancels — unlike the old FIX 85 estimator which double-counted rounding. _floor = reconciliation_floor_unsigned(df_recon, spot, cfg_run) levels["reconciliation_floor_unsigned"] = _floor # FIX 88: derive an unsigned gate from the floor, but CAP it. An uncapped gate # (e.g. 0.35 against a realistic worst case of ~0.10) leaves the unsigned check # unable to fire. The derived gate is floor_p95 × multiplier, capped at # base × unsigned_gate_cap_multiplier (2.0). If the corrected floor STILL brackets # the base gate, model error cannot be separated from publication rounding at this # symbol's gamma precision — the result is INDETERMINATE (precision_limited), # never a pass. _base_gate = cfg_run.unsigned_gate _gate_cap = _base_gate * cfg_run.unsigned_gate_cap_multiplier _brackets = bool(_floor["brackets_gate"]) if _brackets and _floor["floor"] is not None: _derived = _floor["floor"] * cfg_run.unsigned_floor_multiplier _unsigned_gate = min(_derived, _gate_cap) _unsigned_gate_status = "capped" if _derived > _gate_cap else "derived" logger.info("FIX 88: unsigned floor=%.4f brackets base gate %.2f -> " "derived gate %.4f (floor × %.1f), capped at %.2f -> effective %.4f " "(status=%s).", _floor["floor"], _base_gate, _derived, cfg_run.unsigned_floor_multiplier, _gate_cap, _unsigned_gate, _unsigned_gate_status) else: _unsigned_gate = _base_gate _unsigned_gate_status = "base" levels["unsigned_gate_effective"] = _unsigned_gate levels["unsigned_gate_status"] = _unsigned_gate_status _signed_ok = (_signed is None) or (_signed < 0.05) _unsigned_ok = (_unsigned is None) or (_unsigned < _unsigned_gate) # FIX 88: indeterminate state. When the floor brackets the base gate AND the # unsigned error is not cleanly below the base gate, the unsigned check is # precision-limited: we cannot tell model error from rounding noise. This is # neither a pass nor a definitive fail. _precision_limited = bool( _brackets and _unsigned is not None and _unsigned >= _base_gate) if _precision_limited: # FIX 92: profile_reliable is boolean-or-null. "indeterminate" is truthy in # Python, so a consumer's plain `if pass:` would read it as a pass. The # precision-limited state is communicated via unsigned_gate_status instead. profile_reliable = False profile_ok = False # not a clean pass — curve-derived levels stay in doubt _unsigned_gate_status = "precision_limited" levels["unsigned_gate_status"] = _unsigned_gate_status # re-publish (was set pre-branch) else: profile_ok = bool(_signed_ok and _unsigned_ok) profile_reliable = profile_ok # FIX 83: the nested reconciliation.pass is one authoritative flag and must agree # with profile_reliable (boolean: True / False). rec["pass"] = profile_reliable # FIX 90: disambiguate the basis. "both_pass" = both gates passed; "signed" / # "unsigned" = which single gate was breached; "signed+unsigned" = both breached; # "excluded" = reconciliation undefined (all expiries near-settlement); # "indeterminate" = precision-limited (FIX 88). if _precision_limited: levels["reconciliation_pass_basis"] = "indeterminate" elif _signed is None and _unsigned is None: levels["reconciliation_pass_basis"] = "excluded" elif profile_ok: levels["reconciliation_pass_basis"] = "both_pass" else: _breached = [] if not _signed_ok: _breached.append("signed") if not _unsigned_ok: _breached.append("unsigned") levels["reconciliation_pass_basis"] = "+".join(_breached) # FIX 61: scope the reliability flag. Bars use Cboe's REPORTED gamma and are # unaffected by any profile discrepancy. Only the recomputed profile and the # levels derived from it (HVL, GEX Transition, delta-neutral) are in doubt. # FIX 66 / FIX 70: detect Cboe gamma publication precision. Three-state label: # "high" (3+ sig figs) | "adequate" (2) | "coarse" (1). Only "coarse" flips # bars_reliable and shows the chart warning; "adequate" (SMH/SPY, ~1.3% # granularity) is corroborated by their <1% reconciliation and keeps bars reliable. gamma_sig_figs, gamma_low_precision, gamma_label = gamma_precision(df_full) levels["gamma_precision_digits"] = gamma_sig_figs levels["gamma_precision"] = gamma_label # FIX 70: three-state levels["reported_gamma_low_precision"] = gamma_low_precision # derived alias levels["bars_reliable"] = not gamma_low_precision # coarse bars -> unreliable # FIX 92: profile_reliable is boolean-or-null (True = pass, False = fail or # precision-limited). The tri-state semantics live in unsigned_gate_status # ("base" / "derived" / "capped" / "precision_limited") and reconciliation_pass_basis. levels["profile_reliable"] = profile_reliable if gamma_low_precision: logger.warning("FIX 66: gamma precision = %d sig fig(s) (%s) — bars are coarse " "(Cboe 4dp at this price level). bars_reliable=False; " "recomputed profile is the higher-precision object.", gamma_sig_figs, gamma_label) if _precision_limited: # FIX 88: indeterminate — curve-derived levels stay in doubt, but this is NOT # a definitive failure. Mark the levels unreliable and log distinctly. levels["hvl_reliable"] = False levels["gex_transition_reliable"] = False levels["delta_neutral_reliable"] = False logger.warning("PROFILE RECONCILE INDETERMINATE (precision_limited): signed=%s " "unsigned=%s vs base gate %.2f; floor=%.4f brackets the gate, " "so model error cannot be separated from publication rounding at " "this symbol's gamma precision. Curve-derived levels marked " "unreliable; bars unaffected.", f"{_signed:.3f}" if _signed is not None else "n/a", f"{_unsigned:.3f}" if _unsigned is not None else "n/a", _base_gate, _floor["floor"] if _floor["floor"] is not None else float("nan")) elif not profile_ok: # mark the curve-derived levels individually levels["hvl_reliable"] = False levels["gex_transition_reliable"] = False levels["delta_neutral_reliable"] = False worst = rec_by_exp["reconciliation_worst_expiries"] worst_exp = worst[0]["expiry"] if worst else "n/a" worst_gap = worst[0]["dollar_gap"] if worst else float("nan") logger.error("PROFILE RECONCILE FAIL (basis=%s): signed=%s unsigned=%s " "(profile@spot=%.4e vs total=%.4e); worst expiry %s dollar_gap=%.3e. " "Bars unaffected; curve-derived levels marked unreliable.", levels["reconciliation_pass_basis"], f"{_signed:.3f}" if _signed is not None else "n/a", f"{_unsigned:.3f}" if _unsigned is not None else "n/a", rec["profile_at_spot"], rec["total_net_gex"], worst_exp, worst_gap) else: # FIX 68 / FIX 86: always emit the reliability flags so the JSON key set # is identical regardless of whether reconciliation passed (canonical schema). levels["hvl_reliable"] = True levels["gex_transition_reliable"] = True levels["delta_neutral_reliable"] = True logger.info("Reconcile OK (basis=%s): signed=%s unsigned=%s " "(profile@spot=%.4e, total=%.4e)", levels["reconciliation_pass_basis"], f"{_signed:.4f}" if _signed is not None else "n/a", f"{_unsigned:.4f}" if _unsigned is not None else "n/a", rec["profile_at_spot"], rec["total_net_gex"]) levels["zero_greek_contracts_dropped"] = zero_greek_contracts levels["zero_greek_contracts_dropped_full_chain"] = zero_greek_full_chain # FIX 48: expired contracts dropped by the raw calendar-date guard, and # profile outlier contracts dropped by the per-contract dominance guard. levels["expired_contracts_dropped"] = int(df_full.attrs.get("expired_contracts_dropped", 0)) levels["profile_outliers_dropped"] = (gp_outliers + dp_outliers) # FIX 12 / FIX 26: delta_neutral + dex_min on the WIDER dex grid dn, dn_crossings = find_delta_neutral(grid_dex, dp, spot) levels["delta_neutral"] = dn levels["delta_neutral_crossings"] = dn_crossings dmin, dmin_status = dex_min_price(grid_dex, dp) levels["dex_min_price"] = dmin levels["dex_min_status"] = dmin_status if dmin is None: logger.warning("dex_min_price is null (%s) — V-minimum is off the ±%.0f%% grid.", dmin_status, cfg_run.profile_band_dex * 100) # FIX 13 levels["outlier_report"] = build_outlier_report(agg, df_band, cfg_run, spot) # FIX 14 levels["oi_totals"] = build_oi_totals(df_full) # FIX 18 + FIX 25: per-expiry concentration levels["dealer_proxy"] = cfg_run.dealer_proxy gbe = build_gex_by_expiry(df_full, spot, cfg_run) levels["front_expiry_share"] = gbe["front_expiry_share"] levels["front_expiry"] = gbe["front_expiry"] levels["front_expiry_dte"] = gbe["front_expiry_dte"] levels["gex_by_expiry"] = gbe["gex_by_expiry"] levels["max_expiry_share"] = gbe["max_expiry_share"] levels["max_expiry"] = gbe["max_expiry"] levels["max_expiry_dte"] = gbe["max_expiry_dte"] levels["top3_expiry_share"] = gbe["top3_expiry_share"] # FIX 68 / FIX 86: always emit note keys (null when not applicable) so the JSON # key set is identical regardless of data (canonical schema). levels["concentration_note"] = None levels["top3_concentration_note"] = None levels["spread_note"] = None levels["put_heavy_note"] = None if gbe["max_expiry_share"] > 0.40: logger.warning("EXPIRY CONCENTRATION: %s (%dDTE) = %.0f%% of total |GEX|.", gbe["max_expiry"], gbe["max_expiry_dte"], gbe["max_expiry_share"] * 100) levels["concentration_note"] = ( f"{gbe['max_expiry']} ({gbe['max_expiry_dte']}DTE) " f"= {gbe['max_expiry_share']:.0%} of total |GEX|" ) if gbe["top3_expiry_share"] > 0.75: n_dom = sum(1 for r in sorted(gbe["gex_by_expiry"], key=lambda r: r["share"], reverse=True)[:3] if r["share"] > 0) levels["top3_concentration_note"] = f"chart dominated by {n_dom} expiries" # FIX 27 / FIX 46: ephemeral key levels levels_ephemeral = [] # FIX 68: always emit the per-level ephemeral/front-expiry keys (null default). for lvl_name in ("call_resistance", "put_support"): levels[f"{lvl_name}_front_expiry_abs_share"] = None levels[f"{lvl_name}_front_expiry_net_ratio"] = None levels[f"{lvl_name}_ephemeral_note"] = None for lvl_name, strike in (("call_resistance", levels["call_resistance"]), ("put_support", levels["put_support"])): fe = level_front_expiry_pct(df_full, strike, spot, cfg_run) if not fe: continue abs_share = fe["front_expiry_abs_share"] net_ratio = fe.get("front_expiry_net_ratio") front_exp = fe["front_expiry"] levels[f"{lvl_name}_front_expiry_abs_share"] = float(abs_share) if net_ratio is not None: levels[f"{lvl_name}_front_expiry_net_ratio"] = float(net_ratio) if abs_share > 0.50: levels_ephemeral.append(lvl_name) note = f"{lvl_name} {strike:g}: {abs_share:.0%} 0DTE — expires today" if net_ratio is not None and net_ratio > 1.0: note += " (offset by later expiries)" levels[f"{lvl_name}_ephemeral_note"] = note logger.warning("EPHEMERAL LEVEL: %s %s is %.0f%% front-expiry OI (%s).", lvl_name, strike, abs_share * 100, front_exp) levels["levels_ephemeral"] = levels_ephemeral # FIX 31/33: spread detection spread_info = detect_spread_candidates(df_full, spot, cfg_run, increment) levels["spread_candidates"] = spread_info["spread_candidates"] levels["spread_flagged_share"] = spread_info["spread_flagged_share"] levels["sensitivity_smaller_leg_sign_flipped"] = spread_info["sensitivity_smaller_leg_sign_flipped"] if spread_info["spread_flagged_share"] > cfg_run.spread_flag_threshold: n = len(spread_info["spread_candidates"]) pct = spread_info["spread_flagged_share"] levels["spread_note"] = ( f"{n} probable vertical-spread structures = {pct:.0%} of |GEX|; " f"gross-OI proxy overstates net dealer gamma here" ) # FIX 32: ATM IV cross-check against realised vol (FIX 43: reuse early RV) rv = rv_early iv_xc = atm_iv_cross_check(levels.get("atm_iv"), rv, cfg_run) levels["realised_vol_20d"] = iv_xc["realised_vol_20d"] levels["iv_hv_ratio"] = iv_xc["iv_hv_ratio"] levels["vol_regime"] = iv_xc["vol_regime"] # FIX 74: always emit atm_iv_status (null unless an outlier) for a canonical schema. levels["atm_iv_status"] = None if iv_xc.get("iv_hv_outlier"): levels["atm_iv_status"] = "iv_hv_outlier — verify" levels["iv_hv_outlier"] = True logger.warning("ATM IV outlier: iv/hv = %.2f (atm_iv=%.4f, rv=%.4f).", iv_xc["iv_hv_ratio"], levels.get("atm_iv", 0), rv or 0) else: levels["iv_hv_outlier"] = False levels["atm_iv_source_detail"] = _atm_iv_source_detail(df_band, levels, spot) # FIX 14 put-heavy pcr = levels.get("oi_put_call_ratio", 0.0) if pcr > 3.0: logger.warning("put/call OI = %.2f — unusually put-heavy; see oi_totals.", pcr) levels["put_heavy_note"] = f"put/call OI {pcr:.2f} — see per-expiry breakdown" # --- FIX 34: publish endpoint_variant and instrument_class --- levels["endpoint_variant"] = endpoint_variant levels["instrument_class"] = instrument_class # --- FIX 89: provenance stamp. Every artifact carries the producing code version # and the canonical schema version, both from gex/__init__.py. Identical # snapshot_id values across versions can produce different profile values (e.g. # FIX 84 changed the settlement clock, moving rel_err 0.005877 -> 0.006652 for # the same capture), so byte-identity under FIX 75 only holds WITHIN a version — # the history needs the version to be readable. levels["gex_version"] = _GEX_VERSION levels["schema_version"] = _GEX_SCHEMA_VERSION # --- FIX 37 / FIX 47 / FIX 74: index AM-settled front expiry. Always emit the # key (null for non-index instruments) so the canonical schema is identical # across all tickers regardless of instrument class. if instrument_class == "index": front_exp_str = gbe.get("front_expiry") am_settled = False if front_exp_str: try: front_exp_date = datetime.strptime(front_exp_str, "%Y-%m-%d").date() am_settled = is_third_friday(front_exp_date) except Exception: pass levels["front_expiry_am_settled"] = bool(am_settled) if am_settled: logger.info("Front expiry %s is a third Friday (AM-settled index option).", front_exp_str) else: levels["front_expiry_am_settled"] = None # FIX 74: null, not omitted # --- FIX 35: publish bands block --- levels["bands"] = bands # --- FIX 36: snap published levels to strike increment --- # Preserve the UNROUNDED value in *_raw. hvl_raw is already set (unrounded) by # compute_key_levels, so never overwrite an existing *_raw — otherwise the # already-snapped value would clobber the true crossing (e.g. 630.0 over 629.8185). # FIX 68 / FIX 86: always emit *_raw (even when the level is None) so the JSON # key set is identical regardless of data (canonical schema). for key in ("call_resistance", "put_support", "hvl", "gex_transition", "delta_neutral", "dex_min_price"): raw_key = f"{key}_raw" val = levels.get(key) if raw_key not in levels: levels[raw_key] = val # None if the level was not found if val is not None: levels[key] = snap_to_increment(val, increment) # --- FIX 36: timestamp for history CSV --- try: ts_et = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S").replace( tzinfo=ZoneInfo(cfg.source_timestamp_tz)).astimezone(ET) levels["timestamp_et"] = ts_et.strftime("%Y-%m-%d %H:%M") # FIX 75: source_timestamp_age_min is FROZEN at capture. It is the age of the # Cboe source timestamp as of the moment the chain was fetched (captured_at_utc), # NOT as of this render — so re-rendering an old snapshot reports the same # freshness, not "hours stale". The wall-clock delta since capture is published # separately as render_lag_min. When no capture time is available (legacy), fall # back to the render-time age (the pre-v1.7.5 behaviour). now_utc = datetime.now(timezone.utc) if captured_at_utc: captured = datetime.fromisoformat(captured_at_utc) if captured.tzinfo is None: captured = captured.replace(tzinfo=timezone.utc) age_min = (captured - ts_et).total_seconds() / 60.0 render_lag = (now_utc - captured).total_seconds() / 60.0 levels["render_lag_min"] = round(render_lag, 1) else: age_min = (datetime.now(ET) - ts_et).total_seconds() / 60.0 levels["render_lag_min"] = None levels["source_timestamp_age_min"] = round(age_min, 1) if age_min > 45: logger.warning("Source timestamp age %.1f min (>45) for %s.", age_min, symbol) except Exception: levels["timestamp_et"] = ts_str levels["render_lag_min"] = None logger.info("Levels%s: %s", suffix or "", json.dumps({k: round(v, 4) if isinstance(v, float) else v for k, v in levels.items() if not isinstance(v, (dict, list))}, default=str)) # --- FIX 50: publish visible-window profile maxima and compute axis limits --- # Visible window = the same lo/hi the plot uses (key levels + plot_band). _lo = min(spot * (1 - cfg_run.plot_band), levels.get("put_support", spot), levels.get("call_resistance", spot), levels.get("hvl") or spot, levels.get("gex_transition") or spot) _hi = max(spot * (1 + cfg_run.plot_band), levels.get("put_support", spot), levels.get("call_resistance", spot), levels.get("hvl") or spot, levels.get("gex_transition") or spot) _gvis = gp[(grid >= _lo) & (grid <= _hi)] _dvis = dp[(grid_dex >= _lo) & (grid_dex <= _hi)] gex_pmax = float(np.max(np.abs(_gvis))) if len(_gvis) else 0.0 dex_pmax = float(np.max(np.abs(_dvis))) if len(_dvis) else 0.0 levels["gex_profile_max"] = gex_pmax levels["dex_profile_max"] = dex_pmax # Rolling-mode limits (cross-snapshot comparable); None -> fall back to "data". rolling_gex = _rolling_profile_limit(symbol, "gex_profile_max", cfg_run) \ if cfg_run.profile_axis_mode == "rolling" else None rolling_dex = _rolling_profile_limit(symbol, "dex_profile_max", cfg_run) \ if cfg_run.profile_axis_mode == "rolling" else None # FIX 79b: non-destructive writes. Compute the output path up front (same # naming rule as render_chart: {date}_{slot+suffix}.json under outdir/symbol) # and refuse BEFORE doing any expensive work if an existing file carries a # DIFFERENT snapshot_id. A re-render of the SAME snapshot_id is always allowed # (FIX 75's byte-identity guarantee depends on it). --overwrite forces the write. # The date is the ET date — render_chart parses ts_str as UTC (source_timestamp_tz) # and converts to ET before formatting the filename, so mirror that exactly. try: _ts_utc = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S").replace( tzinfo=ZoneInfo(cfg.source_timestamp_tz)) date_str = _ts_utc.astimezone(ET).strftime("%Y-%m-%d") except Exception: date_str = datetime.now(ET).strftime("%Y-%m-%d") out_json_path = Path(cfg.outdir) / symbol / f"{date_str}_{slot + suffix}.json" if not overwrite and out_json_path.exists(): try: with open(out_json_path) as _f: existing_id = json.load(_f).get("snapshot_id") except Exception: existing_id = None if existing_id != snap_id: logger.error("REFUSING to overwrite %s: existing snapshot_id=%r differs " "from incoming %r. Re-render the same snapshot or pass " "--overwrite to force.", out_json_path, existing_id, snap_id) return None # render chart (GEX profile on `grid`, DEX profile on the wider `grid_dex`) display_label = cfg.display_labels.get(symbol, symbol) png_path = render_chart( symbol, agg, grid, gp, grid_dex, dp, levels, ts_str, cfg_run, cfg.outdir, slot + suffix, increment, display_label=display_label, render_bucket=render_bucket_val, rolling_gex_limit=rolling_gex, rolling_dex_limit=rolling_dex, ) # save JSON sibling json_path = png_path.with_suffix(".json") out_data = { "symbol": symbol, "snapshot_id": snap_id, "timestamp": ts_str, "spot": spot, "slot": slot, "variant": suffix.lstrip("_") or "all_expirations", **levels, } with open(json_path, "w") as f: json.dump(out_data, f, indent=2, default=str) logger.info("JSON -> %s", json_path) # save parquet pq_path = png_path.with_suffix(".parquet") agg.to_parquet(pq_path) logger.info("Parquet -> %s", pq_path) # --- FIX 40: append history CSV --- variant_label = suffix.lstrip("_") or "all_expirations" _append_history_csv(symbol, slot, variant_label, out_data, cfg) return png_path def main(): parser = argparse.ArgumentParser(description="Net GEX All Expirations snapshot") parser.add_argument("--tickers", default=None, help="Comma-separated tickers (default: all from config)") parser.add_argument("--slot", default="auto", choices=["auto", "am", "pm"]) parser.add_argument("--from-cache", action="store_true") parser.add_argument("--outdir", default="out") parser.add_argument("--cache-dir", default="data/raw") parser.add_argument("--replay-unsafe", action="store_true", help="FIX 52: bypass ONLY the FIX 49 trading-day/session-window " "check so an out-of-session cached chain can exercise FIX " "48/50. Output is flagged non-publishable and forced to a " "scratch outdir (out_replay/), never gex_out/.") parser.add_argument("--overwrite", action="store_true", help="FIX 79: force-overwrite an existing output whose " "snapshot_id differs. Default is non-destructive: a " "capture that would clobber a different snapshot is " "refused. Re-renders of the same snapshot_id are always " "allowed regardless of this flag.") parser.add_argument("-v", "--verbose", action="store_true") args = parser.parse_args() logging.basicConfig( level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s", ) # FIX 52: replay output must never land in the publishable outdir. Force a scratch # path regardless of --outdir so a stray --replay-unsafe cannot poison gex_out/. outdir = args.outdir if args.replay_unsafe: outdir = "out_replay" logger.error("REPLAY MODE — output forced to scratch outdir %r; not publishable.", outdir) cfg = GexConfig(outdir=outdir, cache_dir=args.cache_dir) # FIX 34: default to all tickers from config if args.tickers: tickers = [t.strip().upper() for t in args.tickers.split(",") if t.strip()] else: tickers = list(cfg.tickers) now_et = datetime.now(ET) if args.slot == "auto" and not args.from_cache: slot = _slot_label(now_et, cfg) if not slot: logger.info("Outside slot window (%s ET). Exiting.", now_et.strftime("%H:%M")) sys.exit(0) if not _is_trading_day(now_et.date()): logger.info("Not an NYSE trading day. Exiting.") sys.exit(0) else: slot = args.slot if args.slot != "auto" else "pm" results = {} for t in tickers: results[t] = run_ticker(t, cfg, slot, from_cache=args.from_cache, replay_unsafe=args.replay_unsafe, overwrite=args.overwrite) failed = [t for t, ok in results.items() if not ok] if failed and len(failed) == len(tickers): logger.error("ALL tickers failed: %s", failed) sys.exit(1) elif failed: logger.warning("Some tickers failed: %s", failed) logger.info("Done. Success: %s", [t for t, ok in results.items() if ok]) # FIX 38: regenerate the page index after every run build_index_json(cfg) if __name__ == "__main__": main()