### Setup Pre-collected Dataset

Source: https://github.com/jon-becker/prediction-market-analysis/blob/main/README.md

Downloads and extracts the pre-collected dataset from Cloudflare R2 Storage. This is a large download.

```bash
make setup
```

--------------------------------

### Install Dependencies with uv

Source: https://github.com/jon-becker/prediction-market-analysis/blob/main/README.md

Installs project dependencies using the uv package manager. Ensure Python 3.9+ is installed.

```bash
uv sync
```

--------------------------------

### Download Pre-collected Dataset

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Download the 36 GiB pre-collected dataset from Cloudflare R2 and extract it to the 'data/' directory using 'make setup'.

```bash
make setup
# Runs: scripts/download.sh
# Extracts data.tar.zst -> data/
```

--------------------------------

### Collect Market and Trade Data

Source: https://github.com/jon-becker/prediction-market-analysis/blob/main/README.md

Initiates the data collection process for market and trade data from prediction market APIs. An interactive menu will guide the selection of indexers. Data is saved to `data/kalshi/` and `data/polymarket/`.

```bash
make index
```

--------------------------------

### Using Kalshi Categories Utility

Source: https://github.com/jon-becker/prediction-market-analysis/blob/main/docs/ANALYSIS.md

Demonstrates how to use utility functions to group Kalshi markets into high-level categories and retrieve associated colors for visualizations. Import necessary functions and constants from the categories module.

```python
from src.analysis.kalshi.util.categories import get_group, get_hierarchy, GROUP_COLORS

# Get high-level group
group = get_group("NFLGAME")  # Returns "Sports"

# Get full hierarchy (group, category, subcategory)
hierarchy = get_hierarchy("NFLGAME")  # Returns ("Sports", "NFL", "Games")

# Use predefined colors for consistent visualizations
color = GROUP_COLORS["Sports"]  # Returns "#1f77b4"

```

--------------------------------

### Collect Market and Trade Data with CLI

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Initiate data collection for Kalshi and Polymarket markets and trades using the CLI. Progress is automatically checkpointed for safe interruption and resumption.

```bash
uv run main.py index
```

--------------------------------

### Discover and run indexers

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Use `Indexer.load()` to automatically discover all subclasses in `src/indexers/`. You can then instantiate and run specific indexers to collect and persist data.

```python
from src.common.indexer import Indexer

# Discover all indexers
indexers = Indexer.load()
for cls in indexers:
    instance = cls()
    print(f"{instance.name}: {instance.description}")
# kalshi_markets: Backfills Kalshi markets data to parquet files
# kalshi_trades:  Backfills Kalshi trades data to parquet files
# polymarket_trades: Backfills Polymarket trades from Polygon blockchain to parquet files

# Run a specific indexer
instance = indexers[0]()
instance.run()
```

--------------------------------

### Using Progress Indicator Context Manager

Source: https://github.com/jon-becker/prediction-market-analysis/blob/main/docs/ANALYSIS.md

Shows how to use the 'progress()' context manager for long-running operations. This displays a spinner to indicate activity during data loading or computation.

```python
def run(self) -> AnalysisOutput:
    with self.progress("Loading trades data"):
        df = con.execute("SELECT * FROM large_table").df()

    with self.progress("Computing aggregations"):
        # expensive computation
        result = df.groupby(...).agg(...)

```

--------------------------------

### Build Serializable Chart Configurations

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Creates serializable chart configurations for web rendering using factory functions. Use for generating various chart types like line, bar, scatter, and heatmap.

```python
from src.common.interfaces.chart import (
    ChartConfig, ChartType, UnitType, ScaleType, Series,
    line_chart, bar_chart, area_chart, pie_chart, scatter_chart, heatmap, treemap,
)

# Line chart with two series and a dashed reference line
data = [{"price": p, "win_rate": p * 0.97 + 1, "perfect": p} for p in range(1, 100)]
cfg = line_chart(
    data,
    x="price",
    y=["win_rate", "perfect"],
    title="Actual Win Rate vs Contract Price",
    strokeDasharrays=[None, "5 5"],
    yUnit=UnitType.PERCENT,
    xLabel="Price (cents)",
    yLabel="Win Rate (%)",
)
print(cfg.to_json())
# {"type": "line", "data": [...], "xKey": "price", "yKeys": ["win_rate", "perfect"], ...}

# Stacked bar chart
volume_data = [{"category": "Sports", "volume": 1200000}, {"category": "Politics", "volume": 980000}]
cfg2 = bar_chart(volume_data, x="category", y="volume", stacked=False, yUnit=UnitType.NUMBER)

# Scatter chart with named series
series = [
    Series(name="Taker",  data=[{"x": 25, "y": 0.8}, {"x": 50, "y": 1.2}]),
    Series(name="Maker",  data=[{"x": 25, "y": -0.3}, {"x": 50, "y": 0.4}]),
]
cfg3 = scatter_chart([], x="x", y="y", series=series, xScale=ScaleType.LINEAR, title="Returns")

# Treemap
tree_data = [{"name": "Sports", "value": 5000, "children": []}]
cfg4 = treemap(tree_data, name="name", value="value")

# Heatmap
heat_data = [{"hour": h, "day": d, "value": h * d} for h in range(24) for d in range(7)]
cfg5 = heatmap(heat_data, x="hour", y="day", value="value", title="Activity Heatmap")
```

--------------------------------

### Run Analysis Scripts with CLI

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Use the CLI to run analysis scripts interactively, by name, or in sequence. Outputs are saved to the 'output/' directory.

```bash
uv run main.py analyze
```

```bash
uv run main.py analyze win_rate_by_price
```

```bash
uv run main.py analyze all
# Output:
# Running: win_rate_by_price
#   png: output/win_rate_by_price.png
#   pdf: output/win_rate_by_price.pdf
#   csv: output/win_rate_by_price.csv
#   json: output/win_rate_by_price.json
```

--------------------------------

### Indexer.load()

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Discovers and loads all available Indexer subclasses from the `src/indexers/` directory.

```APIDOC
## `Indexer.load()` — Discover indexers

Abstract base class for indexers. Subclasses implement `run()` to fetch and persist data. `Indexer.load()` discovers all subclasses in `src/indexers/` automatically.

### Method
```python
Indexer.load() -> list[type[Indexer]]
```

### Response
Returns a list of Indexer subclasses.

### Request Example
```python
from src.common.indexer import Indexer

indexers = Indexer.load()
for cls in indexers:
    instance = cls()
    print(f"{instance.name}: {instance.description}")

# Example of running a specific indexer
if indexers:
    instance = indexers[0]()
    instance.run()
```
```

--------------------------------

### Analysis.progress()

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Displays a tqdm progress spinner to stderr during long-running operations within the `run()` method.

```APIDOC
## `Analysis.progress()` — Progress spinner context manager

Displays a tqdm spinner to stderr during long-running operations inside `run()`.

### Method
```python
with self.progress(message: str):
    # Long-running operation
```

### Parameters
- **message** (str) - The message to display next to the progress spinner.

### Request Example
```python
from src.common.analysis import Analysis, AnalysisOutput
import duckdb

class MyAnalysis(Analysis):
    def run(self) -> AnalysisOutput:
        con = duckdb.connect()

        with self.progress("Loading Kalshi trades (this may take a while)"):
            df = con.execute(
                "SELECT * FROM 'data/kalshi/trades/*.parquet'"
            ).df()

        with self.progress("Computing statistics"):
            stats = df.groupby("yes_price")["count"].sum().reset_index()

        return AnalysisOutput(data=stats)
```
```

--------------------------------

### Compress Data Directory with CLI

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Package the 'data/' directory into a zstd-compressed tar archive for storage or distribution.

```bash
uv run main.py package
# Output:
# Packaging data -> data.tar.zst
# Successfully created data.tar.zst
```

--------------------------------

### Basic Analysis Script Template

Source: https://github.com/jon-becker/prediction-market-analysis/blob/main/docs/ANALYSIS.md

A template for creating a new analysis script. It includes necessary imports, class structure, and a placeholder for data processing and visualization logic. Ensure to define the 'name' and 'description' attributes.

```python
"""Brief description of what this analysis does."""

from pathlib import Path

import duckdb
import matplotlib.pyplot as plt
from matplotlib.figure import Figure

from src.common.analysis import Analysis


class MyAnalysis(Analysis):
    name = "my_analysis"
    description = "Brief description of what this analysis does"

    def run(self) -> tuple[Figure, dict]:
        base_dir = Path(__file__).parent.parent.parent.parent
        kalshi_trades = base_dir / "data" / "kalshi" / "trades"
        kalshi_markets = base_dir / "data" / "kalshi" / "markets"

        con = duckdb.connect()
        df = con.execute(
            f"""
            SELECT yes_price, count, taker_side
            FROM '{kalshi_trades}/*.parquet'
            WHERE yes_price BETWEEN 1 AND 99
            LIMIT 1000
            """
        ).df()

        # Create visualization
        fig, ax = plt.subplots(figsize=(10, 6))
        ax.bar(df["yes_price"], df["count"])
        ax.set_xlabel("Price (cents)")
        ax.set_ylabel("Count")
        ax.set_title("My Analysis")
        plt.tight_layout()

        # Return figure and data dict (for CSV/JSON export)
        return fig, df.to_dict(orient="records")

```

--------------------------------

### Display progress spinner during long operations

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Use `self.progress()` as a context manager within the `run()` method to display a tqdm spinner. This is useful for indicating progress on long-running tasks like data loading or computation.

```python
from src.common.analysis import Analysis, AnalysisOutput
import duckdb
from pathlib import Path

class MyAnalysis(Analysis):
    def __init__(self):
        super().__init__("my_analysis", "Example with progress spinner")

    def run(self) -> AnalysisOutput:
        con = duckdb.connect()

        with self.progress("Loading Kalshi trades (this may take a while)"):
            df = con.execute(
                "SELECT * FROM 'data/kalshi/trades/*.parquet'"
            ).df()

        with self.progress("Computing statistics"):
            stats = df.groupby("yes_price")["count"].sum().reset_index()

        return AnalysisOutput(data=stats)
```

--------------------------------

### Query Parquet Files Directly with DuckDB

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Demonstrates querying Parquet files directly using DuckDB glob patterns for SQL analytics without loading data into memory. This is useful for analyzing large datasets stored in Parquet format.

```python
import duckdb
from pathlib import Path

con = duckdb.connect()

# --- Kalshi: Taker win rate by price bucket (calibration) ---
df_calibration = con.execute("""
    WITH resolved AS (
        SELECT ticker, result
        FROM 'data/kalshi/markets/*.parquet'
        WHERE status = 'finalized' AND result IN ('yes', 'no')
    )
    SELECT
        yes_price AS price,
        COUNT(*) AS trades,
        ROUND(100.0 * SUM(CASE WHEN taker_side = result THEN 1 ELSE 0 END) / COUNT(*), 2) AS taker_win_pct
    FROM 'data/kalshi/trades/*.parquet' t
    JOIN resolved USING (ticker)
    WHERE yes_price BETWEEN 1 AND 99
    GROUP BY yes_price
    ORDER BY yes_price
""").df()

# --- Kalshi: Maker vs taker returns by category ---
df_returns = con.execute(f"""
    WITH resolved AS (
        SELECT ticker, result
        FROM 'data/kalshi/markets/*.parquet'
        WHERE status = 'finalized' AND result IN ('yes', 'no')
    ),
    positions AS (
        SELECT
            CASE
                WHEN event_ticker IS NULL OR event_ticker = '' THEN 'independent'
                ELSE regexp_extract(event_ticker, '^([A-Z0-9]+)', 1)
            END AS category,
            -- Taker P&L (cents per contract)
            CASE WHEN taker_side = result
                 THEN 100 - CASE WHEN taker_side = 'yes' THEN yes_price ELSE no_price END
                 ELSE  -(CASE WHEN taker_side = 'yes' THEN yes_price ELSE no_price END)
            END AS taker_pnl,
            -- Maker P&L
            CASE WHEN taker_side != result
                 THEN 100 - CASE WHEN taker_side = 'yes' THEN no_price ELSE yes_price END
                 ELSE  -(CASE WHEN taker_side = 'yes' THEN no_price ELSE yes_price END)
            END AS maker_pnl
        FROM 'data/kalshi/trades/*.parquet' t
        JOIN 'data/kalshi/markets/*.parquet' m USING (ticker)
        JOIN resolved USING (ticker)
    )
    SELECT category, AVG(taker_pnl) AS avg_taker_return, AVG(maker_pnl) AS avg_maker_return
    FROM positions
    GROUP BY category
    ORDER BY avg_taker_return DESC
""").df()

# --- Polymarket: Daily volume in USD ---
df_poly_vol = con.execute("""
    SELECT
        DATE_TRUNC('day', b.timestamp::TIMESTAMP) AS day,
        SUM(t.maker_amount / 1e6) AS volume_usd
    FROM 'data/polymarket/trades/*.parquet' t
    JOIN 'data/polymarket/blocks/*.parquet' b USING (block_number)
    WHERE t.maker_asset_id = '0'   -- USDC side
    GROUP BY day
    ORDER BY day
""").df()

print(df_calibration.head())
print(df_poly_vol.tail())
```

--------------------------------

### Run Analysis Scripts

Source: https://github.com/jon-becker/prediction-market-analysis/blob/main/README.md

Executes analysis scripts to generate figures and statistics. An interactive menu allows selection of specific analyses or running all of them. Outputs are saved to the `output/` directory.

```bash
make analyze
```

--------------------------------

### Implement Custom Analysis Script

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Subclass the 'Analysis' base class to create custom analysis scripts. The 'run()' method should be overridden to perform the analysis, and the 'save()' method handles exporting outputs to various formats.

```python
from pathlib import Path
import duckdb
import matplotlib.pyplot as plt
import pandas as pd
from src.common.analysis import Analysis, AnalysisOutput
from src.common.interfaces.chart import ChartType, UnitType, ChartConfig, bar_chart

class VolumeByHourAnalysis(Analysis):
    def __init__(self):
        super().__init__(
            name="volume_by_hour",
            description="Total contracts traded by hour of day (Kalshi)",
        )
        self.trades_dir = Path("data/kalshi/trades")

    def run(self) -> AnalysisOutput:
        con = duckdb.connect()

        with self.progress("Querying trades"):
            df = con.execute(f"""
                SELECT
                    HOUR(created_time) AS hour,
                    SUM(count) AS total_contracts
                FROM '{self.trades_dir}/*.parquet'
                GROUP BY hour
                ORDER BY hour
            """).df()

        fig, ax = plt.subplots(figsize=(12, 5))
        ax.bar(df["hour"], df["total_contracts"], color="#4C72B0")
        ax.set_xlabel("Hour of Day (UTC)")
        ax.set_ylabel("Contracts Traded")
        ax.set_title("Kalshi Volume by Hour of Day")
        plt.tight_layout()

        chart = bar_chart(
            df.to_dict("records"),
            x="hour",
            y="total_contracts",
            title="Kalshi Volume by Hour of Day",
            yUnit=UnitType.NUMBER,
            xLabel="Hour (UTC)",
            yLabel="Contracts",
        )

        return AnalysisOutput(figure=fig, data=df, chart=chart)

# Save to output/ in all formats
analysis = VolumeByHourAnalysis()
saved = analysis.save("output/", formats=["png", "pdf", "csv", "json"])
# saved == {"png": Path("output/volume_by_hour.png"), "pdf": ..., "csv": ..., "json": ...}
```

--------------------------------

### Backfill Kalshi markets data

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

The `KalshiMarketsIndexer` fetches all market metadata from the Kalshi API, deduplicates records, and saves them as chunked Parquet files. It supports cursor-based resumption for interrupted backfills.

```python
from src.indexers.kalshi.markets import KalshiMarketsIndexer

# Fetch all markets (full backfill)
indexer = KalshiMarketsIndexer()
indexer.run()
# Resuming from cursor: abc123...
# Fetched 1000 markets (total: 1000, stored: 998)
# Fetched 1000 markets (total: 2000, stored: 1996)
# ...
# Backfill complete: 45231 markets fetched

# Restrict to markets closing within a time window
import datetime
start = int(datetime.datetime(2024, 1, 1).timestamp())
end   = int(datetime.datetime(2024, 12, 31).timestamp())
indexer = KalshiMarketsIndexer(min_close_ts=start, max_close_ts=end)
indexer.run()
```

--------------------------------

### Backfill Kalshi trades data

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

The `KalshiTradesIndexer` concurrently fetches trades for markets with volume ≥ 100, deduplicates by `trade_id`, and saves chunked Parquet files. It skips tickers already fully processed and allows configuration of concurrency and time windows.

```python
from src.indexers.kalshi.trades import KalshiTradesIndexer

# Full backfill with default 10 concurrent workers
indexer = KalshiTradesIndexer()
indexer.run()
# Found 198034 existing trades
# Found 42100 unique markets
# Skipped 41800 already processed, 300 to fetch
# Fetching trades: 100%|████| 300/300 [02:14<00:00]
# Backfill trades complete: 300 markets processed, 12450 trades saved

# Restrict by timestamp and increase parallelism
indexer = KalshiTradesIndexer(
    min_ts=1700000000,
    max_ts=1710000000,
    max_workers=20,
)
indexer.run()
```

--------------------------------

### SQL: Analyze Taker and Maker Positions

Source: https://github.com/jon-becker/prediction-market-analysis/blob/main/docs/ANALYSIS.md

Aggregates contract counts for both taker and maker positions across different price points. This provides a combined view of market participation.

```sql
WITH all_positions AS (
    -- Taker positions
    SELECT
        CASE WHEN taker_side = 'yes' THEN yes_price ELSE no_price END AS price,
        count,
        'taker' AS role
    FROM '{kalshi_trades}/*.parquet'

    UNION ALL

    -- Maker positions (counterparty)
    SELECT
        CASE WHEN taker_side = 'yes' THEN no_price ELSE yes_price END AS price,
        count,
        'maker' AS role
    FROM '{kalshi_trades}/*.parquet'
)
SELECT price, role, SUM(count) AS total_contracts
FROM all_positions
GROUP BY price, role
ORDER BY price

```

--------------------------------

### Create typed analysis output with figure, data, and chart

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

The `AnalysisOutput` dataclass is returned by `Analysis.run()`. It can hold a matplotlib figure, a pandas DataFrame, and a chart configuration. Fields are optional; populate only what your analysis produces.

```python
from src.common.analysis import AnalysisOutput
from src.common.interfaces.chart import line_chart
import matplotlib.pyplot as plt
import pandas as pd

fig, ax = plt.subplots()
ax.plot([1, 2, 3], [10, 20, 15])

df = pd.DataFrame({"x": [1, 2, 3], "y": [10, 20, 15]})

chart = line_chart(
    df.to_dict("records"),
    x="x",
    y="y",
    title="Example Line Chart",
)

output = AnalysisOutput(
    figure=fig,       # matplotlib Figure or FuncAnimation (for GIF)
    data=df,          # DataFrame -> saved as CSV
    chart=chart,      # ChartConfig -> saved as JSON
    metadata={"note": "example"},  # arbitrary dict, not exported
)
```

--------------------------------

### Analysis.save()

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Runs the analysis and saves the figure and data to multiple formats including png, pdf, svg, gif, csv, and json.

```APIDOC
## `Analysis.save()` — Export analysis outputs

Runs the analysis and saves the figure and data to multiple formats. Supported formats: `png`, `pdf`, `svg`, `gif` (for `FuncAnimation`), `csv`, `json`.

### Method
```python
Analysis.save(output_dir: str, formats: list[str], dpi: int = 150)
```

### Parameters
- **output_dir** (str) - The directory to save the output files.
- **formats** (list[str]) - A list of desired output formats.
- **dpi** (int, optional) - The resolution in dots per inch for image formats. Defaults to 150.

### Request Example
```python
from src.common.analysis import Analysis

analyses = Analysis.load()

for cls in analyses:
    instance = cls()
    saved = instance.save(
        output_dir="output/",
        formats=["png", "pdf", "csv", "json"],
        dpi=300,
    )
    for fmt, path in saved.items():
        print(f"  {fmt} -> {path}")
```

### Response
Returns a dictionary mapping each requested format to its saved file path.
```

--------------------------------

### KalshiMarketsIndexer

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Fetches and stores Kalshi markets data. It paginates the Kalshi API, deduplicates records, and writes chunked Parquet files. Supports cursor-based resumption and filtering by closing time.

```APIDOC
## `KalshiMarketsIndexer` — Fetch and store Kalshi markets

Paginates the Kalshi API to collect all market metadata, deduplicates records, and writes chunked Parquet files to `data/kalshi/markets/`. Supports cursor-based resumption.

### Method
```python
KalshiMarketsIndexer(min_close_ts: int = None, max_close_ts: int = None)
```

### Parameters
- **min_close_ts** (int, optional) - Minimum closing timestamp for markets to fetch.
- **max_close_ts** (int, optional) - Maximum closing timestamp for markets to fetch.

### Request Example (Full Backfill)
```python
from src.indexers.kalshi.markets import KalshiMarketsIndexer

indexer = KalshiMarketsIndexer()
indexer.run()
```

### Request Example (Filtered by Time)
```python
import datetime
start = int(datetime.datetime(2024, 1, 1).timestamp())
end   = int(datetime.datetime(2024, 12, 31).timestamp())
indexer = KalshiMarketsIndexer(min_close_ts=start, max_close_ts=end)
indexer.run()
```
```

--------------------------------

### Chunked Parquet Writer with Deduplication

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Manages writing market records to Parquet files, automatically deduplicating by ticker and splitting chunks. Use to efficiently store and manage market data.

```python
from src.common.storage import ParquetStorage
from pathlib import Path
from dataclasses import dataclass
from datetime import datetime

@dataclass
class Market:
    ticker: str
    title: str
    status: str
    volume: int

storage = ParquetStorage(data_dir=Path("data/kalshi/markets"))

markets = [
    Market(ticker="PRES-2024-DJT", title="Trump wins presidency", status="finalized", volume=500000),
    Market(ticker="BTCMAX-2024",   title="BTC above 100k in 2024", status="finalized", volume=120000),
]

total_stored = storage.append_markets(markets)
print(f"Total unique markets stored: {total_stored}")
# Total unique markets stored: 2

# Re-inserting the same tickers is a no-op (deduplication)
total_stored = storage.append_markets(markets)
print(f"After duplicate insert: {total_stored}")
# After duplicate insert: 2
```

--------------------------------

### Package Data Directory

Source: https://github.com/jon-becker/prediction-market-analysis/blob/main/README.md

Compresses the `data/` directory into a zstd-compressed tar archive (`data.tar.zst`) and then removes the original `data/` directory. This is useful for storage or distribution.

```bash
make package
```

--------------------------------

### Save analysis outputs to multiple formats

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Use `Analysis.save()` to export analysis figures and data. Supported formats include 'png', 'pdf', 'svg', 'gif', 'csv', and 'json'. Specify the output directory and desired formats.

```python
from src.common.analysis import Analysis

# Load all available analyses from src/analysis/
analyses = Analysis.load()  # returns list[type[Analysis]]

for cls in analyses:
    instance = cls()
    print(f"{instance.name}: {instance.description}")
    saved = instance.save(
        output_dir="output/",
        formats=["png", "pdf", "csv", "json"],
        dpi=300,
    )
    for fmt, path in saved.items():
        print(f"  {fmt} -> {path}")
```

--------------------------------

### KalshiTradesIndexer

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Fetches and stores Kalshi trades data concurrently. It processes trades for markets with volume >= 100, deduplicates by trade ID, and writes chunked Parquet files. It skips already processed tickers and supports filtering by timestamp and adjusting concurrency.

```APIDOC
## `KalshiTradesIndexer` — Fetch and store Kalshi trades

Concurrently fetches all trades for every market with volume ≥ 100 using a thread pool, deduplicates by `trade_id`, and writes chunked Parquet files to `data/kalshi/trades/`. Skips tickers already fully processed.

### Method
```python
KalshiTradesIndexer(min_ts: int = None, max_ts: int = None, max_workers: int = 10)
```

### Parameters
- **min_ts** (int, optional) - Minimum timestamp for trades to fetch.
- **max_ts** (int, optional) - Maximum timestamp for trades to fetch.
- **max_workers** (int, optional) - The maximum number of worker threads to use for concurrent fetching. Defaults to 10.

### Request Example (Full Backfill)
```python
from src.indexers.kalshi.trades import KalshiTradesIndexer

indexer = KalshiTradesIndexer()
indexer.run()
```

### Request Example (Filtered and Parallelized)
```python
indexer = KalshiTradesIndexer(
    min_ts=1700000000,
    max_ts=1710000000,
    max_workers=20,
)
indexer.run()
```
```

--------------------------------

### Kalshi Market Categorization Utilities

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Maps Kalshi event tickers to a hierarchical categorization (Group, Category, Subcategory) and provides consistent colors for visualizations. Use for organizing and analyzing market data.

```python
from src.analysis.kalshi.util.categories import get_group, get_hierarchy, GROUP_COLORS, CATEGORY_SQL

# High-level group
print(get_group("NFLGAME-2024-KC"))     # "Sports"
print(get_group("BTCD-2024-01-15"))     # "Crypto"
print(get_group("FEDDECISION-2024"))    # "Finance"
print(get_group("PRES-2024-DJT"))       # "Politics"
print(get_group("UNKNOWN-XYZ"))         # "Other"

# Full three-level hierarchy
print(get_hierarchy("NFLGAME-2024-KC"))    # ("Sports", "NFL", "Games")
print(get_hierarchy("BTCMAXY-2024"))       # ("Crypto", "Bitcoin", "Max Yearly")
print(get_hierarchy("FEDDECISION-2024"))   # ("Finance", "Fed", "Decisions")
```

--------------------------------

### AnalysisOutput

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

A typed container for analysis outputs, returned by `Analysis.run()`. It can hold figures, dataframes, chart configurations, and metadata.

```APIDOC
## `AnalysisOutput` — Typed output container

Dataclass returned by `Analysis.run()`. All fields are optional — only populate what your analysis produces.

### Fields
- **figure** (matplotlib.Figure or FuncAnimation, optional) - The matplotlib figure or animation to be saved.
- **data** (pandas.DataFrame, optional) - The data to be saved as CSV.
- **chart** (ChartConfig, optional) - The chart configuration to be saved as JSON.
- **metadata** (dict, optional) - Arbitrary metadata, not exported.

### Request Example
```python
from src.common.analysis import AnalysisOutput
from src.common.interfaces.chart import line_chart
import matplotlib.pyplot as plt
import pandas as pd

fig, ax = plt.subplots()
ax.plot([1, 2, 3], [10, 20, 15])

df = pd.DataFrame({"x": [1, 2, 3], "y": [10, 20, 15]})

chart = line_chart(
    df.to_dict("records"),
    x="x",
    y="y",
    title="Example Line Chart",
)

output = AnalysisOutput(
    figure=fig,
    data=df,
    chart=chart,
    metadata={"note": "example"},
)
```
```

--------------------------------

### SQL: Join Trades with Resolved Kalshi Markets

Source: https://github.com/jon-becker/prediction-market-analysis/blob/main/docs/ANALYSIS.md

Combines trade data with market outcomes for finalized Kalshi markets. This query helps determine if the taker won their trade based on the market's result.

```sql
WITH resolved_markets AS (
    SELECT ticker, result
    FROM '{kalshi_markets}/*.parquet'
    WHERE status = 'finalized'
      AND result IN ('yes', 'no')
)
SELECT
    t.yes_price,
    t.count,
    t.taker_side,
    m.result,
    CASE WHEN t.taker_side = m.result THEN 1 ELSE 0 END AS taker_won
FROM '{kalshi_trades}/*.parquet' t
INNER JOIN resolved_markets m ON t.ticker = m.ticker

```

--------------------------------

### Fetch Polymarket Trades from Polygon Blockchain

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Scans the Polygon blockchain for OrderFilled events. Saves progress to a cursor file for resuming interrupted runs. Use for full backfills or specific block ranges.

```python
from src.indexers.polymarket.trades import PolymarketTradesIndexer

# Full backfill from Polymarket genesis block
indexer = PolymarketTradesIndexer()
indexer.run()
# Resuming from block 23456789
# Fetching trades from block 23456789 to 68901234
# Total blocks: 45,444,445
# Backfilling: 45445it [1:23:10,  9.12 chunks/s, block=68901234, saved=2100000]
# Backfill complete: 2134567 trades saved

# Backfill a specific block range
indexer = PolymarketTradesIndexer(
    from_block=60_000_000,
    to_block=65_000_000,
    chunk_size=500,   # blocks per RPC call
)
indexer.run()
```

--------------------------------

### Extract Category from Event Ticker with DuckDB

Source: https://context7.com/jon-becker/prediction-market-analysis/llms.txt

Uses a SQL query with DuckDB to extract a category from an event ticker using a predefined SQL pattern. Requires the CATEGORY_SQL variable to be defined.

```python
import duckdb
df = duckdb.execute(f"""
    SELECT
        {CATEGORY_SQL} AS category,
        COUNT(*) AS market_count,
        SUM(volume) AS total_volume
    FROM 'data/kalshi/markets/*.parquet'
    GROUP BY category
    ORDER BY total_volume DESC
    LIMIT 10
""").df()
print(df)
```

--------------------------------

### SQL: Extract Category from Event Ticker

Source: https://github.com/jon-becker/prediction-market-analysis/blob/main/docs/ANALYSIS.md

Extracts a category identifier from the 'event_ticker' field in Kalshi market data. Handles null or empty tickers by assigning them to an 'independent' category.

```sql
SELECT
    CASE
        WHEN event_ticker IS NULL OR event_ticker = '' THEN 'independent'
        ELSE regexp_extract(event_ticker, '^([A-Z0-9]+)', 1)
    END AS category,
    COUNT(*) AS market_count
FROM '{kalshi_markets}/*.parquet'
GROUP BY category

```

--------------------------------

### FPMM Collateral Lookup Schema

Source: https://github.com/jon-becker/prediction-market-analysis/blob/main/docs/SCHEMAS.md

This JSON structure maps FPMM contract addresses to their collateral token details. It is used to identify markets collateralized by specific tokens like USDC.

```json
{
  "0x1234...": {
    "collateral_address": "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174",
    "collateral_symbol": "USDC"
  }
}
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.