💬 Forum

LiquidityShockReversalStrategy

♡

uploads/liquidity_shock_reversal_strategy.py · uploaded by 🐋 Ron · first seen 2026-07-18 · ⬇ 2 downloads

Basics mode: futures timeframe: 3m interface version: 3
Settings stoploss: -0.01 hyperopt hyperopt params: 3
Indicators talib technical
Concepts mean_reversion
15 related strategies (⧉ identical code, ≈ similar name)

Each tile is a different kind of check — from an instant code lint to full sandboxed backtests and forward tests on recent data. Not sure what a check actually proves? See the FAQ →

Source

🔒 Log in to copy 🔒 Log in to download 🔒 Raw 
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
# isort: skip_file
# --- Do not remove these imports ---
import logging
import numpy as np
import pandas as pd
from datetime import datetime, timedelta, timezone
from pandas import DataFrame
from typing import Optional, Union
logger = logging.getLogger(__name__)
from freqtrade.strategy import (
    IStrategy,
    Trade,
    Order,
    PairLocks,
    informative,  # @informative decorator
    # Hyperopt Parameters
    BooleanParameter,
    CategoricalParameter,
    DecimalParameter,
    IntParameter,
    RealParameter,
    # timeframe helpers
    timeframe_to_minutes,
    timeframe_to_next_date,
    timeframe_to_prev_date,
    # Strategy helper functions
    merge_informative_pair,
    stoploss_from_absolute,
    stoploss_from_open,
)

# --------------------------------
# Add your lib to import here
import talib.abstract as ta
from technical import qtpylib

class LiquidityShockReversalStrategy(IStrategy):
    INTERFACE_VERSION = 3
    timeframe = "3m"
    
    stoploss = -0.01
    # Liquidity Parameters
    vol_shock_mult = DecimalParameter(2.0, 5.0, default=3.0, space='buy')
    price_impact = DecimalParameter(0.005, 0.05, default=1e-7, space='buy')
    window_size = IntParameter(3, 10, default=5, space='buy')

    can_short = True

    
    
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # Volume Shock Detection
        dataframe['volume_ma'] = dataframe['volume'].rolling(window=self.window_size.value).mean()
        dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_ma']
        dataframe['volume_shares'] = dataframe['volume']*dataframe['close']
        
        # Price Impact
        dataframe['returns'] = dataframe['close'].pct_change()
        dataframe['price_impact'] = dataframe['returns'].abs() / (dataframe['volume'] * dataframe['close'])
        #logger.info(dataframe['price_impact'])

        # Liquidity Shock Score
        dataframe['shock_score'] = (
            (dataframe['volume_ratio'] > self.vol_shock_mult.value) & 
            (dataframe['price_impact'] > self.price_impact.value)
        ).astype(int)
        
        # Mean Reversion Signal
        dataframe['future_return'] = dataframe['close'].shift(-4) / dataframe['close'] - 1
        dataframe['reversal_signal'] = (
            (dataframe['shock_score'] == 1) & 
            (dataframe['returns'] < -self.price_impact.value)
        )
        
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe.loc[
            (
                dataframe['reversal_signal'] &
                (dataframe['volume'] > 0) &
                ( dataframe['volume_shares'] > 100_0000)
            ),
            'enter_short'
        ] = 1
        
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe.loc[
            (
                (dataframe['future_return'] > self.price_impact.value) |
                (dataframe['volume_ratio'] < 1.0)
            ),
            'exit_short'
        ] = 1
        
        return dataframe