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:
  #接收任務(wù)，如果連續(xù)20秒沒有新的任務(wù)，線程退出，否則會一直執(zhí)行
  item=self.producerQueue.get(block=True, timeout=20)
  #阻塞調(diào)用進(jìn)程直到有數(shù)據(jù)可用。如果timeout是個(gè)正整數(shù)，
  #阻塞調(diào)用進(jìn)程最多timeout秒，
  #如果一直無數(shù)據(jù)可用，拋出Empty異常（帶超時(shí)的阻塞調(diào)用）
  except Queue.Empty:
  print("Nothing to do!thread exit!")
  self.thread_stop=True
  break
  #實(shí)現(xiàn)生產(chǎn)者邏輯，生成消費(fèi)者需要處理的內(nèi)容 consumerQueue.put(someItem)
  #還可以邊處理，邊生成新的生產(chǎn)任務(wù)
  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:
  #接收任務(wù)，如果連續(xù)20秒沒有新的任務(wù)，線程退出，否則會一直執(zhí)行
  item=self.consumerQueue.get(block=True, timeout=20)
  #阻塞調(diào)用進(jìn)程直到有數(shù)據(jù)可用。如果timeout是個(gè)正整數(shù)，
  #阻塞調(diào)用進(jìn)程最多timeout秒，
  #如果一直無數(shù)據(jù)可用，拋出Empty異常（帶超時(shí)的阻塞調(diào)用）
  except Queue.Empty:
  print("Nothing to do!thread exit!")
  self.thread_stop=True
  break
  doSomethingAboutConsuming(lock)# 處理消費(fèi)者邏輯,必要時(shí)使用線程鎖 ，如文件操作等
  self.consumerQueue.task_done()
#定義主線程
def main():
 for i in range(n):#定義n個(gè)i消費(fèi)者線程
 t = ThreadRead(producerQueue, consumerQueue)
 t.setDaemon(True)
 t.start()
 producerTasks=[] #定義初始化生產(chǎn)者任務(wù)隊(duì)列
 producerQueue.put(producerTasks)
 for i in range(n):#定義n個(gè)生產(chǎn)者錢程
 t = ThreadWrite(consumerQueue, lock)
 t.setDaemon(True)
 t.start() 
 stock_queue.join()
 data_queue.join()

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ù)

import tushare as ts
ts.get_stock_basics()

  name industry area  pe outstanding  totals totalAssets
code
600606 金豐投資  房產(chǎn)服務(wù) 上海  0.00  51832.01 51832.01 744930.44
002285 世聯(lián)行  房產(chǎn)服務(wù) 深圳 71.04  76352.17 76377.60 411595.28
000861 海印股份  房產(chǎn)服務(wù) 廣東 126.20  83775.50 118413.84 730716.56
000526 銀潤投資  房產(chǎn)服務(wù) 福建 2421.16  9619.50 9619.50  20065.32
000056 深國商  房產(chǎn)服務(wù) 深圳  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

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

parameter：
code:string,股票代碼 e.g. 600848
start:string,開始日期 format：YYYY-MM-DD 為空時(shí)取當(dāng)前日期
end:string,結(jié)束日期 format：YYYY-MM-DD 為空時(shí)取去年今日
autype:string,復(fù)權(quán)類型，qfq-前復(fù)權(quán) hfq-后復(fù)權(quán) None-不復(fù)權(quán)，默認(rèn)為qfq
index:Boolean，是否是大盤指數(shù)，默認(rèn)為False
retry_count : int, 默認(rèn)3,如遇網(wǎng)絡(luò)等問題重復(fù)執(zhí)行的次數(shù)
pause : int, 默認(rèn) 0,重復(fù)請求數(shù)據(jù)過程中暫停的秒數(shù)，防止請求間隔時(shí)間太短出現(xiàn)的問題

return：
date : 交易日期 (index)
open : 開盤價(jià)
high : 最高價(jià)
close : 收盤價(jià)
low : 最低價(jià)
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):
  '''
  用于根據(jù)股票代碼、需要讀取的日期，讀取增量的日行情數(shù)據(jù)，
  :param queue:用于保存需要讀取的股票代碼、起始日期的列表
  :param out_queue:用于保存需要寫入到數(shù)據(jù)庫表的結(jié)果集列表
  :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ù)放回隊(duì)列里面去，以便下次重試。
    time.sleep(10)
    continue
   self.queue.task_done()

class ThreadWrite(threading.Thread):
 def __init__(self, queue, lock, db_engine):
  '''
  :param queue: 某種形式的任務(wù)隊(duì)列，此處為tushare為每個(gè)股票返回的最新日復(fù)權(quán)行情數(shù)據(jù)
  :param lock: 暫時(shí)用連接互斥操作，防止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)境變量，取得相應(yīng)的環(huán)境配置，上線時(shí)不需要再變更代碼
 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()報(bào)錯(cuò)的處理，如果取不到信息則直接用數(shù)據(jù)庫中的股票代碼信息，來獲取增量信息
 #TODO 增加一個(gè)標(biāo)志，如果一個(gè)股票代碼的最新日期不是最新日期，則需標(biāo)記該代碼不需要重新獲取數(shù)據(jù)，即記錄該股票更新日期到了最新工作日，
 df = ts.get_stock_basics()
 df.to_sql('stock_basics',db_engine,if_exists='replace',dtype={'code': CHAR(6)})
 # 計(jì)算距離當(dāng)前日期最大的工作日，以便每日定時(shí)更新
 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()
 # 計(jì)算每只股票當(dāng)前加載的最大工作日期，支持重跑
 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)
 # 開始?xì)w檔前復(fù)權(quán)歷史行情至數(shù)據(jù)庫當(dāng)中，以便可以方便地計(jì)算后續(xù)選股模型

 for i in range(3):#使用3個(gè)線程
  t = ThreadRead(stock_queue, data_queue)
  t.setDaemon(True)
  t.start()
 for code in set(list(df.index)):
  try:
   #如果當(dāng)前股票已經(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:
   #如果某只股票沒有相關(guān)的行情，則默認(rèn)開始日期為2015年1月1日
   startdate='2015-01-01'
  item={}
  item['code']=code
  item['startdate']=startdate
  stock_queue.put(item) # 生成生產(chǎn)者任務(wù)隊(duì)列
 for i in range(3):
  t = ThreadWrite(data_queue, lock, db_engine)
  t.setDaemon(True)
  t.start()
 stock_queue.join()
 data_queue.join()