Ma725BreakStrategy

import numpy as np
import pandas as pd
from pandas import DataFrame
from datetime import datetime, timedelta
from freqtrade.persistence import Trade
from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter
import talib.abstract as ta
import freqtrade.vendor.qtpylib.indicators as qtpylib

class Ma725BreakStrategy(IStrategy):
    """
    Ma725BreakStrategy ported from custom Python implementation.
    
    Original Logic:
    1. Entry: MA7 crosses MA25 after a long trend (20 periods) in the opposite direction.
    2. Filters: RSI, ATR, Time Filter (22:00-01:00 UTC excluded).
    3. Exit: MA7 crosses back.
    4. Stoploss: Mainstream Trailing Stop (User requested).
    """

    INTERFACE_VERSION = 3

    # Minimal ROI designed for the strategy.
    # We rely on trend following (exit on signal) and trailing stop.
    minimal_roi = {
        "0": 100  # Let the strategy decide exit or trailing stop hit
    }

    # Mainstream Stoploss Strategy (User Request)
    stoploss = -0.05
    trailing_stop = True
    trailing_stop_positive = 0.01
    trailing_stop_positive_offset = 0.02
    trailing_only_offset_is_reached = True

    # Timeframe
    timeframe = '15m'

    # Strategy parameters
    # RSI filter: >= 60 for Long, <= 40 for Short (implied symmetry)? 
    # Original code only had 'fail in rsi fall' (ma7>ma25 and rsi<60).
    buy_rsi = IntParameter(60, 90, default=60, space='buy')
    
    can_short = True
    
    # Run "populate_indicators" only for new candle
    process_only_new_candles = True

    # These values can be overridden in the config file
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    # Startup candle count necessary for this strategy
    startup_candle_count: int = 100

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # MA7 and MA25
        dataframe['ma7'] = ta.SMA(dataframe, timeperiod=7)
        dataframe['ma25'] = ta.SMA(dataframe, timeperiod=25)

        # RSI and ATR
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
        dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
        dataframe['atr_mean'] = dataframe['atr'].rolling(window=100).mean()

        # MA25 Slope (Keep Grow / Keep Fall)
        dataframe['ma25_diff'] = dataframe['ma25'].diff()
        dataframe['is_ma25_keep_grow'] = dataframe['ma25_diff'] > 0
        dataframe['is_ma25_keep_fall'] = dataframe['ma25_diff'] < 0
        
        # Trend State
        # Long Trend: MA7 > MA25
        # Short Trend: MA7 < MA25
        dataframe['ma7_gt_ma25'] = (dataframe['ma7'] > dataframe['ma25']).astype(int)
        dataframe['ma7_lt_ma25'] = (dataframe['ma7'] < dataframe['ma25']).astype(int)

        # Calculate duration of the trend
        # We need to know if the trend held for 20 periods
        dataframe['trend_long_streak'] = dataframe['ma7_gt_ma25'].rolling(window=20).sum()
        dataframe['trend_short_streak'] = dataframe['ma7_lt_ma25'].rolling(window=20).sum()

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Based on trade_if_cross_ma logic:
        1. Confirm MA7/MA25 maintained same direction for > 20 periods.
        2. If trend changes (Cross), and Tech Filters passed -> Entry.
        
        Also implements 'Fake Break' logic which bypasses filters.
        """
        
        # Initialize columns
        dataframe['enter_long'] = 0
        dataframe['enter_short'] = 0

        # Calculate Crossovers
        dataframe['cross_above'] = qtpylib.crossed_above(dataframe['ma7'], dataframe['ma25'])
        dataframe['cross_below'] = qtpylib.crossed_below(dataframe['ma7'], dataframe['ma25'])
        
        # Calculate "Recent Crossovers" for Fake Break detection (look back 5 candles)
        # We shift by 1 to check if *previous* 5 candles had a crossover
        dataframe['recent_cross_below'] = dataframe['cross_below'].rolling(window=5).max().shift(1)
        dataframe['recent_cross_above'] = dataframe['cross_above'].rolling(window=5).max().shift(1)

        # 1. Entry Long conditions
        
        # A. Standard Trend Break
        enter_long_standard = (
            dataframe['cross_above'] &
            (dataframe['trend_short_streak'].shift(1) == 20) & 
            (dataframe['is_ma25_keep_fall'] == False) &
            (dataframe['rsi'] >= 60) &
            (dataframe['atr'] >= (dataframe['atr_mean'] * 0.8)) &
            (dataframe['volume'] > 0)
        )
        
        # B. Fake Break (Re-entry)
        # Logic: Current Cross Above + Recent Cross Below (approx < 5 candles ago)
        # Original code does NOT check RSI/ATR for fake break.
        enter_long_fake_break = (
            dataframe['cross_above'] &
            (dataframe['recent_cross_below'] > 0) &
            (dataframe['volume'] > 0)
        )

        dataframe.loc[
            (enter_long_standard | enter_long_fake_break),
            'enter_long'] = 1

        # 2. Entry Short conditions
        
        # A. Standard Trend Break
        enter_short_standard = (
            dataframe['cross_below'] &
            (dataframe['trend_long_streak'].shift(1) == 20) &
            (dataframe['is_ma25_keep_grow'] == False) &
            (dataframe['atr'] >= (dataframe['atr_mean'] * 0.8)) &
            (dataframe['volume'] > 0)
        )
        
        # B. Fake Break (Re-entry)
        # Logic: Current Cross Below + Recent Cross Above
        enter_short_fake_break = (
            dataframe['cross_below'] &
            (dataframe['recent_cross_above'] > 0) &
            (dataframe['volume'] > 0)
        )

        dataframe.loc[
            (enter_short_standard | enter_short_fake_break),
            'enter_short'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Based on trade_if_cross_ma logic:
        The strategy reverses when the signal reverses.
        """
        
        # Initialize
        dataframe['exit_long'] = 0
        dataframe['exit_short'] = 0

        # Exit Long
        dataframe.loc[
            (
                # Signal Reverse: MA7 < MA25
                (dataframe['ma7'] < dataframe['ma25']) &
                (dataframe['volume'] > 0)
            ),
            'exit_long'] = 1

        # Exit Short
        dataframe.loc[
            (
                # Signal Reverse: MA7 > MA25
                (dataframe['ma7'] > dataframe['ma25']) &
                (dataframe['volume'] > 0)
            ),
            'exit_short'] = 1

        return dataframe

    def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
                            time_in_force: str, current_time: datetime, entry_tag: str,
                            side: str, **kwargs) -> bool:
        """
        Checks additional filters: Time Filter and Cool Down.
        """
        
        # 1. Time Filter (22:00 - 01:00 UTC)
        # Original: 22, 23, 0 forbidden
        # Note: current_time is usually UTC in Freqtrade
        if current_time.hour in [22, 23, 0]:
            return False

        # 2. Cool Down Period (Simplified)
        # If the last closed trade was a loss, we might want to skip this trade.
        trades = Trade.get_trades_proxy(pair=pair, is_open=False)
        if trades:
            last_trade = trades[-1]
            # If last trade was a loss
            if last_trade.close_profit < 0:
                # Check how recently it closed. 
                # Original logic: 2 consecutive losses -> 96 periods (24 hours). 1 loss -> 1 period?
                # Let's enforce a 1 hour cool down for any loss for safety in this simulation.
                if (current_time - last_trade.close_date_utc).total_seconds() < 3600:
                    return False

        return True