BTC Prediction Pools Python SDK
Thin, dependency-light Python client for the BTC prediction pools. Predict where BTC/USD lands at settlement — every 5 minutes (btc-5m) or every 4 hours (btc-4h); the closest prediction takes the pool. The SDK wraps the same endpoints documented in the REST API reference.
Install
pip install requests websocketsThe 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. 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"] # [{"price": "...", "amount_usdc": "..."}]
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")