"""Master analysis pipeline for the Polymarket Arrow-Debreu paper. Usage: python scripts/run_analysis.py [--skip-data] [--skip-models] [--skip-paper] """ from __future__ import annotations import argparse import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # make `src` importable import numpy as np import pandas as pd from src.config import ( DATA_PROCESSED, DATA_RAW_POLY, ELECTION_KEY_DATES, FOMC_DATES, CPI_RELEASE_DATES, ) from src.utils import ensure_dirs, setup_logging logger = setup_logging("analysis") # ── Phase 1: Data acquisition ──────────────────────────────────────────────── def run_data_acquisition() -> None: import asyncio from src.api.runner import run_all run_all() # ── Phase 2: Cleaning ──────────────────────────────────────────────────────── def run_cleaning() -> None: from src.cleaning.polymarket_cleaner import clean_all as clean_poly from src.cleaning.macro_cleaner import clean_all as clean_macro from src.cleaning.yahoo_cleaner import clean_all as clean_yahoo clean_poly() clean_macro() clean_yahoo() # ── Phase 3: Models + Figures + Tables ──────────────────────────────────────── def run_models_and_outputs() -> None: from src.models.arrow_debreu import ( test_market_completeness, extract_risk_neutral_density, compute_risk_neutral_moments, detect_arbitrage_violations, state_price_dynamics, ) from src.models.event_study import run_event_study from src.models.price_discovery import granger_causality_matrix, estimate_var from src.models.volatility import ( compute_realized_volatility_measures, estimate_har_rv, estimate_garch, ) from src.models.liquidity import analyze_volume_patterns, market_quality_from_orderbook from src.features.engineering import compute_roll_spread from src.plots.figures import ( fig_state_prices, fig_completeness_test, fig_risk_neutral_density, fig_event_study_cars, fig_granger_heatmap, fig_impulse_response, fig_volatility, fig_liquidity, fig_election_case_study, fig_arbitrage_violations, ) from src.plots.tables import ( table_descriptive_stats, table_summary_stats, table_completeness_test, table_event_study, table_granger_causality, table_var_results, table_garch, table_har_rv, table_liquidity, ) from src.plots.style import save_figure # ── Load cleaned data ───────────────────────────────────────────────── logger.info("Loading cleaned data...") fed_state = pd.read_parquet(DATA_PROCESSED / "fed_june2026_state_prices.parquet") fed_panel = pd.read_parquet(DATA_PROCESSED / "fed_june2026_panel.parquet") election = pd.read_parquet(DATA_PROCESSED / "election_2024_panel.parquet") financial = pd.read_parquet(DATA_PROCESSED / "financial_data.parquet") macro_events = pd.read_parquet(DATA_PROCESSED / "macro_events.parquet") state_cols = [c for c in fed_state.columns if c not in ("state_price_sum", "completeness_deviation")] state_prices = fed_state[state_cols] # ── 1. Arrow-Debreu completeness tests ──────────────────────────────── logger.info("=== Arrow-Debreu Completeness Tests ===") completeness = test_market_completeness(state_prices) logger.info(" Mean sum: %.4f, Std: %.4f, All positive: %s", completeness.mean_sum, completeness.std_sum, completeness.all_positive) state_values = [-50.0, -25.0, 0.0, 25.0, 50.0] rn_density = extract_risk_neutral_density(state_prices, state_values) rn_moments = compute_risk_neutral_moments(rn_density, state_values) arb_violations = detect_arbitrage_violations(state_prices, market_type="state_space") dynamics = state_price_dynamics(state_prices) # Figures fig = fig_state_prices(state_prices) save_figure(fig, "fig1_state_prices") logger.info(" Saved fig1_state_prices") fig = fig_completeness_test(completeness.daily_sums) save_figure(fig, "fig2_completeness_test") logger.info(" Saved fig2_completeness_test") fig = fig_risk_neutral_density(rn_density, state_values) save_figure(fig, "fig3_risk_neutral_density") logger.info(" Saved fig3_risk_neutral_density") fig = fig_arbitrage_violations(arb_violations) save_figure(fig, "fig12_arbitrage_violations") logger.info(" Saved fig12_arbitrage_violations") # Tables table_completeness_test(completeness) logger.info(" Saved table3_completeness") table_summary_stats(state_prices) logger.info(" Saved table2_summary_stats") # ── 2. Event study ──────────────────────────────────────────────────── logger.info("=== Event Study ===") fomc_dates_in_range = [d for d in FOMC_DATES if fed_state.index.min().date() <= d <= fed_state.index.max().date()] cpi_dates_in_range = [d for d in CPI_RELEASE_DATES if fed_state.index.min().date() <= d <= fed_state.index.max().date()] event_results = {} if fomc_dates_in_range: es_fomc = run_event_study(state_prices, fomc_dates_in_range, "FOMC", window=(-3, 3)) event_results["FOMC"] = es_fomc logger.info(" FOMC: %d events", es_fomc.n_events) if cpi_dates_in_range: es_cpi = run_event_study(state_prices, cpi_dates_in_range, "CPI", window=(-3, 3)) event_results["CPI"] = es_cpi logger.info(" CPI: %d events", es_cpi.n_events) # Election event study with broader window election_dates = list(ELECTION_KEY_DATES.values()) election_dates_in_range = [d for d in election_dates if election.index.min().date() <= d <= election.index.max().date()] if election_dates_in_range and "Donald Trump" in election.columns: es_election = run_event_study( election[["Donald Trump"]], election_dates_in_range, "Election", window=(-5, 5), ) event_results["Election"] = es_election logger.info(" Election: %d events", es_election.n_events) if event_results: cars_for_plot = {} for etype, result in event_results.items(): if not result.test_stats.empty: cars_for_plot[etype] = result.test_stats if cars_for_plot: fig = fig_event_study_cars(cars_for_plot, window=(-3, 3)) save_figure(fig, "fig5_event_study_cars") logger.info(" Saved fig5_event_study_cars") table_event_study(event_results) logger.info(" Saved table4_event_study") # ── 3. Price discovery ──────────────────────────────────────────────── logger.info("=== Price Discovery ===") no_change = state_prices.get("No change") if no_change is not None and "SPY_log_return" in financial.columns: aligned = pd.DataFrame({ "Fed_NoChange": no_change.pct_change(), "SPY": financial["SPY_log_return"], "BTC": financial.get("BTC_USD_log_return", pd.Series(dtype=float)), "VIX": financial.get("VIX_log_return", pd.Series(dtype=float)), }).dropna() if len(aligned) > 20: f_stats, p_values = granger_causality_matrix(aligned, max_lag=5) fig = fig_granger_heatmap(p_values) save_figure(fig, "fig7_granger_heatmap") logger.info(" Saved fig7_granger_heatmap") table_granger_causality(f_stats, p_values) logger.info(" Saved table5_granger") var_result = estimate_var(aligned, max_lags=5) if var_result["model"] is not None: table_var_results(var_result) logger.info(" Saved table6_var") if var_result["irf"]: fig = fig_impulse_response(var_result["irf"]) save_figure(fig, "fig8_impulse_response") logger.info(" Saved fig8_impulse_response") else: logger.warning(" Not enough aligned data for price discovery (%d rows)", len(aligned)) else: logger.warning(" Missing data for price discovery analysis") # ── 4. Volatility ───────────────────────────────────────────────────── logger.info("=== Volatility Analysis ===") if no_change is not None: returns = no_change.pct_change().dropna() if len(returns) > 30: rv_measures = compute_realized_volatility_measures(returns) # HAR-RV try: har_result = estimate_har_rv(rv_measures) table_har_rv(har_result) logger.info(" HAR-RV R²: %.4f", har_result.r_squared) except Exception as e: logger.warning(" HAR-RV failed: %s", e) har_result = None # GARCH garch_results = {} for mtype in ["GARCH", "GJR-GARCH"]: try: garch_results[mtype] = estimate_garch(returns, model_type=mtype) logger.info(" %s AIC: %.2f", mtype, garch_results[mtype].aic) except Exception as e: logger.warning(" %s failed: %s", mtype, e) if garch_results: table_garch(garch_results) logger.info(" Saved table7_garch") first_garch = list(garch_results.values())[0] fig = fig_volatility( rv_measures["rv_5"], cond_vol=first_garch.conditional_volatility, ) save_figure(fig, "fig9_volatility") logger.info(" Saved fig9_volatility") # ── 5. Liquidity ────────────────────────────────────────────────────── logger.info("=== Liquidity Analysis ===") # Use volume from discovered markets metadata discovered_path = DATA_RAW_POLY / "discovered_markets.json" if discovered_path.exists(): with open(discovered_path) as f: discovered = json.load(f) volumes = pd.DataFrame([{ "name": m.get("question", "")[:40], "volume": float(m.get("volumeNum", 0)), "liquidity": float(m.get("liquidityNum", 0)), } for m in discovered if float(m.get("volumeNum", 0)) > 0]) if not volumes.empty: metadata = pd.DataFrame([{ "name": m.get("question", "")[:40], "volume": float(m.get("volumeNum", 0)), "liquidity": float(m.get("liquidityNum", 0)), "n_obs": 0, "mean_price": np.nan, "std_price": np.nan, } for m in discovered[:20]]) table_descriptive_stats(metadata) logger.info(" Saved table1_descriptive_stats") # Roll spread for Fed panel if no_change is not None: roll_spread = compute_roll_spread(no_change) fig = fig_liquidity( no_change.rename("Price"), spreads=roll_spread, ) save_figure(fig, "fig10_liquidity") logger.info(" Saved fig10_liquidity") # ── 6. Election case study ──────────────────────────────────────────── logger.info("=== Election Case Study ===") if "Donald Trump" in election.columns: trump = election["Donald Trump"] spy_ret = financial.get("SPY_log_return") fig = fig_election_case_study(trump, spy_returns=spy_ret) save_figure(fig, "fig11_election_case_study") logger.info(" Saved fig11_election_case_study") logger.info("=== All analysis complete ===") # ── Main ────────────────────────────────────────────────────────────────────── def main() -> None: parser = argparse.ArgumentParser(description="Polymarket Arrow-Debreu Analysis Pipeline") parser.add_argument("--skip-data", action="store_true", help="Skip data acquisition") parser.add_argument("--skip-models", action="store_true", help="Skip models/figures/tables") parser.add_argument("--skip-paper", action="store_true", help="Skip paper compilation") args = parser.parse_args() ensure_dirs() if not args.skip_data: logger.info("Phase 1: Data acquisition") run_data_acquisition() logger.info("Phase 2: Cleaning") run_cleaning() if not args.skip_models: logger.info("Phase 3-4: Models, figures, and tables") run_models_and_outputs() if not args.skip_paper: logger.info("Phase 5: Paper compilation") import subprocess result = subprocess.run( ["latexmk", "-pdf", "-interaction=nonstopmode", "main.tex"], cwd="paper", capture_output=True, text=True, ) if result.returncode == 0: logger.info("Paper compiled successfully") else: logger.warning("Paper compilation had issues (check paper/main.log)") if __name__ == "__main__": main()