CrossAssetStrategy

from freqtrade.strategy.interface import IStrategy
from pandas import DataFrame
from typing import Dict

class CrossAssetStrategy(IStrategy):
    timeframe = '1h'
    startup_candle_count = 30
    stoploss = -0.99
    use_custom_stoploss = True
    process_only_new_candles = True
    minimal_roi = {"0": 0.1}

    pairs = {
        "BTC/USDT:USDT": "ETH/USDT:USDT",
        "ETH/USDT:USDT": "BTC/USDT:USDT"
    }

    def populate_indicators(self, df: DataFrame, metadata: dict) -> DataFrame:
        pair = metadata["pair"]
        if pair not in self.pairs:
            return df

        # Load data for the signal pair
        signal_pair = self._get_signal_pair(pair)
        signal_df, _ = self.dp.get_pair_dataframe(pair=signal_pair, timeframe=self.timeframe)

        signal_df['20_high'] = signal_df['high'].rolling(20).max()
        signal_df['3_low'] = signal_df['low'].rolling(3).min()
        signal_df['breakout'] = (signal_df['high'] > signal_df['20_high'].shift(1)).astype(int)

        # Copy relevant indicators into current pair's df
        df['signal_breakout'] = signal_df['breakout']
        df['signal_3_low'] = signal_df['3_low']

        return df

    def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
        if metadata["pair"] not in self.pairs:
            return df

        df.loc[df['signal_breakout'] == 1, 'enter_long'] = 1
        return df

    def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
        df['exit_long'] = 0  # all exit logic in stoploss
        return df

    def confirm_trade_entry(self, *args, **kwargs) -> bool:
        return len(self.wallets.get_all_open_trades()) == 0

    def custom_stoploss(self, pair, trade, current_time, current_rate, current_profit, **kwargs):
        if pair not in self.pairs:
            return 1

        # Figure out which asset generated the signal for this trade
        signal_pair = self._get_signal_pair(pair)

        try:
            signal_df, _ = self.dp.get_pair_dataframe(pair=signal_pair, timeframe=self.timeframe)
            signal_df['3_low'] = signal_df['low'].rolling(3).min()
            idx = signal_df.index.get_loc(trade.open_date_utc, method='nearest')
            trailing_low = signal_df.iloc[idx:]['3_low'].min()

            if current_rate < trailing_low:
                return 0.01
        except Exception:
            pass

        return 1

    def _get_signal_pair(self, current_pair: str) -> str:
        """Returns the pair used as signal for this traded pair"""
        return self.pairs.get(current_pair)