# source: https://raw.githubusercontent.com/aabudeev/freqtrade/9c29560b14047786bc6cae0ac3bd0e5d8ad19b92/user_data/strategies/HybridStrategy.py # pragma pylint: disable=missing-docstring """Hybrid Strategy. Signals confirmed by Indicators.""" from pandas import DataFrame from datetime import datetime import logging import pandas_ta as ta import numpy as np logger = logging.getLogger(__name__) from freqtrade.strategy import IStrategy from freqtrade.persistence import Trade from freqtrade.signals.queue_store import SignalQueueStore from freqtrade.signals.parser import parse_signal_text class Github_aabudeev_freqtrade__HybridStrategy__20260605_063513(IStrategy): """ Strategy that requires BOTH an external signal AND indicator confirmation. """ def __init__(self, config: dict) -> None: super().__init__(config) self.signal_store = SignalQueueStore("/freqtrade/user_data/signals.db") INTERFACE_VERSION = 3 can_short: bool = True minimal_roi = {"0": 10.0} stoploss = -0.99 trailing_stop = True trailing_stop_positive = 0.01 trailing_stop_positive_offset = 0.02 trailing_only_offset_is_reached = True timeframe = "5m" use_custom_stoploss = True process_only_new_candles = False use_exit_signal = False startup_candle_count = 200 order_types = { "entry": "market", "exit": "limit", "stoploss": "market", "stoploss_on_exchange": True, } order_time_in_force = {"entry": "GTC", "exit": "GTC"} def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, side: str, **kwargs) -> float: settings = self.signal_store.get_settings() lev = float(settings.get('hybrid_strategy_leverage', 20.0)) return min(lev, max_leverage) def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: if len(dataframe) < 20: return dataframe dataframe['ema50'] = ta.ema(dataframe['close'], length=50) dataframe['rsi'] = ta.rsi(dataframe['close'], length=14) st = ta.supertrend(dataframe['high'], dataframe['low'], dataframe['close'], length=10, multiplier=3) if st is not None and not st.empty: dataframe['supertrend_direction'] = st.iloc[:, 1] # Order Blocks dataframe['order_block_low'] = np.nan dataframe['order_block_high'] = np.nan for i in range(5, len(dataframe)): if dataframe['low'].iloc[i-3] == dataframe['low'].iloc[i-5:i].min(): dataframe.loc[dataframe.index[i:], 'order_block_low'] = dataframe['low'].iloc[i-3] dataframe.loc[dataframe.index[i:], 'order_block_high'] = dataframe['high'].iloc[i-3] if dataframe['high'].iloc[i-3] == dataframe['high'].iloc[i-5:i].max(): dataframe.loc[dataframe.index[i:], 'order_block_supply_high'] = dataframe['high'].iloc[i-3] dataframe.loc[dataframe.index[i:], 'order_block_supply_low'] = dataframe['low'].iloc[i-3] return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[:, 'enter_long'] = 0 dataframe.loc[:, 'enter_short'] = 0 # 1. Calculate indicator confirmation indicator_long = ( (dataframe['close'] > dataframe['ema50']) & (dataframe['supertrend_direction'] == 1) & (dataframe['rsi'] > 40) & (dataframe['rsi'] < 70) & (dataframe['close'] <= dataframe['order_block_high']) & (dataframe['close'] >= dataframe['order_block_low']) ) indicator_short = ( (dataframe['close'] < dataframe['ema50']) & (dataframe['supertrend_direction'] == -1) & (dataframe['rsi'] < 60) & (dataframe['rsi'] > 30) & (dataframe.get('order_block_supply_high') is not None) & (dataframe['close'] >= dataframe.get('order_block_supply_low', 0)) & (dataframe['close'] <= dataframe.get('order_block_supply_high', 0)) ) # 2. Check for signals waiting = self.signal_store.get_waiting_signals() for sig in waiting: symbol = sig.get('symbol') if symbol and (symbol == metadata['pair'] or symbol.split(':')[0] == metadata['pair'].split(':')[0]): event = parse_signal_text(sig['text']) if event: is_long = (event.side.name == 'LONG') is_short = (event.side.name == 'SHORT') if is_long and indicator_long.iloc[-1]: dataframe.loc[dataframe.index[-1], 'enter_long'] = 1 dataframe.loc[dataframe.index[-1], 'enter_tag'] = f"hybrid_{sig['idempotency_key']}" elif is_short and indicator_short.iloc[-1]: dataframe.loc[dataframe.index[-1], 'enter_short'] = 1 dataframe.loc[dataframe.index[-1], 'enter_tag'] = f"hybrid_{sig['idempotency_key']}" return dataframe def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: str | None, side: str, **kwargs) -> bool: if entry_tag and entry_tag.startswith("hybrid_"): key = entry_tag.replace("hybrid_", "") self.signal_store.mark_status(key, "sent") logger.info(f"Hybrid trade confirmed for {pair}. Signal {key} marked as sent.") return True def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[:, "exit_long"] = 0 dataframe.loc[:, "exit_short"] = 0 return dataframe def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: signal_sl = trade.get_custom_data("signal_sl") if signal_sl is not None: sl_price = float(signal_sl) if not trade.is_short: if sl_price < current_rate: return (sl_price / current_rate) - 1 else: if sl_price > current_rate: return 1 - (sl_price / current_rate) lev = trade.leverage if trade.leverage else 1.0 safety_sl = (0.8 / lev) return -safety_sl def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> str | bool | None: signal_tp = trade.get_custom_data("signal_tp") if signal_tp is not None: tp_price = float(signal_tp) if not trade.is_short: if current_rate >= tp_price: return f"signal_tp_{tp_price}" else: if current_rate <= tp_price: return f"signal_tp_{tp_price}" return None