Skip to content

TradAI Data Library - Design Documentation

Overview

tradai-data is a clean architecture implementation for market data collection and storage, following SOLID principles and Test-Driven Development (TDD).

Test Coverage: Run just test-package libs/tradai-data for current stats.

Architecture

3-Layer Clean Architecture

┌─────────────────────────────────────────────────────────┐
│                    Core Layer                            │
│  - Entities (Value Objects)                             │
│  - Repository Interfaces (Protocols)                     │
│  - Services (Business Logic)                            │
│  - NO external dependencies                             │
└─────────────────────────────────────────────────────────┘
                         ▲
                         │ depends on
┌─────────────────────────────────────────────────────────┐
│              Infrastructure Layer                        │
│  - CCXTRepository (CCXT exchange data)                  │
│  - ArcticAdapter (S3 storage)                           │
│  - Concrete implementations                             │
└─────────────────────────────────────────────────────────┘

Core Layer

Entities (Value Objects)

Purpose: Eliminate code duplication through validated, immutable domain objects.

DateRange

  • Eliminates: 6 duplicate validation sites
  • Features:
  • Start/end validation
  • Duration calculations
  • Contains() check
  • Handles datetime/string/pandas Timestamp
  • Test Coverage: 7 tests
from tradai.data.core.entities import DateRange

# From strings
dr = DateRange.from_strings("2024-01-01", "2024-01-31")

# Properties
dr.duration_days  # 30
dr.contains(datetime(2024, 1, 15))  # True

SymbolList

  • Eliminates: 4 duplicate conversions
  • Features:
  • Automatic deduplication
  • Handles string/list inputs
  • Validation (non-empty)
  • Sorted output for APIs
  • Test Coverage: 7 tests
from tradai.data.core.entities import SymbolList

# From various inputs
sl = SymbolList.from_input("BTC/USDT:USDT")
sl = SymbolList.from_input(["BTC/USDT:USDT", "ETH/USDT:USDT"])

# Deduplication automatic
sl = SymbolList.from_input(["BTC/USDT:USDT", "BTC/USDT:USDT"])
len(sl.symbols)  # 1

# To list (sorted)
sl.to_list()  # ['BTC/USDT:USDT', 'ETH/USDT:USDT']

Timeframe

  • Eliminates: 3 fragile regex parsings
  • Features:
  • Parse string to seconds
  • Supports: s/m/h/d/w/mo
  • Complex formats ("1h 30m")
  • Unit conversions
  • Test Coverage: 7 tests
from tradai.data.core.entities import Timeframe

tf = Timeframe.parse("1h")
tf.seconds  # 3600
tf.minutes  # 60

tf = Timeframe.parse("1h 30m")
tf.seconds  # 5400

OHLCVData

  • Features:
  • Validated DataFrame wrapper
  • Required columns enforced
  • Immutable (copies returned)
  • Symbol filtering
  • Date range extraction
  • Test Coverage: 6 tests
from tradai.data.core.entities import OHLCVData

data = OHLCVData.from_dataframe(df)

data.row_count  # Number of candles
data.symbols  # frozenset of symbols
data.date_range  # DateRange object

# Get data for specific symbol
btc_data = data.get_symbol_data("BTC/USDT:USDT")

MarketSeriesKey (#808 Track D, Inv 1)

  • Eliminates: the v1 symbol-only key that collides two timeframes on one symbol and cannot represent mark/index/premium-index/funding series. Identity lives in the key, never on OHLCVData (which stays identity-free).
  • Features:
  • Frozen, hashable composite identity (exchange, symbol, timeframe, candle_type) + key_schema_version: Literal[1]
  • Canonicalized on construction — exchange via parse_exchange_key (<name>_<mode>), timeframe normalized via Timeframe.parse ("1 hour" → "1h"); extra="forbid"
  • CandleType enum (spot/futures/mark/index/premiumIndex/funding_rate), Freqtrade-aligned
  • Collision-free, versioned ms1_ codec (encode/decode) bounded to ArcticDB's ≤254-char symbol envelope — doubles as the v2 save-manifest sort key; decode rejects non-canonical input
  • Keyed MarketSeriesSaveResult + PartialSeriesStorageError/PartialSeriesReadError (typed by key, in tradai-data)
  • Test Coverage: 18 unit tests
from tradai.data import CandleType, MarketSeriesKey, decode, encode

key = MarketSeriesKey(
    exchange="binance_futures",
    symbol="BTC/USDT:USDT",
    timeframe="1h",
    candle_type=CandleType.FUTURES,
)
sym = encode(key)  # "ms1_..." — the v2 ArcticDB symbol
assert decode(sym) == key  # round-trips; non-canonical input is rejected

RequiredDatasetPlan / DatasetRequirement (#808 Track D, D2.5)

  • Eliminates: the implicit, per-caller notion of "what market data a run needs". A run needs a set of series over specific windows; this is the pure, validated container for that set — what D2.6 derives, D3 resolves against providers, D4 persists, and D6 materializes. No I/O, no provider capability (D3), no derivation (D2.6).
  • Features:
  • DatasetRequirement — frozen (key: MarketSeriesKey, requested_range, required_range, reasons: frozenset[DatasetReason]); required_range always covers requested_range (leading/warmup history), and the key's candle_type must be mode-consistent for its exchange — both enforced at construction
  • DatasetReason provenance set (primary/timeframe_detail/strategy_informative/freqai_feature/futures_auxiliary) — a series can carry several reasons; funding/mark/index provenance lives in CandleType, not here
  • RequiredDatasetPlan — non-empty, one requirement per MarketSeriesKey; from_requirements merges same-key requirements (continuous-cover ranges + unioned reasons) and orders deterministically by encoded key
  • plan_schema_version: Literal[1] + order-independent fingerprint() (SHA-256 over canonical content) — the versioned contract D4 persists and D6 verifies
  • Pure exchange-independent candle_types_for_mode / is_mode_consistent helpers (spot⇒{spot}, futures⇒{futures,mark,index,premiumIndex,funding_rate})
  • Test Coverage: 21 unit tests
from tradai.data import (
    CandleType,
    DatasetReason,
    DatasetRequirement,
    MarketSeriesKey,
    RequiredDatasetPlan,
)
from tradai.data.core.entities import DateRange

req = DatasetRequirement(
    key=MarketSeriesKey(
        exchange="binance_futures",
        symbol="BTC/USDT:USDT",
        timeframe="1h",
        candle_type=CandleType.FUTURES,
    ),
    requested_range=DateRange.from_strings("2024-02-01", "2024-03-01"),
    required_range=DateRange.from_strings("2024-01-01", "2024-03-01"),  # +warmup history
    reasons=frozenset({DatasetReason.PRIMARY}),
)
plan = RequiredDatasetPlan.from_requirements([req])  # non-empty, dedup+merged by key
digest = plan.fingerprint()  # stable, order-independent — D4 persists it, D6 verifies it

StrategyDataSpec / derive_required_dataset_plan (#808 Track D, D2.6)

  • Eliminates: three unsynchronized flat 30-calendar-day warmup pads (entrypoint/training/handler.py:34, infrastructure/backtest_executor.py:43, lambdas/data-collection-proxy/handler.py:476) and the guesswork of "which series does this run actually need". Freqtrade's real rule is per-timeframe and, under FreqAI, includes the training window — so a FreqAI run is short by startup_candle_count candles today and Freqtrade silently moves the backtest start forward (TimeRange.adjust_start_if_necessary). The deriver is pure: no I/O, no strategy loading (that is tradai.strategy.dataset.extractor), no provider capability (D3), no acquisition (D5).
  • Features:
  • StrategyDataSpec — frozen, extra="forbid", versioned (spec_schema_version) + fingerprint(); the static strategy-derived facts (timeframe, startup_candle_count, timeframe_detail, informatives, FreqAI feature params). Not run-specific, so one spec serves every run of a strategy (Track E publishes it)
  • InformativeSeries — one declared informative: timeframe, optional pair pin (None = every whitelist pair), optional non-default candle type
  • ExchangeDataOptions.for_exchange() — Freqtrade's _ft_has auxiliary knobs (mark_ohlcv_price/timeframe, funding_fee_timeframe), including hyperliquid's mark_ohlcv_price="futures"; unknown exchanges take defaults, never a guessed override
  • derive_required_dataset_plan(...) — emits PRIMARY (lead-in), TIMEFRAME_DETAIL (none — Freqtrade loads it with startup_candles=0), STRATEGY_INFORMATIVE and FREQAI_FEATURE (lead-in at their own timeframe), and FUTURES_AUXILIARY funding/mark (none). The FreqAI cross product runs over include_timeframes ∪ the base timeframe — Freqtrade prepends the base itself (config_validation.py:357-362) before building it, and omitting it dropped every corr-pair series at the base timeframe. Symbols normalized via normalize_symbols_for_exchange; required_range.start floored to each series' own candle grid (calendar-correct for w/M)
  • The lead-in itself lives in tradai.common.freqtrade.warmup, differential-tested against Freqtrade's own DataProvider.get_required_startup over 1080 combinations
  • Test Coverage: 46 unit tests (incl. mutation-checked rules and the five real strategies as a regression table)
from tradai.data import InformativeSeries, StrategyDataSpec, derive_required_dataset_plan
from tradai.data.core.entities import DateRange

spec = StrategyDataSpec(
    timeframe="1h", startup_candle_count=50, informatives=(InformativeSeries(timeframe="4h"),)
)
plan = derive_required_dataset_plan(
    spec=spec,
    exchange_key="binance_futures",
    pairs=["BTC/USDT:USDT"],
    requested_range=DateRange.from_strings("2024-03-01", "2024-06-01"),
)
# -> primary 1h (+50 candles lead-in), informative 4h (+50 *4h* candles),
#    funding_rate 1h and mark 1h (no lead-in)

Repositories (Interfaces)

Purpose: Abstract external dependencies following Dependency Inversion Principle.

DataRepository (Protocol)

  • Interface for: Data sources (Binance, mock, etc.)
  • Method: fetch_ohlcv(symbols, date_range, timeframe) -> OHLCVData
  • Test Coverage: 4 tests
from tradai.data.core.repositories import DataRepository


class MyRepository(DataRepository):
    def fetch_ohlcv(self, symbols, date_range, timeframe):
        # Implementation
        return OHLCVData.from_dataframe(df)

DataAdapter (Protocol)

  • Interface for: Storage backends (ArcticDB, PostgreSQL, etc.)
  • Methods:
  • save(data, symbols, latest_query_date)
  • load(symbols, date_range) -> OHLCVData
  • exists(symbols) -> dict[str, bool]
  • get_latest_date(symbols) -> dict[str, datetime]
  • Test Coverage: 9 tests
from tradai.data.core.repositories import DataAdapter


class MyAdapter(DataAdapter):
    def save(self, data, symbols, latest_query_date):
        # Store to backend
        pass

    def load(self, symbols, date_range):
        # Load from backend
        return OHLCVData.from_dataframe(df)

    # ... implement exists, get_latest_date

Services (Business Logic)

Purpose: Coordinate repositories and adapters with NO global state.

DataQueryService

  • Features:
  • Query with caching
  • Storage fallback
  • Dependency injection
  • No StaticScope!
  • Test Coverage: 10 tests

Query Flow: 1. Check in-memory cache (if enabled) 2. Try loading from storage adapter 3. Fetch from repository (source) 4. Save to storage adapter 5. Cache in memory

from tradai.common import ExchangeConfig, TradingMode
from tradai.data.core.services import DataQueryService
from tradai.data.infrastructure.repositories import CCXTRepository

# Setup with dependency injection
config = ExchangeConfig(name="binance", trading_mode=TradingMode.FUTURES)
repository = CCXTRepository(config)
service = DataQueryService(
    repository=repository,
    adapter=None,  # Optional
    enable_cache=True,
)

# Query with convenient API
data = service.query(
    symbols="BTC/USDT:USDT", start_date="2024-01-01", end_date="2024-01-31", timeframe="1h"
)

# Or with value objects
data = service.query(
    symbols=SymbolList.from_input(["BTC/USDT:USDT"]),
    date_range=DateRange.from_strings("2024-01-01", "2024-01-31"),
    timeframe=Timeframe.parse("1h"),
)

CoverageChecker

  • Features:
  • Check data coverage for symbols across date ranges
  • Identify gaps and missing data
  • Used by data-collection service for coverage reporting
  • Test Coverage: Unit tests in tests/unit/test_coverage.py
from tradai.data.core.coverage import CoverageChecker

checker = CoverageChecker(adapter=my_adapter)
report = checker.check_coverage(symbols, date_range, timeframe)

DataCollectionService

  • Features:
  • Batch data collection
  • Incremental updates
  • Automatic storage
  • Test Coverage: 5 tests
from tradai.common import ExchangeConfig, TradingMode
from tradai.data.core.services import DataCollectionService
from tradai.data.infrastructure.repositories import CCXTRepository

config = ExchangeConfig(name="binance", trading_mode=TradingMode.FUTURES)
service = DataCollectionService(repository=CCXTRepository(config), adapter=my_adapter)

# Collect and store
service.collect(
    symbols=["BTC/USDT:USDT", "ETH/USDT:USDT"],
    start_date="2024-01-01",
    end_date="2024-01-31",
    timeframe="1h",
)

# Incremental (only new data since last stored date)
service.collect_incremental(
    symbols="BTC/USDT:USDT", start_date="2024-01-01", end_date="2024-01-31", timeframe="1h"
)

Infrastructure Layer

CCXT Configuration (ccxt_config.py)

Centralized configuration for all CCXT repositories. Single source of truth for exchange-specific limits and supported exchanges.

ExchangeLimits (Pydantic Model)

from tradai.data.infrastructure.repositories import ExchangeLimits

# Exchange-specific configuration
# Note: Rate limiting is handled natively by CCXT (enableRateLimit=True)
limits = ExchangeLimits(
    ohlcv_limit=1000,  # Max candles per request (for pagination)
    max_concurrency=10,  # Max concurrent async requests
    ws_reconnect_delay=1.0,  # WebSocket reconnect delay (seconds)
)

Supported Exchanges

from tradai.data.infrastructure.repositories import (
    SUPPORTED_EXCHANGES,
    get_exchange_limits,
    get_exchange_class,
    is_supported_exchange,
)

# Check supported exchanges (frozenset)
print(SUPPORTED_EXCHANGES)  # {'binance', 'binanceusdm', 'hyperliquid', 'kraken', 'coinbase'}

# Get exchange-specific limits (with sensible defaults for unknown)
limits = get_exchange_limits("binance")
print(limits.ohlcv_limit)  # 1000

# Check if exchange is supported
is_supported_exchange("binance")  # True
is_supported_exchange("unknown")  # False

# Get exchange class from CCXT module (centralized resolution)
import ccxt

exchange_class = get_exchange_class("binance", ccxt)
exchange = exchange_class({"enableRateLimit": True})

CCXT Shared Utilities (ccxt_shared.py)

Common utilities shared across sync, async, and WebSocket repositories. Eliminates code duplication.

from tradai.data.infrastructure.repositories.ccxt_shared import (
    candles_to_dataframe,
    parse_timeframe_ms,
)

# Convert CCXT candles to filtered DataFrame
candles = [[1704067200000, 42000.0, 42500.0, 41800.0, 42300.0, 100.5]]
df = candles_to_dataframe(candles, "BTC/USDT:USDT", date_range)

# Parse timeframe to milliseconds (raises ValueError for invalid input)
ms = parse_timeframe_ms("1h")  # 3600000
ms = parse_timeframe_ms("1d")  # 86400000

CCXTRepository

Concrete implementation of DataRepository using CCXT. Supports any CCXT-compatible exchange.

  • Features:
  • Fetches from any CCXT-supported exchange (Binance, Hyperliquid, Kraken, etc.)
  • Automatic pagination (exchange-specific limits)
  • Error handling for API failures with partial success support
  • Multi-symbol support
  • Converts CCXT → OHLCVData
  • Test Coverage: 12 tests, 100% coverage
from tradai.common import ExchangeConfig, TradingMode
from tradai.data.infrastructure.repositories import CCXTRepository
from tradai.data.core.entities import SymbolList, DateRange, Timeframe

# Binance Futures
config = ExchangeConfig(name="binance", trading_mode=TradingMode.FUTURES)
repository = CCXTRepository(config)

# Fetch OHLCV data
data = repository.fetch_ohlcv(
    symbols=SymbolList.from_input(["BTC/USDT:USDT"]),
    date_range=DateRange.from_strings("2024-01-01", "2024-01-31"),
    timeframe=Timeframe.parse("1h"),
)

# Hyperliquid DEX with credentials from Secrets Manager
config = ExchangeConfig.from_secret("tradai/prod/hyperliquid", name="hyperliquid")
repo = CCXTRepository(config)

AsyncCCXTRepository

Async implementation of DataRepository using ccxt.async_support. Provides 3-5x performance improvement for multi-symbol fetching.

  • Features:
  • Concurrent symbol fetching with asyncio.gather()
  • Semaphore-based concurrency control (respects exchange rate limits)
  • Same API as CCXTRepository (async)
  • Test Coverage: 86%
from tradai.common import ExchangeConfig, TradingMode
from tradai.data.infrastructure.repositories import AsyncCCXTRepository

config = ExchangeConfig(name="binance", trading_mode=TradingMode.FUTURES)
repo = AsyncCCXTRepository(config)

# Fetch multiple symbols concurrently (3-5x faster)
data = await repo.fetch_ohlcv(
    symbols=SymbolList.from_input(["BTC/USDT:USDT", "ETH/USDT:USDT", "SOL/USDT:USDT"]),
    date_range=DateRange.from_strings("2024-01-01", "2024-01-31"),
    timeframe=Timeframe.parse("1h"),
)

# Required cleanup
await repo.close()

CCXTProRepository

WebSocket implementation for real-time OHLCV streaming via CCXT Pro.

  • Features:
  • Real-time candle streaming via WebSocket
  • Automatic reconnection with exponential backoff
  • Multi-symbol watching (where supported by exchange)
  • Trades-to-OHLCV fallback for unsupported exchanges
  • Test Coverage: 69%
from tradai.common import ExchangeConfig, TradingMode
from tradai.data.infrastructure.repositories import CCXTProRepository

config = ExchangeConfig(name="binance", trading_mode=TradingMode.FUTURES)
repo = CCXTProRepository(config)

await repo.connect()
try:
    async for event in repo.watch_ohlcv(symbols, timeframe):
        print(f"{event.symbol}: {event.close}")
finally:
    await repo.disconnect()

ResilientDataRepository

Decorator that wraps any DataRepository with circuit breaker protection for fault tolerance.

  • Features:
  • Circuit breaker pattern for automatic failure detection
  • Wraps any DataRepository without modifying its implementation
  • CCXT's built-in rate limiting + retry handles the rest
  • Single-exchange only (no backup exchange failover)
from tradai.common import ExchangeConfig, TradingMode
from tradai.data.infrastructure.repositories import CCXTRepository, ResilientDataRepository

config = ExchangeConfig(name="binance", trading_mode=TradingMode.FUTURES)
repo = ResilientDataRepository(CCXTRepository(config))

ResilientStreamRepository

Resilient streaming repository with automatic REST API fallback when WebSocket is unavailable.

  • Features:
  • Circuit breaker for WebSocket failures
  • Automatic fallback to REST API (CCXTRepository) when WebSocket is down
  • Graceful reconnection with exponential backoff
from tradai.data.infrastructure.repositories import (
    CCXTProRepository,
    CCXTRepository,
    ResilientStreamRepository,
)

resilient = ResilientStreamRepository(CCXTProRepository(config), CCXTRepository(config))
async for event in resilient.watch_ohlcv(symbols, timeframe):
    print(f"{event.symbol}: {event.close}")

FreqtradeDataWriter

Converter that writes OHLCVData to Freqtrade-compatible feather files.

  • Converts OHLCVData to Freqtrade's expected feather format
  • Handles symbol-to-filename conversion (e.g., BTC/USDT:USDT -> BTC_USDT_USDT-1h-futures.feather)

ArcticBacktestExecutor

Backtest executor that composes ArcticAdapter, FreqtradeDataWriter, and FreqtradeBacktester.

  • Satisfies BacktestExecutor protocol via structural typing
  • Fetches OHLCV data from ArcticDB, writes feather files, and runs Freqtrade backtests
  • Includes startup candle padding for indicator warmup

ArcticAdapter

Concrete DataAdapter implementation using ArcticDB with S3 backend for time-series storage.

  • Features:
  • Save OHLCV data to ArcticDB on S3
  • Load data from ArcticDB with date range filtering
  • Check symbol existence
  • Get latest stored dates per symbol

MarketSeriesAdapterV2 — composite-key store (#808 Track D, v2)

ArcticMarketSeriesAdapterV2 / InMemoryMarketSeriesAdapterV2 implement MarketSeriesAdapterV2 (see Repositories), storing one series per encode(MarketSeriesKey) symbol in a physically separate ArcticDB library (arctic_market_series_library, default market_series_v2) — additive alongside the v1 DataAdapter; v1/v2 reads never mix.

  • Features:
  • Multiple timeframes + mark/index/premium-index/funding series per pair (v1 collapses them)
  • Per-key independent commits (save_series → MarketSeriesSaveResult; partial failure → PartialSeriesStorageError with the failed keys, for retry-only-failed migration)
  • All-or-error reads (load_series returns the complete mapping or raises PartialSeriesReadError)
  • effective_start_date / last_candle_date preserved across saves (no wholesale metadata erase)
  • Key/value invariant: a value must carry exactly its key's symbol
  • Factory: create_market_series_adapter(bucket, library_name) (Darwin/no-bucket → InMemory)
  • Implementation: the Arctic v2 class composes an internal v1 ArcticAdapter (pointed at the v2 library) and reuses its connection / frame-prep / metadata-merge helpers — the v1 class is unchanged.
  • Test Coverage: 10 adapter unit tests + requires_emulator Arctic-over-S3 equivalence vs InMemory

Data Quality (quality/)

Data quality validation subsystem for detecting gaps, anomalies, and freshness issues in OHLCV data.

DataQualityService (quality/service.py)

  • Orchestrates data quality checks including gap detection, freshness validation, and anomaly detection
  • Produces DataQualityReport aggregating all findings

Quality Entities (quality/entities.py)

  • AnomalyType — Enum of detectable anomalies (price spike, zero volume, invalid OHLC, etc.)
  • DataGap — Represents a gap in time series data
  • DataQualityReport — Aggregated quality report for a symbol

GapDetector (quality/detectors/gap_detector.py)

  • Detects gaps in time series data where expected candles are missing

AnomalyDetector (quality/detectors/anomaly_detector.py)

  • Detects price/volume anomalies: price spikes, zero volumes, invalid OHLC relationships, duplicate timestamps

SOLID Principles Applied

Single Responsibility Principle (SRP)

  • Entities: Only data validation and conversions
  • Repositories: Only data fetching/storage
  • Services: Only business logic coordination

Open/Closed Principle (OCP)

  • Protocols: New repositories/adapters can be added without modifying core
  • Value Objects: Extensible through composition

Liskov Substitution Principle (LSP)

  • Any DataRepository implementation is substitutable
  • Any DataAdapter implementation is substitutable
  • All value objects are immutable and consistent

Interface Segregation Principle (ISP)

  • DataRepository: Single focused method (fetch_ohlcv)
  • DataAdapter: Focused methods for storage operations
  • No fat interfaces

Dependency Inversion Principle (DIP)

  • Services depend on: DataRepository, DataAdapter (abstractions)
  • Infrastructure depends on: Same abstractions
  • No dependencies on: Concrete implementations

DRY (Don't Repeat Yourself)

Before Migration

# Date validation (repeated 6 times)
start_date = to_datetime(start_date)
end_date = to_datetime(end_date)
verify_date_range(start_date, end_date)

# Symbol preparation (repeated 4 times)
unique_syms = frozenset((symbols,)) if isinstance(symbols, str) else frozenset(symbols)
if not unique_syms:
    raise ValueError("Symbols cannot be empty.")

# Timeframe parsing (repeated 3 times)
tf_seconds = to_seconds(timeframe)
pattern = r"(\d+)\s*([a-z]+)"
# ... fragile regex

After Migration

# Date validation (1 place)
date_range = DateRange.from_strings("2024-01-01", "2024-01-31")

# Symbol preparation (1 place)
symbols = SymbolList.from_input(["BTC/USDT:USDT"])

# Timeframe parsing (1 place)
timeframe = Timeframe.parse("1h")

Eliminated Patterns

❌ Global State (StaticScope)

# OLD (bad)
scope = StaticScope.instance()  # Global singleton
cache = scope.data_source_cache

# NEW (good)
service = DataQueryService(repository=repository, adapter=adapter, enable_cache=True)

❌ Dead Code

  • PlaceHolderSource: Incomplete stub removed
  • SourceFactory: Overly complex Enum pattern removed
  • 8 unused imports: Removed
  • 6 commented code blocks: Removed

❌ Tight Coupling

# OLD (bad)
class Binance(DataSource):  # Tightly coupled to DataSource
    def query(...):
        scope = StaticScope.instance()  # Global state
        # ... implementation

# NEW (good)
class CCXTRepository(DataRepository):  # Depends on abstraction
    def fetch_ohlcv(...):
        # No global state, configurable via ExchangeConfig
        # ... implementation

Testing Strategy

Test-Driven Development (TDD)

  1. RED: Write failing tests
  2. GREEN: Implement minimal code to pass
  3. REFACTOR: Improve code quality

Test Coverage

Run just test-package libs/tradai-data for current coverage. Tests cover:

  • Core Entities (Protocols, value objects)
  • Core Services (DataCollectionService, AsyncDataCollectionService)
  • Infrastructure (ArcticAdapter, CCXTRepository)
  • Integration tests (requires MiniStack)

Test Categories

Unit Tests

  • Mock all external dependencies
  • Test business logic in isolation
  • Fast execution (~6 seconds)

Integration Tests

  • Test with real CCXT (optional)
  • Marked with @pytest.mark.integration
  • Skipped by default
# Run unit tests
uv run pytest -m "not integration"

# Run integration tests
uv run pytest -m integration

# Run all tests
uv run pytest

Usage Examples

Simple Query

from tradai.common import ExchangeConfig, TradingMode
from tradai.data.core.services import DataQueryService
from tradai.data.infrastructure.repositories import CCXTRepository

config = ExchangeConfig(name="binance", trading_mode=TradingMode.FUTURES)
repository = CCXTRepository(config)
service = DataQueryService(repository=repository)

data = service.query(
    symbols="BTC/USDT:USDT", start_date="2024-01-01", end_date="2024-01-31", timeframe="1h"
)

print(f"Fetched {data.row_count} candles")

With Caching

from tradai.common import ExchangeConfig, TradingMode
from tradai.data.infrastructure.repositories import CCXTRepository

config = ExchangeConfig(name="binance", trading_mode=TradingMode.FUTURES)
service = DataQueryService(repository=CCXTRepository(config), enable_cache=True)

# First call - fetches from exchange
data1 = service.query("BTC/USDT:USDT", "2024-01-01", "2024-01-31", "1h")

# Second call - uses cache
data2 = service.query("BTC/USDT:USDT", "2024-01-01", "2024-01-31", "1h")

With Storage Adapter

from tradai.common import ExchangeConfig, TradingMode
from tradai.data.infrastructure.adapters.arctic_adapter import ArcticAdapter
from tradai.data.infrastructure.repositories import CCXTRepository

config = ExchangeConfig(name="binance", trading_mode=TradingMode.FUTURES)
adapter = ArcticAdapter(bucket="my-bucket", library_name="ohlcv")
service = DataQueryService(repository=CCXTRepository(config), adapter=adapter, enable_cache=True)

# Query flow: Cache → Storage → Source → Store & Cache
data = service.query("BTC/USDT:USDT", "2024-01-01", "2024-01-31", "1h")

Batch Collection

from tradai.common import ExchangeConfig, TradingMode
from tradai.data.core.services import DataCollectionService
from tradai.data.infrastructure.repositories import CCXTRepository

config = ExchangeConfig(name="binance", trading_mode=TradingMode.FUTURES)
collector = DataCollectionService(repository=CCXTRepository(config), adapter=adapter)

# Collect multiple symbols
collector.collect(
    symbols=["BTC/USDT:USDT", "ETH/USDT:USDT", "SOL/USDT:USDT"],
    start_date="2024-01-01",
    end_date="2024-01-31",
    timeframe="1h",
)

Incremental Updates

# Only fetches data since last stored date
collector.collect_incremental(
    symbols="BTC/USDT:USDT", start_date="2024-01-01", end_date="2024-01-31", timeframe="1h"
)

Migration from Old Codebase

Removed Files (485 lines)

  • source.py (330 lines) - Replaced by entities + repositories + CCXTRepository
  • adapter.py (155 lines) - Replaced by DataAdapter Protocol + concrete implementations

Code Reduction

  • Old: 485 lines with global state, duplication, dead code
  • New: 257 lines with clean architecture, SOLID, DRY
  • Reduction: 47% less code, 100% more maintainable

Breaking Changes

  1. No StaticScope: Use dependency injection
  2. Different API: Services instead of direct class usage
  3. Value Objects: Use entities for type safety

Migration Guide

# OLD
from libs.data.src.source import Binance

binance = Binance()
df = binance.query(
    symbols="BTC/USDT:USDT", start_date="2024-01-01", end_date="2024-01-31", timeframe="1h"
)

# NEW
from tradai.common import ExchangeConfig, TradingMode
from tradai.data.core.services import DataQueryService
from tradai.data.infrastructure.repositories import CCXTRepository

config = ExchangeConfig(name="binance", trading_mode=TradingMode.FUTURES)
service = DataQueryService(repository=CCXTRepository(config))
data = service.query(
    symbols="BTC/USDT:USDT", start_date="2024-01-01", end_date="2024-01-31", timeframe="1h"
)
df = data.to_dataframe()

Performance

Optimizations

  • Caching: In-memory cache for repeated queries
  • Pagination: Automatic for large date ranges
  • Validation: Early validation with value objects
  • Immutability: Thread-safe by default

Benchmarks (TODO)

  • Query with cache hit: <1ms
  • Query with cache miss: ~500ms (network bound)
  • Value object creation: <0.1ms

Future Enhancements

Planned Features

  1. PostgresAdapter: SQL-based storage option

Completed Features (Previously Planned)

  • ArcticAdapter — Implemented (infrastructure/adapters/arctic_adapter.py)
  • Rate Limiting — Exchange-specific limits via ccxt_config.py
  • Retry Logic — ResilientDataRepository with circuit breaker and retry
  • Multiple Exchanges — Binance, Hyperliquid, Kraken, Coinbase supported
  • WebSocket Support — CCXTProRepository and WebSocketDataCollector
  • Coverage Checking — CoverageChecker in core/coverage.py

Extensibility

  • Add new repositories: Implement DataRepository
  • Add new adapters: Implement DataAdapter
  • Add new services: Use existing repositories/adapters
  • All via dependency injection, no core changes needed

Conclusion

The tradai-data library demonstrates: - ✅ Clean Architecture - ✅ SOLID Principles - ✅ DRY (Don't Repeat Yourself) - ✅ Test-Driven Development - ✅ Comprehensive Test Coverage - ✅ Zero Dead Code - ✅ Type Safety (100% type hints) - ✅ Thread Safety (immutable entities)

Result: Production-ready, maintainable, extensible data layer for TradAI platform.