324 lines
13 KiB
Python
324 lines
13 KiB
Python
# --- 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=20250721-
|
|
# 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.05, # 5% 利润即可退出
|
|
"30": 0.03, # 30分钟后3%利润退出
|
|
"60": 0.02, # 1小时后2%利润退出
|
|
"120": 0.01 # 2小时后1%利润退出
|
|
}
|
|
|
|
stoploss = -0.03 # 3%止损
|
|
|
|
# 时间框架
|
|
timeframe = '1m'
|
|
|
|
# 指标参数
|
|
macd_fast = 12
|
|
macd_slow = 26
|
|
macd_signal = 9
|
|
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)
|
|
|
|
# 零轴判断
|
|
dataframe['above_zero'] = (dataframe['macd'] > 0) & (dataframe['macdsignal'] > 0)
|
|
dataframe['below_zero'] = (dataframe['macd'] < 0) & (dataframe['macdsignal'] < 0)
|
|
dataframe['cross_zero'] = (
|
|
(dataframe['macd'].shift(1) < 0) & (dataframe['macd'] > 0) |
|
|
(dataframe['macdsignal'].shift(1) < 0) & (dataframe['macdsignal'] > 0)
|
|
)
|
|
|
|
# 高位空形态检测
|
|
# 高位空：MACD黄白线处于高位，K线缓慢上涨或横盘，能量柱衰减，形成夹角
|
|
dataframe['high_position'] = (
|
|
# MACD黄白线远离零轴（高位）
|
|
((dataframe['macd'] > 50) & (dataframe['macdsignal'] > 50)) |
|
|
((dataframe['macd'] < -50) & (dataframe['macdsignal'] < -50))
|
|
)
|
|
# 能量柱衰减检测
|
|
dataframe['histogram_decreasing'] = dataframe['macdhist'] < dataframe['macdhist'].shift(1)
|
|
dataframe['histogram_increasing'] = dataframe['macdhist'] > dataframe['macdhist'].shift(1)
|
|
|
|
# 高位空形态：高位 + 能量柱衰减 + 黄白线横盘
|
|
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黄白线横盘（变化不大）
|
|
(abs(dataframe['macd'] - dataframe['macd'].shift(3)) < 0.05) &
|
|
(abs(dataframe['macdsignal'] - dataframe['macdsignal'].shift(3)) < 0.05)
|
|
)
|
|
|
|
|
|
# 归零轴检测
|
|
dataframe['near_zero'] = (
|
|
(abs(dataframe['macd']) < 0.1) & (abs(dataframe['macdsignal']) < 0.1)
|
|
)
|
|
|
|
# 价格与EMA52关系
|
|
dataframe['price_above_ema52'] = dataframe['close'] > dataframe['ema_52']
|
|
dataframe['price_below_ema52'] = dataframe['close'] < dataframe['ema_52']
|
|
dataframe['price_near_ema52'] = abs(dataframe['close'] - dataframe['ema_52']) / dataframe['ema_52'] < 0.01
|
|
|
|
# 背离检测
|
|
dataframe = self.detect_divergence(dataframe)
|
|
|
|
# 跳空检测
|
|
dataframe = self.detect_gaps(dataframe)
|
|
|
|
return dataframe
|
|
|
|
def detect_divergence(self, dataframe: DataFrame) -> DataFrame:
|
|
"""
|
|
检测背离形态
|
|
"""
|
|
# 顶背离检测
|
|
dataframe['top_divergence'] = False
|
|
dataframe['bottom_divergence'] = False
|
|
|
|
for i in range(self.divergence_lookback, len(dataframe)):
|
|
# 顶背离：价格创新高，MACD未创新高
|
|
if (dataframe['close'].iloc[i] > dataframe['close'].iloc[i-self.divergence_lookback:i].max() and
|
|
dataframe['macd'].iloc[i] < dataframe['macd'].iloc[i-self.divergence_lookback:i].max() and
|
|
dataframe['above_zero'].iloc[i]):
|
|
dataframe.loc[dataframe.index[i], 'top_divergence'] = True
|
|
|
|
# 底背离：价格创新低，MACD未创新低
|
|
if (dataframe['close'].iloc[i] < dataframe['close'].iloc[i-self.divergence_lookback:i].min() and
|
|
dataframe['macd'].iloc[i] > dataframe['macd'].iloc[i-self.divergence_lookback:i].min() and
|
|
dataframe['below_zero'].iloc[i]):
|
|
dataframe.loc[dataframe.index[i], 'bottom_divergence'] = True
|
|
|
|
return dataframe
|
|
|
|
def detect_gaps(self, dataframe: DataFrame) -> DataFrame:
|
|
"""
|
|
检测跳空形态
|
|
"""
|
|
# 连续跳空检测
|
|
dataframe['continuous_gap'] = False
|
|
dataframe['separate_gap'] = False
|
|
|
|
for i in range(5, len(dataframe)):
|
|
# 连续跳空：能量柱连续增长
|
|
if (dataframe['histogram_increasing'].iloc[i-2:i+1].all() and
|
|
dataframe['macdhist'].iloc[i] > 0 and
|
|
dataframe['macdhist'].iloc[i] > dataframe['macdhist'].iloc[i-1]):
|
|
dataframe.loc[dataframe.index[i], 'continuous_gap'] = True
|
|
|
|
# 分立跳空：能量柱被反向能量柱分隔
|
|
if (i > 10 and
|
|
dataframe['macdhist'].iloc[i] > 0 and
|
|
dataframe['macdhist'].iloc[i-5:i].min() < 0 and
|
|
dataframe['macdhist'].iloc[i] > dataframe['macdhist'].iloc[i-5:i].max()):
|
|
dataframe.loc[dataframe.index[i], 'separate_gap'] = True
|
|
|
|
return dataframe
|
|
|
|
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
"""
|
|
买入信号生成
|
|
"""
|
|
conditions = []
|
|
|
|
# 条件1: 底背离确认买点
|
|
conditions.append(
|
|
dataframe['bottom_divergence'] &
|
|
dataframe['below_zero'] &
|
|
dataframe['price_near_ema52']
|
|
)
|
|
|
|
# 条件2: 单位调整周期内的连续跳空背离
|
|
conditions.append(
|
|
dataframe['continuous_gap'] &
|
|
dataframe['below_zero'] &
|
|
dataframe['near_zero']
|
|
)
|
|
|
|
# 条件3: 底部形态V字反转
|
|
conditions.append(
|
|
dataframe['price_above_ema52'] &
|
|
dataframe['near_zero'] &
|
|
dataframe['histogram_increasing'] &
|
|
(dataframe['close'] > dataframe['close'].shift(5))
|
|
)
|
|
|
|
# 条件4: 抢底原理（第三阶段背离/动能不足）
|
|
conditions.append(
|
|
dataframe['below_zero'] &
|
|
dataframe['near_zero'] &
|
|
dataframe['histogram_decreasing'] &
|
|
(dataframe['macd'] > dataframe['macd'].shift(3)) # MACD开始收敛
|
|
)
|
|
|
|
# 条件5: 归零轴反弹
|
|
conditions.append(
|
|
dataframe['near_zero'] &
|
|
dataframe['price_near_ema52'] &
|
|
dataframe['histogram_increasing'] &
|
|
(dataframe['close'] > dataframe['close'].shift(1))
|
|
)
|
|
|
|
# 条件6: 零轴之下高位空形态（归零轴需求）
|
|
conditions.append(
|
|
dataframe['high_position_empty'] &
|
|
dataframe['below_zero']
|
|
)
|
|
|
|
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']
|
|
)
|
|
|
|
# 条件3: 穿零轴下跌
|
|
conditions.append(
|
|
dataframe['cross_zero'] &
|
|
dataframe['price_below_ema52'] &
|
|
(dataframe['macd'] < 0)
|
|
)
|
|
|
|
# 条件4: 能量柱隐形形态（无能量配合的上涨）
|
|
conditions.append(
|
|
dataframe['above_zero'] &
|
|
(dataframe['macdhist'] < 0) &
|
|
(dataframe['close'] > dataframe['close'].shift(1))
|
|
)
|
|
|
|
# 条件5: 线段背离（价格创新高但MACD未创新高）
|
|
conditions.append(
|
|
dataframe['above_zero'] &
|
|
(dataframe['close'] > dataframe['close'].shift(10).max()) &
|
|
(dataframe['macd'] < dataframe['macd'].shift(10).max())
|
|
)
|
|
|
|
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 |