10历史数据服务的实操课_W

1、2019/2/2110【数据】历史数据服务的实操课 - 华尔街见闻搜索首页资讯快讯行情日历主题专栏数据APP商城会员 投资全球更要投资自己我的订阅/premium/articles/34818255/10参与评论10【数据】历史数据服务的实操课来自特辑30堂实战课跑通量化交 收藏作者： 陈晓优微信微博2019-02-21 13:36字数 7,558阅读需 19分钟最近更新易小白学量化从0-1直达中级648 人已购上一篇:09 【数据】选择： 合约指数vs连续合约 马上添加助教微信（jwzhujiao2） 发送订单截图申请名额本期注意点：本节课程用时

2、约30分钟，包含代码讲解，为更好理解本课，请至少预留一小时用于学习本期提要09 【数据】选择：合约指数vs连续合约2019-02-19更新08 【数据】获取历史数据的实操课2019-02-14更新三步骤：数据的获取、数据的处理以及数据的保存 历史数据服务的工作流程与代码实现框架重点：详解data service的代码实现（两部分实操内容） 本课相关代码下载 点击下载本课程的第二次课后作业大家好，欢迎来到Quant全实战成长计划第一阶的第十集，在上一集里面我们比较了合约指数和连续合约之间具体的优劣，以及做出了我们的选择。那么在这一集里面我们就要来实现一个叫做历史数据服务的东西，听起来有点玄幻，其

3、实本质上就是一个每天定时自动把新的远端（不管是市场、还是说其他什么来源生成的）数据给同步到我们本地自己的数据库里面的这么一个工具，我们自己来把它给实现出来。数据日常运维三步骤首先我们来看看什么是数据的日常运维。通常所有的数据运维不管你是个人也好，还是说是在一个机构里面，比如说你在一个私募基金，比如说你在一个量化自营的，其实大家数 据运维都差不了太远。首先我们通常需要一个叫做数据服务商的接口，比如说我们这里选择的就是RQdata， 第二个我们需要一个python的程序，在这里就是我们自己写的data service，最后需要一个用来保存数据的数据库，我们选择的是MangoDB以上三步分别对应的是

4、数据的获取、数据的处理以及数据的保存。那么经过这三个步骤之后，所有的数据都在我们本地的数据库里面，后续就可以来做我们策略的回测研究，或者说做我们实盘交易，我们每天策略开始跑之前的状态初始化。有了这个数据之后，后面两个工作就是非常自然而然水到渠成了，基本上你就不用再去管说要 去处理各种数据细节等等，对于你来说可能就是一个无感的事情。这边需要强调一下，就这里我们提到用RQdata也好，用MangoDB也好,都只是一种选择。 那么在我们课程中可能我们就选择这两个工具了，但是在你未来的不管你自己做交易也 好，还是去某家机构也好，其实你都有很多很多其它的选择。比如说数据服务的话，你去 了机构你可以选择更

5、贵的，比如说万得宏汇的数据，它可能更全可能更快，然后数据库方 面，你可能选择比如说有的机构喜欢用My circle，有的机构喜欢用hdf，还有一些用Ktb这样的比较专业的数据库等等，但它本质上都是一个数据库。Data Service的整个工作流程OK那么接下来就是我们中间的data service（历史数据服务）到底是怎么写的，它的整个架构大概是怎么样的，那么先讲一下本节课代码，和之前第八课第六课的时候一样，同样是有内容的，大家自己下一下，然后把它给跑起来，边跑边看效果会更好一些。首先我们来看一下我们的每天要做的事情，那么历史数据服务的更新通常是到达了某个预 定的时间，这个时间可能是比如说盘中

6、更新的话，我的每一分钟我可能每一个小时去更新 一次数据，如果是日线级别的，我可能每天到了下午收盘后某个时间点我就来更新我的数 据， 反正总之它是到达某个预定的时间才来运行的。到达某个预定时间后，我们要来启动我们下载数据的子进程，然后来执行下载的这么一个工作。那么在子进程里面我们首先会下载当日的数据，然后把远端的数据服务商的提供的这么一个数据格式，比如说它可能是个dataframe，可能是一个csv文件，或者说一些json 文件，或者说直接就是数据库的原始数据流等等，我们把它转化为本地我们自己的程序里面用的格式。因为必然，你很少说别人用的东西和你完全一样的，很多时候都是要做这么 一个转化的。转化

7、完了之后我们就要把数据更新到我们的数据库里面，那么这里我用了一个词是更新， 而不是插入。插入数据库你仅仅是把新的数据插进去，但是有的时候要么你下载的数据本 身可能就是有重复的，因为你可能下载过去250天的数据，那么每一天更新的时候，它其 实只是窗口平移的一天， 之前数据还是会给你的， 在这样一个情况下你必然不能重复插入，所以你就只能采用更新。还有一种情况可能尤其是在盘中的时候，你比如说现在你跟 你这个程序运行这个时间点碰巧某个一分钟线它刚走了一半，但是你当时去获取的时候， 也把这个没走完的分钟线给获取过来，更新到你的数据库里，过了几分钟之后你再次去获 取的时候，那个一分钟线的数据其实已经走完了

8、，也就是说它具体的数据已经发生变化了， 那么必然你数据库里面的保存的数据也要对应的给更新变化掉。那么这一步完成之后，我们就可以关闭我们的下载进程了。然后直到下一次再到达某个预定的时间，再来启动我们的下载子进程，不断的循环往复，这个就是我们历史数据服务整 体上的这么一个工作的流程。你可以选择人工去做这样一件事情，你每天到了比如说下午4点钟的时候，像我们第八节课里面提到过的，就python函数dailybarnysimple把它直接给调用一下， 然后它会自动完成这个过程，然后到第二天你再来做，这个也可以，但是无疑比较麻烦，而且也不是我们 学量化学编程的这么一个目标。所以我们会使用一个叫做“监控守护

9、进程”的东西来实现（自动化），那么这个你可以理解为另外一个独立的程序，他做的事情就是每天到了特定时间点就来去启动具体负责下载子进程，也就是说上面这三件事情可能主要在监控守护进程里面完成的，下面三件事情则是在我们的具体下载数据子进程里面去完成的。数据相关的代码结构到这里我们就可以来看看和数据相关的一些代码，那么先把这个讲完，然后我们再来看代 码。首先数据的下载我们试着通过 rq.get_price ， 我们之前已经讲过了。数据的处理是rq.get_price ，我们调用这个函数之后会获得一个pandas里面的dateframe数据结构的这么一个对象，那么大家可以把它理解为一个python内存中的

10、一个巨大的数据表，这个数据表里面它最少有两维了，一维是我们的时间序列的时间戳，另外一维比如说open、high、low等等不同的字段。那么我们要接下来要做的事情就是遍历整张数据表，然后把每一行的数据转化成我们自 己， 比如说像这里我们用的vn.py，那自然就是要把它转化成vn.py里面K线bar对象的这么 一个结构定义，所以我们这里写了一个叫generateVtBar的函数来实现。当我们转化完之后，我们就可以把它插入到数据库里面了，这里我们是使用了pymongo的，下面有collection.replace_one函数来实现。 实操部分（1）数据相关代码 （8 分钟的讲解）OK那么接下来我们就

11、来看代码 dataService.py。数据获取 df=rq.get_price()首先是我们的data service.py，第一步我们要来看get_price:df=rq.get_price(symbol,frequency=1d,fields=FIELDS, end_date=datetime.now().strftime(%Y%m%d)第一个你要传进去的参数就叫做symbol，你要获取的期货合约行情，或者说你如果买了RQdata的股票的数据权限的话，就是它股票的这么一个代码。frequence，因为我们这里是日线，所以都是one day(1D),然后如果你是要分钟，比如说一分钟就是1M

12、,60分钟小时线的话就是60M等等以此类推。然后fields是你要获取的数据字段的定义，一般情况下我们当然是只想要我们需要的这些字 段，因为不需要的字段，第一个你下载来了也没用，第二个会因为传输的数据量更大会导致你下载速度会变慢，所以说我们通常需要去告诉它我们到底要哪些字段。那么这里大写的FIELDS这个叫做常量， 一般在python里面我们常量定义都是通过大写字母来实现的。它的定义在这个顶上：FIELDS = open, high, low, close, volume我们只要open、high、low close、volume高开低收和成交量这五个字段就行了。以及我们要的我们的当天我们要下

13、载数据，最后的结束日期是什么时候，就是现在最新的 这一天。那么调用这一个函数之后，我们就能够获取这个叫做dataframe对象df。dataframe转化成需要的格式在获得df之后，我们下一步要做的是遍历它，来把每一行传到我们generateVtBar的函数里面。那么我们可以看一下：for ix, row in df.iterrows():bar= generateVtBar(row, symbol)d = bar. _dict_#bar里面对应的有很多个字段（见下方代码），其实在python里面也可以把它表示成一种字典的形式。flt =datetime:bar.datetime #过滤，保证

14、每一根K线唯一性cl.replace_one(flt, d,True)这里是forix，ix是这个行的下标；以及我们的row是每一行的这么一个对象，iterrows是迭代的意思，就是一行每一行的这样遍历的意思。那么在这样一个遍历的过程中，首先每一行里面我们都有一个叫row的这么一个对象，我们就把它传到generateVtBar这个函数上面， 然后我们可以看到我们的把它转化成了我们vn.py里面的K线bar对象，那么转化的方法在这里：def generateVtBar( row, symbol):bar = VtBarDate()bar.symbol =symbolbar.vtSymbol =s

15、ymbolbar.open =rowopenbar.high =rowhighbar.low =rowlowbar.close =rowclosebar.volume =rowvolumebar.datetime =bar.date = bar.datetime.strftime(%Y%m%d)bar.time = bar.datetime.strftime(%H%m%s)return bar首先我们创建一个vn.py里面的bar对象，barVtBardata（），然后我们把所有的字段复 制上去，比如说symbol，下载的数据代码里面是不带有这一行的合约代码数据的，把它传 进来

16、之后，然后open、high、low、close直接复制上去，这里下载的dataframe对象里面每 一行的数据本身就是数值，所以在这里我们不用做额外的任何转换，但是对于某些数据 服务它这边下载出来的可能只是一个字符串，你就需要做相应的转化了，比如说open的 话， 就等于float（），然后open，也就是说你要转化成浮点数。然后volume你可能要用一个int 把它转化成整数等等。然后date time这里的话是row的每一行的名字就是datatime， 因为 我们刚刚提到过这个表它的列的这么一个index，它就是时间序列的这么一个日期时间戳， 然后我们把日期和时间都做一个相应的转化，最后

17、我们就可以返回K线的对象。那么在这里拿到K线对象之后，我们通过访问bar._dict_的方法这个是python里面的一个叫做magic method（魔法方法）去获取bar的对象。因为我们可以看到bar里面对应的有这么多个字段，其实在python里面你也可以把它表示成一种字典的形式。比如说然后引号symbol，然后后面对应的就是symbol具体的数值等等，所有python里面的对象本质上背后都是有一个字典在做这个支撑的，那么你也可以相应的通过访问它的下划线dict来直接获取到这个字典。那么获取到这个字典之后，我们就可以把它更新到我们的数据库里面。但是更新的时候我们之前也提到，那么对于某一个这个

18、合约某一个特定的时间周期， 5分钟1小时1分钟等等，那么它的点上面K线的唯一性的确定就是通过它的时间戳来确定的， 也就是说在特定的合约特定的时间周期上，某一个特定的时间点的K线永远只应该有一根。如果后面一次我们更新一根新的进去，应该要把老的那个时间点的k线给覆盖掉，所 以在这里我们需要一个叫做flt过滤的这么一个东西，它的flt就是date我们的日期时间戳的 这么一个对象，注意这是一个对象，它同时包含了日期和时间戳，那么来保证这根K线的唯 一 性。在这里我们调这个叫做cl.replace_one：首先第一个参数是flt，就过滤条件；第二个是我们具体要去放进去的这么一个数据d；然后第三个是叫做u

19、psert，如果没有这个数据的话， 你是否要新建,那对于我们来说肯定是ture的，因为如果数据库里面没有的话，我们就插一 条新的进去，有的话就把它给更新掉。然后这样就完成了这么一个把dataframe里面每一行数据转化成我们自己需要的这么一个数据格式，并且插入到数据库里面的这么一个操作。在这里大家还可以看到这里有一个叫做CL的概念，CL就是collection，它等于数据库。你 用一个方括号里面取一个symbol就可以得到collection集合，它比较类似于我们传统circle 类数据库里面的表。cl =dbMinutesymbolcl.ensure_index(datetime, ASCE

20、NDING), unique = Ture)然后这里还有一个ensure_index来创建一个叫索引，因为我们的时间戳使用datetime来实现的，包括我们后续的，比如说你要去查询数据库里的数据，去获取过去一年两年的数据的时候，你肯定是基于时间戳，从两年前的今天到今天的数据。那么在这样一个查询的过程中，如果时间戳已经提前做好了一个类似于目录那样的东西，查询的速度必然会快很多。所以在这里我们就会做一个叫做ensure_index创建索引的这么一个操作。然后我们和数据相关的代码就讲完了，其实也是我们所有的和子进程相关的代码。那么这些代码本质上我们都已经把它封装到了一个单一的函数里面，比如说在这里我

21、们刚刚讲的例子是downloadMiniuteBarBySymbol 以及我们之前用的downloadDailyBarBySymbol ，还有这里的downloadTickBarBySymbol。你无论是下分钟线日线还是下tick 你把以上的这三个逻辑其实都可以封到一个函数里面这样对于你来说，你只要告诉它OK我就要下这个合约的行情，它就会给你把你要下的数据给拉到本地来。 数据相关代码 over 2019/2/2110【数据】历史数据服务的实操课 - 华尔街见闻监控相关的代码结构接下来我们就要看看和自动运维监控相关的代码。首先我们需要在子进程里面运行的就是downloadDailyBarBySy

22、mbol （因为我们这边课程里面用的海龟的例子使用的日线， 所以我是downloadDailyBarBySymbol， 如果你是自己其他的策略，比如说日内CTA策略，你可能是downloadMiniuteBarBySymbol或者其他的一些不同的时间结构，反正根据你自己的需求来）第二个是在父进程里面要实现三个功能：首先我要等到特定的时间再运行，我不能说我要下载日线，但是我在盘中还没收盘的时候我就去下，那肯定没有意义的。第二个我们要有一个非交易日过滤。我阿里云上买了个服务器，我在上面部署好之后把父进程一启动，然后它就永远的运行下去了，除非到某一天我上去把它停掉，否则它每天会自动的去把这个数据下载

23、到本地，它自然就是永远在运行的。但是对于我们来说， 非交易日的时候也就是说比如说周六周 肯定都是没有必要去下的。可能对于国内的法定节假日，目前我们没有特别好的判断函数直接来获取，所以我们无法判断，但至少周六周日我是可以过滤掉，那这么一个情况我们也要相应地做一个过滤。最后我们需要通过一个叫做multiprocessing多进程库的Process对象来管理，比如说创 建我们子进程里面就downloadDailyBarBySymbol这么一个子进程，以及在它完成之后把它给安全销毁这么一个操作。 实操部分（2）监控相关代码 （7 分钟的讲解） 我们还是来看监控相关的代码 run.py。首先在这个代码头

24、部的里面，把之前的dataService.py做了import，导入dataService.py里面的downloadDailyBarBySymbol和setting。补充讲解dataService中的setting：config = open(config.json)setting = jason.load(config)其中，config.json中的内容：“rqUsername”:,rqPassword:,symbols:I99,IF99, RB99, TA99大家可以看一下是我们的config文件通过jason的模块把它加载进来之后得到的一个字典。其中的具体内容上节课里面我们讲过， 主

25、要是rqUsername和rqPassword， 这是你的RQdata的用户名和 ，然后我们第三个字段也是这节课我们才加上叫做symbols，也/premium/articles/34818256/102019/2/2110【数据】历史数据服务的实操课 - 华尔街见闻就你具体要写每日更新每日自动更新的这些合约的代码是什么,我这里加上I99、IF99、RB99和TA99，也就是我们的指数的数据。子进程运行的函数在我们的runChildProcess()，就是子进程里面运行的函数就很简单了:def runChildProcess():print _*6

26、0print 开始更新日K线数据for symbol in settingsymbols:downloadDailyBarBySymbol(symbol)print日K线数据更新完毕print _*60首先上面加一个print，然后我们把setting的symbols里面symbol做一个遍历，把每个合约的行情都给更新一下，那么就这么讲，其实本质上就这两行代码，设其它print只是一个输出 而已。子进程代码很简单， 因为本质上逻辑都已经在downloadDailyBarBySymbol里面实现掉了。父进程详解runParentProcess()父进程里面就稍微复杂一点了：我们首先要有一个对于已

27、经更新过的数据日期的一个记录。updateDate =None然后我们运行的是一个while true的无限循环，那么这个无限循环除非你去点关闭按钮把它给关掉，否则在不管是windows还是linux也好，它都会一直地运行下去的。在运行的过程时候每轮检查我们等待五秒，如果你不加这个的话，它会一直疯狂的检 查，直到把你的某一个cpu的核给占满，这个是第一个是浪费算力的行为，如果你是用 自己的电脑话也会浪费电，同时也会使得你某个cpu上一直在跑，就无用的计算，那就 会导致于你的电脑有一个核就用不了了，如果你是单核电脑的话就会卡死等等。然后我们去获取一下当前的时间，我们叫datetime.now的函

28、数，以及它下面的返回的是一个data time的对象，然后我们再调用一个叫做time的函数获取当前的探对象，因为我们只关心时间不关心日期，然后我们把输出一下,告诉你当天的时间是几点.然后我们做一个简单的交易日过滤，同样的调datetime.now，但是这次我们要发挥的是一 个 叫 做 date 的 对 象 ， 然 后 我 们 还 需 要 这 个 叫 做 weekday 的 对 象 ， datetime.now.weekday，就是今天是周几，那么周几和我们自然的记录法有点区别，就 是它的记录是从零开始的，也就周一是零，周日是六这么一个情况。因为我们要过滤掉周 周日，所以我们要把五和六的这样情况

29、做一个判断。如果今天weekday是在五和六这两个数字之中，不管是哪一个，我们都做一个contiune，就是进入下一轮循环，我没有必要去做后面的判断，我直接就去等于就过滤掉。然后我们来看一下，因为我只希望在每天下午5点后才下载数据。因为每天正常中国期货市场最迟收盘的是我们的国债期货3:15收盘后，数据服务商它一般需要一定的时间来 清洗数据，然后才能把这个清洗好数据入库做到当天数据库里面。所以我们会一般选择收盘后过一个小时，或者你有一些数据服务上最少过半小时，一般过一个多小时保险一点，这里我们就到五点之后再做这个事情，如果没到五点我同样continue进入下的循 环，不做其他事情。大家注意到每进

30、入到下一轮循环里面第一件事情就会等待五秒，在等待过程中，尽管五秒对于人来说很短，但是对于CPU来说已经非常长了，你的电脑有足够的时间去做其他事情，也就几乎是不卡的，对于你这个CPU的运算几乎是没有影响。/premium/articles/34818257/102019/2/2110【数据】历史数据服务的实操课 - 华尔街见闻然后我们来看一下，如果说我们的update date就已经更新数据的日期和等于today的话， 就等于今天，就我今天已经更新过了，同样我也没有必要重复更新了。那么同样是 一个continue。也就是说这边三个continue就做

31、完了我们所有在下载数据之前要做的这么一个过滤的操作。过滤都过滤完了，如果剩下的情况就是我真正要下数据的时候，在这个时候我就要去创建 这个 multiprocessing.Process 创 建 一 个 子 进 程 对 象 ， 它 唯 一 的 一 个 入 参 就 是target=runChildProcess，运行子进程的这么一个函数。那么这样我们就创建了一个我们 的子进程的对象。然后我们就用这个子进程的start的函数就是启动它，然后我们输出一下叫做启动子进程， 然后我们再调一个p.join，join就是等待子进程的结束。如果没有结束的话，这句函数会一 直卡在这，直到它结束了之后才会到下一条里

32、面叫做去输出子进程已关闭这么一个情况。同时最后我们也会在真正把数据更新做完了之后，我们把当天的日期更新到updateDate的 缓存的对象上面， 那么这样在下一轮再运行到这里的时候， 因为updateDate等于today， 因为我已经更新过了，所以它不会重复的去更新。也就实现了一个叫做每天只更新一次的 控制。那么在任何时候这也是一个写程序的一个比较好的习惯，我们都要想到没有必要的 话尽量不要去浪费我们计算机的资源，没有必要的话应该尽可能的让我们计算机少跑无谓 的运行。养成这个习惯之后，你会发现你写出来程序的整洁度以及它的有效程度会好很多。最后在这里我们需要在这个叫if_name_=“ mai

33、n ”，这个是python的标准的，你要运行某个文件的主函数写法，这里你要运行一个叫RunParentProcess启动守护进程。 监控相关代码 over 我们也是简单的来看一下运行的效果。因为我在录制的时候现在是中午12点，所以我把这里的时间稍微改一下.先给大家看一下不改的效果，大家可以看到这边正式第一次输出，第一轮运行当天时间是12:11:54，然后你可以看到下一次就是59秒，再下次是04秒等等，它反正每五秒钟检查一次。那么因为没到下午5点钟，所以我们这边先把它给解除掉，按ctlc就可以把这个程序给结束掉，如果不按的话，这个程序会永远地运行下去。然后在这里我们把时间改一下，如果我们就改了1

34、2点，因为我希望它立即运行一下，这个就交给大家自己去运行，跑一跑效果。Q：那么有些人可能会问说为什么要采用这么一个父子进程的模式，我直接在同样一个进 程内写一个while循环做检查不行吗？A：那么这里的主要原因是因为下数据也好，还是说另外一种情况，你可能自己连CDP去 录制数据的时候，你都有可能会触发各种各样的异常，那么因为考虑到我们数据服务是一 旦我把它运行起来，我希望它能够一直跑下去的，我肯定不希望说因为某个小的异常它就 终止运行了，所以我们采用子进程父子进程这么一个模式，把核心逻辑放到子进程里面去 跑， 那么对于我们的程序的稳健性可靠性来说会更高一些。 第二次课后作业 OK那么从上一次作业到现在，我们已经看过了我们海龟策略的简单的回测、如何去获取 数据，那么回测尽管你可能现在还不知道海龟策略具体代码的原理，不管怎么样你已经跑 过了。那么我们接下来就要来做我们的第二次的课后策略了。/premium/articles/348182510/10你要做的就是把上一节课第09课里面我们做的最后的两组不同的数据，一组是合约指 数，一组是平滑处理后的连续价格合约，那么这两者做一个回测对比。那么合约指数是99结尾的，