# source: https://raw.githubusercontent.com/Eksz3lsi0r/DOLLAMNY/f6178a8e9385159556a6443262fc338e43d6f88c/example_calibrated_ml_strategy.py
#!/usr/bin/env python3
"""
Example Calibrated ML Strategy

This strategy demonstrates the integration of ML calibration with FreqTrade,
showing how to use calibrated probabilities for trading decisions.

Features:
- ML model calibration using Platt scaling or Isotonic regression
- Threshold optimization for precision-recall trade-offs
- Integration with FreqAI framework
- Comprehensive hyperopt parameter support
- Calibration quality monitoring

Usage:
    freqtrade backtesting --strategy Github_Eksz3lsi0r_DOLLAMNY__example_calibrated_ml_strategy__20250916_061219 --freqai-model-type RandomForest
    freqtrade hyperopt --strategy Github_Eksz3lsi0r_DOLLAMNY__example_calibrated_ml_strategy__20250916_061219 --spaces buy sell ml_calibration
"""

from functools import reduce
import logging
import numpy as np
import pandas as pd
from pathlib import Path
import sys
from typing import Any


# Add freqtrade to path
sys.path.insert(0, str(Path(__file__).parent))

from freqtrade.strategy import IStrategy

# Import calibration parameters
from hyperopt_parameters import (
    ml_brier_score_threshold,
    # Standard ML parameters
    ml_buy_threshold,
    ml_calibrated_buy_threshold,
    ml_calibrated_sell_threshold,
    ml_calibration_cv_folds,
    # ML Calibration Parameters
    ml_calibration_enabled,
    ml_calibration_method,
    ml_confidence_threshold,
    ml_min_calibration_quality,
    ml_min_recall,
    ml_model_type,
    ml_precision_tolerance,
    ml_sell_threshold,
    ml_target_precision,
)

# Import calibration modules
from ml_calibration_system import MLCalibrationWrapper


logger = logging.getLogger(__name__)


class Github_Eksz3lsi0r_DOLLAMNY__example_calibrated_ml_strategy__20250916_061219(IStrategy):
    """
    Example strategy demonstrating ML calibration integration.

    This strategy uses calibrated ML predictions for trading decisions,
    with comprehensive threshold optimization and calibration quality monitoring.
    """

    # Strategy interface version
    INTERFACE_VERSION = 3

    # Strategy parameters
    minimal_roi = {"0": 0.1, "10": 0.05, "20": 0.02, "30": 0.01}

    stoploss = -0.1
    trailing_stop = True
    trailing_stop_positive = 0.02
    trailing_stop_positive_offset = 0.05
    trailing_only_offset_is_reached = True

    # Timeframe
    timeframe = "5m"

    # Startup candle count
    startup_candle_count: int = 30

    # Calibration parameters
    calibration_enabled = ml_calibration_enabled
    calibration_method = ml_calibration_method
    calibration_cv_folds = ml_calibration_cv_folds

    # Threshold optimization parameters
    target_precision = ml_target_precision
    min_recall = ml_min_recall
    precision_tolerance = ml_precision_tolerance

    # Calibrated thresholds
    calibrated_buy_threshold = ml_calibrated_buy_threshold
    calibrated_sell_threshold = ml_calibrated_sell_threshold

    # Calibration quality thresholds
    min_calibration_quality = ml_min_calibration_quality
    brier_score_threshold = ml_brier_score_threshold

    # Standard ML parameters
    ml_buy_threshold = ml_buy_threshold
    ml_sell_threshold = ml_sell_threshold
    ml_confidence_threshold = ml_confidence_threshold
    ml_model_type = ml_model_type

    # Calibration state
    calibrated_model = None
    calibration_quality = {}
    optimal_thresholds = {}

    def informative_pairs(self) -> list[tuple[str, str]]:
        """
        Define additional informative pairs for the strategy.
        """
        pairs = self.dp.current_whitelist()
        informative_pairs = [(pair, self.timeframe) for pair in pairs]
        return informative_pairs

    def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        """
        Populate technical indicators for the strategy.
        """
        # RSI
        dataframe["rsi"] = self.rsi(dataframe, timeperiod=14)

        # Moving averages
        dataframe["ema_short"] = self.ema(dataframe, timeperiod=9)
        dataframe["ema_long"] = self.ema(dataframe, timeperiod=21)

        # MACD
        macd = self.macd(dataframe)
        dataframe["macd"] = macd["macd"]
        dataframe["macdsignal"] = macd["macdsignal"]
        dataframe["macdhist"] = macd["macdhist"]

        # Bollinger Bands
        bollinger = self.bollinger_bands(dataframe, timeperiod=20, stddev=2)
        dataframe["bb_lowerband"] = bollinger["lower"]
        dataframe["bb_middleband"] = bollinger["mid"]
        dataframe["bb_upperband"] = bollinger["upper"]
        dataframe["bb_percent"] = (dataframe["close"] - dataframe["bb_lowerband"]) / (
            dataframe["bb_upperband"] - dataframe["bb_lowerband"]
        )

        # Volume indicators
        dataframe["volume_sma"] = dataframe["volume"].rolling(window=20).mean()
        dataframe["volume_ratio"] = dataframe["volume"] / dataframe["volume_sma"]

        # Price features
        dataframe["price_change"] = dataframe["close"].pct_change()
        dataframe["high_low_ratio"] = dataframe["high"] / dataframe["low"]

        # Lagged features for ML
        for lag in [1, 2, 3, 5]:
            dataframe[f"close_lag_{lag}"] = dataframe["close"].shift(lag)
            dataframe[f"volume_lag_{lag}"] = dataframe["volume"].shift(lag)
            dataframe[f"rsi_lag_{lag}"] = dataframe["rsi"].shift(lag)

        return dataframe

    def get_features(self, dataframe: pd.DataFrame) -> pd.DataFrame:
        """
        Extract features for ML model training.
        """
        feature_columns = [
            "rsi",
            "ema_short",
            "ema_long",
            "macd",
            "macdsignal",
            "macdhist",
            "bb_percent",
            "volume_ratio",
            "price_change",
            "high_low_ratio",
        ]

        # Add lagged features
        for lag in [1, 2, 3, 5]:
            feature_columns.extend([f"close_lag_{lag}", f"volume_lag_{lag}", f"rsi_lag_{lag}"])

        # Select features and remove NaN
        features_df = dataframe[feature_columns].dropna()

        return features_df

    def get_labels(self, dataframe: pd.DataFrame) -> pd.Series:
        """
        Generate labels for ML model training.
        """
        # Simple price direction prediction
        # 1 if price goes up in next 5 candles, 0 otherwise
        future_price = dataframe["close"].shift(-5)
        labels = (future_price > dataframe["close"]).astype(int)

        return labels.dropna()

    def train_ml_model(self, dataframe: pd.DataFrame, metadata: dict) -> Any:
        """
        Train ML model with calibration.
        """
        logger.info(f"Training ML model for {metadata['pair']}")

        # Extract features and labels
        features_df = self.get_features(dataframe)
        labels_df = self.get_labels(dataframe)

        # Align indices
        common_idx = features_df.index.intersection(labels_df.index)
        features_df = features_df.loc[common_idx]
        labels_df = labels_df.loc[common_idx]

        if len(features_df) == 0:
            logger.warning("No valid features found for training")
            return None

        # Split data
        from sklearn.model_selection import train_test_split

        X_train, X_test, y_train, y_test = train_test_split(
            features_df.values, labels_df.values, test_size=0.3, random_state=42, stratify=labels_df.values
        )
        X_train, X_val, y_train, y_val = train_test_split(
            X_train, y_train, test_size=0.3, random_state=42, stratify=y_train
        )

        # Create base model
        base_model = self._create_base_model()

        if self.calibration_enabled.value:
            # Create calibrated model
            self.calibrated_model = MLCalibrationWrapper(
                base_model=base_model,
                calibration_method=self.calibration_method.value,
                cv_folds=self.calibration_cv_folds.value,
            )

            # Fit and calibrate
            self.calibrated_model.fit(X_train, y_train, X_val, y_val)

            # Find optimal thresholds
            try:
                self.optimal_thresholds = self.calibrated_model.find_optimal_thresholds(
                    target_precision=self.target_precision.value,
                    min_recall=self.min_recall.value,
                    precision_tolerance=self.precision_tolerance.value,
                )

                logger.info(f"Optimal thresholds found: {self.optimal_thresholds}")

            except Exception as e:
                logger.warning(f"Threshold optimization failed: {e}")
                self.optimal_thresholds = {"threshold": 0.5}

            # Store calibration quality
            self.calibration_quality = self.calibrated_model.calibration_quality

            logger.info(f"Calibration quality: {self.calibration_quality}")

            return self.calibrated_model

        else:
            # Use uncalibrated model
            base_model.fit(X_train, y_train)
            return base_model

    def _create_base_model(self):
        """Create base ML model."""
        if self.ml_model_type.value == "RandomForest":
            from sklearn.ensemble import RandomForestClassifier

            return RandomForestClassifier(n_estimators=100, random_state=42)
        elif self.ml_model_type.value == "XGBoost":
            from xgboost import XGBClassifier

            return XGBClassifier(random_state=42)
        elif self.ml_model_type.value == "LightGBM":
            from lightgbm import LGBMClassifier

            return LGBMClassifier(random_state=42)
        else:
            from sklearn.ensemble import RandomForestClassifier

            return RandomForestClassifier(n_estimators=100, random_state=42)

    def get_ml_predictions(self, dataframe: pd.DataFrame) -> pd.DataFrame:
        """
        Get ML predictions for the dataframe.
        """
        if self.calibrated_model is None:
            logger.warning("No ML model available for predictions")
            return dataframe

        # Extract features
        features_df = self.get_features(dataframe)

        if len(features_df) == 0:
            logger.warning("No valid features for predictions")
            return dataframe

        try:
            # Get predictions
            if self.calibration_enabled.value:
                # Use calibrated model
                probabilities = self.calibrated_model.predict_proba(features_df.values)

                # Use optimal threshold if available
                threshold = self.optimal_thresholds.get("threshold", self.calibrated_buy_threshold.value)
                predictions = self.calibrated_model.predict(features_df.values, threshold)

                # Add calibration metadata
                dataframe.loc[features_df.index, "ml_calibration_method"] = self.calibration_method.value
                dataframe.loc[features_df.index, "ml_calibration_quality"] = self.calibration_quality.get(
                    "brier_score_calibrated", 0.0
                )

            else:
                # Use uncalibrated model
                probabilities = self.calibrated_model.predict_proba(features_df.values)
                predictions = (probabilities[:, 1] >= self.ml_buy_threshold.value).astype(int)
                threshold = self.ml_buy_threshold.value

            # Add predictions to dataframe
            dataframe.loc[features_df.index, "ml_prediction"] = predictions
            dataframe.loc[features_df.index, "ml_probability"] = probabilities[:, 1]
            dataframe.loc[features_df.index, "ml_threshold"] = threshold

            # Add confidence score
            dataframe.loc[features_df.index, "ml_confidence"] = np.abs(probabilities[:, 1] - 0.5) * 2

        except Exception as e:
            logger.error(f"ML prediction failed: {e}")
            dataframe["ml_prediction"] = 0
            dataframe["ml_probability"] = 0.5
            dataframe["ml_threshold"] = 0.5
            dataframe["ml_confidence"] = 0.0

        return dataframe

    def populate_buy_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        """
        Define buy signal logic.
        """
        # Train ML model if not already trained
        if self.calibrated_model is None:
            self.train_ml_model(dataframe, metadata)

        # Get ML predictions
        dataframe = self.get_ml_predictions(dataframe)

        # Define buy conditions
        conditions = []

        # ML prediction condition
        if self.calibration_enabled.value:
            # Use calibrated predictions
            ml_condition = (
                (dataframe["ml_prediction"] == 1)
                & (dataframe["ml_probability"] >= self.calibrated_buy_threshold.value)
                & (dataframe["ml_confidence"] >= self.ml_confidence_threshold.value)
            )

            # Add calibration quality check
            if "ml_calibration_quality" in dataframe.columns:
                ml_condition &= dataframe["ml_calibration_quality"] <= self.brier_score_threshold.value

        else:
            # Use standard ML predictions
            ml_condition = (
                (dataframe["ml_prediction"] == 1)
                & (dataframe["ml_probability"] >= self.ml_buy_threshold.value)
                & (dataframe["ml_confidence"] >= self.ml_confidence_threshold.value)
            )

        conditions.append(ml_condition)

        # Technical indicator conditions
        rsi_condition = dataframe["rsi"] < 30
        ema_condition = dataframe["ema_short"] > dataframe["ema_long"]
        bb_condition = dataframe["close"] < dataframe["bb_lowerband"]

        conditions.extend([rsi_condition, ema_condition, bb_condition])

        # Combine conditions
        dataframe.loc[reduce(lambda x, y: x & y, conditions), "buy"] = 1

        return dataframe

    def populate_sell_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        """
        Define sell signal logic.
        """
        # Get ML predictions
        dataframe = self.get_ml_predictions(dataframe)

        # Define sell conditions
        conditions = []

        # ML prediction condition
        if self.calibration_enabled.value:
            # Use calibrated predictions
            ml_condition = (dataframe["ml_prediction"] == 0) & (
                dataframe["ml_probability"] <= self.calibrated_sell_threshold.value
            )
        else:
            # Use standard ML predictions
            ml_condition = (dataframe["ml_prediction"] == 0) & (
                dataframe["ml_probability"] <= self.ml_sell_threshold.value
            )

        conditions.append(ml_condition)

        # Technical indicator conditions
        rsi_condition = dataframe["rsi"] > 70
        ema_condition = dataframe["ema_short"] < dataframe["ema_long"]
        bb_condition = dataframe["close"] > dataframe["bb_upperband"]

        conditions.extend([rsi_condition, ema_condition, bb_condition])

        # Combine conditions
        dataframe.loc[reduce(lambda x, y: x & y, conditions), "sell"] = 1

        return dataframe

    def custom_stoploss(self, pair: str, trade, current_time, current_rate, current_profit, **kwargs) -> float:
        """
        Custom stoploss logic using ML predictions.
        """
        if self.calibrated_model is None:
            return self.stoploss

        # Get current dataframe
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)

        if dataframe.empty:
            return self.stoploss

        # Get latest ML prediction
        latest = dataframe.iloc[-1]

        if "ml_probability" in latest and "ml_calibration_quality" in latest:
            # Adjust stoploss based on ML confidence and calibration quality
            ml_confidence = latest.get("ml_confidence", 0.5)
            calibration_quality = latest.get("ml_calibration_quality", 0.2)

            # More aggressive stoploss for high confidence, well-calibrated predictions
            if ml_confidence > 0.8 and calibration_quality < 0.1:
                return self.stoploss * 0.5  # Tighter stoploss
            elif ml_confidence < 0.3 or calibration_quality > 0.3:
                return self.stoploss * 1.5  # Wider stoploss

        return self.stoploss

    def custom_exit(self, pair: str, trade, current_time, current_rate, current_profit, **kwargs) -> str | None:
        """
        Custom exit logic using ML predictions.
        """
        if self.calibrated_model is None:
            return None

        # Get current dataframe
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)

        if dataframe.empty:
            return None

        # Get latest ML prediction
        latest = dataframe.iloc[-1]

        if "ml_prediction" in latest and "ml_calibration_quality" in latest:
            # Exit if ML prediction changes and calibration quality is good
            ml_prediction = latest.get("ml_prediction", 0)
            calibration_quality = latest.get("ml_calibration_quality", 0.2)

            if ml_prediction == 0 and calibration_quality < 0.15 and current_profit > 0.01:  # Only exit with profit
                return "ml_signal_exit"

        return None


def create_calibration_demo():
    """Create a demonstration of the calibration system."""
    print("ML Calibration Strategy Demo")
    print("=" * 50)

    # Create strategy instance
    strategy = Github_Eksz3lsi0r_DOLLAMNY__example_calibrated_ml_strategy__20250916_061219()

    # Print calibration parameters
    print("Calibration Parameters:")
    print(f"  Calibration Enabled: {strategy.calibration_enabled.value}")
    print(f"  Calibration Method: {strategy.calibration_method.value}")
    print(f"  CV Folds: {strategy.calibration_cv_folds.value}")
    print(f"  Target Precision: {strategy.target_precision.value}")
    print(f"  Min Recall: {strategy.min_recall.value}")
    print(f"  Precision Tolerance: {strategy.precision_tolerance.value}")

    print("\nThreshold Parameters:")
    print(f"  Calibrated Buy Threshold: {strategy.calibrated_buy_threshold.value}")
    print(f"  Calibrated Sell Threshold: {strategy.calibrated_sell_threshold.value}")
    print(f"  Min Calibration Quality: {strategy.min_calibration_quality.value}")
    print(f"  Brier Score Threshold: {strategy.brier_score_threshold.value}")

    print("\nStrategy Features:")
    print("  - ML model calibration using Platt scaling or Isotonic regression")
    print("  - Threshold optimization for precision-recall trade-offs")
    print("  - Calibration quality monitoring")
    print("  - Dynamic stoploss based on ML confidence")
    print("  - Custom exit logic using calibrated predictions")
    print("  - Comprehensive hyperopt parameter support")


if __name__ == "__main__":
    create_calibration_demo()