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

add more stuff

Browse Source
This commit is contained in:
jackyu66git
2025-07-11 01:57:21 +08:00
parent 5d58de1b6b
commit 4f98924295
5 changed files with 572 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
strategies/BB9033.py
+280
View File
@@ -0,0 +1,280 @@
# --- 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 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
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):
"""
布林带ATR反转策略
基于ATR动态调整布林带轨道，实现反转交易
交易逻辑：
- 做多：价格跌破下轨后反转
- 做空：价格突破上轨后反转
- 止盈：价格触及对侧轨道
- 止损：基于ATR动态设置
"""
INTERFACE_VERSION: int = 3
# 策略参数
bb_length = 54 # 布林带长度
atr_multiplier = 2.5 # ATR乘数(轨道)
atr_stop_multiplier = 5.6 # ATR乘数(止损)
atr_length = 17 # ATR计算周期
# Minimal ROI designed for the strategy.
# This attribute will be overridden if the config file contains "minimal_roi"
minimal_roi = {
"0": 0.5
}
# 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'
time3 = 60
# Trailing stop loss
trailing_stop = False
# 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, atr_length) + 10
# 存储每个交易的止损价格
trade_stop_prices: Dict[str, float] = {}
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
计算技术指标
"""
dataframe_3 = resample_to_interval(dataframe, self.get_ticker_indicator() * self.time3)
# 计算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
# 计算突破条件（与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.time3)
break_above_upper = 'resample_{}_break_above_upper'.format(self.get_ticker_indicator()*self.time3)
# 做多条件：价格跌破下轨
dataframe.loc[
(
(dataframe[break_below_lower] == True) & # 价格跌破下轨，明确检查True值
(dataframe[break_below_lower].notna()) # 确保不是NaN
),
'enter_long'] = 1
# 做空条件：价格突破上轨
dataframe.loc[
(
(dataframe[break_above_upper] == True) & # 价格突破上轨，明确检查True值
(dataframe[break_above_upper].notna()) # 确保不是NaN
),
'enter_short'] = 1
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.time3)
bb_lower_str = 'resample_{}_bb_lower'.format(self.get_ticker_indicator()*self.time3)
# 多头止盈：价格突破上轨（与Pine Script一致）
dataframe.loc[
(
(dataframe['close'] > dataframe[bb_upper_str]) & # 当前价格突破上轨
(dataframe[bb_upper_str].notna()) & # 确保布林带上轨不是NaN
(dataframe['close'].notna()) # 确保收盘价不是NaN
),
'exit_long'] = 1
# 空头止盈：价格跌破下轨（与Pine Script一致）
dataframe.loc[
(
(dataframe['close'] < dataframe[bb_lower_str]) & # 当前价格跌破下轨
(dataframe[bb_lower_str].notna()) & # 确保布林带下轨不是NaN
(dataframe['close'].notna()) # 确保收盘价不是NaN
),
'exit_short'] = 1
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线的收盘价
last_close = dataframe.iloc[-2]['close'] # 上一个完整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.time3)
bb_lower_str = 'resample_{}_bb_lower'.format(self.get_ticker_indicator() * self.time3)
atr_str = 'resample_{}_atr'.format(self.get_ticker_indicator() * self.time3)
# 确保技术指标有效
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.time3)
# 找到最接近开仓时间的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 get_ticker_indicator(self):
return int(self.timeframe[:-1])
strategies/ChanLun_BTC_30.py
+152 -67
View File
@@ -22,7 +22,7 @@ logger = logging.getLogger(__name__)
# freqtrade plot-dataframe --strategy ChanLun_BTC_30 --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 ChanLun_BTC_30 --strategy-path ./user_data/Chan/strategies
# freqtrade backtesting -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy ChanLun_BTC_30 --strategy-path ./user_data/Chan/strategies --timerange=20250520-
# freqtrade backtesting -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy ChanLun_BTC_30 --strategy-path ./user_data/Chan/strategies --timerange=20250510-20250520
# freqtrade download-data -c ./user_data/Chan/config/ChanLun_BTC_30.json -t 1m --pairs BTC/USDT:USDT --timerange=20250405-
# freqtrade hyperopt --hyperopt-loss SharpeHyperOptLossDaily --spaces roi stoploss --strategy ChanLun_BTC_30 --strategy-path ./user_data/Chan/strategies -c ./user_data/Chan/config/ChanLun_BTC_30.json -e 200 --timerange=20250201-20250401
@@ -37,20 +37,20 @@ class ChanLun_BTC_30(IStrategy):
# 30m and 1h
minimal_roi = {
"0": 0.60,
"0": 0.15,
"360": 0.2,
"640": 0.1,
"1200": 0
}
# 5m and 15m
minimal_roi = {
minimal_roi_1 = {
"0": 0.1,
"60": 0.05,
"120": 0.02,
"240": 0
}
# 15m and 30m
minimal_roi = {
minimal_roi_1 = {
"0": 0.1,
"240": 0.05,
"480": 0.03,
@@ -64,15 +64,16 @@ class ChanLun_BTC_30(IStrategy):
}
can_short = True
lev = 1.0
stoploss = -0.01
#use_custom_stoploss = True
stoploss = -0.2 # 设置为很大的负值，让custom_stoploss来控制
use_custom_stoploss = False # 启用自定义止损
trailing_stop = False
trailing_stop_positive = 0.025
trailing_stop_positive_offset = 0.045
trailing_only_offset_is_reached = False
#position_adjustment_enable = True
# 启用仓位调整功能以支持分批止盈
position_adjustment_enable = True
startup_candle_count = 780
time5 = 5
@@ -80,16 +81,18 @@ class ChanLun_BTC_30(IStrategy):
time30 = 30
time60 = 60
time4h = 240
time30 = 60
time30 = 3
last_time = datetime.now()
chan = ChanLun()
chanpy = ChanPY()
classifier = ChanLunClassifier(None)
last_order = None
last_trade = None
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# resample our dataframes
dataframe_3 = resample_to_interval(dataframe, self.get_ticker_indicator() * 3)
dataframe_5 = resample_to_interval(dataframe, self.get_ticker_indicator() * 5)
dataframe_15 = resample_to_interval(dataframe, self.get_ticker_indicator() * 15)
dataframe_30 = resample_to_interval(dataframe, self.get_ticker_indicator() * 30)
@@ -104,6 +107,7 @@ class ChanLun_BTC_30(IStrategy):
#dataframe_1w = resample_to_interval(dataframe, self.get_ticker_indicator() * 10080)
#dataframe_1m = resample_to_interval(dataframe, self.get_ticker_indicator() * 43200)
dataframe = self.add_indicators(dataframe)
dataframe_3 = self.add_indicators(dataframe_3)
dataframe_5 = self.add_indicators(dataframe_5)
dataframe_15 = self.add_indicators(dataframe_15)
dataframe_30 = self.add_indicators(dataframe_30)
@@ -113,9 +117,9 @@ class ChanLun_BTC_30(IStrategy):
#self.chan.plot_dual(dataframe_5, dataframe_30)
chanpy_state = self.chanpy.get_bsp_state(dataframe_5)
dataframe_5['chanpy_state'] = chanpy_state
state_list = self.chan.get_klc_state_list(dataframe_60)
dataframe_60['state'] = state_list
dataframe_60['fx'] = state_list
state_list = self.chan.get_klc_state_list(dataframe_3)
dataframe_3['state'] = state_list
dataframe_3['fx'] = state_list
#bi_list_1 = self.chan.get_bi_list(dataframe)
#bi_list_5 = self.chan.get_bi_list(dataframe_5)
#bi_list_15 = self.chan.get_bi_list(dataframe_15)
@@ -130,6 +134,7 @@ class ChanLun_BTC_30(IStrategy):
self.print_seg(dataframe_5)
print("-------------------------------------------------------------------------------")
self.last_time = datetime.now()
dataframe = resampled_merge(dataframe, dataframe_3)
dataframe = resampled_merge(dataframe, dataframe_5)
#dataframe = resampled_merge(dataframe, dataframe_15)
#dataframe = resampled_merge(dataframe, dataframe_30)
@@ -157,12 +162,17 @@ class ChanLun_BTC_30(IStrategy):
bb120 = ta.BBANDS(df, timeperiod=120, nbdevup=3.0, nbdevdn=3.0, matype=0)
bb30 = ta.BBANDS(df, timeperiod=90, nbdevup=3.0, nbdevdn=3.0, matype=0)
bb302 = ta.BBANDS(df, timeperiod=90, nbdevup=2.0, nbdevdn=2.0, matype=0)
# 计算布林带中轨（移动平均线）
bb30_middle = ta.SMA(df, timeperiod=90)
# 手动计算布林带 %B 指标 (BBP)
# %B = (Price - Lower Band) / (Upper Band - Lower Band)
bbp365 = (df['close'] - bb365['lowerband']) / (bb365['upperband'] - bb365['lowerband'])
bbp120 = (df['close'] - bb120['lowerband']) / (bb120['upperband'] - bb120['lowerband'])
bbp30 = (df['close'] - bb30['lowerband']) / (bb30['upperband'] - bb30['lowerband'])
bbp302 = (df['close'] - bb302['lowerband']) / (bb302['upperband'] - bb302['lowerband'])
df['atr'] = ta.ATR(df, timeperiod=14)
df['bbup365'] = bb365['upperband']
df['bblow365'] = bb365['lowerband']
df['bbp365'] = bbp365
@@ -171,6 +181,7 @@ class ChanLun_BTC_30(IStrategy):
df['bbp120'] = bbp120
df['bbup30'] = bb30['upperband']
df['bblow30'] = bb30['lowerband']
df['bbmiddle30'] = bb30_middle # 添加bb30中轨
df['bbp30'] = bbp30
df['bbup302'] = bb302['upperband']
df['bblow302'] = bb302['lowerband']
@@ -215,6 +226,129 @@ class ChanLun_BTC_30(IStrategy):
new_exitprice = proposed_rate - 50
return new_exitprice
def adjust_trade_position(self, trade: Trade, current_time: datetime,
current_rate: float, current_profit: float,
min_stake: Optional[float], max_stake: float,
current_entry_rate: float, current_exit_rate: float,
current_entry_profit: float, current_exit_profit: float,
**kwargs) -> Optional[float]:
"""
基于布林带的分批止盈逻辑
"""
# 获取当前数据
dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
if dataframe is None or len(dataframe) == 0:
return None
last_candle = dataframe.iloc[-1]
# 获取布林带数据
bb30_middle = last_candle['bbmiddle30']
bb30_upper = last_candle['bbup30']
bb30_lower = last_candle['bblow30']
bb302_upper = last_candle['bbup302']
bb302_lower = last_candle['bblow302']
# 获取交易的状态标记
first_tp_triggered = trade.get_custom_data(key="first_tp_triggered", default=False)
second_tp_triggered = trade.get_custom_data(key="second_tp_triggered", default=False)
if trade.is_short:
# 做空逻辑
if not first_tp_triggered and current_rate <= bb30_middle:
# 第一次止盈：价格跌到bb30中轨，止盈50%
logger.info(f"做空第一次止盈触发：价格{current_rate} <= BB30中轨{bb30_middle}")
trade.set_custom_data(key="first_tp_triggered", value=True)
trade.set_custom_data(key="new_stoploss", value=trade.open_rate) # 设置止损为开仓价
return -(trade.amount * 0.5) # 减少50%仓位
elif first_tp_triggered and not second_tp_triggered and current_rate <= bb302_lower:
# 第二次止盈：继续跌到bb302下轨，止盈剩余仓位的60%
logger.info(f"做空第二次止盈触发：价格{current_rate} <= BB302下轨{bb302_lower}")
trade.set_custom_data(key="second_tp_triggered", value=True)
trade.set_custom_data(key="new_stoploss", value=bb30_middle) # 移动止损到bb30中轨
remaining_amount = trade.amount * 0.5 # 剩余50%
return -(remaining_amount * 0.6) # 减少剩余仓位的60%
else:
# 做多逻辑
if not first_tp_triggered and current_rate >= bb30_middle:
# 第一次止盈：价格涨到bb30中轨，止盈50%
logger.info(f"做多第一次止盈触发：价格{current_rate} >= BB30中轨{bb30_middle}")
trade.set_custom_data(key="first_tp_triggered", value=True)
trade.set_custom_data(key="new_stoploss", value=trade.open_rate) # 设置止损为开仓价
return -(trade.amount * 0.5) # 减少50%仓位
elif first_tp_triggered and not second_tp_triggered and current_rate >= bb302_upper:
# 第二次止盈：继续涨到bb302上轨，止盈剩余仓位的60%
logger.info(f"做多第二次止盈触发：价格{current_rate} >= BB302上轨{bb302_upper}")
trade.set_custom_data(key="second_tp_triggered", value=True)
trade.set_custom_data(key="new_stoploss", value=bb30_middle) # 移动止损到bb30中轨
remaining_amount = trade.amount * 0.5 # 剩余50%
return -(remaining_amount * 0.6) # 减少剩余仓位的60%
return None
def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime,
current_rate: float, current_profit: float, after_fill: bool,
**kwargs) -> float | None:
"""
动态止损逻辑
"""
# 检查是否有自定义的新止损价格（分批止盈后的动态止损）
new_stoploss_price = trade.get_custom_data(key="new_stoploss")
if new_stoploss_price:
logger.info(f"使用动态止损价格: {new_stoploss_price}")
return stoploss_from_absolute(new_stoploss_price, current_rate, is_short=trade.is_short)
# 如果没有ATR数据，使用固定的5%止损作为备用
logger.warning(f"未找到开仓时ATR数据，使用默认5%止损")
return -0.05
def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float,
current_profit: float, **kwargs):
"""
自定义退出逻辑 - 处理最终止盈条件
"""
# 获取当前数据
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if dataframe is None or len(dataframe) == 0:
return None
last_candle = dataframe.iloc[-1]
# 获取布林带数据
bb30_upper = last_candle['bbup30']
bb30_lower = last_candle['bblow30']
# 检查是否已经触发过前两次止盈
first_tp_triggered = trade.get_custom_data(key="first_tp_triggered", default=False)
second_tp_triggered = trade.get_custom_data(key="second_tp_triggered", default=False)
if trade.is_short:
# 做空：如果价格跌到bb30下轨，全部止盈
if first_tp_triggered and second_tp_triggered and current_rate <= bb30_lower:
logger.info(f"做空最终止盈触发：价格{current_rate} <= BB30下轨{bb30_lower}")
return "short_final_tp_bb30_lower"
else:
# 做多：如果价格涨到bb30上轨，全部止盈
if first_tp_triggered and second_tp_triggered and current_rate >= bb30_upper:
logger.info(f"做多最终止盈触发：价格{current_rate} >= BB30上轨{bb30_upper}")
return "long_final_tp_bb30_upper"
# 原有退出逻辑
if trade.is_short:
last_high = trade.get_custom_data(key="entry_candle_high")
if last_high and current_rate > last_high:
return "Relay Top FX exit"
else:
last_low = trade.get_custom_data(key="entry_candle_low")
if last_low and current_rate < last_low:
return "Relay Bottom FX exit"
return None
def confirm_trade_entry1(self, pair: str, order_type: str, amount: float, rate: float,
time_in_force: str, current_time: datetime, entry_tag: str | None,
side: str, **kwargs) -> bool:
@@ -248,36 +382,8 @@ class ChanLun_BTC_30(IStrategy):
return stoploss_from_absolute(last_low, current_rate, is_short=trade.is_short)
# return maximum stoploss value, keeping current stoploss price unchanged
return None
def custom_exit1(self, pair: str, trade: Trade, current_time: datetime, current_rate: float,
current_profit: float, **kwargs):
#dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
#last_candle = dataframe.iloc[-1].squeeze()
"""
# Above 20% profit, sell when rsi < 80
if current_profit > 0.2:
if last_candle["rsi"] < 80:
return "rsi_below_80"
# Between 2% and 10%, sell if EMA-long above EMA-short
if 0.02 < current_profit < 0.1:
if last_candle["emalong"] > last_candle["emashort"]:
return "ema_long_below_80"
# Sell any positions at a loss if they are held for more than one day.
if current_profit < 0.0 and (current_time - trade.open_date_utc).days >= 1:
return "unclog"
"""
if trade.is_short:
last_high = trade.get_custom_data(key="entry_candle_high")
if last_high and current_rate > last_high:
#print(trade.open_date, last_high, current_rate, "Relay Top FX exit")
return "Relay Top FX exit"
else:
last_low = trade.get_custom_data(key="entry_candle_low")
if last_low and current_rate < last_low:
#print(trade.open_date, last_low, current_rate, "Relay Bottom FX exit")
return "Relay Bottom FX exit"
def order_filled1(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> None:
def order_filled(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> None:
"""
Called right after an order fills.
Will be called for all order types (entry, exit, stoploss, position adjustment).
@@ -290,33 +396,12 @@ class ChanLun_BTC_30(IStrategy):
# Obtain pair dataframe (just to show how to access it)
dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
last_candle = dataframe.iloc[-1].squeeze()
ema5 = 'resample_{}_ema5'.format(self.get_ticker_indicator()*self.time30)
ema10 = 'resample_{}_ema10'.format(self.get_ticker_indicator()*self.time30)
ema26 = 'resample_{}_ema26'.format(self.get_ticker_indicator()*self.time30)
ema52 = 'resample_{}_ema52'.format(self.get_ticker_indicator()*self.time30)
#print(last_candle[ema5], last_candle[ema10], last_candle[ema26], last_candle[ema52])
#print(last_candle['close'])
klc_list = self.chan.get_klc_list(resample_to_interval(dataframe, self.get_ticker_indicator() * self.time30))
bi_list = self.chan.cal_bi_list(klc_list)
if self.last_order is None:
if trade.is_short and klc_list[-2].last_top_klc:
if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side):
last_high = klc_list[-2].last_top_klc.high
print(klc_list[-2].start_time, "--------------------------------", order.order_date, order.side, last_high)
trade.set_custom_data(key="entry_candle_high", value=last_high)
else:
if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side) and klc_list[-2].last_bottom_klc:
last_low = klc_list[-2].last_bottom_klc.low
trade.set_custom_data(key="entry_candle_low", value=last_low)
print(klc_list[-2].start_time, "--------------------------------", order.order_date, order.side, last_low)
#print(trade.open_date, trade.close_date, last_high, last_low, order.ft_order_side, klc_list[-2].start_time, klc_list[-2].end_time)
self.last_order = order
else:
if self.last_order.side != order.side:
self.last_order = None
trade.set_custom_data(key="entry_candle_high", value=None)
trade.set_custom_data(key="entry_candle_low", value=None)
self.last_trade = trade
# 保存开仓时的ATR值用于止损计算
if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side):
entry_atr = last_candle['atr']
trade.set_custom_data(key="entry_atr", value=entry_atr)
logger.info(f"保存开仓时ATR值: {entry_atr}")
return None
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
state_str = 'resample_{}_state'.format(self.get_ticker_indicator()*self.time30)
strategies/bollinger_atr_strategy.pine
+136
View File
@@ -0,0 +1,136 @@
//@version=6
strategy("布林带ATR反转策略", shorttitle="BBB_ATR", overlay=true, default_qty_type=strategy.percent_of_equity, default_qty_value=100, calc_on_every_tick=true)
// 输入参数
bb_length = input.int(90, "布林带长度", minval=10, maxval=200)
atr_multiplier = input.float(3.0, "ATR乘数(轨道)", minval=1.0, maxval=10.0, step=0.1)
atr_stop_multiplier = input.float(3.0, "ATR乘数(止损)", minval=0.5, maxval=5.0, step=0.1)
atr_length = input.int(9, "ATR计算周期", minval=5, maxval=50)
// 显示设置
show_bands = input.bool(true, "显示布林带")
show_signals = input.bool(true, "显示信号")
// 计算移动平均线(中线)
bb_middle = ta.sma(close, bb_length)
// 计算ATR
atr_value = ta.atr(atr_length)
// 计算上下轨
bb_upper = bb_middle + (atr_multiplier * atr_value)
bb_lower = bb_middle - (atr_multiplier * atr_value)
// 显示布林带
plot(show_bands ? bb_middle : na, "中线", color=color.blue, linewidth=2)
plot(show_bands ? bb_upper : na, "上轨", color=color.red, linewidth=2)
plot(show_bands ? bb_lower : na, "下轨", color=color.green, linewidth=2)
// 交易条件
// 做空条件：价格突破上轨
short_condition = close > bb_upper and close[1] <= bb_upper[1]
// 做多条件：价格跌破下轨
long_condition = close < bb_lower and close[1] >= bb_lower[1]
// 做空止盈条件：价格跌破下轨
short_take_profit = close < bb_lower
// 做多止盈条件：价格突破上轨
long_take_profit = close > bb_upper
// 记录入场价格和止损位
var float long_entry_price = na
var float short_entry_price = na
var float long_stop_loss = na
var float short_stop_loss = na
// 执行交易逻辑
if strategy.position_size == 0
if long_condition
strategy.entry("做多", strategy.long)
long_entry_price := close
long_stop_loss := close - (atr_stop_multiplier * atr_value)
if short_condition
strategy.entry("做空", strategy.short)
short_entry_price := close
short_stop_loss := close + (atr_stop_multiplier * atr_value)
// 多头仓位管理
if strategy.position_size > 0
// 止盈：价格突破上轨
if long_take_profit
strategy.close("做多", comment="多头止盈")
long_entry_price := na
long_stop_loss := na
// 止损：价格跌破止损位
else if close <= long_stop_loss
strategy.close("做多", comment="多头止损")
long_entry_price := na
long_stop_loss := na
// 空头仓位管理
if strategy.position_size < 0
// 止盈：价格跌破下轨
if short_take_profit
strategy.close("做空", comment="空头止盈")
short_entry_price := na
short_stop_loss := na
// 止损：价格突破止损位
else if close >= short_stop_loss
strategy.close("做空", comment="空头止损")
short_entry_price := na
short_stop_loss := na
// 显示信号
if show_signals
if long_condition and strategy.position_size == 0
label.new(bar_index, low, "做多", color=color.green, style=label.style_label_up, size=size.normal, textcolor=color.white)
if short_condition and strategy.position_size == 0
label.new(bar_index, high, "做空", color=color.red, style=label.style_label_down, size=size.normal, textcolor=color.white)
if long_take_profit and strategy.position_size > 0
label.new(bar_index, high, "多头止盈", color=color.green, style=label.style_label_down, size=size.small, textcolor=color.white)
if short_take_profit and strategy.position_size < 0
label.new(bar_index, low, "空头止盈", color=color.red, style=label.style_label_up, size=size.small, textcolor=color.white)
// 显示止损线
plot(strategy.position_size > 0 and not na(long_stop_loss) ? long_stop_loss : na, "多头止损", color=color.red, style=plot.style_linebr, linewidth=1)
plot(strategy.position_size < 0 and not na(short_stop_loss) ? short_stop_loss : na, "空头止损", color=color.red, style=plot.style_linebr, linewidth=1)
// 信息表格
if barstate.islast
var table info_table = table.new(position.top_right, 2, 10, bgcolor=color.white, border_width=1)
table.cell(info_table, 0, 0, "布林带ATR反转策略", text_color=color.black, bgcolor=color.gray)
table.cell(info_table, 1, 0, "", text_color=color.black, bgcolor=color.gray)
table.cell(info_table, 0, 1, "布林带长度", text_color=color.black)
table.cell(info_table, 1, 1, str.tostring(bb_length), text_color=color.black)
table.cell(info_table, 0, 2, "ATR轨道乘数", text_color=color.black)
table.cell(info_table, 1, 2, str.tostring(atr_multiplier), text_color=color.black)
table.cell(info_table, 0, 3, "ATR止损乘数", text_color=color.black)
table.cell(info_table, 1, 3, str.tostring(atr_stop_multiplier), text_color=color.black)
table.cell(info_table, 0, 4, "当前ATR", text_color=color.black)
table.cell(info_table, 1, 4, str.tostring(math.round(atr_value, 4)), text_color=color.black)
table.cell(info_table, 0, 5, "上轨价位", text_color=color.black)
table.cell(info_table, 1, 5, str.tostring(math.round(bb_upper, 2)), text_color=color.black)
table.cell(info_table, 0, 6, "下轨价位", text_color=color.black)
table.cell(info_table, 1, 6, str.tostring(math.round(bb_lower, 2)), text_color=color.black)
table.cell(info_table, 0, 7, "当前价格", text_color=color.black)
table.cell(info_table, 1, 7, str.tostring(math.round(close, 2)), text_color=color.black)
table.cell(info_table, 0, 8, "仓位状态", text_color=color.black)
position_text = strategy.position_size > 0 ? "多头" : strategy.position_size < 0 ? "空头" : "空仓"
table.cell(info_table, 1, 8, position_text, text_color=color.black)
table.cell(info_table, 0, 9, "价格位置", text_color=color.black)
price_position = close > bb_upper ? "上轨之上" : close < bb_lower ? "下轨之下" : "轨道之间"
table.cell(info_table, 1, 9, price_position, text_color=color.black)