Back

Stock Predictions Python SDK

A thin, dependency-light Python client for the stock prediction REST + WebSocket API. Covers 3,700+ markets across NYSE, NASDAQ, AMEX, and OTC exchanges. The SDK wraps the same endpoints documented in the REST API reference.

REST & WebSocket Python SDK MCP Tools All Stock Markets

Install

pip install requests websockets

The SDK is a single file you can drop into your project. No package install required — it talks to the public API.

Quickstart

from stock_predictions import StockPredictionClient

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

# List all markets
markets = client.markets()
print(f"{len(markets)} stock prediction markets")

# Find AAPL
aapl = client.find("AAPL")

# Read the CLOB orderbook
book = client.orderbook(aapl["id"], "AAPL")
print("best_bid:", book["best_bid"], "best_ask:", book["best_ask"])

# Place a buy order — AAPL beats S&P 500 — at $0.55
order = client.place_order(
    market_id=aapl["id"],
    symbol="AAPL",
    side="buy",
    price=0.55,
    size=10,
)

Browse by Sector

# Filter markets by sector
markets = client.markets()
materials = [m for m in markets if m["assets"][0]["sector"] == "Materials"]
tech = [m for m in markets if m["assets"][0]["sector"] == "Technology"]
print(f"Materials: {len(materials)}, Tech: {len(tech)}")

Streaming Trades

import asyncio
from stock_predictions import StockPredictionStream

async def main():
    aapl = client.find("AAPL")
    async with StockPredictionStream(market_id=aapl["id"]) as stream:
        async for event in stream:
            if event["type"] == "trade":
                t = event["data"]
                print(f"{t['symbol']} {t['taker_side']} {t['size']}@{t['price']}")

asyncio.run(main())

SDK Source (paste this file)

"""stock_predictions.py — minimal client for EventTrader stock prediction markets."""
import json
import requests
import websockets

class StockPredictionClient:
    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 markets(self):
        r = self.s.get(f"{self.base}/api/v1/otc/markets", timeout=10)
        r.raise_for_status()
        return r.json()["markets"]

    def find(self, ticker: str):
        for m in self.markets():
            for a in m.get("assets", []):
                if a["symbol"].upper() == ticker.upper():
                    return m
        raise ValueError(f"No market found for {ticker}")

    def orderbook(self, market_id: str, symbol: str):
        r = self.s.get(f"{self.base}/api/v1/otc/orderbook/{market_id}/{symbol}", timeout=10)
        r.raise_for_status()
        return r.json()

    def trades(self, market_id: str, symbol: str, limit: int = 200):
        r = self.s.get(f"{self.base}/api/v1/otc/trades/{market_id}/{symbol}",
                       params={"limit": limit}, timeout=10)
        r.raise_for_status()
        return r.json()

    def place_order(self, market_id: str, symbol: str, side: str, price: float, size: float):
        r = self.s.post(f"{self.base}/api/v1/otc/order",
                        json={"market_id": market_id, "symbol": symbol,
                              "side": side, "price": price, "size": size}, timeout=15)
        r.raise_for_status()
        return r.json()

    def cancel_order(self, market_id: str, order_id: str, symbol: str):
        r = self.s.delete(f"{self.base}/api/v1/otc/order/{market_id}/{order_id}",
                          params={"symbol": symbol}, timeout=10)
        r.raise_for_status()
        return r.json()

    def bet(self, market_id: str, symbol: str, amount: float, user_address: str):
        r = self.s.post(f"{self.base}/api/v1/wta/markets/{market_id}/bet",
                        json={"symbol": symbol, "amount": amount, "user_address": user_address}, timeout=15)
        r.raise_for_status()
        return r.json()

class StockPredictionStream:
    def __init__(self, market_id: str, base_url: str = "wss://cymetica.com"):
        self.url = f"{base_url}/ws/wta/markets/{market_id}"
        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

Scan for stocks with biggest pools

markets = client.markets()
with_pool = [m for m in markets if m.get("current_epoch", {}).get("total_pool", 0) > 0]
with_pool.sort(key=lambda m: m["current_epoch"]["total_pool"], reverse=True)
for m in with_pool[:10]:
    ticker = m["assets"][0]["symbol"]
    pool = m["current_epoch"]["total_pool"]
    print(f"{ticker}: ${pool:.2f} pool")

Stream live spread via WebSocket

import asyncio

async def watch_spread():
    m = client.find("AAPL")
    async with StockPredictionStream(m["id"]) as stream:
        async for event in stream:
            if event["type"] == "orderbook_update":
                d = event["data"]
                print(d["best_bid"], "->", d["best_ask"], " spread:", d["spread"])

asyncio.run(watch_spread())

Simple mean-reversion bot

m = client.find("TSLA")
trades = client.trades(m["id"], "TSLA", limit=50)["trades"]
if not trades:
    raise SystemExit("No trades yet")
prices = [float(t["price"]) for t in trades]
avg = sum(prices) / len(prices)
last = prices[-1]
if last < avg - 0.03:
    client.place_order(m["id"], "TSLA", "buy", round(last + 0.01, 2), 5)
elif last > avg + 0.03:
    client.place_order(m["id"], "TSLA", "sell", round(last - 0.01, 2), 5)

EventTrader