Back

Prediction Pools Python SDK

Thin, dependency-light Python client for the prediction pools. Predict where the index lands at settlement — BTC/USD every 5 minutes (btc-5m) or every 4 hours (btc-4h), or the live event index every 15 minutes (event-5m — historical slug, 15-minute cadence — the tradable-event pool on the landing page; its index is the price-weighted basket of the assets moved by the featured news headline). The closest prediction takes the pool. Everything below works for all three pools — pass pool="event-5m" to trade the landing-page event. The SDK wraps the same endpoints documented in the REST API reference.

REST & WebSocket Python SDK MCP Tools 5-Minute Pool 4-Hour Pool

Install

pip install requests websockets

The SDK is a single file. Drop it in your project — no package install required.

Quickstart

from btc_pool import BtcPoolClient

client = BtcPoolClient(
    base_url="https://cymetica.com",
    email="you@example.com",
    password="…",
)

# Read the 5-minute pool
s = client.state("btc-5m")
rnd = s["entry_round"]
print(f"round {rnd['round_epoch']}: {rnd['entries']} entries, "
      f"pool ${rnd['pool_usdc']}, locks at {rnd['locks_at']}")
print("live index:", s["index"]["price"])

# Enter: predict BTC/USD at settlement, stake $1.00 ($0.10 min, $1,000,000 max)
ticket = client.enter(predicted_price="65450.00", amount_usdc="1.00", pool="btc-5m")
print("entered:", ticket)

# Your ticket, all-time pool profit, + last 20 outcomes
me = client.me("btc-5m")
print("my entry:", me["entry"], "— net profit:", me["profit_usdc"])

# Changed your mind? Cancel for a full refund any time before lock,
# then re-enter at a new price while the round is still open.
client.cancel("btc-5m")

Reading the Prediction Tape

entry_round.predictions is the anonymized public tape — every live prediction's price and stake, no user data. id is an opaque rung handle (stable for the life of the round) so a client can apply pool_update deltas to a single rung instead of re-fetching. The same data drives the Prediction Book on the pool page.

s = client.state("btc-5m")
idx = float(s["index"]["price"])
preds = s["entry_round"]["predictions"]   # [{"id": 4812, "price": "...", "amount_usdc": "...", "side": "up"}]
above = [p for p in preds if float(p["price"]) > idx]
below = [p for p in preds if float(p["price"]) <= idx]
print(f"tape: {len(above)} above / {len(below)} below the index at {idx}")

Streaming a Pool

The WebSocket wraps every message as {"type", "data", "ts"}. One socket carries all pools — filter on data.pool where present.

import asyncio
from btc_pool import BtcPoolStream

async def main():
    async with BtcPoolStream() as stream:
        async for event in stream:
            t, d = event.get("type"), event.get("data") or {}
            if t == "index_tick":
                print("BTC/USD", d["price"])
            elif t == "pool_update" and d.get("pool") == "btc-5m":
                print(f"entry: {d['entries']} tickets, pool ${d['pool_usdc']}")
            elif t == "round_update" and d.get("pool") == "btc-5m":
                print(f"round {d['round_epoch']} → {d['status']}")
            elif t == "win_event":
                print("winner:", d)

asyncio.run(main())

SDK Source (paste this file)

"""btc_pool.py — minimal client for EventTrader BTC prediction pools."""
import json
import requests
import websockets

BASE_PATH = "/api/v1/btc-pool"


class BtcPoolClient:
    def __init__(self, base_url: str, email: str | None = None, password: str | None = None):
        self.base = base_url.rstrip("/")
        self.s = requests.Session()
        self.token = None
        if email and password:
            r = self.s.post(f"{self.base}/auth/login",
                            json={"email": email, "password": password}, timeout=15)
            r.raise_for_status()
            self.token = r.json()["access_token"]
            self.s.headers.update({"Authorization": f"Bearer {self.token}"})

    def state(self, pool: str = "btc-4h"):
        r = self.s.get(f"{self.base}{BASE_PATH}/state", params={"pool": pool}, timeout=10)
        r.raise_for_status()
        return r.json()

    def enter(self, predicted_price, amount_usdc, pool: str = "btc-4h"):
        # predicted_price > 0; amount_usdc between 0.10 and 1,000,000
        # (cumulative per user per round). Pass Decimals as strings.
        r = self.s.post(
            f"{self.base}{BASE_PATH}/enter",
            json={"predicted_price": str(predicted_price),
                  "amount_usdc": str(amount_usdc), "pool": pool},
            timeout=15,
        )
        r.raise_for_status()
        return r.json()

    def cancel(self, pool: str = "btc-4h"):
        # Full refund of your entry — allowed strictly until the round locks.
        r = self.s.post(f"{self.base}{BASE_PATH}/cancel",
                        json={"pool": pool}, timeout=15)
        r.raise_for_status()
        return r.json()

    def me(self, pool: str = "btc-4h"):
        r = self.s.get(f"{self.base}{BASE_PATH}/me", params={"pool": pool}, timeout=10)
        r.raise_for_status()
        return r.json()

    def history(self, pool: str = "btc-4h", limit: int = 24):
        r = self.s.get(f"{self.base}{BASE_PATH}/history",
                       params={"pool": pool, "limit": limit}, timeout=10)
        r.raise_for_status()
        return r.json()


class BtcPoolStream:
    def __init__(self, base_url: str = "wss://cymetica.com"):
        self.url = f"{base_url}/ws/btc-pool"
        self._ws = None
    async def __aenter__(self):
        self._ws = await websockets.connect(self.url)
        return self
    async def __aexit__(self, *a):
        if self._ws:
            await self._ws.close()
    def __aiter__(self):
        return self
    async def __anext__(self):
        msg = await self._ws.recv()
        return json.loads(msg)

Examples

Momentum entry — predict a drift from the live index

s = client.state("btc-5m")
idx = float(s["index"]["price"])
# Naive: assume the last round's move repeats
last = s.get("last_round") or {}
drift = 0.0
if last.get("p0") and last.get("p1"):
    drift = float(last["p1"]) - float(last["p0"])
client.enter(predicted_price=f"{idx + drift:.2f}", amount_usdc="0.50", pool="btc-5m")

Round-outcome tracker

h = client.history("btc-5m", limit=50)
settled = [r for r in h["rounds"] if r["status"] == "settled"]
voided  = [r for r in h["rounds"] if r["status"] == "voided"]
dists = [float(r["winning_distance"]) for r in settled if r["winning_distance"]]
print(f"{len(settled)} settled / {len(voided)} voided")
if dists:
    print(f"median winning distance: ${sorted(dists)[len(dists)//2]:.2f}")

Wait for settlement, then check your payout

import asyncio
from btc_pool import BtcPoolStream

async def wait_for_settle(pool="btc-5m"):
    async with BtcPoolStream() as stream:
        async for event in stream:
            d = event.get("data") or {}
            if (event.get("type") == "round_update"
                    and d.get("pool") == pool and d.get("status") == "settled"):
                return d

rnd = asyncio.run(wait_for_settle())
me = client.me("btc-5m")
mine = [h for h in me["history"] if h["round_epoch"] == rnd["round_epoch"]]
print("outcome:", mine[0] if mine else "no ticket in that round")