A comprehensive, type-safe, async-first Python SDK for Dome API. Features include market data, wallet analytics, order tracking, and cross-platform market matching for prediction markets. For detailed API documentation, visit DomeApi.io.
# Using pip
pip install dome-api-sdk
# Using poetry
poetry add dome-api-sdk
# Using pipenv
pipenv install dome-api-sdkfrom dome_api_sdk import DomeClient
# Initialize the client with your API key
dome = DomeClient({"api_key": "your-dome-api-key-here"})
# Get market price
market_price = dome.polymarket.markets.get_market_price({
"token_id": "98250445447699368679516529207365255018790721464590833209064266254238063117329"
})
print(f"Market Price: {market_price.price}")The SDK accepts the following configuration options:
from dome_api_sdk import DomeClient
config = {
"api_key": "your-api-key", # Authentication token (required)
"base_url": "https://api.domeapi.io/v1", # Base URL (optional)
"timeout": 30.0, # Request timeout (optional)
}
client = DomeClient(config)You can also configure the SDK using environment variables:
export DOME_API_KEY="your-api-key"from dome_api_sdk import DomeClient
# Will automatically use DOME_API_KEY from environment
client = DomeClient()The Dome SDK provides access to the following API endpoints, organized by platform:
All Polymarket endpoints are accessed through dome.polymarket.*:
| Category | Method | Description | Endpoint Path |
|---|---|---|---|
| Markets | markets.get_market_price() |
Get current or historical market price by token ID | /polymarket/market-price/{token_id} |
| Markets | markets.get_candlesticks() |
Get historical candlestick data for a market | /polymarket/candlesticks/{condition_id} |
| Markets | markets.get_markets() |
Get market data with filtering (slug, tags, status, etc.) | /polymarket/markets |
| Markets | markets.get_orderbooks() |
Get historical orderbook snapshots for an asset | /polymarket/orderbooks |
| Orders | orders.get_orders() |
Get order data with filtering (market, user, time range, etc.) | /polymarket/orders |
| WebSocket | websocket.subscribe() |
Subscribe to real-time order events via WebSocket | wss://ws.domeapi.io/{api_key} |
| WebSocket | websocket.unsubscribe() |
Unsubscribe from order events | wss://ws.domeapi.io/{api_key} |
| WebSocket | websocket.get_active_subscriptions() |
Get all active subscriptions | N/A |
| Wallet | wallet.get_wallet_pnl() |
Get realized profit and loss (PnL) for a wallet | /polymarket/wallet/pnl/{wallet_address} |
| Activity | activity.get_activity() |
Get trading activity (MERGE, SPLIT, REDEEM) for a user | /polymarket/activity |
All Kalshi endpoints are accessed through dome.kalshi.*:
| Category | Method | Description | Endpoint Path |
|---|---|---|---|
| Markets | markets.get_markets() |
Get Kalshi market data with filtering | /kalshi/markets |
| Orderbooks | orderbooks.get_orderbooks() |
Get historical Kalshi orderbook snapshots | /kalshi/orderbooks |
Cross-platform market matching endpoints are accessed through dome.matching_markets.*:
| Method | Description | Endpoint Path |
|---|---|---|
get_matching_markets() |
Find equivalent markets across platforms by Polymarket slug or Kalshi ticker | /matching-markets/sports/ |
get_matching_markets_by_sport() |
Find equivalent markets by sport and date | /matching-markets/sports/{sport}/ |
Get current or historical market prices:
from dome_api_sdk import DomeClient
dome = DomeClient({"api_key": "your-api-key"})
# Current price
price = dome.polymarket.markets.get_market_price({
"token_id": "1234567890"
})
print(f"Current Price: {price.price}")
# Historical price
historical_price = dome.polymarket.markets.get_market_price({
"token_id": "1234567890",
"at_time": 1740000000 # Unix timestamp
})
print(f"Historical Price: {historical_price.price}")Get historical candlestick data for market analysis:
from dome_api_sdk import DomeClient
dome = DomeClient({"api_key": "your-api-key"})
candlesticks = dome.polymarket.markets.get_candlesticks({
"condition_id": "0x4567b275e6b667a6217f5cb4f06a797d3a1eaf1d0281fb5bc8c75e2046ae7e57",
"start_time": 1640995200,
"end_time": 1672531200,
"interval": 60 # 1 = 1m, 60 = 1h, 1440 = 1d
})
print(f"Candlesticks: {len(candlesticks.candlesticks)}")Get market data with filtering and search:
from dome_api_sdk import DomeClient
dome = DomeClient({"api_key": "your-api-key"})
# Get markets by status
markets = dome.polymarket.markets.get_markets({
"status": "open",
"limit": 20,
"min_volume": 100000
})
print(f"Markets: {len(markets.markets)}")
# Get markets by slug(s)
markets_filtered = dome.polymarket.markets.get_markets({
"market_slug": ["bitcoin-up-or-down-july-25-8pm-et"],
"limit": 10
})
# Get markets by tags
markets_by_tags = dome.polymarket.markets.get_markets({
"tags": ["crypto", "politics"],
"status": "open"
})Get historical orderbook snapshots:
from dome_api_sdk import DomeClient
dome = DomeClient({"api_key": "your-api-key"})
orderbooks = dome.polymarket.markets.get_orderbooks({
"token_id": "18823838997443878656879952590502524526556504037944392973476854588563571859850",
"start_time": 1760470000000, # milliseconds
"end_time": 1760480000000, # milliseconds
"limit": 100
})
print(f"Orderbook snapshots: {len(orderbooks.snapshots)}")Get order data with filtering:
from dome_api_sdk import DomeClient
dome = DomeClient({"api_key": "your-api-key"})
# Get orders by market slug
orders = dome.polymarket.orders.get_orders({
"market_slug": "bitcoin-up-or-down-july-25-8pm-et",
"limit": 50,
"offset": 0,
"start_time": 1640995200,
"end_time": 1672531200
})
print(f"Orders: {len(orders.orders)}")
# Get orders by user
user_orders = dome.polymarket.orders.get_orders({
"user": "0x7c3db723f1d4d8cb9c550095203b686cb11e5c6b",
"limit": 100
})
# Get orders with array filters
orders_array = dome.polymarket.orders.get_orders({
"market_slug": ["slug1", "slug2"],
"limit": 50
})Subscribe to real-time Polymarket order data via WebSocket. The SDK automatically handles reconnection with exponential backoff and subscription management.
import asyncio
from dome_api_sdk import DomeClient, WebSocketOrderEvent
async def main():
dome = DomeClient({"api_key": "your-api-key"})
ws_client = dome.polymarket.websocket
# Define event handler
def on_order_event(event: WebSocketOrderEvent):
print(f"New order: {event.data.side} {event.data.shares_normalized} shares")
print(f"Market: {event.data.market_slug}")
print(f"User: {event.data.user}")
print(f"Price: {event.data.price}")
# Connect and subscribe
await ws_client.connect()
subscription_id = await ws_client.subscribe(
users=["0x6031b6eed1c97e853c6e0f03ad3ce3529351f96d"],
on_event=on_order_event
)
print(f"Subscribed with ID: {subscription_id}")
# Keep running to receive events
await asyncio.sleep(60)
# Unsubscribe when done
await ws_client.unsubscribe(subscription_id)
await ws_client.disconnect()
asyncio.run(main())import asyncio
from dome_api_sdk import DomeClient, WebSocketOrderEvent
async def main():
dome = DomeClient({"api_key": "your-api-key"})
ws_client = dome.polymarket.websocket
def on_order_event(event: WebSocketOrderEvent):
print(f"Order event: {event.data}")
# Context manager handles connection/disconnection
async with ws_client:
subscription_id = await ws_client.subscribe(
users=["0x6031b6eed1c97e853c6e0f03ad3ce3529351f96d"],
on_event=on_order_event
)
# Keep running
await asyncio.sleep(60)
# Unsubscribe before exiting context
await ws_client.unsubscribe(subscription_id)
asyncio.run(main())import asyncio
from dome_api_sdk import DomeClient, WebSocketOrderEvent
async def main():
dome = DomeClient({"api_key": "your-api-key"})
ws_client = dome.polymarket.websocket
def on_order_event(event: WebSocketOrderEvent):
print(f"Order from subscription {event.subscription_id}: {event.data.user}")
await ws_client.connect()
# Subscribe to multiple users
sub1 = await ws_client.subscribe(
users=["0x6031b6eed1c97e853c6e0f03ad3ce3529351f96d"],
on_event=on_order_event
)
sub2 = await ws_client.subscribe(
users=["0x7c3db723f1d4d8cb9c550095203b686cb11e5c6b"],
on_event=on_order_event
)
# Get all active subscriptions
active = ws_client.get_active_subscriptions()
print(f"Active subscriptions: {len(active)}")
for sub in active:
print(f" - {sub.subscription_id}: {sub.request['filters']['users']}")
await asyncio.sleep(60)
# Unsubscribe from all
for sub in active:
await ws_client.unsubscribe(sub.subscription_id)
await ws_client.disconnect()
asyncio.run(main())The SDK automatically handles reconnection with exponential backoff (up to 10 retries) and re-subscribes to all active subscriptions:
import asyncio
from dome_api_sdk import DomeClient, WebSocketOrderEvent
async def main():
dome = DomeClient({"api_key": "your-api-key"})
ws_client = dome.polymarket.websocket
def on_order_event(event: WebSocketOrderEvent):
print(f"Order: {event.data}")
await ws_client.connect()
subscription_id = await ws_client.subscribe(
users=["0x6031b6eed1c97e853c6e0f03ad3ce3529351f96d"],
on_event=on_order_event
)
# If connection drops, SDK will:
# 1. Attempt to reconnect with exponential backoff (1s, 2s, 4s, 8s, ...)
# 2. Automatically re-subscribe to all active subscriptions
# 3. Continue receiving events seamlessly
await asyncio.sleep(300) # Run for 5 minutes
await ws_client.unsubscribe(subscription_id)
await ws_client.disconnect()
asyncio.run(main())- Automatic Reconnection: Exponential backoff up to 10 retries
- Subscription Management: Track and manage all active subscriptions
- Event Handling: Callback-based event processing
- Re-subscription: Automatically re-subscribes on reconnection
- Type Safety: Full type hints for all WebSocket messages and events
Get realized profit and loss for a wallet:
from dome_api_sdk import DomeClient
dome = DomeClient({"api_key": "your-api-key"})
wallet_pnl = dome.polymarket.wallet.get_wallet_pnl({
"wallet_address": "0x7c3db723f1d4d8cb9c550095203b686cb11e5c6b",
"granularity": "day",
"start_time": 1726857600,
"end_time": 1758316829
})
print(f"PnL data points: {len(wallet_pnl.pnl_over_time)}")Get trading activity (MERGE, SPLIT, REDEEM) for a user:
from dome_api_sdk import DomeClient
dome = DomeClient({"api_key": "your-api-key"})
activity = dome.polymarket.activity.get_activity({
"user": "0x7c3db723f1d4d8cb9c550095203b686cb11e5c6b",
"start_time": 1726857600,
"end_time": 1758316829,
"limit": 50
})
print(f"Activities: {len(activity.activities)}")Get Kalshi market data:
from dome_api_sdk import DomeClient
dome = DomeClient({"api_key": "your-api-key"})
# Get Kalshi markets
kalshi_markets = dome.kalshi.markets.get_markets({
"status": "open",
"limit": 20,
"min_volume": 10000000 # in cents
})
print(f"Kalshi markets: {len(kalshi_markets.markets)}")
# Get Kalshi markets by ticker(s)
kalshi_filtered = dome.kalshi.markets.get_markets({
"market_ticker": ["KXNFLGAME-25AUG16ARIDEN-ARI"],
"limit": 10
})Get historical Kalshi orderbook snapshots:
from dome_api_sdk import DomeClient
dome = DomeClient({"api_key": "your-api-key"})
kalshi_orderbooks = dome.kalshi.orderbooks.get_orderbooks({
"ticker": "KXNFLGAME-25AUG16ARIDEN-ARI",
"start_time": 1760470000000, # milliseconds
"end_time": 1760480000000, # milliseconds
"limit": 100
})
print(f"Kalshi orderbook snapshots: {len(kalshi_orderbooks.snapshots)}")Find equivalent markets across different platforms:
from dome_api_sdk import DomeClient
dome = DomeClient({"api_key": "your-api-key"})
# By Polymarket market slugs
matching_markets = dome.matching_markets.get_matching_markets({
"polymarket_market_slug": ["nfl-ari-den-2025-08-16"]
})
print(f"Matching Markets: {len(matching_markets.markets)}")
# By Kalshi event tickers
matching_markets_kalshi = dome.matching_markets.get_matching_markets({
"kalshi_event_ticker": ["KXNFLGAME-25AUG16ARIDEN"]
})
print(f"Kalshi Markets: {len(matching_markets_kalshi.markets)}")
# By sport and date
matching_markets_by_sport = dome.matching_markets.get_matching_markets_by_sport({
"sport": "nfl",
"date": "2025-08-16"
})
print(f"Sport Markets: {len(matching_markets_by_sport.markets)}")The SDK provides comprehensive error handling:
from dome_api_sdk import DomeClient
dome = DomeClient({"api_key": "your-api-key"})
try:
result = dome.polymarket.markets.get_market_price({
"token_id": "invalid-token"
})
except ValueError as error:
if "API Error" in str(error):
print(f"API Error: {error}")
else:
print(f"Network Error: {error}")The SDK includes a comprehensive integration test that makes live calls to the real API endpoints to verify everything works correctly.
# Run integration tests with your API key
python -m dome_api_sdk.tests.integration_test YOUR_API_KEYThis smoke test covers all endpoints with various parameter combinations and provides detailed results.
- Clone the repository:
git clone https://github.com/dome/dome-sdk-py.git
cd dome-sdk-py- Install development dependencies:
make dev-setup- Run tests:
make test- Run type checking:
make type-check- Run linting:
make lintThis project is licensed under the MIT License - see the LICENSE file for details.
- Kurush Dubash - kurush@domeapi.com
- Kunal Roy - kunal@domeapi.com