stockapis 0.1.7

StockAPIS SDK - Python client for cryptocurrency historical data (trades, klines)

StockAPIs

Fully typed Python SDK for cryptocurrency historical data (trades, klines) from Binance, Bybit, and OKX.

Features

• All kline intervals — 1m, 5m, 15m, 1h, 4h, 1d and more
• Full type safety — Pydantic models with IDE autocomplete
• Typed data objects — OHLCV and Trade with numpy arrays
• Async & sync — Both StockAPIs and StockAPIsSync clients
• CLI — Command-line interface for quick exports
• Progress tracking — Callbacks for long-running operations
• Auto-caching — Local disk cache for repeated queries

Installation

pip install stockapis

Quick Start

Typed OHLCV Data (Recommended)

from stockapis import StockAPIs, Exchange

async with StockAPIs(api_key="your-api-key") as api:
    ohlcv = await api.exports.export_to_ohlcv(
        symbol="BTCUSDT",
        exchange=Exchange.BINANCE,
        from_date="2025-01-01",
        to_date="2025-01-07",
        interval="1h",
    )

    # Typed access with autocomplete
    print(f"Symbol: {ohlcv.symbol}")           # str
    print(f"Candles: {len(ohlcv)}")            # int
    print(f"Latest close: {ohlcv.close[-1]}")  # numpy float64
    print(f"Latest time: {ohlcv.datetime[-1]}")  # datetime
    print(f"Volatility: {ohlcv.volatility:.2%}")  # float (computed)

Sync Client

from stockapis import StockAPIsSync, Exchange

with StockAPIsSync(api_key="your-api-key") as api:
    ohlcv = api.exports.export_to_ohlcv(
        symbol="BTCUSDT",
        exchange=Exchange.BINANCE,
        from_date="2025-01-01",
        to_date="2025-01-07",
        interval="1h",
    )
    print(f"Loaded {len(ohlcv)} candles")

Type Safety

All API responses are Pydantic models with full type hints:

from stockapis import StockAPIs, Exchange, DataType

async with StockAPIs(api_key="your-key") as api:
    # Create a download request
    request = await api.downloads.create(
        symbol="BTCUSDT",
        exchange=Exchange.BINANCE,
        data_type=DataType.KLINES,
        from_date="2025-01-01",
        to_date="2025-01-07",
        interval="1m",
    )

    # DownloadRequest is a Pydantic model - full IDE support
    print(request.id)              # int
    print(request.symbol.symbol)   # str (nested SymbolNested model)
    print(request.symbol.exchange) # SymbolNestedExchange enum
    print(request.from_date)       # date
    print(request.to_date)         # date
    print(request.created_at)      # datetime
    print(request.status)          # DownloadRequestStatus enum
    print(request.progress_percent)  # float

    # Access segments (list of typed DownloadSegment)
    for segment in request.segments:
        print(segment.target_date)      # date
        print(segment.status)           # DownloadSegmentStatus enum
        print(segment.records_ingested) # int
        print(segment.completed_at)     # datetime | None

Response Models

Model	Description
DownloadRequest	Download job with segments
DownloadSegment	Individual day segment
ExportJob	Export job status
DataAvailabilityResponse	Coverage info
AvailabilityGridResponse	Monthly grid
SymbolsWithDataResponse	Available symbols

All models have proper types: date, datetime, enums, nested models.

Typed Data Objects

OHLCV (Candlestick Data)

from stockapis import OHLCV

ohlcv = await api.exports.export_to_ohlcv(...)

# Typed numpy arrays
ohlcv.timestamp   # NDArray[np.int64]  — Unix ms
ohlcv.datetime    # list[datetime]    — Python datetime
ohlcv.open        # NDArray[np.float64]
ohlcv.high        # NDArray[np.float64]
ohlcv.low         # NDArray[np.float64]
ohlcv.close       # NDArray[np.float64]
ohlcv.volume      # NDArray[np.float64]

# Metadata
ohlcv.symbol      # str
ohlcv.exchange    # str
ohlcv.interval    # str

# Computed properties
ohlcv.returns     # NDArray[np.float64] — log returns
ohlcv.volatility  # float — annualized volatility
len(ohlcv)        # int — number of candles

Trade Data

from stockapis import Trade

trades = await api.exports.export_to_trades(...)

# Typed numpy arrays
trades.timestamp      # NDArray[np.int64]
trades.datetime       # list[datetime]
trades.price          # NDArray[np.float64]
trades.quantity       # NDArray[np.float64]
trades.is_buyer_maker # NDArray[np.bool_]

# Metadata
trades.symbol         # str
trades.exchange       # str
len(trades)           # int — number of trades

Status Handling

SDK properly handles all statuses including partial data availability:

from stockapis import (
    DownloadRequestStatus,
    SUCCESS_DOWNLOAD_STATUSES,
    TERMINAL_REQUEST_STATUSES,
)

# Wait for download
request = await api.downloads.download_and_wait(...)

# Check status with enums
status = DownloadRequestStatus(request.status)

if status in SUCCESS_DOWNLOAD_STATUSES:
    # COMPLETED, SKIPPED, or PARTIAL — data is available
    print(f"Success: {request.total_records_ingested} records")

elif status == DownloadRequestStatus.DEFERRED:
    # Data not yet available from exchange (1-2 day delay)
    print("Data will be available tomorrow")

elif status == DownloadRequestStatus.FAILED:
    # Check error in segments
    for seg in request.segments:
        if seg.error_message:
            print(f"{seg.target_date}: {seg.error_message}")

Status Reference

Request statuses (DownloadRequestStatus):

Status	Description
PENDING	Not started
PROCESSING	In progress
COMPLETED	All done
SKIPPED	Data already exists
PARTIAL	Some completed, some deferred
DEFERRED	All segments deferred
FAILED	At least one failed

Note: PARTIAL is a success — available data was downloaded. DEFERRED occurs when requesting data too close to today (exchanges publish with 1-2 day delay).

Export Methods

To DataFrame (Polars)

df = await api.exports.export_to_dataframe(
    symbol="BTCUSDT",
    exchange=Exchange.BINANCE,
    data_type=DataType.KLINES,
    from_date="2025-01-01",
    to_date="2025-01-07",
    interval="1m",
)
print(f"Loaded {len(df)} rows")

To File

path = await api.exports.export_to_file(
    symbol="BTCUSDT",
    exchange=Exchange.BINANCE,
    data_type=DataType.KLINES,
    from_date="2025-01-01",
    to_date="2025-01-07",
    interval="1m",
    output_path="btcusdt.csv",
    on_progress=lambda p: print(f"Progress: {p}%"),
)

Download to QuestDB

result = await api.downloads.download_to_questdb(
    symbol="BTCUSDT",
    exchange=Exchange.BINANCE,
    data_type=DataType.TRADES,
    from_date="2025-01-01",
    to_date="2025-01-31",
)
print(f"Records: {result['total_records']}")

CLI

# Set API key
export STOCKAPIS_API_KEY=your-api-key

# Export klines to file
stockapis export BTCUSDT -e binance --from 2025-01-01 --to 2025-01-07 -i 1m -o data.csv

# Download trades to QuestDB
stockapis download BTCUSDT -e binance --from 2025-01-01 --to 2025-01-31 --type trades

# Check availability
stockapis availability BTCUSDT -e binance --from 2024-01-01 --to 2024-12-31

# List jobs
stockapis list downloads
stockapis list exports

# Check status
stockapis status 123

Low-Level API

For fine-grained control:

# Create request
request = await api.downloads.create(
    symbol="BTCUSDT",
    exchange=Exchange.BINANCE,
    data_type=DataType.TRADES,
    from_date="2025-01-01",
    to_date="2025-01-07",
)

# Poll progress
async for req in api.downloads.poll_progress(request.id):
    print(f"Progress: {req.progress_percent}%")

# Or wait for completion
completed = await api.downloads.wait_for_completion(
    request.id,
    poll_interval=5.0,
    timeout=3600,
)

Check Data Availability

# Range availability (with interval for klines)
avail = await api.downloads.check_availability(
    symbol="BTCUSDT",
    exchange=Exchange.BINANCE,
    data_type=DataType.KLINES,
    from_date="2024-01-01",
    to_date="2024-12-31",
    interval="1m",  # Check specific interval
)
print(f"Coverage: {avail.coverage_percent}%")
print(f"Missing days: {avail.missing_days}")  # list[date]

# Monthly grid
grid = await api.downloads.get_availability_grid(
    symbol="BTCUSDT",
    exchange=Exchange.BINANCE,
    data_type=DataType.KLINES,
    interval="1m",
    year=2024,
)
for month in grid.months:
    print(f"{month.month}/{month.year}: {month.status}")

Configuration

# Production (default)
api = StockAPIs(api_key="sk_live_xxx")

# Development
api = StockAPIs(api_key="demo_stockapis_2024", mode="dev")

# Custom URL
api = StockAPIs(api_key="your-key", base_url="http://custom:9000")

API Keys

Type	Prefix	Usage
Demo	demo_	Testing
Production	sk_	Production

Supported Intervals

All standard kline intervals are supported:

Interval	Description
1m	1 minute
3m	3 minutes
5m	5 minutes
15m	15 minutes
30m	30 minutes
1h	1 hour
2h	2 hours
4h	4 hours
6h	6 hours
8h	8 hours
12h	12 hours
1d	1 day
3d	3 days
1w	1 week
1M	1 month

Enums

from stockapis import (
    Exchange,           # BINANCE, BYBIT, OKX
    MarketType,         # SPOT, LINEAR, INVERSE
    DataType,           # KLINES, TRADES
    Granularity,        # DAILY, MONTHLY
    FileFormat,         # CSV, PARQUET, JSON
    Compression,        # NONE, GZIP, ZIP
    DownloadStatus,     # Segment statuses
    DownloadRequestStatus,  # Request statuses
    ExportJobStatus,    # Export statuses
    AvailabilityStatus, # COMPLETE, PARTIAL, EMPTY
    KlineInterval,      # Type alias for intervals
)

Error Handling

import httpx

try:
    df = await api.exports.export_to_dataframe(...)
except httpx.HTTPStatusError as e:
    print(f"API error {e.response.status_code}")
except ValueError as e:
    print(f"Validation error: {e}")
except TimeoutError as e:
    print(f"Timeout: {e}")
except ImportError as e:
    print(f"Missing polars: {e}")

License

MIT