# source: https://raw.githubusercontent.com/shaojintian/crypto_backtesting/75ba72665f1a6afaa985e9e0e853330d6d2bd848/freq_trade/user_data/strategies/illiquidity_strategy.py # pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file # --- Do not remove these imports --- import numpy as np import pandas as pd from datetime import datetime, timedelta, timezone from pandas import DataFrame from typing import Optional, Union 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, ) class Github_shaojintian_crypto_backtesting__illiquidity_strategy__20250211_105003(IStrategy): INTERFACE_VERSION = 3 timeframe = "3m" # 0.5% stoploss = -0.010 max_volume_ratio = 0.25 # 25% of volume can_short = True use_custom_stoploss: bool = False trailing_stop: bool = True trailing_stop_positive: float | None = 0.03 trailing_stop_positive_offset: float = 0.0 trailing_only_offset_is_reached = False def custom_stoploss( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs, ) -> float | None: """ Custom stoploss logic, returning the new distance relative to current_rate (as ratio). e.g. returning -0.05 would create a stoploss 5% below current_rate. The custom stoploss can never be below self.stoploss, which serves as a hard maximum loss. For full documentation please go to https://www.freqtrade.io/en/latest/strategy-advanced/ When not implemented by a strategy, returns the initial stoploss value. Only called when use_custom_stoploss is set to True. :param pair: Pair that's currently analyzed :param trade: trade object. :param current_time: datetime object, containing the current datetime :param current_rate: Rate, calculated based on pricing settings in exit_pricing. :param current_profit: Current profit (as ratio), calculated based on current_rate. :param after_fill: True if the stoploss is called after the order was filled. :param **kwargs: Ensure to keep this here so updates to this won't break your strategy. :return float: New stoploss value, relative to the current_rate """ return self.stoploss def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: Optional[float], max_stake: float, entry_tag: Optional[str], side: str, **kwargs) -> float: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) current_candle = dataframe.iloc[-1] # Calculate volume in USDT volume_usdt = current_candle['volume'] * current_candle['close'] # Max stake is 25% of volume max_stake_volume = volume_usdt * self.max_volume_ratio # Return minimum between proposed stake and volume-based stake return min(proposed_stake, max_stake_volume) def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: str | None, side: str, **kwargs) -> float: """ Customize leverage for each new trade. This method is only called in futures mode. :param pair: Pair that's currently analyzed :param current_time: datetime object, containing the current datetime :param current_rate: Rate, calculated based on pricing settings in exit_pricing. :param proposed_leverage: A leverage proposed by the bot. :param max_leverage: Max leverage allowed on this pair :param entry_tag: Optional entry_tag (buy_tag) if provided with the buy signal. :param side: "long" or "short" - indicating the direction of the proposed trade :return: A leverage amount, which is between 1.0 and max_leverage. """ return 1.0 # Volume multiplier parameter volume_multiplier = IntParameter( low=20, high=100, default=50, space='buy', optimize=True ) # Window size for volume average volume_window = IntParameter( low=50, high=200, default=100, space='buy', optimize=True ) amihud_threshold = DecimalParameter(low=0.05, high=0.05, default=0.05, space='buy', optimize=True) def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Add volume and illiquidity indicators""" # Calculate rolling volume average dataframe['volume_mean'] = dataframe['volume'].rolling( window=self.volume_window.value ).mean() # Calculate volume ratio dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_mean'] dataframe['volume_shares'] = dataframe['volume'] * dataframe['close'] # Price change percentage dataframe['price_pct'] = dataframe['close'].pct_change() * 100 # Calculate Amihud illiquidity ratio (absolute return / volume) dataframe['amihud_ratio'] = (dataframe['price_pct'].abs() / dataframe['volume']) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Entry signals based on Amihud illiquidity ratio and volume pump""" dataframe.loc[ ( # Volume is X times higher than average (dataframe['volume_ratio'] > self.volume_multiplier.value) & # Amihud illiquidity ratio is higher, indicating higher volatility (dataframe['amihud_ratio'] > self.amihud_threshold.value) & # Ensure there is enough trading volume (dataframe['volume_shares'] > 10_0000) & # Price has increased (dataframe['price_pct'] > 0) ), 'enter_long' ] = 1 dataframe.loc[ ( # Volume is X times higher than average (dataframe['volume_ratio'] > self.volume_multiplier.value) & # Amihud illiquidity ratio is higher, indicating higher volatility (dataframe['amihud_ratio'] > self.amihud_threshold.value) & # Ensure there is enough trading volume (dataframe['volume_shares'] > 10_0000) & # Price has decreased (dataframe['price_pct'] < 0) ), 'enter_short' ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Exit signals when Amihud illiquidity ratio normalizes""" # Exit long positions when Amihud ratio decreases (liquidity improves) dataframe.loc[ ( # Amihud ratio decreases, liquidity improves (dataframe['amihud_ratio'] < self.amihud_threshold.value/2) ), 'exit_long' ] = 1 # Exit short positions when Amihud ratio decreases (liquidity improves) dataframe.loc[ ( # Amihud ratio decreases, liquidity improves (dataframe['amihud_ratio'] < self.amihud_threshold.value/2) ), 'exit_short' ] = 1 return dataframe