KrakenSwingStrategy

# 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
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

class KrakenSwingStrategy(IStrategy):
    """
    Kraken-optimized swing trading strategy for BTC/USD and ETH/USD pairs.
    
    Designed for single $1000 positions targeting 4-6% minimum moves
    to overcome Kraken's 0.8% round-trip fee structure.
    
    Based on research from Kraken Trading Pairs Analysis and 
    Implementation Roadmap documents.
    """
    # Strategy interface version - allow new iterations of the strategy interface.
    INTERFACE_VERSION = 3
    # Optimal timeframe for swing trading on Kraken
    timeframe = '1h'
    # Can this strategy go short?
    can_short: bool = False
    # Minimal ROI designed for strategy - targeting 4-6% minimum moves
    # 6% immediate target
    # 5% after 40 minutes  
    # 4% after 80 minutes (minimum for fee efficiency)
    # 1% after 2 hours (safety exit)
    minimal_roi = {'0': 0.06, '40': 0.05, '80': 0.04, '120': 0.01}
    # Optimal stoploss for single position strategy
    stoploss = -0.03  # 3% maximum loss per trade ($30 on $1000 position)
    # Trailing stoploss
    trailing_stop = False
    # Hyperoptable parameters
    buy_rsi_period = 14
    buy_rsi_value = 35
    sell_rsi_value = 70
    # Buy/Sell signal optimization spaces
    buy_bb_lower_offset = 0.02  # 2% below lower BB
    sell_bb_upper_offset = 0.02  # 2% above upper BB

    def informative_pairs(self):
        """
        Define additional, informative pair/interval combinations to be cached from the exchange.
        """
        return []

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Adds several different TA indicators to the given DataFrame
        
        Performance Note: For preprocessing, consider using new columns via:
        dataframe['ema12'] = ta.EMA(dataframe, timeperiod=12)
        """
        # RSI - Primary momentum indicator
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.buy_rsi_period)
        # Bollinger Bands - Volatility and mean reversion
        bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=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'])
        dataframe['bb_width'] = (dataframe['bb_upperband'] - dataframe['bb_lowerband']) / dataframe['bb_middleband']
        # MACD - Trend confirmation
        macd = ta.MACD(dataframe)
        dataframe['macd'] = macd['macd']
        dataframe['macdsignal'] = macd['macdsignal']
        dataframe['macdhist'] = macd['macdhist']
        # EMA - Trend direction
        dataframe['ema12'] = ta.EMA(dataframe, timeperiod=12)
        dataframe['ema26'] = ta.EMA(dataframe, timeperiod=26)
        # Volume indicators
        dataframe['volume_sma'] = ta.SMA(dataframe['volume'], timeperiod=20)
        # ADX - Trend strength (for filtering weak signals)
        dataframe['adx'] = ta.ADX(dataframe)
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Based on TA indicators, populates the entry signal for the given dataframe
        
        Entry Criteria for 4-6% swing moves:
        1. RSI oversold but recovering (momentum building)
        2. Price near lower Bollinger Band (mean reversion setup)
        3. MACD showing bullish divergence
        4. Strong trend confirmation from EMA
        5. Above average volume (institutional interest)
        """
        conditions = []
        # RSI recovery from oversold
        conditions.append((dataframe['rsi'] > self.buy_rsi_value) & (dataframe['rsi'].shift(1) <= self.buy_rsi_value))
        # Price bouncing off lower Bollinger Band
        conditions.append((dataframe['close'] <= dataframe['bb_lowerband'] * (1 + self.buy_bb_lower_offset)) & (dataframe['close'] > dataframe['bb_lowerband'] * (1 - self.buy_bb_lower_offset)))
        # MACD bullish signal
        conditions.append((dataframe['macd'] > dataframe['macdsignal']) & (dataframe['macdhist'] > 0))
        # EMA trend confirmation (short above long)
        conditions.append(dataframe['ema12'] > dataframe['ema26'])
        # Volume confirmation (above average)
        conditions.append(dataframe['volume'] > dataframe['volume_sma'])
        # ADX shows strong trend (above 25)
        conditions.append(dataframe['adx'] > 25)
        # Bollinger Band width shows sufficient volatility for 4-6% moves
        conditions.append(dataframe['bb_width'] > 0.04)  # 4% minimum width
        if conditions:
            dataframe.loc[reduce(lambda x, y: x & y, conditions), 'enter_long'] = 1
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Based on TA indicators, populates the exit signal for the given dataframe
        
        Exit Criteria for profit-taking:
        1. RSI overbought (momentum exhausted)
        2. Price approaching upper Bollinger Band
        3. MACD showing bearish divergence
        4. Volume declining (profit-taking phase)
        """
        conditions = []
        # RSI overbought
        conditions.append(dataframe['rsi'] > self.sell_rsi_value)
        # Price at upper Bollinger Band
        conditions.append(dataframe['close'] >= dataframe['bb_upperband'] * (1 - self.sell_bb_upper_offset))
        # MACD bearish signal
        conditions.append((dataframe['macd'] < dataframe['macdsignal']) | (dataframe['macdhist'] < 0))
        # Volume declining (optional exit signal)
        volume_declining = dataframe['volume'] < dataframe['volume'].shift(1)
        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:
        """
        Custom stoploss logic for Kraken optimization
        
        - Maintains 3% maximum loss
        - No trailing stop to avoid Kraken API rate limit issues
        """
        return self.stoploss

    def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> Optional[Union[str, bool]]:
        """
        Custom sell logic for profit optimization on Kraken
        
        Ensures we hit minimum 4% profit targets to overcome fees
        """
        # Force exit if we hit 6% profit (optimal target)
        if current_profit >= 0.06:
            return 'profit_target_6pct'
        # Consider exit if we hit 4% minimum and other conditions met
        if current_profit >= 0.04:
            # Check if momentum is declining (basic check)
            return None  # Let normal exit signals handle this
        return None

    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:
        """
        Customize leverage for each new trade. Only called when margin trading is enabled.
        
        For Kraken spot trading, this returns 1.0 (no leverage)
        """
        return 1.0

def reduce(function, iterable, initializer=None):
    """Python reduce function for combining conditions"""
    it = iter(iterable)
    if initializer is None:
        value = next(it)
    else:
        value = initializer
    for element in it:
        value = function(value, element)
    return value