SekkaCVF60

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 SekkaCVF60(IStrategy):
    """
    Mean-Reversion VWAP + Net Volume Z-Score strategy.
    Full plan: user_data/strategies/plan/cvf-strategy.md
    """

    timeframe = '1h'
    can_short = True

    minimal_roi = {}
     ## -- TRAILING SETUP --
    trailing_stop = True
    trailing_stop_positive_offset = 0.03
    trailing_stop_positive = 0.005
    trailing_only_offset_is_reached = True

    stoploss = -0.03

    startup_candle_count = 300

    # --- Fixed Parameters (Adjusted for 1h) ---
    VWAP_PERIOD = 14       # 14h rolling VWAP
    ZSCORE_PERIOD = 24     # 24h EMA z-score lookback (needed for statistical validity)
    ATR_PERIOD = 14        # Standard 14h ATR
    ATR_RANK_PERIOD = 24   # 24h ATR percentile window
    

    # --- Hyperopt Parameters ---
    wick_rejection = DecimalParameter(0.5, 0.8, default=0.6, space='buy', optimize=True)
    buy_zscore = DecimalParameter(1.0, 1.8, default=1.4, space='buy', optimize=True)
    sell_zscore = DecimalParameter(-1.8, -1.0, default=-1.4, space='sell', optimize=True)
    vwap_long_dist = DecimalParameter(0.01, 0.08, default=0.02, space='buy', optimize=True)
    vwap_short_dist = DecimalParameter(0.01, 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('60min').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