# --- Do not remove these libs --- from statistics import median from freqtrade.strategy import IStrategy, stoploss_from_absolute import sys import os # 添加父目录到系统路径 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from chanlun import ChanLun from chanlun.core.ChanEnum import Chan_FX_TYPE, Chan_KLC_FX, Chan_BI_DIR, Chan_KLC_FX, Chan_BSP_TYPE # -------------------------------- from technical.util import resample_to_interval, resampled_merge import talib.abstract as ta from pandas import DataFrame import pandas as pd from datetime import datetime, timedelta from typing import Optional import logging logger = logging.getLogger(__name__) class ChanLun_BTC_5m(IStrategy): """ChanLun_BTC_5m: 5m B3 signals with trailing stop exit.""" INTERFACE_VERSION: int = 3 timeframe = '5m' minimal_roi = {"0": 100} can_short = True enable_long = True enable_short = False lev = 1.0 stoploss = -0.3 use_custom_stoploss = True trailing_stop = False use_exit_signal = True position_adjustment_enable = False startup_candle_count = 500 chan = ChanLun() def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe = self.add_indicators(dataframe) bsp_signal_data = self.chan.get_bsp_signal_data(dataframe) for column, values in bsp_signal_data.items(): dataframe[column] = values return dataframe def add_indicators(self, df): df = self.add_base_indicators(df) base_interval = self.get_ticker_indicator() for interval in (5, 15, 60): if interval <= base_interval: df = self.copy_base_indicators_to_resample(df, interval) continue resampled = resample_to_interval(df, interval) resampled = self.add_base_indicators(resampled) df = resampled_merge(df, resampled) return df def copy_base_indicators_to_resample(self, df, interval): prefix = f'resample_{interval}_' for column in ( 'date', 'open', 'high', 'low', 'close', 'volume', 'atr', 'macd', 'macdsignal', 'macdhist', 'ema24', 'ema52', 'atr_ratio', 'resistance_240', 'support_240', 'trend' ): if column in df.columns: df[f'{prefix}{column}'] = df[column] return df def add_base_indicators(self, df): fast = 12 slow = 26 period = 9 macd = ta.MACD(df, fastperiod=fast, slowperiod=slow, signalperiod=period) df['atr'] = ta.ATR(df, timeperiod=14) df['macd'] = macd['macd'] df['macdsignal'] = macd['macdsignal'] df['macdhist'] = macd['macdhist'] df['ema24'] = ta.EMA(df, timeperiod=24) df['ema52'] = ta.EMA(df, timeperiod=52) df['atr_ratio'] = df['atr'] / df['close'] df['resistance_240'] = df['high'].rolling(240).max().shift(1) df['support_240'] = df['low'].rolling(240).min().shift(1) df['trend'] = 0 df.loc[(df['close'] > df['ema52']) & (df['ema24'] >= df['ema52']), 'trend'] = 1 df.loc[(df['close'] < df['ema52']) & (df['ema24'] <= df['ema52']), 'trend'] = -1 return df def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: min_atr_ratio = 0.0005 long_min_sr_distance_r = 1.0 short_min_sr_distance_r = 0.8 long_space_ratio = (dataframe['resistance_240'].shift(1) - dataframe['close'].shift(1)) / dataframe['close'].shift(1) short_space_ratio = (dataframe['close'].shift(1) - dataframe['support_240'].shift(1)) / dataframe['close'].shift(1) long_tf_aligned = ( (dataframe['resample_5_trend'].shift(1) == 1).astype(int) + (dataframe['resample_15_trend'].shift(1) == 1).astype(int) + (dataframe['resample_60_trend'].shift(1) == 1).astype(int) ) >= 2 short_tf_aligned = ( (dataframe['resample_5_trend'].shift(1) == -1).astype(int) + (dataframe['resample_15_trend'].shift(1) == -1).astype(int) + (dataframe['resample_60_trend'].shift(1) == -1).astype(int) ) >= 2 dataframe.loc[ ( self.enable_long & (dataframe['bsp_state'].shift(1) == -1) & (dataframe['bsp_risk_ratio'].shift(1) > 0) & (long_space_ratio >= dataframe['bsp_risk_ratio'].shift(1) * long_min_sr_distance_r) & (dataframe['macdhist'].shift(1) > 0) & (dataframe['atr_ratio'].shift(1) >= min_atr_ratio) & (dataframe['trend'].shift(1) == 1) & long_tf_aligned ), ['enter_long', 'enter_tag']] = (1, 'long_signal_chan') dataframe.loc[ ( self.enable_short & (dataframe['bsp_state'].shift(1) == 1) & (dataframe['bsp_risk_ratio'].shift(1) > 0) & (short_space_ratio >= dataframe['bsp_risk_ratio'].shift(1) * short_min_sr_distance_r) & (dataframe['macdhist'].shift(1) < 0) & (dataframe['atr_ratio'].shift(1) >= min_atr_ratio) & (dataframe['trend'].shift(1) == -1) & short_tf_aligned ), ['enter_short', 'enter_tag']] = (1, 'short_signal_chan') return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['exit_long'] = 0 dataframe['exit_short'] = 0 return dataframe def custom_exit(self, pair: str, trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs): # Time-based exit only - trailing stop handles profit taking elapsed = current_time - trade.open_date_utc if elapsed >= timedelta(hours=72) and current_profit < 0.005: return 'time_stop_72h' return None def custom_stoploss(self, pair: str, trade, current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs) -> float | None: # Initialize stored state if not trade.get_custom_data('trail_activated'): trade.set_custom_data('trail_activated', False) trade.set_custom_data('max_profit', 0.0) # Read bsp_stop_price from signal try: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) > 0: entry_rows = dataframe[dataframe['date'] <= trade.open_date_utc] signal_rows = entry_rows.tail(3) signal_rows = signal_rows[signal_rows['bsp_risk_ratio'] > 0] if len(signal_rows) > 0: signal_candle = signal_rows.iloc[-1] trade.set_custom_data('bsp_stop_price', float(signal_candle['bsp_stop_price'])) except Exception: pass max_profit = max(float(trade.get_custom_data('max_profit')), current_profit) trade.set_custom_data('max_profit', max_profit) trail_activated = trade.get_custom_data('trail_activated') # Stage 1: Initial stop at bsp_stop with -5% floor if not trail_activated: if max_profit >= 0.02: # Activate trail: move stop to breakeven trade.set_custom_data('trail_activated', True) sl = stoploss_from_absolute(trade.open_rate, current_rate, is_short=trade.is_short) return max(sl, -0.005) else: bsp_stop = trade.get_custom_data('bsp_stop_price') if bsp_stop: sl = stoploss_from_absolute(float(bsp_stop), current_rate, is_short=trade.is_short) return min(sl, -0.05) return -0.05 else: # Stage 2: Trail from max profit if max_profit >= 0.04: trail_offset = 0.02 # Trail 2% behind max trail_price = trade.open_rate * (1 + max_profit - trail_offset) sl = stoploss_from_absolute(trail_price, current_rate, is_short=trade.is_short) return max(sl, -0.02) elif max_profit >= 0.02: # Breakeven to 1% trail sl = stoploss_from_absolute(trade.open_rate * 1.005, current_rate, is_short=trade.is_short) return max(sl, -0.005) else: sl = stoploss_from_absolute(trade.open_rate * 0.998, current_rate, is_short=trade.is_short) return max(sl, -0.02) def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: return self.lev def get_ticker_indicator(self): return int(self.timeframe[:-1])