Skip to content

tradai-common Design Document

Overview

tradai-common provides shared utilities, base classes, and AWS integrations for the TradAI platform. This library follows SOLID principles and applies DRY patterns throughout.

Architecture Decisions

What We Keep (Improved)

  1. base_service.py - Base class for services with Hydra config loading

  2. ADD: Docstrings, settings_class validation

  3. KEEP: LoggerMixin composition pattern

  4. base_settings.py - Pydantic settings with S3 support

  5. ADD: Path validation, proper return types

  6. KEEP: Strict mode, frozen entities

  7. logger_mixin.py - Logging mixin for composition

  8. KEEP: Already well-designed

  9. common.py - Utility functions for time, quantization

  10. KEEP: Excellent type hints

  11. ADD: More validation where needed

  12. cache.py - Caching mechanisms

  13. KEEP: Already well-designed with Final types

  14. mlflow_client.py - MLflow REST client

  15. REMOVE: All test code (lines 530-550)

  16. IMPROVE: Error handling

What We Fix (Security)

  1. secrets_manager.py (rename from secrets_maneger.py)

  2. FIX: Thread-safe with lock

  3. FIX: Secret name from environment variables
  4. FIX: Proper error handling

  5. REMOVE: Test code

  6. ecr_util.py - ECS job manager

  7. FIX: All AWS config from environment variables

  8. REMOVE: Hardcoded ARNs, subnets, security groups
  9. REMOVE: Test code (lines 519-560)
  10. KEEP: Async patterns, CloudWatch streaming

What We Add (SOLID)

  1. exceptions.py - Custom exception hierarchy

  2. NEW: TradAIError base class

  3. NEW: ValidationError, NotFoundError, ConfigurationError, etc.

  4. repositories.py - Repository pattern ABCs

  5. NEW: Repository[T] generic base

  6. NEW: DataRepository for market data

  7. NEW: Dependency Inversion principle

  8. s3_utils.py - S3 path parsing (DRY)

  9. NEW: S3Path frozen dataclass

  10. NEW: Centralized parsing logic

What We Skip (Too Complex/Refactor Later)

  1. scope.py - Thread-unsafe, overly complex

  2. SKIP: Will redesign with dependency injection later

  3. REASON: 500+ lines, global state, tight coupling

  4. log.py - 50+ methods, too complex

  5. SKIP: Will simplify later if needed

  6. REASON: Logging is already handled by logger_mixin

  7. strategy_config_mixin.py - Service-specific logic

  8. SKIP: Move to strategy service layer later

  9. REASON: Violates Single Responsibility

Design Patterns Applied

1. Repository Pattern (Dependency Inversion)

# Abstract interface
class DataRepository(ABC):
    @abstractmethod
    def get_ohlcv(...) -> pd.DataFrame:
        pass

# Concrete implementation
class ArcticDBDataRepository(DataRepository):
    def get_ohlcv(...) -> pd.DataFrame:
        # Implementation

2. Frozen Entities (Immutability)

from pydantic import BaseModel


class Strategy(BaseModel):
    name: str
    version: str

    class Config:
        frozen = True  # Immutable

3. Mixin Pattern (Composition over Inheritance)

class LoggerMixin:
    @property
    def logger(self) -> logging.Logger:
        ...

class MyService(LoggerMixin):
    # Gets logging for free

4. Thread-Safe Singleton with lru_cache

from functools import lru_cache


@lru_cache(maxsize=1)
def get_secrets_client():
    return boto3.client("secretsmanager")

5. Factory Pattern (Already exists in source.py)

class SourceFactory:
    @staticmethod
    def create(kind: str) -> DataSource: ...

6. Handler Registry Pattern (entrypoint/base.py)

Enables dependency inversion for mode handlers, avoiding circular imports:

from tradai.common.entrypoint.base import StrategyEntrypoint
from tradai.common.entrypoint.settings import TradingMode
from tradai.common.protocols import ModeHandler

# Services register handlers using decorator
@StrategyEntrypoint.register_handler(TradingMode.BACKTEST)
class BacktestHandler:
    def __init__(self, entrypoint: StrategyEntrypoint) -> None:
        self._ep = entrypoint

    def run(self) -> int:
        # Execute backtest logic
        return 0

# Handler is invoked via registry lookup (no direct import)
handler_factory = StrategyEntrypoint.get_handler(TradingMode.BACKTEST)
handler = handler_factory(entrypoint)
return handler.run()

Benefits: - Breaks circular dependencies (common → strategy-service) - Services own their handlers - Base class remains generic

Module Organization

tradai/common/
├── __init__.py              # Public API exports
├── exceptions.py            # Custom exceptions (NEW)
├── repositories.py          # Repository ABCs (NEW)
├── base_service.py          # Service base class (IMPROVED)
├── base_settings.py         # Settings base class (IMPROVED)
├── logger_mixin.py          # Logging mixin (KEEP)
├── aws/                     # AWS integrations
│   ├── __init__.py
│   ├── secrets_manager.py   # Secrets (FIXED)
│   ├── s3_utils.py          # S3 utilities (NEW)
│   └── ecr_util.py          # ECS manager (FIXED)
├── utils/                   # Utility functions
│   ├── __init__.py
│   ├── common.py            # Time, quantization (KEEP)
│   ├── cache.py             # Caching (KEEP)
│   └── mlflow_client.py     # MLflow client (IMPROVED)
└── types.py                 # Type aliases (NEW)

Trading Mode Safety

LIVE Mode Fail-Fast (entrypoint/trading.py)

LIVE trading requires valid configuration - empty or invalid configs cause immediate failure:

def _load_strategy_config(self) -> dict[str, Any]:
    is_live = self._settings.trading_mode == TradingMode.LIVE
    try:
        config_dict = loader.load_config(...)
        self._validate_config_for_mode(config_dict)
        return config_dict
    except Exception as e:
        if is_live:
            # LIVE mode: fail immediately - never trade with empty config
            raise TradingError(f"Failed to load strategy config for LIVE trading: {e}")
        # DRY_RUN: warn and continue (for development/testing)
        self.logger.warning(f"Config load failed, using empty config: {e}")
        return {}

Unified StrategyConfig (strategy_config_loader.py)

Single canonical schema for strategy configuration:

class StrategyConfig(BaseModel):
    model_config = ConfigDict(frozen=True, extra="allow")

    name: str = Field(..., min_length=1)
    version: str = Field(..., pattern=r"^\d+\.\d+\.\d+$")
    timeframe: str = Field(..., pattern=r"^\d+[mhd]$")
    pairs: list[str] = Field(default_factory=list)
    stake_currency: str = Field(default="USDT")
    exchange: str = Field(default="binance")
    parameters: dict[str, Any] = Field(default_factory=dict)
    buy_params: dict[str, Any] = Field(default_factory=dict)
    sell_params: dict[str, Any] = Field(default_factory=dict)
    freqai: dict[str, Any] | None = Field(default=None)

Services should import from tradai.common: 

from tradai.common import StrategyConfig  # Canonical schema

Security Improvements

  1. No Hardcoded Credentials

  2. All secrets from AWS Secrets Manager

  3. Secret names from environment variables

  4. No Hardcoded Infrastructure IDs

  5. All AWS ARNs, subnets, security groups from env vars

  6. Configuration factory pattern

  7. Thread Safety

  8. Thread locks for shared state

  9. Immutable entities with Pydantic frozen=True

  10. Input Validation

  11. S3 path validation

  12. File path sanitization
  13. Pydantic validation throughout

Code Quality Standards

  1. Type Hints: 100% coverage (strict mypy)
  2. Docstrings: Google style for all public APIs
  3. Test Coverage: 85%+ target
  4. Absolute Imports: Enforced via Ruff
  5. No Dead Code: Remove all test blocks from production files

Dependencies

Core:

  • pydantic>=2.0.0 (validation, immutability)
  • pydantic-settings>=2.0.0 (environment-based config)
  • boto3>=1.35.0 (AWS SDK)

Utilities:

  • redis>=5.0.0 (caching)
  • python-json-logger>=2.0.7 (structured logging)
  • PyYAML>=6.0.1 (config files)
  • requests>=2.32.0 (HTTP client)
  • omegaconf>=2.3.0 (Hydra compatibility)
  • mlflow-skinny>=2.18.0 (experiment tracking)
  • docker>=7.0.0 (ECS job manager)

Testing Strategy

  1. Unit Tests (fast, isolated):

  2. Mock all external dependencies

  3. Test business logic only

  4. Use pytest-mock

  5. Integration Tests (slower, real services):

  6. LocalStack for AWS services

  7. Test actual AWS interactions

  8. Test Structure:

tests/
├── unit/
│   ├── test_exceptions.py
│   ├── test_base_settings.py
│   ├── aws/
│   │   ├── test_secrets_manager.py
│   │   └── test_s3_utils.py
│   └── utils/
│       └── test_cache.py
└── integration/
    └── aws/
        └── test_ecr_util.py

Migration from Original

Removed (Security/Quality)

  • ❌ secrets_maneger.py test code (lines 56-58)
  • ❌ mlflow_client.py test code (lines 530-550)
  • ❌ ecr_util.py test code (lines 519-560)
  • ❌ ecr_util.py hardcoded ARNs (lines 16-26)
  • ❌ Commented dead code in mlflow_client.py

Renamed (Correctness)

  • ✅ secrets_maneger.py → secrets_manager.py

Skipped (Complexity)

  • ⏭ scope.py (700+ lines, will redesign)
  • ⏭ log.py (50+ methods, use logger_mixin instead)
  • ⏭ strategy_config_mixin.py (move to services layer)

Added (Architecture)

  • ✅ exceptions.py (custom exception hierarchy)
  • ✅ repositories.py (repository ABCs)
  • ✅ aws/s3_utils.py (DRY S3 parsing)
  • ✅ types.py (type aliases)

Success Criteria

  • [ ] All 4 security vulnerabilities fixed
  • [ ] 85%+ test coverage
  • [ ] 100% type hint coverage
  • [ ] No hardcoded credentials or infrastructure IDs
  • [ ] All public APIs documented
  • [ ] Thread-safe implementations
  • [ ] Absolute imports only