核心代码
加载数据
def load_csv_data(data_path, size=None, start=None, end=None):
return bt.feeds.GenericCSVData(
dataname=data_path,
nullvalue=0.0,
fromdate=start,
todate=end,
dtformat="%Y-%m-%d %H:%M:%S",
timeframe=bt.TimeFrame.Minutes,
datetime=0,
high=1,
low=2,
open=3,
close=4,
volume=5,
openinterest=-1
)
def create_cerebro(cash=10000.0, commission=0.01, stake=1, strategy=None):
"""
:param data: 数据
:param cash: 初始资金
:param commission: 佣金率
:param stake: 交易单位大小
:param strategy: 交易策略
:return:
"""
cerebro = bt.Cerebro()
# 设置启动资金
cerebro.broker.setcash(cash)
# 设置交易单位大小
cerebro.addsizer(bt.sizers.FixedSize, stake=stake)
# 设置佣金率为千分之一
cerebro.broker.setcommission(commission)
# 显示回测过程中的买入和卖出信号
cerebro.addobserver(bt.observers.Value)
# 显示了回测过程中的买入和卖出信号
cerebro.addobserver(bt.observers.BuySell)
return cerebro
class TurtleTradingStrategy(bt.Strategy):
params = dict(
N1=40, # 唐奇安通道上轨的t
N2=30, # 唐奇安通道下轨的t
ATR_T=40, # ATR的周期T
printlog=False,
)
def log(self, txt, dt=None, doprint=False):
if self.params.printlog or doprint:
dt = dt or self.datas[0].datetime.date(0)
print(f"{dt.isoformat()},{txt}")
def __init__(self):
self.order = None
self.buy_count = 0 # 记录买入次数
self.last_price = 0 # 记录买入价格
# 第一个标的沪深300主力合约的close、high、low 行情数据
self.close = self.datas[0].close
self.high = self.datas[0].high
self.low = self.datas[0].low
# 计算唐奇安通道上轨：过去最高价
self.DonchianH = bt.ind.Highest(self.high(-1), period=self.p.N1, subplot=True)
# 计算唐奇安通道下轨：过去最低价
self.DonchianL = bt.ind.Lowest(self.low(-1), period=self.p.N2, subplot=True)
# 生成唐奇安通道上轨突破：close>DonchianH，取值为1.0；反之为 -1.0
self.CrossoverH = bt.ind.CrossOver(self.close(0), self.DonchianH, subplot=False)
# 生成唐奇安通道下轨突破:
self.CrossoverL = bt.ind.CrossOver(self.close(0), self.DonchianL, subplot=False)
# ATR
self.ATR = bt.talib.ATR(self.high, self.low, self.close, timeperiod=self.p.ATR_T, subplot=True)
def next(self):
# 如果还有订单在执行中，就不做新的仓位调整
if self.order:
return
# 如果当前持有多单
if self.position.size > 0:
# 多单加仓:价格上涨了买入价的0.5的ATR且加仓次数少于等于3次
if self.datas[0].close > self.last_price + 0.5 * self.ATR[0] and self.buy_count <= 4:
# print("if self.datas[0].close >self.last_price + 0.5*self.ATR[0] and self.buy_count <= 4:")
# print("self.buy_count",self.buy_count)
# 计算建仓单位：self.ATR*期货合约乘数300*保证金比例0.1
self.buy_unit = max((self.broker.getvalue() * 0.005) / (self.ATR * 300 * 0.1), 1)
self.buy_unit = int(self.buy_unit) # 交易单位为手
# self.sizer.p.stake = self.buy_unit
self.order = self.buy(size=self.buy_unit)
self.last_price = self.position.price # 获取买入价格
self.buy_count = self.buy_count + 1
# 多单止损：当价格回落2倍ATR时止损平仓
elif self.datas[0].close < (self.last_price - 2 * self.ATR[0]):
# print("elif self.datas[0].close < (self.last_price - 2*self.ATR[0]):")
self.order = self.sell(size=abs(self.position.size))
self.buy_count = 0
# 多单止盈：当价格突破10日最低点时止盈离场 平仓
elif self.CrossoverL < 0:
# print("self.CrossoverL < 0")
self.order = self.sell(size=abs(self.position.size))
self.buy_count = 0
# 如果当前持有空单
elif self.position.size < 0:
# 空单加仓:价格小于买入价的0.5的ATR且加仓次数少于等于3次
if self.datas[0].close < self.last_price - 0.5 * self.ATR[0] and self.buy_count <= 4:
# print("self.datas[0].close<self.last_price-0.5*self.ATR[0] and self.buy_count <= 4")
# 计算建仓单位：self.ATR*期货合约乘数300*保证金比例0.1
self.buy_unit = max((self.broker.getvalue() * 0.005) / (self.ATR * 300 * 0.1), 1)
self.buy_unit = int(self.buy_unit) # 交易单位为手
# self.sizer.p.stake = self.buy_unit
self.order = self.sell(size=self.buy_unit)
self.last_price = self.position.price # 获取买入价格
self.buy_count = self.buy_count + 1
# 空单止损：当价格上涨至2倍ATR时止损平仓
elif self.datas[0].close < (self.last_price + 2 * self.ATR[0]):
# print("self.datas[0].close < (self.last_price+2*self.ATR[0])")
self.order = self.buy(size=abs(self.position.size))
self.buy_count = 0
# 多单止盈：当价格突破20日最高点时止盈平仓
elif self.CrossoverH > 0:
# print("self.CrossoverH>0")
self.order = self.buy(size=abs(self.position.size))
self.buy_count = 0
else: # 如果没有持仓，等待入场时机
# 入场: 价格突破上轨线且空仓时，做多
if self.CrossoverH > 0 and self.buy_count == 0:
# print("if self.CrossoverH > 0 and self.buy_count == 0:")
# 计算建仓单位：self.ATR*期货合约乘数300*保证金比例0.1
self.buy_unit = max((self.broker.getvalue() * 0.005) / (self.ATR * 300 * 0.1), 1)
self.buy_unit = int(self.buy_unit) # 交易单位为手
self.order = self.buy(size=self.buy_unit)
self.last_price = self.position.price # 记录买入价格
self.buy_count = 1 # 记录本次交易价格
# 入场: 价格跌破下轨线且空仓时，做空
elif self.CrossoverL < 0 and self.buy_count == 0:
# print("self.CrossoverL < 0 and self.buy_count == 0")
# 计算建仓单位：self.ATR*期货合约乘数300*保证金比例0.1
self.buy_unit = max((self.broker.getvalue() * 0.005) / (self.ATR * 300 * 0.1), 1)
self.buy_unit = int(self.buy_unit) # 交易单位为手
self.order = self.sell(size=self.buy_unit)
self.last_price = self.position.price # 记录买入价格
self.buy_count = 1 # 记录本次交易价格
# 打印订单日志
def notify_order(self, order):
order_status = ["Created", "Submitted", "Accepted", "Partial",
"Completed", "Canceled", "Expired", "Margin", "Rejected"]
# 未被处理的订单
if order.status in [order.Submitted, order.Accepted]:
self.log("ref:%.0f, name: %s, Order: %s" % (order.ref,
order.data._name,
order_status[order.status]))
return
# 已经处理的订单
if order.status in [order.Partial, order.Completed]:
if order.isbuy():
self.log(
"BUY EXECUTED, status: %s, ref:%.0f, name: %s, Size: %.2f, Price: %.2f, Cost: %.2f, Comm %.2f" %
(order_status[order.status], # 订单状态
order.ref, # 订单编号
order.data._name, # 股票名称
order.executed.size, # 成交量
order.executed.price, # 成交价
order.executed.value, # 成交额
order.executed.comm)) # 佣金
else: # Sell
self.log(
"SELL EXECUTED, status: %s, ref:%.0f, name: %s, Size: %.2f, Price: %.2f, Cost: %.2f, Comm %.2f" %
(order_status[order.status],
order.ref,
order.data._name,
order.executed.size,
order.executed.price,
order.executed.value,
order.executed.comm))
elif order.status in [order.Canceled, order.Margin, order.Rejected, order.Expired]:
# 订单未完成
self.log("ref:%.0f, name: %s, status: %s" % (
order.ref, order.data._name, order_status[order.status]))
self.order = None
def notify_trade(self, trade):
# 交易刚打开时
if trade.justopened:
self.log("Trade Opened, name: %s, Size: %.2f,Price: %.2f" % (
trade.getdataname(), trade.size, trade.price))
# 交易结束
elif trade.isclosed:
self.log("Trade Closed, name: %s, GROSS %.2f, NET %.2f, Comm %.2f" % (
trade.getdataname(), trade.pnl, trade.pnlcomm, trade.commission))
# 更新交易状态
else:
self.log("Trade Updated, name: %s, Size: %.2f,Price: %.2f" % (
trade.getdataname(), trade.size, trade.price))
def stop(self):
self.log(f"(组合线:{self.p.N1},{self.p.N2},{self.p.ATR_T})
期末总资金: {self.broker.getvalue():.2f}", doprint=True)
入场：价格突破20日价格高点时，入场；加仓：价格继续上涨至0.5倍ATR，再次加仓，加仓次数不超过3次；止损：价格回落2倍ATR时止损离场；止盈：价格突破10日最低点时止盈离场；做空与做多的逻辑相反。
目前从回测结果来看，该策略明显没跑赢大盘，下篇文章优化该策略，使得策略可以获得更高的收益
交易信号
回测结果
| 初始资金 | 10000 |
| 期货品种 | ETH |
| 时间级别 | 1H |
| 回测时间 | 2017.7.15 - 2022.7.24 |
| 倍数 | 1 |
| 手续费 | 1% |
| 总盈利 | 993 |
唐奇安通道+ATR
文章为作者独立观点，不代表股票交易接口观点
相关文章
股民评论