# source: https://raw.githubusercontent.com/hankniel/freqtrade/83ba4265808f598d5e1f810daa10ebea1726a645/user_data/strategies/freqai/prod/ConsumerSwingShortStrategy.py
"""
Consumer Short Strategy, receive indicators from SwingProducerStrategy.
Generate short signal only.
Used in conjunction with SwingLongStrategy.
"""

import logging
from datetime import datetime

from pandas import DataFrame

from freqtrade.strategy import (IStrategy, merge_informative_pair)

from freqtrade.loggers import logger
logger.name = __name__

class Github_hankniel_freqtrade__ConsumerSwingShortStrategy__20251014_102948(IStrategy):
    """
    Example strategy showing how the user connects their own
    IFreqaiModel to the strategy.
    
    Warning! This is a showcase of functionality,
    which means that it is designed to show various functions of FreqAI
    and it runs on all computers. We use this showcase to help users
    understand how to build a strategy, and we use it as a benchmark
    to help debug possible problems.
    
    This means this is *not* meant to be run live in production.
    """
    
    timeframe = '15m'
    
    minimal_roi = {"0": 0.2}
    
    plot_config = {
        "main_plot": {},
        "subplots": {
            "&-s_close": {"&-s_close": {"color": "blue"}},
            "do_predict": {
                "do_predict": {"color": "brown"},
            },
        },
    }
    
    process_only_new_candles = False #! required for consumers
    stoploss = -0.4 # Using 20x leverage - 2% price drop
    use_exit_signal = True
    # this is the maximum period fed to talib (timeframe independent)
    startup_candle_count: int = 400
    can_short = True
    
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Use the websocket api to get pre-populated indicators from another freqtrade instance.
        Use `self.dp.get_producer_df(pair)` to get the dataframe
        """
        pair = metadata['pair']
        timeframe = self.timeframe
        
        producer_pairs = self.dp.get_producer_pairs()
        # You can specify which producer to get pairs from via:
        # self.dp.get_producer_pairs("my_other_producer")
        
        # This func returns the analyzed dataframe, and when it was analyzed
        producer_df, _ = self.dp.get_producer_df(pair)
        # You can get other data if the producer makes it available:
        # self.dp.get_producer_df(
        #   pair,
        #   timeframe="1h",
        #   candle_type=CandleType.SPOT,
        #   producer_name="my_other_producer"
        # )
        
        if not producer_df.empty:
            # If you plan on passing the producer's entry/exit signal directly,
            # specify ffill=False or it will have unintended results
            #* Here we use custom entry/exit signals of consumer
            merged_df = merge_informative_pair(dataframe, producer_df,
                                               timeframe, timeframe,
                                               append_timeframe=False,
                                               suffix='default')
            return merged_df
        else:
            raise ValueError(f"No producer dataframe found for pair {pair}")
    
    def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame:        
        if 'short_position' in df.columns:
            df.loc[:, ['enter_short', 'enter_tag']] = (df['short_position'], 'short')
        
        return df
    
    def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
        if 'exit_short_position' in df.columns:
            df.loc[:, ['exit_short', 'exit_tag']] = (df['exit_short_position'], 'exit_short')
        
        return df
    
    def confirm_trade_entry(
        self,
        pair: str,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        current_time,
        entry_tag,
        side: str,
        **kwargs,
    ) -> bool:
        df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        last_candle = df.iloc[-1].squeeze()
        
        if side == "long":
            if rate > (last_candle["close"] * (1 + 0.0025)):
                return False
        else:
            if rate < (last_candle["close"] * (1 - 0.0025)):
                return False
        
        return True
    
    def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float,
                            proposed_stake: float, min_stake: float | None, max_stake: float,
                            leverage: float, entry_tag: str | None, side: str,
                            **kwargs) -> float:
        # Using static stake amount for now, 10% available balance each entry
        return self.wallets.get_total_stake_amount() / self.config["max_open_trades"]
    
    def leverage(self, pair: str, current_time: datetime, current_rate: float,
                 proposed_leverage: float, max_leverage: float, entry_tag: str | None, side: str,
                 **kwargs) -> float:
        return 20.0
    
    # TODO: Implement custom_exit(), custom_stake_amount(), leverage(), custom_stoploss(), adjust_trade_position()