jack/Chan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add more strategies

Browse Source
This commit is contained in:
jackyu66git
2025-07-21 20:31:25 +08:00
parent d75c202975
commit 6907ce7d8c
6 changed files with 379 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
strategies/BB9033.py
+58 -41
View File
@@ -5,8 +5,12 @@ 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
@@ -28,38 +32,37 @@ logger = logging.getLogger(__name__)
class BB9033(IStrategy):
"""
布林带ATR反转策略
基于ATR动态调整布林带轨道，实现反转交易
布林带反转策略（与Pine Script保持一致）
基于EMA和标准差计算布林带，实现反转交易
交易逻辑：
- 做多：价格跌破下轨后反转
- 做空：价格突破上轨后反转
- 止盈：价格触及对侧轨道
- 做多止盈：价格减去0.5倍ATR突破上轨
- 做空止盈：价格跌破下轨
- 止损：基于ATR动态设置
"""
INTERFACE_VERSION: int = 3
# 策略参数
bb_length = 90 # 布林带长度
atr_multiplier = 4.2 # ATR乘数(轨道)
atr_stop_multiplier = 1.8 # ATR乘数(止损)
atr_length = 14 # ATR计算周期
# 策略参数（与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 = {
"0": 0.5
}
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
timeframe = '3m'
time = 30
time = 5
# Trailing stop loss
trailing_stop = False
lev = 1.0
@@ -67,26 +70,27 @@ class BB9033(IStrategy):
process_only_new_candles = False
# Number of candles the strategy requires before producing valid signals
startup_candle_count: int = max(bb_length, atr_length) + 10
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)
# 计算布林带（使用标准方法：移动平均线 ± 标准差倍数）
bb_upper, bb_middle, bb_lower = ta.BBANDS(dataframe_3['close'], timeperiod=self.bb_length, nbdevup=self.atr_multiplier, nbdevdn=self.atr_multiplier, matype=0)
dataframe_3['bb_upper'] = bb_upper
dataframe_3['bb_lower'] = bb_lower
for i in range(1700, 1800):
print(dataframe_3.iloc[i])
# 计算布林带（使用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 = (
@@ -98,16 +102,20 @@ class BB9033(IStrategy):
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
@@ -118,23 +126,25 @@ class BB9033(IStrategy):
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) & # 价格跌破下轨，明确检查True值
(dataframe[break_below_lower].notna()) # 确保不是NaN
(dataframe[break_below_lower] == True) &
(pd.notna(dataframe[break_below_lower]))
),
'enter_long'] = 1
['enter_long', 'enter_tag']] = (1, 'long_signal_chan')
# 做空条件：价格突破上轨
# 检测空头信号：价格突破上轨
dataframe.loc[
(
(dataframe[break_above_upper] == True) & # 价格突破上轨，明确检查True值
(dataframe[break_above_upper].notna()) # 确保不是NaN
(dataframe[break_above_upper] == True) &
(pd.notna(dataframe[break_above_upper]))
),
'enter_short'] = 1
['enter_short', 'enter_tag']] = (1, 'short_signal_chan')
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
@@ -142,24 +152,30 @@ class BB9033(IStrategy):
"""
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一致）
# 做多止盈条件：价格突破上轨（与Pine Script保持一致）
dataframe.loc[
(
(dataframe['close'] > dataframe[bb_upper_str]) & # 当前价格突破上轨
(dataframe[bb_upper_str].notna()) & # 确保布林带上轨不是NaN
(dataframe['close'].notna()) # 确保收盘价不是NaN
(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'] = 1
['exit_long', 'exit_tag']] = (1, 'long_close_signal_chan')
# 空头止盈：价格跌破下轨（与Pine Script一致）
# 做空止盈条件：价格跌破下轨（与Pine Script保持一致）
dataframe.loc[
(
(dataframe['close'] < dataframe[bb_lower_str]) & # 当前价格跌破下轨
(dataframe[bb_lower_str].notna()) & # 确保布林带下轨不是NaN
(dataframe['close'].notna()) # 确保收盘价不是NaN
(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'] = 1
['exit_short', 'exit_tag']] = (1, 'short_close_signal_chan')
return dataframe
@@ -180,8 +196,9 @@ class BB9033(IStrategy):
if len(dataframe) < 2:
return self.stoploss
# 使用上一个K线的收盘价
last_close = dataframe.iloc[-2]['close'] # 上一个完整K线的收盘价
# 使用上一个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]
@@ -266,7 +283,7 @@ class BB9033(IStrategy):
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}")
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:
"""