ticksupply 0.2.0

Official Python client for Ticksupply market data API

Ticksupply Python Client

Official Python client library for the Ticksupply market data API.

Installation

pip install ticksupply

Quick Start

Synchronous Usage

from ticksupply import Client

# API key from TICKSUPPLY_API_KEY environment variable
client = Client()

# Or with explicit API key
client = Client(api_key="key_xxx.secret")

# List available exchanges
for exchange in client.exchanges.list():
    print(f"{exchange.code}: {exchange.display_name}")

# Search for instruments
result = client.exchanges.list_instruments("binance", search="BTC", limit=10)
for inst in result.items:
    print(f"{inst.symbol}: {inst.base}/{inst.quote}")

# Create subscription
sub = client.subscriptions.create(datastream_id=12345)

# Get, pause, resume, delete
sub = client.subscriptions.get("sub_xxx")
client.subscriptions.pause("sub_xxx")
client.subscriptions.resume("sub_xxx")
client.subscriptions.delete("sub_xxx")

# Get activity spans
spans = client.subscriptions.list_spans("sub_xxx")
for span in spans:
    print(f"{span.started_at} - {span.ended_at or 'ongoing'}")

Asynchronous Usage

import asyncio
from ticksupply import AsyncClient

async def main():
    async with AsyncClient() as client:
        # List exchanges
        exchanges = await client.exchanges.list()
        for exchange in exchanges:
            print(f"{exchange.code}: {exchange.display_name}")

        # Iterate over all subscriptions
        async for sub in client.subscriptions.list_all():
            print(f"{sub.id}: {sub.status}")

asyncio.run(main())

Configuration

The client can be configured via constructor arguments or environment variables:

Parameter	Environment Variable	Default
api_key	TICKSUPPLY_API_KEY	(required)
base_url	-	https://api.ticksupply.com/v1
timeout	-	30.0 seconds
max_retries	-	3

Error Handling

from ticksupply import Client
from ticksupply.exceptions import (
    NotFoundError,
    RateLimitError,
    AuthenticationError,
)

client = Client()

try:
    sub = client.subscriptions.get("invalid-uuid")
except NotFoundError as e:
    print(f"Subscription not found: {e.message}")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except AuthenticationError as e:
    print(f"Invalid API key: {e.message}")

API Reference

Client Resources

• client.exchanges - List exchanges, instruments, stream types, and datastreams
• client.subscriptions - Manage data stream subscriptions
• client.exports - Create and download historical data exports
• client.availability - Check data availability

Exchanges

# List all exchanges
exchanges = client.exchanges.list()

# Get instruments for an exchange
instruments = client.exchanges.list_instruments("binance", search="BTC", limit=10)

Subscriptions

# List subscriptions
page = client.subscriptions.list(limit=10)

# Auto-paginate through all
for sub in client.subscriptions.list_all():
    print(sub.id)

# Create subscription
sub = client.subscriptions.create(datastream_id=12345)

# Get, pause, resume, delete
sub = client.subscriptions.get("sub_xxx")
client.subscriptions.pause("sub_xxx")
client.subscriptions.resume("sub_xxx")
client.subscriptions.delete("sub_xxx")

# Get activity spans
spans = client.subscriptions.list_spans("sub_xxx")
for span in spans:
    print(f"{span.started_at} - {span.ended_at or 'ongoing'}")

Exports

from datetime import datetime, timezone

# Create export
job = client.exports.create(
    datastream_id=12345,
    start_time=datetime(2025, 1, 1, tzinfo=timezone.utc),
    end_time=datetime(2025, 1, 2, tzinfo=timezone.utc),
)

# List exports
page = client.exports.list()

# Get export status
job = client.exports.get("exp_xxx")

# Get download URLs (when status is "succeeded")
download = client.exports.get_download("exp_xxx")
print(f"Files: {download.count}, Total: {download.total_bytes} bytes")
for artifact in download.artifacts:
    print(f"  {artifact.filename}: {artifact.url}")

Availability

# Check data availability for a datastream
avail = client.availability.get(datastream_id=12345)
print(f"Datastream: {avail.datastream.exchange}/{avail.datastream.instrument}")
for range_ in avail.ranges:
    print(f"{range_.from_ns} - {range_.to_ns}: ~{range_.rows_estimate} rows")

Development

# Create virtual environment (one time)
python3 -m venv .venv

# Install dev dependencies
.venv/bin/pip install -e ".[dev]"

# Run tests
.venv/bin/pytest

# Linting
.venv/bin/ruff check
.venv/bin/ruff format src/ tests/

# Type checking
.venv/bin/mypy src/

# All checks
.venv/bin/ruff check && .venv/bin/mypy src/ && .venv/bin/pytest

License

MIT