mmflow SDK · Python

Python SDK

Single-file, **standard-library-only** Python client for the mmflow Open API. No requests, no httpx, no aiohttp — just urllib. Tested on CPython 3.10+. MIT licensed.

Download mmflow.py ← REST reference TypeScript SDK →OpenAPI spec

Install

stdlib only · no pip

Copy file into your project

Drop mmflow.py anywhere on PYTHONPATH. Zero deps; stdlib only.

curl -O https://mmflow.ai/sdk/mmflow.py

Or paste into your script

Single file, MIT licensed — vendor it next to your code.

python -c "import urllib.request; urllib.request.urlretrieve('https://mmflow.ai/sdk/mmflow.py', 'mmflow.py')"

Quickstart

from mmflow import MmflowClient

mm = MmflowClient()                       # default https://mmflow.ai/api/v1
# mm = MmflowClient(base_url="http://localhost:3000/api/v1")

# REST
oi = mm.perps_oi(coins=["BTC", "ETH"]).data
print(oi[0]["coin"], oi[0]["oiUsd"])

# SSE
for ev in mm.stream_whales(min_usd=250_000):
    if ev.type == "whale":
        print(ev.data["coin"], ev.data["notionalUsd"])
    if ev.type == "bye":
        break

Recipes

one snippet per resource family

Perps OI snapshot

Cross-venue open interest for two coins.

from mmflow import MmflowClient

mm = MmflowClient()
resp = mm.perps_oi(coins=["BTC", "ETH"])
for row in resp.data:
    print(row["coin"], row["venue"], row["oiUsd"])

Concurrent fetches

Use threads to hit two endpoints in parallel.

from concurrent.futures import ThreadPoolExecutor
from mmflow import MmflowClient

mm = MmflowClient()
with ThreadPoolExecutor(max_workers=2) as pool:
    f_funding = pool.submit(mm.perps_funding, coins=["BTC", "ETH"])
    f_liqs = pool.submit(mm.perps_liquidations, coins=["BTC", "ETH"])
funding, liqs = f_funding.result(), f_liqs.result()
fund_by_coin = {r["coin"]: r["rate"] for r in funding.data}
for row in liqs.data:
    print(row["coin"], "liq24h=", row["long24hUsd"] + row["short24hUsd"],
          "funding=", fund_by_coin.get(row["coin"], 0))

Options GEX surface

BTC dealer gamma exposure — call wall, put wall, gamma flip.

gex = mm.options_gex(underlying="BTC").data
print("spot", gex["spot"])
print("totalGex", gex["totalGex"])
print("callWall", gex["callWall"])
print("putWall", gex["putWall"])
print("gammaFlip", gex["gammaFlip"])

IV smile + term structure

Full vol surface ready to fit your own smile model.

skew = mm.options_skew(underlying="BTC").data
for exp in skew["termStructure"]:
    print(exp["expiryLabel"], exp["dteDays"], "ATM IV=", exp["atmIv"])
first_expiry = next(iter(skew["smileByExpiry"]))
smile = skew["smileByExpiry"][first_expiry]
print("strikes in smile:", len(smile))

Cross-venue spot ticker

BTC + ETH price across Coinbase, Binance, Bitfinex, Kraken.

spot = mm.markets_spot(coins=["BTC", "ETH"]).data
by_coin = {}
for row in spot:
    by_coin.setdefault(row["coin"], {})[row["venue"]] = row["price"]
print(by_coin)

Polymarket events

Top 5 active markets by liquidity.

events = mm.polymarket_events(limit=5).data
for ev in events:
    print(ev["question"][:60],
          "yes=", round(ev["yesPrice"], 3),
          "liq=", round(ev["liquidity"]))

On-chain valuation

BTC MVRV-Z, NUPL, Puell Multiple in one call.

v = mm.onchain_valuation().data
print("MVRV-Z:", v["mvrvZScore"])
print("NUPL:", v["nupl"])
print("Puell:", v["puellMultiple"])
print("as of:", v["asOfDate"])

Live whale tape (SSE)

Stream HL whale prints over $250k.

for ev in mm.stream_whales(min_usd=250_000):
    if ev.type == "whale":
        w = ev.data
        print(w["coin"], w["side"], round(w["notionalUsd"]))
    elif ev.type == "bye":
        break

Live HL trade stream

BTC prints ≥ $50k.

from datetime import datetime, timezone

for ev in mm.stream_trades(coin="BTC", min_usd=50_000):
    if ev.type != "trade":
        continue
    t = ev.data
    when = datetime.fromtimestamp(t["ts"] / 1000, timezone.utc).isoformat()
    print(when, t["side"], t["size"], "@", t["price"])

Funding rate change alerts

Fire only when |delta| ≥ 1bp on BTC / ETH.

for ev in mm.stream_funding(coins=["BTC", "ETH"], min_delta_bps=1):
    if ev.type != "funding":
        continue
    f = ev.data
    bps = (f.get("delta") or 0) * 1e4
    print(f["coin"], "rate=", f["rate"], "delta=", round(bps, 2), "bps")

Errors

Non-2xx HTTP responses raise MmflowApiError with .status and .body attributes:

from mmflow import MmflowClient, MmflowApiError

mm = MmflowClient()
try:
    mm.options_gex(underlying="XYZ")
except MmflowApiError as e:
    print(e.status, e.body)

Looking for the TypeScript version? TypeScript SDK