BTCSwingStrategy

# 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 BTCSwingStrategy(IStrategy):
    """
    BTC-optimized swing trading strategy for BTC/USD on Kraken.
    
    Designed for Bitcoin's 7.55% volatility and steady trend patterns.
    Uses 4h primary timeframe with 1h entry timing for precision.
    
    Target: 4-6% profit moves, 65%+ win rate
    Based on research from Strategy Analysis Report.
    """
    # Strategy interface version
    INTERFACE_VERSION = 3
    # BTC-optimized timeframe (4h primary for trend identification)
    timeframe = '4h'
    # Use 1h for entry timing precision
    informative_timeframe = '1h'
    # Can this strategy go short?
    can_short = False
    # BTC-specific ROI targeting 4-6% moves with scale-out approach
    # 8% aggressive target
    # 6% after 4 hours (optimal target) 
    # 4% after 8 hours (minimum for fees)
    # 1% after 16 hours (safety exit)
    minimal_roi = {'0': 0.08, '60': 0.06, '120': 0.04, '240': 0.01}
    # BTC-optimized stoploss (tighter due to lower volatility)
    stoploss = -0.025  # 2.5% maximum loss (BTC's steadier patterns)
    # Trailing stoploss
    trailing_stop = False
    # BTC-specific parameters
    # Trend confirmation parameters
    ema_short_period = 20
    ema_long_period = 50
    # RSI parameters for BTC's momentum patterns
    rsi_period = 14
    rsi_buy_threshold = 40  # Less oversold for BTC
    rsi_sell_threshold = 65  # Earlier exit for BTC
    # Volume confirmation
    volume_sma_period = 20
    # ADX trend strength
    adx_min_strength = 25

    def informative_pairs(self):
        """
        Define additional informative pairs for 1h entry timing
        """
        pairs = self.dp.current_whitelist()
        informative_pairs = []
        for pair in pairs:
            informative_pairs.append((pair, self.informative_timeframe))
        return informative_pairs

    def populate_indicators(self, dataframe, metadata):
        """
        Populate indicators optimized for BTC swing trading
        """
        # EMA trend indicators (primary for BTC)
        dataframe['ema_short'] = ta.EMA(dataframe, timeperiod=self.ema_short_period)
        dataframe['ema_long'] = ta.EMA(dataframe, timeperiod=self.ema_long_period)
        dataframe['ema_trend'] = dataframe['ema_short'] > dataframe['ema_long']
        # RSI for momentum
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period)
        # Bollinger Bands for volatility
        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'])
        # MACD for trend confirmation
        macd = ta.MACD(dataframe)
        dataframe['macd'] = macd['macd']
        dataframe['macdsignal'] = macd['macdsignal']
        dataframe['macdhist'] = macd['macdhist']
        # Volume indicators
        dataframe['volume_sma'] = ta.SMA(dataframe['volume'], timeperiod=self.volume_sma_period)
        dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma']
        # ADX for trend strength
        dataframe['adx'] = ta.ADX(dataframe)
        # Support/Resistance levels (simple approach)
        dataframe['resistance'] = dataframe['high'].rolling(window=10).max()
        dataframe['support'] = dataframe['low'].rolling(window=10).min()
        # Price position relative to support/resistance
        dataframe['price_position'] = (dataframe['close'] - dataframe['support']) / (dataframe['resistance'] - dataframe['support'])
        return dataframe

    @informative('1h')
    def populate_indicators_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        1h timeframe indicators for precise entry timing
        """
        # Fast RSI for entry timing
        dataframe['rsi_1h'] = ta.RSI(dataframe, timeperiod=14)
        # Fast EMA for immediate trend
        dataframe['ema_fast_1h'] = ta.EMA(dataframe, timeperiod=12)
        dataframe['ema_slow_1h'] = ta.EMA(dataframe, timeperiod=26)
        # Volume surge detection
        dataframe['volume_sma_1h'] = ta.SMA(dataframe['volume'], timeperiod=20)
        dataframe['volume_surge_1h'] = dataframe['volume'] > dataframe['volume_sma_1h'] * 1.5
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        BTC-optimized entry conditions for 4-6% swing moves
        
        Entry Logic:
        1. 4h trend confirmation (EMA alignment)
        2. RSI recovery from oversold (but not too deep)
        3. MACD bullish alignment
        4. Volume confirmation
        5. 1h timing confirmation
        6. Support/resistance positioning
        """
        conditions = []
        # Primary trend confirmation (4h timeframe)
        conditions.append(dataframe['ema_trend'])  # Short EMA > Long EMA
        # RSI in favorable zone (not too oversold for BTC)
        # Not overbought
        conditions.append((dataframe['rsi'] > self.rsi_buy_threshold) & (dataframe['rsi'] < 60))
        # MACD bullish alignment
        conditions.append((dataframe['macd'] > dataframe['macdsignal']) & (dataframe['macdhist'] > 0))
        # Volume confirmation (institutional interest)
        conditions.append(dataframe['volume_ratio'] > 1.2)
        # ADX shows trend strength
        conditions.append(dataframe['adx'] > self.adx_min_strength)
        # Price position favorable (not at resistance)
        conditions.append(dataframe['price_position'] < 0.8)
        # 1h timeframe confirmation for entry timing
        conditions.append(dataframe['rsi_1h'] > 35)  # 1h RSI recovery
        conditions.append(dataframe['ema_fast_1h'] > dataframe['ema_slow_1h'])  # 1h trend aligned
        # Bollinger Band positioning (near lower band for entry)
        conditions.append(dataframe['bb_percent'] < 0.3)
        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:
        """
        BTC-optimized exit conditions for profit-taking
        
        Exit Logic:
        1. RSI overbought (momentum exhausted)
        2. MACD bearish divergence
        3. Volume declining
        4. Price at upper Bollinger Band
        5. 1h timeframe shows weakness
        """
        conditions = []
        # RSI overbought (BTC-specific threshold)
        conditions.append(dataframe['rsi'] > self.rsi_sell_threshold)
        # MACD bearish signal
        conditions.append((dataframe['macd'] < dataframe['macdsignal']) | (dataframe['macdhist'] < 0))
        # Volume declining (profit-taking phase)
        conditions.append(dataframe['volume_ratio'] < 0.8)
        # Price at upper Bollinger Band
        conditions.append(dataframe['bb_percent'] > 0.8)
        # 1h timeframe showing weakness
        conditions.append(dataframe['rsi_1h'] > 70)
        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:
        """
        BTC-optimized stoploss logic
        """
        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]]:
        """
        BTC-optimized profit-taking logic
        
        Scale-out approach: 4%, 6%, 8% targets
        """
        # Scale out at 8% (aggressive target for BTC)
        if current_profit >= 0.08:
            return 'btc_profit_8pct'
        # Scale out at 6% (optimal target)
        if current_profit >= 0.06:
            return 'btc_profit_6pct'
        # Consider exit at 4% minimum if other conditions met
        if current_profit >= 0.04:
            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:
        """
        No leverage for spot trading on Kraken
        """
        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