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
    """

    # Default Freqtrade pre-trade protections (issue #443).
    #
    # Defined as a plain class attribute (matching IStrategy's `protections: list`),
    # NOT an @property: hyperopt runs strategies across processes via pickle, which
    # cannot serialise bound methods/properties. These also apply in live/dry-run
    # (Freqtrade reads strategy.protections), not only during backtesting.
    # Backtesting additionally requires the --enable-protections flag, which TradAI
    # passes by default (see FreqtradeCLIBuilder / BacktestConfig.enable_protections).
    # Strategies may override or set to [] to opt out.
    protections: list[dict[str, Any]] = [
        {
            "method": "MaxDrawdown",
            "calculation_mode": "equity",
            "lookback_period_candles": 48,
            "max_allowed_drawdown": 0.20,
            "stop_duration_candles": 12,
            "trade_limit": 20,
        },
        {
            "method": "StoplossGuard",
            "lookback_period_candles": 24,
            "trade_limit": 4,
            "stop_duration_candles": 4,
        },
        {
            "method": "CooldownPeriod",
            "stop_duration_candles": 2,
        },
    ]

    @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

    # Issue #460 — slippage modeling.
    # FreqtradeBacktester._prepare_config writes `tradai_slippage_bps` into runtime_config;
    # the hooks below bias limit-order prices toward worse fills.
    #
    # WHERE THIS IS EFFECTIVE:
    #   * Live trading / dry-run: full effect. Long entries are placed at a higher limit,
    #     which fills more aggressively against the live orderbook → realistic worse fill.
    #   * Backtest: PARTIAL effect only. Freqtrade's
    #     `get_valid_entry_price_and_stake` constrains custom_entry_price within the
    #     simulated candle's [low, high] range
    #     (https://github.com/freqtrade/freqtrade/blob/develop/freqtrade/optimize/backtesting.py).
    #     A long-entry bumped above the candle high is clamped to that high, so backtest
    #     fills only worsen by the distance from proposed_rate to candle-high, not by the
    #     full `slippage_bps`. Operators reading total_fees in backtest output should treat
    #     slippage_bps as an upper bound on the *intent*, not a literal applied cost.
    #
    # Market orders bypass these hooks (Freqtrade contract). Subclasses that override
    # custom_*_price must call super() or apply_tradai_slippage, or opt out via
    # _tradai_slippage_managed = True; __init_subclass__ warns otherwise.

    _tradai_slippage_managed: ClassVar[bool] = False
    """Set True on a subclass to opt out of the slippage-bypass warning."""

    def apply_tradai_slippage(self, rate: float, *, is_long: bool, is_entry: bool) -> float:
        """Bias `rate` toward a worse fill by `tradai_slippage_bps` from self.config.

        Public so subclasses overriding custom_*_price can still apply the
        platform-managed assumption to their own computed price. Returns `rate`
        unchanged when slippage is zero (preserves float identity).

        Worse fill = pay-up for long entries and short exits, give-up for long
        exits and short entries — i.e. positive when `is_long == is_entry`.
        """
        try:
            factor = float(self.config.get("tradai_slippage_bps", 0)) / 10_000.0
        except (AttributeError, TypeError, ValueError):
            # AttributeError: self.config absent/None (defensive — Freqtrade sets it).
            return rate
        if factor == 0:
            return rate
        sign = 1 if is_long == is_entry else -1
        return rate * (1 + sign * factor)

    def custom_entry_price(
        self,
        pair: str,
        trade: Trade | None,
        current_time: datetime,
        proposed_rate: float,
        entry_tag: str | None,
        side: str,
        **kwargs: Any,
    ) -> float:
        """Bias the entry price by `tradai_slippage_bps` (Issue #460).

        Signature matches Freqtrade's IStrategy contract; `trade` is None on the first
        entry order. Freqtrade caps the deviation by `custom_price_max_distance_ratio`
        (default 2%); FreqtradeBacktester widens that cap to match the slippage so
        larger values aren't silently clamped.
        """
        return self.apply_tradai_slippage(proposed_rate, is_long=(side == "long"), is_entry=True)

    def custom_exit_price(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        proposed_rate: float,
        current_profit: float,
        exit_tag: str | None,
        **kwargs: Any,
    ) -> float:
        """Bias the exit price by `tradai_slippage_bps` (Issue #460). See custom_entry_price."""
        return self.apply_tradai_slippage(proposed_rate, is_long=not trade.is_short, is_entry=False)

    def __init_subclass__(cls, **kwargs: Any) -> None:
        # Heuristic: read the override's source and grep for `super().<hook>` or
        # `apply_tradai_slippage`. Known limits (deveshmoper round-2 review #4 —
        # accepted as best-effort):
        #   * False NEGATIVE: a comment or string literal inside the override that
        #     mentions either token silences the warning. The override would still
        #     bypass slippage at runtime.
        #   * False POSITIVE: an override that delegates the slippage adjustment
        #     through a helper method without naming either token in its own source
        #     gets warned. Workaround: name `apply_tradai_slippage` once in a
        #     comment, or set `_tradai_slippage_managed = True`.
        # The explicit `_tradai_slippage_managed` opt-out is the supported escape
        # hatch when the heuristic is wrong for a given subclass. A stricter check
        # (e.g. AST walk) was rejected as over-engineering for a warning-only path.
        super().__init_subclass__(**kwargs)
        # Opt-out is INHERITED (getattr): a subclass of a strategy that already set
        # `_tradai_slippage_managed = True` is itself managed and must not re-warn.
        if getattr(cls, "_tradai_slippage_managed", False):
            return
        for hook_name in ("custom_entry_price", "custom_exit_price"):
            # Override detection is CLASS-LOCAL (cls.__dict__) on purpose: only warn
            # about a hook THIS class defines. An inherited override was already
            # checked when its defining class was created; getattr here would re-warn
            # every subclass and even flag the base hook itself.
            override = cls.__dict__.get(hook_name)
            if override is None:
                continue
            try:
                source = inspect.getsource(override)
            except (OSError, TypeError):  # interactive class / C-impl
                continue
            if f"super().{hook_name}" in source or "apply_tradai_slippage" in source:
                continue
            warnings.warn(
                f"{cls.__name__} overrides {hook_name} without calling "
                f"super().{hook_name} or self.apply_tradai_slippage — "
                f"slippage modeling (tradai_slippage_bps) will be bypassed for this "
                f"strategy, which can overstate backtest fills (Issue #460). "
                f"Either delegate via super(), apply self.apply_tradai_slippage(...) "
                f"on your computed price, or set "
                f"`_tradai_slippage_managed = True` on the class to opt out.",
                category=RuntimeWarning,
                stacklevel=2,
            )

    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