pmlab

A generic, plugin-based ML framework for Polymarket prediction markets.

Installation · Quickstart · Architecture · Writing a Plugin · CLI · API Reference · Roadmap · Contributing

What is pmlab?

pmlab is a Python library that lets you build, backtest, and paper-trade machine learning models for any Polymarket prediction market — weather, sports, politics, crypto, or anything else.

The framework is completely domain-agnostic: you write a small plugin that tells pmlab how to find markets, generate features, and resolve outcomes. pmlab handles everything else — edge calculation, walk-forward backtesting, champion promotion with a hard gate, paper trading with stale-signal guards, and settlement.

Battle-tested design — the architecture is extracted from two live production systems:

• polymarket-tmax-lab — weather/temperature markets (active since 2025)
• f1-polymarket-lab — Formula 1 race outcome markets

Installation

pip install pmlab

Or with uv:

uv add pmlab

Requirements: Python 3.12+

Quickstart

1. Implement a plugin

from pmlab import MarketPlugin, MarketSpec, OutcomeBin

class MyPlugin(MarketPlugin):
    family = "my_markets"

    def discover_markets(self, **kwargs) -> list[MarketSpec]:
        return [...]

    def fetch_features(self, spec: MarketSpec, horizon: str, **kwargs) -> dict[str, float]:
        return {"feature_trend": 0.65, "feature_volume": 1200.0}

    def fetch_truth(self, spec: MarketSpec, **kwargs) -> float | str | None:
        return 31.5

    def build_training_row(self, spec: MarketSpec, horizon: str, **kwargs) -> dict | None:
        features = self.fetch_features(spec, horizon)
        truth = self.fetch_truth(spec)
        if truth is None:
            return None
        winning_label = spec.resolve_winning_bin(float(truth))
        return {"market_id": spec.market_id, "decision_horizon": horizon,
                "winning_label": winning_label, "market_price": 0.3, **features}

2. Run a backtest

import pandas as pd
from pmlab import LGBMForecaster
from pmlab.backtest.rolling_origin import rolling_origin_eval

panel = pd.read_parquet("data/historical_panel.parquet")
model = LGBMForecaster()
result = rolling_origin_eval(panel, model, stride=30, min_train_rows=100)

print(f"Total trades: {len(result.trades)}")
print(f"Total PnL:    {result.trades['realized_pnl'].sum():.2f}")
print(f"Hit rate:     {(result.trades['realized_pnl'] > 0).mean():.1%}")

# The trade log composes straight into metrics and the go/no-go gate —
# no manual column wrangling.
from pmlab import compute_metrics
from pmlab.backtest.holdout_gate import HoldoutGateResult

metrics = compute_metrics(result.trades)          # derives outcome from PnL
gate = HoldoutGateResult.evaluate(result.trades)  # aggregate GO / NO_GO
print(metrics, "->", gate.decision)

3. Evaluate calibration

from pmlab import brier_decomposition, reliability_data

# Murphy (1973) decomposition: BS = reliability - resolution + uncertainty
result = brier_decomposition(y_true, y_prob)
print(f"Brier score:  {result.brier_score:.4f}")
print(f"Skill score:  {result.skill_score:.4f}")
print(f"Reliability:  {result.reliability:.4f}")
print(f"Resolution:   {result.resolution:.4f}")

# Data for a reliability diagram
centers, mean_pred, frac_pos = reliability_data(y_true, y_prob, n_bins=10)

4. Promote a champion

from pmlab import ChampionManifest, HoldoutGateResult

gate = HoldoutGateResult.evaluate(
    trades=result.trades,
    required_segments=["Buenos Aires", "Atlanta", "Dallas"],
    min_trades_per_segment=40,
    min_pnl_per_segment=0.0,
)
print(f"Gate decision: {gate.decision}")  # "GO" or "NO_GO"

if gate.decision == "GO":
    manifest = ChampionManifest.publish(
        model=model,
        gate=gate,
        output_dir="artifacts/public_models",
        plugin_family="my_markets",
    )
    print(f"Allowed segments: {manifest.get_allowed_segments()}")
# If gate.decision == "NO_GO", publish() raises ValueError — by design.

5. Paper trade with Kelly sizing

from pmlab import EdgeSignal, PaperBroker, kelly_stake_size

# Kelly-sized position (quarter-Kelly, max 5% of bankroll per trade)
stake = kelly_stake_size(
    win_prob=0.65,
    entry_price=0.35,
    bankroll=200.0,
    fraction=0.25,
    max_exposure=0.05,
)

broker = PaperBroker(
    trades_path="artifacts/ops/forward_paper_trades.json",
    allowed_segments=manifest.get_allowed_segments(),
    flat_stake=stake,
)
new_trades = broker.record(signals)

6. Generate an HTML report

from pmlab import generate_report

trades = broker.load_trades()
generate_report(
    trades,
    output_path="reports/session_report.html",
    title="May 2026 Paper Trading",
    brier_score=result_diagnostics.brier_score,
)

Opens a self-contained dark-themed HTML with equity curve, per-segment breakdown, and full trade log.

7. Go live (dry-run first)

from pmlab import LiveBroker

# Test without sending real orders (L1 private key + L2 API credentials)
with LiveBroker(
    private_key="0x...",          # L1 ECDSA private key
    api_key="...",
    api_secret="...",
    api_passphrase="...",
    dry_run=True,
) as broker:
    receipt = broker.place_order(token_id="tok_abc", side="BUY", price=0.35, size=14.28)
    print(receipt.status)  # "dry_run"

# Check balance before going live
with LiveBroker(
    private_key="0x...",
    api_key="...",
    api_secret="...",
    api_passphrase="...",
) as broker:
    balance = broker.get_balance()
    print(f"Available: ${balance:.2f} USDC")

Architecture

Polymarket API
      │
      ▼
MarketPlugin.discover_markets()  ──►  list[MarketSpec]
      │
      ├── fetch_features(spec, horizon)       ──►  dict[str, float]
      ├── fetch_truth(spec)                   ──►  float | str | None
      └── build_training_row(spec, horizon)   ──►  dict | None
                │
    [pmlab.features]  ← add_lags, rolling_stats, encode_cyclical, onehot, clip_outliers
                │
                ▼
        rolling_origin_eval(panel, model)      ◄── LGBMForecaster
                │
                ├── BacktestMetrics  (PnL, hit_rate, avg_edge)
                ├── brier_decomposition()      ◄── calibration diagnostics
                │
                └── HoldoutGate.evaluate(required_segments)
                            │
                          GO │ NO_GO → ValueError (hard gate)
                            │
                  ChampionManifest.publish(model, gate)
                            │
                     champion.json + champion.pkl
                            │
              ┌─────────────┴──────────────┐
              │                            │
        PaperBroker                   LiveBroker
        (paper trades)                (real CLOB orders)
        flat or kelly sizing          dry_run=True to test
              │
        SettlementEngine.settle_all(specs)
              │
        generate_report() → HTML report

Module map

Writing a Plugin

See the full guide at docs/plugin-authoring.md.

The four methods

class MarketPlugin(ABC):
    family: str  # unique identifier

    def discover_markets(self, **kwargs) -> list[MarketSpec]: ...
    def fetch_features(self, spec: MarketSpec, horizon: str, **kwargs) -> dict[str, float]: ...
    def fetch_truth(self, spec: MarketSpec, **kwargs) -> float | str | None: ...
    def build_training_row(self, spec: MarketSpec, horizon: str, **kwargs) -> dict | None: ...

    # Optional — override if your data source has a finalization lag:
    def is_truth_final(self, spec: MarketSpec, **kwargs) -> bool: ...

Dependency injection (recommended for testing)

class MyPlugin(MarketPlugin):
    family = "my_markets"

    def __init__(self, gamma_client=None):
        self._gamma = gamma_client or GammaClient()

# In tests — inject a mock, never hit the network:
mock_gamma = MagicMock()
plugin = MyPlugin(gamma_client=mock_gamma)

Feature Engineering

pmlab.features provides common transforms so plugins don't reimplement them:

from pmlab import add_lags, add_rolling_stats, encode_cyclical, encode_onehot, clip_outliers

# Lag features (with optional group-by for panel data)
df = add_lags(df, cols=["temp"], lags=[1, 3, 7], group_by="city")

# Rolling statistics
df = add_rolling_stats(df, cols=["temp"], windows=[7, 14], stats=["mean", "std"])

# Cyclical encoding (e.g. hour-of-day, day-of-year)
df["hour_sin"], df["hour_cos"] = encode_cyclical(df["hour"], period=24)

# One-hot encoding (drop_first=True avoids multicollinearity)
df = encode_onehot(df, cols=["city"], drop_first=True)

# Outlier clipping (IQR or z-score)
df = clip_outliers(df, cols=["temp", "humidity"], method="iqr", iqr_factor=3.0)

All transforms return a copy — originals are never mutated.

Data Layer — reproducibility & no-lookahead ingestion

pmlab.data is the reproducibility backbone: point-in-time snapshots, a typed panel builder, and leakage guards.

from pmlab import FeatureSnapshotStore, build_panel, check_no_leakage

# 1. Capture point-in-time snapshots of OPEN markets over time. When a market
#    later resolves, join its truth to the snapshot taken before decision time —
#    the features a decision could actually have seen.
store = FeatureSnapshotStore("~/.pmlab/snapshots")
store.append("politics", [
    {"market_id": "m1", "captured_at": "2025-01-05T00:00:00Z",
     "feature_spread": 0.02, "feature_vol": 1.2e6},
])
seen = store.as_of("politics", cutoff="2025-01-06")   # latest snapshot per market before the cutoff

# 2. Assemble a schema-validated panel from plugin rows (drops None, coerces
#    features to float, raises a clear error on missing columns / bad dates).
panel = build_panel(plugin.build_training_row(s, "resolution") for s in specs)

# 3. Guard against label leakage — the silent killer. Raises LeakageError on any
#    feature that separates the outcome suspiciously well.
check_no_leakage(panel)                    # or build_panel(rows, check_leakage=True)

# 4. Close the loop: when markets resolve, build a no-lookahead panel by joining
#    each market's pre-decision snapshot to its realized truth.
from pmlab import build_panel_from_snapshots

resolutions = pd.DataFrame([
    {"market_id": "m1", "decision_date": "2025-01-10", "winning_label": "Yes"},
])
train = build_panel_from_snapshots(store, "politics", resolutions)  # uses snapshot < decision_date

Async Market Access

For scanning many markets in parallel, use the async clients:

import asyncio
from pmlab import AsyncGammaClient, AsyncClobClient

async def scan():
    async with AsyncGammaClient() as gamma:
        markets = await gamma.fetch_markets(tag="temperature", keyword="highest")

    token_ids = [m["conditionId"] for m in markets]
    async with AsyncClobClient(concurrency=20) as clob:
        prices = await clob.fetch_prices(token_ids)

    return markets, prices

markets, prices = asyncio.run(scan())

TTL disk cache

Avoid redundant API calls during repeated scans:

from pmlab import GammaClient, DiskCache

cache = DiskCache(cache_dir=".pmlab_cache", ttl_seconds=3600)
client = GammaClient(cache=cache)

markets = client.fetch_markets(tag="temperature")  # fetched from API
markets = client.fetch_markets(tag="temperature")  # served from cache

Position Sizing

from pmlab import flat_stake_size, kelly_fraction, kelly_stake_size

# Flat stake — always risk $5 regardless of edge
shares = flat_stake_size(flat_stake=5.0, entry_price=0.35)

# Kelly fraction — what % of bankroll to wager?
f = kelly_fraction(win_prob=0.65, entry_price=0.35, fraction=0.25)  # quarter-Kelly

# Kelly stake — USDC amount, capped at max_exposure % of bankroll
stake = kelly_stake_size(
    win_prob=0.65,
    entry_price=0.35,
    bankroll=500.0,
    fraction=0.25,
    max_exposure=0.05,  # never more than 5% of bankroll per trade
)

Advanced Modeling

Every forecaster below implements the same MarketForecaster interface (fit / predict_proba / save / load), so they compose freely and drop into the backtest, champion gate, and brokers unchanged.

Ensembles

Blend several models to cut variance — a weighted average of member probabilities:

from pmlab import EnsembleForecaster, LGBMForecaster, SklearnForecaster

ens = EnsembleForecaster(
    forecasters=[
        LGBMForecaster(),
        SklearnForecaster(estimator="random_forest", n_estimators=200),
        SklearnForecaster(estimator="logistic_regression"),
    ],
    weights=[2.0, 1.0, 1.0],  # optional; defaults to equal weight
)
ens.fit(X_train, y_train)
proba = ens.predict_proba(X_test)

Probability calibration

Edge is computed from probabilities, so calibration matters. Wrap any forecaster to calibrate its positive-class output on a held-out split:

from pmlab import CalibratedForecaster, LGBMForecaster

clf = CalibratedForecaster(LGBMForecaster(), method="isotonic")  # or "sigmoid" (Platt)
clf.fit(X_train, y_train)
calibrated = clf.predict_proba(X_test)  # calibrated (n, 2) probabilities

Or calibrate raw scores directly:

from pmlab import SigmoidCalibrator, IsotonicCalibrator

cal = SigmoidCalibrator()          # Platt scaling — sample-efficient
cal.fit(raw_scores, true_labels)
p = cal.transform(new_scores)

Conformal prediction sets

Turn a point probability into a prediction set with a distribution-free coverage guarantee: on exchangeable data, the set contains the true label with probability ≥ 1 − alpha, no matter how well-specified the base model is. A singleton set is a confident call; a full set is the model saying "toss-up".

from pmlab import ConformalForecaster, LGBMForecaster

clf = ConformalForecaster(LGBMForecaster(), alpha=0.1)  # target 90% coverage
clf.fit(X_train, y_train)

sets = clf.predict_set(X_test)   # list[set[int]] — e.g. {1}, {0, 1}, {0}
proba = clf.predict_proba(X_test)  # still a drop-in forecaster

Multiclass calibration

Calibration extends to multi-outcome markets (F1 winner, multi-candidate races) via one-vs-rest — just pass a multiclass target:

from pmlab import CalibratedForecaster, SklearnForecaster, multiclass_brier

clf = CalibratedForecaster(SklearnForecaster("random_forest"), method="isotonic")
clf.fit(X_train, y_train)          # y has 3+ classes
proba = clf.predict_proba(X_test)  # calibrated (n, n_classes), rows sum to 1

score = multiclass_brier(y_test, proba)   # Brier + climatology skill score
print(score.brier_score, score.skill_score)

Hyperparameter tuning (no-lookahead)

TunedForecaster searches with Optuna but scores every trial through rolling_origin_eval, so model selection is walk-forward — it never shuffles the future into the past. Requires the tune extra (pip install pmlab[tune]):

from pmlab import TunedForecaster, SklearnForecaster

tuner = TunedForecaster(
    build_fn=lambda p: SklearnForecaster(estimator="random_forest", **p),
    param_space=lambda t: {"n_estimators": t.suggest_int("n_estimators", 50, 400, step=50)},
    n_trials=30,
    metric="total_pnl",      # or "mean_pnl" / "hit_rate"
)
tuner.tune(panel)            # panel = the same format rolling_origin_eval expects
print(tuner.best_params_, tuner.best_score_)
proba = tuner.predict_proba(X_test)   # refit on the full panel, ready to use

Leakage-aware cross-validation

For time-series model selection outside the backtest, use splitters that respect time instead of sklearn.KFold:

import numpy as np
from pmlab import purged_kfold, embargoed_split

idx = np.arange(len(panel))            # rows sorted by decision_date
for train, test in purged_kfold(idx, n_splits=5, embargo=3):
    ...   # test is a contiguous block; train purges an embargo band around it

for train, test in embargoed_split(idx, n_splits=5, embargo=3):
    ...   # expanding-origin walk-forward: train is always strictly earlier

Bundled Plugins

WeatherTmaxPlugin

Handles "Highest temperature in [city] on [date]?" markets. Features: ECMWF forecast temperature, ensemble spread, lead time, city baseline. Truth: official observations via Wunderground / NOAA / CWA.

SportsF1Plugin

Handles "F1 [GP] winner?" and similar categorical race markets. Categorical bins — winning label is a driver/team name string.

CLI Reference

pmlab version                                              # print version
pmlab status                                               # paper trade status
pmlab scan-markets --plugin weather_tmax                   # discover open markets
pmlab record-trades --plugin weather_tmax --min-edge 0.20  # record paper trades
pmlab settle-trades --plugin weather_tmax                  # settle resolved positions
pmlab backtest --plugin weather_tmax --stride 30           # walk-forward backtest
pmlab promote-champion model.pkl --gate-path gate.json \   # promote if gate is GO
               --plugin weather_tmax

Workspace isolation

scripts/pmlab-workspace ops_daily pmlab scan-markets --plugin weather_tmax
scripts/pmlab-workspace historical_real pmlab backtest --plugin weather_tmax --stride 30

API Reference

Core

Features

Markets

Modeling

Execution

Reports

Key Design Decisions

Hard gate on champion promotion — ChampionManifest.publish() raises ValueError if gate.decision != "GO". Not configurable. Investigate failures; don't bypass the gate.

champion.json is the source of truth — PaperBroker reads allowed_segments from champion.json, not from any intermediate benchmark file that could be overwritten by a failed retrain.

No-lookahead guarantee — rolling_origin_eval trains strictly on rows with decision_date < eval_date. This is asserted in the test suite.

Settlement waits for finalization — SettlementEngine only settles when plugin.is_truth_final(spec) returns True. Data sources with finalization lags can override this method.

paper → live transition — Use LiveBroker(dry_run=True) to validate signals against the real CLOB API without sending orders. Flip dry_run=False only when validators pass.

Development

git clone https://github.com/ArtBreguez/polymarket-lab
cd polymarket-lab
uv sync --extra dev

uv run pytest                              # all tests
uv run pytest --cov=src/pmlab             # with coverage report
uv run ruff check src/ tests/             # lint
uv run ruff check src/ tests/ --fix       # auto-fix
uv run mypy src/                          # type check
uv build                                  # build wheel + sdist

Project Status

Tests: 237 passing · Coverage: 95.7% · Python: 3.12

Contributing

See CONTRIBUTING.md for the full guide.

Quick version:

1. Fork the repo
2. Create a branch: git checkout -b feat/my-plugin
3. Write your plugin in src/pmlab/plugins/<family>/
4. Add tests in tests/plugins/<family>/ — TDD required
5. Run uv run pytest && uv run ruff check src/ tests/
6. Open a pull request

License

MIT — see LICENSE for details.