DualAssetSelectionStrategy

# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
# isort: skip_file
# --- Do not remove these libs ---
import numpy as np
import pandas as pd
from pandas import DataFrame
from datetime import datetime
from typing import Optional, Union, List
from functools import reduce

from freqtrade.strategy import (
    IStrategy,
    Trade,
    Order,
    PairLocks,
    informative,
    merge_informative_pair,
    stoploss_from_open,
    stoploss_from_absolute,
)

# --------------------------------
# Add your lib to import here
import talib.abstract as ta
import freqtrade.vendor.qtpylib.indicators as qtpylib

# Import our custom pair selector
from user_data.strategies.pair_selector import PairSelector, calculate_setup_indicators


class DualAssetSelectionStrategy(IStrategy):
    """
    Master Dual-Asset Selection Strategy for Kraken BTC/USD and ETH/USD trading.

    Implements the research-defined approach:
    - Scans both BTC/USD and ETH/USD daily for best setups
    - Uses 7+ out of 10 scoring quality gate
    - Targets 60% BTC trades (swing), 40% ETH trades (momentum)
    - Single $1000 position focus with 4-6% minimum profit targets

    Based on Strategy Analysis Report and Kraken Trading Pairs Analysis.
    """

    # Strategy interface version
    INTERFACE_VERSION = 3

    # Dynamic timeframe (adjusted based on selected pair)
    timeframe = "1h"  # Default, will be overridden by pair selection

    # Can this strategy go short?
    can_short = False

    # Dynamic ROI (adjusted based on selected pair/strategy)
    minimal_roi = {
        "0": 0.08,  # 8% target (adjusted per pair)
        "60": 0.06,  # 6% after 1 hour
        "120": 0.04,  # 4% after 2 hours (minimum for fees)
        "240": 0.01,  # 1% after 4 hours (safety exit)
    }

    # Dynamic stoploss (adjusted based on pair volatility)
    stoploss = -0.03  # 3% default (2.5% for BTC, 3% for ETH)

    # Trailing stoploss
    trailing_stop = False

    # Dual-asset selection parameters
    startup_candle_count: int = 50  # Enough for all indicators

    # Pair selection settings
    quality_gate_threshold = 7.0  # 7+ out of 10 to trade
    btc_allocation_target = 0.60  # 60% BTC trades
    eth_allocation_target = 0.40  # 40% ETH trades

    # Initialize pair selector
    def __init__(self, config: dict) -> None:
        super().__init__(config)
        self.pair_selector = PairSelector()
        self.latest_selection_result = None

    def informative_pairs(self):
        """
        Define informative pairs for both BTC and ETH analysis
        """
        return [
            ("BTC/USD", "4h"),  # BTC trend context
            ("ETH/USD", "4h"),  # ETH trend context
        ]

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Run all indicator calculations for the asset.
        """
        pair = metadata["pair"]
        asset_type = "BTC" if "BTC" in pair else "ETH"

        # Use the centralized indicator calculation function
        dataframe = calculate_setup_indicators(dataframe, asset_type=asset_type)

        # Merge informative data (for 4h trend context)
        if self.dp:
            informative_btc = self.dp.get_pair_dataframe("BTC/USD", "4h")
            informative_eth = self.dp.get_pair_dataframe("ETH/USD", "4h")

            if informative_btc is not None and not informative_btc.empty:
                informative_btc = self.populate_indicators_btc_4h(informative_btc, {})
                dataframe = merge_informative_pair(
                    dataframe, informative_btc, self.timeframe, "4h", ffill=True
                )

            if informative_eth is not None and not informative_eth.empty:
                informative_eth = self.populate_indicators_eth_4h(informative_eth, {})
                dataframe = merge_informative_pair(
                    dataframe, informative_eth, self.timeframe, "4h", ffill=True
                )

        return dataframe

    @informative("4h", "BTC/USD")
    def populate_indicators_btc_4h(
        self, dataframe: DataFrame, metadata: dict
    ) -> DataFrame:
        """4h BTC trend context"""
        dataframe["ema_trend_4h"] = ta.EMA(dataframe, timeperiod=20) > ta.EMA(
            dataframe, timeperiod=50
        )
        dataframe["rsi_4h"] = ta.RSI(dataframe, timeperiod=14)
        macd_4h = ta.MACD(dataframe)
        dataframe["trend_strength_4h"] = macd_4h["macd"] > macd_4h["macdsignal"]
        return dataframe

    @informative("4h", "ETH/USD")
    def populate_indicators_eth_4h(
        self, dataframe: DataFrame, metadata: dict
    ) -> DataFrame:
        """4h ETH trend context"""
        dataframe["ema_trend_4h"] = ta.EMA(dataframe, timeperiod=20) > ta.EMA(
            dataframe, timeperiod=50
        )
        dataframe["rsi_4h"] = ta.RSI(dataframe, timeperiod=14)
        macd_4h = ta.MACD(dataframe)
        dataframe["trend_strength_4h"] = macd_4h["macd"] > macd_4h["macdsignal"]
        dataframe["volume_trend_4h"] = dataframe["volume"] > ta.SMA(
            dataframe["volume"], timeperiod=10
        )
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Perform dual-asset selection and generate entry signal for the best pair.
        """
        # Initialize entry column
        dataframe["enter_long"] = 0

        # The dataprovider is required for multi-asset analysis
        if not self.dp:
            return dataframe

        # We only need to run the selection logic on one of the pairs.
        # Arbitrarily choose BTC to be the "main" one to avoid duplicate processing.
        if metadata["pair"] != "BTC/USD":
            return dataframe

        # Get the latest data for both assets
        btc_df = self.dp.get_pair_dataframe("BTC/USD", self.timeframe)
        eth_df = self.dp.get_pair_dataframe("ETH/USD", self.timeframe)

        # Ensure we have enough data to analyze
        if (
            btc_df.empty
            or len(btc_df) < self.startup_candle_count
            or eth_df.empty
            or len(eth_df) < self.startup_candle_count
        ):
            return dataframe

        # Run the pair selection logic
        selection_result = self.pair_selector.select_best_pair(btc_df, eth_df)
        self.latest_selection_result = selection_result  # Cache for other methods

        # If a pair was selected, find its corresponding dataframe and set the signal
        if selection_result and selection_result["selected_pair"]:
            selected_pair_name = selection_result["selected_pair"]["pair"]

            if selected_pair_name == "BTC/USD":
                # Set enter_long on the last candle of the BTC dataframe
                btc_df.loc[btc_df.index[-1], "enter_long"] = 1

            elif selected_pair_name == "ETH/USD":
                # Set enter_long on the last candle of the ETH dataframe
                # Note: We are modifying a different dataframe here. Freqtrade will process it.
                eth_df.loc[eth_df.index[-1], "enter_long"] = 1

        # Return the original dataframe. If BTC was selected, it will have the entry signal.
        # If ETH was selected, its dataframe was modified in place.
        return btc_df if metadata["pair"] == "BTC/USD" else eth_df

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Dynamic exit logic based on selected strategy type
        """
        pair = metadata["pair"]

        # Initialize exit signal
        dataframe["exit_long"] = 0

        if "BTC" in pair:
            dataframe = self._populate_btc_exit(dataframe)
        elif "ETH" in pair:
            dataframe = self._populate_eth_exit(dataframe)

        return dataframe

    def _populate_btc_exit(self, dataframe: DataFrame) -> DataFrame:
        """BTC swing trading exit conditions"""
        conditions = []

        # RSI overbought
        conditions.append(dataframe["rsi"] > 65)

        # MACD bearish
        conditions.append(
            (dataframe["macd"] < dataframe["macdsignal"]) | (dataframe["macdhist"] < 0)
        )

        # Volume declining
        conditions.append(dataframe["volume_ratio"] < 0.8)

        # Bollinger Band upper region
        conditions.append(dataframe["bb_percent"] > 0.8)

        if conditions:
            dataframe.loc[reduce(lambda x, y: x | y, conditions), "exit_long"] = 1

        return dataframe

    def _populate_eth_exit(self, dataframe: DataFrame) -> DataFrame:
        """ETH momentum trading exit conditions"""
        conditions = []

        # RSI extremely overbought
        conditions.append(dataframe["rsi"] > 75)

        # MACD momentum declining
        conditions.append(
            (dataframe["macd"] < dataframe["macdsignal"])
            | (dataframe["macdhist"] < dataframe["macdhist"].shift(1))
        )

        # Volume declining
        conditions.append(dataframe["volume_ratio"] < 0.7)

        # Overextended above BB
        conditions.append(dataframe["bb_percent"] > 1.1)

        # Momentum weakening
        conditions.append(~dataframe["momentum_positive"])

        if conditions:
            dataframe.loc[reduce(lambda x, y: x | y, conditions), "exit_long"] = 1

        return dataframe

    def custom_stoploss(
        self,
        pair: str,
        trade: "Trade",
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> float:
        """
        Dynamic stoploss based on pair volatility
        """
        if "BTC" in pair:
            return -0.025  # 2.5% for BTC (lower volatility)
        elif "ETH" in pair:
            return -0.030  # 3.0% for ETH (higher volatility)
        else:
            return self.stoploss

    def custom_sell(
        self,
        pair: str,
        trade: "Trade",
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> Optional[Union[str, bool]]:
        """
        Dynamic profit-taking based on asset and strategy type
        """
        if "BTC" in pair:
            # BTC swing trading targets
            if current_profit >= 0.08:
                return "btc_profit_8pct"
            elif current_profit >= 0.06:
                return "btc_profit_6pct"
            elif current_profit >= 0.04:
                return None  # Let exit signals handle

        elif "ETH" in pair:
            # ETH momentum trading targets
            if current_profit >= 0.09:
                return "eth_profit_9pct"
            elif current_profit >= 0.07:
                return "eth_profit_7pct"
            elif current_profit >= 0.05:
                return None  # Let exit signals handle

        return None

    def confirm_trade_entry(
        self,
        pair: str,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        current_time: datetime,
        entry_tag: Optional[str],
        side: str,
        **kwargs,
    ) -> bool:
        """
        Confirm the trade and record it for allocation tracking.
        """
        # Record the trade for allocation tracking
        if (
            self.latest_selection_result
            and self.latest_selection_result["selected_pair"]
        ):
            selected_pair_info = self.latest_selection_result["selected_pair"]
            if selected_pair_info["pair"] == pair:
                self.pair_selector.record_trade(
                    pair=pair, score=selected_pair_info["total_score"]
                )
                print(f"Trade confirmed for {pair}. Recorded for allocation tracking.")
                return True

        print(f"Trade confirmation failed for {pair}. No valid selection found.")
        return False

    def leverage(
        self,
        pair: str,
        current_time: datetime,
        current_rate: float,
        proposed_leverage: float,
        max_leverage: float,
        entry_tag: Optional[str],
        side: str,
        **kwargs,
    ) -> float:
        """
        No leverage for Kraken spot trading
        """
        return 1.0