Mean-Reversion VWAP + Net Volume Z-Score strategy. Full plan: user_data/strategies/plan/cvf-strategy.md
Timeframe
1h
Direction
Long & Short
Stoploss
-3.0%
Trailing Stop
Yes
ROI
N/A
Interface Version
N/A
Startup Candles
300
Indicators
3
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