302 lines
12 KiB
Python
302 lines
12 KiB
Python
|
|
# --- Do not remove these libs ---
|
|
from freqtrade.strategy import IStrategy
|
|
from typing import Dict, List
|
|
from functools import reduce
|
|
from pandas import DataFrame
|
|
import numpy as np
|
|
import pandas as pd
|
|
# --------------------------------
|
|
|
|
# 设置pandas选项以避免FutureWarning
|
|
pd.set_option('future.no_silent_downcasting', True)
|
|
|
|
import talib.abstract as ta
|
|
import freqtrade.vendor.qtpylib.indicators as qtpylib
|
|
from technical.util import resample_to_interval, resampled_merge
|
|
from freqtrade.persistence import Trade, Order
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
# freqtrade plot-dataframe --strategy BB9033 --datadir user_data/data/binance -c ./user_data/ChanLun_SOL_30.json --timerange=20250309-
|
|
|
|
# freqtrade trade -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy BB9033 --strategy-path ./user_data/Chan/strategies
|
|
# freqtrade backtesting -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy BB9033 --strategy-path ./user_data/Chan/strategies --timerange=20250623-
|
|
# freqtrade download-data -c ./user_data/Chan/config/ChanLun_BTC_30.json -t 1m --pairs BTC/USDT:USDT --timerange=20250501-
|
|
# freqtrade hyperopt --hyperopt-loss SharpeHyperOptLossDaily --spaces roi stoploss --strategy BB9033 --strategy-path ./user_data/Chan/strategies -c ./user_data/Chan/config/ChanLun_BTC_30.json -e 200 --timerange=20250201-20250401
|
|
|
|
# sudo docker compose run --rm chan_btc backtesting -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy BB9033 --strategy-path ./user_data/Chan/strategies --timerange=20250101-
|
|
# sudo docker compose run --rm chan_btc download-data -c ./user_data/Chan/config/ChanLun_BTC_30.json --pairs BTC/USDT:USDT -t 1m --timerange 20240101-
|
|
# sudo docker compose run --rm chan_btc trade -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy BB9033 --strategy-path ./user_data/Chan/strategies
|
|
|
|
class BB9033(IStrategy):
|
|
"""
|
|
布林带反转策略（与Pine Script保持一致）
|
|
基于EMA和标准差计算布林带，实现反转交易
|
|
|
|
交易逻辑：
|
|
- 做多：价格跌破下轨后反转
|
|
- 做空：价格突破上轨后反转
|
|
- 做多止盈：价格减去0.5倍ATR突破上轨
|
|
- 做空止盈：价格跌破下轨
|
|
- 止损：基于ATR动态设置
|
|
"""
|
|
|
|
INTERFACE_VERSION: int = 3
|
|
|
|
# 策略参数（与Pine Script保持一致）
|
|
bb_length = 41 # 布林带长度
|
|
atr_multiplier = 2.3 # 布林带倍数
|
|
atr_stop_multiplier = 3 # 止损ATR倍数
|
|
atr_length = 11 # ATR计算周期
|
|
|
|
# Minimal ROI designed for the strategy.
|
|
# This attribute will be overridden if the config file contains "minimal_roi"
|
|
minimal_roi = {
|
|
}
|
|
can_short = True
|
|
# Optimal stoploss designed for the strategy
|
|
# This attribute will be overridden if the config file contains "stoploss"
|
|
stoploss = -0.3
|
|
use_custom_stoploss = True
|
|
|
|
# Optimal timeframe for the strategy
|
|
time = 5
|
|
# Trailing stop loss
|
|
trailing_stop = False
|
|
lev = 1.0
|
|
# Run "populate_indicators" only for new candle
|
|
process_only_new_candles = False
|
|
|
|
# Number of candles the strategy requires before producing valid signals
|
|
startup_candle_count: int = max(bb_length*time, atr_length*time) + 10
|
|
|
|
# 存储每个交易的止损价格
|
|
trade_stop_prices: Dict[str, float] = {}
|
|
|
|
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
"""
|
|
计算技术指标（与Pine Script保持一致）
|
|
"""
|
|
dataframe_3 = resample_to_interval(dataframe, self.get_ticker_indicator() * self.time)
|
|
|
|
# 计算ATR（用于止损计算）
|
|
dataframe_3['atr'] = ta.ATR(dataframe_3, timeperiod=self.atr_length)
|
|
|
|
# 计算布林带（使用EMA作为基础，与Pine Script保持一致）
|
|
bb_basis = ta.EMA(dataframe_3['close'], timeperiod=self.bb_length)
|
|
bb_dev = self.atr_multiplier * ta.STDDEV(dataframe_3['close'], timeperiod=self.bb_length)
|
|
dataframe_3['bb_upper'] = bb_basis + bb_dev
|
|
dataframe_3['bb_middle'] = bb_basis
|
|
dataframe_3['bb_lower'] = bb_basis - bb_dev
|
|
|
|
# 计算突破条件（与Pine Script保持一致）
|
|
# 确保所有用于计算的数据都不是NaN
|
|
valid_data = (
|
|
dataframe_3['close'].notna() &
|
|
dataframe_3['bb_upper'].notna() &
|
|
dataframe_3['bb_lower'].notna() &
|
|
dataframe_3['close'].shift(1).notna() &
|
|
dataframe_3['bb_upper'].shift(1).notna() &
|
|
dataframe_3['bb_lower'].shift(1).notna()
|
|
)
|
|
|
|
# 做空条件：价格突破上轨
|
|
dataframe_3['break_above_upper'] = (
|
|
(dataframe_3['close'] > dataframe_3['bb_upper']) &
|
|
(dataframe_3['close'].shift(1) <= dataframe_3['bb_upper'].shift(1)) &
|
|
valid_data
|
|
)
|
|
|
|
# 做多条件：价格跌破下轨
|
|
dataframe_3['break_below_lower'] = (
|
|
(dataframe_3['close'] < dataframe_3['bb_lower']) &
|
|
(dataframe_3['close'].shift(1) >= dataframe_3['bb_lower'].shift(1)) &
|
|
valid_data
|
|
)
|
|
|
|
dataframe = resampled_merge(dataframe, dataframe_3)
|
|
return dataframe
|
|
|
|
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
"""
|
|
Based on TA indicators, populates the entry trend columns
|
|
"""
|
|
break_below_lower = 'resample_{}_break_below_lower'.format(self.get_ticker_indicator()*self.time)
|
|
break_above_upper = 'resample_{}_break_above_upper'.format(self.get_ticker_indicator()*self.time)
|
|
|
|
# 检测多头信号：价格跌破下轨
|
|
dataframe.loc[
|
|
(
|
|
(dataframe[break_below_lower] == True) &
|
|
(pd.notna(dataframe[break_below_lower]))
|
|
),
|
|
['enter_long', 'enter_tag']] = (1, 'long_signal_chan')
|
|
|
|
# 检测空头信号：价格突破上轨
|
|
dataframe.loc[
|
|
(
|
|
(dataframe[break_above_upper] == True) &
|
|
(pd.notna(dataframe[break_above_upper]))
|
|
),
|
|
['enter_short', 'enter_tag']] = (1, 'short_signal_chan')
|
|
|
|
return dataframe
|
|
|
|
|
|
|
|
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
"""
|
|
Based on TA indicators, populates the exit trend columns
|
|
"""
|
|
bb_upper_str = 'resample_{}_bb_upper'.format(self.get_ticker_indicator()*self.time)
|
|
bb_lower_str = 'resample_{}_bb_lower'.format(self.get_ticker_indicator()*self.time)
|
|
atr_str = 'resample_{}_atr'.format(self.get_ticker_indicator()*self.time)
|
|
close_str = 'resample_{}_close'.format(self.get_ticker_indicator()*self.time)
|
|
low_str = 'resample_{}_low'.format(self.get_ticker_indicator()*self.time)
|
|
|
|
# 做多止盈条件：价格突破上轨（与Pine Script保持一致）
|
|
dataframe.loc[
|
|
(
|
|
(dataframe[close_str] > dataframe[bb_upper_str]) & # close-atr_value*0.5 > bb_upper
|
|
(dataframe[bb_upper_str].notna()) & # 确保布林带上轨不是NaN
|
|
(dataframe[close_str].notna()) & # 确保收盘价不是NaN
|
|
(dataframe[atr_str].notna()) & # 确保ATR不是NaN
|
|
(len(self.trade_stop_prices) > 0)
|
|
),
|
|
['exit_long', 'exit_tag']] = (1, 'long_close_signal_chan')
|
|
|
|
# 做空止盈条件：价格跌破下轨（与Pine Script保持一致）
|
|
dataframe.loc[
|
|
(
|
|
(dataframe[low_str] < dataframe[bb_lower_str]) & # low < bb_lower
|
|
(dataframe[bb_lower_str].notna()) & # 确保布林带下轨不是NaN
|
|
(dataframe[low_str].notna()) & # 确保最低价不是NaN
|
|
(len(self.trade_stop_prices) > 0)
|
|
),
|
|
['exit_short', 'exit_tag']] = (1, 'short_close_signal_chan')
|
|
|
|
return dataframe
|
|
|
|
def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime,
|
|
current_rate: float, current_profit: float,
|
|
after_fill: bool, **kwargs) -> float:
|
|
"""
|
|
自定义止损逻辑：使用开单时记录的ATR止损价格
|
|
"""
|
|
|
|
# 检查是否有存储的止损价格
|
|
trade_id = str(trade.id)
|
|
if trade_id not in self.trade_stop_prices:
|
|
return self.stoploss
|
|
|
|
# 获取最新的K线数据
|
|
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
|
if len(dataframe) < 2:
|
|
return self.stoploss
|
|
|
|
# 使用上一个K线的收盘价（重采样后的数据）
|
|
close_str = 'resample_{}_close'.format(self.get_ticker_indicator()*self.time)
|
|
last_close = dataframe.iloc[-2][close_str] # 上一个完整K线的收盘价
|
|
|
|
stop_price = self.trade_stop_prices[trade_id]
|
|
|
|
if trade.is_short:
|
|
# 空头止损：上一个K线收盘价超过止损价格时触发止损
|
|
if last_close >= stop_price:
|
|
stop_loss_pct = -abs((last_close - stop_price) / last_close)
|
|
else:
|
|
stop_loss_pct = 1.0 # 不触发止损
|
|
else:
|
|
# 多头止损：上一个K线收盘价低于止损价格时触发止损
|
|
if last_close <= stop_price:
|
|
stop_loss_pct = -abs((stop_price - last_close) / last_close)
|
|
else:
|
|
stop_loss_pct = 1.0 # 不触发止损
|
|
|
|
# 确保止损不会比默认止损更宽松
|
|
return max(stop_loss_pct, self.stoploss)
|
|
|
|
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
|
|
time_in_force: str, current_time: datetime, entry_tag: str,
|
|
side: str, **kwargs) -> bool:
|
|
"""
|
|
确认交易进场
|
|
"""
|
|
|
|
# 获取最新数据进行最终确认
|
|
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
|
|
|
if len(dataframe) == 0:
|
|
return False
|
|
|
|
latest_candle = dataframe.iloc[-1]
|
|
|
|
# 使用重采样后的字段名
|
|
bb_upper_str = 'resample_{}_bb_upper'.format(self.get_ticker_indicator() * self.time)
|
|
bb_lower_str = 'resample_{}_bb_lower'.format(self.get_ticker_indicator() * self.time)
|
|
atr_str = 'resample_{}_atr'.format(self.get_ticker_indicator() * self.time)
|
|
|
|
# 确保技术指标有效
|
|
if (np.isnan(latest_candle[bb_upper_str]) or
|
|
np.isnan(latest_candle[bb_lower_str]) or
|
|
np.isnan(latest_candle[atr_str])):
|
|
return False
|
|
|
|
return True
|
|
|
|
def order_filled(self, pair: str, trade: Trade, order: Order, current_time: datetime,
|
|
**kwargs) -> None:
|
|
"""
|
|
当订单填充时的回调函数
|
|
在开仓时记录基于开单时ATR的止损价格
|
|
"""
|
|
|
|
# 处理开仓订单（包括做多和做空）
|
|
if (order.ft_order_side == 'buy' or order.ft_order_side == 'sell') and trade.is_open:
|
|
# 获取开仓时的数据
|
|
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
|
|
|
if len(dataframe) == 0:
|
|
return
|
|
|
|
# 获取开仓时的ATR值
|
|
atr_str = 'resample_{}_atr'.format(self.get_ticker_indicator() * self.time)
|
|
|
|
# 找到最接近开仓时间的K线
|
|
open_candle = dataframe.iloc[-1] # 使用最新的K线作为开仓时的数据
|
|
atr_value = open_candle[atr_str]
|
|
|
|
if not np.isnan(atr_value) and atr_value > 0:
|
|
# 计算止损价格并存储
|
|
atr_stop_distance = self.atr_stop_multiplier * atr_value
|
|
|
|
if trade.is_short:
|
|
# 空头止损：入场价 + ATR止损距离
|
|
stop_price = trade.open_rate + atr_stop_distance
|
|
else:
|
|
# 多头止损：入场价 - ATR止损距离
|
|
stop_price = trade.open_rate - atr_stop_distance
|
|
|
|
# 使用trade_id作为key存储止损价格
|
|
self.trade_stop_prices[str(trade.id)] = stop_price
|
|
|
|
#logger.info(f"交易 {trade.id} 开仓，记录止损价格: {stop_price}, ATR: {atr_value}, 开仓价: {trade.open_rate}")
|
|
logger.info(f"{current_time} {pair} {trade.open_rate} {stop_price} {atr_value}")
|
|
def trade_exit(self, pair: str, trade: Trade, order: Order, current_time: datetime,
|
|
**kwargs) -> None:
|
|
"""
|
|
当交易退出时的回调函数
|
|
清理存储的止损价格记录
|
|
"""
|
|
trade_id = str(trade.id)
|
|
if trade_id in self.trade_stop_prices:
|
|
del self.trade_stop_prices[trade_id]
|
|
logger.info(f"交易 {trade.id} 已关闭，清理止损价格记录")
|
|
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]) |