Mean-Reversion VWAP + Net Volume Z-Score strategy. Full plan: user_data/strategies/plan/cvf-strategy.md
Timeframe
5m
Direction
Long & Short
Stoploss
-3.0%
Trailing Stop
Yes
ROI
N/A
Interface Version
N/A
Startup Candles
300
Indicators
3
freqtrade/freqtrade-strategies
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 SekkaCVF(IStrategy):
"""
Mean-Reversion VWAP + Net Volume Z-Score strategy.
Full plan: user_data/strategies/plan/cvf-strategy.md
"""
timeframe = '5m'
can_short = True
minimal_roi = {}
## -- TRAILING SETUP --
trailing_stop = True
trailing_stop_positive_offset = 0.05
trailing_stop_positive = 0.005
trailing_only_offset_is_reached = True
stoploss = -0.03
startup_candle_count = 300
# --- Fixed Parameters ---
VWAP_PERIOD = 288 # 24h rolling VWAP
ZSCORE_PERIOD = 48 # 4h EMA z-score lookback
ATR_PERIOD = 48 # 4h ATR
ATR_RANK_PERIOD = 288 # 24h ATR percentile window
TIMEOUT_CANDLES = 48 # 4h time-based exit
# --- Hyperopt Parameters ---
wick_rejection = DecimalParameter(0.4, 0.75, default=0.55, space='buy', optimize=True)
buy_zscore = DecimalParameter(1.8, 2.5, default=2.0, space='buy', optimize=True)
sell_zscore = DecimalParameter(-2.5, -1.8, default=-2.0, 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('5min').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
def custom_exit(self, pair: str, trade: Trade, current_time: datetime,
current_rate: float, current_profit: float, **kwargs):
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
#if dataframe.empty:
# return None
#vwap = dataframe.iloc[-1]['vwap']
# Take profit at VWAP touch
#if trade.is_short and current_rate <= vwap:
# return 'tp_vwap_touch'
#if not trade.is_short and current_rate >= vwap:
# return 'tp_vwap_touch'
# Time-based safety exit (4 hours)
#if current_time - trade.open_date_utc >= timedelta(minutes=self.TIMEOUT_CANDLES * 5):
# return 'timeout_4h'
return None