SekkaCVF15

import logging
import pandas as pd
import numpy as np
import talib.abstract as ta
from freqtrade.strategy import IStrategy, DecimalParameter
from freqtrade.persistence import Trade
from pandas import DataFrame
from pathlib import Path
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)


class SekkaCVF15(IStrategy):
    """
    Mean-Reversion VWAP + Net Volume Z-Score strategy.
    Full plan: user_data/strategies/plan/cvf-strategy.md
    """

    timeframe = '15m'
    can_short = True

    minimal_roi = {}
     ## -- TRAILING SETUP --
    trailing_stop = True
    trailing_stop_positive_offset = 0.02
    trailing_stop_positive = 0.01
    trailing_only_offset_is_reached = True

    stoploss = -0.03

    startup_candle_count = 300

    # --- Fixed Parameters ---
    # --- Fixed Parameters (Adjusted for 15m) ---
    VWAP_PERIOD = 192      # 48h rolling VWAP (48 * 4 candles)
    ZSCORE_PERIOD = 16     # 4h EMA z-score lookback (4 * 4 candles)
    ATR_PERIOD = 16        # 4h ATR
    ATR_RANK_PERIOD = 96   # 24h ATR percentile window
    TIMEOUT_CANDLES = 16   # 4h time-based exit

    # --- Hyperopt Parameters ---
    wick_rejection = DecimalParameter(0.5, 0.8, default=0.6, space='buy', optimize=True)
    buy_zscore = DecimalParameter(1.3, 3.2, default=1.6, space='buy', optimize=True)
    sell_zscore = DecimalParameter(-3.2, -1.3, default=-1.6, space='sell', optimize=True)
    vwap_long_dist = DecimalParameter(0.03, 0.08, default=0.02, space='buy', optimize=True)
    vwap_short_dist = DecimalParameter(0.03, 0.08, default=0.02, space='sell', optimize=True)

    def bot_start(self, **kwargs):
        self.net_vol_cache = {}

    def compute_net_volume(self, pair: str) -> pd.DataFrame:
        pair_filename = pair.replace('/', '_').replace(':', '_')
        data_dir = Path(self.config['datadir'])

        if self.config.get('trading_mode', 'spot') == 'futures' and data_dir.name != 'futures':
            data_dir = data_dir / 'futures'

        trades_path = data_dir / f"{pair_filename}-trades.feather"
        if not trades_path.exists():
            raise FileNotFoundError(f"Trades file not found: {trades_path}")

        trades_df = pd.read_feather(trades_path)
        if not pd.api.types.is_datetime64_any_dtype(trades_df['timestamp']):
            trades_df['timestamp'] = pd.to_datetime(trades_df['timestamp'], unit='ms', utc=True)
        trades_df.set_index('timestamp', inplace=True)

        trades_df['buy_vol'] = trades_df['amount'].where(trades_df['side'] == 'buy', 0)
        trades_df['sell_vol'] = trades_df['amount'].where(trades_df['side'] != 'buy', 0)

        resampled = trades_df.resample('15min').agg(
            buy_vol=('buy_vol', 'sum'),
            sell_vol=('sell_vol', 'sum')
        ).fillna(0)

        resampled['v_net'] = resampled['buy_vol'] - resampled['sell_vol']
        return resampled[['v_net']]

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        pair = metadata['pair']

        if pair not in self.net_vol_cache:
            self.net_vol_cache[pair] = self.compute_net_volume(pair)

        dataframe.index = pd.to_datetime(dataframe['date'], utc=True)
        dataframe = dataframe.join(self.net_vol_cache[pair], how='left')
        dataframe['v_net'] = dataframe['v_net'].fillna(0)
        dataframe.reset_index(drop=True, inplace=True)

        # Rolling VWAP (24h)
        typical_price = (dataframe['high'] + dataframe['low'] + dataframe['close']) / 3
        pv = typical_price * dataframe['volume']
        dataframe['vwap'] = (
            pv.rolling(self.VWAP_PERIOD).sum() /
            dataframe['volume'].rolling(self.VWAP_PERIOD).sum().clip(lower=1e-8)
        )

        # EMA Z-Score of Net Volume
        mu_ema = dataframe['v_net'].ewm(span=self.ZSCORE_PERIOD, adjust=False).mean()
        sigma = dataframe['v_net'].rolling(self.ZSCORE_PERIOD).std()
        dataframe['zscore'] = (dataframe['v_net'] - mu_ema) / (sigma + 1e-10)

        # ATR percentile filter
        dataframe['atr'] = ta.ATR(dataframe, timeperiod=self.ATR_PERIOD)
        dataframe['atr_pct'] = dataframe['atr'].rolling(self.ATR_RANK_PERIOD).rank(pct=True)

        # Wick rejection helper
        dataframe['candle_range'] = dataframe['high'] - dataframe['low']

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        wr = self.wick_rejection.value

        long_signal = (
            (dataframe['close'] < (dataframe['vwap'] * (1 - self.vwap_long_dist.value))) &
            (dataframe['zscore'] > self.buy_zscore.value) &
            (dataframe['close'] > dataframe['low'] + (dataframe['candle_range'] * wr)) &
            (dataframe['atr_pct'] > 0.25)
        )

        short_signal = (
            (dataframe['close'] > (dataframe['vwap'] * (1 + self.vwap_short_dist.value))) &
            (dataframe['zscore'] < self.sell_zscore.value) &
            (dataframe['close'] < dataframe['high'] - (dataframe['candle_range'] * wr)) &
            (dataframe['atr_pct'] > 0.25)
        )

        dataframe.loc[long_signal, ['enter_long', 'enter_tag']] = (1, 'cvf_long_zscore_vwap')
        dataframe.loc[short_signal, ['enter_short', 'enter_tag']] = (1, 'cvf_short_zscore_vwap')

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        return dataframe