tradai-strategy Design Document¶

from tradai.strategy import TradAIStrategy, StrategyMetadata
from tradai.strategy.enums import StrategyCategory
import talib

class MyStrategy(TradAIStrategy):
    """Example trading strategy."""

    timeframe = '1h'
    can_short = True
    stoploss = -0.08

    def get_metadata(self) -> StrategyMetadata:
        return StrategyMetadata(
            name="MyStrategy",
            version="1.0.0",
            description="RSI mean reversion",
            timeframe=self.timeframe,
            can_short=self.can_short,
            category=StrategyCategory.MEAN_REVERSION,
            tags=["rsi", "mean-reversion"],
        )

    def populate_indicators(self, dataframe, metadata):
        dataframe['rsi'] = talib.RSI(dataframe['close'])
        return dataframe

    def populate_entry_trend(self, dataframe, metadata):
        dataframe.loc[dataframe['rsi'] < 30, 'enter_long'] = 1
        return dataframe

    def populate_exit_trend(self, dataframe, metadata):
        dataframe.loc[dataframe['rsi'] > 70, 'exit_long'] = 1
        return dataframe

def get_metadata(self) -> StrategyMetadata:
    return StrategyMetadata(
        name="LiveMomentumBot",
        version="2.1.0",
        description="Production momentum strategy",
        timeframe="4h",
        category=StrategyCategory.MOMENTUM,
        # Live trading fields (LM001)
        trading_modes=["backtest", "dry-run", "live"],
        exchange="binance",
        pairs=["BTC/USDT:USDT", "ETH/USDT:USDT"],
        stake_currency="USDT",
        status=StrategyStatus.ACTIVE,
    )

from tradai.strategy.freqai import FreqAIConfigBuilder

class MLStrategy(TradAIStrategy):
    """FreqAI-enabled strategy."""

    def get_metadata(self) -> StrategyMetadata:
        return StrategyMetadata(
            name="MLStrategy",
            version="1.0.0",
            description="ML-based predictions",
            timeframe="1h",
            category=StrategyCategory.TREND_FOLLOWING,
            tags=["freqai", "ml", "gradient-boosting"],
        )

    @property
    def freqai_config(self) -> dict:
        return FreqAIConfigBuilder(
            model_name="LightGBMClassifier",
            label_period_candles=24,
            feature_parameters={
                "include_corr_pairlist": ["BTC/USDT:USDT"],
                "indicator_periods_candles": [10, 20, 50],
            },
        ).build()

from tradai.strategy import StrategyValidator
from tradai.strategy.enums import ValidationLevel

validator = StrategyValidator(level=ValidationLevel.STANDARD)
result = validator.validate(MyStrategy)

if not result.is_valid:
    for error in result.errors:
        print(f"Error: {error}")

# TradAIStrategy IS an IStrategy
class TradAIStrategy(IStrategy, StrategyDebugMixin):
    # All IStrategy methods work directly
    # Additional TradAI features added on top

class StrategyDebugMixin:
    """Debug utilities mixed into TradAIStrategy."""

    def check_nan_indicators(self, df, pair): ...
    def print_indicator_summary(self, df, pair): ...

class TradAIStrategy(IStrategy, StrategyDebugMixin):
    # Gets debug methods without inheritance hierarchy

class StrategyMetadata(BaseModel):
    model_config = ConfigDict(frozen=True)
    # All fields immutable after creation

class StrategyCategory(str, Enum):
    """str inheritance for JSON serialization."""
    TREND_FOLLOWING = "trend-following"
    MEAN_REVERSION = "mean-reversion"

# Base types (validation_entities.py)
class CheckSeverity(str, Enum):
    ERROR = "error"      # Blocks execution/merge
    WARNING = "warning"  # Review required, doesn't block
    INFO = "info"        # Informational only

class ValidationIssue(BaseModel):
    """Base class for all validation issues."""
    rule_id: str
    severity: CheckSeverity
    message: str
    details: dict[str, Any] = {}

class MetricValidationIssue(ValidationIssue):
    """For threshold violations (sanity, CI)."""
    actual_value: float | int | None
    threshold: float | int
    suggestion: str = ""

# Concrete implementations
class SanityWarning(MetricValidationIssue):     # sanity/entities.py
    rule_id: SanityRuleId

class CIViolation(MetricValidationIssue):       # ci/backtest_gate.py
    rule_id: CIGateRuleId
    is_blocking: bool = True

class ValidationResultMixin:
    """Provides consistent filtering/counting across all validation results."""

    @property
    def errors(self) -> list[ValidationIssue]:
        """Get all ERROR-level issues."""

    @property
    def warnings_only(self) -> list[ValidationIssue]:
        """Get all WARNING-level issues."""

    @property
    def error_count(self) -> int: ...
    @property
    def warning_count(self) -> int: ...
    @property
    def has_errors(self) -> bool: ...
    @property
    def has_warnings(self) -> bool: ...

# Usage in result classes
class SanityCheckResult(BaseModel, ValidationResultMixin):
    passed: bool
    warnings: list[SanityWarning]  # Mixin finds this automatically

class CIValidationResult(BaseModel, ValidationResultMixin):
    passed: bool
    violations: list[CIViolation]  # Mixin finds this automatically

from tradai.strategy.testing import (
    create_ohlcv_dataframe,      # Realistic OHLCV data
    create_trending_dataframe,   # Trending price data
    create_oversold_dataframe,   # Oversold conditions
    sample_ohlcv_fixture,        # Pytest fixture
)

# Create test data with reproducible seeds
df = create_ohlcv_dataframe(
    n_candles=200,
    base_price=50000.0,
    seed=42,  # Uses np.random.Generator for thread-safety
)

class MyStrategy(TradAIStrategy):
    """Example with proper risk configuration."""

    # Required risk parameters
    stoploss = -0.08  # -8% max loss per trade
    minimal_roi = {
        "0": 0.20,   # 20% take profit immediately
        "60": 0.10,  # 10% after 1 hour
        "120": 0.05, # 5% after 2 hours
    }

    # Position limits
    max_open_trades = 3
    stake_amount = "unlimited"  # Or fixed amount
    tradable_balance_ratio = 0.2  # Use 20% of balance

    # Advanced: Trailing stoploss
    trailing_stop = True
    trailing_stop_positive = 0.01
    trailing_only_offset_is_reached = True

    # Exchange-level protection
    stoploss_on_exchange = True  # Server-side stoploss

# Copy from: cookiecutter-tradai-strategy/{{cookiecutter.strategy_name}}/strategy.py

# Risk Management (REQUIRED - configure before live trading)
stoploss = -0.05  # -5% max loss per trade (adjust for your risk tolerance)
minimal_roi = {"0": 0.10, "60": 0.05}  # Take profit targets

# Position Limits (REQUIRED)
max_open_trades = 2  # Maximum concurrent positions
stake_amount = "unlimited"  # Or fixed: 100.0 for $100 per trade

# Optional Protections
trailing_stop = False  # Enable for trend-following strategies
stoploss_on_exchange = True  # Recommended for live trading
use_exit_signal = True  # Use exit signals from strategy