PYTHON多线程行情抓取工具实现
思路
借助python当中threading模块与Queue模块组合可以方便的实现基于生产者-消费者模型的多线程模型。Jimmy大神的tushare一直是广大python数据分析以及业余量化爱好者喜爱的免费、开源的python财经数据接口包。
平时一直有在用阿里云服务器通过tushare的接口自动落地相关财经数据,但日复权行情数据以往在串行下载的过程当中,速度比较慢,有时遇到网络原因还需要重下。每只股票的行情下载过程中都需要完成下载、落地2个步骤,一个可能需要网络开销、一个需要数据库mysql的存取开销。2者原本就可以独立并行执行,是个典型的“生产者-消费者”模型。
基于queue与threading模块的线程使用一般采用以下的套路:
producerQueue=Queue()
consumerQueue=Queue()
lock = threading.Lock()
class producerThead(threading.Thread):def __init__(self, producerQueue,consumerQueue):self.producerQueue=producerQueueself.consumerQueue=consumerQueuedef run(self):while not self.thread_stop:try:#接收任务,如果连续20秒没有新的任务,线程退出,否则会一直执行item=self.producerQueue.get(block=True, timeout=20)#阻塞调用进程直到有数据可用。如果timeout是个正整数,#阻塞调用进程最多timeout秒,#如果一直无数据可用,抛出Empty异常(带超时的阻塞调用)except Queue.Empty:print("Nothing to do!thread exit!")self.thread_stop=Truebreak#实现生产者逻辑,生成消费者需要处理的内容 consumerQueue.put(someItem)#还可以边处理,边生成新的生产任务doSomethingAboutProducing()self.producerQueue.task_done()def stop(self):self.thread_stop = Trueclass consumerThead(threading.Thread):def __init__(self,lock, consumerQueue):self.consumerQueue=consumerQueuedef run(self):while true:try:#接收任务,如果连续20秒没有新的任务,线程退出,否则会一直执行item=self.consumerQueue.get(block=True, timeout=20)#阻塞调用进程直到有数据可用。如果timeout是个正整数,#阻塞调用进程最多timeout秒,#如果一直无数据可用,抛出Empty异常(带超时的阻塞调用)except Queue.Empty:print("Nothing to do!thread exit!")self.thread_stop=TruebreakdoSomethingAboutConsuming(lock)# 处理消费者逻辑,必要时使用线程锁 ,如文件操作等self.consumerQueue.task_done()
#定义主线程
def main():for i in range(n):#定义n个i消费者线程t = ThreadRead(producerQueue, consumerQueue)t.setDaemon(True)t.start()producerTasks=[] #定义初始化生产者任务队列producerQueue.put(producerTasks)for i in range(n):#定义n个生产者钱程t = ThreadWrite(consumerQueue, lock)t.setDaemon(True)t.start() stock_queue.join()data_queue.join()
相关接口
1,股票列表信息接口
作用
获取沪深上市公司基本情况。属性包括:code,代码 name,名称 industry,所属行业 area,地区 pe,市盈率 outstanding,流通股本(亿) totals,总股本(亿) totalAssets,总资产(万) liquidAssets,流动资产 fixedAssets,固定资产 reserved,公积金 reservedPerShare,每股公积金 esp,每股收益 bvps,每股净资 pb,市净率 timeToMarket,上市日期 undp,未分利润 perundp, 每股未分配 rev,收入同比(%) profit,利润同比(%) gpr,毛利率(%) npr,净利润率(%) holders,股东人数
- 调用方法
import tushare as ts
ts.get_stock_basics()
- 返回效果
name industry area pe outstanding totals totalAssets
code
600606 金丰投资 房产服务 上海 0.00 51832.01 51832.01 744930.44
002285 世联行 房产服务 深圳 71.04 76352.17 76377.60 411595.28
000861 海印股份 房产服务 广东 126.20 83775.50 118413.84 730716.56
000526 银润投资 房产服务 福建 2421.16 9619.50 9619.50 20065.32
000056 深国商 房产服务 深圳 0.00 14305.55 26508.14 787195.94
600895 张江高科 园区开发 上海 171.60 154868.95 154868.95 1771040.38
600736 苏州高新 园区开发 江苏 48.68 105788.15 105788.15 2125485.75
600663 陆家嘴 园区开发 上海 47.63 135808.41 186768.41 4562074.50
600658 电子城 园区开发 北京 19.39 58009.73 58009.73 431300.19
600648 外高桥 园区开发 上海 65.36 81022.34 113534.90 2508100.75
600639 浦东金桥 园区开发 上海 57.28 65664.88 92882.50 1241577.00
600604 市北高新 园区开发 上海 692.87 33352.42 56644.92 329289.50
2,日复权行情接口
- 作用
提供股票上市以来所有历史数据,默认为前复权,读取后存到本地,作为后续分析的基础
- 调用方法
ts.get_h_data('002337', start='2015-01-01', end='2015-03-16') #两个日期之间的前复权数据parameter:
code:string,股票代码 e.g. 600848
start:string,开始日期 format:YYYY-MM-DD 为空时取当前日期
end:string,结束日期 format:YYYY-MM-DD 为空时取去年今日
autype:string,复权类型,qfq-前复权 hfq-后复权 None-不复权,默认为qfq
index:Boolean,是否是大盘指数,默认为False
retry_count : int, 默认3,如遇网络等问题重复执行的次数
pause : int, 默认 0,重复请求数据过程中暂停的秒数,防止请求间隔时间太短出现的问题return:
date : 交易日期 (index)
open : 开盘价
high : 最高价
close : 收盘价
low : 最低价
volume : 成交量
amount : 成交金额
- 返回结果
open high close low volume amount
date
2015-03-16 13.27 13.45 13.39 13.00 81212976 1073862784
2015-03-13 13.04 13.38 13.37 13.00 40548836 532739744
2015-03-12 13.29 13.95 13.28 12.96 71505720 962979904
2015-03-11 13.35 13.48 13.15 13.00 59110248 780300736
2015-03-10 13.16 13.67 13.59 12.72 105753088 1393819776
2015-03-09 13.77 14.73 14.13 13.70 139091552 1994454656
2015-03-06 12.17 13.39 13.39 12.17 89486704 1167752960
2015-03-05 12.79 12.80 12.17 12.08 26040832 966927360
2015-03-04 13.96 13.96 13.30 12.58 26636174 1060270720
2015-03-03 12.17 13.10 13.10 12.05 19290366 733336768
实现
废话不多说,直接上代码,
- 生产者线程,读取行情
class ThreadRead(threading.Thread):def __init__(self, queue, out_queue):'''用于根据股票代码、需要读取的日期,读取增量的日行情数据,:param queue:用于保存需要读取的股票代码、起始日期的列表:param out_queue:用于保存需要写入到数据库表的结果集列表:return:'''threading.Thread.__init__(self)self.queue = queueself.out_queue = out_queuedef run(self):while true:item = self.queue.get()time.sleep(0.5)try:df_h_data = ts.get_h_data(item['code'], start=item['startdate'], retry_count=10, pause=0.01)if df_h_data is not None and len(df_h_data)>0:df_h_data['secucode'] = item['code']df_h_data.index.name = 'date'print df_h_data.index,item['code'],item['startdate']df_h_data['tradeday'] = df_h_data.index.strftime('%Y-%m-%d')self.out_queue.put(df_h_data)except Exception, e:print str(e)self.queue.put(item) # 将没有爬取成功的数据放回队列里面去,以便下次重试。time.sleep(10)continueself.queue.task_done()
- 消费者线程,本地存储
class ThreadWrite(threading.Thread):def __init__(self, queue, lock, db_engine):''':param queue: 某种形式的任务队列,此处为tushare为每个股票返回的最新日复权行情数据:param lock: 暂时用连接互斥操作,防止mysql高并发,后续可尝试去掉:param db_engine: mysql数据库的连接对象:return:no'''threading.Thread.__init__(self)self.queue = queueself.lock = lockself.db_engine = db_enginedef run(self):while True:item = self.queue.get()self._save_data(item)self.queue.task_done()def _save_data(self, item):with self.lock:try:item.to_sql('cron_dailyquote', self.db_engine, if_exists='append', index=False)except Exception, e: # 如果是新股,则有可能df_h_data是空对象,因此需要跳过此类情况不处理print str(e)
- 定义主线程
from Queue import Queue
stock_queue = Queue()
data_queue = Queue()
lock = threading.Lock()
def main():'''用于测试多线程读取数据:return:'''#获取环境变量,取得相应的环境配置,上线时不需要再变更代码global stock_queueglobal data_queueconfig=os.getenv('FLASK_CONFIG')if config == 'default':db_url='mysql+pymysql://root:******@localhost:3306/python?charset=utf8mb4'else:db_url='mysql+pymysql://root:******@localhost:3306/test?charset=utf8mb4'db_engine = create_engine(db_url, echo=True)conn = db_engine.connect()#TODO 增加ts.get_stock_basics()报错的处理,如果取不到信息则直接用数据库中的股票代码信息,来获取增量信息#TODO 增加一个标志,如果一个股票代码的最新日期不是最新日期,则需标记该代码不需要重新获取数据,即记录该股票更新日期到了最新工作日,df = ts.get_stock_basics()df.to_sql('stock_basics',db_engine,if_exists='replace',dtype={'code': CHAR(6)})# 计算距离当前日期最大的工作日,以便每日定时更新today=time.strftime('%Y-%m-%d',time.localtime(time.time()))s1=("select max(t.date) from cron_tradeday t where flag=1 and t.date <='"+ today+"'")selectsql=text(s1)maxTradeay = conn.execute(selectsql).first()# 计算每只股票当前加载的最大工作日期,支持重跑s = ("select secucode,max(t.tradeday) from cron_dailyquote t group by secucode ")selectsql = text(s)result = conn.execute(selectsql) # 执行查询语句df_result = pd.DataFrame(result.fetchall())df_result.columns=['stockcode','max_tradeday']df_result.set_index(df_result['stockcode'],inplace=True)# 开始归档前复权历史行情至数据库当中,以便可以方便地计算后续选股模型for i in range(3):#使用3个线程t = ThreadRead(stock_queue, data_queue)t.setDaemon(True)t.start()for code in set(list(df.index)):try:#如果当前股票已经是最新的行情数据,则直接跳过,方便重跑。#print maxTradeay[0],df_result.loc[code].values[1]if df_result.loc[code].values[1] == maxTradeay[0]:continuestartdate=getLastNdate(df_result.loc[code].values[1],1)except Exception, e:#如果某只股票没有相关的行情,则默认开始日期为2015年1月1日startdate='2015-01-01'item={}item['code']=codeitem['startdate']=startdatestock_queue.put(item) # 生成生产者任务队列for i in range(3):t = ThreadWrite(data_queue, lock, db_engine)t.setDaemon(True)t.start()stock_queue.join()data_queue.join()
- 执行效果
原本需要2,3个小时才能执行完成的每日复权行情增量落地,有效缩短至了1小时以内,这里线程数并不上越多越好,由于复权行情读的是新浪接口,在高并发情况下会返回HTTP 503服务器过载的错误,另外高并发下可能需要使用IP代理池,下载的时段也需要尝试多个时段进行。初次尝试,如果有更好的方法或者哪里有考虑不周的地方欢迎留言建议或者指正。
PYTHON多线程行情抓取工具实现相关推荐
- 脱贫利器 | PYTHON多线程行情抓取工具实现
思路 借助python当中threading模块与Queue模块组合可以方便的实现基于生产者-消费者模型的多线程模型.Jimmy大神的tushare一直是广大python数据分析以及业余量化爱好者喜爱 ...
- python爬取百度贴吧中的所有邮箱_使用 Python 编写多线程爬虫抓取百度贴吧邮箱与手机号...
原标题:使用 Python 编写多线程爬虫抓取百度贴吧邮箱与手机号 不知道大家过年都是怎么过的,反正栏主是在家睡了一天,醒来的时候登QQ发现有人找我要一份贴吧爬虫的源代码,想起之前练手的时候写过一个抓 ...
- 转一个python写的多线程 代理服务器 抓取,保存,验证程序
用php写过一个,不过由于 php 不支持多线程,抓取和验证速度都非常的慢 (尽管libcurl可以实现多线程抓取,但他也只限于抓取网页这个功能,抓回来的数据进行再处理很麻烦). 于是决定用pytho ...
- python爬网页数据用什么_初学者如何用“python爬虫”技术抓取网页数据?
原标题:初学者如何用"python爬虫"技术抓取网页数据? 在当今社会,互联网上充斥着许多有用的数据.我们只需要耐心观察并添加一些技术手段即可获得大量有价值的数据.而这里的&quo ...
- Python爬虫:抓取智联招聘岗位信息和要求(进阶版)
本文的文字及图片来源于网络,仅供学习.交流使用,不具有任何商业用途,版权归原作者所有,如有问题请及时联系我们以作处理 以下文章来源于腾讯云 作者:王强 ( 想要学习Python?Python学习交流群 ...
- 独家 | 手把手教你用Python进行Web抓取(附代码)
作者:Kerry Parker 翻译:田晓宁 校对:丁楠雅 本文约2900字,建议阅读10分钟. 本教程以在Fast Track上收集百强公司的数据为例,教你抓取网页信息. 作为一名数据科学家,我在工 ...
- python爬电影_使用Python多线程爬虫爬取电影天堂资源
最近花些时间学习了一下Python,并写了一个多线程的爬虫程序来获取电影天堂上资源的迅雷下载地址,代码已经上传到GitHub上了,需要的同学可以自行下载.刚开始学习python希望可以获得宝贵的意见. ...
- python lxml使用_使用lxml和Python进行Web抓取的简介
python lxml使用 by Timber.io 由Timber.io 使用lxml和Python进行Web抓取的简介 (An Intro to Web Scraping with lxml an ...
- hacker代码_如何仅用7行R代码构建Hacker News Frontpage抓取工具
hacker代码 by AMR 通过AMR 如何仅用7行R代码构建Hacker News Frontpage抓取工具 (How to build a Hacker News Frontpage scr ...
最新文章
- 33.搜索插件——autocomplete
- page和其它几个内置对象(转)
- 完全理解NIO Selector
- lua--面向对象使用middleclass
- c# 四舍五入、上取整、下取整
- LoRaWAN的四大优势及适用领域
- 前端框架开始学习Vue(一)
- buck变换器设计matlab_在数字控制系统中实现LLC变换器的电流模式控制的思考
- 模块间相互调用需要注意的问题
- 求解汉诺塔问题(提示, 使用递归)
- python制作病毒_python做病毒
- 吐血推荐:无解的完成图
- python入门书籍推荐,看这篇文章就够,请!
- python自动投递_利用python如何在前程无忧高效投递简历
- mapi java_使用协议“ mapi://”从Java在Outlook中打开邮件
- face_recognition人脸识别框架
- Latex排版 Chapter1基本使用(文档结构插入表格、图片)
- 2013年IT审计师认证的详细资料
- 无法超越,1994 电影史上的巅峰
- [MySQL]复杂查询(进阶)