思路

借助python當中threading模塊與Queue模塊組合可以方便的實現(xiàn)基于生產(chǎn)者-消費者模型的多線程模型佑女。Jimmy大神的tushare一直是廣大python數(shù)據(jù)分析以及業(yè)余量化愛好者喜愛的免費、開源的python財經(jīng)數(shù)據(jù)接口包也糊。

平時一直有在用阿里云服務器通過tushare的接口自動落地相關財經(jīng)數(shù)據(jù)着降，但日復權行情數(shù)據(jù)以往在串行下載的過程當中辜荠，速度比較慢箍鼓，有時遇到網(wǎng)絡原因還需要重下樊销。每只股票的行情下載過程中都需要完成下載饵蒂、落地2個步驟声诸，一個可能需要網(wǎng)絡開銷、一個需要數(shù)據(jù)庫mysql的存取開銷退盯。2者原本就可以獨立并行執(zhí)行彼乌，是個典型的“生產(chǎn)者-消費者”模型泻肯。

基于queue與threading模塊的線程使用一般采用以下的套路：

producerQueue=Queue()
consumerQueue=Queue()
lock = threading.Lock()
class producerThead(threading.Thread):
    def __init__(self, producerQueue,consumerQueue):
        self.producerQueue=producerQueue
        self.consumerQueue=consumerQueue



    def run(self):
        while not self.thread_stop:
            try:
                #接收任務，如果連續(xù)20秒沒有新的任務慰照，線程退出灶挟，否則會一直執(zhí)行
                item=self.producerQueue.get(block=True, timeout=20)
                #阻塞調(diào)用進程直到有數(shù)據(jù)可用。如果timeout是個正整數(shù)毒租，
                #阻塞調(diào)用進程最多timeout秒稚铣，
                #如果一直無數(shù)據(jù)可用，拋出Empty異常（帶超時的阻塞調(diào)用）
            except Queue.Empty:
                print("Nothing to do!thread exit!")
                self.thread_stop=True
                break
            #實現(xiàn)生產(chǎn)者邏輯墅垮，生成消費者需要處理的內(nèi)容 consumerQueue.put(someItem)
            #還可以邊處理惕医，邊生成新的生產(chǎn)任務
            doSomethingAboutProducing()
            self.producerQueue.task_done()
    def stop(self):
        self.thread_stop = True

class consumerThead(threading.Thread):
    def __init__(self,lock, consumerQueue):
        self.consumerQueue=consumerQueue
    def run(self):
        while true:
            try:
                #接收任務，如果連續(xù)20秒沒有新的任務算色，線程退出抬伺，否則會一直執(zhí)行
                item=self.consumerQueue.get(block=True, timeout=20)
                #阻塞調(diào)用進程直到有數(shù)據(jù)可用。如果timeout是個正整數(shù)灾梦，
                #阻塞調(diào)用進程最多timeout秒峡钓，
                #如果一直無數(shù)據(jù)可用，拋出Empty異常（帶超時的阻塞調(diào)用）
            except Queue.Empty:
                print("Nothing to do!thread exit!")
                self.thread_stop=True
                break
            doSomethingAboutConsuming(lock)# 處理消費者邏輯,必要時使用線程鎖 若河，如文件操作等
            self.consumerQueue.task_done()
#定義主線程
def main():
    for i in range(n):#定義n個i消費者線程
        t = ThreadRead(producerQueue, consumerQueue)
        t.setDaemon(True)
        t.start()
    producerTasks=[] #定義初始化生產(chǎn)者任務隊列
    producerQueue.put(producerTasks)
    for i in range(n):#定義n個生產(chǎn)者錢程
        t = ThreadWrite(consumerQueue, lock)
        t.setDaemon(True)
        t.start()    
    stock_queue.join()
    data_queue.join()

相關接口

1,股票列表信息接口

• 作用
  獲取滬深上市公司基本情況能岩。屬性包括：

code,代碼
name,名稱
industry,所屬行業(yè)
area,地區(qū)
pe,市盈率
outstanding,流通股本(億)
totals,總股本(億)
totalAssets,總資產(chǎn)(萬)
liquidAssets,流動資產(chǎn)
fixedAssets,固定資產(chǎn)
reserved,公積金
reservedPerShare,每股公積金
esp,每股收益
bvps,每股凈資
pb,市凈率
timeToMarket,上市日期
undp,未分利潤
perundp, 每股未分配
rev,收入同比(%)
profit,利潤同比(%)
gpr,毛利率(%)
npr,凈利潤率(%)
holders,股東人數(shù)

• 調(diào)用方法

import tushare as ts
ts.get_stock_basics()

• 返回效果

        name    industry    area       pe   outstanding     totals  totalAssets
code
600606   金豐投資     房產(chǎn)服務   上海     0.00     51832.01   51832.01    744930.44
002285    世聯(lián)行     房產(chǎn)服務   深圳    71.04     76352.17   76377.60    411595.28
000861   海印股份     房產(chǎn)服務   廣東   126.20     83775.50  118413.84    730716.56
000526   銀潤投資     房產(chǎn)服務   福建  2421.16      9619.50    9619.50     20065.32
000056    深國商     房產(chǎn)服務   深圳     0.00     14305.55   26508.14    787195.94
600895   張江高科     園區(qū)開發(fā)   上海   171.60    154868.95  154868.95   1771040.38
600736   蘇州高新     園區(qū)開發(fā)   江蘇    48.68    105788.15  105788.15   2125485.75
600663    陸家嘴     園區(qū)開發(fā)   上海    47.63    135808.41  186768.41   4562074.50
600658    電子城     園區(qū)開發(fā)   北京    19.39     58009.73   58009.73    431300.19
600648    外高橋     園區(qū)開發(fā)   上海    65.36     81022.34  113534.90   2508100.75
600639   浦東金橋     園區(qū)開發(fā)   上海    57.28     65664.88   92882.50   1241577.00
600604   市北高新     園區(qū)開發(fā)   上海   692.87     33352.42   56644.92    329289.50

2,日復權行情接口

• 作用
  提供股票上市以來所有歷史數(shù)據(jù)，默認為前復權萧福，讀取后存到本地拉鹃，作為后續(xù)分析的基礎
• 調(diào)用方法

ts.get_h_data('002337', start='2015-01-01', end='2015-03-16') #兩個日期之間的前復權數(shù)據(jù)

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毛俏，是否是大盤指數(shù)炭庙，默認為False
retry_count : int, 默認3,如遇網(wǎng)絡等問題重復執(zhí)行的次數(shù)
pause : int, 默認 0,重復請求數(shù)據(jù)過程中暫停的秒數(shù)饲窿，防止請求間隔時間太短出現(xiàn)的問題

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

實現(xiàn)

廢話不多說，直接上代碼焕蹄，

• 生產(chǎn)者線程逾雄，讀取行情

class ThreadRead(threading.Thread):
    def __init__(self, queue, out_queue):
        '''
        用于根據(jù)股票代碼、需要讀取的日期腻脏，讀取增量的日行情數(shù)據(jù)鸦泳，
        :param queue:用于保存需要讀取的股票代碼、起始日期的列表
        :param out_queue:用于保存需要寫入到數(shù)據(jù)庫表的結果集列表
        :return:
        '''
        threading.Thread.__init__(self)
        self.queue = queue
        self.out_queue = out_queue
    def 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) # 將沒有爬取成功的數(shù)據(jù)放回隊列里面去永品，以便下次重試做鹰。
                time.sleep(10)
                continue

            self.queue.task_done()

• 消費者線程，本地存儲

class ThreadWrite(threading.Thread):
    def __init__(self, queue, lock, db_engine):
        '''
        :param queue: 某種形式的任務隊列鼎姐，此處為tushare為每個股票返回的最新日復權行情數(shù)據(jù)
        :param lock:  暫時用連接互斥操作钾麸，防止mysql高并發(fā)更振，后續(xù)可嘗試去掉
        :param db_engine:  mysql數(shù)據(jù)庫的連接對象
        :return:no
        '''
        threading.Thread.__init__(self)
        self.queue = queue
        self.lock = lock
        self.db_engine = db_engine

    def 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():
    '''
    用于測試多線程讀取數(shù)據(jù)
    :return:
    '''
    #獲取環(huán)境變量肯腕，取得相應的環(huán)境配置，上線時不需要再變更代碼
    global stock_queue
    global data_queue
    config=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()報錯的處理钥平，如果取不到信息則直接用數(shù)據(jù)庫中的股票代碼信息实撒，來獲取增量信息
    #TODO 增加一個標志，如果一個股票代碼的最新日期不是最新日期涉瘾，則需標記該代碼不需要重新獲取數(shù)據(jù)知态，即記錄該股票更新日期到了最新工作日，
    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)  # 執(zhí)行查詢語句
    df_result = pd.DataFrame(result.fetchall())
    df_result.columns=['stockcode','max_tradeday']
    df_result.set_index(df_result['stockcode'],inplace=True)
    # 開始歸檔前復權歷史行情至數(shù)據(jù)庫當中，以便可以方便地計算后續(xù)選股模型

    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:
            #如果當前股票已經(jīng)是最新的行情數(shù)據(jù)囚巴，則直接跳過,方便重跑原在。
            #print maxTradeay[0],df_result.loc[code].values[1]
            if df_result.loc[code].values[1] == maxTradeay[0]:
                continue
            startdate=getLastNdate(df_result.loc[code].values[1],1)
        except Exception, e:
            #如果某只股票沒有相關的行情，則默認開始日期為2015年1月1日
            startdate='2015-01-01'
        item={}
        item['code']=code
        item['startdate']=startdate
        stock_queue.put(item) # 生成生產(chǎn)者任務隊列
    for i in range(3):
        t = ThreadWrite(data_queue, lock, db_engine)
        t.setDaemon(True)
        t.start()
    stock_queue.join()
    data_queue.join()

• 執(zhí)行效果
  原本需要2彤叉，3個小時才能執(zhí)行完成的每日復權行情增量落地庶柿，有效縮短至了1小時以內(nèi)，這里線程數(shù)并不上越多越好秽浇，由于復權行情讀的是新浪接口浮庐，在高并發(fā)情況下會返回HTTP 503服務器過載的錯誤，另外高并發(fā)下可能需要使用IP代理池柬焕，下載的時段也需要嘗試多個時段進行审残。初次嘗試，如果有更好的方法或者哪里有考慮不周的地方歡迎留言建議或者指正斑举。