量化交易框架开发实践（一）

量化交易框架开发实践（一）

        量化交易平台指支持通过对数据进行多维度的定量分析，结合发现的特征定制策略，并能够基于历史数据对策略进行回测，最后支持实盘买卖的交易平台。

        从业务流上看，量化交易可以分解成：行情获取->数据清洗->指标计算->策略开发->策略回测->模拟/实盘交易；其中，指标计算(alpha,risk)+策略开发(算法+交易信号)可以被视为组合构建模块，而策略回测和实盘交易可以被视为执行模块。从框架上看，底层标的几乎可以是金融市场上所有产品，包括股票、期货、外汇、商品、数字货币等。当然，也可以应用到基金诊断、基金组合的构建和回测。

        从业务价值上看，最核心的组件自然是策略开发。各家量化机构也都是从不同维度进行因子挖掘，从而捕获交易机会。结合当下火热的GPT4，令人向往的就是策略开发是否可能借助GPT的发展而带来变化。早在2020年，Facebook提出的Pluribus已经在6人德扑桌上击败了顶尖的人类选手，而Pluribus和以往的AI也都不一样，它不需要预先设定最优交易策略，不需要提前设置在哪些牌型就下注，基于pot size和stack size决定是否要raise等等（相当于量化双均线策略、动量策略等），而是根据牌局的变化动态生成下注的信号。但Pluribus并没有开源，所以对于策略生成（AIGS）无法窥之细节。

        AIGC和AIGS是否能够被通用智能AGI统一目前还是很难讲的，因为在一些复杂游戏中，策略生成后，执行效果的评估和时间强相关，它其实是一个无限游戏。而无限游戏的目标并不是最大化/最小化某个指标，而是假设未来是未知的，所有人的胜负概率都一样。关键是赢的时候多赢一些，输的时候少输一些。好的策略就是选择合适的牌桌，设法让游戏一直继续，不要造成下牌桌的局面。从而在游戏过程中等待对手犯错获利。很明显，无限游戏不同于有限游戏，后者基于给定的目标总能找到某个逻辑进行分解。比如AutoGPT我用起来感觉就是一款优秀的基于特定目标求解的框架。

        话说回来，目前的主流量化框架包括：zipline、QuantLib、Backtrader、Finance-Py、vnpy等；这些开源框架各有特点，主要的差异体现在高性能、功能丰富度、实盘交易支持度、社区活跃度方面。

功能丰富度：QuantLib和VNPY功能最为强大,学习难度最大；而Finance-Py和PyAlgoTrade最为简单，学习难度也是最小的

实盘交易：QuantLib和VNPY更侧重高频与量化基金实盘；而PyAlgoTrade和Finance-Py侧重学习与小额实盘；

社区活跃度:VNPY和QuantLib社区最为活跃,更新速度较好,而Pyalgotrade已经不更新了。

高性能：C++编写的QuantLib、python编写的VNPY在性能方面也很优秀；

尤其是在支持的交易品种方面，vnpy已经支持了市面上几乎所有金融产品的交易，包括期货、股票、期权、外汇、数字货币；

所以综上考虑，个人小规模实践完全可以从PyAlgoTrade入手，再选择VNPY进行深入。

        我选择了聚宽平台作为行情获取来源，注册账号后，pip install qdatasdk就可以开始使用。jqdata的api做了速率的限制，循环调用数据获取接口会抛异常。这里，我使用backoff模块控制api调用频率。下载好数据就通过pandas写入本地文件夹，通过证券代码进行索引，以建立一个本地股票数据池。

#获取单个股票的行情数据@on_exception(expo, Exception, max_tries=5)def get_stock_data(code, start_date, end_date, source='local'):if source == 'jqdata':df = jq.get_price(code, start_date=start_date, end_date=end_date,         frequency='daily', fields=None, skip_paused=False, fq='pre', count=None)# 如果df不为空，则将index转换为datetime类型if pty:df.index = pd.to_datetime(df.index)return df#指定第1列作为indexdf = pd.read_csv('data/stock_data/' + code + '.csv',index_col=0)if pty:df.index = pd.to_datetime(df.index)df = df[start_date:end_date]else:print('error in get_stock_data: ', code, ' is empty')return df

         股票数据池可以分成三个部分：行情、基本面、估值；分别调用get_price、get_fundamentals、get_financial_data接口；我们通常也会通过市场指数来初始化股票池，这就要用到get_all_securities，get_index_stocks，get_index_weights三个数据接口；

#获取股票的估值数据pe_ration, pb_ration, ps_ration, pcf_ration等
#codes支持包含多只股票，比如['000001.XSHE', '600000.XSHG']
@on_exception(expo, Exception, max_tries=5)
def get_valuation_data(codes, start_date, end_date, source='jqdata'):df = pd.DataFrame()#优先从本地获取数据#读取valuation_data目录下的valuation_data.csv文件if ists('data/valuation_data/valuation_data.csv'):df = pd.read_csv('data/valuation_data/valuation_data.csv', index_col=0, parse_dates=True)#如果df包括了codes中的所有股票，那么就直接返回if set(codes).issubset(set(df['code'])):return dfif source == 'jqdata':#获取估值数据#获取start_date和end_date之间的最每个月估值数据#比如start_date是2018-01-01，end_date是2020-01-01，那么就获取2019-12-01,2020-01-01的估值数据statDate = []#start_date是个string类型，需要解析出yearstart_year = int(start_date.split('-')[0])end_year = int(end_date.split('-')[0])end_month = int(end_date.split('-')[1]) for year in range(start_year, end_year+1):#explain:下面这句话的意思是，如果year不等于end_year，那么就取13个月，如果year等于end_year，那么就取end_month个月final_month = 13 if year != end_year else end_month + 1for month in range(1, final_month):statDate.append(str(year) + '-' + str(month).zfill(2) + '-01')#遍历statDate，获取每个月的估值数据for date in statDate:df1 = jq.get_fundamentals(jq.query(jq.valuation,jq.income).filter(de.in_(codes)), date=date)if pty:df = df.append(df1)#以添加数据的方式保存到本地if pty:if ists('data/valuation_data/valuation_data.csv'):df.to_csv('data/valuation_data/valuation_data.csv', mode='a', header=None)_csv('data/valuation_data/valuation_data.csv')return df

        数据池建立好之后，我们可以开始进行一些指标的计算，包括最大回撤、股票涨跌幅、单次收益率、累计收益率等

#计算股价涨跌幅
def calculate_change_pct(df):#计算股票涨跌幅df['pct_change'] = df['close'].pct_change()#nan填充为0df['pct_change'] = df['pct_change'].fillna(0)print(df.head(10))return df#过滤交易信号
def filter_trading_signal(df):#过滤出buy和sell信号不为0的数据temp_df = df[(df['buy']!=0) | (df['sell']!=0)]#如果有连续buy的信号，则只保留第一个buy信号，其余的buy信号过滤掉#如果有连续sell的信号，则只保留第一个sell信号，其余的sell信号过滤掉temp_df['buy'] = temp_df['buy'].replace(1,0)temp_df['sell'] = temp_df['sell'].replace(-1,0)return temp_df#计算股票单次交易的收益率，平仓时股价市值-开仓时股价市值 / 开仓时股价市值
# 开仓时bug=1，平仓时sell=-1
def calculate_single_trade_return(df):#计算单次交易的收益率，单次交易的收益率 = (卖出时股价市值 - 最近一次的买入时股价市值) / 最近一次买入时的股价市值#不能用pct_change，因为pct_change表示的是当前股价和前一天股价的涨跌幅，而不是买入时的股价和卖出时的股价的涨跌幅#当遇到sell时，应该找上一次buy的股价市值，df['sell'] = -1时，找到上一次df['buy'] = 1的股价市值#df['sell'] = -1时，找到上一次df['buy'] = 1的股价市值temp_df = filter_trading_signal(df)temp_df['single_trade_return'] = (temp_df['close'] - temp_df['close'].shift(1)) / temp_df['close'].shift(1)#temp_df合并给df，如果temp_df中没有对应的数据，则用0填充df = df.merge(temp_df[['single_trade_return']],how='left',left_index=True,right_index=True)df['single_trade_return'] = df['single_trade_return'].fillna(0)#通过iplot绘制收益率曲线，横坐标是时间，纵坐标是收益率df['single_trade_return'].iplot(kind='line',title='single trade return')return df#计算累计收益率
def calcualte_cum_trade_return(df):#计算累计收益率,需要-1吗？因为是累计收益率，不是累计收益df['cum_trade_return'] = (df['single_trade_return'] + 1).cumprod() - 1#通过iplot绘制累计收益率曲线，横坐标是时间，纵坐标是收益率df['cum_trade_return'].iplot(kind='line',title='cum trade return')return df#计算股票的最大回撤：windows表示天数，在windows天内的最大回撤 = （windows天内的最大值 - windows天内的最小值) / windows天内的最大值
def calculate_max_drawdown(df,window=7):#计算最大回撤,raw=False表示传入的是一个series，而不是一个dataframe。rolling(window)表示计算window天内的最大回撤，返回的是一个seriesdf['max_drawdown'] = df['close'].rolling(window).apply(lambda x:(x.max() - x.min()) / x.max(),raw=False)#通过iplot绘制最大回撤曲线，横坐标是时间，纵坐标是最大回撤df['max_drawdown'].iplot(kind='line',title='max drawdown')return df

        然后，基于PyAlgoTrade框架进行数据对接；PyAlgoTrade内置了对quandl平台（类似聚宽）的支持，我们主要是对中国市场进行实践，所以并不能调用quandl的下载数据接口，使用build_feed进行构建feed即可。

        我们定义一个简单的策略进行测试，策略分为两部分：
1，如果收盘价高于SMA（10），那么就生成买入信号
2，如果已经购买了该股票，而收盘价低于SMA(10),那么就生成卖出信号

#定义策略，策略分为两部分
#1，如果收盘价高于SMA（15），那么就生成买入信号
#2，如果已经购买了该股票，而收盘价低于SMA(15),那么就生成卖出信号class MyStrategy(strategy.BacktestingStrategy):def __init__(self, feed, instrument):super(MyStrategy, self).__init__(feed, 1000)#10000是初始资金#保存股票代码self.__instrument = instrument#定义SMA指标self.__sma = ma.SMA(feed[instrument].getCloseDataSeries(), 10)#保存上一次的订单self.__position = None#获取股票代码def getInstrument(self):return self.__instrument#获取SMA指标def getSMA(self):return self.__sma#监听买入和 卖出事件def onEnterOk(self, position):execInfo = EntryOrder().getExecutionInfo()self.info("BUY at $%.2f" % (Price()))def onEnterCanceled(self, position):self.__position = Nonedef onExitOk(self, position):execInfo = ExitOrder().getExecutionInfo()self.info("SELL at $%.2f" % (Price()))self.__position = Nonedef onExitCanceled(self, position):# If the exit was canceled, re-submit it.self.__itMarket()#策略执行函数def onBars(self, bars):#如果SMA指标没有准备好，那么就什么都不做if self.__sma[-1] is None:return#获取当前的barbar = bars[self.__instrument]#如果没有持有股票，而且收盘价高于SMA（15），那么就生成买入信号if self.__position is None:Price() > self.__sma[-1]:#计算可以购买的股票数量#购买股票self.__position = Long(self.__instrument, 10, True)#如果已经购买了该股票，而收盘价低于SMA(15),那么就生成卖出信号；exitActive表示是否已经有订单Price() < self.__sma[-1] and not self.__itActive():#卖出                self.__itMarket()# print("sell shares: at price of ",Price()) # #获取数据
instruments = ['000001.XSHE']
feeds = quandlfeed.Feed()
feeds.addBarsFromCSV("000001.XSHE", "./data/stock_data/000001.XSHE.csv")
#创建策略
myStrategy = MyStrategy(feeds, "000001.XSHE")
#创建回测引擎
plt = plotter.StrategyPlotter(myStrategy)
#设置绘图参数
InstrumentSubplot("000001.XSHE").addDataSeries("SMA", SMA())
# 绘制每次交易的收益率
returnsAnalyzer = urns.Returns()
myStrategy.attachAnalyzer(returnsAnalyzer)
OrCreateSubplot("returns").addDataSeries("Simple returns", Returns())#运行策略,
myStrategy.setDebugMode(True)
myStrategy.run()#绘图
plt.plot()
# plt.show()
#打印最终结果
print("Final portfolio value: $%.2f" % Result())

        有了一个跑通的基础框架之后，我们可以逐步分解研究下PyAlgoTrade：

PyAlgoTrade框架主要包含六个部分，最重要的是strategies和feeds：

策略：Strategies；

回测数据：Feeds；

交易经纪人：Brokers；

时间序列数据：DataSeries；

技术分析：Technicals

优化器：Optimizer

        从前面的代码我们可以看到，PyAlgoTrade当前支持从csv文件中获取数据，形成feeds；我们也可以对不同的金融产品来源构建不同的feed，比如从mysql、sqlite中读取，或者读取比特币数据。csvfeed从csv文件中读取数据后，就会开始解析Date,Open,High,Low,Close,Volume,Adj Close从而生成。（PS：下载聚宽数据后，需要对colum进行设置，否则会报数据格式出错的问题。）

        接下来，我们研究下PyAlgoTrade的事件驱动机制，从而实现状态一致性；它和MetaTrader 4的设计理念是一致的，MQL也是采用事件回调来计算指标或者进行EA交易。