Strategy Entities¶

class TradAIStrategy(IStrategy, StrategyDebugMixin):  # type: ignore[misc]  # IStrategy has no stubs
    """
    TradAI strategy base class.

    Note: Includes pickle support for hyperopt multiprocessing. See:
    https://github.com/freqtrade/freqtrade/issues/10287

    All TradAI strategies must extend this class and implement:
    1. get_metadata() - Return StrategyMetadata (REQUIRED)
    2. populate_indicators() - Calculate indicators (Freqtrade REQUIRED)
    3. populate_entry_trend() - Define entry signals (Freqtrade REQUIRED)
    4. populate_exit_trend() - Define exit signals (Freqtrade REQUIRED)

    Optional methods with default implementations:
    - validate_configuration() - Custom config validation
    - get_test_config() - Test configuration parameters

    Debug methods (from StrategyDebugMixin):
    - check_nan_indicators(df, pair) - Detect NaN in indicator columns
    - print_indicator_summary(df, pair) - Print indicator values for latest candle

    All Freqtrade IStrategy methods remain available:
    - custom_stoploss(), custom_exit(), leverage()
    - confirm_trade_entry(), confirm_trade_exit()
    - bot_start(), bot_loop_start()
    - And all other IStrategy lifecycle hooks

    Example:
        >>> class MyStrategy(TradAIStrategy):
        ...     timeframe = '1h'
        ...     stoploss = -0.08
        ...
        ...     def get_metadata(self):
        ...         return StrategyMetadata(
        ...             name="MyStrategy",
        ...             version="1.0.0",
        ...             description="My trading strategy",
        ...             timeframe=self.timeframe,
        ...             category=StrategyCategory.MEAN_REVERSION,
        ...             tags=["macd", "rsi"]
        ...         )
        ...
        ...     def populate_indicators(self, dataframe, metadata):
        ...         # Calculate indicators using TA-Lib
        ...         import talib
        ...         dataframe['rsi'] = talib.RSI(dataframe['close'])
        ...         # Debug: check for NaN (output when TRADAI_DEBUG=1)
        ...         self.check_nan_indicators(dataframe, metadata.get('pair'))
        ...         return dataframe
        ...
        ...     def populate_entry_trend(self, dataframe, metadata):
        ...         # Define entry logic
        ...         dataframe.loc[dataframe['rsi'] < 30, 'enter_long'] = 1
        ...         return dataframe
        ...
        ...     def populate_exit_trend(self, dataframe, metadata):
        ...         # Define exit logic (or leave empty for stoploss-only)
        ...         return dataframe
    """

    @abstractmethod
    def get_metadata(self) -> StrategyMetadata:
        """
        Return strategy metadata (REQUIRED).

        This method must be implemented by all TradAI strategies.
        Metadata is used for:
        - Strategy registration in MLflow
        - Docker image labels
        - Strategy discovery/listing
        - Documentation generation

        Returns:
            StrategyMetadata: Pydantic model with strategy information

        Example:
            >>> def get_metadata(self):
            ...     return StrategyMetadata(
            ...         name="PascalStrategyV2",
            ...         version="2.0.0",
            ...         description="MACD mean reversion with RSI filter",
            ...         timeframe=self.timeframe,
            ...         can_short=self.can_short,
            ...         category=StrategyCategory.MEAN_REVERSION,
            ...         tags=["macd", "rsi", "mean-reversion"],
            ...         parameters={
            ...             "macd_fast": self.macd_fast,
            ...             "macd_slow": self.macd_slow,
            ...         }
            ...     )
        """
        ...

    def validate_configuration(self, config: dict[str, Any]) -> bool:
        """
        Validate strategy-specific configuration (optional).

        Override this method to add custom validation logic for your strategy.
        Called during strategy initialization to ensure config is valid.

        Args:
            config: Freqtrade configuration dictionary

        Returns:
            bool: True if valid, raises exception if invalid

        Example:
            >>> def validate_configuration(self, config):
            ...     # Ensure ML model file exists
            ...     ml_path = config.get('ml_preds_filepath')
            ...     if ml_path and not os.path.exists(ml_path):
            ...         raise ValueError(f"ML predictions file not found: {ml_path}")
            ...     return True
        """
        # Default implementation: no additional validation
        return True

    def get_test_config(self) -> dict[str, Any]:
        """
        Return test configuration for unit/integration testing (optional).

        Override this method to provide strategy-specific test parameters.
        Used by testing utilities to run automated tests on your strategy.

        Returns:
            dict: Test configuration with symbols, timerange, etc.

        Example:
            >>> def get_test_config(self):
            ...     return {
            ...         "symbols": ["BTC/USDT:USDT", "ETH/USDT:USDT"],
            ...         "timerange": "20230101-20230201",
            ...         "timeframe": self.timeframe,
            ...         "stake_amount": 1000,
            ...         "dry_run": True,
            ...     }
        """
        # Default implementation: minimal config
        return {}

class StrategyMetadata(BaseModel):
    """
    Standardized metadata for all TradAI strategies.

    This model defines required and optional metadata fields that every
    TradAI strategy must provide via get_metadata() method.

    Required Fields:
        name: Strategy class name (e.g., 'PascalStrategyV2')
        version: Semantic version (e.g., '2.0.0')
        description: Short description of strategy logic
        timeframe: Primary timeframe (e.g., '1h', '15m')
        category: Trading style category (enum)

    Optional Fields:
        can_short: Supports short positions (default: False)
        tags: Searchable tags (default: [])
        author: Strategy author (default: 'TradAI Team')
        status: Lifecycle status (default: TESTING)
        parameters: Configurable parameters with defaults (default: {})
        min_freqtrade_version: Minimum Freqtrade version required

    Example:
        >>> metadata = StrategyMetadata(
        ...     name="PascalStrategyV2",
        ...     version="2.0.0",
        ...     description="MACD mean reversion with RSI filter",
        ...     timeframe="1h",
        ...     category=StrategyCategory.MEAN_REVERSION,
        ...     can_short=True,
        ...     tags=["macd", "rsi", "mean-reversion"]
        ... )
    """

    model_config = ConfigDict(frozen=True, use_enum_values=True)

    # Required fields
    name: str = Field(..., description="Strategy class name")
    version: str = Field(..., description="Semantic version (X.Y.Z)")
    description: str = Field(..., description="Short description of strategy logic")
    timeframe: str = Field(..., description="Primary timeframe (e.g., '1h', '15m')")
    category: StrategyCategory = Field(..., description="Trading style category")

    # Optional fields with defaults
    can_short: bool = Field(default=False, description="Supports short positions")
    tags: list[str] = Field(default_factory=list, description="Searchable tags")
    author: str = Field(default="TradAI Team", description="Strategy author")
    status: StrategyStatus = Field(default=StrategyStatus.TESTING, description="Lifecycle status")
    parameters: dict[str, Any] = Field(
        default_factory=dict, description="Configurable parameters with defaults"
    )
    min_freqtrade_version: str | None = Field(
        default=None, description="Minimum Freqtrade version required"
    )
    created_at: datetime = Field(
        default_factory=lambda: datetime.now(UTC), description="Creation timestamp"
    )

    # Live trading fields (LM001)
    trading_modes: list[str] = Field(
        default_factory=lambda: ["backtest"],
        description="Supported modes (backtest, hyperopt, dry-run, live)",
    )
    exchange: str | None = Field(
        default=None,
        description="Target exchange for live trading (e.g., 'binance')",
    )
    pairs: list[str] = Field(
        default_factory=list,
        description="Trading pairs list (e.g., ['BTC/USDT:USDT'])",
    )
    stake_currency: str | None = Field(
        default=None,
        description="Base currency for staking (e.g., 'USDT')",
    )
    configuration_file: str | None = Field(
        default=None,
        description="Path to strategy configuration file (S3 or local)",
    )

    @field_validator("version")
    @classmethod
    def validate_version(cls, v: str) -> str:
        """
        Validate semantic versioning format (X.Y.Z).

        Args:
            v: Version string to validate

        Returns:
            Validated version string

        Raises:
            ValueError: If version doesn't match X.Y.Z format
        """
        if not _SEMVER_PATTERN.match(v):
            raise ValueError(f"Version must be semantic versioning format (X.Y.Z): {v}")
        return v

    @field_validator("trading_modes")
    @classmethod
    def validate_trading_modes(cls, v: list[str]) -> list[str]:
        """
        Validate trading modes are valid values.

        Valid modes: backtest, hyperopt, dry-run, live

        Args:
            v: List of trading mode strings

        Returns:
            Validated list of trading modes

        Raises:
            ValueError: If any mode is not valid
        """
        valid_modes = {"backtest", "hyperopt", "dry-run", "live"}
        for mode in v:
            if mode not in valid_modes:
                raise ValueError(f"Invalid trading mode '{mode}'. Valid modes: {valid_modes}")
        return v

    @field_validator("min_freqtrade_version")
    @classmethod
    def validate_min_freqtrade_version(cls, v: str | None) -> str | None:
        """
        Validate min_freqtrade_version is a valid version string.

        Freqtrade uses versions like '2024.1', '2024.11' (YYYY.N format).
        The packaging.Version class can validate these formats.

        Args:
            v: Version string to validate

        Returns:
            Validated version string or None

        Raises:
            ValueError: If version string is invalid
        """
        if v is None:
            return v
        try:
            Version(v)  # Validates it's a valid version string
        except InvalidVersion as e:
            raise ValueError(
                f"min_freqtrade_version must be a valid version (e.g., '2024.1' or '2024.11'): {e}"
            ) from e
        return v

    def to_mlflow_tags(self) -> list[dict[str, str]]:
        """
        Convert metadata to MLflow registry tags format.

        Returns:
            List of tag dictionaries with 'key' and 'value' fields

        Example:
            >>> tags = metadata.to_mlflow_tags()
            >>> tags[0]
            {'key': 'strategy_name', 'value': 'PascalStrategyV2'}
        """
        tags = [
            {"key": "strategy_name", "value": self.name},
            {"key": "strategy_version", "value": self.version},
            {"key": "timeframe", "value": self.timeframe},
            {"key": "can_short", "value": str(self.can_short)},
            {"key": "category", "value": self.category},
            {"key": "author", "value": self.author},
            {"key": "status", "value": self.status},
        ]

        # Add individual tag entries
        for i, tag in enumerate(self.tags):
            tags.append({"key": f"tag_{i}", "value": tag})

        # Live trading tags (LM001)
        tags.append({"key": "trading_modes", "value": ",".join(self.trading_modes)})

        if self.exchange:
            tags.append({"key": "exchange", "value": self.exchange})

        if self.pairs:
            tags.append({"key": "pairs", "value": ",".join(self.pairs)})

        if self.stake_currency:
            tags.append({"key": "stake_currency", "value": self.stake_currency})

        if self.configuration_file:
            tags.append({"key": "strategy.configuration_file", "value": self.configuration_file})

        return tags

    def to_docker_labels(self) -> dict[str, str]:
        """
        Convert metadata to Docker image labels format.

        Uses 'tradai.' prefix for all labels.

        Returns:
            Dictionary of label key-value pairs

        Example:
            >>> labels = metadata.to_docker_labels()
            >>> labels['tradai.strategy.name']
            'PascalStrategyV2'
        """
        labels = {
            "tradai.strategy.name": self.name,
            "tradai.strategy.version": self.version,
            "tradai.strategy.timeframe": self.timeframe,
            "tradai.strategy.author": self.author,
            "tradai.strategy.trading_modes": ",".join(self.trading_modes),
        }

        if self.exchange:
            labels["tradai.strategy.exchange"] = self.exchange

        if self.pairs:
            labels["tradai.strategy.pairs"] = ",".join(self.pairs)

        if self.stake_currency:
            labels["tradai.strategy.stake_currency"] = self.stake_currency

        return labels

class StrategyCategory(str, Enum):
    """Trading strategy category/style."""

    TREND_FOLLOWING = "trend-following"
    MEAN_REVERSION = "mean-reversion"
    BREAKOUT = "breakout"
    MOMENTUM = "momentum"
    ARBITRAGE = "arbitrage"
    SCALPING = "scalping"
    GRID_TRADING = "grid-trading"
    DCA = "dca"
    SWING_TRADING = "swing-trading"
    MARKET_MAKING = "market-making"
    STATISTICAL_ARBITRAGE = "statistical-arbitrage"
    EVENT_DRIVEN = "event-driven"
    SENTIMENT_BASED = "sentiment-based"
    MULTI_FACTOR = "multi-factor"
    ML_DRIVEN = "ml-driven"
    PORTFOLIO_REBALANCING = "portfolio-rebalancing"

class CheckSeverity(str, Enum):
    """Severity levels for validation checks.

    Used across all validation systems (preflight, sanity, CI gate) to indicate
    the severity of issues found.

    Attributes:
        ERROR: Critical issue that blocks execution/merge
        WARNING: Issue that should be reviewed but doesn't block
        INFO: Informational message, no action required
    """

    ERROR = "error"
    WARNING = "warning"
    INFO = "info"

class ValidationIssue(BaseModel):
    """Base class for all validation issues.

    Provides common fields shared across preflight, sanity, and CI validation
    results. Specific validation systems extend this with additional fields.

    Attributes:
        rule_id: Identifier for the validation rule that triggered
        severity: Severity level of the issue
        message: Human-readable description of the issue
        details: Additional structured data about the issue
    """

    rule_id: str
    severity: CheckSeverity
    message: str
    details: dict[str, Any] = Field(default_factory=dict)

    model_config = {"frozen": True}

    @property
    def is_error(self) -> bool:
        """Check if this issue is an error (blocking)."""
        return self.severity == CheckSeverity.ERROR

    @property
    def is_warning(self) -> bool:
        """Check if this issue is a warning (non-blocking)."""
        return self.severity == CheckSeverity.WARNING

    @property
    def is_info(self) -> bool:
        """Check if this issue is informational (no action required)."""
        return self.severity == CheckSeverity.INFO

class MetricValidationIssue(ValidationIssue):
    """Validation issue for metric threshold violations.

    Extends ValidationIssue with fields specific to numeric metric checks,
    used by SanityWarning and CIViolation.

    Attributes:
        actual_value: The actual metric value from backtest/validation
        threshold: The threshold that was exceeded/not met
        suggestion: Recommended action to investigate (optional)
    """

    actual_value: float | int | None
    threshold: float | int
    suggestion: str = ""

    def format_comparison(self, comparison: str = "vs") -> str:
        """Format the metric comparison for display.

        Args:
            comparison: Comparison word/symbol (e.g., 'vs', '>', '<')

        Returns:
            Formatted string like "actual: 0.5 vs threshold: 1.0"
        """
        actual_str = f"{self.actual_value}" if self.actual_value is not None else "N/A"
        return f"actual: {actual_str} {comparison} threshold: {self.threshold}"

class ValidationResultMixin:
    """Mixin providing common aggregation properties for validation results.

    Add this mixin to validation result classes (PreflightResult, SanityCheckResult,
    CIValidationResult) to get consistent filtering and counting of issues.

    The inheriting class must have an attribute `issues: list[ValidationIssue]` or
    override the `_get_issues` method.
    """

    def _get_issues(self) -> list[ValidationIssue]:
        """Get the list of issues to aggregate.

        Override this method if the issues attribute has a different name.

        Returns:
            List of ValidationIssue instances
        """
        # Default: look for common attribute names
        for attr in ("issues", "warnings", "violations", "checks"):
            if hasattr(self, attr):
                value = getattr(self, attr)
                # Ensure it's a list before returning
                if isinstance(value, list):
                    return value
        return []

    @property
    def errors(self) -> list[ValidationIssue]:
        """Get all ERROR-level issues."""
        return [i for i in self._get_issues() if i.severity == CheckSeverity.ERROR]

    @property
    def warnings_only(self) -> list[ValidationIssue]:
        """Get all WARNING-level issues (excluding errors)."""
        return [i for i in self._get_issues() if i.severity == CheckSeverity.WARNING]

    @property
    def error_count(self) -> int:
        """Count of ERROR-level issues."""
        return len(self.errors)

    @property
    def warning_count(self) -> int:
        """Count of WARNING-level issues."""
        return len(self.warnings_only)

    @property
    def has_errors(self) -> bool:
        """Check if there are any error-level issues."""
        return self.error_count > 0

    @property
    def has_warnings(self) -> bool:
        """Check if there are any warning-level issues."""
        return self.warning_count > 0