# source: https://raw.githubusercontent.com/Cryptoseen/fvg/2cb9dc50ffcfb2344390134501b8694b69e435ca/FVG_Trading_Strategy.py # coding: utf-8 """FVG trading strategy for Freqtrade 2025.4.""" from __future__ import annotations from datetime import datetime import json import logging import os from typing import Optional import numpy as np import pandas as pd from pandas import DataFrame from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter from freqtrade.persistence import Trade logger = logging.getLogger(__name__) class Github_Cryptoseen_fvg__FVG_Trading_Strategy__20250622_130223(IStrategy): """Simple strategy implementing Fair Value Gap logic.""" minimal_roi = { "0": 0.05, "30": 0.02, "60": 0.01, "120": 0, } stoploss = -0.10 timeframe = "5m" can_short = True position_adjustment_enable = True fvg_threshold = DecimalParameter(0.001, 0.01, default=0.003, decimals=4, optimize=True, space="buy", load=True) fvg_min_volume = IntParameter(100, 100000, default=1000, optimize=True, space="buy", load=True) rr_ratio_1 = DecimalParameter(1.0, 3.0, default=1.0, decimals=2, optimize=True, space="buy", load=True) rr_ratio_2 = DecimalParameter(2.0, 5.0, default=2.0, decimals=2, optimize=True, space="buy", load=True) rr_ratio_3 = DecimalParameter(3.0, 7.0, default=3.0, decimals=2, optimize=True, space="buy", load=True) rr_ratio_4 = DecimalParameter(4.0, 9.0, default=4.0, decimals=2, optimize=True, space="buy", load=True) rr_ratio_5 = DecimalParameter(5.0, 10.0, default=5.0, decimals=2, optimize=True, space="buy", load=True) _custom_info_file_path = "user_data/fvgbots_custom_info.json" _custom_info: dict[str, dict] = {} def __init__(self, config: dict) -> None: super().__init__(config) self.load_custom_info() def load_custom_info(self) -> None: if os.path.exists(self._custom_info_file_path): try: with open(self._custom_info_file_path, "r") as f: self._custom_info = json.load(f) logger.info("Custom info loaded successfully") except Exception as exc: # pylint: disable=broad-except logger.warning("Error loading custom info: %s", exc) self._custom_info = {} else: self._custom_info = {} def save_custom_info(self) -> None: try: with open(self._custom_info_file_path, "w") as f: json.dump(self._custom_info, f, indent=4) except Exception as exc: # pylint: disable=broad-except logger.error("Error saving custom info: %s", exc) def informative_pairs(self): return [] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["fvg_long_gap"] = np.where( dataframe["high"].shift(2) < dataframe["low"], dataframe["low"] - dataframe["high"].shift(2), 0, ) dataframe["fvg_short_gap"] = np.where( dataframe["low"].shift(2) > dataframe["high"], dataframe["low"].shift(2) - dataframe["high"], 0, ) dataframe["fvg_supply_zone_middle"] = (dataframe["high"].shift(1) + dataframe["low"].shift(1)) / 2 dataframe["fvg_demand_zone_lower"] = dataframe["low"].shift(1) dataframe["long_stoploss"] = dataframe["fvg_demand_zone_lower"] * (1 - 0.01) dataframe["short_stoploss"] = dataframe["fvg_supply_zone_middle"] * (1 + 0.01) dataframe["long_stoploss"].fillna(method="ffill", inplace=True) dataframe["short_stoploss"].fillna(method="ffill", inplace=True) dataframe["entering_long_trade"] = np.where( (dataframe["fvg_long_gap"] > self.fvg_threshold.value) & (dataframe["volume"] > self.fvg_min_volume.value) & (dataframe["close"] > dataframe["fvg_demand_zone_lower"]), 1, 0, ) dataframe["entering_short_trade"] = np.where( (dataframe["fvg_short_gap"] > self.fvg_threshold.value) & (dataframe["volume"] > self.fvg_min_volume.value) & (dataframe["close"] < dataframe["fvg_supply_zone_middle"]), 1, 0, ) dataframe["long_breakeven"] = dataframe["close"] * 20 - 0.0001 dataframe["short_breakeven"] = dataframe["close"] * 20 + 0.0001 dataframe["long_breakeven"].fillna(method="ffill", inplace=True) dataframe["short_breakeven"].fillna(method="ffill", inplace=True) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Set entry signals and tags for Fair Value Gap trades.""" long_cond = dataframe["entering_long_trade"] == 1 dataframe.loc[long_cond, ["enter_long", "enter_tag"]] = [1, "fvg_long"] if self.can_short: short_cond = dataframe["entering_short_trade"] == 1 dataframe.loc[short_cond, ["enter_short", "enter_tag"]] = [1, "fvg_short"] return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["cd_type"] = np.nan dataframe["cd_body_spread"] = np.nan dataframe["cd_body_spread_stddev_1"] = np.nan dataframe["cd_vol_spread"] = np.nan dataframe["cd_vol_spread_stddev_1"] = np.nan dataframe.loc[ (dataframe["cd_type"] == "bear") & (dataframe["cd_body_spread"] > dataframe["cd_body_spread_stddev_1"]) & (dataframe["cd_vol_spread"] > dataframe["cd_vol_spread_stddev_1"]), "exit_long", ] = 1 dataframe.loc[ (dataframe["cd_type"] == "bull") & (dataframe["cd_body_spread"] > dataframe["cd_body_spread_stddev_1"]) & (dataframe["cd_vol_spread"] > dataframe["cd_vol_spread_stddev_1"]), "exit_short", ] = 1 return dataframe def confirm_trade_entry( self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, **kwargs, ) -> bool: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1] if pair not in self._custom_info: self._custom_info[pair] = {} if pd.notna(last_candle.get("long_stoploss")) and pd.notna(last_candle.get("short_stoploss")): if order_type == "long": risk_amount = rate - last_candle["long_stoploss"] if risk_amount <= 0: logger.warning("[%s] Long entry risk non-positive", pair) return False self._custom_info[pair]["1_rr_long_profit_target"] = rate + risk_amount * self.rr_ratio_1.value self._custom_info[pair]["2_rr_long_profit_target"] = rate + risk_amount * self.rr_ratio_2.value self._custom_info[pair]["3_rr_long_profit_target"] = rate + risk_amount * self.rr_ratio_3.value self._custom_info[pair]["4_rr_long_profit_target"] = rate + risk_amount * self.rr_ratio_4.value self._custom_info[pair]["5_rr_long_profit_target"] = rate + risk_amount * self.rr_ratio_5.value self._custom_info[pair]["long_breakeven"] = last_candle["long_breakeven"] self._custom_info[pair]["pair_target_reached_status"] = None self._custom_info[pair]["scaling_out_position_sizing_status"] = None elif order_type == "short": risk_amount = last_candle["short_stoploss"] - rate if risk_amount <= 0: logger.warning("[%s] Short entry risk non-positive", pair) return False self._custom_info[pair]["1_rr_short_profit_target"] = rate - risk_amount * self.rr_ratio_1.value self._custom_info[pair]["2_rr_short_profit_target"] = rate - risk_amount * self.rr_ratio_2.value self._custom_info[pair]["3_rr_short_profit_target"] = rate - risk_amount * self.rr_ratio_3.value self._custom_info[pair]["4_rr_short_profit_target"] = rate - risk_amount * self.rr_ratio_4.value self._custom_info[pair]["5_rr_short_profit_target"] = rate - risk_amount * self.rr_ratio_5.value self._custom_info[pair]["short_breakeven"] = last_candle["short_breakeven"] self._custom_info[pair]["pair_target_reached_status"] = None self._custom_info[pair]["scaling_out_position_sizing_status"] = None self.save_custom_info() return True logger.warning("[%s] Stoploss levels not available", pair) return False def custom_exit( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs, ) -> Optional[str]: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1] info = self._custom_info.get(pair) if info is None: return None if current_profit < 0: if trade.trade_direction == "long" and current_rate < last_candle["long_stoploss"]: return "long_stoploss_hit" if trade.trade_direction == "short" and current_rate > last_candle["short_stoploss"]: return "short_stoploss_hit" return None if trade.trade_direction == "long": try: if current_rate >= info["5_rr_long_profit_target"] and info.get("pair_target_reached_status") != "5_rr_long_profit_target_reached": info["pair_target_reached_status"] = "5_rr_long_profit_target_reached" self.save_custom_info() return "5_rr_long" if current_rate >= info["2_rr_long_profit_target"] and info.get("pair_target_reached_status") != "2_rr_long_profit_target_reached": info["pair_target_reached_status"] = "2_rr_long_profit_target_reached" self.save_custom_info() return "2_rr_long" if current_rate >= info["1_rr_long_profit_target"] and info.get("pair_target_reached_status") != "1_rr_long_profit_target_reached": info["pair_target_reached_status"] = "1_rr_long_profit_target_reached" self.save_custom_info() return "1_rr_long" except KeyError: pass if current_profit > 0 and current_profit >= info.get("long_breakeven", float("inf")) and info.get("pair_target_reached_status") != "breakeven_reached": info["pair_target_reached_status"] = "breakeven_reached" self.save_custom_info() return "long_breakeven" elif trade.trade_direction == "short": try: if current_rate <= info["5_rr_short_profit_target"] and info.get("pair_target_reached_status") != "5_rr_short_profit_target_reached": info["pair_target_reached_status"] = "5_rr_short_profit_target_reached" self.save_custom_info() return "5_rr_short" if current_rate <= info["2_rr_short_profit_target"] and info.get("pair_target_reached_status") != "2_rr_short_profit_target_reached": info["pair_target_reached_status"] = "2_rr_short_profit_target_reached" self.save_custom_info() return "2_rr_short" if current_rate <= info["1_rr_short_profit_target"] and info.get("pair_target_reached_status") != "1_rr_short_profit_target_reached": info["pair_target_reached_status"] = "1_rr_short_profit_target_reached" self.save_custom_info() return "1_rr_short" except KeyError: pass if current_profit > 0 and current_profit >= info.get("short_breakeven", float("inf")) and info.get("pair_target_reached_status") != "breakeven_reached": info["pair_target_reached_status"] = "breakeven_reached" self.save_custom_info() return "short_breakeven" if trade.trade_direction == "long" and last_candle.get("exit_long") == 1: return "exit_long_trend_signal" if trade.trade_direction == "short" and last_candle.get("exit_short") == 1: return "exit_short_trend_signal" return None def custom_stoploss( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs, ) -> float: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1] if current_profit < 0: if trade.trade_direction == "long": return last_candle["long_stoploss"] if trade.trade_direction == "short": return last_candle["short_stoploss"] return -1 def adjust_trade_position( self, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, last_candle: dict, info: dict, **kwargs, ) -> Optional[float]: pair = trade.pair if pair not in self._custom_info: return None if trade.trade_direction == "long": try: if ( current_rate >= self._custom_info[pair]["1_rr_long_profit_target"] and self._custom_info[pair]["pair_target_reached_status"] == "1_rr_long_profit_target_reached" and self._custom_info[pair]["scaling_out_position_sizing_status"] != "scaling_out_half_when_1_rr_long_profit_target_reached" ): self._custom_info[pair]["scaling_out_position_sizing_status"] = "scaling_out_half_when_1_rr_long_profit_target_reached" self.save_custom_info() return trade.stake_amount * 0.5 except KeyError: pass try: if ( current_rate >= self._custom_info[pair]["2_rr_long_profit_target"] and self._custom_info[pair]["pair_target_reached_status"] == "2_rr_long_profit_target_reached" and self._custom_info[pair]["scaling_out_position_sizing_status"] != "scaling_out_one_forth_when_2_rr_long_profit_target_reached" ): self._custom_info[pair]["scaling_out_position_sizing_status"] = "scaling_out_one_forth_when_2_rr_long_profit_target_reached" self.save_custom_info() return trade.stake_amount * 0.25 except KeyError: pass try: if ( current_rate >= self._custom_info[pair]["5_rr_long_profit_target"] and self._custom_info[pair]["pair_target_reached_status"] == "5_rr_long_profit_target_reached" and self._custom_info[pair]["scaling_out_position_sizing_status"] != "scaling_out_remainder_at_5rr" ): self._custom_info[pair]["scaling_out_position_sizing_status"] = "scaling_out_remainder_at_5rr" self.save_custom_info() remaining = trade.amount if hasattr(trade, "calc_remaining_amount"): try: remaining = trade.calc_remaining_amount() except Exception: # pylint: disable=broad-except pass return remaining except KeyError: pass elif trade.trade_direction == "short": try: if ( current_rate <= self._custom_info[pair]["1_rr_short_profit_target"] and self._custom_info[pair]["pair_target_reached_status"] == "1_rr_short_profit_target_reached" and self._custom_info[pair]["scaling_out_position_sizing_status"] != "scaling_out_half_when_1_rr_short_profit_target_reached" ): self._custom_info[pair]["scaling_out_position_sizing_status"] = "scaling_out_half_when_1_rr_short_profit_target_reached" self.save_custom_info() return trade.stake_amount * 0.5 except KeyError: pass try: if ( current_rate <= self._custom_info[pair]["2_rr_short_profit_target"] and self._custom_info[pair]["pair_target_reached_status"] == "2_rr_short_profit_target_reached" and self._custom_info[pair]["scaling_out_position_sizing_status"] != "scaling_out_one_forth_when_2_rr_short_profit_target_reached" ): self._custom_info[pair]["scaling_out_position_sizing_status"] = "scaling_out_one_forth_when_2_rr_short_profit_target_reached" self.save_custom_info() return trade.stake_amount * 0.25 except KeyError: pass try: if ( current_rate <= self._custom_info[pair]["5_rr_short_profit_target"] and self._custom_info[pair]["pair_target_reached_status"] == "5_rr_short_profit_target_reached" and self._custom_info[pair]["scaling_out_position_sizing_status"] != "scaling_out_remainder_at_5rr_short" ): self._custom_info[pair]["scaling_out_position_sizing_status"] = "scaling_out_remainder_at_5rr_short" self.save_custom_info() remaining = trade.amount if hasattr(trade, "calc_remaining_amount"): try: remaining = trade.calc_remaining_amount() except Exception: # pylint: disable=broad-except pass return remaining except KeyError: pass return None def bot_start(self, **kwargs) -> None: self.load_custom_info() def bot_shutdown(self, **kwargs) -> None: self.save_custom_info()