< {end_date}"cur.execute(sql)data = http://www.kingceram.com/post/cur.fetchall()data = pd.DataFrame(data)data = data.rename(columns={0:"date", 1: "stock_code", 2: "open", 3: "high", 4: "low", 5: "close", 6: "volume"})stock_data = http://www.kingceram.com/post/self.data_convert(data)cur.close()conn.close()return stock_datadef p_l(self, test_data):df = test_dataopen = df.open.valueshigh = df.high.valueslow = df.low.valuesclose= df.close.values# 总的样本数量n = len(close)# 使用talib计算移动均线ma1 = ta.SMA(df.close.values,timeperiod=self.L1)ma2 = ta.SMA(df.close.values,timeperiod=self.L2)# 计算趋势con_long = ma1> ma2con_short = ma1 < ma2trend = np.zeros(n)trend[con_long] = 1trend[con_short] = -1# 仓位变化 ， 比如从﹣1到1 ， 变化为2sig = np.zeros(n)# 当前总仓位pos = np.zeros(n)# 新仓位的开仓价pce = np.zeros(n)# 保存交易信息trade_info = [" " for i in range(n)]# 每次开仓一手new_pos = 1for i in range(self.L2, n):# 正常情况下 ， 仓位保持不变pos[i] = pos[i-1]# 昨天收盘 ， 新出现多头趋势 ， 开盘就开多头仓（如果有空头仓 ， 则先平空头仓）if trend[i-1] > 0 and pos[i-1] <= 0:# 目标仓位pos[i] = new_pos# 计算仓位变化sig[i] = new_pos-pos[i-1]# 记录交易价格pce[i] = open[i]# 记录交易log信息trade_info[i]=u'long at %s' %(pce[i])# 昨天收盘 ， 新出现空头趋势 ， 开盘就开空头仓（如果有多头仓 ， 则先平多头仓）elif trend[i-1] < 0 and pos[i-1] >= 0:# 目标仓位pos[i] = -new_pos# 计算仓位变化sig[i] = -new_pos-pos[i-1]# 记录交易价格pce[i] = open[i]# 记录交易log信息trade_info[i]=u'short at %s' %(pce[i])df = pd.DataFrame({'open':open, 'high':high, 'low' :low, 'close':close, 'ma1':ma1, 'ma2':ma2, 'trend':trend, \'sig':sig, 'pce':pce, 'pos':pos, 'trade_info':trade_info},\columns=['open','high','low','close','ma1','ma2','trend','sig','pos','pce','trade_info'])df['new_pos'] = df['pos'] - df['pos'].shift(1)df['old_pos'] = df['pos'] - df['new_pos']df['p&l_new'] = (df['close']-df['pce']) * df['new_pos']df['p&l_old'] = (df['close']-df['close'].shift(1)) * df['old_pos']df['p&l'] = df['p&l_new'] + df['p&l_old']del df['new_pos']del df['old_pos']del df['p&l_new']del df['p&l_old']df=df.dropna()return dfsd = StockData(3, 7)stock_code = '002245' #'600519'start_date = '20230104'end_date = '20230904'stock_dt = sd.get_stock_data(stock_code, start_date, end_date)# print(stock_dt)net_pl = sd.p_l(stock_dt)np.sum(net_pl['p&l'])
结果为：
3、简介
本部分将简单介绍一下开源的事件驱动系统的使用 。这里使用的是 0.18版本 。并没有自带 ， 所以读者需要自行下载安装 ， 这里不再介绍 。
pip install pyalgotrade
首先是数据问题 ， 是将数据封装在其提供的feed类中 ， 读取数据的方式是读取csv文件 ， 是不能直接读取数据框的数据 。需要从数据源中获取数据 ， 例如CSV文件、MySQL数据库或者在线数据API（如Yahoo ） 。如果您想使用数据框中的数据 ， 您需要先将其读入到支持的数据源中 ， 然后再使用 。您可以将数据框转换为CSV文件或者MySQL数据库 ， 然后使用的数据接口来读取数据 。
因为从本地读取csv文件极不方便 ， 因此我直接从数据库拉取数据 ， 下面是完整的回测代码：
#coding=utf-8#mac is Moving Average Crossoverimport pymysqlimport numpy as npimport pandas as pdfrom pyalgotrade import barfrom datetime import datetimefrom pyalgotrade.technical import mafrom pyalgotrade import strategy,plotterfrom pyalgotrade.bar import BasicBar, Frequency# from pyalgotrade.barfeed import BarFeedfrom pyalgotrade.stratanalyzer import returns, sharpefrom pyalgotrade.barfeed.csvfeed import GenericBarFeed # 从csv或者数据库读取数据class MyStrategy(strategy.BacktestingStrategy):# 初始化bar线 ， stock_code ， 均线1和2的周期 ， 本金def __init__(self,feed, stock_code, smaPeriod1, smaPeriod2, par):if feed is None:returnelse:strategy.BacktestingStrategy.__init__(self, feed, par)self.__stock_code = stock_codeself.__sma1=ma.EMA(feed[stock_code].getPriceDataSeries(),smaPeriod1)self.__sma2=ma.EMA(feed[stock_code].getPriceDataSeries(),smaPeriod2)self.__shortPos = Noneself.__longPos = None# 获取到bars线数据重新初始化函数def setFeed(self,feed, stock_code, smaPeriod1, smaPeriod2, par):self.__feed = feedself.__init__(feed, stock_code, smaPeriod1, smaPeriod2, par)# 拉取数据库对数据进行转换def data_convert(self, sql_data):# sql_data["date"] = [str(date_str) for date_str in sql_data["date"]]# sql_data["date"] = [datetime.strptime(date_str, "%Y-%m-%d").strftime("%Y-%m-%d %H:%M:%S") for date_str in sql_data["date"]]sql_data["open"] = sql_data['open'].astype(float)sql_data["high"] = sql_data['high'].astype(float)sql_data["low"] = sql_data['low'].astype(float)sql_data["close"] = sql_data['close'].astype(float)sql_data["volume"] = sql_data['volume'].astype(float)sql_data = http://www.kingceram.com/post/sql_data.dropna()return sql_data# 返回计算MA的元组数值def getSMA(self):return self.__sma1,self.__sma2# 持仓进入市场被取消时触发 ， 检查被取消的持仓是否匹配long和short ， 反映持仓能够正确更新def onEnterCanceled(self, position):if self.__shortPos == position:self.__shortPos = Noneelif self.__longPos == position:self.__longPos = Noneelse:assert False# 检查持仓交易成功进入def onEnterOk(self, position):execInfo = position.getEntryOrder().getExecutionInfo()self.info('buy at %.2f' % (execInfo.getPrice()))# 检查持仓交易成功退出def onExitOk(self, position):execInfo = position.getExitOrder().getExecutionInfo()self.info('sell at %.2f' % (execInfo.getPrice()))if self.__shortPos == position:self.__shortPos = Noneelif self.__longPos == position:self.__longPos = Noneelse:assert False# 交易信号 ， 短线MA＞长线MA进入多头(退出空头)def onExitCanceled(self, position):position.exitMarket()def enterLongSignal(self, bar):return self.__sma1[-1]> self.__sma2[-1]def exitLongSignal(self):return self.__sma1[-1]