"""Tests for gex.compute, gex.greeks, and gex.plot.""" import json from datetime import date, datetime, time, timedelta from zoneinfo import ZoneInfo import numpy as np import pandas as pd import pytest from gex.compute import ( aggregate, atm_expected_move, build_gex_by_expiry, build_oi_totals, build_outlier_report, compute_hvl_candidates, compute_levels, detect_increment, detect_render_spacing, dex_min_price, dex_profile, filter_contracts, filter_contracts_band, filter_contracts_full, find_delta_neutral, find_hvl, gex_profile, reconcile, reconcile_by_expiry, render_bucket, time_to_expiry_years, total_net_gex_from_contracts, ) from gex.config import GexConfig from gex.greeks import bs_gamma _ET = ZoneInfo("America/New_York") # --------------------------------------------------------------------------- # 1. Synthetic chain fixture: 2 strikes, 1 call + 1 put each # --------------------------------------------------------------------------- @pytest.fixture def synthetic_chain(): """2 strikes (100, 110), 1 call + 1 put each, known gamma/delta/OI.""" return [ {"strike": 100.0, "expiry": date(2026, 8, 15), "cp": "C", "iv": 0.25, "oi": 1000, "volume": 500, "delta": 0.60, "gamma": 0.04, "vega": 0.12, "theta": -0.05, "theo": 8.5, "bid": 8.4, "ask": 8.6}, {"strike": 100.0, "expiry": date(2026, 8, 15), "cp": "P", "iv": 0.28, "oi": 800, "volume": 300, "delta": -0.40, "gamma": 0.04, "vega": 0.11, "theta": -0.04, "theo": 3.2, "bid": 3.1, "ask": 3.3}, {"strike": 110.0, "expiry": date(2026, 8, 15), "cp": "C", "iv": 0.22, "oi": 600, "volume": 200, "delta": 0.35, "gamma": 0.03, "vega": 0.10, "theta": -0.04, "theo": 4.1, "bid": 4.0, "ask": 4.2}, {"strike": 110.0, "expiry": date(2026, 8, 15), "cp": "P", "iv": 0.30, "oi": 900, "volume": 400, "delta": -0.65, "gamma": 0.03, "vega": 0.09, "theta": -0.03, "theo": 6.8, "bid": 6.7, "ask": 6.9}, ] def test_aggregate_net_gex(synthetic_chain): """Assert net_gex, dex, and put-call ratios to 1e-6.""" cfg = GexConfig(strike_band=0.50, dte_max=365) spot = 105.0 today = date(2026, 7, 24) df = filter_contracts(synthetic_chain, spot, cfg, today) assert len(df) == 4, f"Expected 4 contracts, got {len(df)}" agg = aggregate(df, spot, cfg) M = 100 S2 = spot * spot gex_call_100 = 0.04 * 1000 * M * S2 * 0.01 gex_put_100 = -0.04 * 800 * M * S2 * 0.01 net_100 = gex_call_100 + gex_put_100 gex_call_110 = 0.03 * 600 * M * S2 * 0.01 gex_put_110 = -0.03 * 900 * M * S2 * 0.01 net_110 = gex_call_110 + gex_put_110 assert abs(agg.loc[100.0, "gex_call"] - gex_call_100) < 1e-6 assert abs(agg.loc[100.0, "gex_put"] - gex_put_100) < 1e-6 assert abs(agg.loc[100.0, "net_gex"] - net_100) < 1e-6 assert abs(agg.loc[110.0, "gex_call"] - gex_call_110) < 1e-6 assert abs(agg.loc[110.0, "gex_put"] - gex_put_110) < 1e-6 assert abs(agg.loc[110.0, "net_gex"] - net_110) < 1e-6 dex_100 = (0.60 * 1000 + (-0.40) * 800) * M * spot dex_110 = (0.35 * 600 + (-0.65) * 900) * M * spot assert abs(agg.loc[100.0, "dex"] - dex_100) < 1e-6 assert abs(agg.loc[110.0, "dex"] - dex_110) < 1e-6 total_gex_put = abs(gex_put_100 + gex_put_110) total_gex_call = gex_call_100 + gex_call_110 expected_gex_pcr = total_gex_put / total_gex_call total_oi_put = 800 + 900 total_oi_call = 1000 + 600 expected_oi_pcr = total_oi_put / total_oi_call levels = compute_levels(agg, np.array([100.0, 110.0]), np.array([-1e6, 1e6]), spot, 10.0) assert abs(levels["gex_put_call_ratio"] - expected_gex_pcr) < 1e-6 assert abs(levels["oi_put_call_ratio"] - expected_oi_pcr) < 1e-6 # --------------------------------------------------------------------------- # 2. bs_gamma sanity # --------------------------------------------------------------------------- def test_bs_gamma_atm_gt_otm(): S, T, iv, r, q = 100.0, 30 / 252, 0.25, 0.04, 0.0 assert bs_gamma(S, 100.0, T, iv, r, q) > bs_gamma(S, 120.0, T, iv, r, q) def test_bs_gamma_decays_with_T(): S, K, iv, r, q = 100.0, 150.0, 0.25, 0.04, 0.0 gamma_long = bs_gamma(S, K, 1.0, iv, r, q) gamma_short = bs_gamma(S, K, 0.001, iv, r, q) assert gamma_short < gamma_long assert gamma_short < 1e-10 # --------------------------------------------------------------------------- # 3. HVL interpolation on hand-built profile (find_hvl now takes spot, returns 4) # --------------------------------------------------------------------------- def test_hvl_zero_crossing(): grid = np.array([90.0, 95.0, 100.0, 105.0, 110.0]) profile = np.array([-5e6, -2e6, -0.5e6, 1.5e6, 4e6]) # crossing between 100 (-0.5M) and 105 (+1.5M) -> 101.25 hvl, rule, conf, crossings = find_hvl(grid, profile, spot=100.0) assert rule == "zero_crossing" assert conf == "high" assert abs(hvl - 101.25) < 1e-6, f"Expected 101.25, got {hvl}" assert len(crossings) == 1 def test_hvl_no_crossing_midpoint(): """FIX 28: inflection fallback deleted. All-positive profile with no crossing returns midpoint with 'no_crossing_midpoint' rule and 'low' confidence.""" grid = np.array([90.0, 95.0, 100.0, 105.0, 110.0]) profile = np.array([1e6, 2e6, 3e6, 2e6, 1e6]) # all positive, no crossing hvl, rule, conf, crossings = find_hvl(grid, profile, spot=100.0) assert rule == "no_crossing_midpoint" assert conf == "low" # --------------------------------------------------------------------------- # 4. Golden-image smoke test # --------------------------------------------------------------------------- def test_golden_image_smoke(tmp_path): from gex.plot import render_chart cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) spot = 105.0 strikes = np.arange(95, 116, 5.0) agg = pd.DataFrame({ "gex_call": np.random.uniform(1e6, 5e6, len(strikes)), "gex_put": -np.random.uniform(1e6, 5e6, len(strikes)), "dex": np.random.uniform(-2e7, 2e7, len(strikes)), "oi_call": np.random.randint(100, 5000, len(strikes)), "oi_put": np.random.randint(100, 5000, len(strikes)), }, index=strikes) agg["net_gex"] = agg["gex_call"] + agg["gex_put"] grid = np.arange(90.0, 121.0, 1.0) gp = np.linspace(-3e6, 3e6, len(grid)) dp = np.linspace(-1e9, 1e9, len(grid)) # simulated DEX (monotonic) increment = 5.0 levels = compute_levels(agg, grid, gp, spot, increment) png = render_chart("TEST", agg, grid, gp, grid, dp, levels, "2026-07-24 15:44:00", cfg, str(tmp_path), "pm", increment) assert png.exists(), f"PNG not found: {png}" assert png.stat().st_size > 50_000, f"PNG too small: {png.stat().st_size} bytes" # --------------------------------------------------------------------------- # 5. detect_increment # --------------------------------------------------------------------------- def test_detect_increment(): assert detect_increment(np.array([100, 105, 110, 115, 120, 125, 130])) == 5.0 assert detect_increment(np.array([50, 51, 52, 53, 54, 55])) == 1.0 # =========================================================================== # NEW TESTS (v1.1.0 audit fixes) # =========================================================================== # --------------------------------------------------------------------------- # 6. FIX 1: timestamp UTC -> ET # --------------------------------------------------------------------------- def test_timestamp_utc_to_et(): """'2026-07-24 19:35:00' UTC -> '2026-07-24 15:35 EDT'.""" from gex.plot import format_timestamp label = format_timestamp("2026-07-24 19:35:00", source_tz="UTC") assert label == "2026-07-24 15:35 EDT", f"got {label!r}" # --------------------------------------------------------------------------- # 7. FIX 2: profile uses the FULL chain (no strike-band truncation) # --------------------------------------------------------------------------- def test_profile_uses_full_chain(): """Heavy call OI at spot*1.20 must shift the profile zero-crossing. If the profile were fed the truncated (±12%) band, that far-OTM call OI would be discarded and the crossing would not move — proving no truncation. """ cfg = GexConfig(strike_band=0.12, profile_band=0.25) spot = 100.0 today = date(2026, 7, 24) exp = date(2026, 8, 15) base = [ {"strike": 95.0, "expiry": exp, "cp": "P", "iv": 0.30, "oi": 5000, "volume": 0, "delta": -0.4, "gamma": 0.03, "vega": 0.1, "theta": -0.05, "theo": 3.0, "bid": 2.9, "ask": 3.1}, {"strike": 105.0, "expiry": exp, "cp": "C", "iv": 0.25, "oi": 1000, "volume": 0, "delta": 0.4, "gamma": 0.03, "vega": 0.1, "theta": -0.05, "theo": 3.0, "bid": 2.9, "ask": 3.1}, ] far_call = {"strike": 120.0, "expiry": exp, "cp": "C", "iv": 0.22, "oi": 50000, "volume": 0, "delta": 0.2, "gamma": 0.01, "vega": 0.1, "theta": -0.03, "theo": 2.0, "bid": 1.9, "ask": 2.1} grid = np.arange(spot * 0.75, spot * 1.25 + 0.5, 0.5) df_without = filter_contracts_full(base, spot, cfg, today) df_with = filter_contracts_full(base + [far_call], spot, cfg, today) # the far call (strike 120 = spot*1.20) is OUTSIDE the ±12% band (112) but in the full frame assert (df_with["strike"] == 120.0).any(), "far call must survive full filter" assert not (filter_contracts(base + [far_call], spot, cfg, today)["strike"] == 120.0).any(), \ "far call must be dropped by the band filter" agg = aggregate(df_with, spot, cfg) gp_without, _ = gex_profile(df_without, spot, cfg, grid) # FIX 14: agg dropped gp_with, _ = gex_profile(df_with, spot, cfg, grid) # the profiles must differ where the far call OI has influence (high strikes) assert not np.allclose(gp_without, gp_with), \ "profile must change when far-OTM call OI is added (full chain, not truncated)" # specifically, adding call OI raises the profile in the upper wing upper = grid > 112 assert gp_with[upper].mean() > gp_without[upper].mean() # --------------------------------------------------------------------------- # 8. FIX 12: DEX profile is V-shaped (NOT monotonic) — interior minimum # --------------------------------------------------------------------------- def test_dex_profile_v_shape(): """Put OI at spot*1.15 + call OI at spot*0.85 => V-shaped DEX with an interior minimum (supersedes the old monotonicity test, which was wrong). At low s the high-strike puts are deep ITM (delta ~ -1) so dex ~ -100*s*OI_put decreases in s; at high s the low-strike calls dominate and dex increases. The minimum must be strictly interior (not index 0 or -1). """ cfg = GexConfig(profile_band=0.40) spot = 100.0 today = date(2026, 7, 24) exp = date(2026, 8, 15) chain = [ # puts at spot*1.15 = 115 {"strike": 115.0, "expiry": exp, "cp": "P", "iv": 0.25, "oi": 20000, "volume": 0, "delta": -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, # calls at spot*0.85 = 85 {"strike": 85.0, "expiry": exp, "cp": "C", "iv": 0.25, "oi": 20000, "volume": 0, "delta": 0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, ] df_full = filter_contracts_full(chain, spot, cfg, today) grid = np.arange(spot * 0.60, spot * 1.40 + 0.5, 0.5) dp, _ = dex_profile(df_full, cfg, grid, spot) imin = int(np.argmin(dp)) assert 0 < imin < len(dp) - 1, \ f"DEX minimum must be strictly interior, got index {imin} of {len(dp)}" # and it is genuinely V-shaped: both ends above the minimum assert dp[0] > dp[imin] and dp[-1] > dp[imin], "DEX profile must be V-shaped" # --------------------------------------------------------------------------- # 9. FIX 10: three x-axes (bars + GEX profile + DEX profile), zeros aligned # --------------------------------------------------------------------------- def test_three_axis_zero_aligned(tmp_path): """FIX 10: separate twin axis per profile. All three x-axes share y (strikes) and their zeros must coincide in display coords (warn-not-assert in render).""" import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt from gex.plot import render_chart cfg = GexConfig(outdir=str(tmp_path)) spot = 100.0 strikes = np.arange(90.0, 111.0, 2.0) agg = pd.DataFrame({ "gex_call": np.linspace(1e6, 4e6, len(strikes)), "gex_put": -np.linspace(2e6, 1e6, len(strikes)), "dex": np.zeros(len(strikes)), "oi_call": np.full(len(strikes), 1000), "oi_put": np.full(len(strikes), 1200), }, index=strikes) agg["net_gex"] = agg["gex_call"] + agg["gex_put"] grid = np.arange(85.0, 116.0, 0.5) gp = np.linspace(-3e6, 3e6, len(grid)) dp = np.linspace(-1e9, 1e9, len(grid)) levels = compute_levels(agg, grid, gp, spot, 2.0) captured = {} orig = plt.Figure.savefig def spy(self, *a, **k): captured["fig"] = self return orig(self, *a, **k) plt.Figure.savefig = spy try: render_chart("TEST", agg, grid, gp, grid, dp, levels, "2026-07-24 15:44:00", cfg, str(tmp_path), "pm", 2.0) finally: plt.Figure.savefig = orig fig = captured["fig"] ax = fig.axes[0] gex_axes = [a for a in fig.axes if a is not ax and a.get_xlabel().startswith("GEX Profile")] dex_axes = [a for a in fig.axes if a is not ax and a.get_xlabel().startswith("DEX Profile")] assert len(gex_axes) == 1, "GEX profile axis must exist" assert len(dex_axes) == 1, "DEX profile axis must exist" # colour-matching: GEX axis yellow, DEX axis orange assert gex_axes[0].xaxis.label.get_color() == cfg.gex_profile_color assert dex_axes[0].xaxis.label.get_color() == cfg.dex_color # all three zeros coincide fig.canvas.draw() d0 = ax.transData.transform((0.0, 0.0))[0] dg = gex_axes[0].transData.transform((0.0, 0.0))[0] dd = dex_axes[0].transData.transform((0.0, 0.0))[0] assert abs(d0 - dg) < 0.5, f"GEX-axis zero misaligned: {abs(d0-dg):.3f}px" assert abs(d0 - dd) < 0.5, f"DEX-axis zero misaligned: {abs(d0-dd):.3f}px" # --------------------------------------------------------------------------- # 10. FIX 7: robust x-limits resist a single outlier bar # --------------------------------------------------------------------------- def test_robust_xlim_outlier(): """One 8e8 outlier among ~1e7 bars must NOT dominate the axis (FIX 7). The robust limit is clean(1.15 * max(p97, median*8)). p97 excludes the single outlier (it sits in the ~1e7 bulk); the median*8 floor caps the axis near the bulk. The result is far below the naive max-based limit (1.15*8e8 = 920M), which is the actual defect being fixed — the outlier no longer flattens the mid-window bars to hairlines. """ from gex.plot import _clean_xlim net = np.array([1e7] * 100 + [8e8]) nz = np.abs(net[net != 0]) p97 = np.percentile(nz, 97) med8 = np.median(nz) * 8 xlim = _clean_xlim(1.15 * max(p97, med8)) naive = 1.15 * 8e8 # the outlier must not set the axis: robust limit is a small fraction of naive assert xlim < 0.2 * naive, f"xlim={xlim:.2e} should be << naive {naive:.2e}" # and it is governed by the bulk (median*8 = 8e7), not the outlier — i.e. within # one clean-step rounding of 1.15*med8 (100M is the clean step above 92M) assert xlim <= 1.15 * med8 * 1.10 + 1e7, f"xlim={xlim:.2e} should track the bulk" # p97 itself must have rejected the outlier assert p97 < 1e8, f"p97={p97:.2e} should sit in the bulk, not at the outlier" # --------------------------------------------------------------------------- # 11. FIX 6: HVL low-confidence flag when there is no crossing # --------------------------------------------------------------------------- def test_hvl_low_confidence_flag(): """With an all-positive FLAT profile (no zero-crossing), HVL is null with status 'no_flip_in_range' (FIX 29). No inflection fallback (FIX 28).""" strikes = np.arange(90.0, 111.0, 5.0) agg = pd.DataFrame({ "gex_call": np.full(len(strikes), 1e6), "gex_put": np.full(len(strikes), -0.5e6), "dex": np.zeros(len(strikes)), "oi_call": np.full(len(strikes), 1000), "oi_put": np.full(len(strikes), 800), }, index=strikes) agg["net_gex"] = agg["gex_call"] + agg["gex_put"] grid = np.arange(85.0, 116.0, 1.0) profile = np.full(len(grid), 2e6) # all positive, flat -> no crossing cfg = GexConfig() levels = compute_levels(agg, grid, profile, spot=100.0, increment=5.0, cfg=cfg) # flat profile: no zero crossing -> hvl is None, status no_flip_in_range assert levels["hvl"] is None assert levels["hvl_status"] == "no_flip_in_range" assert levels["hvl_rule"] == "zero_cross" assert levels["hvl_confidence"] == "n/a" # =========================================================================== # NEW TESTS (v1.2.0 audit fixes) # =========================================================================== # --------------------------------------------------------------------------- # 12. FIX 12: delta_neutral picks the crossing NEAREST spot; crossings listed # --------------------------------------------------------------------------- def test_delta_neutral_nearest_spot(): """A V-shaped DEX profile crosses zero twice; pick the one nearest spot.""" grid = np.array([80.0, 90.0, 100.0, 110.0, 120.0]) # crosses zero near 85 (rising) and near 115 (rising again) — two crossings dp = np.array([-2e9, -0.5e9, 1e9, -0.5e9, -2e9]) # not monotonic, two sign changes nearest, crossings = find_delta_neutral(grid, dp, spot=100.0) assert len(crossings) >= 1 # nearest must be the crossing closest to spot=100 assert nearest == min(crossings, key=lambda c: abs(c - 100.0)) def test_dex_min_price_interior(): """dex_min_price returns (price, status) at the V minimum; interior -> 'interior'.""" grid = np.array([80.0, 90.0, 100.0, 110.0, 120.0]) dp = np.array([-1e9, -3e9, -5e9, -3e9, -1e9]) # min at index 2 (100.0) price, status = dex_min_price(grid, dp) assert price == 100.0 assert status == "interior" # --------------------------------------------------------------------------- # 13. FIX 13: outlier_report has top strikes with per-expiry breakdown # --------------------------------------------------------------------------- def test_outlier_report_structure(synthetic_chain): cfg = GexConfig(strike_band=0.50) spot = 105.0 today = date(2026, 7, 24) df = filter_contracts(synthetic_chain, spot, cfg, today) agg = aggregate(df, spot, cfg) report = build_outlier_report(agg, df, cfg, spot, top_n=5) assert "top_strikes" in report assert len(report["top_strikes"]) > 0 top = report["top_strikes"][0] for field in ("strike", "net_gex", "oi_call", "oi_put", "by_expiry"): assert field in top, f"missing {field} in outlier_report entry" # per-expiry breakdown entries carry expiry + oi + gex exp0 = top["by_expiry"][0] for field in ("expiry", "oi_call", "oi_put", "gex"): assert field in exp0, f"missing {field} in per-expiry breakdown" # --------------------------------------------------------------------------- # 14. FIX 14: oi_totals totals + DTE buckets; gex_profile has no agg arg # --------------------------------------------------------------------------- def test_oi_totals(synthetic_chain): cfg = GexConfig(strike_band=0.50) spot = 105.0 today = date(2026, 7, 24) df_full = filter_contracts_full(synthetic_chain, spot, cfg, today) totals = build_oi_totals(df_full) # calls: 1000 + 600 = 1600; puts: 800 + 900 = 1700 assert totals["call_oi"] == 1600 assert totals["put_oi"] == 1700 assert totals["n_contracts"] == 4 assert totals["n_expiries"] == 1 # all OI lands in exactly one DTE bucket bucket_sum = sum(totals["oi_by_dte_bucket"].values()) assert bucket_sum == 1600 + 1700 def test_gex_profile_signature_no_agg(): """FIX 14: gex_profile no longer takes agg (contracts_df, spot, cfg, grid).""" import inspect sig = inspect.signature(gex_profile) params = list(sig.parameters.keys()) assert "agg" not in params, f"gex_profile still has 'agg': {params}" assert params[0] == "contracts_df" # =========================================================================== # NEW TESTS (v1.3.0 audit fixes) # =========================================================================== def _load_smh_fixture(): """Load the cached SMH fixture and return (contracts, spot, ts_str).""" import glob, gzip, json, os cache_dir = os.path.join(os.path.dirname(__file__), "..", "data", "raw") paths = sorted(glob.glob(os.path.join(cache_dir, "SMH_*.json.gz"))) assert paths, f"No cached SMH fixture found in {cache_dir}" with gzip.open(paths[-1], "rt", encoding="utf-8") as f: data = json.load(f) from gex.fetch import parse_chain contracts, spot, ts_str = parse_chain(data, "SMH") return contracts, spot, ts_str # --------------------------------------------------------------------------- # 1. FIX 15: reconcile within tolerance on the cached SMH fixture # --------------------------------------------------------------------------- def test_reconcile_within_tolerance(): """rel_err < 0.05 on the cached SMH fixture (bars and profiles agree at spot).""" contracts, spot, ts_str = _load_smh_fixture() cfg = GexConfig() snap_et = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S").replace( tzinfo=ZoneInfo(cfg.source_timestamp_tz)).astimezone(_ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) df_band = filter_contracts_band(df_full, spot, cfg) agg = aggregate(df_band, spot, cfg) increment = detect_increment(agg.index.to_numpy()) step = max(increment / 5.0, 0.10) grid = np.arange(spot * (1 - cfg.profile_band), spot * (1 + cfg.profile_band) + step, step) gp, _ = gex_profile(df_full, spot, cfg, grid) total_full = total_net_gex_from_contracts(df_full, spot, cfg) rec = reconcile(total_full, grid, gp, spot) assert rec["pass"], f"reconciliation failed: rel_err={rec['rel_err']:.4f}" assert rec["rel_err"] < 0.05, f"rel_err={rec['rel_err']:.4f} >= 0.05" # --------------------------------------------------------------------------- # 2. FIX 15: continuous time-to-expiry # --------------------------------------------------------------------------- def test_time_to_expiry_continuous(): """FIX 65: calendar-time T for ALL expiries (not just 0DTE). T = minutes_to_settlement / (365*24*60). Settlement = 16:00 ET (09:30 for AM-settled).""" cfg = GexConfig() expiry = date(2026, 7, 24) # same day as the snapshot YEAR_MIN = 365.0 * 24.0 * 60.0 floor = 1.0 / (252.0 * 13.0) # ≈ 2.67 calendar-hours; caps gamma as T->0 # at 12:00 ET: 240 clock-minutes to 16:00 settlement (above the floor) now_1200 = datetime(2026, 7, 24, 12, 0, tzinfo=_ET) T_1200 = time_to_expiry_years(expiry, now_1200, cfg) assert abs(T_1200 - 240.0 / YEAR_MIN) < 1e-12, f"T at 12:00 = {T_1200}" # at 09:30 ET: 390 clock-minutes (6.5h) to 16:00 settlement now_0930 = datetime(2026, 7, 24, 9, 30, tzinfo=_ET) T_0930 = time_to_expiry_years(expiry, now_0930, cfg) assert abs(T_0930 - 390.0 / YEAR_MIN) < 1e-12, f"T at 09:30 = {T_0930}" # at 15:35 ET: only 25 clock-minutes left, below the floor -> T = floor now_1535 = datetime(2026, 7, 24, 15, 35, tzinfo=_ET) T_1535 = time_to_expiry_years(expiry, now_1535, cfg) assert T_1535 == floor, f"T at 15:35 should be floor {floor}, got {T_1535}" # after 16:00 ET: settlement passed -> minutes floored at 0 -> T = floor now_1630 = datetime(2026, 7, 24, 16, 30, tzinfo=_ET) T_1630 = time_to_expiry_years(expiry, now_1630, cfg) assert T_1630 == floor, f"T after 16:00 should be floor {floor}, got {T_1630}" # AM-settled same-day: settlement at 09:30. At 09:00 only 30 min remain, # below the floor -> T = floor (the 09:30 settlement clock is selected). T_am = time_to_expiry_years(expiry, datetime(2026, 7, 24, 9, 0, tzinfo=_ET), cfg, am_settled=True) assert T_am == floor, f"T am-settled at 09:00 should be floor {floor}, got {T_am}" # FIX 65: MULTI-DAY path also uses calendar time. Snapshot Fri 07-24 15:35 ET, # expiry Tue 07-28: 4 calendar days + 25 min = 5785 min to 16:00 ET settlement. later = date(2026, 7, 28) T_multi = time_to_expiry_years(later, now_1535, cfg) expected_multi = 5785.0 / YEAR_MIN assert abs(T_multi - expected_multi) < 1e-12, \ f"multi-day T should be {expected_multi}, got {T_multi}" # sanity: multi-day T >> same-day T assert T_multi > T_1200 # --------------------------------------------------------------------------- # 3. FIX 16: three HVL candidates, default inflection, method-sensitivity flag # --------------------------------------------------------------------------- def test_hvl_zero_cross_only(): """FIX 29: HVL is always the zero crossing nearest spot. compute_hvl_candidates (deprecated wrapper) returns zero_cross as the only candidate, no inflection.""" cfg = GexConfig() # hvl_rule = "zero_cross" spot = 100.0 grid = np.arange(80.0, 120.5, 0.5) # profile crosses zero near 105 profile = (grid - 105.0) * 1e6 strikes = np.arange(85.0, 116.0, 5.0) agg = pd.DataFrame({ "gex_call": np.linspace(1e6, 4e6, len(strikes)), "gex_put": -np.linspace(3e6, 1e6, len(strikes)), "dex": np.zeros(len(strikes)), "oi_call": np.full(len(strikes), 1000), "oi_put": np.full(len(strikes), 1200), }, index=strikes) agg["net_gex"] = agg["gex_call"] + agg["gex_put"] result = compute_hvl_candidates(grid, profile, agg, spot, cfg) # only zero_cross in candidates assert "zero_cross" in result["hvl_candidates"] assert result["hvl_candidates"]["zero_cross"] is not None assert abs(result["hvl_candidates"]["zero_cross"] - 105.0) < 1.0 assert result["hvl_rule_used"] == "zero_cross" assert result["hvl_confidence"] == "high" # inflection retired assert result["hvl_inflection_status"] == "retired_v1.5.0" assert result["hvl_spread_pct"] == 0.0 # --------------------------------------------------------------------------- # 4. FIX 17: ATM IV rejects garbage # --------------------------------------------------------------------------- def test_atm_iv_rejects_garbage(): """Chain whose only near-spot contract has iv=0.83 and OI=3 -> rejected.""" cfg = GexConfig() spot = 100.0 today = date(2026, 7, 24) # only one expiry with DTE >= 5, one strike near spot, OI=3 (fails OI>=100) chain = [ {"strike": 100.0, "expiry": date(2026, 8, 15), "cp": "C", "iv": 0.83, "oi": 3, "volume": 0, "delta": 0.5, "gamma": 0.03, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, ] df = filter_contracts_full(chain, spot, cfg, today) result = atm_expected_move(df, spot) assert result["atm_iv"] is None, f"atm_iv should be None, got {result['atm_iv']}" assert result["atm_iv_status"].startswith("rejected"), \ f"status should start with 'rejected', got {result['atm_iv_status']}" # --------------------------------------------------------------------------- # 5. FIX 18: front_expiry_share # --------------------------------------------------------------------------- def test_front_expiry_share(): """Fixture where the front expiry dominates Σ|net_gex| -> front_expiry_share > 0.40.""" cfg = GexConfig() spot = 100.0 today = date(2026, 7, 24) # front expiry (0DTE): put-heavy -> large negative net GEX # later expiry: call-heavy -> smaller positive net GEX chain = [ # front expiry (0DTE): put OI >> call OI -> net GEX ∝ (10000-60000)*0.05 = -2500 {"strike": 100.0, "expiry": date(2026, 7, 24), "cp": "C", "iv": 0.30, "oi": 10000, "volume": 0, "delta": 0.5, "gamma": 0.05, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, {"strike": 100.0, "expiry": date(2026, 7, 24), "cp": "P", "iv": 0.30, "oi": 60000, "volume": 0, "delta": -0.5, "gamma": 0.05, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, # later expiry: call OI >> put OI -> net GEX ∝ (40000-10000)*0.03 = +900 {"strike": 100.0, "expiry": date(2026, 8, 15), "cp": "C", "iv": 0.25, "oi": 40000, "volume": 0, "delta": 0.5, "gamma": 0.03, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, {"strike": 100.0, "expiry": date(2026, 8, 15), "cp": "P", "iv": 0.25, "oi": 10000, "volume": 0, "delta": -0.5, "gamma": 0.03, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, ] df_full = filter_contracts_full(chain, spot, cfg, today) result = build_gex_by_expiry(df_full, spot, cfg) # front share = 2500 / (2500 + 900) ≈ 0.735 > 0.40 assert result["front_expiry_share"] > 0.40, \ f"front_expiry_share {result['front_expiry_share']:.3f} should be > 0.40" assert result["front_expiry_dte"] == 0 assert len(result["gex_by_expiry"]) == 2 # --------------------------------------------------------------------------- # 6. FIX 19: no symlog # --------------------------------------------------------------------------- def test_no_symlog(tmp_path): """Render and assert ax.get_xscale() == 'linear' (symlog removed).""" import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt from gex.plot import render_chart cfg = GexConfig(outdir=str(tmp_path)) spot = 100.0 strikes = np.arange(90.0, 111.0, 2.0) agg = pd.DataFrame({ "gex_call": np.linspace(1e6, 4e6, len(strikes)), "gex_put": -np.linspace(2e6, 1e6, len(strikes)), "dex": np.zeros(len(strikes)), "oi_call": np.full(len(strikes), 1000), "oi_put": np.full(len(strikes), 1200), }, index=strikes) agg["net_gex"] = agg["gex_call"] + agg["gex_put"] grid = np.arange(85.0, 116.0, 0.5) gp = np.linspace(-3e6, 3e6, len(grid)) dp = np.linspace(-1e9, 1e9, len(grid)) levels = compute_levels(agg, grid, gp, spot, 2.0, cfg) captured = {} orig = plt.Figure.savefig def spy(self, *a, **k): captured["fig"] = self return orig(self, *a, **k) plt.Figure.savefig = spy try: render_chart("TEST", agg, grid, gp, grid, dp, levels, "2026-07-24 15:44:00", cfg, str(tmp_path), "pm", 2.0) finally: plt.Figure.savefig = orig fig = captured["fig"] ax = fig.axes[0] assert ax.get_xscale() == "linear", f"xscale should be 'linear', got {ax.get_xscale()}" # --------------------------------------------------------------------------- # 7. FIX 20: layout guard clear # --------------------------------------------------------------------------- def test_layout_guard_clear(tmp_path, caplog): """Render the SMH fixture; assert the layout guard logs no warning.""" import logging import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt from gex.plot import render_chart contracts, spot, ts_str = _load_smh_fixture() cfg = GexConfig(outdir=str(tmp_path)) snap_et = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S").replace( tzinfo=ZoneInfo(cfg.source_timestamp_tz)).astimezone(_ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) df_band = filter_contracts_band(df_full, spot, cfg) agg = aggregate(df_band, spot, cfg) increment = detect_increment(agg.index.to_numpy()) step = max(increment / 5.0, 0.10) grid = np.arange(spot * (1 - cfg.profile_band), spot * (1 + cfg.profile_band) + step, step) gp, _ = gex_profile(df_full, spot, cfg, grid) dp, _ = dex_profile(df_full, cfg, grid, spot) levels = compute_levels(agg, grid, gp, spot, increment, cfg) # add put_heavy_note to trigger the note rendering path levels["put_heavy_note"] = "put/call OI 4.84 — see per-expiry breakdown" with caplog.at_level(logging.WARNING, logger="gex.plot"): render_chart("SMH", agg, grid, gp, grid, dp, levels, ts_str, cfg, str(tmp_path), "pm", increment) layout_warnings = [r for r in caplog.records if "LAYOUT" in r.getMessage()] assert len(layout_warnings) == 0, \ f"Layout guard logged {len(layout_warnings)} warning(s): {[r.getMessage() for r in layout_warnings]}" # =========================================================================== # NEW TESTS (v1.4.0 audit fixes) # =========================================================================== # --------------------------------------------------------------------------- # 1. FIX 22: gamma_condition matches sign of profile at spot # --------------------------------------------------------------------------- def test_gamma_condition_matches_sign(): """sign(profile_at_spot) must always agree with gamma_condition, on both the all-expirations and exfront fixtures.""" contracts, spot, ts_str = _load_smh_fixture() cfg = GexConfig() snap_et = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S").replace( tzinfo=ZoneInfo(cfg.source_timestamp_tz)).astimezone(_ET) for exclude_front in (False, True): df_full = filter_contracts_full(contracts, spot, cfg, snap_et) if exclude_front: exps = sorted(df_full["expiry"].unique()) if len(exps) > 1: df_full = df_full[df_full["expiry"] != exps[0]] df_band = filter_contracts_band(df_full, spot, cfg) agg = aggregate(df_band, spot, cfg) increment = detect_increment(agg.index.to_numpy()) step = max(increment / 5.0, 0.10) grid = np.arange(spot * (1 - cfg.profile_band), spot * (1 + cfg.profile_band) + step, step) gp, _ = gex_profile(df_full, spot, cfg, grid) # interpolate profile at spot profile_at_spot = float(np.interp(spot, grid, gp)) levels = compute_levels(agg, grid, gp, spot, increment, cfg, profile_at_spot=profile_at_spot) expected = "POSITIVE" if profile_at_spot > 0 else "NEGATIVE" assert levels["gamma_condition"] == expected, \ f"gamma_condition={levels['gamma_condition']} but profile_at_spot={profile_at_spot:.2e} " \ f"(exclude_front={exclude_front})" assert levels["gamma_condition_basis"] == "sign of simulated GEX profile at spot" assert levels["net_gex_at_spot"] == profile_at_spot # --------------------------------------------------------------------------- # 2. FIX 23: HVL indeterminate when spread > 10% # --------------------------------------------------------------------------- def test_hvl_single_level_with_distance(): """FIX 29: HVL is always a single defined level (zero crossing nearest spot). No 'indeterminate'. Distance is information, not a defect. On the SMH fixture the zero crossing is ~629.8, far from spot (557.09) -> regime note says 'far from spot'.""" contracts, spot, ts_str = _load_smh_fixture() cfg = GexConfig() snap_et = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S").replace( tzinfo=ZoneInfo(cfg.source_timestamp_tz)).astimezone(_ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) df_band = filter_contracts_band(df_full, spot, cfg) agg = aggregate(df_band, spot, cfg) increment = detect_increment(agg.index.to_numpy()) step = max(increment / 5.0, 0.10) grid = np.arange(spot * (1 - cfg.profile_band), spot * (1 + cfg.profile_band) + step, step) gp, _ = gex_profile(df_full, spot, cfg, grid) levels = compute_levels(agg, grid, gp, spot, increment, cfg) # HVL is a single number, not None, not indeterminate assert levels["hvl"] is not None, "HVL must be a single defined level" assert levels["hvl_status"] == "ok" assert levels["hvl_rule"] == "zero_cross" # distance annotation present assert levels["hvl_distance_pct"] is not None assert levels["hvl_regime_note"] is not None # on this fixture the crossing is far from spot (>8%) assert abs(levels["hvl_distance_pct"]) > 0.08 assert "far from spot" in levels["hvl_regime_note"] # no indeterminate artifacts assert levels["hvl_spread_pct"] == 0.0 assert levels["hvl_range"] is None # --------------------------------------------------------------------------- # 3. FIX 24: ATM IV liquidity gate # --------------------------------------------------------------------------- def test_atm_iv_liquidity_gate(): """Expiry with OI < max(5000, 2% chain OI) must be rejected.""" cfg = GexConfig() spot = 100.0 today = date(2026, 7, 24) # one expiry with tiny OI (100 contracts total) -> fails liquidity gate chain = [ {"strike": 100.0, "expiry": date(2026, 8, 15), "cp": "C", "iv": 0.25, "oi": 50, "volume": 0, "delta": 0.5, "gamma": 0.03, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, {"strike": 100.0, "expiry": date(2026, 8, 15), "cp": "P", "iv": 0.25, "oi": 50, "volume": 0, "delta": -0.5, "gamma": 0.03, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, ] df = filter_contracts_full(chain, spot, cfg, today) result = atm_expected_move(df, spot) assert result["atm_iv"] is None, f"atm_iv should be None, got {result['atm_iv']}" assert "rejected" in result["atm_iv_status"], \ f"status should contain 'rejected', got {result['atm_iv_status']}" # --------------------------------------------------------------------------- # 4. FIX 25: max_expiry_share fires warning, not front_expiry_share # --------------------------------------------------------------------------- def test_max_expiry_share(): """max_expiry_share keys on the argmax expiry, not the front one.""" cfg = GexConfig() spot = 100.0 today = date(2026, 7, 24) # front expiry (0DTE): small GEX; later expiry: dominant GEX chain = [ # front expiry: tiny {"strike": 100.0, "expiry": date(2026, 7, 24), "cp": "C", "iv": 0.30, "oi": 100, "volume": 0, "delta": 0.5, "gamma": 0.05, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, # later expiry: dominant {"strike": 100.0, "expiry": date(2026, 8, 15), "cp": "C", "iv": 0.25, "oi": 100000, "volume": 0, "delta": 0.5, "gamma": 0.03, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, ] df_full = filter_contracts_full(chain, spot, cfg, today) result = build_gex_by_expiry(df_full, spot, cfg) # max_expiry should be the later one (08-15), not the front (07-24) assert result["max_expiry"] == "2026-08-15" assert result["max_expiry_share"] > 0.90, \ f"max_expiry_share {result['max_expiry_share']:.3f} should be > 0.90" # front_expiry_share is separate and small assert result["front_expiry_share"] < 0.10 # top3_expiry_share >= max_expiry_share assert result["top3_expiry_share"] >= result["max_expiry_share"] # --------------------------------------------------------------------------- # 5. FIX 26: interior_extremum detects grid boundary # --------------------------------------------------------------------------- def test_interior_extremum_boundary(): """argmin at the grid edge -> (None, 'at_grid_boundary').""" grid = np.array([80.0, 90.0, 100.0, 110.0, 120.0]) # min at index 0 (edge) dp_edge = np.array([-5e9, -3e9, -1e9, 1e9, 3e9]) price, status = dex_min_price(grid, dp_edge) assert price is None assert status == "at_grid_boundary" # min at index 1 (within edge_tol=2 of edge) -> also boundary dp_near = np.array([-3e9, -5e9, -1e9, 1e9, 3e9]) price2, status2 = dex_min_price(grid, dp_near) assert price2 is None assert status2 == "at_grid_boundary" # min at index 2 (interior) -> returns price dp_int = np.array([-1e9, -3e9, -5e9, -3e9, -1e9]) price3, status3 = dex_min_price(grid, dp_int) assert price3 == 100.0 assert status3 == "interior" # --------------------------------------------------------------------------- # 6. FIX 27: ephemeral level detection # --------------------------------------------------------------------------- def test_ephemeral_levels(): """level_front_expiry_pct returns >0.50 for a level dominated by 0DTE OI.""" from gex.compute import level_front_expiry_pct cfg = GexConfig() spot = 100.0 today = date(2026, 7, 24) # put support at 95: 90% from 0DTE, 10% from later expiry chain = [ # 0DTE: dominant put OI at 95 {"strike": 95.0, "expiry": date(2026, 7, 24), "cp": "P", "iv": 0.30, "oi": 90000, "volume": 0, "delta": -0.5, "gamma": 0.05, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, # later expiry: small put OI at 95 {"strike": 95.0, "expiry": date(2026, 8, 15), "cp": "P", "iv": 0.25, "oi": 10000, "volume": 0, "delta": -0.5, "gamma": 0.03, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, # call resistance at 105: mostly later expiry (not ephemeral) {"strike": 105.0, "expiry": date(2026, 7, 24), "cp": "C", "iv": 0.30, "oi": 1000, "volume": 0, "delta": 0.5, "gamma": 0.05, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, {"strike": 105.0, "expiry": date(2026, 8, 15), "cp": "C", "iv": 0.25, "oi": 9000, "volume": 0, "delta": 0.5, "gamma": 0.03, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, ] df_full = filter_contracts_full(chain, spot, cfg, today) # put support at 95: 90% front expiry -> ephemeral fe_95 = level_front_expiry_pct(df_full, 95.0, spot, cfg) assert fe_95["front_expiry_abs_share"] > 0.50, f"put_support at 95 should be >50% front expiry, got {fe_95['front_expiry_abs_share']:.2%}" assert fe_95["front_expiry"] == "2026-07-24" # call resistance at 105: 10% front expiry -> NOT ephemeral fe_105 = level_front_expiry_pct(df_full, 105.0, spot, cfg) assert fe_105["front_expiry_abs_share"] < 0.50, f"call_resistance at 105 should be <50% front expiry, got {fe_105['front_expiry_abs_share']:.2%}" # =========================================================================== # NEW TESTS (v1.5.0 audit fixes) # =========================================================================== # --------------------------------------------------------------------------- # 1. FIX 28: no published level sits at a search/mask boundary # --------------------------------------------------------------------------- def test_no_level_at_search_boundary(): """FIX 28 regression: no published level may equal a mask/search boundary to within one grid step. The old inflection rule walked to the mask boundary (550 + 3.75 = 553.8175 exactly). This test asserts that never happens again.""" contracts, spot, ts_str = _load_smh_fixture() cfg = GexConfig() snap_et = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S").replace( tzinfo=ZoneInfo(cfg.source_timestamp_tz)).astimezone(_ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) df_band = filter_contracts_band(df_full, spot, cfg) agg = aggregate(df_band, spot, cfg) increment = detect_increment(agg.index.to_numpy()) step = max(increment / 5.0, 0.10) grid = np.arange(spot * (1 - cfg.profile_band), spot * (1 + cfg.profile_band) + step, step) gp, _ = gex_profile(df_full, spot, cfg, grid) levels = compute_levels(agg, grid, gp, spot, increment, cfg) # the old mask boundary was at 553.8175 (= 550 + 1.5*2.5). Check no published # level is within one grid step of that value. old_mask_boundary = 553.8175 published_levels = [ levels["call_resistance"], levels["put_support"], levels["hvl"], levels.get("gex_transition"), levels.get("delta_neutral"), ] for lvl in published_levels: if lvl is None: continue dist = abs(lvl - old_mask_boundary) assert dist > step, \ f"Level {lvl} is within one grid step ({step}) of the old mask boundary " \ f"{old_mask_boundary} (dist={dist:.4f}). This is the FIX 28 regression." # --------------------------------------------------------------------------- # 2. FIX 29: HVL distance annotation and regime note # --------------------------------------------------------------------------- def test_hvl_distance_and_regime_note(): """FIX 29: hvl_distance_pct and hvl_regime_note are published. The note buckets are: |d|<=3% -> 'near spot', 3-8% -> 'moderately distant', >8% -> 'far from spot'.""" from gex.compute import compute_hvl cfg = GexConfig() spot = 100.0 grid = np.arange(80.0, 120.5, 0.5) # crossing at 101 (1% from spot) -> "near spot" profile_near = (grid - 101.0) * 1e6 info = compute_hvl(grid, profile_near, spot, cfg) assert info["hvl"] is not None assert abs(info["hvl_distance_pct"]) <= 0.03 assert "near spot" in info["hvl_regime_note"] # crossing at 106 (6% from spot) -> "moderately distant" profile_mod = (grid - 106.0) * 1e6 info2 = compute_hvl(grid, profile_mod, spot, cfg) assert 0.03 < abs(info2["hvl_distance_pct"]) <= 0.08 assert "moderately distant" in info2["hvl_regime_note"] # crossing at 115 (15% from spot) -> "far from spot" profile_far = (grid - 115.0) * 1e6 info3 = compute_hvl(grid, profile_far, spot, cfg) assert abs(info3["hvl_distance_pct"]) > 0.08 assert "far from spot" in info3["hvl_regime_note"] # --------------------------------------------------------------------------- # 3. FIX 30: gex_transition persistence check # --------------------------------------------------------------------------- def test_gex_transition_persistence(): """FIX 30: gex_transition requires 3 consecutive strikes negative before and positive after the flip. A noisy single-strike flip must be rejected.""" from gex.compute import compute_gex_transition spot = 100.0 # persistent flip: 4 negative strikes then 4 positive strikes strikes_p = np.array([90.0, 92.0, 94.0, 96.0, 98.0, 100.0, 102.0, 104.0]) net_p = np.array([-5e6, -4e6, -3e6, -2e6, -1e6, 1e6, 2e6, 3e6]) agg_p = pd.DataFrame({"net_gex": net_p}, index=strikes_p) result_p = compute_gex_transition(agg_p, spot, persistence=3) assert result_p["gex_transition"] is not None assert result_p["gex_transition_status"] == "ok" assert result_p["gex_transition_distance_pct"] is not None # noisy flip: only 1 negative strike before the flip -> rejected strikes_n = np.array([90.0, 92.0, 94.0, 96.0, 98.0, 100.0, 102.0, 104.0]) net_n = np.array([1e6, 2e6, 3e6, 2e6, -1e6, 1e6, 2e6, 3e6]) agg_n = pd.DataFrame({"net_gex": net_n}, index=strikes_n) result_n = compute_gex_transition(agg_n, spot, persistence=3) assert result_n["gex_transition"] is None assert result_n["gex_transition_status"] == "no_persistent_flip" # --------------------------------------------------------------------------- # 4. FIX 31: spread detection finds the 520/500 and 522.5/517.5 pairs # --------------------------------------------------------------------------- def test_spread_detection(): """FIX 31: detect_spread_candidates identifies the 2026-07-31 put spread structures (520/500 and 522.5/517.5) in the SMH fixture.""" from gex.compute import detect_spread_candidates contracts, spot, ts_str = _load_smh_fixture() cfg = GexConfig() snap_et = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S").replace( tzinfo=ZoneInfo(cfg.source_timestamp_tz)).astimezone(_ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) df_band = filter_contracts_band(df_full, spot, cfg) agg = aggregate(df_band, spot, cfg) increment = detect_increment(agg.index.to_numpy()) result = detect_spread_candidates(df_full, spot, cfg, increment) candidates = result["spread_candidates"] assert len(candidates) > 0, "should find at least one spread candidate" # the 520/500 pair must be present pair_520_500 = [c for c in candidates if c["strike_low"] == 500.0 and c["strike_high"] == 520.0 and c["right"] == "P" and "2026-07-31" in c["expiry"]] assert len(pair_520_500) == 1, \ f"520/500 put pair not found. Candidates: {[(c['strike_low'], c['strike_high']) for c in candidates]}" # the 522.5/517.5 pair must be present pair_522_517 = [c for c in candidates if c["strike_low"] == 517.5 and c["strike_high"] == 522.5 and c["right"] == "P" and "2026-07-31" in c["expiry"]] assert len(pair_522_517) == 1, \ f"522.5/517.5 put pair not found. Candidates: {[(c['strike_low'], c['strike_high']) for c in candidates]}" # flagged share > 0.20 -> sensitivity must be present assert result["spread_flagged_share"] > 0.20, \ f"flagged share {result['spread_flagged_share']:.4f} should be > 0.20" sens = result["sensitivity_smaller_leg_sign_flipped"] assert sens is not None, "sensitivity_smaller_leg_sign_flipped must be present when flagged" assert "total_net_gex" in sens assert "hvl" in sens assert "illustrative" in sens["note"].lower() # --------------------------------------------------------------------------- # 5. FIX 32: realised vol and IV cross-check # --------------------------------------------------------------------------- def test_realised_vol_and_iv_cross_check(): """FIX 32: compute_realised_vol returns annualised vol from closes; atm_iv_cross_check publishes iv_hv_ratio and vol_regime.""" from gex.compute import compute_realised_vol, atm_iv_cross_check # synthetic closes: 21 daily prices with ~1% daily moves np.random.seed(42) closes = [100.0] for _ in range(20): closes.append(closes[-1] * (1 + np.random.normal(0, 0.01))) rv = compute_realised_vol(closes, days=20) assert rv is not None assert 0.05 < rv < 0.50, f"realised vol {rv:.4f} outside plausible range" # iv_hv_ratio with atm_iv = 0.30 cfg = GexConfig() xc = atm_iv_cross_check(0.30, rv, cfg) assert xc["iv_hv_ratio"] is not None assert xc["vol_regime"] in ("IV > HV", "IV < HV") assert xc["iv_hv_outlier"] is False # ratio ~1-2, not an outlier # outlier case: atm_iv = 2.0 with rv = 0.15 -> ratio > 2.5 xc2 = atm_iv_cross_check(2.0, 0.15, cfg) assert xc2["iv_hv_ratio"] > 2.5 assert xc2["iv_hv_outlier"] is True # None inputs -> graceful xc3 = atm_iv_cross_check(None, rv, cfg) assert xc3["iv_hv_ratio"] is None xc4 = atm_iv_cross_check(0.30, None, cfg) assert xc4["iv_hv_ratio"] is None # =========================================================================== # NEW TESTS (v1.6.0) # =========================================================================== # --------------------------------------------------------------------------- # 1. FIX 34: test_add_ticker_no_code_change # --------------------------------------------------------------------------- def test_add_ticker_no_code_change(tmp_path): """Appending a symbol to cfg.tickers must run end-to-end with no other module modified. Uses a mocked chain (no network).""" from gex.snapshot import _process_and_render from gex.compute import filter_contracts_full cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) # append a new ticker to the list cfg.tickers = ["NDX", "SPY", "SMH", "QQQ"] assert "QQQ" in cfg.tickers # build a minimal synthetic chain for QQQ spot = 450.0 today = date(2026, 7, 24) exp = date(2026, 8, 15) chain = [] for strike in np.arange(400, 501, 5.0): for cp in ("C", "P"): chain.append({ "strike": float(strike), "expiry": exp, "cp": cp, "iv": 0.25, "oi": 5000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) df_full = filter_contracts_full(chain, spot, cfg, today) assert not df_full.empty # run the pipeline — must not raise png = _process_and_render("QQQ", cfg, "pm", "test_snap", "2026-07-24 15:59:00", spot, df_full, suffix="", endpoint_variant="plain", instrument_class="equity_etf") assert png.exists() json_path = png.with_suffix(".json") assert json_path.exists() with open(json_path) as f: data = json.load(f) assert data["symbol"] == "QQQ" assert data["endpoint_variant"] == "plain" assert data["instrument_class"] == "equity_etf" assert "bands" in data # --------------------------------------------------------------------------- # 2. FIX 35: test_bands_scale_with_iv # --------------------------------------------------------------------------- def test_bands_scale_with_iv(): """A 20% IV fixture and a 60% IV fixture at the same spot produce profile_band differing by roughly 3x, both inside band_limits.""" from gex.compute import compute_bands cfg = GexConfig() bands_20 = compute_bands(0.20, cfg) bands_60 = compute_bands(0.60, cfg) # sigma_30d scales linearly with IV assert bands_60["sigma_30d"] > bands_20["sigma_30d"] ratio = bands_60["sigma_30d"] / bands_20["sigma_30d"] assert abs(ratio - 3.0) < 0.1, f"sigma ratio {ratio:.2f} should be ~3.0" # profile_band scales ~3x (both within clip limits) pb_ratio = bands_60["profile_band"] / bands_20["profile_band"] assert 2.5 < pb_ratio < 3.5, f"profile_band ratio {pb_ratio:.2f} should be ~3x" # both within band_limits lo, hi = cfg.band_limits["profile_band"] assert lo <= bands_20["profile_band"] <= hi assert lo <= bands_60["profile_band"] <= hi # band_basis is "atm_iv" for both assert bands_20["band_basis"] == "atm_iv" assert bands_60["band_basis"] == "atm_iv" # fallback case bands_fb = compute_bands(None, cfg) assert bands_fb["band_basis"] == "fallback" assert bands_fb["atm_iv_used"] == cfg.atm_iv_fallback # --------------------------------------------------------------------------- # 3. FIX 36: test_grid_fixed_count # --------------------------------------------------------------------------- def test_grid_fixed_count(): """Grid length == profile_grid_points for spot 500 and spot 25000.""" cfg = GexConfig() n = cfg.profile_grid_points for spot in (500.0, 25000.0): grid = np.linspace(spot * (1 - cfg.profile_band), spot * (1 + cfg.profile_band), n) assert len(grid) == n, f"grid length {len(grid)} != {n} for spot={spot}" # resolution check: spot*2*band/n (allow float tolerance) expected_res = spot * 2 * cfg.profile_band / n actual_res = grid[1] - grid[0] assert abs(actual_res - expected_res) / expected_res < 0.01, \ f"resolution {actual_res:.6f} vs expected {expected_res:.6f}" # DEX grid also fixed count grid_dex = np.linspace(500 * (1 - cfg.profile_band_dex), 500 * (1 + cfg.profile_band_dex), n) assert len(grid_dex) == n # =========================================================================== # NEW TESTS (v1.6.1 five-fix) # =========================================================================== # --------------------------------------------------------------------------- # 1. FIX 43: test_oi_gates_scale_with_chain # --------------------------------------------------------------------------- def _bracketing_chain(spot, total_oi, base_expiry): """Two strikes bracketing spot on one DTE>=5 expiry, OI split so both the expiry-level and per-contract chain-relative gates pass.""" per_strike = total_oi / 4.0 # 4 contracts (2 strikes x C/P) rows = [] for strike in (spot - 5.0, spot + 5.0): for cp in ("C", "P"): rows.append({ "strike": strike, "expiry": base_expiry, "cp": cp, "iv": 0.25, "oi": per_strike, "volume": 0, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.04, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, "dte": 10, }) return pd.DataFrame(rows) def test_oi_gates_scale_with_chain(): """An 87k-OI chain and a 2M-OI chain BOTH resolve atm_iv; the small chain's per-contract floor must be below 50 (proving the gates are relative, not the old absolute 250).""" spot = 100.0 base_expiry = date(2026, 8, 15) small = _bracketing_chain(spot, 87_000, base_expiry) large = _bracketing_chain(spot, 2_000_000, base_expiry) res_small = atm_expected_move(small, spot) res_large = atm_expected_move(large, spot) assert res_small["atm_iv"] is not None, \ f"small chain should resolve atm_iv, got status={res_small['atm_iv_status']}" assert res_large["atm_iv"] is not None, \ f"large chain should resolve atm_iv, got status={res_large['atm_iv_status']}" # chain-relative per-contract floor for the small chain must be below 50 small_contract_floor = max(0.0002 * 87_000, 10.0) # = 17.4 assert small_contract_floor < 50, \ f"small-chain contract floor {small_contract_floor} should be < 50" # --------------------------------------------------------------------------- # 2. FIX 44: test_render_spacing # --------------------------------------------------------------------------- def test_render_spacing(): """Strikes listed every 25 but populated every 100 yield render_spacing == 100 while strike_increment stays 25.""" # listed grid: every 25 from 100..500 listed = np.arange(100.0, 500.0 + 1, 25.0) # populated (non-trivial |net_gex|) only every 100 net_gex = np.where(np.isclose(listed % 100.0, 0.0), 1000.0, 0.0) inc = detect_increment(listed) rs = detect_render_spacing(listed, net_gex, inc) assert inc == 25.0, f"strike_increment should stay 25, got {inc}" assert rs == 100.0, f"render_spacing should be 100, got {rs}" # --------------------------------------------------------------------------- # FIX 60 / FIX 69: test_render_bucket # --------------------------------------------------------------------------- def test_render_bucket(): """FIX 69: bucket = ladder value CLOSEST to raw = window_span / 40, then clamped to at least strike_increment. SMH (span ~98, increment 2.5) -> raw 2.45 -> closest 2.5 -> max(2.5, 2.5) = 2.5. NDX (span ~2310, increment 10) -> raw 57.75 -> closest 50 -> max(50, 10) = 50. SPY (span ~16, increment 1) -> raw 0.4 -> closest 0.5 -> max(0.5, 1) = 1.0. """ assert render_bucket(98.0, 2.5) == 2.5 # SMH: raw 2.45 -> 2.5 assert render_bucket(2310.0, 10.0) == 50.0 # NDX: raw 57.75 -> 50 assert render_bucket(16.0, 1.0) == 1.0 # SPY: raw 0.4 -> 0.5 -> clamped to 1.0 # bucket never drops below strike_increment assert render_bucket(5.0, 10.0) == 10.0 # raw 0.125 -> 0.5 -> clamped to 10 # bar count lands near target for the two real cases assert 25 <= 2310.0 / render_bucket(2310.0, 10.0) <= 50 assert 30 <= 98.0 / render_bucket(98.0, 2.5) <= 50 # --------------------------------------------------------------------------- # FIX 63 / FIX 64: reconcile_by_expiry ranks by dollar gap + unsigned rel_err # --------------------------------------------------------------------------- def test_reconcile_by_expiry_dollar_rank_and_unsigned(): """FIX 63: worst expiries rank by abs dollar gap (not rel_err), each with a gap_share of the total dollar gap. FIX 64: rel_err_unsigned is published.""" cfg = GexConfig() spot = 100.0 # Two expiries. Expiry A: large reported GEX, recomputed matches closely # (small rel_err but the absolute gap can still be big). Expiry B: tiny # reported GEX with a huge rel_err but a small absolute gap. Dollar ranking # must surface A first even though B has the worse rel_err. rows = [] # expiry A: one call, reported gamma 0.05, OI 10000 -> large GEX rows.append({"strike": 100.0, "expiry": date(2026, 8, 15), "cp": "C", "iv": 0.20, "oi": 10000, "gamma": 0.05, "T": 20 / 252.0}) # expiry B: one call, reported gamma 0.05, OI 10 -> tiny GEX rows.append({"strike": 100.0, "expiry": date(2026, 8, 22), "cp": "C", "iv": 0.20, "oi": 10, "gamma": 0.05, "T": 25 / 252.0}) df = pd.DataFrame(rows) out = reconcile_by_expiry(df, spot, cfg) worst = out["reconciliation_worst_expiries"] assert len(worst) == 2 # ranked by dollar gap descending -> the big-OI expiry A first assert worst[0]["dollar_gap"] >= worst[1]["dollar_gap"] assert worst[0]["expiry"] == "2026-08-15" # gap_share sums to 1 across all entries assert abs(sum(r["gap_share"] for r in worst) - 1.0) < 1e-9 # each entry carries rel_err AND dollar_gap for r in worst: assert "rel_err" in r and "dollar_gap" in r and "gap_share" in r # FIX 64: unsigned rel_err published and non-negative assert out["rel_err_unsigned"] is not None assert out["rel_err_unsigned"] >= 0.0 # --------------------------------------------------------------------------- # FIX 66: gamma_precision # --------------------------------------------------------------------------- def test_gamma_precision(): """FIX 66 / FIX 70: detect Cboe's 4dp gamma publication precision. NDX-style (gamma ~0.0001): 1 sig fig -> coarse, low_precision=True. SMH-style (gamma ~0.0012): 2 sig figs -> adequate, low_precision=False. SPY-style (gamma ~0.0123): 3 sig figs -> high, low_precision=False.""" from gex.compute import gamma_precision # NDX: all strikes report exactly 0.0001 (1 sig fig at 4dp) -> coarse df_ndx = pd.DataFrame({"gamma": [0.0001, 0.0001, 0.0001, 0.0002, 0.0001]}) sig_figs, low, label = gamma_precision(df_ndx) assert sig_figs == 1, f"NDX gamma should be 1 sig fig, got {sig_figs}" assert low is True, "NDX should be flagged low precision" assert label == "coarse" # SMH: gamma ~0.0012 (2 sig figs) -> adequate, bars stay reliable df_smh = pd.DataFrame({"gamma": [0.0012, 0.0011, 0.0013, 0.0012]}) sig_figs2, low2, label2 = gamma_precision(df_smh) assert sig_figs2 == 2, f"SMH gamma should be 2 sig figs, got {sig_figs2}" assert low2 is False, "SMH (adequate) should NOT flip bars_reliable" assert label2 == "adequate" # SPY: gamma ~0.0123 (3 sig figs at 4dp) -> high df_spy = pd.DataFrame({"gamma": [0.0123, 0.0118, 0.0131, 0.0120]}) sig_figs3, low3, label3 = gamma_precision(df_spy) assert sig_figs3 >= 3, f"SPY gamma should be >= 3 sig figs, got {sig_figs3}" assert low3 is False assert label3 == "high" # empty / all-zero: default to high, not low precision df_empty = pd.DataFrame({"gamma": []}) sig_figs4, low4, label4 = gamma_precision(df_empty) assert sig_figs4 == 4 and low4 is False and label4 == "high" # --------------------------------------------------------------------------- # FIX 67: compute_bands band_basis = atm_iv_single_strike # --------------------------------------------------------------------------- def test_compute_bands_single_strike_basis(): """FIX 67: when atm_iv_status='single_strike_no_interpolation', band_basis must be 'atm_iv_single_strike', not 'atm_iv'.""" from gex.compute import compute_bands cfg = GexConfig() # normal two-strike interpolation bands_ok = compute_bands(0.25, cfg, atm_iv_status="ok") assert bands_ok["band_basis"] == "atm_iv" # single-strike (no interpolation) bands_ss = compute_bands(0.25, cfg, atm_iv_status="single_strike_no_interpolation") assert bands_ss["band_basis"] == "atm_iv_single_strike", \ f"expected atm_iv_single_strike, got {bands_ss['band_basis']}" # same numeric bands (the value is the same; only the label differs) assert abs(bands_ok["sigma_30d"] - bands_ss["sigma_30d"]) < 1e-12 # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # FIX 97: the ex-front variant was removed. The old FIX 68 test # (test_exfront_identical_keys) that guarded the exfront path's key set against # drift from the main pipeline is RETIRED — with a single pipeline there is no # second path to drift. The canonical-schema guarantee is now covered by # test_canonical_schema_enumerates_index_json over every published entry. # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # FIX 78: zero-greek dropped-contract count (single pipeline, FIX 97) # --------------------------------------------------------------------------- def test_zero_greek_count_dropped(monkeypatch): """FIX 78 / FIX 97: zero_greek_contracts_dropped and zero_greek_contracts_dropped_full_chain are published on the single all-expirations pipeline. With the ex-front variant removed (FIX 97) the two counts are equal — both describe the full chain. The fields remain in the canonical schema so downstream consumers keep working.""" import matplotlib matplotlib.use("Agg") from gex.snapshot import _process_and_render import gex.snapshot as snap_mod import json, os, tempfile from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo monkeypatch.setattr(snap_mod, "_load_recent_closes", lambda symbol, cfg: [99.0, 100.0, 101.0, 100.5, 100.2, 100.8]) cfg = GexConfig() spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) later = today + timedelta(days=7) # Two expiries; 3 contracts with gamma=0 (rule 4) across the full chain. contracts = [] for i, exp in enumerate((today, later)): for j, k in enumerate((95.0, 100.0, 105.0)): for cp in ("C", "P"): zero_gamma = (i == 0 and j < 2) or (i == 1 and j == 0 and cp == "C") contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 5000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.0 if zero_gamma else 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 15, 35, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) assert not df_full.empty zero_greek = sum(1 for c in contracts if c["oi"] > 0 and (c["iv"] <= 0 or c["gamma"] == 0)) ts_str = "2026-07-24 19:35:29" snap_id = "test_fix78" with tempfile.TemporaryDirectory() as tmpdir: cfg_tmp = GexConfig(outdir=tmpdir) _process_and_render("TEST", cfg_tmp, "am", snap_id, ts_str, spot, df_full, suffix="", instrument_class="equity_etf", zero_greek_contracts=zero_greek, zero_greek_full_chain=zero_greek) main_json = os.path.join(tmpdir, "TEST", "2026-07-24_am.json") with open(main_json) as f: main = json.load(f) # Both fields present (canonical schema) assert "zero_greek_contracts_dropped" in main assert "zero_greek_contracts_dropped_full_chain" in main # Single pipeline: variant count == full-chain count assert main["zero_greek_contracts_dropped"] == zero_greek assert main["zero_greek_contracts_dropped_full_chain"] == zero_greek # --------------------------------------------------------------------------- # FIX 74: canonical schema — identical key set across instrument classes # --------------------------------------------------------------------------- def test_canonical_schema_across_instrument_classes(): """FIX 74: every output file must match ONE canonical key set regardless of instrument class. Instrument-class-specific fields (front_expiry_am_settled) are emitted as null for non-index tickers rather than omitted, so an index run (NDX) and an equity/ETF run (SMH/SPY) produce the exact same JSON keys.""" from gex.snapshot import _process_and_render import json, tempfile, os from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo cfg = GexConfig() spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) later = today + timedelta(days=7) contracts = [] for exp in (today, later): for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 8000 if cp == "P" else 3000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 15, 35, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) ts_str = "2026-07-24 19:35:29" with tempfile.TemporaryDirectory() as tmpdir: cfg_tmp = GexConfig(outdir=tmpdir) # index instrument (NDX-style) _process_and_render("IDX", cfg_tmp, "am", "test_fix74", ts_str, spot, df_full, suffix="", instrument_class="index") # equity/ETF instrument (SMH/SPY-style) _process_and_render("ETF", cfg_tmp, "am", "test_fix74", ts_str, spot, df_full, suffix="", instrument_class="equity_etf") with open(os.path.join(tmpdir, "IDX", "2026-07-24_am.json")) as f: idx = json.load(f) with open(os.path.join(tmpdir, "ETF", "2026-07-24_am.json")) as f: etf = json.load(f) idx_keys = set(idx.keys()) etf_keys = set(etf.keys()) missing = idx_keys - etf_keys extra = etf_keys - idx_keys assert not missing, f"ETF MISSING keys vs index: {sorted(missing)}" assert not extra, f"ETF has EXTRA keys vs index: {sorted(extra)}" # the instrument-class field is present in BOTH, null for the ETF assert "front_expiry_am_settled" in idx assert "front_expiry_am_settled" in etf assert etf["front_expiry_am_settled"] is None assert isinstance(idx["front_expiry_am_settled"], bool) # atm_iv_status is present in both (null unless an outlier) assert "atm_iv_status" in idx and "atm_iv_status" in etf # --------------------------------------------------------------------------- # FIX 86: canonical schema over EVERY entry enumerated from index.json # --------------------------------------------------------------------------- def test_canonical_schema_enumerates_index_json(): """FIX 86: the canonical-schema guarantee must hold over EVERY snapshot that index.json advertises, not just a fixed six-file sample. index.json lists all published snapshots (every ticker/date/slot); the test enumerates each entry, loads its JSON, and asserts they all share ONE identical key set. Before FIX 86 the mislabeled 2026-07-27 AM files (v1.7.5, no reconciliation_scope / bands.atm_iv_source_expiry) were listed in index.json but not covered by the six-file sample, so the drift went undetected. Runs against the live public gex_out (GEX_OUT env var, defaulting to the VPS public path). Skips when index.json is absent (e.g. a fresh checkout).""" import json, os base = os.environ.get("GEX_OUT", "/home/allofthesewords/public_html/gex_out") idx_path = os.path.join(base, "index.json") if not os.path.exists(idx_path): import pytest pytest.skip("no index.json at %s (set GEX_OUT to the public gex_out)" % base) with open(idx_path) as f: idx = json.load(f) # enumerate every (ticker, date, slot) entry entries = [] for ticker, dates in idx.get("snapshots", {}).items(): for date_str, slots in dates.items(): for slot in slots: entries.append((ticker, date_str, slot)) assert entries, "index.json lists no snapshots" keysets = {} for ticker, date_str, slot in entries: p = os.path.join(base, ticker, "%s_%s.json" % (date_str, slot)) assert os.path.exists(p), "index.json lists %s but file is missing: %s" % ( "%s/%s_%s" % (ticker, date_str, slot), p) with open(p) as f: d = json.load(f) keysets["%s/%s_%s" % (ticker, date_str, slot)] = set(d.keys()) ref_name, ref = next(iter(keysets.items())) for name, ks in keysets.items(): missing = ref - ks extra = ks - ref assert not missing and not extra, ( "canonical schema drift: %s vs %s — missing %s, extra %s" % ( name, ref_name, sorted(missing), sorted(extra))) # --------------------------------------------------------------------------- # FIX 75: re-rendering a snapshot is byte-identical apart from render_lag_min # --------------------------------------------------------------------------- def test_rerender_byte_identical_except_render_lag(monkeypatch): """FIX 75: source_timestamp_age_min is frozen at capture, so re-rendering the SAME snapshot (same captured_at_utc) produces a byte-identical JSON apart from the explicitly whitelisted render_lag_min (the wall-clock delta since capture, which legitimately advances between renders). Before FIX 75 the age was computed at render time, so two renders of identical data reported different freshness.""" from gex.snapshot import _process_and_render import gex.snapshot as snap_mod import json, tempfile, os from datetime import date, datetime, timedelta, timezone from zoneinfo import ZoneInfo # determinism: pin realised-vol input so neither the Yahoo network call nor the # outdir feedback loop (which reads prior JSON) can vary between the two renders. monkeypatch.setattr(snap_mod, "_load_recent_closes", lambda symbol, cfg: [99.0, 100.0, 101.0, 100.5, 100.2, 100.8]) cfg = GexConfig() spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) later = today + timedelta(days=7) contracts = [] for exp in (today, later): for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 8000 if cp == "P" else 3000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 15, 35, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) ts_str = "2026-07-24 19:35:29" # a frozen capture time, as persisted by fetch at capture captured = "2026-07-24T19:36:00+00:00" WHITELIST = {"render_lag_min"} # the ONLY field allowed to differ between renders with tempfile.TemporaryDirectory() as tmpdir: cfg_tmp = GexConfig(outdir=tmpdir) _process_and_render("TEST", cfg_tmp, "am", "test_fix75", ts_str, spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc=captured) path = os.path.join(tmpdir, "TEST", "2026-07-24_am.json") with open(path) as f: first = f.read() first_obj = json.loads(first) # re-render the identical snapshot (same capture time) into the same file _process_and_render("TEST", cfg_tmp, "am", "test_fix75", ts_str, spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc=captured) with open(path) as f: second = f.read() second_obj = json.loads(second) # the frozen age must be identical across renders assert first_obj["source_timestamp_age_min"] == second_obj["source_timestamp_age_min"] # render_lag_min is published and is a number (capture time is in the past) assert isinstance(first_obj["render_lag_min"], (int, float)) # every field except the whitelist must match exactly all_keys = set(first_obj) | set(second_obj) for k in all_keys - WHITELIST: assert first_obj.get(k) == second_obj.get(k), ( f"field {k!r} differs between renders: " f"{first_obj.get(k)!r} != {second_obj.get(k)!r}") # --------------------------------------------------------------------------- # FIX 71: render_spacing must equal render_bucket (single source of truth) # --------------------------------------------------------------------------- def test_render_spacing_equals_render_bucket(): """FIX 71 / FIX 97: render_spacing is an ALIAS of render_bucket — the bucket is computed once in snapshot.py from the plot-band window and drives both the published fields and the renderer's bar aggregation. They must never disagree. (FIX 97 removed the ex-front variant; there is now a single pipeline.)""" from gex.snapshot import _process_and_render import json, tempfile, os from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo cfg = GexConfig() spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) later = today + timedelta(days=7) contracts = [] for exp in (today, later): for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 5000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 15, 35, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) ts_str = "2026-07-24 19:35:29" snap_id = "test_fix71" with tempfile.TemporaryDirectory() as tmpdir: cfg_tmp = GexConfig(outdir=tmpdir) _process_and_render("TEST", cfg_tmp, "am", snap_id, ts_str, spot, df_full, suffix="", instrument_class="equity_etf") path = os.path.join(tmpdir, "TEST", "2026-07-24_am.json") with open(path) as f: d = json.load(f) assert "render_bucket" in d, "render_bucket missing" assert "render_spacing" in d, "render_spacing missing" assert d["render_spacing"] == d["render_bucket"], ( f"render_spacing ({d['render_spacing']}) != " f"render_bucket ({d['render_bucket']})") # bucket must be at least the strike increment assert d["render_bucket"] >= d["strike_increment"] # --------------------------------------------------------------------------- # FIX 73: dual-threshold pass (signed < 0.05 AND unsigned < 0.10) + denominators # --------------------------------------------------------------------------- def test_reconciliation_dual_gate_and_denominators(): """FIX 73: profile_reliable requires BOTH rel_err < 0.05 AND rel_err_unsigned < 0.10. reconciliation_pass_basis names the binding metric; rel_err_denominator and rel_err_unsigned_denominator are published so a variant scored against a smaller book is visible.""" from gex.snapshot import _process_and_render import json, tempfile, os from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo cfg = GexConfig() spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) later = today + timedelta(days=7) contracts = [] for exp in (today, later): for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): # FIX 73 test: asymmetric OI (puts heavier) so per-expiry reported # GEX does NOT cancel to zero — a symmetric chain would make # Σ|reported| ≈ 0 and rel_err_unsigned_denominator = 0. contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 8000 if cp == "P" else 3000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 15, 35, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) ts_str = "2026-07-24 19:35:29" with tempfile.TemporaryDirectory() as tmpdir: cfg_tmp = GexConfig(outdir=tmpdir) _process_and_render("TEST", cfg_tmp, "am", "test_fix73", ts_str, spot, df_full, suffix="", instrument_class="equity_etf") path = os.path.join(tmpdir, "TEST", "2026-07-24_am.json") with open(path) as f: d = json.load(f) # new fields present assert "reconciliation_pass_basis" in d assert "rel_err_denominator" in d assert "rel_err_unsigned_denominator" in d assert d["rel_err_denominator"] > 0 assert d["rel_err_unsigned_denominator"] > 0 # the gate logic must be self-consistent with the published errors signed = d["reconciliation"]["rel_err"] unsigned = d["rel_err_unsigned"] expected_ok = (signed < 0.05) and (unsigned is None or unsigned < 0.10) assert d["profile_reliable"] == expected_ok, ( f"profile_reliable={d['profile_reliable']} but signed={signed} " f"unsigned={unsigned} -> expected {expected_ok}") # pass_basis is one of the documented values assert d["reconciliation_pass_basis"] in ( "signed", "unsigned", "signed+unsigned") # FIX 76: denominator convention is published and is "variant" assert d["reconciliation_denominator_basis"] == "variant" # --------------------------------------------------------------------------- # FIX 77: exercise the UNSIGNED branch of the FIX 73 gate # --------------------------------------------------------------------------- def test_unsigned_branch_binds_gate(monkeypatch, tmp_path): """FIX 77: the dual gate's unsigned condition must actually bind on real-shaped data, not only in the abstract. Engineered fixture: two multi-day expiries whose reported gammas deviate from Black-Scholes in OPPOSITE directions but with equal OI weight, so the per-expiry gaps CANCEL in the signed sum (signed ~0.003 < 0.05) yet ADD in the unsigned sum (unsigned ~0.63 >= 0.10). The gate must therefore fail on the unsigned branch: reconciliation_pass_basis == "unsigned", profile_reliable == False, and the reconciliation-failure banner renders. This confirms the basis string is genuinely derived, not defaulting to "signed".""" import matplotlib matplotlib.use("Agg") from matplotlib.figure import Figure from gex.snapshot import _process_and_render import gex.snapshot as snap_mod import json, os from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo # determinism: pin realised-vol input (no network, no outdir feedback) monkeypatch.setattr(snap_mod, "_load_recent_closes", lambda symbol, cfg: [99.0, 100.0, 101.0, 100.5, 100.2, 100.8]) # FIX 85 isolation: this test verifies the dual-gate MECHANICS under a FIXED # 0.10 gate. Mock the Monte Carlo floor to a non-bracketing result so the # adaptive per-symbol threshold does not absorb the engineered unsigned error. monkeypatch.setattr(snap_mod, "reconciliation_floor_unsigned", lambda df, spot, cfg, **kw: { "method": "grid_rounding", "rounding_interval": 0.5e-4, "floor": 0.002, "brackets_gate": False}) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) # rep0 + rep1 = 0.106 keeps the signed sum near zero (matches the BS profile # total at spot); the split (0.03 vs 0.076) puts the two per-expiry gaps on # opposite sides so they cancel signed but add unsigned. contracts = [] for dte, rep_gamma in ((10, 0.03), (40, 0.076)): exp = today + timedelta(days=dte) for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 9000 if cp == "P" else 1000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": rep_gamma, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 15, 35, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) ts_str = "2026-07-24 19:35:29" # spy on fig.text to confirm the reconciliation-failure banner renders texts = [] orig_text = Figure.text def spy_text(self, x, y, s, *a, **k): texts.append(str(s)) return orig_text(self, x, y, s, *a, **k) monkeypatch.setattr(Figure, "text", spy_text) _process_and_render("TEST", cfg, "am", "test_fix77", ts_str, spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T19:36:00+00:00") path = os.path.join(str(tmp_path), "TEST", "2026-07-24_am.json") with open(path) as f: d = json.load(f) signed = d["reconciliation"]["rel_err"] unsigned = d["rel_err_unsigned"] # the engineered window: signed passes, unsigned fails assert signed < 0.05, f"fixture broken: signed={signed:.4f} should be < 0.05" assert unsigned >= 0.10, f"fixture broken: unsigned={unsigned:.4f} should be >= 0.10" # the gate binds on the UNSIGNED branch assert d["reconciliation_pass_basis"] == "unsigned", ( f"expected basis 'unsigned', got {d['reconciliation_pass_basis']!r} " f"(signed={signed:.4f}, unsigned={unsigned:.4f})") assert d["profile_reliable"] is False # FIX 83: the nested reconciliation.pass must agree with profile_reliable. # Before FIX 83 it was signed-only (True here, since signed < 0.05), contradicting # profile_reliable == False. One authoritative pass flag. assert d["reconciliation"]["pass"] is d["profile_reliable"], ( "reconciliation.pass must equal profile_reliable (FIX 83)") assert d["reconciliation"]["pass"] is False # FIX 95: the reconciliation banner NO LONGER renders on the chart image. The # PNG is caveat-free; the failure status lives in the page's collapsible #
section (driven by the JSON fields asserted above). Confirm the # image carries no reconciliation commentary. banners = [t for t in texts if "RECONCILIATION" in t.upper()] assert not banners, ( "FIX 95: reconciliation status must NOT render on the PNG; " f"found fig.text: {banners}") # --------------------------------------------------------------------------- # FIX 48: test_expired_contracts_dropped # --------------------------------------------------------------------------- def test_expired_contracts_dropped(): """A chain containing an expiry one day BEFORE the snapshot date must be excluded; expired_contracts_dropped == 1; and the profile maximum must stay within 3x of total_net_gex (i.e. no expired-contract gamma explosion).""" cfg = GexConfig() spot = 100.0 snap_et = datetime(2026, 7, 24, 15, 35, tzinfo=_ET) today = snap_et.date() # one EXPIRED contract (expiry yesterday) with huge OI right at the money — # pre-FIX-48 this would survive (DTE floored to 0) and explode the profile. expired = {"strike": 100.0, "expiry": today - timedelta(days=1), "cp": "C", "iv": 0.30, "oi": 500000, "volume": 0, "delta": 0.5, "gamma": 0.10, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1} # valid contracts on a live expiry bracketing spot live = [] for strike in (95.0, 105.0): for cp in ("C", "P"): live.append({"strike": strike, "expiry": date(2026, 8, 15), "cp": cp, "iv": 0.25, "oi": 5000, "volume": 0, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.04, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}) df_full = filter_contracts_full([expired] + live, spot, cfg, snap_et) # the expired contract must be gone assert df_full.attrs.get("expired_contracts_dropped") == 1, \ f"expected 1 expired dropped, got {df_full.attrs.get('expired_contracts_dropped')}" assert (df_full["expiry"] >= today).all(), "no expired contract should survive the filter" # profile maximum must stay within 3x of total_net_gex (no explosion) grid = np.linspace(spot * 0.8, spot * 1.2, 400) gp, _ = gex_profile(df_full, spot, cfg, grid) total = total_net_gex_from_contracts(df_full, spot, cfg) assert np.isfinite(gp).all(), "profile must be finite" gp_max = float(np.max(np.abs(gp))) assert gp_max <= 3.0 * abs(total), \ f"profile max {gp_max:.3g} exceeds 3x total_net_gex {abs(total):.3g} — gamma explosion" # --------------------------------------------------------------------------- # v1.6.2 item 1: --from-cache must not be blocked by the age_min refusal, # while the NYSE-trading-day and 09:30-16:15 ET window checks stay active. # --------------------------------------------------------------------------- def test_from_cache_skips_age_refusal_only(tmp_path, monkeypatch): """Three cases, wall-clock pinned to a Saturday so the age check is live: (a) cached Friday-15:35-ET snapshot replayed with from_cache=True renders (age refusal skipped); the SAME snapshot with from_cache=False refuses (age > 240 min). (b) cached Saturday-stamped snapshot still refuses (not a trading day) even with from_cache=True. (c) cached 23:44-ET snapshot still refuses (outside window) even with from_cache=True. """ import gex.snapshot as snap_mod spot = 100.0 friday = date(2026, 7, 24) # an NYSE trading day saturday = date(2026, 7, 25) # not a trading day def make_chain(exp): chain = [] for strike in (95.0, 100.0, 105.0): for cp in ("C", "P"): chain.append({"strike": strike, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 5000, "volume": 0, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.04, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}) return chain # pin wall-clock to a fixed Saturday 16:00 ET so the age check is exercised fake_now = datetime(2026, 7, 25, 16, 0, 0, tzinfo=_ET) class FakeDatetime(datetime): @classmethod def now(cls, tz=None): return fake_now if tz is None else fake_now.astimezone(tz) monkeypatch.setattr(snap_mod, "datetime", FakeDatetime) rendered = [] monkeypatch.setattr(snap_mod, "_process_and_render", lambda *a, **k: rendered.append(a[0] if a else k.get("symbol")) or (a[0] if a else "ok")) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) def run(ts_str, from_cache): chain = make_chain(friday) monkeypatch.setattr(snap_mod, "fetch_chain", lambda symbol, cfg, from_cache=False: ({"data": {}}, "snap_test", "plain")) monkeypatch.setattr(snap_mod, "parse_chain", lambda data, symbol: (list(chain), spot, ts_str)) return snap_mod.run_ticker("SMH", cfg, "pm", from_cache=from_cache) # (a) Friday 15:50 ET (19:50 UTC) — valid trading day + pm window, but age > 240 min assert run("2026-07-24 19:50:00", from_cache=True) is True, \ "cached Friday-15:50 snapshot must render under --from-cache (age refusal skipped)" assert len(rendered) == 1, "exactly one render expected for the cached Friday replay" assert run("2026-07-24 19:50:00", from_cache=False) is False, \ "the same stale snapshot must refuse WITHOUT --from-cache (age > 240 min)" assert len(rendered) == 1, "no additional render for the non-cache stale refusal" # (b) Saturday 10:05 ET (14:05 UTC) — not a trading day; refuses even from cache assert run("2026-07-25 14:05:00", from_cache=True) is False, \ "Saturday-stamped snapshot must refuse even under --from-cache" # (c) Friday 23:44 ET (Sat 03:44 UTC) — outside 09:30-16:15; refuses even from cache assert run("2026-07-25 03:44:00", from_cache=True) is False, \ "23:44-ET snapshot must refuse even under --from-cache" # --------------------------------------------------------------------------- # v1.6.4: a firing profile outlier guard MUST surface the red FAULT footnote # --------------------------------------------------------------------------- def test_outlier_guard_fault_footnote_rendered(tmp_path, monkeypatch): """The outlier guard is kept as drop-and-alert (v1.6.4): its loudness is load-bearing, so if it fires (drops a contract) the chart MUST carry the red 'FAULT' footnote. This test fails if a firing guard does not produce it. """ import matplotlib matplotlib.use("Agg") from matplotlib.figure import Figure from gex.plot import render_chart cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) spot = 105.0 strikes = np.arange(95, 116, 5.0) agg = pd.DataFrame({ "gex_call": np.random.uniform(1e6, 5e6, len(strikes)), "gex_put": -np.random.uniform(1e6, 5e6, len(strikes)), "dex": np.random.uniform(-2e7, 2e7, len(strikes)), "oi_call": np.random.randint(100, 5000, len(strikes)), "oi_put": np.random.randint(100, 5000, len(strikes)), }, index=strikes) agg["net_gex"] = agg["gex_call"] + agg["gex_put"] grid = np.arange(90.0, 121.0, 1.0) gp = np.linspace(-3e6, 3e6, len(grid)) dp = np.linspace(-1e9, 1e9, len(grid)) increment = 5.0 levels = compute_levels(agg, grid, gp, spot, increment) # simulate a guard firing: one contract dropped levels["profile_outliers_dropped"] = [ {"strike": 100.0, "expiry": "2026-07-24", "value": 1.2e9, "price_width_pct": 1.1} ] # capture every fig.text() string so we can assert the footnote was emitted texts = [] orig_text = Figure.text def spy_text(self, x, y, s, *a, **k): texts.append(str(s)) return orig_text(self, x, y, s, *a, **k) monkeypatch.setattr(Figure, "text", spy_text) png = render_chart("TEST", agg, grid, gp, grid, dp, levels, "2026-07-24 15:44:00", cfg, str(tmp_path), "pm", increment) assert png.exists(), f"PNG not found: {png}" fault_notes = [t for t in texts if "FAULT" in t and "profile outlier guard" in t] assert fault_notes, ( "a firing outlier guard MUST render the red FAULT footnote; " f"fig.text calls were: {texts}" ) assert "100" in fault_notes[0], "footnote must name the dropped strike" # --------------------------------------------------------------------------- # v1.6.5 FIX 52: --replay-unsafe bypasses ONLY the session check, forces a # scratch outdir, and stamps replay_unsafe into the levels JSON. # --------------------------------------------------------------------------- def test_replay_unsafe_forces_scratch_outdir_and_stamp(tmp_path, monkeypatch): """Two parts: (a) replay_unsafe=True on an out-of-session (Saturday) chain renders, passes replay_unsafe=True down to _process_and_render, and main() forces the outdir to the scratch path out_replay/. (b) the SAME out-of-session chain with replay_unsafe=False refuses (returns False) — the session guard stays active without the flag. """ import gex.snapshot as snap_mod spot = 100.0 saturday = date(2026, 7, 25) # not an NYSE trading day def make_chain(exp): chain = [] for strike in (95.0, 100.0, 105.0): for cp in ("C", "P"): chain.append({"strike": strike, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 5000, "volume": 0, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.04, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}) return chain # pin wall-clock to a fixed Saturday so the age check is not the deciding factor fake_now = datetime(2026, 7, 25, 16, 0, 0, tzinfo=_ET) class FakeDatetime(datetime): @classmethod def now(cls, tz=None): return fake_now if tz is None else fake_now.astimezone(tz) monkeypatch.setattr(snap_mod, "datetime", FakeDatetime) captured = [] monkeypatch.setattr( snap_mod, "_process_and_render", lambda *a, **k: captured.append(k.get("replay_unsafe")) or "ok") chain = make_chain(saturday) monkeypatch.setattr(snap_mod, "fetch_chain", lambda symbol, cfg, from_cache=False: ({"data": {}}, "snap_test", "plain")) # Saturday 10:05 ET (14:05 UTC) — out of session (not a trading day) monkeypatch.setattr(snap_mod, "parse_chain", lambda data, symbol: (list(chain), spot, "2026-07-25 14:05:00")) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) # (a) replay_unsafe=True renders and stamps replay_unsafe=True downstream assert snap_mod.run_ticker("NDX", cfg, "pm", from_cache=True, replay_unsafe=True) is True, \ "out-of-session chain must render under --replay-unsafe" assert captured == [True], \ f"replay_unsafe=True must reach _process_and_render; got {captured}" # (b) replay_unsafe=False refuses the same out-of-session chain captured.clear() assert snap_mod.run_ticker("NDX", cfg, "pm", from_cache=True, replay_unsafe=False) is False, \ "out-of-session chain must REFUSE without --replay-unsafe" assert captured == [], "no render expected for the refused run" # (c) main() forces the outdir to the scratch path under --replay-unsafe monkeypatch.setattr("sys.argv", ["snapshot", "--tickers", "NDX", "--slot", "pm", "--from-cache", "--replay-unsafe"]) seen_cfg = {} monkeypatch.setattr(snap_mod, "run_ticker", lambda symbol, cfg, slot, **k: seen_cfg.update(outdir=cfg.outdir) or True) monkeypatch.setattr(snap_mod, "build_index_json", lambda cfg: None) snap_mod.main() assert seen_cfg["outdir"] == "out_replay", \ f"--replay-unsafe must force outdir to out_replay/, got {seen_cfg['outdir']!r}" # --------------------------------------------------------------------------- # FIX 79a: slot source-window check at write time # --------------------------------------------------------------------------- def test_slot_window_rejects_out_of_window_capture(tmp_path, monkeypatch): """FIX 79a: a capture whose ET time falls outside the requested slot's window must be refused at write time, not just at fetch. A 15:55-ET capture with slot='am' (window 09:48-10:12) must return False; a 10:00-ET capture with slot='pm' (window 15:47-16:11) must also return False.""" import gex.snapshot as snap_mod from datetime import date, datetime from zoneinfo import ZoneInfo spot = 100.0 friday = date(2026, 7, 24) ET = ZoneInfo("America/New_York") def make_chain(): chain = [] for strike in (95.0, 100.0, 105.0): for cp in ("C", "P"): chain.append({"strike": strike, "expiry": friday, "cp": cp, "iv": 0.25, "oi": 5000, "volume": 0, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.04, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}) return chain # Pin wall-clock to Friday 16:00 ET so the age check passes for in-session captures fake_now = datetime(2026, 7, 24, 16, 0, 0, tzinfo=ET) class FakeDatetime(datetime): @classmethod def now(cls, tz=None): return fake_now if tz is None else fake_now.astimezone(tz) monkeypatch.setattr(snap_mod, "datetime", FakeDatetime) rendered = [] monkeypatch.setattr(snap_mod, "_process_and_render", lambda *a, **k: rendered.append(a[0] if a else k.get("symbol")) or (a[0] if a else "ok")) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) def run(ts_str, slot): chain = make_chain() monkeypatch.setattr(snap_mod, "fetch_chain", lambda symbol, cfg, from_cache=False: ({"data": {}}, "snap_test", "plain")) monkeypatch.setattr(snap_mod, "parse_chain", lambda data, symbol: (list(chain), spot, ts_str)) return snap_mod.run_ticker("SMH", cfg, slot, from_cache=True) # 15:55 ET (19:55 UTC) with slot="am" — outside am window (09:48-10:12) assert run("2026-07-24 19:55:00", "am") is False, \ "15:55-ET capture must be refused for slot='am'" assert len(rendered) == 0, "no render expected for out-of-window capture" # 10:00 ET (14:00 UTC) with slot="pm" — outside pm window (15:47-16:11) assert run("2026-07-24 14:00:00", "pm") is False, \ "10:00-ET capture must be refused for slot='pm'" assert len(rendered) == 0, "no render expected for out-of-window capture" # 10:00 ET with slot="am" — inside am window, should succeed assert run("2026-07-24 14:00:00", "am") is True, \ "10:00-ET capture must be accepted for slot='am'" assert len(rendered) == 1, "exactly one render expected for in-window capture" # 15:55 ET with slot="pm" — inside pm window, should succeed rendered.clear() assert run("2026-07-24 19:55:00", "pm") is True, \ "15:55-ET capture must be accepted for slot='pm'" assert len(rendered) == 1, "exactly one render expected for in-window capture" def test_slot_window_uses_capture_time_not_source_ts(tmp_path, monkeypatch): """FIX 79a (capture-time path): the slot window must be checked against the CAPTURE time (when the fetch happened), NOT the Cboe source timestamp. The source feed is delayed ~15 min, so a fetch at 10:00 ET carries data timestamped ~10:15 ET. Checking the source timestamp would reject every legitimate AM capture. This test pins the production scenario: capture at 10:00 ET (in the AM window 09:30–12:00, FIX 86b) with source ts 10:15 ET must be ACCEPTED for slot='am', because the capture time governs.""" import gex.snapshot as snap_mod from datetime import date, datetime from zoneinfo import ZoneInfo spot = 100.0 friday = date(2026, 7, 24) ET = ZoneInfo("America/New_York") def make_chain(): chain = [] for strike in (95.0, 100.0, 105.0): for cp in ("C", "P"): chain.append({"strike": strike, "expiry": friday, "cp": cp, "iv": 0.25, "oi": 5000, "volume": 0, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.04, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}) return chain # Pin wall-clock so the >240-min age refusal does not interfere. fake_now = datetime(2026, 7, 24, 16, 0, 0, tzinfo=ET) class FakeDatetime(datetime): @classmethod def now(cls, tz=None): return fake_now if tz is None else fake_now.astimezone(tz) monkeypatch.setattr(snap_mod, "datetime", FakeDatetime) rendered = [] monkeypatch.setattr(snap_mod, "_process_and_render", lambda *a, **k: rendered.append(a[0] if a else k.get("symbol")) or (a[0] if a else "ok")) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) def run(source_ts_str, captured_at_utc, slot): chain = make_chain() # fetch_chain returns _captured_at_utc so run_ticker uses the capture-time path monkeypatch.setattr( snap_mod, "fetch_chain", lambda symbol, cfg, from_cache=False: ( {"data": {}, "_captured_at_utc": captured_at_utc}, "snap_test", "plain")) monkeypatch.setattr(snap_mod, "parse_chain", lambda data, symbol: (list(chain), spot, source_ts_str)) return snap_mod.run_ticker("SMH", cfg, slot, from_cache=True) # Capture at 10:00 ET (14:00 UTC, in am window) but Cboe source ts 10:15 ET # (14:15 UTC, OUTSIDE am window). Must be ACCEPTED — capture time governs. assert run("2026-07-24 14:15:00", "2026-07-24T14:00:00+00:00", "am") is True, \ "10:00-ET capture with delayed 10:15-ET source ts must be accepted for slot='am'" assert len(rendered) == 1, "exactly one render expected" # Capture at 15:55 ET (19:55 UTC, in pm window) but source ts 10:00 ET # (14:00 UTC, an am time). Must be REFUSED for slot='am' — the capture happened # in the pm window, so it must not be written as an am file. rendered.clear() assert run("2026-07-24 14:00:00", "2026-07-24T19:55:00+00:00", "am") is False, \ "15:55-ET capture must be refused for slot='am' even if source ts is an am time" assert len(rendered) == 0, "no render expected for out-of-window capture" # --------------------------------------------------------------------------- # FIX 79b: non-destructive writes # --------------------------------------------------------------------------- def test_non_destructive_write_refuses_different_snapshot_id(monkeypatch, tmp_path): """FIX 79b: if an output JSON already exists with a DIFFERENT snapshot_id, the write must be refused (return None) and the existing file must NOT be modified. A re-render of the SAME snapshot_id must always succeed.""" import matplotlib matplotlib.use("Agg") from gex.snapshot import _process_and_render import gex.snapshot as snap_mod import json, os from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo monkeypatch.setattr(snap_mod, "_load_recent_closes", lambda symbol, cfg: [99.0, 100.0, 101.0, 100.5, 100.2, 100.8]) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) contracts = [] for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": today, "cp": cp, "iv": 0.25, "oi": 5000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 15, 55, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) ts_str = "2026-07-24 19:55:00" # First write with snapshot_id "snap_A" result_a = _process_and_render("TEST", cfg, "pm", "snap_A", ts_str, spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T19:56:00+00:00") assert result_a is not None, "first write must succeed" json_path = os.path.join(str(tmp_path), "TEST", "2026-07-24_pm.json") assert os.path.exists(json_path), f"JSON not found: {json_path}" with open(json_path) as f: data_a = json.load(f) assert data_a["snapshot_id"] == "snap_A" # Second write with a DIFFERENT snapshot_id "snap_B" — must be refused result_b = _process_and_render("TEST", cfg, "pm", "snap_B", ts_str, spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T19:56:00+00:00") assert result_b is None, "write with different snapshot_id must be refused" with open(json_path) as f: data_after = json.load(f) assert data_after["snapshot_id"] == "snap_A", \ f"existing file must NOT be clobbered; got {data_after['snapshot_id']!r}" # Third write with the SAME snapshot_id "snap_A" — must succeed (re-render) result_c = _process_and_render("TEST", cfg, "pm", "snap_A", ts_str, spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T19:56:00+00:00") assert result_c is not None, "re-render of same snapshot_id must succeed" with open(json_path) as f: data_c = json.load(f) assert data_c["snapshot_id"] == "snap_A" def test_overwrite_flag_forces_clobber(monkeypatch, tmp_path): """FIX 79b: with overwrite=True, a different snapshot_id MUST clobber the existing file.""" import matplotlib matplotlib.use("Agg") from gex.snapshot import _process_and_render import gex.snapshot as snap_mod import json, os from datetime import date, datetime from zoneinfo import ZoneInfo monkeypatch.setattr(snap_mod, "_load_recent_closes", lambda symbol, cfg: [99.0, 100.0, 101.0, 100.5, 100.2, 100.8]) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) contracts = [] for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": today, "cp": cp, "iv": 0.25, "oi": 5000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 15, 55, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) ts_str = "2026-07-24 19:55:00" # First write with snapshot_id "snap_A" _process_and_render("TEST", cfg, "pm", "snap_A", ts_str, spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T19:56:00+00:00") json_path = os.path.join(str(tmp_path), "TEST", "2026-07-24_pm.json") with open(json_path) as f: assert json.load(f)["snapshot_id"] == "snap_A" # Overwrite with snapshot_id "snap_B" and overwrite=True — must succeed result = _process_and_render("TEST", cfg, "pm", "snap_B", ts_str, spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T19:56:00+00:00", overwrite=True) assert result is not None, "overwrite=True must force the write" with open(json_path) as f: data = json.load(f) assert data["snapshot_id"] == "snap_B", \ f"overwrite=True must clobber; got {data['snapshot_id']!r}" # --------------------------------------------------------------------------- # FIX 80: reconciliation guard against near-settlement expiries # --------------------------------------------------------------------------- def test_reconciliation_excludes_near_settlement(monkeypatch, tmp_path): """FIX 80: an expiry inside min_minutes_to_settlement of settlement must be excluded from the reconciliation numerator AND denominator, but still plotted (bars use the full chain). reconciliation_scope must be 'excl_near_settlement' and reconciliation_excluded_expiries must list the expiry with its minutes_to_settlement. When no expiry is near settlement, scope is 'full'.""" import matplotlib matplotlib.use("Agg") from gex.snapshot import _process_and_render import gex.snapshot as snap_mod import json, os from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo monkeypatch.setattr(snap_mod, "_load_recent_closes", lambda symbol, cfg: [99.0, 100.0, 101.0, 100.5, 100.2, 100.8]) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) spot = 100.0 ET = ZoneInfo("America/New_York") # Snapshot at 15:55 ET. Front expiry = today (0DTE) settles at 16:00 ET -> 5 min # to settlement (< 30 threshold -> excluded). Second expiry = +7 days -> included. today = date(2026, 7, 24) later = today + timedelta(days=7) contracts = [] for exp in (today, later): for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 9000 if cp == "P" else 1000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) # ts_str is the Cboe source timestamp (UTC); 19:55 UTC = 15:55 ET. snap_et = datetime(2026, 7, 24, 15, 55, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) assert not df_full.empty ts_str = "2026-07-24 19:55:00" _process_and_render("TEST", cfg, "pm", "test_fix80", ts_str, spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T19:56:00+00:00") json_path = os.path.join(str(tmp_path), "TEST", "2026-07-24_pm.json") with open(json_path) as f: d = json.load(f) # The 0DTE front expiry is near settlement and must be excluded. assert d["reconciliation_scope"] == "excl_near_settlement" excluded = d["reconciliation_excluded_expiries"] assert len(excluded) == 1, f"expected 1 excluded expiry, got {excluded}" assert excluded[0]["expiry"] == str(today) assert excluded[0]["minutes_to_settlement"] < 30 # The near-settlement expiry is NOT in the worst-expiry reconciliation list # (it was dropped from the reconciliation entirely). worst_expiries = {e["expiry"] for e in d["reconciliation_worst_expiries"]} assert str(today) not in worst_expiries, \ "near-settlement expiry must not appear in reconciliation_worst_expiries" # Bars still reflect the full chain: total_net_gex_full uses ALL expiries. assert d["total_net_gex_full"] != 0 # --- Control: a snapshot with NO near-settlement expiry -> scope 'full' --- # Move the snapshot to mid-morning so the front expiry has > 30 min to settlement. snap_et2 = datetime(2026, 7, 24, 11, 0, tzinfo=ET) # 11:00 ET -> 300 min to 16:00 df_full2 = filter_contracts_full(contracts, spot, cfg, snap_et2) ts_str2 = "2026-07-24 15:00:00" # 15:00 UTC = 11:00 ET _process_and_render("TEST", cfg, "am", "test_fix80_full", ts_str2, spot, df_full2, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T15:01:00+00:00") json_path2 = os.path.join(str(tmp_path), "TEST", "2026-07-24_am.json") with open(json_path2) as f: d2 = json.load(f) assert d2["reconciliation_scope"] == "full" assert d2["reconciliation_excluded_expiries"] == [] # --------------------------------------------------------------------------- # FIX 82: publish the pass-1 ATM IV source so the atm_iv divergence is visible # --------------------------------------------------------------------------- def test_bands_publish_atm_iv_source(monkeypatch, tmp_path): """FIX 82: bands.atm_iv_used comes from PASS 1 (_pass1_atm_iv on df_coarse, DTE-filtered) while the top-level atm_iv comes from PASS 2 (atm_expected_move on df_band, strike-filtered). The two passes run on different contract sets and can legitimately pick different expiries/strikes (live SMH: 0.5752 vs 0.5861). The fix publishes bands.atm_iv_source_expiry and bands.atm_iv_source_strikes so the divergence is auditable. Both keys must ALWAYS be present (canonical schema), and must be populated whenever pass-1 resolved a real atm_iv.""" import matplotlib matplotlib.use("Agg") from gex.snapshot import _process_and_render import gex.snapshot as snap_mod import json, os from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo monkeypatch.setattr(snap_mod, "_load_recent_closes", lambda symbol, cfg: [99.0, 100.0, 101.0, 100.5, 100.2, 100.8]) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) later = today + timedelta(days=7) # DTE 7 -> eligible for pass-1 (DTE >= 5) contracts = [] for exp in (today, later): for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 9000 if cp == "P" else 1000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 11, 0, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) ts_str = "2026-07-24 15:00:00" # 15:00 UTC = 11:00 ET _process_and_render("TEST", cfg, "am", "test_fix82", ts_str, spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T15:01:00+00:00") json_path = os.path.join(str(tmp_path), "TEST", "2026-07-24_am.json") with open(json_path) as f: d = json.load(f) bands = d["bands"] # Canonical schema: both keys ALWAYS present (null when pass-1 didn't resolve). assert "atm_iv_source_expiry" in bands, \ f"bands keys: {sorted(bands.keys())}" assert "atm_iv_source_strikes" in bands, \ f"bands keys: {sorted(bands.keys())}" # When pass-1 resolved a real atm_iv (basis starts with 'atm_iv'), the source # must be published; when it fell back, the source is null. if bands["band_basis"].startswith("atm_iv"): assert bands["atm_iv_source_expiry"] is not None assert bands["atm_iv_source_strikes"] is not None assert isinstance(bands["atm_iv_source_strikes"], list) # The source strikes must bracket or equal spot (100.0). assert any(abs(k - spot) <= 5.0 for k in bands["atm_iv_source_strikes"]), \ f"source strikes {bands['atm_iv_source_strikes']} should be near spot {spot}" else: assert bands["atm_iv_source_expiry"] is None assert bands["atm_iv_source_strikes"] is None # When both passes pick the SAME expiry, the top-level atm_iv and # bands.atm_iv_used agree (live NDX case: both 0.286). When they differ, the # published source makes the divergence visible. Either way the two source # fields are internally consistent with the resolver output. if bands["atm_iv_source_expiry"] is not None and \ d.get("atm_iv_expiry") == bands["atm_iv_source_expiry"] and \ d.get("atm_iv_source_strikes") == bands["atm_iv_source_strikes"]: assert abs(d["atm_iv"] - bands["atm_iv_used"]) < 1e-9, \ "same source expiry+strikes must yield identical atm_iv" # --------------------------------------------------------------------------- # FIX 84: instrument-class-aware settlement time # --------------------------------------------------------------------------- def test_settlement_time_instrument_class(): """FIX 84: the settlement clock depends on instrument class. PM-settled index options settle at 16:15 ET (not 16:00), AM-settled index expiries at 09:30 ET one day earlier, equity/ETF at 16:00 ET. Using 16:00 for a PM-settled index understates time-to-settlement by 15 min — for a 0DTE NDX captured at 15:54 that is 6 min vs the true 21 min, pushing gamma ~1.9x too high.""" from datetime import date, datetime, time as dtime from zoneinfo import ZoneInfo from gex.compute import _settlement_time, minutes_to_settlement, time_to_expiry_years ET = ZoneInfo("America/New_York") # --- settlement clock --- assert _settlement_time(am_settled=False, instrument_class="index") == dtime(16, 15) assert _settlement_time(am_settled=True, instrument_class="index") == dtime(9, 30) assert _settlement_time(am_settled=False, instrument_class="equity_etf") == dtime(16, 0) assert _settlement_time(am_settled=True, instrument_class="equity_etf") == dtime(9, 30) # --- minutes_to_settlement: 0DTE NDX captured at 15:54:22 ET --- expiry = date(2026, 7, 27) snap = datetime(2026, 7, 27, 15, 54, 22, tzinfo=ET) mts_index = minutes_to_settlement(expiry, snap, am_settled=False, instrument_class="index") mts_etf = minutes_to_settlement(expiry, snap, am_settled=False, instrument_class="equity_etf") # index: 16:15 - 15:54:22 = 20.63 min (NOT 6 min) assert abs(mts_index - 20.63) < 0.1, f"index 0DTE should be ~20.6 min, got {mts_index}" # equity/ETF: 16:00 - 15:54:22 = 5.63 min assert abs(mts_etf - 5.63) < 0.1, f"etf 0DTE should be ~5.6 min, got {mts_etf}" # --- T shares the same assumption (the user's explicit concern) --- # At 0DTE both minutes are below the T floor (~160 min), so T is floored # identically — the gamma cap dominates. Verify the shared clock instead with # an above-floor case: a +1-day expiry captured at 11:00 ET. Index settles # 16:15 (315 min), equity/ETF settles 16:00 (300 min) — both above the floor, # so the index T must be strictly larger. cfg = GexConfig() expiry_tmr = date(2026, 7, 28) snap_mid = datetime(2026, 7, 27, 11, 0, tzinfo=ET) T_index = time_to_expiry_years(expiry_tmr, snap_mid, cfg, am_settled=False, instrument_class="index") T_etf = time_to_expiry_years(expiry_tmr, snap_mid, cfg, am_settled=False, instrument_class="equity_etf") assert T_index > T_etf, "index T must exceed equity/ETF T (16:15 vs 16:00 clock)" # the difference is exactly 15 minutes of settlement time delta_min = (T_index - T_etf) * (365.0 * 24.0 * 60.0) assert abs(delta_min - 15.0) < 0.01, \ f"T difference should be exactly 15 min, got {delta_min}" def test_settlement_time_published_per_expiry(monkeypatch, tmp_path): """FIX 84: reconciliation_excluded_expiries entries carry settlement_time_et so the settlement assumption is auditable. An index 0DTE captured at 15:54 ET has 20.6 min to settlement (16:15 clock) — under the 30-min threshold, so it is excluded and its settlement_time_et is published as '16:15'.""" import matplotlib matplotlib.use("Agg") from gex.snapshot import _process_and_render import gex.snapshot as snap_mod import json, os from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo monkeypatch.setattr(snap_mod, "_load_recent_closes", lambda symbol, cfg: [99.0, 100.0, 101.0, 100.5, 100.2, 100.8]) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 27) later = today + timedelta(days=7) contracts = [] for exp in (today, later): for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 9000 if cp == "P" else 1000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) # 15:54 ET = 19:54 UTC. Index 0DTE -> 20.6 min to settlement (< 30 -> excluded). snap_et = datetime(2026, 7, 27, 15, 54, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et, instrument_class="index") ts_str = "2026-07-27 19:54:00" _process_and_render("NDX", cfg, "pm", "test_fix84", ts_str, spot, df_full, suffix="", instrument_class="index", captured_at_utc="2026-07-27T19:55:00+00:00") json_path = os.path.join(str(tmp_path), "NDX", "2026-07-27_pm.json") with open(json_path) as f: d = json.load(f) excluded = d["reconciliation_excluded_expiries"] today_excl = [e for e in excluded if e["expiry"] == str(today)] assert today_excl, f"0DTE index expiry should be excluded, got {excluded}" entry = today_excl[0] assert entry["settlement_time_et"] == "16:15", \ f"index PM settlement must be 16:15, got {entry['settlement_time_et']}" # 21 min to settlement (16:15 - 15:54), NOT 6 min assert abs(entry["minutes_to_settlement"] - 21.0) < 0.2, \ f"index 0DTE should be ~21 min, got {entry['minutes_to_settlement']}" # --------------------------------------------------------------------------- # FIX 85: empirical unsigned reconciliation floor (Monte Carlo) # --------------------------------------------------------------------------- def test_reconciliation_floor_unsigned_shape(): """FIX 93: reconciliation_floor_unsigned is DETERMINISTIC — rounding to a fixed grid is not random, so the floor is a single scalar, not a Monte Carlo distribution. Returns method, rounding_interval, floor, brackets_gate.""" from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo from gex.compute import reconciliation_floor_unsigned cfg = GexConfig() spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) contracts = [] for dte in (7, 30): exp = today + timedelta(days=dte) for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 9000 if cp == "P" else 1000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 11, 0, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) floor = reconciliation_floor_unsigned(df_full, spot, cfg) # FIX 93: deterministic scalar, no n_draws/median/p95 spread. assert floor["method"] == "grid_rounding" assert floor["rounding_interval"] == 0.5e-4 assert floor["floor"] is not None assert floor["floor"] >= 0.0 assert isinstance(floor["brackets_gate"], bool) # Determinism: same input -> same output, no RNG. floor2 = reconciliation_floor_unsigned(df_full, spot, cfg) assert floor2["floor"] == floor["floor"], "floor must be deterministic" def test_unsigned_gate_adapts_to_floor(monkeypatch, tmp_path): """FIX 85: when the floor's p95 brackets the unsigned gate (0.10), the effective unsigned threshold becomes floor_p95 × multiplier (1.5), and unsigned_gate_effective is published. A symbol whose unsigned error sits between 0.10 and the derived threshold then passes on the unsigned branch — NOT because of hand-tuning, but because the gate is dominated by rounding noise. When the floor does NOT bracket, the fixed 0.10 gate applies.""" import matplotlib matplotlib.use("Agg") from gex.snapshot import _process_and_render import gex.snapshot as snap_mod import json, os from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo monkeypatch.setattr(snap_mod, "_load_recent_closes", lambda symbol, cfg: [99.0, 100.0, 101.0, 100.5, 100.2, 100.8]) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) later = today + timedelta(days=7) contracts = [] for exp in (today, later): for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 9000 if cp == "P" else 1000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 11, 0, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) ts_str = "2026-07-24 15:00:00" # --- Case A: floor brackets the gate -> adaptive threshold = floor × 1.5 --- monkeypatch.setattr(snap_mod, "reconciliation_floor_unsigned", lambda df, spot, cfg, **kw: { "method": "grid_rounding", "rounding_interval": 0.5e-4, "floor": 0.12, "brackets_gate": True}) _process_and_render("BRK", cfg, "am", "test_fix85a", ts_str, spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T15:01:00+00:00") with open(os.path.join(str(tmp_path), "BRK", "2026-07-24_am.json")) as f: da = json.load(f) assert da["reconciliation_floor_unsigned"]["brackets_gate"] is True assert abs(da["unsigned_gate_effective"] - 0.12 * 1.5) < 1e-9, \ f"adaptive gate should be floor 0.12 × 1.5 = 0.18, got {da['unsigned_gate_effective']}" # --- Case B: floor does NOT bracket -> fixed 0.10 gate applies --- monkeypatch.setattr(snap_mod, "reconciliation_floor_unsigned", lambda df, spot, cfg, **kw: { "method": "grid_rounding", "rounding_interval": 0.5e-4, "floor": 0.02, "brackets_gate": False}) _process_and_render("NBR", cfg, "am", "test_fix85b", ts_str, spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T15:01:00+00:00") with open(os.path.join(str(tmp_path), "NBR", "2026-07-24_am.json")) as f: db = json.load(f) assert db["reconciliation_floor_unsigned"]["brackets_gate"] is False assert abs(db["unsigned_gate_effective"] - 0.10) < 1e-9, \ f"non-bracketing gate should stay 0.10, got {db['unsigned_gate_effective']}" # --------------------------------------------------------------------------- # FIX 87: the grid_rounding floor estimator does not double-count rounding # --------------------------------------------------------------------------- def test_floor_grid_rounding_does_not_double_count(): """FIX 87: the corrected estimator rounds the recomputed (truth) gamma to the 4dp publication grid and measures rounded-vs-unrounded. Because BOTH sides derive from the recomputed gamma, the genuine model error cancels and only quantisation noise remains — so the floor must be far smaller than the old FIX 85 estimator (which perturbed already-rounded reported gamma and folded in model error). For an NDX-like chain (spot ~28000, 1-sig-fig gamma) the old estimator gave median ~0.166 / p95 ~0.236; the grid_rounding method must fall well below that.""" from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo from gex.compute import reconciliation_floor_unsigned cfg = GexConfig() spot = 28000.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) contracts = [] for dte in (7, 30): exp = today + timedelta(days=dte) for k in (27500.0, 28000.0, 28500.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 9000 if cp == "P" else 1000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.0001, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 11, 0, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) floor = reconciliation_floor_unsigned(df_full, spot, cfg) assert floor["method"] == "grid_rounding" # The whole point of FIX 87: the corrected floor is well below the old 0.236. # FIX 93: deterministic scalar, not a distribution. assert floor["floor"] < 0.15, \ f"grid_rounding floor ({floor['floor']:.4f}) should be well below the old " \ f"double-counted 0.236 — model error must cancel, leaving only rounding" # --------------------------------------------------------------------------- # FIX 88: gate cap + indeterminate (precision_limited) state # --------------------------------------------------------------------------- def test_gate_capped_and_indeterminate(monkeypatch, tmp_path): """FIX 88: a derived gate is capped at 2× the base gate (0.20). When the floor brackets the base gate AND the unsigned error is not cleanly below the base gate, the result is precision-limited: profile_reliable == False (boolean, FIX 92), unsigned_gate_status == "precision_limited"/"capped", and the nested reconciliation.pass agrees (False). Passing and precision-limited must be distinct — precision-limited is NOT a pass.""" import matplotlib matplotlib.use("Agg") from gex.snapshot import _process_and_render import gex.snapshot as snap_mod import json, os from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo monkeypatch.setattr(snap_mod, "_load_recent_closes", lambda symbol, cfg: [99.0, 100.0, 101.0, 100.5, 100.2, 100.8]) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) later = today + timedelta(days=7) contracts = [] for exp in (today, later): for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 9000 if cp == "P" else 1000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 11, 0, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) ts_str = "2026-07-24 15:00:00" # Floor brackets the base gate (p95 = 0.12 > 0.10). The derived gate would be # 0.12 × 1.5 = 0.18 (< cap 0.20). Engineer the unsigned error to 0.11 — above the # base gate (0.10) so it is precision-limited, but below the derived gate (0.18). # We force this by mocking the floor AND checking the indeterminate path triggers # on the unsigned error being >= base gate while the floor brackets. monkeypatch.setattr(snap_mod, "reconciliation_floor_unsigned", lambda df, spot, cfg, **kw: { "method": "grid_rounding", "rounding_interval": 0.5e-4, "floor": 0.12, "brackets_gate": True}) # Build a chain whose unsigned error lands in (0.10, 0.18): reuse the FIX 77 # opposite-deviation fixture but scaled down. Two expiries, reported gammas that # cancel signed but add unsigned to ~0.12. contracts2 = [] for dte, rep_gamma in ((10, 0.024), (40, 0.028)): exp = today + timedelta(days=dte) for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts2.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 9000 if cp == "P" else 1000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": rep_gamma, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) df2 = filter_contracts_full(contracts2, spot, cfg, snap_et) _process_and_render("IND", cfg, "am", "test_fix88", ts_str, spot, df2, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T15:01:00+00:00") with open(os.path.join(str(tmp_path), "IND", "2026-07-24_am.json")) as f: d = json.load(f) unsigned = d["rel_err_unsigned"] # The gate is capped: derived 0.18 < cap 0.20, so effective is 0.18 (status derived). assert d["unsigned_gate_status"] in ("derived", "capped", "precision_limited") assert d["unsigned_gate_effective"] <= 0.20 + 1e-9, \ f"gate must be capped at 2× base (0.20), got {d['unsigned_gate_effective']}" # FIX 92: if the unsigned error is >= base gate (0.10) and the floor brackets, # the result is precision-limited: profile_reliable is False (boolean, NOT the # truthy string "indeterminate"), unsigned_gate_status == "precision_limited", # and the nested reconciliation.pass agrees (False). if unsigned is not None and unsigned >= 0.10: assert d["profile_reliable"] is False, \ f"unsigned={unsigned:.4f} >= base gate with bracketing floor must be " \ f"False (boolean), got {d['profile_reliable']!r}" assert d["unsigned_gate_status"] == "precision_limited" assert d["reconciliation_pass_basis"] == "indeterminate" assert d["reconciliation"]["pass"] is False, \ "nested pass must agree with boolean profile_reliable (FIX 83/92)" def test_gate_cap_enforced_at_2x(monkeypatch, tmp_path): """FIX 88: when the derived gate (p95 × 1.5) EXCEEDS the 2× cap, the effective gate is clamped to exactly 2× base (0.20) and status is 'capped'. This is the guard against an un-fireable gate (the 0.3535 case from v1.7.7).""" import matplotlib matplotlib.use("Agg") from gex.snapshot import _process_and_render import gex.snapshot as snap_mod import json, os from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo monkeypatch.setattr(snap_mod, "_load_recent_closes", lambda symbol, cfg: [99.0, 100.0, 101.0, 100.5, 100.2, 100.8]) # Floor = 0.30 -> derived = 0.45 -> capped at 0.20. monkeypatch.setattr(snap_mod, "reconciliation_floor_unsigned", lambda df, spot, cfg, **kw: { "method": "grid_rounding", "rounding_interval": 0.5e-4, "floor": 0.30, "brackets_gate": True}) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) later = today + timedelta(days=7) contracts = [] for exp in (today, later): for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 8000 if cp == "P" else 3000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 11, 0, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) _process_and_render("CAP", cfg, "am", "test_fix88cap", "2026-07-24 15:00:00", spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T15:01:00+00:00") with open(os.path.join(str(tmp_path), "CAP", "2026-07-24_am.json")) as f: d = json.load(f) # FIX 92: when the floor brackets AND the unsigned error >= base gate, the # precision_limited branch fires and overrides the status. Both "capped" and # "precision_limited" are valid here; the key assertion is that the gate IS 0.20. assert d["unsigned_gate_status"] in ("capped", "precision_limited"), \ f"expected capped or precision_limited, got {d['unsigned_gate_status']!r}" assert abs(d["unsigned_gate_effective"] - 0.20) < 1e-9, \ f"derived 0.45 must be capped at 0.20, got {d['unsigned_gate_effective']}" # --------------------------------------------------------------------------- # FIX 89: provenance stamp in the canonical key set # --------------------------------------------------------------------------- def test_provenance_stamp_in_every_artifact(tmp_path): """FIX 89: every artifact carries gex_version and schema_version, populated from gex/__init__.py. Both are part of the canonical key set (present regardless of instrument class or reconciliation outcome).""" import matplotlib matplotlib.use("Agg") from gex.snapshot import _process_and_render import gex.snapshot as snap_mod import gex import json, os from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo monkeypatch_free = snap_mod._load_recent_closes snap_mod._load_recent_closes = lambda s, c: [99.0, 100.0, 101.0, 100.5] try: cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) later = today + timedelta(days=7) contracts = [] for exp in (today, later): for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 8000 if cp == "P" else 3000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 11, 0, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) _process_and_render("PROV", cfg, "am", "test_fix89", "2026-07-24 15:00:00", spot, df_full, suffix="", instrument_class="equity_etf") with open(os.path.join(str(tmp_path), "PROV", "2026-07-24_am.json")) as f: d = json.load(f) assert d["gex_version"] == gex.__version__ assert d["schema_version"] == gex.__schema_version__ assert d["gex_version"] == "1.8.0" finally: snap_mod._load_recent_closes = monkeypatch_free # --------------------------------------------------------------------------- # FIX 90: excluded share + disambiguated pass basis # --------------------------------------------------------------------------- def test_excluded_share_and_basis_disambiguation(monkeypatch, tmp_path): """FIX 90: reconciliation_excluded_share discloses the share of Σ|GEX| excluded by the near-settlement guard (the pass covers 1 - share, not 100%), and reconciliation_pass_basis distinguishes 'both_pass' from a single-gate breach.""" import matplotlib matplotlib.use("Agg") from gex.snapshot import _process_and_render import gex.snapshot as snap_mod import json, os from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo monkeypatch.setattr(snap_mod, "_load_recent_closes", lambda symbol, cfg: [99.0, 100.0, 101.0, 100.5, 100.2, 100.8]) # Non-bracketing floor so the base 0.10 gate applies and a clean pass is possible. monkeypatch.setattr(snap_mod, "reconciliation_floor_unsigned", lambda df, spot, cfg, **kw: { "method": "grid_rounding", "rounding_interval": 0.5e-4, "floor": 0.002, "brackets_gate": False}) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) later = today + timedelta(days=7) contracts = [] for exp in (today, later): for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 8000 if cp == "P" else 3000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) # Capture at 15:54 ET so the 0DTE (today) expiry is near-settlement (< 30 min for # an equity/ETF settling at 16:00) and gets excluded — exercising the share field. snap_et = datetime(2026, 7, 24, 15, 54, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) _process_and_render("SHR", cfg, "pm", "test_fix90", "2026-07-24 19:54:00", spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T19:54:30+00:00") with open(os.path.join(str(tmp_path), "SHR", "2026-07-24_pm.json")) as f: d = json.load(f) # excluded_share is published and in [0, 1]; with a 0DTE excluded it should be > 0. assert "reconciliation_excluded_share" in d share = d["reconciliation_excluded_share"] assert 0.0 <= share <= 1.0 if d["reconciliation_scope"] == "excl_near_settlement": assert share > 0.0, "an excluded 0DTE must yield a positive excluded share" else: assert share == 0.0 # basis is one of the disambiguated values assert d["reconciliation_pass_basis"] in ( "both_pass", "signed", "unsigned", "signed+unsigned", "excluded", "indeterminate") # a clean pass must read "both_pass", not the old ambiguous "signed"/"unsigned" if d["profile_reliable"] is True: assert d["reconciliation_pass_basis"] == "both_pass" # --------------------------------------------------------------------------- # FIX 91: per-contract GEX uses spot^2, not strike^2 (regression test) # --------------------------------------------------------------------------- def test_per_contract_gex_uses_spot_squared(): """FIX 91: per-contract GEX must equal gamma x OI x spot^2 x M x 0.01, with spot pinned. The bug was that build_outlier_report used each strike's own K^2, which made the breakdown disagree with its parent bar by (K/S)^2.""" from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo from gex.compute import aggregate, build_outlier_report cfg = GexConfig() spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) exp = today + timedelta(days=7) # One call, one put, known gamma and OI. contracts = [ {"strike": 95.0, "expiry": exp, "cp": "C", "iv": 0.25, "oi": 1000, "volume": 100, "delta": 0.6, "gamma": 0.03, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, {"strike": 95.0, "expiry": exp, "cp": "P", "iv": 0.25, "oi": 800, "volume": 100, "delta": -0.4, "gamma": 0.03, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1}, ] snap_et = datetime(2026, 7, 24, 11, 0, tzinfo=ET) df = filter_contracts(contracts, spot, cfg, snap_et) agg = aggregate(df, spot, cfg) # Per-contract GEX = sign * |gamma| * OI * M * spot^2 * 0.01 M = cfg.contract_multiplier S2 = spot * spot call_gex = 1.0 * 0.03 * 1000 * M * S2 * 0.01 put_gex = -1.0 * 0.03 * 800 * M * S2 * 0.01 expected_net = call_gex + put_gex assert abs(agg.loc[95.0, "net_gex"] - expected_net) < 1e-6, \ f"bar net_gex must use spot^2: expected {expected_net}, got {agg.loc[95.0, 'net_gex']}" # The outlier breakdown must agree with the bar (FIX 91b invariant). report = build_outlier_report(agg, df, cfg, spot, top_n=5) top = report["top_strikes"][0] by_sum = sum(e["gex"] for e in top["by_expiry"]) assert abs(by_sum - top["net_gex"]) < max(1.0, abs(top["net_gex"]) * 1e-9), \ f"FIX 91b: sum(by_expiry[].gex)={by_sum} must equal net_gex={top['net_gex']}" # --------------------------------------------------------------------------- # FIX 91b: decomposition reconciles to its parent (invariant test) # --------------------------------------------------------------------------- def test_outlier_breakdown_reconciles_to_bar(synthetic_chain): """FIX 91b: for every entry in outlier_report.top_strikes, the sum of the per-expiry breakdown must equal the bar's net_gex within float tolerance. This is the invariant that would have caught the K^2 bug.""" from datetime import date cfg = GexConfig(strike_band=0.50) spot = 105.0 today = date(2026, 7, 24) df = filter_contracts(synthetic_chain, spot, cfg, today) agg = aggregate(df, spot, cfg) report = build_outlier_report(agg, df, cfg, spot, top_n=5) for entry in report["top_strikes"]: by_sum = sum(e["gex"] for e in entry["by_expiry"]) tol = max(1.0, abs(entry["net_gex"]) * 1e-9) assert abs(by_sum - entry["net_gex"]) < tol, \ f"strike {entry['strike']}: sum(by_expiry)={by_sum} != net_gex={entry['net_gex']}" # --------------------------------------------------------------------------- # FIX 92: profile_reliable is boolean-or-null; tri-state lives in unsigned_gate_status # --------------------------------------------------------------------------- def test_profile_reliable_is_boolean(monkeypatch, tmp_path): """FIX 92: profile_reliable must be True or False (boolean), never the truthy string 'indeterminate'. A consumer's plain `if pass:` must not read precision-limited as a pass. The tri-state semantics live in unsigned_gate_status.""" import matplotlib matplotlib.use("Agg") from gex.snapshot import _process_and_render import gex.snapshot as snap_mod import json, os from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo monkeypatch.setattr(snap_mod, "_load_recent_closes", lambda symbol, cfg: [99.0, 100.0, 101.0, 100.5, 100.2, 100.8]) # Force a bracketing floor so the precision_limited path fires. monkeypatch.setattr(snap_mod, "reconciliation_floor_unsigned", lambda df, spot, cfg, **kw: { "method": "grid_rounding", "rounding_interval": 0.5e-4, "floor": 0.15, "brackets_gate": True}) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) later = today + timedelta(days=7) contracts = [] for exp in (today, later): for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 9000 if cp == "P" else 1000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 11, 0, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) _process_and_render("BOOL", cfg, "am", "test_fix92", "2026-07-24 15:00:00", spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T15:01:00+00:00") with open(os.path.join(str(tmp_path), "BOOL", "2026-07-24_am.json")) as f: d = json.load(f) # profile_reliable must be a Python bool, never a string. assert isinstance(d["profile_reliable"], bool), \ f"profile_reliable must be bool, got {type(d['profile_reliable']).__name__}: {d['profile_reliable']!r}" # The nested reconciliation.pass must also be bool. assert isinstance(d["reconciliation"]["pass"], bool), \ f"reconciliation.pass must be bool, got {type(d['reconciliation']['pass']).__name__}" # If precision-limited fired, unsigned_gate_status carries the tri-state. if d["unsigned_gate_status"] == "precision_limited": assert d["profile_reliable"] is False assert d["reconciliation_pass_basis"] == "indeterminate" # --------------------------------------------------------------------------- # FIX 93: floor is deterministic (no n_draws, no median/p95 spread) # --------------------------------------------------------------------------- def test_floor_is_deterministic_scalar(): """FIX 93: reconciliation_floor_unsigned returns a single deterministic scalar, not a Monte Carlo distribution. The keys n_draws, median, p95 must NOT be present.""" from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo from gex.compute import reconciliation_floor_unsigned cfg = GexConfig() spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) contracts = [] for dte in (7, 30): exp = today + timedelta(days=dte) for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 9000 if cp == "P" else 1000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 11, 0, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) floor = reconciliation_floor_unsigned(df_full, spot, cfg) # Must have the deterministic keys. assert "floor" in floor assert "method" in floor and floor["method"] == "grid_rounding" assert "brackets_gate" in floor # Must NOT have the old Monte Carlo keys. assert "n_draws" not in floor, "n_draws must be removed (FIX 93)" assert "median" not in floor, "median must be removed (FIX 93)" assert "p95" not in floor, "p95 must be removed (FIX 93)" # Determinism: two calls give identical results. floor2 = reconciliation_floor_unsigned(df_full, spot, cfg) assert floor2["floor"] == floor["floor"] # --------------------------------------------------------------------------- # FIX 94: reconciliation_excluded_share uses net basis consistent with gex_by_expiry # --------------------------------------------------------------------------- def test_excluded_share_net_basis(monkeypatch, tmp_path): """FIX 94: reconciliation_excluded_share must use the same net basis as gex_by_expiry.share (numerator: Sigma|net GEX per excluded expiry|; denominator: Sigma|net GEX per strike|). Previously mixed gross numerator with gross denominator, inconsistent with gex_by_expiry.""" import matplotlib matplotlib.use("Agg") from gex.snapshot import _process_and_render import gex.snapshot as snap_mod import json, os from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo monkeypatch.setattr(snap_mod, "_load_recent_closes", lambda symbol, cfg: [99.0, 100.0, 101.0, 100.5, 100.2, 100.8]) cfg = GexConfig(outdir=str(tmp_path), cache_dir=str(tmp_path / "cache")) spot = 100.0 ET = ZoneInfo("America/New_York") today = date(2026, 7, 24) later = today + timedelta(days=7) contracts = [] for exp in (today, later): for k in (95.0, 100.0, 105.0): for cp in ("C", "P"): contracts.append({ "strike": k, "expiry": exp, "cp": cp, "iv": 0.25, "oi": 9000 if cp == "P" else 1000, "volume": 100, "delta": 0.5 if cp == "C" else -0.5, "gamma": 0.02, "vega": 0.1, "theta": -0.05, "theo": 5.0, "bid": 4.9, "ask": 5.1, }) snap_et = datetime(2026, 7, 24, 11, 0, tzinfo=ET) df_full = filter_contracts_full(contracts, spot, cfg, snap_et) _process_and_render("NET", cfg, "am", "test_fix94", "2026-07-24 15:00:00", spot, df_full, suffix="", instrument_class="equity_etf", captured_at_utc="2026-07-24T15:01:00+00:00") with open(os.path.join(str(tmp_path), "NET", "2026-07-24_am.json")) as f: d = json.load(f) # reconciliation_excluded_share must be a float in [0, 1]. share = d["reconciliation_excluded_share"] assert isinstance(share, float) assert 0.0 <= share <= 1.0, f"excluded_share must be in [0,1], got {share}" # gex_by_expiry must use "net_gex" (FIX 94 rename from sum_abs_gex). for row in d.get("gex_by_expiry", []): assert "net_gex" in row, f"gex_by_expiry row missing net_gex: {row.keys()}" assert "sum_abs_gex" not in row, "sum_abs_gex must be renamed to net_gex (FIX 94)"