ironflow-sdk 0.3.0

Python SDK for Ironflow market data API — on-chain derivatives data

ironflow

Python SDK for the Ironflow market data API — on-chain derivatives data.

• Async-first with httpx and websockets
• Typed dataclass responses for all endpoints
• Automatic pagination with await page.next()
• Typed WebSocket streams with async generators
• Full type annotations (py.typed)

Install

pip install ironflow-sdk

Requires Python 3.10+.

Quick Start

from ironflow_sdk import Ironflow

async with Ironflow("if_your_api_key") as client:
    # Get recent trades
    page = await client.trades("BTC-PERP", limit=10)
    for trade in page.data:
        print(trade.price, trade.size, trade.side)

    # Paginate
    if page.has_more:
        next_page = await page.next()

Configuration

client = Ironflow(
    "if_your_api_key",
    base_url="https://api.ironflow.sh",  # default
    source="hyperliquid",                # default
    timeout=30.0,                        # request timeout in seconds (default: 30)
)

REST Endpoints

Market Data

async with Ironflow("if_your_api_key") as client:
    # Trades
    trades = await client.trades("BTC-PERP", limit=100)

    # Fills (requires address)
    fills = await client.fills("0xabc...", market="ETH-PERP")

    # Order book snapshot
    book = await client.book("BTC-PERP")
    # book.bids[0].price, book.bids[0].size

    # OHLCV candles
    candles = await client.candles("BTC-PERP", "1h", limit=24)

    # Liquidations
    liqs = await client.liquidations("BTC-PERP")

    # Funding rates
    rates = await client.funding("BTC-PERP")

    # Open interest
    oi = await client.open_interest("BTC-PERP")

    # Mark prices
    prices = await client.mark_prices("BTC-PERP")

    # Deposits & withdrawals
    deps = await client.deposits(address="0xabc...")
    wds = await client.withdrawals(address="0xabc...")

    # Order statuses
    orders = await client.order_statuses(address="0xabc...")

    # Vault operations
    vaults = await client.vault_operations(vault="HLP")

Analytics (Builder+)

flows = await client.net_flows(interval="1d", limit=7)
# flows[0].deposits, flows[0].withdrawals, flows[0].net_flow

levels = await client.liquidation_levels("BTC-PERP", bucket_size=100)
# levels[0].price_bucket, levels[0].count, levels[0].total_size

order_flow = await client.order_flow()
# order_flow[0].total_orders, order_flow[0].filled_orders, order_flow[0].fill_rate

leaderboard = await client.vault_leaderboard()
# leaderboard[0].vault, leaderboard[0].total_deposits, leaderboard[0].net_flow

funding = await client.funding_stats("BTC-PERP", interval="8h")
# funding[0].avg_rate, funding[0].min_rate, funding[0].max_rate

Triggers

# Create a webhook trigger
trigger = await client.create_trigger(
    name="whale-alert",
    channel="trades",
    webhook_url="https://your-server.com/hook",
    rule={"condition": {"field": "data.size", "op": "gt", "value": "100"}},
)

# List, toggle, delete
triggers = await client.list_triggers()
await client.toggle_trigger(trigger.id, is_active=False)
await client.delete_trigger(trigger.id)

# Test a rule without creating
matched = await client.test_trigger(
    rule={"condition": {"field": "data.size", "op": "gt", "value": "10"}},
    event={"data": {"size": "50", "market": "BTC-PERP"}},
)
print(matched)  # True

Cohorts (Enterprise)

cohorts = await client.list_cohorts()
addrs = await client.cohort_addresses("top_pnl_30d")
await client.create_cohort("my_whales", ["0xabc...", "0xdef..."])
await client.delete_cohort("my_whales")

Export (Builder+)

data = await client.export_data(
    "fills",
    market="BTC-PERP",
    format="csv",
)
# data is bytes — write to file or parse with pandas
import pandas as pd
df = pd.read_csv(io.BytesIO(data))

Status

status = await client.status()
print(status.status)  # "ok"
print(status.venues["hyperliquid"].streams["fills"].age_seconds)

Info API Proxy

meta = await client.info({"type": "metaAndAssetCtxs"})

WebSocket Streaming

All stream methods return typed async generators:

# Real-time trades
async for trade in client.stream.trades("BTC-PERP"):
    print(trade.price, trade.size, trade.side)

# Wallet fills (copy trading)
async for fill in client.stream.fills("0xleader"):
    print(fill.side, fill.size, fill.market)

# Order book updates
async for snap in client.stream.book("ETH-PERP"):
    print(snap.bids[0].price, snap.asks[0].price)

# Filter by minimum size
async for trade in client.stream.trades("BTC-PERP", min_size="10"):
    pass  # Only trades >= 10 BTC

Available streams: trades, fills, book, book_l4, orders, liquidations, funding_rates, deposits, withdrawals, vault_operations.

Error Handling

from ironflow_sdk import Ironflow, RateLimitError, AuthError, TierError, QueryWindowError

try:
    trades = await client.trades("BTC-PERP")
except RateLimitError as e:
    print(f"Rate limited, retry in {e.retry_after_ms}ms")
except AuthError:
    print("Invalid API key")
except TierError:
    print("Upgrade your plan for this endpoint")
except QueryWindowError:
    print("Time range too wide for your tier")

Pagination

All list endpoints return a Page[T] with async pagination:

page = await client.trades("BTC-PERP", limit=1000)
all_trades = list(page.data)

while page.has_more:
    page = await page.next()
    all_trades.extend(page.data)

License

MIT