900字范文 > Python量化交易策略及回测系统

Python量化交易策略及回测系统

时间：2021-06-06 08:57:44

相关推荐

Python量化交易策略及回测系统

前言：行文思路

由于本文篇幅较长，而且文中关于python数据分析的知识点、python金融量化的知识点较多，因此在文章的开头先给出本文整体思路，以便读者能够拥有较为清晰的脉络通读全文。

第一部分：模块导入，主要是将后面需要用到的模块进行导入（简单，非重点）

第二部分：数据获取，鉴于在网络上股票数据不易找到，Wind金融终端等数据库数据收费，通过多方查找，终于让我找到了能够免费下载股票数据的模块，建议大家收藏（简单，非重点）

第三部分：将股票数据转化为新的数据类型，通过上面的方法下载下来的数据类型是我们常见的DataFrame，虽然pandas的功能已经很强大了，但是为了加入新的数据指标以及方便下一步操作，最好还是将DataFrame数据转化为一种新的数据类型（较难，非重点）

第四部分：策略编写，也就是利用代码将我们在股票市场中的交易原则表达出来（较难，重点）

第五部分：回测系统编写，股票回测即是基于历史已经发生过的真实行情数据，在历史上某一段时间内，模拟真实金融市场中股票的买入、卖出，得出这个时间段内的盈利率等数据（较难，重点）

第六部分：实例化策略并回测得到收益，继承一个策略类，得到一个实际的例子，利用股票数据回测得出结果（简单，重点）

1、模块导入

import akshare as akimport numpy as npimport pandas as pdimport matplotlib.pyplot as pltfrom collections import namedtuplefrom collections import OrderedDictfrom collections.abc import Iterablefrom functools import reducefrom abc import ABC, abstractmethod

akshare：用于下载股票的交易数据(前复权)

collections：对基本数据类型做进一步处理,实现特定的数据类型

abc：主要定义了基本类和最基本的抽象方法，可以为子类定义共有的接口API，不需要具体实现

2、数据获取

# 获取某一只股票一段时间的数据stock_sz300750_df = ak.stock_zh_a_daily(symbol="sz300750", start_date="0103", end_date="1231", adjust="qfq")# list_date = list(stock_sz300750_df['date'])# stock_sz300750_df.index = list_datestock_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_arrayself.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_arraydef _init_days(self, date_array):# 日期序列date_array = [str(date) for date in date_array]return date_arraydef _init_sma_short(self, price_array):# 5日移动平均线s_short = price_array.rolling(window=5, min_periods=5).mean()return s_shortdef _init_sma_long(self, price_array):# 30日移动平均线s_long = price_array.rolling(window=30, min_periods=30).mean()return s_longdef _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_dictdef 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.0for day in want_days:change_sum += day.changereturn change_sum

相同指标：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): # 只能被继承，不能实例化# 交易策略抽象基类@abstractmethoddef buy_strategy(self, *args, **kwargs):# 买入策略基类pass@abstractmethoddef sell_strategy(self, *args, **kwargs):# 卖出策略基类passclass TradeLoopBack(object):# 交易回测系统def __init__(self, trade_days, trade_strategy):"""使用上面封装的StockTradeDays类和编写的交易策略类TradeStrategyBase类初始化交易系统:param trade_days: StockTradeDays交易数据序列:param trade_strategy: TradeStrategyBase交易策略"""self.trade_days = trade_daysself.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 = -1def 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_shorttoday_long = trade_day.s_long# 昨日移动平均线值yesterday_short = trade_days[trade_ind - 1].s_shortyesterday_long = trade_days[trade_ind - 1].s_longif today_short > today_long and yesterday_short < yesterday_long:# 买入条件成立：self.keep_stock_day = 1def 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_shorttoday_long = trade_day.s_long# 昨日移动平均线值yesterday_short = trade_days[trade_ind - 1].s_shortyesterday_long = trade_days[trade_ind - 1].s_longif 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 = 0self.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].changeif today_down and yesterday_down and down_rate > self.s_buy_change_threshold:# 买入条件成立：连涨两天，涨幅超过s_buy_change_thresholdself.keep_stock_day += 1def 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].changeif today_down and yesterday_down and down_rate < self.s_sell_change_threshold:# 卖出条件成立：连跌两天，跌幅超过s_sell_change_thresholdself.keep_stock_day = 0@propertydef s_buy_change_threshold(self):# getter获取属性函数return self.__s_buy_change_threshold@s_buy_change_threshold.setterdef 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)@propertydef s_sell_change_threshold(self):# getter获取属性函数return self.__s_sell_change_threshold@s_sell_change_threshold.setterdef 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、实例化策略

# 实例化策略1trade_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())

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

# 实例化策略2trade_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 = -1for ind, day in enumerate(trade_days):short2 = day.s_shortlong2 = day.s_longshort1 = trade_days[ind - 1].s_shortlong1 = trade_days[ind - 1].s_longif pd.isna(long1):continueif flag == 1:changes_list_1.append(day.change)print("日期：{}，持有中".format(day.date))if short2 > long2 and short1 < long1:flag = 1print("日期：{}，买入策略执行".format(day.date))if short2 < long2 and short1 > long1:flag = 0print("日期：{}，卖出策略执行".format(day.date))print('回测策略1总盈亏为：{}%'.format(reduce(lambda a, b: a + b, changes_list_1) * 100))plt.plot(np.array(changes_list_1).cumsum())

结果：

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

结果：