trading-data-client 0.1.0

Python client library for Trading Data Server - fetch historical and real-time market data

Trading Data Client

A Python client library for the Trading Data Server. Provides a simple, intuitive interface for fetching historical OHLCV data and subscribing to real-time market data streams.

Features

• 📊 Historical Data: Fetch OHLCV bars via REST API
• 🔴 Real-Time Streaming: Subscribe to live market data via ZeroMQ
• ⏯️ Historical Playback: Replay historical data at controlled speeds
• 🔒 Type Safe: Full type hints with Pydantic models
• 🧵 Thread Safe: Safe for concurrent use
• 🎯 Simple API: Intuitive interface with context manager support
• ⚡ High Performance: Efficient data handling with minimal overhead
• ✅ Well Tested: 99% test coverage with 104 passing tests

Installation

pip install trading-data-client

Quick Start

Historical Data

from trading_data_client import TradingDataClient
from datetime import datetime, timedelta

# Create client
client = TradingDataClient(server_url="http://localhost:8000")

# Fetch historical bars
bars = client.get_historical_bars(
    symbol="AAPL",
    start=datetime.now() - timedelta(days=30),
    end=datetime.now(),
    timeframe="1d"
)

# Process bars
for bar in bars:
    print(f"{bar.timestamp}: O={bar.open} H={bar.high} L={bar.low} C={bar.close} V={bar.volume}")

client.close()

Real-Time Streaming

from trading_data_client import TradingDataClient

def on_bar(bar):
    print(f"New bar: {bar.symbol} @ {bar.timestamp}: Close={bar.close}")

# Create client
client = TradingDataClient(
    server_url="http://localhost:8000",
    zmq_address="tcp://localhost:5555"
)

# Subscribe to real-time stream
subscription_id = client.subscribe("AAPL", "1m", callback=on_bar, poll_interval=60)

# Keep running
try:
    client.run()  # Blocks until interrupted
except KeyboardInterrupt:
    client.close()

Historical Playback

from trading_data_client import TradingDataClient
from datetime import datetime, timedelta

def on_playback_bar(bar):
    print(f"Playback: {bar.timestamp}: Close={bar.close}")

client = TradingDataClient(
    server_url="http://localhost:8000",
    zmq_address="tcp://localhost:5555"
)

# Replay last 7 days at 10x speed
playback_id = client.playback(
    symbol="AAPL",
    start=datetime.now() - timedelta(days=7),
    end=datetime.now(),
    timeframe="1m",
    speed=10.0,
    callback=on_playback_bar
)

# Stop after some time
import time
time.sleep(60)
client.stop_playback(playback_id)
client.close()

Context Manager

from trading_data_client import TradingDataClient
from datetime import datetime, timedelta

with TradingDataClient() as client:
    bars = client.get_historical_bars(
        "AAPL",
        datetime.now() - timedelta(days=7),
        datetime.now(),
        "1d"
    )
    print(f"Fetched {len(bars)} bars")
# Automatic cleanup

Configuration

Environment Variables

export TRADING_SERVER_URL="http://localhost:8000"
export TRADING_ZMQ_ADDRESS="tcp://localhost:5555"
export TRADING_CLIENT_TIMEOUT=30
export TRADING_CLIENT_MAX_RETRIES=3

Programmatic Configuration

client = TradingDataClient(
    server_url="http://production-server:8000",
    zmq_address="tcp://production-server:5555",
    timeout=60,
    max_retries=5
)

Documentation

• Quick Start Guide
• API Reference
• Examples
• Trading Data Server

Requirements

• Python 3.10+
• Trading Data Server running and accessible

Dependencies

• requests - HTTP client for REST API
• pyzmq - ZeroMQ Python bindings for streaming
• pydantic - Data validation and serialization
• python-dateutil - Date/time parsing

Development

Install Development Dependencies

pip install -e ".[dev]"

Run Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=trading_data_client --cov-report=html

# Run specific test file
pytest tests/unit/test_models.py

# Run with timeout enforcement
pytest --timeout=60

Test Coverage

The library has 99% test coverage with 104 comprehensive tests:

• ✅ HistoricalClient: 100% coverage (21 tests)
• ✅ StreamClient: 100% coverage (26 tests)
• ✅ Main Client: 97% coverage (24 tests)
• ✅ Models: 94% coverage (15 tests)
• ✅ Utils: 100% coverage (14 tests)
• ✅ Config: 100% coverage (6 tests)

All tests pass in < 10 seconds. See FINAL_TEST_REPORT.md for details.

Type Checking

mypy trading_data_client/

Code Formatting

black trading_data_client/ tests/

Linting

flake8 trading_data_client/

Examples

See the examples/ directory for more usage examples:

• fetch_historical.py - Fetching historical data
• stream_realtime.py - Real-time streaming
• playback_historical.py - Historical playback
• context_manager.py - Context manager usage

Error Handling

from trading_data_client import (
    TradingDataClient,
    ConnectionError,
    ClientError,
    ServerError
)

client = TradingDataClient()

try:
    bars = client.get_historical_bars("AAPL", start, end, "1d")
except ConnectionError as e:
    print(f"Cannot connect to server: {e}")
except ClientError as e:
    print(f"Client error {e.status_code}: {e}")
except ServerError as e:
    print(f"Server error {e.status_code}: {e}")
finally:
    client.close()

Thread Safety

• get_historical_bars() - Thread-safe, can be called concurrently
• subscribe() / unsubscribe() - Thread-safe subscription management
• Callbacks are invoked from a background thread - ensure your callback is thread-safe

Performance

• Historical Data: < 100ms for cached data (server-dependent)
• Streaming: < 10ms from message receipt to callback invocation
• Throughput: Can handle 1000+ messages/second
• Memory: Minimal overhead, no buffering

License

MIT License - see LICENSE file for details.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Support

• GitHub Issues: https://github.com/example/trading-data-client/issues
• Documentation: https://trading-data-client.readthedocs.io

Changelog

See CHANGELOG.md for version history.