上海股票交易所数据接口-Python 量化交易学习笔记海龟策略

策略思想

唐奇安通道+ATR

交易信号

入场：价格突破20日价格高点时，入场；加仓：价格继续上涨至0.5倍ATR，再次加仓，加仓次数不超过3次；止损：价格回落2倍ATR时止损离场；止盈：价格突破10日最低点时止盈离场；做空与做多的逻辑相反。

回测结果

初始资金	10000
期货品种	ETH
时间级别	1H
回测时间	2017.7.15 - 2022.7.24
倍数	1
手续费	1%
总盈利	993

核心代码

加载数据

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)

主程序

if __name__ == "__main__":
    path = "D:\work\git\Tools\static\data\ETHUSDT_1h.csv"
    data = load_csv_data(path)
    cerebro = create_cerebro()
    cerebro.adddata(data)
    cerebro.addstrategy(TurtleTradingStrategy)
    cerebro.run()
    cerebro.plot()

目前从回测结果来看，该策略明显没跑赢大盘，下篇文章优化该策略，使得策略可以获得更高的收益

文章为作者独立观点，不代表股票程序化软件自动交易接口观点