量化交易学习笔记（1） 突破策略

策略思想

指标

使用SMA和收盘价

买入信号

价格出现连续4根k线的收盘价下跌，并且收盘价突破sma50

订单

出现交易信号后，价格回调2%，添加买入订单，止损价格当前价格×（1-0.05），止盈价格当前价格×（1+0.1）

回测结果

核心代码

突破策略

import datetime

import backtrader as bt

from utils import load_csv_data


class BaseStrategy(bt.Strategy):
    p = {
        "islog": False
    }

    def log(self, message):
        if self.p['islog']:
            print('[{}] [{}]'.format(bt.num2date(self.data.datetime[0]), message))
        pass

    def notify_order(self, order):
        log_txt = ""
        if order.status in [bt.Order.Submitted, bt.Order.Accepted]:
            return
        if order.status == order.Completed:
            if order.isbuy():
                log_txt = '买:%.2f  量:%s  持仓:%s' % (order.executed.price, order.executed.size, self.position.size)
                self.log(log_txt)
            else:
                log_txt = '卖:%.2f 量:%s  持仓:%s' % (order.executed.price, order.executed.size, self.position.size)
                self.log(log_txt)
        if order.status in [bt.Order.Close]:
            log_txt = '平:%.2f 量:%s  持仓:%s' % (order.executed.price, order.executed.size, self.position.size)
            self.log(log_txt)


    def next_open(self):
        # 参数优化
        if self.is_optimize_params():
            self.optimize_params()


class BreakthroughStrategy(bt.Strategy):
    """
    突破策略
    价格出现连续4根k线的收盘价下跌，并且收盘价突破sma50
    """
    params = dict(
        break_through=0.03,
        callback=0.02,  # 价格回调比例
        period=50,  # sma周期
        down_day=4,  # 连续下跌天数
        stop_loss=0.05,  # 止损比例
        take_profit=0.1,  # 止盈比例
        validity_day=3,  # 订单有效期
        expired_day=1000,  # 订单失效期
    )

    def notify_order(self, order):
        if order.status == order.Completed:
            self.holdstart = len(self)
        if not order.alive() and order.ref in self.orefs:
            self.orefs.remove(order.ref)

    def __init__(self):
        self.holdstart = None
        self.dataclose = self.datas[0].close  # 收盘价
        self.sma = bt.ind.SMA(period=self.p.period, plot=True)  # SMA
        self.orefs = list()  # order列表，用于存储尚未执行完成的订单

    def next(self):
        # 有尚未执行的订单
        if self.orefs:
            return
            # 尚未进场
        if not self.position:

            # 获取近几日收盘价用于判断是否连续下跌
            last_closes = list()
            for i in range(1, self.p.down_day + 1):
                last_closes.append(self.dataclose[-i])
            # 连续N日下跌 在 sma上方
            if last_closes == sorted(last_closes, reverse=True) and self.dataclose[0] > self.sma[0]:
                p1 = self.dataclose[0] * (1.0 - self.p.callback)
                p2 = p1 - self.p.stop_loss * p1
                p3 = p1 + self.p.take_profit * p1
                # 计算订单有效期
                validity_day = datetime.timedelta(self.p.validity_day)
                expired_day = valid3 = datetime.timedelta(self.p.expired_day)
                size = self.broker.getcash() / self.data.high[0]
                # 使用bracket orders设置买入卖出
                if size is None:
                    size = 1
                os = self.buy_bracket(size=size,
                                      price=p1, valid=validity_day,
                                      stopprice=p2, stopargs=dict(valid=expired_day),
                                      limitprice=p3, limitargs=dict(valid=valid3), )
                # 保存激活的的订单
                self.orefs = [o.ref for o in os]


class ThreeMovingAverage(bt.Strategy):
    params = dict(
        short_period=5,
        median_period=20,
        long_period=60,
        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.close = self.datas[0].close
        self.s_ma = bt.ind.SMA(period=int(self.p.short_period))
        self.m_ma = bt.ind.SMA(period=int(self.p.median_period))
        self.l_ma = bt.ind.SMA(period=int(self.p.long_period))
        # 捕获做多信号
        # 短期均线在中期均线上方，且中期均取也在长期均线上方，三线多头排列，取值为1；反之，取值为0
        self.signal1 = bt.And(self.m_ma > self.l_ma, self.s_ma > self.m_ma)
        # 做多信号，求上面 self.signal1 的环比增量，可以判断得到第一次同时满足上述条件的时间，第一次满足条件为1，其余条件为0
        self.long_signal = bt.If((self.signal1 - self.signal1(-1)) > 0, 1, 0)
        # 做多平仓信号，短期均线下穿长期均线时，取值为1；反之取值为0
        self.close_long_signal = bt.ind.CrossDown(self.s_ma, self.m_ma)
        # 捕获做空信号和平仓信号，与做多相反
        self.signal2 = bt.And(self.m_ma < self.l_ma, self.s_ma < self.m_ma)
        self.short_signal = bt.If((self.signal2 - self.signal2(-1)) > 0, 1, 0)
        self.close_short_signal = bt.ind.CrossUp(self.s_ma, self.m_ma)

    def next(self):
        #         self.log(self.sell_signal[0],doprint=True)
        #         self.log(type(self.position.size),doprint=True)
        # 如果还有订单在执行中，就不做新的仓位调整

        # 如果当前持有多单
        if self.position.size > 0:
            #             self.log(self.position.size,doprint=True)
            # 平仓设置,出现平仓信号进行平仓
            if self.close_long_signal == 1:
                self.order = self.sell(size=abs(self.position.size))
        # 如果当前持有空单
        elif self.position.size < 0:
            # 平仓设置，出现平仓信号进行平仓
            if self.close_short_signal == 1:
                self.order = self.buy(size=abs(self.position.size))
        else:  # 如果没有持仓，等待入场时机
            # 入场: 出现做多信号，做多，开四分之一仓位
            if self.long_signal == 1:
                self.buy_unit = int(self.broker.getcash() / self.close[0] / 4)
                self.order = self.buy(size=self.buy_unit)
            # 入场: 出现做空信号，做空，开四分之一仓位
            elif self.short_signal == 1:
                self.sell_unit = int(self.broker.getcash() / self.close[0] / 4)
                self.order = self.sell(size=self.sell_unit)

    # 打印订单日志
    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 stop(self):
        self.log(
            f'(组合线：{self.p.short_period},{self.p.median_period},{self.p.long_period}); 期末总资金: {self.broker.getvalue():.2f}',
            doprint=False)


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


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(BreakthroughStrategy)
    cerebro.run()
    cerebro.plot()
    print(cerebro.broker.getvalue() - 10000)

主程序

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(BreakthroughStrategy)
    cerebro.run()
    print(".2f" % cerebro.broker.getvalue() - 10000)
    cerebro.plot()

总结

目前从回测结果来看，该策略在破产的的边缘试探，表现结果十分糟糕。下一篇文章会对该策略进行优化，使策略可持续盈利，加油！！！！