stock_untitl.py
from pymongo import ASCENDING
from database import DB_CONN
from datetime import datetime,timedelta
# In[2]:
def get_trading_dates(begin_date=None, end_date=None):
? ? """
? ? 獲取指定日期范圍的按照正序排列的交易日列表
? ? 如果沒(méi)有指定日期范圍，則獲取從當(dāng)期日期向前365個(gè)自然日內(nèi)的所有交易日
? ? :param begin_date: 開(kāi)始日期
? ? :param end_date: 結(jié)束日期
? ? :return: 日期列表
? ? """
? ? # 當(dāng)前日期
? ? now = datetime.now()
? ? # 開(kāi)始日期，默認(rèn)今天向前的365個(gè)自然日
? ? if begin_date is None:
? ? ? ? # 當(dāng)前日期減去365天
? ? ? ? one_year_ago = now - timedelta(days=365)
? ? ? ? # 轉(zhuǎn)化為str類型
? ? ? ? begin_date = one_year_ago.strftime('%Y-%m-%d')
? ? # 結(jié)束日期默認(rèn)為今天
? ? if end_date is None:
? ? ? ? end_date = now.strftime('%Y-%m-%d')
? ? # 用上證綜指000001作為查詢條件，因?yàn)橹笖?shù)是不會(huì)停牌的，所以可以查詢到所有的交易日
? ? daily_cursor = DB_CONN.daily.find(
? ? ? ? {'code': '000001', 'date': {'$gte': begin_date, '$lte': end_date}, 'index': True},
? ? ? ? sort=[('date', ASCENDING)],
? ? ? ? projection={'date': True, '_id': False})
? ? # 轉(zhuǎn)換為日期列表
? ? dates = [x['date'] for x in daily_cursor]
? ? return dates
def get_all_codes():
? ? """
? ? 獲取所有股票代碼列表
? ? :return: 股票代碼列表
? ? """
? ? # 通過(guò)distinct函數(shù)拿到所有不重復(fù)的股票代碼列表
? ? return DB_CONN.basic.distinct('code')
# In[4]:
if __name__ == '__main__':
? ? get_all_codes()
basic_crawler:
import traceback
from datetime import datetime,timedelta
import tushare as ts
from pymongo import MongoClient
from pandas.io import json
from pymongo import UpdateOne
from stock_util import get_trading_dates
DB_CONN = MongoClient('mongodb://127.0.0.1:27017')['quant_01']
#?從tushare獲取股票基礎(chǔ)數(shù)據(jù)，保存到本地的MongoDB數(shù)據(jù)庫(kù)中
def crawl_basic(begin_date=None, end_date=None):
? ? """
? ? 抓取指定時(shí)間范圍內(nèi)的股票基礎(chǔ)信息
? ? :param begin_date: 開(kāi)始日期
? ? :param end_date: 結(jié)束日期
? ? """
? ? # 如果沒(méi)有指定開(kāi)始日期，則默認(rèn)為前一日
? ? if begin_date is None:
? ? ? ? begin_date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
? ? # 如果沒(méi)有指定結(jié)束日期，則默認(rèn)為前一日
? ? if end_date is None:
? ? ? ? end_date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
? ? # 獲取指定日期范圍的所有交易日列表
? ? all_dates = get_trading_dates(begin_date, end_date)
? ? # 按照每個(gè)交易日抓取
? ? for date in all_dates:
? ? ? ? try:
? ? ? ? ? ? # 抓取當(dāng)日的基本信息
? ? ? ? ? ? crawl_basic_at_date(date)
? ? ? ? except:
? ? ? ? ? ? print('抓取股票基本信息時(shí)出錯(cuò)，日期：%s' % date, flush=True)
def crawl_basic_at_date(date):
? ? """
? ? 從Tushare抓取指定日期的股票基本信息
? ? :param date: 日期
? ? """
? ? # 從TuShare獲取基本信息，index是股票代碼列表
? ? df_basics = ts.get_stock_basics(date)
? ? # 如果當(dāng)日沒(méi)有基礎(chǔ)信息，在不做操作
? ? if df_basics is None:
? ? ? ? return
? ? # 初始化更新請(qǐng)求列表
? ? update_requests = []
? ? # 獲取所有股票代碼集合
? ? codes = list(set(df_basics.index))? #codes = list(set(df_basics.index))[:2]
? ? # 按照股票代碼提取所有數(shù)據(jù)
? ? for code in codes:
? ? ? ? # 獲取一只股票的數(shù)據(jù)
? ? ? ? doc = dict(df_basics.loc[code])
? ? ? ? try:
? ? ? ? ? ? # API返回的數(shù)據(jù)中，上市日期是一個(gè)int類型。將上市日期，20180101轉(zhuǎn)換為2018-01-01的形式
? ? ? ? ? ? time_to_market = datetime \
? ? ? ? ? ? ? ? .strptime(str(doc['timeToMarket']), '%Y%m%d') \
? ? ? ? ? ? ? ? .strftime('%Y-%m-%d')
? ? ? ? ? ? # 將總股本和流通股本轉(zhuǎn)為數(shù)字
? ? ? ? ? ? totals = float(doc['totals'])
? ? ? ? ? ? outstanding = float(doc['outstanding'])
? ? ? ? ? ? # 組合成基本信息文檔
? ? ? ? ? ? doc.update({
? ? ? ? ? ? ? ? # 股票代碼
? ? ? ? ? ? ? ? 'code': code,
? ? ? ? ? ? ? ? # 日期
? ? ? ? ? ? ? ? 'date': date,
? ? ? ? ? ? ? ? # 上市日期
? ? ? ? ? ? ? ? 'timeToMarket': time_to_market,
? ? ? ? ? ? ? ? # 流通股本
? ? ? ? ? ? ? ? 'outstanding': outstanding,
? ? ? ? ? ? ? ? # 總股本
? ? ? ? ? ? ? ? 'totals': totals
? ? ? ? ? ? })
? ? ? ? ? ? # 生成更新請(qǐng)求，需要按照code和date創(chuàng)建索引
? ? ? ? ? ? # tushare
? ? ? ? ? ? # numpy.int64/numpy.float64等數(shù)據(jù)類型，保存到mongodb時(shí)無(wú)法序列化。
? ? ? ? ? ? # 解決辦法：這里使用pandas.json強(qiáng)制轉(zhuǎn)換成json字符串，然后再轉(zhuǎn)換成dict。int64/float64轉(zhuǎn)換成int,float
? ? ? ? ? ? update_requests.append(
? ? ? ? ? ? ? ? UpdateOne(
? ? ? ? ? ? ? ? ? ? {'code': code, 'date': date},
? ? ? ? ? ? ? ? ? ? {'$set': json.loads(json.dumps(doc))}, upsert=True))
? ? ? ? except:
? ? ? ? ? ? print('發(fā)生異常，股票代碼：%s，日期：%s' % (code, date), flush=True)
? ? ? ? ? ? print(doc, flush=True)
? ? ? ? ? ? print(traceback.print_exc())
? ? # 如果抓到了數(shù)據(jù)
? ? if len(update_requests) > 0:
? ? ? ? update_result = DB_CONN['basic'].bulk_write(update_requests, ordered=False)
? ? ? ? print('抓取股票基本信息，日期：%s, 插入：%4d條，更新：%4d條' %
? ? ? ? ? ? ? (date, update_result.upserted_count, update_result.modified_count), flush=True)
if __name__ == '__main__':
? ? crawl_basic('2017-01-01', '2017-12-31')
