[人工智能]Python量化交易策略及回测系统

前言：行文思路

由于本文篇幅较长，而且文中关于python数据分析的知识点、python金融量化的知识点较多，因此在文章的开头先给出本文整体思路，以便读者能够拥有较为清晰的脉络通读全文。
第一部分：模块导入，主要是将后面需要用到的模块进行导入（简单，非重点）
第二部分：数据获取，鉴于在网络上股票数据不易找到，Wind金融终端等数据库数据收费，通过多方查找，终于让我找到了能够免费下载股票数据的模块，建议大家收藏（简单，非重点）
第三部分：将股票数据转化为新的数据类型，通过上面的方法下载下来的数据类型是我们常见的DataFrame，虽然pandas的功能已经很强大了，但是为了加入新的数据指标以及方便下一步操作，最好还是将DataFrame数据转化为一种新的数据类型（较难，非重点）
第四部分：策略编写，也就是利用代码将我们在股票市场中的交易原则表达出来（较难，重点）
第五部分：回测系统编写，股票回测即是基于历史已经发生过的真实行情数据，在历史上某一段时间内，模拟真实金融市场中股票的买入、卖出，得出这个时间段内的盈利率等数据（较难，重点）
第六部分：实例化策略并回测得到收益，继承一个策略类，得到一个实际的例子，利用股票数据回测得出结果（简单，重点）

1、模块导入

import akshare as ak
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from collections import namedtuple
from collections import OrderedDict
from collections.abc import Iterable
from functools import reduce
from abc import ABC, abstractmethod

akshare：用于下载股票的交易数据(前复权)
collections：对基本数据类型做进一步处理,实现特定的数据类型
abc：主要定义了基本类和最基本的抽象方法，可以为子类定义共有的接口API，不需要具体实现

2、数据获取

# 获取某一只股票一段时间的数据
stock_sz300750_df = ak.stock_zh_a_daily(symbol="sz300750", start_date="20200103", end_date="20211231", adjust="qfq")
# list_date = list(stock_sz300750_df['date'])
# stock_sz300750_df.index = list_date
stock_sz300750_df.head()

函数ak.stock_zh_a_daily()用于获取A股股票数据
symbol为股票代码，sh为上交所股票，sz为深交所股票；strat_date、end_date分别为股票数据开始时间、结束时间；adjust默认为不复权的数据， qfq是返回前复权后的数据，hfq是 返回后复权后的数据

3、股票数据类型转换

由于后面写了两个量化交易策略，而且策略中的有部分指标不相同，所以在这一部分以及下面回测系统两部分面向对象编程，有部分函数只用于策略一，有部分只用于策略二。

# 将股票数据转化为新的数据类型
class StockTradeDays(object):
    def __init__(self, price_array, date_array=None):
        self.price_array = price_array
        self.date_array = self._init_days(date_array)
        self.change_array = self._init_change()
        self.s_short = self._init_sma_short(price_array)
        self.s_long = self._init_sma_long(price_array)
        self.stock_dict = self._init_stock_dict()

    def _init_change(self):
        # 收益率

        change_array = self.price_array.pct_change(periods=1)
        return change_array

    def _init_days(self, date_array):
        # 日期序列
        
        date_array = [str(date) for date in date_array]
        return date_array

    def _init_sma_short(self, price_array):
        # 5日移动平均线
        
        s_short = price_array.rolling(window=5, min_periods=5).mean()
        return s_short
    
    def _init_sma_long(self, price_array):
        # 30日移动平均线
        
        s_long = price_array.rolling(window=30, min_periods=30).mean()
        return s_long
    
    def _init_stock_dict(self):
        # 将股票的日期、收盘价、涨跌幅转化为一个新的数据类型

        stock_namedtuple = namedtuple('stock', ('date', 'price', 'change', 's_short', 's_long'))

        # 使用以被赋值的date_array等进行OrderedDict的组装
        stock_dict = OrderedDict((date, stock_namedtuple(date, price, change, s_short, s_long)) for date, price, change, s_short, s_long 
                                 in zip(self.date_array, self.price_array, self.change_array, self.s_short, self.s_long))
        return stock_dict

   
    def filter_stock(self, want_up=True, want_calc_sum=False):
        # 判断交易日股票是上涨还是下跌
        # Python中的三目表达式的写法
        filter_func = (lambda p_day: p_day.change > 0) if want_up else (lambda p_day: p_day.change < 0)
        # 使用filter_func做筛选函数
        want_days = list(filter(filter_func, self.stock_dict.values()))

        if not want_calc_sum:
            return want_days

        # 需要计算涨跌幅和
        change_sum = 0.0
        for day in want_days:
            change_sum += day.change
        return change_sum    
   
    #下面的__str__，__iter__， __getitem__， __len__,这些以双下划线开头结尾的名字表示系统保留定义
    def __str__(self):
        return str(self.stock_dict)
    
    __repr__ = __str__

    def __iter__(self):        
    # 通过代理stock_dict的迭代，yield元素
        for key in self.stock_dict:
            yield self.stock_dict[key]

    def __getitem__(self, ind):
        date_key = self.date_array[ind]
        return self.stock_dict[date_key]

    def __len__(self):
        return len(self.stock_dict)

相同指标：date、price、change
策略一：s_short、s_long分别为5日移动平均线和30日移动平均线；可以根据自己的需求更改参数数据
策略二：函数filter_stock()，用于判断交易日股票是上涨还是下跌

最后将DataFrame数据转换为自定义数据类型OrderedDict

trade_days = StockTradeDays(stock_sz300750_df['close'], stock_sz300750_df['date'])
if isinstance(trade_days, Iterable) :
    for day in trade_days:
        print(day) 

4、回测系统编写

class TradeStrategyBase(ABC, object):   # 只能被继承，不能实例化
     # 交易策略抽象基类

    @abstractmethod
    def buy_strategy(self, *args, **kwargs):
        # 买入策略基类
        pass

    @abstractmethod
    def sell_strategy(self, *args, **kwargs):
        # 卖出策略基类
        pass


class TradeLoopBack(object):
    # 交易回测系统

    def __init__(self, trade_days, trade_strategy):
        """
        使用上面封装的StockTradeDays类和编写的交易策略类
        TradeStrategyBase类初始化交易系统
        :param trade_days: StockTradeDays交易数据序列
        :param trade_strategy: TradeStrategyBase交易策略
        """
        self.trade_days = trade_days
        self.trade_strategy = trade_strategy
        # 交易盈亏结果序列
        self.profit_array = []

    def execute_trade(self):
        # 执行交易回测
        
        for ind, day in enumerate(self.trade_days):
        # 以时间驱动，完成交易回测
            
            if self.trade_strategy.keep_stock_day == 1 or self.trade_strategy.keep_stock_day > 0:
                # 如果有持有股票，加入交易盈亏结果序列
                self.profit_array.append(day.change)
#                 print("日期：{}，持有中".format(day.date))

            # hasattr: 用来查询对象有没有实现某个方法
            if hasattr(self.trade_strategy, 'buy_strategy'):
                # 买入策略执行
                self.trade_strategy.buy_strategy(ind, day, self.trade_days)
#                 if self.trade_strategy.keep_stock_day == 1:
#                     print("日期：{}，买入策略执行".format(day.date))

            if hasattr(self.trade_strategy, 'sell_strategy'):
                # 卖出策略执行
                self.trade_strategy.sell_strategy(ind, day, self.trade_days)
#                 if self.trade_strategy.keep_stock_day == 0:
#                     print("日期：{}，卖出策略执行".format(day.date))

execute_trade()函数中，利用循环遍历整一个交易时段，将获得的每日股票数据传给交易策略进行判断，最终确定是买入、卖出还是持有

5、策略编写

class TradeStrategy1(TradeStrategyBase):
    """
        交易策略1: 利用5日移动平均线与30日移动平均线交叉点进行股票买卖
        当5日移动平均线从下往上穿过30日移动平均线时，买入股票并持有
        当5日移动平均线从上往下穿过30日移动平均线时，卖出股票
    """
    def __init__(self, stock_df):        
        self.keep_stock_day = -1

    def buy_strategy(self, trade_ind, trade_day, trade_days):
        if not pd.isna(trade_days[trade_ind - 1].s_long):
        # 只有当长期移动平均线的数据有了，才能进行下一步操作    
            
            # 今日移动平均线值
            today_short = trade_day.s_short
            today_long = trade_day.s_long

            # 昨日移动平均线值
            yesterday_short = trade_days[trade_ind - 1].s_short
            yesterday_long = trade_days[trade_ind - 1].s_long

            if today_short > today_long and yesterday_short < yesterday_long:
                # 买入条件成立：
                self.keep_stock_day = 1

    def sell_strategy(self, trade_ind, trade_day, trade_days):
        if not pd.isna(trade_days[trade_ind - 1].s_long):
        # 只有当长期移动平均线的数据有了，才能进行下一步操作            
            
            # 今日移动平均线值
            today_short = trade_day.s_short
            today_long = trade_day.s_long

            # 昨日移动平均线值
            yesterday_short = trade_days[trade_ind - 1].s_short
            yesterday_long = trade_days[trade_ind - 1].s_long

            if today_short < today_long and yesterday_short > yesterday_long:
                # 卖出条件成立：
                self.keep_stock_day = 0

移动平均线是将一定时期内的股票价格加以平均，把不同时间的平均值连接起来形成一根MA，利用长短期的移动平均线交叉点观察股票价格变动趋势的一种技术指标。因此，只有到了第30天才可以获得30日移动平均值，才可能进行买卖。
判断买入条件：当短期移动平均线从下往上穿过长期移动平均线时，可以认为短期内股价的趋势向上，股价可能会上涨
判断卖出条件：当短期移动平均线从上往下穿过长期移动平均线时，可以认为短期内股价的趋势向下，股价可能会下跌

class TradeStrategy2(TradeStrategyBase):
    """
        交易策略2: 追涨杀跌策略，当股价连续两个交易日上涨
        且上涨幅度超过阀值默认s_buy_change_threshold()，买入股票并持有
        当股价连续两个交易日下跌
        且下跌幅度超过阀值默认s_sell_change_threshold()，卖出股票
    """
    def __init__(self):
        self.keep_stock_day = 0
        self.s_buy_change_threshold = 0.05      # 上涨买入阀值
        self.s_sell_change_threshold = -0.05   #下跌卖出阀值

    def buy_strategy(self, trade_ind, trade_day, trade_days):
        if self.keep_stock_day == 0 and trade_ind >= 1:
            """
                当没有持有股票的时候self.keep_stock_day == 0 并且
                trade_ind >= 1, 不是交易开始的第一天，因为需要yesterday数据
            """
            # trade_day.change > 0 bool：今天是否股价上涨
            today_down = trade_day.change > 0
            # 昨天是否股价上涨
            yesterday_down = trade_days[trade_ind - 1].change > 0
            # 两天总涨幅
            down_rate = trade_day.change + trade_days[trade_ind - 1].change
            if today_down and yesterday_down and down_rate > self.s_buy_change_threshold:
                # 买入条件成立：连涨两天，涨幅超过s_buy_change_threshold
                self.keep_stock_day += 1

    def sell_strategy(self, trade_ind, trade_day, trade_days):
        # trade_day.change < 0 bool：今天是否股价下跌
        today_down = trade_day.change < 0
        # 昨天是否股价下跌
        yesterday_down = trade_days[trade_ind - 1].change < 0
        # 两天总跌幅
        down_rate = trade_day.change + trade_days[trade_ind - 1].change
        if today_down and yesterday_down and down_rate < self.s_sell_change_threshold:
            # 卖出条件成立：连跌两天，跌幅超过s_sell_change_threshold
            self.keep_stock_day = 0

    @property
    def s_buy_change_threshold(self):
        # getter获取属性函数
        return self.__s_buy_change_threshold

    @s_buy_change_threshold.setter
    def s_buy_change_threshold(self, s_buy_change_threshold):
        # setter属性赋值
        if not isinstance(s_buy_change_threshold, float):
            """
                上涨阀值需要为float类型
            """
            raise TypeError('buy_change_threshold must be float!')
        # 上涨阀值只取小数点后两位
        self.__s_buy_change_threshold = round(s_buy_change_threshold, 2)

    @property
    def s_sell_change_threshold(self):
        # getter获取属性函数
        return self.__s_sell_change_threshold

    @s_sell_change_threshold.setter
    def s_sell_change_threshold(self, s_sell_change_threshold):
        # setter属性赋值
        if not isinstance(s_sell_change_threshold, float):
            """
                上涨阀值需要为float类型
            """
            raise TypeError('buy_change_threshold must be float!')
        # 上涨阀值只取小数点后两位
        self.__s_sell_change_threshold = round(s_sell_change_threshold, 2)

策略二可以认为是非理性人在股票市场中交易时，遇到多日上涨且上涨幅度较大时，会认为股票有继续上涨的趋势，为了获利所以买入股票；但当某一股票连续下跌且下跌幅度超过心理预期时，会认为股票又继续下跌的趋势，为了止损卖出股票。
策略二中买入股票条件为：当股价连续两个交易日上涨且上涨幅度超过0.05，买入股票并持有
卖出条件为：当股价连续两个交易日下跌且下跌幅度超过-0.05，卖出股票
相关参数可以根据需求修改

6、实例化策略

# 实例化策略1
trade_strategy1 = TradeStrategy1(stock_sz300750_df)
trade_loop_back = TradeLoopBack(trade_days, trade_strategy1)
trade_loop_back.execute_trade()
print('回测策略1总盈亏为：{}%'.format(reduce(lambda a, b: a + b, trade_loop_back.profit_array) * 100))
plt.plot(np.array(trade_loop_back.profit_array).cumsum())

经过前面的所有步骤之后，就可以实例化一个交易策略，利用交易数据进行回测，可得到相应的结果：

# 实例化策略2
trade_strategy2 = TradeStrategy2()
trade_loop_back = TradeLoopBack(trade_days, trade_strategy2)
trade_loop_back.execute_trade()
print('回测策略2总盈亏为：{}%'.format(reduce(lambda a, b: a + b, trade_loop_back.profit_array) * 100))
plt.plot(np.array(trade_loop_back.profit_array).cumsum())

结果：

非面向对象的编程

由于对面向对象编程不太擅长，所以我对两个策略又分别写了新的程序，以判断上文面向对象程序是否正确

changes_list_1 = []
flag = -1
for ind, day in enumerate(trade_days):
    short2 = day.s_short
    long2 = day.s_long
    short1 = trade_days[ind - 1].s_short
    long1 = trade_days[ind - 1].s_long
    if pd.isna(long1):
        continue
    if flag == 1:
        changes_list_1.append(day.change)
        print("日期：{}，持有中".format(day.date))
    if short2 > long2 and short1 < long1:
        flag = 1
        print("日期：{}，买入策略执行".format(day.date))
    if short2 < long2 and short1 > long1:
        flag = 0
        print("日期：{}，卖出策略执行".format(day.date))
        
print('回测策略1总盈亏为：{}%'.format(reduce(lambda a, b: a + b, changes_list_1) * 100))
plt.plot(np.array(changes_list_1).cumsum())

结果：

# 策略2
changes_list_2 = []
flag = 0
for ind, day in enumerate(trade_days):
    today_down = day.change
    yesterday_down = trade_days[ind - 1].change
    if flag > 0:
        changes_list_2.append(day.change)
        print("日期：{}，持有中".format(day.date))
    if today_down > 0 and yesterday_down > 0 and today_down + yesterday_down > 0.01:
        flag += 1
        print("日期：{}，买入策略执行".format(day.date))
    if today_down < 0 and yesterday_down < 0 and today_down + yesterday_down < -0.01:
        flag = 0
        print("日期：{}，卖出策略执行".format(day.date))
        
print('回测策略2总盈亏为：{}%'.format(reduce(lambda a, b: a + b, changes_list_2) * 100))
plt.plot(np.array(changes_list_2).cumsum())

结果：

总结

以上策略只用于量化分析，并不适合用于实际交易，之所以有较高的盈利，得益于宁王领衔的新能源板块的强势，大家也可以试试其他的股票，比如药明康德（代码：SH603259)


可以看出策略对该股票进行回测交易时，获得的盈利并不客观，甚至出现较大的亏损，因此，需要对相关策略进行参数调整修改，或者发掘其他更为有效的策略……