backtrader_单标的_海龟交易年化112最大回撤192

一.效果

二.具体实现

1.实现海龟交易类

class TurtleStrategy(bt.Strategy):
    # 默认参数
    params = (
        ("long_period", 25),
        ("short_period", 15),
        ("printlog", False),
    )
    def __init__(self):
        # 初始化订单信息
        self.order = None
        # 买入价格
        self.buyprice = 0
        # 买入手续费
        self.buycomm = 0
        # 买入数量
        self.buy_size = 0
        # 买入次数
        self.buy_count = 0
        # 计算唐奇安通道的上轨线
        # 海龟交易法则中的唐奇安通道和平均波幅ATR
        self.H_line = bt.indicators.Highest(
            self.data.high(-1), period=self.p.long_period
        )
        # 计算唐奇安通道的下轨线
        self.L_line = bt.indicators.Lowest(
            self.data.low(-1), period=self.p.short_period
        )
        # 计算真实波幅TR
        self.TR = bt.indicators.Max(
            (self.data.high(0) - self.data.low(0)),
            abs(self.data.close(-1) - self.data.high(0)),
            abs(self.data.close(-1) - self.data.low(0)),
        )
        # 计算平均波幅ATR
        self.ATR = bt.indicators.SimpleMovingAverage(self.TR, period=14)
        # 判断价格是否上穿上轨线，生成买入信号
        # 价格与上下轨线的交叉
        self.buy_signal = bt.ind.CrossOver(self.data.close(0), self.H_line)
        # 判断价格是否下穿下轨线，生成卖出信号
        self.sell_signal = bt.ind.CrossOver(self.data.close(0), self.L_line)
    def next(self):
        if self.order:
            return
        # 判断是否满足入场条件：价格突破上轨线且空仓
        # 入场：价格突破上轨线且空仓时
        if self.buy_signal > 0 and self.buy_count == 0:
            # 计算买入数量
            self.buy_size = self.broker.getvalue() * 0.01 / self.ATR
            self.buy_size = int(self.buy_size / 100) * 100
            # 设置买入数量
            self.sizer.p.stake = self.buy_size
            # 买入计数加1
            self.buy_count = 1
            # 输出空仓时买入的信息
            print("空仓时买入", self.buy_size)
            # 执行买入操作，并保存订单
            self.order = self.buy()
        # 判断是否满足加仓条件：价格上涨了买入价的0.5的ATR且加仓次数少于3次（含）
        # 加仓：价格上涨了买入价的0.5的ATR且加仓次数少于3次（含）
        elif (
            self.data.close > self.buyprice + 0.5 * self.ATR[0]
            and self.buy_count > 0
            and self.buy_count <= 4
        ):
            # 输出加仓买入的信息
            print("加仓买入")
            # 计算加仓数量
            self.buy_size = self.broker.getvalue() * 0.01 / self.ATR
            self.buy_size = int(self.buy_size / 100) * 100
            # 设置加仓数量
            self.sizer.p.stake = self.buy_size
            # 执行加仓操作，并保存订单
            self.order = self.buy()
            # 加仓计数加1
            self.buy_count += 1
        # 判断是否满足离场条件：价格跌破下轨线且持仓
        # 离场：价格跌破下轨线且持仓时
        elif self.sell_signal < 0 and self.buy_count > 0:
            # 输出平仓信号卖出的信息
            print("平仓信号卖出")
            # 执行卖出操作，并保存订单
            self.order = self.sell()
            # 清空买入计数
            self.buy_count = 0
        # 判断是否满足止损条件：价格跌破买入价的2个ATR且持仓
        # 止损：价格跌破买入价的2个ATR且持仓时
        elif self.data.close < (self.buyprice - 2 * self.ATR[0]) and self.buy_count > 0:
            # 输出止损信号卖出的信息
            print("止损信号卖出")
            # 执行卖出操作，并保存订单
            self.order = self.sell()
            # 清空买入计数
            self.buy_count = 0
    # 交易记录日志（默认不打印结果）
    def log(self, txt, dt=None, doprint=False):
        # 如果参数printlog为True或者doprint为True
        if self.params.printlog or doprint:
            # 如果dt为空，则使用self.datas[0].datetime.date(0)作为默认值
            dt = dt or self.datas[0].datetime.date(0)
            # 打印日志，格式为日期和txt
            print(f"{dt.isoformat()},{txt}")
    # 记录交易执行情况（默认不输出结果）
    def notify_order(self, order):
        # 如果order为submitted/accepted,返回空
        if order.status in [order.Submitted, order.Accepted]:
            return
        # 如果order为buy/sell executed,报告价格结果
        if order.status in [order.Completed]:
            if order.isbuy():
                # 买入
                self.log(
                    f"买入:\n价格:{order.executed.price},\
                成本:{order.executed.value},\
                手续费:{order.executed.comm}"
                )
                # 更新买入价格
                self.buyprice = order.executed.price
                # 更新买入手续费
                self.buycomm = order.executed.comm
            else:
                # 卖出
                self.log(
                    f"卖出:\n价格：{order.executed.price},\
                成本: {order.executed.value},\
                手续费{order.executed.comm}"
                )
            # 更新已执行条数
            self.bar_executed = len(self)
        # 如果指令取消/交易失败, 报告结果
        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.log("交易失败")
        # 清除订单
        self.order = None
    # 记录交易收益情况（可省略，默认不输出结果）
    def notify_trade(self, trade):
        # 判断交易是否已关闭
        if not trade.isclosed:
            # 如果未关闭，则返回不执行后续操作
            return
        # 输出策略收益信息
        self.log(f"策略收益：\n毛收益 {trade.pnl:.2f}, 净收益 {trade.pnlcomm:.2f}")
    def stop(self):
        # 打印日志，显示组合线参数和期末总资金
        self.log(
            f"(组合线：{self.p.long_period},{self.p.short_period})； 期末总资金: {self.broker.getvalue():.2f}",
            doprint=True,
        )

2.添加TraderSizer

class TradeSizer(bt.Sizer):
    params = (("stake", 1),)
    def _getsizing(self, comminfo, cash, data, isbuy):
        # 如果isbuy为真，则返回持仓数量
        if isbuy:
            return self.p.stake
        # 获取当前持仓信息
        position = self.broker.getposition(data)
        # 如果持仓数量为0
        if not position.size:
            # 返回0
            return 0
        else:
            # 返回持仓数量
            return position.size

3.回测

cerebro = bt.Cerebro()
cerebro.broker.setcash(100 * 10000.0)
cerebro.addanalyzer(bt.analyzers.PyFolio, _name="pyfolio")
data = bt.feeds.PandasData(dataname=df)
cerebro.adddata(data)
cerebro.addsizer(TradeSizer) #添加交易资金分配器
print("初始资金: %.2f" % cerebro.broker.getvalue())
cerebro.addstrategy(TurtleStrategy)
strats = cerebro.run()
print("期末资金: %.2f" % cerebro.broker.getvalue())

4.分析

strat0 = strats[0]
pyfolio = strat0.analyzers.getbyname("pyfolio")
returns, positions, transactions, gross_lev = pyfolio.get_pf_items()
import empyrical
print("累计收益：", round(empyrical.cum_returns_final(returns), 3))
print("年化收益：", round(empyrical.annual_return(returns), 3))
print("最大回撤：", round(empyrical.max_drawdown(returns), 3))
print("夏普比", round(empyrical.sharpe_ratio(returns), 3))
print("卡玛比", round(empyrical.calmar_ratio(returns), 3))