# -*- encode:utf-8 -*- # 上期所 import time import re import datetime import requests import pymongo import settings import threading import pandas as pd from queue import Queue, Empty from log import Logger log = Logger('logs/shfe.log') class CrawlData(threading.Thread): """爬取数据类""" def __init__(self, q): super(CrawlData, self).__init__() # 数据存储队列 self.q = q # url self.url = settings.API['shfe'] # 请求头 self.headers = { 'Host': 'www.shfe.com.cn', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.103 Safari/537.36', 'Referer': 'http://www.shfe.com.cn/statements/dataview.html?paramid=delaymarket_all', } # 连接失败重试次数 self.retry = 3 def run(self): # 获取数据库最新的一条时间 last_time = self.get_last_time() last_time = last_time.date() today = datetime.date.today() while last_time <= today: # 排除周六日情况 if last_time.isoweekday != 6 and last_time.isoweekday != 7: # 请求url url = self.url % last_time.strftime('%Y%m%d') # 超时次数 time_out = 0 while time_out < self.retry: try: # log.logger.debug('开始爬取 %s' % url) response = requests.get(url) except Exception as e: log.logger.warning('连接失败, 错误内容: %s, url: %s' % (e, url), exc_info=True) time_out += 1 continue # 如果是404，说明当天没有数据 if response.status_code == 404: break try: # 把数据转换成json data = response.json() except Exception as e: log.logger.error('数据转成json失败, 错误内容: %s, url: %s' % (e, url), exc_info=True) time_out += 1 continue # 如果数据没有report_date，则添加一个时间 data.setdefault('report_date', last_time.strftime('%Y%m%d')) # 如果是有数据的 if data['o_cursor']: # 把数据放进队列里 self.q.put(data) break # 如果超时次数等于设定的重试次数，说明没有成功爬取到数据 if time_out == self.retry: log.logger.error('获取数据失败 %s' % url) # 停止爬取 break # 日期加1天 last_time += datetime.timedelta(days=1) def get_last_time(self): """查询数据库最后一条的时间""" # 连接数据库 if not settings.MONGODB['AUTHMECHANISM']: client = pymongo.MongoClient( host=settings.MONGODB['HOST'], port=settings.MONGODB['PORT'], username=settings.MONGODB['USERNAME'], password=settings.MONGODB['PASSWORD'], authSource=settings.MONGODB['AUTHSOURCE'] ) else: client = pymongo.MongoClient( host=settings.MONGODB['HOST'], port=settings.MONGODB['PORT'], username=settings.MONGODB['USERNAME'], password=settings.MONGODB['PASSWORD'], authSource=settings.MONGODB['AUTHSOURCE'], authMechanism=settings.MONGODB['AUTHMECHANISM'] ) db = client[settings.DB_NAME] date_list = [] # 查询每个表中最小的日期 for collection_name in settings.COLLECTION_NAMES.values(): collection = db[collection_name] data = collection.find_one({'exchange': 'shfe'}, sort=[('date', -1)]) # 上期所期货最早时间是2002年1月7日 if data: date_list.append(data['date']) else: date_list.append(datetime.datetime(2002, 1, 7)) return min(date_list) class ParseData(threading.Thread): """处理数据类""" def __init__(self, q, trade_q, short_q, long_q): super(ParseData, self).__init__() # 数据队列 self.q = q self.trade_q = trade_q self.short_q = short_q self.long_q = long_q def run(self): global EXIT_FLAG_PARSER while not EXIT_FLAG_PARSER: # 采用非堵塞获取队列数据 try: data = self.q.get(timeout=1) self.parse_data(data) self.q.task_done() except Empty: pass except Exception as e: log.logger.error('数据处理线程出错，时间：%s，错误信息：%s' % (data['report_date'], e), exc_info=True) self.q.task_done() def parse_data(self, data): """处理数据""" # 日期 date = datetime.datetime.strptime(data['report_date'], '%Y%m%d') # log.logger.debug('正在处理 %s' % date) # 把数据转换成DataFrame df = pd.DataFrame(data['o_cursor']) # 如果排名是-1或0，是品种的一个总情况，忽略跳过 df = df[(df['RANK']!=-1) & (df['RANK']!=0)] # 把数据中的空格去掉 df['INSTRUMENTID'] = df['INSTRUMENTID'].str.strip() df['PARTICIPANTABBR1'] = df['PARTICIPANTABBR1'].str.strip() df['PARTICIPANTABBR2'] = df['PARTICIPANTABBR2'].str.strip() df['PARTICIPANTABBR3'] = df['PARTICIPANTABBR3'].str.strip() # 数据类型转换 df['RANK'] = df['RANK'].astype('int32') # 按照合约进行分组 for contract, son_df in df.groupby(df['INSTRUMENTID']): # RANK=999的是对合约的小结 new_df = son_df[son_df['RANK']!=999] for i in range(1, 4): # 如果合约小结没有数据，则跳过 if int(son_df.loc[son_df['RANK']==999, 'CJ%s' % i] == ''): continue # 构造新的DataFrame temp_df = pd.DataFrame({'rank': new_df['RANK'], 'name': new_df['PARTICIPANTABBR%s' % i], 'volume': new_df['CJ%s' % i], 'volumeDiff': new_df['CJ%s_CHG' % i]}).reset_index(drop=True).sort_values('rank') # 把空的数据删除，并转换数据类型 temp_df = temp_df[temp_df['name'] != ''] temp_df[['volume', 'volumeDiff']] = temp_df[['volume', 'volumeDiff']].astype('int32') # 整理全部数据 temp_dict = { 'exchange': 'shfe', 'goods': re.match(r'[^\d]+', contract).group(), 'symbol': 'shfe_%s' % contract.lower(), 'date': date, 'volume': int(son_df.loc[son_df['RANK']==999, 'CJ%s' % i]), 'volumeDiff': int(son_df.loc[son_df['RANK']==999, 'CJ%s_CHG' % i]), 'data': temp_df.to_dict('records'), } # 把数据放进队列 if i == 1: self.trade_q.put(temp_dict) elif i == 3: self.short_q.put(temp_dict) elif i == 2: self.long_q.put(temp_dict) class InsertData(threading.Thread): """插入数据类""" def __init__(self, q, collection_name): super(InsertData, self).__init__() self.q = q if not settings.MONGODB['AUTHMECHANISM']: self.client = pymongo.MongoClient( host=settings.MONGODB['HOST'], port=settings.MONGODB['PORT'], username=settings.MONGODB['USERNAME'], password=settings.MONGODB['PASSWORD'], authSource=settings.MONGODB['AUTHSOURCE'] ) else: self.client = pymongo.MongoClient( host=settings.MONGODB['HOST'], port=settings.MONGODB['PORT'], username=settings.MONGODB['USERNAME'], password=settings.MONGODB['PASSWORD'], authSource=settings.MONGODB['AUTHSOURCE'], authMechanism=settings.MONGODB['AUTHMECHANISM'] ) self.db = self.client[settings.DB_NAME] self.collection = self.db[collection_name] self.collection.create_index([('date', 1), ('symbol', 1)]) def run(self): global EXIT_FLAG_INSERTER while not EXIT_FLAG_INSERTER: # 采用非堵塞获取队列数据 try: data = self.q.get(timeout=1) self.insert_data(data) self.q.task_done() except Empty: pass except Exception as e: log.logger.error('插入数据线程出错, 时间：%s，错误内容：%s' % (data['date'], e), exc_info=True) self.q.task_done() self.client.close() def insert_data(self, data): date = data['date'] symbol = data['symbol'] try: # log.logger.debug('正在插入 %s %s' % (symbol, date)) self.collection.replace_one({'date': date, 'symbol': symbol}, data, True) except Exception as e: log.logger.error('插入数据出错 %s' % data) def main(): start = time.time() log.logger.info('-'*50+' start '+'-'*50) log.logger.info('开始上期所大户持仓爬虫程序') # 数据处理线程退出信号 global EXIT_FLAG_PARSER EXIT_FLAG_PARSER = False # 数据插入线程退出信号 global EXIT_FLAG_INSERTER EXIT_FLAG_INSERTER = False # 爬虫数据队列 q = Queue() # 成交量排名数据队列 trade_q = Queue() # 持卖单量排名数据队列 short_q = Queue() # 持买单量排名数据队列 long_q = Queue() # 开启爬虫 crawler = CrawlData(q) crawler.start() # 开启数据处理 parser1 = ParseData(q, trade_q, short_q, long_q) parser2 = ParseData(q, trade_q, short_q, long_q) parser1.start() parser2.start() # 开启数据插入 insert1 = InsertData(trade_q, settings.COLLECTION_NAMES['TRADE']) insert2 = InsertData(short_q, settings.COLLECTION_NAMES['SHORT']) insert3 = InsertData(long_q, settings.COLLECTION_NAMES['LONG']) insert1.start() insert2.start() insert3.start() # 等待爬虫线程结束 crawler.join() # 等待数据处理完成 q.join() # 通知数据处理线程可以结束了 EXIT_FLAG_PARSER = True # 等待处理线程结束 parser1.join() parser2.join() # 等待其他数据队列完成 trade_q.join() short_q.join() long_q.join() # 通知数据插入线程已经没有其他数据了 EXIT_FLAG_INSERTER = True # 等待数据插入线程结束 insert1.join() insert2.join() insert3.join() log.logger.info('上期所大户持仓数据已更新完成') log.logger.info('共耗时%ss' % (time.time()-start)) log.logger.info('-'*50+' end '+'-'*50) if __name__ == "__main__": main()