# --- 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_Classifier import ChanLunClassifier from ChanEnum import Chan_FX_TYPE, Chan_KLC_FX, Chan_BI_DIR, Chan_KLC_FX from ChanPY import ChanPY # -------------------------------- from technical.util import resample_to_interval, resampled_merge import talib.abstract as ta from pandas import DataFrame from datetime import datetime, timedelta from freqtrade.persistence import Trade, Order from typing import Optional import logging import numpy as np import pandas as pd from functools import reduce logger = logging.getLogger(__name__) ### Now you can use logger.info('asfd') to log # freqtrade plot-dataframe --strategy ChanLun_BTC_K --datadir user_data/data/binance -c ./user_data/Chan/config/ChanLun_BTC_K.json --timerange=20250309- # freqtrade trade -c ./user_data/Chan/config/ChanLun_BTC_K.json --strategy ChanLun_BTC_K --strategy-path ./user_data/Chan/strategies # freqtrade backtesting -c ./user_data/Chan/config/ChanLun_BTC_K.json --strategy ChanLun_BTC_K --strategy-path ./user_data/Chan/strategies --timerange=20250701- # freqtrade download-data -c ./user_data/Chan/config/ChanLun_BTC_K.json -t 1m --pairs BTC/USDT:USDT --timerange=20250405- # freqtrade hyperopt --hyperopt-loss SharpeHyperOptLossDaily --spaces roi stoploss --strategy ChanLun_BTC_K --strategy-path ./user_data/Chan/strategies -c ./user_data/Chan/config/ChanLun_BTC_K.json -e 200 --timerange=20250201-20250401 # sudo docker compose run --rm chan_btc backtesting -c ./user_data/Chan/config/ChanLun_BTC_K.json --strategy ChanLun_BTC_K --strategy-path ./user_data/Chan/strategies --timerange=20250101- # sudo docker compose run --rm chan_btc download-data -c ./user_data/Chan/config/ChanLun_BTC_K.json --pairs BTC/USDT:USDT -t 1m --timerange 20240101- # sudo docker compose run --rm chan_btc trade -c ./user_data/Chan/config/ChanLun_BTC_K.json --strategy ChanLun_BTC_K --strategy-path ./user_data/Chan/strategies class ChanLun_BTC_K(IStrategy): INTERFACE_VERSION: int = 3 # 策略参数 minimal_roi = { "0": 0.004, # 0.4% "15": 0.006, # 15分钟0.6% "30": 0.008, # 30分钟0.8% "60": 0.01 # 60分钟1.0% } stoploss = -0.03 # 3%止损 use_custom_stoploss = True startup_candle_count = 200 def get_ticker_indicator(self) -> int: """返回基础时间框架的分钟数（如 '1m' -> 1）。""" tf = str(self.timeframe).strip().lower() if tf.endswith('m'): return int(tf[:-1]) if tf.endswith('h'): return int(tf[:-1]) * 60 if tf.endswith('d'): return int(tf[:-1]) * 60 * 24 return 1 # 时间框架 timeframe = '5m' # 指标参数 macd_fast = 24 macd_slow = 52 macd_signal = 18 ema_short = 24 ema_long = 52 # 背离检测参数 divergence_lookback = 20 # 背离检测回看周期 min_divergence_bars = 5 # 最小背离确认K线数 def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 计算技术指标 """ # MACD指标 macd = ta.MACD(dataframe, fastperiod=self.macd_fast, slowperiod=self.macd_slow, signalperiod=self.macd_signal) dataframe['macd'] = macd['macd'] dataframe['macdsignal'] = macd['macdsignal'] dataframe['macdhist'] = macd['macdhist'] # EMA均线 dataframe['ema_24'] = ta.EMA(dataframe, timeperiod=self.ema_short) dataframe['ema_52'] = ta.EMA(dataframe, timeperiod=self.ema_long) # 多时间周期（3x、5x、15x）聚合与指标 base_min = self.get_ticker_indicator() intervals = { 'x3': base_min * 3, 'x5': base_min * 5, 'x15': base_min * 15, 'x60': base_min * 60, } def build_htf(df_resampled: DataFrame, suffix: str) -> DataFrame: macd_htf = ta.MACD(df_resampled, fastperiod=self.macd_fast, slowperiod=self.macd_slow, signalperiod=self.macd_signal) df_resampled[f'macd_{suffix}'] = macd_htf['macd'] df_resampled[f'macdsignal_{suffix}'] = macd_htf['macdsignal'] df_resampled[f'macdhist_{suffix}'] = macd_htf['macdhist'] df_resampled[f'ema_24_{suffix}'] = ta.EMA(df_resampled, timeperiod=self.ema_short) df_resampled[f'ema_52_{suffix}'] = ta.EMA(df_resampled, timeperiod=self.ema_long) # ATR及其百分比（用于波动过滤/动态止损） df_resampled[f'atr_{suffix}'] = ta.ATR(df_resampled, timeperiod=14) df_resampled[f'atr_pct_{suffix}'] = df_resampled[f'atr_{suffix}'] / df_resampled['close'] # 近零轴/方向 zero_dist = np.sqrt(np.square(df_resampled[f'macd_{suffix}']) + np.square(df_resampled[f'macdsignal_{suffix}'])) zero_dist_ema = zero_dist.ewm(span=50, adjust=False).mean() zero_eps = zero_dist_ema * 0.2 df_resampled[f'above_zero_{suffix}'] = (df_resampled[f'macd_{suffix}'] > 0) & (df_resampled[f'macdsignal_{suffix}'] > 0) df_resampled[f'below_zero_{suffix}'] = (df_resampled[f'macd_{suffix}'] < 0) & (df_resampled[f'macdsignal_{suffix}'] < 0) df_resampled[f'near_zero_{suffix}'] = (np.abs(df_resampled[f'macd_{suffix}']) < zero_eps) & (np.abs(df_resampled[f'macdsignal_{suffix}']) < zero_eps) df_resampled[f'hist_increasing_{suffix}'] = df_resampled[f'macdhist_{suffix}'] > df_resampled[f'macdhist_{suffix}'].shift(1) df_resampled[f'hist_decreasing_{suffix}'] = df_resampled[f'macdhist_{suffix}'] < df_resampled[f'macdhist_{suffix}'].shift(1) # 高位：远离零轴 df_resampled[f'high_position_{suffix}'] = zero_dist > (zero_dist_ema * 1.5) # 金叉/死叉 df_resampled[f'macd_cross_up_{suffix}'] = (df_resampled[f'macd_{suffix}'] > df_resampled[f'macdsignal_{suffix}']) & (df_resampled[f'macd_{suffix}'].shift(1) <= df_resampled[f'macdsignal_{suffix}'].shift(1)) df_resampled[f'macd_cross_down_{suffix}'] = (df_resampled[f'macd_{suffix}'] < df_resampled[f'macdsignal_{suffix}']) & (df_resampled[f'macd_{suffix}'].shift(1) >= df_resampled[f'macdsignal_{suffix}'].shift(1)) cols = [ 'date', 'close', f'macd_{suffix}', f'macdsignal_{suffix}', f'macdhist_{suffix}', f'ema_24_{suffix}', f'ema_52_{suffix}', f'atr_{suffix}', f'atr_pct_{suffix}', f'above_zero_{suffix}', f'below_zero_{suffix}', f'near_zero_{suffix}', f'hist_increasing_{suffix}', f'hist_decreasing_{suffix}', f'high_position_{suffix}', f'macd_cross_up_{suffix}', f'macd_cross_down_{suffix}' ] # 确保返回独立副本，避免下游在 resampled_merge 内部触发 SettingWithCopyWarning return df_resampled.loc[:, cols].copy() for suf, minutes in intervals.items(): df_res = resample_to_interval(dataframe, minutes) df_htf = build_htf(df_res, suf) dataframe = resampled_merge(dataframe, df_htf) # 动态阈值与距离定义 # 距离零轴的合成距离，用于高位/近零判定 dataframe['macd_abs'] = np.abs(dataframe['macd']) dataframe['macdsignal_abs'] = np.abs(dataframe['macdsignal']) dataframe['zero_dist'] = np.sqrt(np.square(dataframe['macd']) + np.square(dataframe['macdsignal'])) dataframe['zero_dist_ema'] = dataframe['zero_dist'].ewm(span=50, adjust=False).mean() # 近零动态阈值（零轴“无限接近”的量化） dataframe['zero_eps'] = (dataframe['zero_dist_ema'] * 0.2).clip(lower=1e-8) # 零轴判断（方向与近零） dataframe['above_zero'] = (dataframe['macd'] > 0) & (dataframe['macdsignal'] > 0) dataframe['below_zero'] = (dataframe['macd'] < 0) & (dataframe['macdsignal'] < 0) dataframe['near_zero_fast'] = dataframe['macd_abs'] < dataframe['zero_eps'] dataframe['near_zero_slow'] = dataframe['macdsignal_abs'] < dataframe['zero_eps'] dataframe['near_zero'] = dataframe['near_zero_fast'] & dataframe['near_zero_slow'] # MACD与信号线穿越零轴（当根事件，用于阶段/线段识别） dataframe['cross_zero_up'] = ( ((dataframe['macd'].shift(1) <= 0) & (dataframe['macd'] > 0)) | ((dataframe['macdsignal'].shift(1) <= 0) & (dataframe['macdsignal'] > 0)) ) dataframe['cross_zero_down'] = ( ((dataframe['macd'].shift(1) >= 0) & (dataframe['macd'] < 0)) | ((dataframe['macdsignal'].shift(1) >= 0) & (dataframe['macdsignal'] < 0)) ) dataframe['cross_zero'] = dataframe['cross_zero_up'] | dataframe['cross_zero_down'] # 价格触碰/接近EMA52（文档：K线触碰EMA52附近） dataframe['price_above_ema52'] = dataframe['close'] > dataframe['ema_52'] dataframe['price_below_ema52'] = dataframe['close'] < dataframe['ema_52'] dataframe['price_near_ema52'] = (np.abs(dataframe['close'] - dataframe['ema_52']) / dataframe['ema_52']) < 0.003 dataframe['touch_zero_by_price'] = dataframe['price_near_ema52'] # MACD白线（DIF）无限接近零轴（文档：白线靠近零轴） dataframe['touch_zero_by_macd'] = dataframe['near_zero_fast'] # 有效击穿/突破EMA52与零轴（延后确认，信号在确认K线产生，避免前瞻） # 上破EMA52，并在2根K线后仍然在其上 cond_break_up = (dataframe['close'].shift(2) > dataframe['ema_52'].shift(2)) & ( (dataframe['close'].shift(3) <= dataframe['ema_52'].shift(3)) ) dataframe['effective_break_ema52_up'] = cond_break_up.fillna(False) # 下破EMA52，并在2根K线后仍然在其下 cond_break_down = (dataframe['close'].shift(2) < dataframe['ema_52'].shift(2)) & ( (dataframe['close'].shift(3) >= dataframe['ema_52'].shift(3)) ) dataframe['effective_break_ema52_down'] = cond_break_down.fillna(False) # 黄线（慢线：DEA）有效击穿零轴（2根K线后确认） cond_dea_up = (dataframe['macdsignal'].shift(2) > 0) & (dataframe['macdsignal'].shift(3) <= 0) cond_dea_down = (dataframe['macdsignal'].shift(2) < 0) & (dataframe['macdsignal'].shift(3) >= 0) dataframe['effective_dea_cross_up'] = cond_dea_up.fillna(False) dataframe['effective_dea_cross_down'] = cond_dea_down.fillna(False) # 高位空形态检测 # 高位空：MACD黄白线处于高位，K线缓慢上涨或横盘，能量柱衰减，形成夹角 # 高位：距离零轴远离，采用动态阈值（> 1.5x 距离均值） dataframe['high_position'] = dataframe['zero_dist'] > (dataframe['zero_dist_ema'] * 1.5) # 能量柱衰减检测 dataframe['histogram_decreasing'] = dataframe['macdhist'] < dataframe['macdhist'].shift(1) dataframe['histogram_increasing'] = dataframe['macdhist'] > dataframe['macdhist'].shift(1) # 线条“横盘”（变化不大）：3根之前差值很小 dataframe['macd_flat_3'] = (np.abs(dataframe['macd'] - dataframe['macd'].shift(3)) < dataframe['zero_eps']) dataframe['macdsignal_flat_3'] = (np.abs(dataframe['macdsignal'] - dataframe['macdsignal'].shift(3)) < dataframe['zero_eps']) # 高位空形态：高位 + 能量柱衰减 + 黄白线横盘 dataframe['high_position_empty'] = ( dataframe['high_position'] & dataframe['histogram_decreasing'] & # K线缓慢上涨或横盘（价格变化不大） (abs(dataframe['close'] - dataframe['close'].shift(3)) / dataframe['close'].shift(3) < 0.02) & # MACD黄白线横盘（变化不大） dataframe['macd_flat_3'] & dataframe['macdsignal_flat_3'] ) # 归零轴四种走势（近似量化） # 1) 触碰EMA52（由上至下或下至上） dataframe['zero_touch_ema52'] = dataframe['price_near_ema52'] # 2) 白线无限接近零轴 dataframe['zero_near_fastline'] = dataframe['near_zero_fast'] # 3) 零轴粘合：刚穿零轴后，|黄白线|均小，hist不释放反向能量柱，斜率小（横向） small_lines = (dataframe['macd_abs'] < dataframe['zero_eps'] * 1.2) & (dataframe['macdsignal_abs'] < dataframe['zero_eps'] * 1.2) same_side_hist = ( ((dataframe['macdhist'] >= 0) & dataframe['cross_zero_up']) | ((dataframe['macdhist'] <= 0) & dataframe['cross_zero_down']) ) dataframe['zero_axis_adhesion'] = small_lines & same_side_hist # 4) K线先触碰EMA52，而黄白线未归零 dataframe['zero_touch_price_first'] = dataframe['price_near_ema52'] & (~dataframe['near_zero']) # 零轴纠缠：黄白线反复在近零区上下缠绕（5根内多次变号或绝对值很小） near_zero_many = dataframe['near_zero'].rolling(5).sum() >= 3 sign_flip_fast = (np.sign(dataframe['macd']) != np.sign(dataframe['macd'].shift(1))) sign_flip_slow = (np.sign(dataframe['macdsignal']) != np.sign(dataframe['macdsignal'].shift(1))) dataframe['zero_axis_entanglement'] = near_zero_many | (sign_flip_fast & sign_flip_slow & dataframe['near_zero']) # 零轴倒挂：靠近零轴、能量柱衰减形成夹角、黄白线交叉并释放反向能量柱 macd_cross = ((dataframe['macd'] > dataframe['macdsignal']) & (dataframe['macd'].shift(1) <= dataframe['macdsignal'].shift(1))) | ( (dataframe['macd'] < dataframe['macdsignal']) & (dataframe['macd'].shift(1) >= dataframe['macdsignal'].shift(1)) ) hist_flip = np.sign(dataframe['macdhist']) != np.sign(dataframe['macdhist'].shift(1)) dataframe['zero_axis_inverted'] = dataframe['near_zero'] & dataframe['histogram_decreasing'] & macd_cross & hist_flip # 隐形形态：无能量配合 # 高位隐形：远离零轴、价格继续拉升/下跌，但hist未释放同向能量 dataframe['hidden_high_bull'] = dataframe['above_zero'] & dataframe['high_position'] & (dataframe['close'] > dataframe['close'].shift(1)) & (dataframe['macdhist'] <= 0) dataframe['hidden_high_bear'] = dataframe['below_zero'] & dataframe['high_position'] & (dataframe['close'] < dataframe['close'].shift(1)) & (dataframe['macdhist'] >= 0) # 归零轴隐形：近零时应释放的支撑/压力能量未出现 -> 可能反向穿零 dataframe['hidden_zero_bull_fail'] = dataframe['near_zero'] & dataframe['price_near_ema52'] & (dataframe['macdhist'] <= 0) dataframe['hidden_zero_bear_fail'] = dataframe['near_zero'] & dataframe['price_near_ema52'] & (dataframe['macdhist'] >= 0) # 斜率/拐点/金叉死叉（上下文门控） dataframe['ema24_slope_up'] = dataframe['ema_24'] > dataframe['ema_24'].shift(1) dataframe['ema24_slope_down'] = dataframe['ema_24'] < dataframe['ema_24'].shift(1) dataframe['hist_turn_up'] = (dataframe['macdhist'] > dataframe['macdhist'].shift(1)) & (dataframe['macdhist'].shift(1) <= dataframe['macdhist'].shift(2)) dataframe['hist_turn_down'] = (dataframe['macdhist'] < dataframe['macdhist'].shift(1)) & (dataframe['macdhist'].shift(1) >= dataframe['macdhist'].shift(2)) dataframe['macd_cross_up'] = (dataframe['macd'] > dataframe['macdsignal']) & (dataframe['macd'].shift(1) <= dataframe['macdsignal'].shift(1)) dataframe['macd_cross_down'] = (dataframe['macd'] < dataframe['macdsignal']) & (dataframe['macd'].shift(1) >= dataframe['macdsignal'].shift(1)) # 线段与单位调整周期（近似） - 需在背离检测之前生成 # 线段：以黄线穿零轴划分段（上穿为上涨线段，下穿为下跌线段） seg_change = (((dataframe['macdsignal'] <= 0) & (dataframe['macdsignal'].shift(1) > 0)) | ((dataframe['macdsignal'] >= 0) & (dataframe['macdsignal'].shift(1) < 0))) dataframe['segment_id'] = seg_change.cumsum().fillna(0).astype(int) # 单位调整周期：由近零出发-远离-回到近零（用近零作为粗略起止标记） dataframe['near_zero_flag'] = dataframe['near_zero'].astype(int) dataframe['unit_cycle_id'] = (dataframe['near_zero_flag'].diff().fillna(0) > 0).cumsum().astype(int) # 背离检测（线段内） dataframe = self.detect_divergence(dataframe) # 跳空检测 dataframe = self.detect_gaps(dataframe) # V字反转：近零+收敛+突破横盘区 price_break = dataframe['close'] > dataframe['close'].rolling(10).max().shift(1) macd_converge = dataframe['histogram_decreasing'].rolling(4).sum() >= 3 dataframe['v_reversal'] = dataframe['near_zero'] & dataframe['price_above_ema52'] & macd_converge & price_break # 抢底原理（第三阶段：底背离/动能不足触发） momentum_lack = (dataframe['below_zero'] & dataframe['histogram_decreasing'] & (dataframe['close'] <= dataframe['close'].shift(1))) dataframe['bottom_snap_buy'] = dataframe['near_zero'] & (dataframe['bottom_divergence'] | momentum_lack) # 归零轴强支撑（零轴粘合 + EMA52支撑） dataframe['zero_adhesion_support'] = dataframe['zero_axis_adhesion'] & dataframe['price_near_ema52'] & dataframe['price_above_ema52'] # 高周期门控（5x 与 60x），注意列名经过 resampled_merge 改名：resample_{minutes}_ min3 = intervals['x3'] min5 = intervals['x5'] min15 = intervals['x15'] min60 = intervals['x60'] ema24_5 = f'resample_{min5}_ema_24_x5' ema52_5 = f'resample_{min5}_ema_52_x5' above0_5 = f'resample_{min5}_above_zero_x5' near0_5 = f'resample_{min5}_near_zero_x5' macds_5 = f'resample_{min5}_macdsignal_x5' d5 = f'resample_{min5}_date' close5 = f'resample_{min5}_close' ema24_60 = f'resample_{min60}_ema_24_x60' ema52_60 = f'resample_{min60}_ema_52_x60' macds_60 = f'resample_{min60}_macdsignal_x60' above0_60 = f'resample_{min60}_above_zero_x60' near0_60 = f'resample_{min60}_near_zero_x60' date_60 = f'resample_{min60}_date' dataframe['htf_buy_gate'] = ( (dataframe.get(ema24_60, np.nan) > dataframe.get(ema52_60, np.nan)) & ((dataframe.get(above0_60, False)) | (dataframe.get(near0_60, False)) | (dataframe.get(macds_60, np.nan) >= 0)) ).fillna(False) dataframe['htf_sell_gate'] = ( (dataframe.get(ema24_60, np.nan) < dataframe.get(ema52_60, np.nan)) | (dataframe.get(macds_60, np.nan) < 0) ).fillna(False) # 1m 对 60m 均线关系 dataframe['price_above_ema52_x60'] = (dataframe['close'] > dataframe.get(ema52_60, np.nan)).fillna(False) dataframe['price_near_ema52_x60'] = (np.abs(dataframe['close'] - dataframe.get(ema52_60, np.nan)) / dataframe.get(ema52_60, np.nan) < 0.003).fillna(False) # 提供无前缀别名，供买卖条件使用（60m） dataframe['near_zero_x60'] = dataframe.get(near0_60, False) dataframe['macdsignal_x60'] = dataframe.get(macds_60, np.nan) dataframe['ema24_x60'] = dataframe.get(ema24_60, np.nan) dataframe['ema52_x60'] = dataframe.get(ema52_60, np.nan) dataframe['hist_increasing_x60'] = dataframe.get(f'resample_{min60}_hist_increasing_x60', False) dataframe['hist_decreasing_x60'] = dataframe.get(f'resample_{min60}_hist_decreasing_x60', False) dataframe['macd_cross_up_60m'] = dataframe.get(f'resample_{min60}_macd_cross_up_x60', False) dataframe['macd_cross_down_60m'] = dataframe.get(f'resample_{min60}_macd_cross_down_x60', False) # 60m边界触发：仅在新60m开始的一根1m上允许交易（直接比较，避免时区转换问题） d60 = dataframe.get(date_60) dataframe['is_new_60m'] = d60.ne(d60.shift(1)).fillna(False) # 5m边界（用于5m信号仅在新5m产生） d5s = dataframe.get(d5) dataframe['is_new_5m'] = d5s.ne(d5s.shift(1)).fillna(False) # 60m开始后前5分钟内也允许交易 if 'date' in dataframe.columns: dt_delta60 = (dataframe['date'] - d60) dataframe['within_first_60m5'] = dt_delta60.dt.total_seconds().div(60).between(0, 10).fillna(False) else: dataframe['within_first_60m5'] = dataframe['is_new_60m'] # 60m波动过滤 dataframe['atr_pct_x60'] = dataframe.get(f'resample_{min60}_atr_pct_x60', np.nan) dataframe['htf_vol_ok'] = (dataframe['atr_pct_x60'] > 0.0005).fillna(False) # 价格接近1h EMA24（小回踩判定） dataframe['price_near_ema24_x60'] = (np.abs(dataframe['close'] - dataframe.get('ema24_x60', np.nan)) / dataframe.get('ema24_x60', np.nan) < 0.0015).fillna(False) # 5m别名与突破判定 dataframe['near_zero_x5'] = dataframe.get(near0_5, False) dataframe['macd_cross_up_5m'] = dataframe.get(f'resample_{min5}_macd_cross_up_x5', False) dataframe['macd_cross_down_5m'] = dataframe.get(f'resample_{min5}_macd_cross_down_x5', False) dataframe['ema24_x5'] = dataframe.get(ema24_5, np.nan) dataframe['ema52_x5'] = dataframe.get(ema52_5, np.nan) dataframe['close_x5'] = dataframe.get(close5, np.nan) # 5m突破：新5m且收盘突破近20根5m最高收盘 dataframe['breakout_5m'] = ( dataframe['is_new_5m'] & (dataframe['close_x5'] > dataframe['close_x5'].rolling(20).max().shift(1)) ).fillna(False) return dataframe def detect_divergence(self, dataframe: DataFrame) -> DataFrame: """ 检测背离形态 """ # 顶/底背离（限制在同一线段内比较，避免跨段） df = dataframe df['top_divergence'] = False df['bottom_divergence'] = False # 线段内的滚动极值（不跨段） seg_group = df.groupby('segment_id', group_keys=False) seg_close_max_prev = seg_group['close'].apply(lambda s: s.cummax().shift(1)) seg_macd_max_prev = seg_group['macd'].apply(lambda s: s.cummax().shift(1)) seg_close_min_prev = seg_group['close'].apply(lambda s: s.cummin().shift(1)) seg_macd_min_prev = seg_group['macd'].apply(lambda s: s.cummin().shift(1)) cond_top = (df['close'] > seg_close_max_prev) & (df['macd'] < seg_macd_max_prev) & df['above_zero'] cond_bottom = (df['close'] < seg_close_min_prev) & (df['macd'] > seg_macd_min_prev) & df['below_zero'] df.loc[cond_top.fillna(False), 'top_divergence'] = True df.loc[cond_bottom.fillna(False), 'bottom_divergence'] = True return df def detect_gaps(self, dataframe: DataFrame) -> DataFrame: """ 检测跳空形态 """ # 连续跳空/分立跳空（单位周期内的近似判定） df = dataframe df['continuous_gap'] = False df['separate_gap'] = False # 能量柱包含在黄白线之内：|hist| <= max(|macd|, |signal|) hist_within_lines = df['macdhist'].abs() <= np.maximum(df['macd_abs'], df['macdsignal_abs']) for i in range(5, len(df)): # 连续跳空：同向能量柱在单位周期中由衰减转为增长，且能量柱包含在线内 if ( df['histogram_increasing'].iloc[i-2:i+1].all() and hist_within_lines.iloc[i-2:i+1].all() and ((df['macdhist'].iloc[i] > 0) | (df['macdhist'].iloc[i] < 0)) and # 近似单位周期：最近出现过near_zero且当前未再次near_zero (df['near_zero'].iloc[i-10:i].any()) and (not df['near_zero'].iloc[i]) ): df.loc[df.index[i], 'continuous_gap'] = True # 分立跳空：两个同向能量堆被反向能量柱分隔，且远离零轴 if i > 10: recent_hist = df['macdhist'].iloc[i-10:i+1] same_dir = (recent_hist.max() > 0 and recent_hist.min() < 0) far_from_zero = df['high_position'].iloc[i] if same_dir and far_from_zero: # 当前是同向新峰，且此前5根内出现过反向柱 if (df['macdhist'].iloc[i] > 0 and (recent_hist.iloc[-6:-1] < 0).any() and df['macdhist'].iloc[i] > recent_hist.iloc[:-1].max()) or ( df['macdhist'].iloc[i] < 0 and (recent_hist.iloc[-6:-1] > 0).any() and df['macdhist'].iloc[i] < recent_hist.iloc[:-1].min() ): df.loc[df.index[i], 'separate_gap'] = True return df def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 买入信号生成 """ conditions = [] # 条件1: 60m门控 + 底背离确认买点（同段内） conditions.append( dataframe['bottom_divergence'] & dataframe['below_zero'] & (dataframe['price_near_ema52_x60'] | dataframe['near_zero_x60'] | dataframe['near_zero']) & (dataframe['ema24_slope_up']) & dataframe['htf_vol_ok'] & (dataframe['htf_buy_gate']) & (dataframe['is_new_60m'] | dataframe['within_first_60m5']) ) # 条件2: 单位调整周期内的连续跳空背离 conditions.append( dataframe['continuous_gap'] & dataframe['below_zero'] & dataframe['near_zero'] & (dataframe['ema24_slope_up']) & dataframe['htf_vol_ok'] & (dataframe['price_near_ema52_x60'] | dataframe['near_zero_x60'] | dataframe['price_near_ema24_x60']) & (dataframe['htf_buy_gate']) ) # 条件3: 底部形态V字反转 或 5m突破 conditions.append( ( dataframe['v_reversal'] & dataframe['macd_cross_up'] | (dataframe['breakout_5m'] & dataframe['macd_cross_up_5m']) ) & (dataframe['price_above_ema52_x60'] | (dataframe['macdsignal_x60'] >= 0)) & dataframe['htf_vol_ok'] & (dataframe['htf_buy_gate']) ) # 条件4: 抢底原理（第三阶段：底背离/动能不足叠加） conditions.append( dataframe['bottom_snap_buy'] & (dataframe['macd_cross_up'] | dataframe['hist_turn_up']) & (dataframe['price_near_ema52_x60'] | dataframe['near_zero_x60'] | dataframe['price_near_ema24_x60']) & (dataframe['htf_buy_gate']) & dataframe['htf_vol_ok'] ) # 条件5: 归零轴反弹（近零+EMA52附近+能量回升） conditions.append( dataframe['near_zero'] & (dataframe['price_near_ema52_x60'] | dataframe['price_near_ema24_x60']) & dataframe['histogram_increasing'] & (dataframe['close'] > dataframe['close'].shift(1)) & (dataframe['ema24_slope_up']) & (dataframe['htf_buy_gate']) & dataframe['htf_vol_ok'] ) # 条件6: 零轴粘合强支撑买入（文档强调强支撑、弱反弹） conditions.append( dataframe['zero_adhesion_support'] & dataframe['macd_cross_up'] & (dataframe['price_above_ema52_x60'] | dataframe['near_zero_x60'] | dataframe['price_near_ema24_x60']) & (dataframe['htf_buy_gate']) & dataframe['htf_vol_ok'] ) if conditions: dataframe.loc[ reduce(lambda x, y: x | y, conditions), 'enter_long'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 卖出信号生成 """ conditions = [] # 条件1: 顶背离确认卖点（同段内） conditions.append( dataframe['top_divergence'] & dataframe['above_zero'] ) # 条件2: 高位空形态 conditions.append( dataframe['high_position_empty'] & dataframe['above_zero'] & (dataframe['macd_cross_down'] | dataframe['hist_turn_down']) ) # 条件3: 有效穿零轴下跌（慢线有效击穿 + EMA52下） conditions.append( (dataframe['effective_dea_cross_down'] | dataframe['cross_zero_down'] | dataframe['htf_sell_gate']) & dataframe['price_below_ema52'] & (dataframe['ema24_slope_down']) ) # 条件4: 能量柱隐形形态（无能量配合的上涨） conditions.append( dataframe['hidden_high_bull'] & (dataframe['macd_cross_down'] | dataframe['hist_turn_down']) ) # 条件5: 零轴倒挂（弱支撑弱反弹，易继续下行） conditions.append( dataframe['zero_axis_inverted'] & dataframe['above_zero'] ) if conditions: dataframe.loc[ reduce(lambda x, y: x | y, conditions), 'exit_long'] = 1 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: Optional[str], side: str, **kwargs) -> bool: """ 交易确认 """ # 获取当前数据 dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # 买入确认 if side == 'buy': # 确保MACD在零轴下方且有反弹迹象 if not (last_candle['below_zero'] or last_candle['near_zero']): return False # 确保价格接近EMA52 if not last_candle['price_near_ema52']: return False # 卖出确认 elif side == 'sell': # 确保MACD在零轴上方且有下跌迹象 if not (last_candle['above_zero'] or last_candle['near_zero']): return False return True 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].squeeze() # 如果出现顶背离，立即止损 if last_candle['top_divergence']: return -0.01 # 1%止损 # 如果价格跌破EMA52，止损 if last_candle['price_below_ema52'] and current_profit < 0: return -0.02 # 2%止损 # 如果MACD穿零轴向下，止损 if last_candle['cross_zero'] and last_candle['macd'] < 0: return -0.015 # 1.5%止损 return self.stoploss