tradai-strategy¶

from tradai.strategy import TradAIStrategy, StrategyMetadata, StrategyValidator

from tradai.strategy import TradAIStrategy, StrategyMetadata
import talib.abstract as ta


class MyStrategy(TradAIStrategy):
    """Simple trend-following strategy."""

    metadata = StrategyMetadata(
        name="MyStrategy",
        version="1.0.0",
        author="TradAI",
        category="trend_following",
        description="EMA crossover strategy",
    )

    # Strategy parameters
    ema_fast = 12
    ema_slow = 26

    # Freqtrade settings
    minimal_roi = {"0": 0.1}
    stoploss = -0.05
    timeframe = "1h"

    def populate_indicators(self, dataframe, metadata):
        """Calculate technical indicators."""
        dataframe['ema_fast'] = ta.EMA(dataframe, timeperiod=self.ema_fast)
        dataframe['ema_slow'] = ta.EMA(dataframe, timeperiod=self.ema_slow)
        return dataframe

    def populate_entry_trend(self, dataframe, metadata):
        """Define entry signals."""
        dataframe.loc[
            (dataframe['ema_fast'] > dataframe['ema_slow']) &
            (dataframe['ema_fast'].shift(1) <= dataframe['ema_slow'].shift(1)),
            'enter_long'
        ] = 1
        return dataframe

    def populate_exit_trend(self, dataframe, metadata):
        """Define exit signals."""
        dataframe.loc[
            (dataframe['ema_fast'] < dataframe['ema_slow']) &
            (dataframe['ema_fast'].shift(1) >= dataframe['ema_slow'].shift(1)),
            'exit_long'
        ] = 1
        return dataframe

from tradai.strategy import StrategyMetadata, StrategyCategory, StrategyStatus

metadata = StrategyMetadata(
    name="TrendFollowingStrategy",
    version="1.0.0",
    description="EMA crossover with RSI filter",
    timeframe="1h",
    category=StrategyCategory.TREND_FOLLOWING,
    # Optional fields
    author="TradAI Team",
    tags=["trend", "ema", "rsi"],
    can_short=False,
    status=StrategyStatus.TESTING,
    exchange="binance",
    pairs=["BTC/USDT:USDT", "ETH/USDT:USDT"],
)

from tradai.strategy import StrategyValidator, ValidationLevel

# Validate strategy class
result = StrategyValidator.validate_strategy(
    strategy_class=MyStrategy,
    level=ValidationLevel.STRICT
)

if result["valid"]:
    print("Strategy passed validation")
else:
    for error in result["errors"]:
        print(f"Error: {error}")
    for warning in result["warnings"]:
        print(f"Warning: {warning}")

from tradai.strategy import validate_dataframe

# Validate indicator DataFrame
errors = validate_dataframe(
    dataframe,
    required_columns=['ema_fast', 'ema_slow', 'rsi'],
    check_nan=True,
    check_inf=True
)

if errors:
    for error in errors:
        print(f"DataFrame issue: {error}")

from tradai.strategy import CIBacktestGate, CIBacktestThresholds

# Define quality thresholds
thresholds = CIBacktestThresholds(
    min_sharpe_ratio=1.0,
    max_drawdown_pct=20.0,
    min_trade_count=50,
    min_win_rate=0.4,
    require_positive_returns=True,
)

# Create gate
gate = CIBacktestGate(thresholds=thresholds)

# Validate backtest results
result = gate.validate(backtest_result)

if result.passed:
    print("Strategy ready for deployment")
else:
    print(result.to_summary())
    for violation in result.violations:
        print(f"  {violation.rule_id}: {violation.message}")

from tradai.strategy.ci import (
    CIBacktestGate,
    CIBacktestThresholds,
    CIValidationResult,
    CIGateRuleId,      # Enum of gate rules
    CIViolation,       # Individual violation details
)

# Available gate rules
CIGateRuleId.SHARPE_TOO_LOW
CIGateRuleId.DRAWDOWN_TOO_HIGH
CIGateRuleId.INSUFFICIENT_TRADES
CIGateRuleId.NEGATIVE_RETURNS
CIGateRuleId.WIN_RATE_TOO_LOW

from tradai.strategy.preflight import (
    PreflightValidationService,
    PreflightResult,
    CheckResult,
    CheckSeverity,
)
from datetime import datetime

service = PreflightValidationService()

# Run all checks
result: PreflightResult = await service.validate(
    strategy_name="MyStrategy",
    symbols=["BTC/USDT:USDT", "ETH/USDT:USDT"],
    start_date=datetime(2024, 1, 1),
    end_date=datetime(2024, 6, 1),
    timeframe="1h",
    strict=False
)

print(f"Valid: {result.valid}, Can proceed: {result.can_proceed}")

for check in result.checks:
    status = "PASS" if check.passed else "FAIL"
    print(f"[{status}] {check.check_name}: {check.message}")

from tradai.strategy.preflight import (
    PreflightContext,          # Validation context
    DataAvailabilityResult,    # Data availability per symbol
)

from tradai.strategy.sanity import (
    SanityCheckService,
    SanityCheckResult,
    SanityRuleId,
    SanityWarning,
)

service = SanityCheckService()

# Check backtest results for issues
result: SanityCheckResult = service.check(backtest_result)

if result.warnings:
    for warning in result.warnings:
        print(f"[{warning.rule_id}] {warning.message}")

from tradai.strategy.sanity import SanityRuleId

# Available rules
SanityRuleId.WIN_RATE_TOO_HIGH      # Suspiciously high win rate
SanityRuleId.RETURNS_TOO_HIGH       # Unrealistic returns
SanityRuleId.NO_TRADES              # No trades executed
SanityRuleId.SHARPE_INCONSISTENT    # Inconsistent Sharpe ratio
SanityRuleId.PROFIT_FACTOR_EXTREME  # Extreme profit factor
SanityRuleId.MAX_DRAWDOWN_EXTREME   # Extreme max drawdown
SanityRuleId.TRADE_COUNT_LOW        # Insufficient trade count

from tradai.strategy import TradAIStrategy, StrategyDebugMixin


class MyStrategy(StrategyDebugMixin, TradAIStrategy):
    """Strategy with debug capabilities."""

    def populate_indicators(self, dataframe, metadata):
        dataframe['ema'] = ta.EMA(dataframe, timeperiod=20)

        # Debug: log indicator values
        self.debug_indicator('ema', dataframe['ema'].iloc[-1])

        return dataframe

    def populate_entry_trend(self, dataframe, metadata):
        # Debug: log signal generation
        self.debug_signal('entry_check', {
            'ema': dataframe['ema'].iloc[-1],
            'close': dataframe['close'].iloc[-1],
        })

        return dataframe

from tradai.strategy.freqai import CSVLoader
from pathlib import Path

loader = CSVLoader(path=Path("predictions/my_predictions.csv"))
predictions, do_predict = loader.load(dataframe, data_kitchen)

from tradai.strategy.freqai import MLflowLoader
from tradai.common import MLflowAdapter

adapter = MLflowAdapter(base_url="http://localhost:5000")
loader = MLflowLoader(
    adapter=adapter,
    model_uri="models:/MyStrategy/Production",  # or version: "models:/MyStrategy/3"
    validate_on_init=True  # Validates model exists and is READY
)

# Use in strategy
predictions, do_predict = loader.load(dataframe, data_kitchen)

from tradai.strategy.freqai import FreqAIConfigBuilder
from tradai.common import MLflowAdapter

mlflow_adapter = MLflowAdapter(base_url="http://localhost:5000")
builder = FreqAIConfigBuilder(mlflow_adapter=mlflow_adapter)

# Build inference config from registered model
config = builder.build_inference_config(
    base_config={"freqai": {"enabled": True}},
    model_uri="models:/MyModel/Production"
)

# Or build training config
training_config = builder.build_training_config(
    base_config=base_config,
    model_type="LightGBMClassifier",
    training_period=365,
    backtest_period=30
)

# Validate model exists before deployment
if builder.validate_model_exists("models:/MyModel/Production"):
    print("Model ready for deployment")

from tradai.strategy.freqai import (
    FreqAIConfigBuilder,       # Config builder
    TradAIPredictionModel,     # Inference-only model
    TrainingPredictionModel,   # Walk-forward training model
    CSVLoader,                 # Load predictions from CSV
    MLflowLoader,              # Load predictions from MLflow
)

from tradai.strategy import TradAIStrategy
import talib.abstract as ta


class MLStrategy(TradAIStrategy):
    """FreqAI-enabled ML strategy with MLflow predictions."""

    def populate_indicators(self, dataframe, metadata):
        # Add features for ML model
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
        dataframe['macd'] = ta.MACD(dataframe)['macd']
        dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
        return dataframe

    def feature_engineering_expand_all(self, dataframe, period, **kwargs):
        """Define features for ML model (called for each symbol).

        Features prefixed with '%-' are used as ML features.
        """
        dataframe['%-rsi'] = ta.RSI(dataframe, timeperiod=period)
        dataframe['%-macd'] = ta.MACD(dataframe, fastperiod=period)['macd']
        dataframe['%-atr'] = ta.ATR(dataframe, timeperiod=period)
        return dataframe

    def feature_engineering_expand_basic(self, dataframe, metadata, **kwargs):
        """Add non-indicator features (OHLCV transformations)."""
        dataframe['%-pct_change'] = dataframe['close'].pct_change()
        dataframe['%-volume_change'] = dataframe['volume'].pct_change()
        return dataframe

    def set_freqai_targets(self, dataframe, metadata, **kwargs):
        """Define prediction target.

        Target column must be named '&-<name>'.
        """
        # Predict next candle direction
        dataframe['&-target'] = (
            (dataframe['close'].shift(-1) > dataframe['close']).astype(int)
        )
        return dataframe

from tradai.strategy import apply_filters

def populate_entry_trend(self, dataframe, metadata):
    # First, generate entry signals
    dataframe.loc[
        (dataframe['ema_fast'] > dataframe['ema_slow']) &
        (dataframe['rsi'] < 70),
        'enter_long'
    ] = 1

    # Then, apply filters (operates on enter_long/enter_short columns)
    # - Removes signals when volume < 50% of 20-period average
    # - Removes long signals when price below SMA50
    # - Removes short signals when price above SMA50
    dataframe = apply_filters(dataframe)

    return dataframe

from tradai.strategy import apply_volatility_filter

# Filter out signals when ATR% is outside acceptable range
# Default: Keep signals when 0.5% < ATR/price < 3.0%
dataframe = apply_volatility_filter(
    dataframe,
    min_atr_multiplier=0.5,   # Minimum ATR as % of price (filter boring markets)
    max_atr_multiplier=3.0,   # Maximum ATR as % of price (filter risky markets)
)

from tradai.strategy import SignalDirection

direction = SignalDirection.LONG   # Long/buy signal
direction = SignalDirection.SHORT  # Short/sell signal

from tradai.strategy import StrategyCategory

category = StrategyCategory.TREND_FOLLOWING  # Trend-following strategies
category = StrategyCategory.MEAN_REVERSION   # Mean reversion strategies
category = StrategyCategory.BREAKOUT         # Breakout strategies
category = StrategyCategory.MOMENTUM         # Momentum strategies
category = StrategyCategory.ARBITRAGE        # Arbitrage strategies

from tradai.strategy import ValidationLevel

level = ValidationLevel.BASIC     # Minimum checks
level = ValidationLevel.STANDARD  # Standard checks
level = ValidationLevel.STRICT    # All checks

from tradai.strategy import (
    BacktestStatus,     # Status of backtest (PENDING, RUNNING, COMPLETED, FAILED)
    StrategyStatus,     # Strategy lifecycle status
    PredictionSource,   # Source of ML predictions (CSV, MLFLOW, LIVE)
)

from tradai.strategy import StrategyError, IndicatorCalculationError, InsufficientDataError

try:
    result = calculate_indicator(dataframe)
except IndicatorCalculationError as e:
    logger.error(f"Indicator failed: {e}")
    raise StrategyError("Strategy cannot proceed") from e

# Direct linting (advanced use)
from tradai.strategy.linting import StrategyLinter, lint_file

linter = StrategyLinter()
issues = linter.lint(strategy_class)

for issue in issues:
    print(f"Line {issue.line}: [{issue.severity}] {issue.message}")

# Or lint a file directly
issues = lint_file("strategies/my_strategy.py")

from tradai.strategy.preflight import PreflightValidationService

service = PreflightValidationService()
result = await service.validate(strategy_class=MyStrategy, config=config)
# Linting is included as part of preflight checks

# conftest.py
from tradai.strategy.testing.fixtures import (
    sample_dataframe,
    sample_ohlcv,
    mock_strategy,
)

def test_strategy_indicators(sample_dataframe):
    strategy = MyStrategy({})
    result = strategy.populate_indicators(sample_dataframe, {})

    assert 'ema_fast' in result.columns
    assert 'ema_slow' in result.columns
    assert not result['ema_fast'].isna().all()

from tradai.strategy import (
    # Base class
    TradAIStrategy,
    LoggerMixin,

    # CI Validation
    CIBacktestGate,
    CIBacktestThresholds,
    CIValidationResult,

    # Debug
    StrategyDebugMixin,
    debug,

    # Metadata
    StrategyMetadata,

    # Validation
    StrategyValidator,
    validate_dataframe,

    # Filters
    apply_filters,
    apply_volatility_filter,

    # Enums
    SignalDirection,
    StrategyStatus,
    StrategyCategory,
    BacktestStatus,
    ValidationLevel,
    PredictionSource,

    # Exceptions
    StrategyError,
    IndicatorCalculationError,
    InsufficientDataError,
)

# Preflight validation
from tradai.strategy.preflight import PreflightValidationService, PreflightResult

# Sanity checks
from tradai.strategy.sanity import SanityCheckService, SanityCheckResult

# FreqAI integration
from tradai.strategy.freqai import FreqAIConfigBuilder, TradAIPredictionModel

# CI details
from tradai.strategy.ci import CIGateRuleId, CIViolation

# Linting (advanced)
from tradai.strategy.linting import StrategyLinter

from datetime import datetime
from typing import ClassVar

import pandas as pd
from freqtrade.strategy import IStrategy

from tradai.strategy import (
    TradAIStrategy,
    StrategyMetadata,
    SignalDirection,
    StrategyCategory,
    apply_filters,
    validate_dataframe,
)


class MomentumCrossStrategy(TradAIStrategy, IStrategy):
    """EMA crossover strategy with volume and trend filters."""

    # Freqtrade configuration
    INTERFACE_VERSION: ClassVar[int] = 3
    timeframe = "1h"
    stoploss = -0.05
    trailing_stop = True
    trailing_stop_positive = 0.02
    startup_candle_count = 200

    # Strategy parameters
    ema_fast_period = 12
    ema_slow_period = 26
    rsi_period = 14
    rsi_overbought = 70
    rsi_oversold = 30

    @classmethod
    def get_metadata(cls) -> StrategyMetadata:
        """Strategy metadata for registry and validation."""
        return StrategyMetadata(
            name="MomentumCrossStrategy",
            version="1.2.0",
            author="TradAI Team",
            description="EMA crossover with RSI confirmation",
            category=StrategyCategory.MOMENTUM,
            pairs=["BTC/USDT:USDT", "ETH/USDT:USDT"],
            timeframe="1h",
            min_history_days=30,
            tags=["momentum", "crossover", "trend-following"],
        )

    def populate_indicators(
        self,
        dataframe: pd.DataFrame,
        metadata: dict,
    ) -> pd.DataFrame:
        """Calculate technical indicators."""
        # Validate input
        validate_dataframe(dataframe)

        # Moving averages
        dataframe["ema_fast"] = dataframe["close"].ewm(
            span=self.ema_fast_period, adjust=False
        ).mean()
        dataframe["ema_slow"] = dataframe["close"].ewm(
            span=self.ema_slow_period, adjust=False
        ).mean()

        # RSI
        delta = dataframe["close"].diff()
        gain = delta.where(delta > 0, 0).rolling(window=self.rsi_period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=self.rsi_period).mean()
        rs = gain / loss.replace(0, 1e-10)
        dataframe["rsi"] = 100 - (100 / (1 + rs))

        # Volume SMA for filter
        dataframe["volume_sma"] = dataframe["volume"].rolling(window=20).mean()

        return dataframe

    def populate_entry_trend(
        self,
        dataframe: pd.DataFrame,
        metadata: dict,
    ) -> pd.DataFrame:
        """Generate entry signals."""
        dataframe["enter_long"] = 0
        dataframe["enter_short"] = 0

        # Long: EMA cross up + RSI oversold recovery
        long_conditions = (
            (dataframe["ema_fast"] > dataframe["ema_slow"])
            & (dataframe["ema_fast"].shift(1) <= dataframe["ema_slow"].shift(1))
            & (dataframe["rsi"] > self.rsi_oversold)
            & (dataframe["rsi"] < 50)
            & (dataframe["volume"] > dataframe["volume_sma"])
        )
        dataframe.loc[long_conditions, "enter_long"] = 1

        # Short: EMA cross down + RSI overbought decline
        short_conditions = (
            (dataframe["ema_fast"] < dataframe["ema_slow"])
            & (dataframe["ema_fast"].shift(1) >= dataframe["ema_slow"].shift(1))
            & (dataframe["rsi"] < self.rsi_overbought)
            & (dataframe["rsi"] > 50)
            & (dataframe["volume"] > dataframe["volume_sma"])
        )
        dataframe.loc[short_conditions, "enter_short"] = 1

        # Apply standard filters
        dataframe = apply_filters(dataframe)

        return dataframe

    def populate_exit_trend(
        self,
        dataframe: pd.DataFrame,
        metadata: dict,
    ) -> pd.DataFrame:
        """Generate exit signals."""
        dataframe["exit_long"] = 0
        dataframe["exit_short"] = 0

        # Exit long on RSI overbought
        dataframe.loc[dataframe["rsi"] > self.rsi_overbought, "exit_long"] = 1

        # Exit short on RSI oversold
        dataframe.loc[dataframe["rsi"] < self.rsi_oversold, "exit_short"] = 1

        return dataframe

from datetime import datetime, timedelta

from tradai.strategy.preflight import PreflightValidationService, PreflightResult
from tradai.data.core.entities import SymbolList, Timeframe, DateRange
from tradai.data.infrastructure.repositories import CCXTRepository
from tradai.common import ExchangeConfig, TradingMode

# Import your strategy
from my_strategies import MomentumCrossStrategy


def validate_strategy_for_deployment() -> PreflightResult:
    """Run preflight checks before deploying strategy."""

    # 1. Create validation service
    service = PreflightValidationService()

    # 2. Get strategy metadata
    metadata = MomentumCrossStrategy.get_metadata()

    # 3. Configure validation parameters
    symbols = SymbolList.from_input(metadata.pairs)
    timeframe = Timeframe.parse(metadata.timeframe)
    date_range = DateRange(
        start=datetime.now() - timedelta(days=metadata.min_history_days),
        end=datetime.now(),
    )

    # 4. Optional: provide data repository for data availability checks
    config = ExchangeConfig(name="binance", trading_mode=TradingMode.FUTURES)
    data_repo = CCXTRepository(config=config)

    # 5. Run all preflight checks
    result = service.validate(
        strategy_class=MomentumCrossStrategy,
        symbols=symbols,
        timeframe=timeframe,
        date_range=date_range,
        data_repository=data_repo,
    )

    # 6. Process results
    print(f"Validation passed: {result.passed}")
    print(f"Total checks: {len(result.checks)}")

    for check in result.checks:
        status = "✅" if check.passed else "❌"
        print(f"  {status} {check.name}: {check.message}")

        if check.details:
            for key, value in check.details.items():
                print(f"      {key}: {value}")

    # 7. Return result for CI/CD integration
    return result


# Run validation
if __name__ == "__main__":
    result = validate_strategy_for_deployment()
    exit(0 if result.passed else 1)

from typing import ClassVar

import pandas as pd
from freqtrade.strategy import IStrategy

from tradai.strategy import (
    TradAIStrategy,
    StrategyMetadata,
    StrategyCategory,
)
from tradai.strategy.freqai import (
    FreqAIConfigBuilder,
    TradAIPredictionModel,
    PredictionSource,
)


class MLMomentumStrategy(TradAIStrategy, IStrategy):
    """FreqAI-powered momentum strategy with LightGBM."""

    INTERFACE_VERSION: ClassVar[int] = 3
    timeframe = "1h"
    stoploss = -0.05
    startup_candle_count = 300  # Extra candles for ML training

    # FreqAI configuration
    freqai_config = FreqAIConfigBuilder(
        model_name="LightGBMRegressor",
        train_period_days=30,
        backtest_period_days=7,
        live_retrain_hours=24,
        identifier="ml_momentum_v1",
    ).with_feature_parameters(
        include_timeframes=["1h", "4h"],
        include_corr_pairlist=["BTC/USDT:USDT"],
        label_period_candles=24,
        include_shifted_candles=2,
    ).with_data_split(
        train_test_split=0.85,
        shuffle=False,
    ).build()

    @classmethod
    def get_metadata(cls) -> StrategyMetadata:
        return StrategyMetadata(
            name="MLMomentumStrategy",
            version="2.0.0",
            author="TradAI Team",
            description="LightGBM momentum prediction",
            category=StrategyCategory.ML_HYBRID,
            pairs=["BTC/USDT:USDT", "ETH/USDT:USDT"],
            timeframe="1h",
            min_history_days=60,
            freqai_enabled=True,
            tags=["ml", "lightgbm", "momentum"],
        )

    def feature_engineering_expand_all(
        self,
        dataframe: pd.DataFrame,
        period: int,
        metadata: dict,
        **kwargs,
    ) -> pd.DataFrame:
        """Generate features for all timeframes."""
        # RSI
        delta = dataframe["close"].diff()
        gain = delta.where(delta > 0, 0).rolling(window=period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
        rs = gain / loss.replace(0, 1e-10)
        dataframe[f"%-rsi-period_{period}"] = 100 - (100 / (1 + rs))

        # Bollinger Bands width
        sma = dataframe["close"].rolling(window=period).mean()
        std = dataframe["close"].rolling(window=period).std()
        dataframe[f"%-bb-width-period_{period}"] = (std * 2) / sma

        # Price momentum
        dataframe[f"%-momentum-period_{period}"] = (
            dataframe["close"] / dataframe["close"].shift(period) - 1
        )

        return dataframe

    def feature_engineering_expand_basic(
        self,
        dataframe: pd.DataFrame,
        metadata: dict,
        **kwargs,
    ) -> pd.DataFrame:
        """Generate base timeframe features only."""
        # MACD
        exp1 = dataframe["close"].ewm(span=12, adjust=False).mean()
        exp2 = dataframe["close"].ewm(span=26, adjust=False).mean()
        dataframe["%-macd"] = exp1 - exp2
        dataframe["%-macd-signal"] = dataframe["%-macd"].ewm(span=9, adjust=False).mean()

        # Volume ratio
        dataframe["%-volume-ratio"] = (
            dataframe["volume"] / dataframe["volume"].rolling(window=20).mean()
        )

        return dataframe

    def set_freqai_targets(
        self,
        dataframe: pd.DataFrame,
        metadata: dict,
        **kwargs,
    ) -> pd.DataFrame:
        """Define prediction targets."""
        # Predict future returns (regression)
        dataframe["&-target"] = (
            dataframe["close"].shift(-24) / dataframe["close"] - 1
        )
        return dataframe

    def populate_indicators(
        self,
        dataframe: pd.DataFrame,
        metadata: dict,
    ) -> pd.DataFrame:
        """Run FreqAI prediction."""
        # FreqAI handles feature engineering and prediction
        dataframe = self.freqai.start(dataframe, metadata, self)

        # Prediction confidence threshold
        dataframe["prediction_confidence"] = dataframe.get(
            "&-prediction_confidence", 0.5
        )

        return dataframe

    def populate_entry_trend(
        self,
        dataframe: pd.DataFrame,
        metadata: dict,
    ) -> pd.DataFrame:
        """Generate ML-based entry signals."""
        dataframe["enter_long"] = 0
        dataframe["enter_short"] = 0

        # Get prediction from FreqAI
        prediction = dataframe.get("&-prediction", 0)
        confidence = dataframe["prediction_confidence"]

        # Long: positive prediction with high confidence
        long_conditions = (prediction > 0.01) & (confidence > 0.6)
        dataframe.loc[long_conditions, "enter_long"] = 1

        # Short: negative prediction with high confidence
        short_conditions = (prediction < -0.01) & (confidence > 0.6)
        dataframe.loc[short_conditions, "enter_short"] = 1

        return dataframe

    def populate_exit_trend(
        self,
        dataframe: pd.DataFrame,
        metadata: dict,
    ) -> pd.DataFrame:
        """Generate ML-based exit signals."""
        dataframe["exit_long"] = 0
        dataframe["exit_short"] = 0

        prediction = dataframe.get("&-prediction", 0)

        # Exit when prediction reverses
        dataframe.loc[prediction < -0.005, "exit_long"] = 1
        dataframe.loc[prediction > 0.005, "exit_short"] = 1

        return dataframe