SekkaTrend

# ================================================================
# SekkaTrend Universal 1m – HFT Mean-Reversion Strategy
# ---------------------------------------------------------------
# Long:  Stoch(14) < 10 & RSI(5) < 35 & ATR(7) expanding & MACD > 0
# Short: Price > VWAP(30) by >1.0% & ROC(15) > 1.0%
# Exit:  Take Profit / Stop Loss (Option B)
# ================================================================

import logging
from datetime import datetime
from typing import Optional

import numpy as np
import pandas as pd
import talib.abstract as ta
from pandas import DataFrame
from freqtrade.strategy import IStrategy, DecimalParameter

logger = logging.getLogger(__name__)

class SekkaTrend(IStrategy):
    """
    SekkaTrend (Universal 1m)
    Validated on multiple assets (TAO, CRV, ASTER). Predicts next-candle
    mean reversion using extreme overbought/oversold conditions.
    """
    
    # Strategy interface version
    INTERFACE_VERSION = 3

    # Optimal timeframe for the strategy
    timeframe = "1m"

    # Can this strategy go short?
    can_short = True

    # Use custom exit signals
    use_exit_signal = True
    exit_profit_only = False
    
    # Needs sufficient warmup for 30-min VWAP and 26-period MACD
    startup_candle_count = 60

    # -- Hyperoptable Entry Parameters --
    # LONG
    p_stoch_lower = DecimalParameter(5.0, 15.0, default=10.0, space="buy", optimize=True)
    p_rsi_lower = DecimalParameter(25.0, 45.0, default=35.0, space="buy", optimize=True)
    
    # SHORT
    p_vwap_dist = DecimalParameter(0.5, 2.0, default=1.0, space="sell", optimize=True)
    p_roc_spike = DecimalParameter(0.5, 2.0, default=1.0, space="sell", optimize=True)

    # -- Hyperoptable Custom Exit Parameters (Option B) --
    # Tight TP and SL for 1m candle scalping (Default TP: +0.20%, SL: -0.30%)
    TP_RATE = DecimalParameter(0.001, 0.005, default=0.003, decimals=4, space="sell", optimize=True)
    SL_RATE = DecimalParameter(-0.005, -0.001, default=-0.001, decimals=4, space="sell", optimize=True)

    # Hard stoploss fallback (custom_exit handles the actual tight sl)
    stoploss = -0.05
    minimal_roi = {"0": 100}  # Rely entirely on custom_exit TP

    trailing_stop = False

    # ------------------ Plot Config ------------------
    plot_config = {
        "main_plot": {
            "vwap30": {"color": "orange"},
        },
        "subplots": {
            "Oscillators": {
                "stoch_k14": {"color": "cyan"},
                "rsi5": {"color": "magenta"},
            },
            "Volatility": {
                "atr7": {"color": "purple"},
            },
            "Momentum": {
                "macd": {"color": "blue"},
            },
            "Short Triggers": {
                "vwap_dist": {"color": "red"},
                "roc15": {"color": "yellow"},
            }
        },
    }

    def populate_indicators(self, df: DataFrame, metadata: dict) -> DataFrame:
        """Calculate all TA indicators required for the Universal 1m Strategy."""
        
        # 1. Stochastic %K(14) - Manual calculation to ensure raw %K without default smoothing
        low_min = df['low'].rolling(14).min()
        high_max = df['high'].rolling(14).max()
        df['stoch_k14'] = ((df['close'] - low_min) / (high_max - low_min)) * 100
        
        # 2. RSI(5)
        df['rsi5'] = ta.RSI(df, timeperiod=5)
        
        # 3. ATR(7)
        df['atr7'] = ta.ATR(df, timeperiod=7)
        
        # 4. MACD(8, 26, 9)
        macd_obj = ta.MACD(df, fastperiod=8, slowperiod=26, signalperiod=9)
        df['macd'] = macd_obj['macd']  # The MACD line
        
        # 5. VWAP(30) - 30m rolling VWAP (Volume-Weighted Average Price)
        pv_sum = (df['close'] * df['volume']).rolling(30).sum()
        v_sum = df['volume'].rolling(30).sum()
        df['vwap30'] = pv_sum / v_sum
        
        # Distance from VWAP as a percentage
        df['vwap_dist'] = ((df['close'] - df['vwap30']) / df['vwap30']) * 100
        
        # 6. ROC(15)
        df['roc15'] = ta.ROC(df, timeperiod=15)
        
        return df

    def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
        """Populate Long and Short entry signals."""
        
        df.loc[:, "enter_long"] = 0
        df.loc[:, "enter_short"] = 0

        # ------------------
        # LONG ENTRY: "Buy the Dip in an Uptrend"
        # ------------------
        long_condition = (
            (df['stoch_k14'] < self.p_stoch_lower.value) &
            (df['rsi5'] < self.p_rsi_lower.value) &
            (df['atr7'] > df['atr7'].shift(1)) &     # ATR Expanding (Current > Previous)
            (df['macd'] > 0)
        )
        
        df.loc[long_condition, "enter_long"] = 1

        # ------------------
        # SHORT ENTRY: "Fade the Overextended Rally"
        # ------------------
        short_condition = (
            (df['vwap_dist'] > self.p_vwap_dist.value) &
            (df['roc15'] > self.p_roc_spike.value)
        )
        
        df.loc[short_condition, "enter_short"] = 1

        return df

    def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
        """Clear exit trends to rely exclusively on custom_exit TP/SL limits."""
        df.loc[:, "exit_long"] = 0
        df.loc[:, "exit_short"] = 0
        return df

    def custom_exit(self, pair: str, trade, current_time: datetime,
                    current_rate: float, current_profit: float, **kwargs) -> Optional[str]:
        """Option B: Strict Target Profit and Stop Loss."""
        
        # Check Take Profit
        if current_profit >= self.TP_RATE.value:
            return "TAKE_PROFIT"
        
        # Check Stop Loss
        if current_profit <= self.SL_RATE.value:
            return "STOP_LOSS"
            
        return None