"""Clean Polymarket raw data and build structured panels for analysis.""" from __future__ import annotations import json from pathlib import Path import numpy as np import pandas as pd from src.config import ( DATA_PROCESSED, DATA_RAW_POLY, ELECTION_2024_EVENT_ID, FED_JUNE_2026_EVENT_ID, ) from src.utils import setup_logging logger = setup_logging("polymarket_cleaner") # --------------------------------------------------------------------------- # Low-level loaders # --------------------------------------------------------------------------- def load_event_markets(event_id: int) -> list[dict]: """Load event JSON and extract market info with token IDs.""" path = DATA_RAW_POLY / f"event_{event_id}.json" if not path.exists(): logger.warning("Event file not found: %s", path) return [] with open(path) as f: event_data = json.load(f) results: list[dict] = [] for m in event_data.get("markets", []): tokens = json.loads(m.get("clobTokenIds", "[]")) if tokens: results.append({ "id": m["id"], "name": m.get("groupItemTitle", m.get("question", "")[:40]), "question": m.get("question", ""), "yes_token": tokens[0], "no_token": tokens[1] if len(tokens) > 1 else "", "closed": m.get("closed", False), "prices": m.get("outcomePrices", ""), "volume": float(m.get("volumeNum", m.get("volume", 0) or 0)), }) logger.info("Loaded %d markets from event %d", len(results), event_id) return results def load_price_history(token_id: str) -> pd.DataFrame: """Load price history JSON for a token, return DataFrame with DatetimeIndex. Files are named ``price_history_{token_id[:16]}.json``. Content is a list of ``{"t": unix_ts, "p": float}``. Returns a DataFrame with a datetime index (``date``) and a ``price`` column. """ short_id = token_id[:16] path = DATA_RAW_POLY / f"price_history_{short_id}.json" if not path.exists(): logger.warning("Price history file not found: %s", path) return pd.DataFrame(columns=["price"]) with open(path) as f: raw = json.load(f) if not raw: logger.warning("Empty price history for token %s", short_id) return pd.DataFrame(columns=["price"]) df = pd.DataFrame(raw) # The CLOB daily point is stamped at 00:00 UTC of the *following* day and # represents the close of the preceding UTC day; label it by the day it # closes (revision 2026-10-17; the earlier code labelled it one day late). df["date"] = (pd.to_datetime(df["t"], unit="s", utc=True) - pd.Timedelta(minutes=5)).dt.normalize() df["price"] = df["p"].astype(float) # Some data may have duplicates for the same day (intra-day ticks); # keep the last observation per calendar day. df = df.drop_duplicates(subset="date", keep="last") df = df.set_index("date").sort_index()[["price"]] df.index = df.index.tz_localize(None) # drop tz for easier merging logger.info("Loaded price history for %s: %d observations", short_id, len(df)) return df # --------------------------------------------------------------------------- # Panel builders # --------------------------------------------------------------------------- def build_market_panel(event_id: int) -> pd.DataFrame: """Build panel of prices for all markets in an event. Columns = market names (``groupItemTitle``), Index = date, Values = prices. """ markets = load_event_markets(event_id) if not markets: logger.warning("No markets found for event %d", event_id) return pd.DataFrame() series_dict: dict[str, pd.Series] = {} for m in markets: df = load_price_history(m["yes_token"]) if not df.empty: series_dict[m["name"]] = df["price"] if not series_dict: logger.warning("No price data for event %d", event_id) return pd.DataFrame() panel = pd.DataFrame(series_dict) panel = panel.sort_index() # Forward-fill small gaps (weekends / missing days) panel = panel.ffill(limit=3) logger.info( "Built market panel for event %d: %d dates x %d markets", event_id, len(panel), len(panel.columns), ) return panel def build_arrow_debreu_state_prices(event_id: int = FED_JUNE_2026_EVENT_ID) -> pd.DataFrame: """Build Arrow-Debreu state-price matrix for the Fed June 2026 neg-risk group. Each market corresponds to one Arrow-Debreu state (50+ bps decrease, 25 bps decrease, no change, 25 bps increase, 50+ bps increase). The state price equals the market YES price. Adds: ``state_price_sum`` -- sum of state prices per day (should be close to 1.0). ``completeness_deviation`` -- ``state_price_sum - 1.0``. """ panel = build_market_panel(event_id) if panel.empty: return panel # Canonical ordering of the state space canonical_order = [ "50+ bps decrease", "25 bps decrease", "No change", "25 bps increase", "50+ bps increase", ] # Reorder columns to canonical order, keeping only those present cols = [c for c in canonical_order if c in panel.columns] panel = panel[cols] panel["state_price_sum"] = panel[cols].sum(axis=1) panel["completeness_deviation"] = panel["state_price_sum"] - 1.0 logger.info( "Arrow-Debreu state prices: mean sum = %.4f, mean |deviation| = %.4f", panel["state_price_sum"].mean(), panel["completeness_deviation"].abs().mean(), ) return panel def build_election_panel() -> pd.DataFrame: """Build panel for Election 2024 markets (Trump, Biden/Harris, etc.).""" return build_market_panel(ELECTION_2024_EVENT_ID) # --------------------------------------------------------------------------- # Master cleaning # --------------------------------------------------------------------------- def clean_all() -> None: """Run all Polymarket cleaning and save Parquet files to ``data/processed/``.""" DATA_PROCESSED.mkdir(parents=True, exist_ok=True) # 1. Fed June 2026 -- Arrow-Debreu state prices fed_panel = build_arrow_debreu_state_prices() if not fed_panel.empty: out = DATA_PROCESSED / "fed_june2026_state_prices.parquet" fed_panel.to_parquet(out) logger.info("Saved %s (%d rows)", out.name, len(fed_panel)) # 2. Fed June 2026 -- raw market panel (without AD columns) fed_raw = build_market_panel(FED_JUNE_2026_EVENT_ID) if not fed_raw.empty: out = DATA_PROCESSED / "fed_june2026_panel.parquet" fed_raw.to_parquet(out) logger.info("Saved %s (%d rows)", out.name, len(fed_raw)) # 3. Election 2024 panel election = build_election_panel() if not election.empty: out = DATA_PROCESSED / "election_2024_panel.parquet" election.to_parquet(out) logger.info("Saved %s (%d rows)", out.name, len(election)) logger.info("=== Polymarket cleaning complete ===")