补充停牌的日K数据
问题
从TuShare获取的数据，停牌日是没有数据的，这将会在回测时，不能直接参与账户的净值计算，导致账户的净值以及收益计算不准确。
停盘
股票由于某种消息或进行某种活动引起股价的连续上涨或下跌，由证券交易所暂停其在股票市场上进行交易。待情况澄清或企业恢复正常后，再复牌在交易所挂牌交易。
解决方法
1、 增加is_trading字段，用于区分停牌日还是交易日
2、补充停牌日的日k数据，更加当前数据现状，填充一个交易日的close、volume、high、low为停牌前最后一个交易日的close、volume为0，is_trading为false
填充指定时间段的is_trading
流程图
代码实现
1、获取所有交易日列表
由于指数比如上证指数（000001）是不会停牌的，因此可通过指数来获得交易日期
def get_trading_date(begin_date= None,end_date=None):
"""
获取指定日期范围的按照正序排列的交易日列表
如果没有指定日期范围，则获取从当期日期向前365个自然日内的所有交易日
:param begin_date: 开始日期
:param end_date: 结束日期
:return: 交易日期列表
"""
#当前日期
now = datetime.now()
#开始日期，默认当前日期向前365个自然日
if begin_date is None:
#当前日期减去365天
one_year_ago = now - timedelta(days=365)
#转换成str类型
begin_date = one_year_ago.strftime("%Y-%m-%d")
#结束日期默认为今天
if end_date is None:
end_date = now.strftime("%Y-%m-%d")
#用上证指数000001作为查询条件，因为指数是不会停盘的，所以可以查询到所有的交易日期
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}
)
#转换日期列表
dates = [x['date'] for x in daily_cursor]
return dates
2、填充某个日行情数据的is_trading，并更新数据库（同样采用bulk_write将更新数据写入数据集）
def fill_is_trading_between(begin_date=None,end_date=None):
"""
填充指定时间段内的is_trading字段
:param begin_date :开始日期
:param end_date :结束日期
"""
#获取指定日期范围的左右交易日子列表，按日期正序排列
all_dates = get_trading_date(begin_date,end_date)
#循环填充所有交易日的is_trading字段
for date in all_dates:
#填充daily数据集
fill_single_date_is_trading(date,'daily')
#填充daily_hfq数据集
fill_single_date_is_trading(date,'daily_hfq')
def fill_single_date_is_trading(date,collection_name):
"""
填充某一个日行情的数据集的is_trading
:param date: 日期
:param collection_name: 集合名称
"""
print('填充字段，字段名：is_trading，日期：%s,数据集：%s' %(date,collection_name))
daily_cursor = DB_CONN[collection_name].find(
{'date':date},
projection={'code':True,'volume':True,'_id':False},
batch_size = 1000
})
update_requests = []
for daily in daily_cursor:
#当日成交量大于0，则为交易状态
is_trading = daily['volume']>0
update_requests.append(
UpdateOne(
{'code':daily['code'],'date':date},
{'$set':{'is_trading':is_trading}}
)
)
if len(update_requests):
update_result = DB_CONN[collection_name].bulk_write(update_requests,ordered=False)
print("填充字段，字段名：is_trading，日期：%s，数据集：%s，更新：%4d"%
(date,collection_name,update_result.modified_count),flush=True)
获取股票基本信息
用途
- 获取每日的股票列表
主要字段
- 股票代码
- 股票名称
- 股本（总股本、流通股本）
- 上市日期
- 日期
通过采用TuShare中get_stock_basics接口获取股票的基本信息
get_stock_basics()接口信息：
def get_stock_basics(date=None):
"""
获取沪深上市公司基本情况
Parameters
date:日期YYYY-MM-DD，默认为上一个交易日，目前只能提供2016-08-09之后的历史数据
Return
--------
DataFrame
code,代码
name,名称
industry,细分行业
area,地区
pe,市盈率
outstanding,流通股本
totals,总股本(万)
totalAssets,总资产(万)
liquidAssets,流动资产
fixedAssets,固定资产
reserved,公积金
reservedPerShare,每股公积金
eps,每股收益
bvps,每股净资
pb,市净率
timeToMarket,上市日期
"""
详细代码实现
from database import DB_CONN
from datetime import datetime,timedelta
import tushare as ts
from pymongo import UpdateOne
from stock_util import get_trading_date
"""
从tushare中获取股票的基础数据，保存在本地MongoDB中
"""
def crawl_basic(begin_date=None,end_date=None):
"""
抓取指定时间范围的股票基础信息
:param begin_date:开始时间
:param end_date:结束时间
"""
#如果没有指定日期，则默认为前一日
if begin_date is None :
begin_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
if end_date is None:
end_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
#获取指定日期范围的所有交易日列表
all_dates = get_trading_date(begin_date,end_date)
#按每个交易日抓取
for date in all_dates:
try:
#抓取当日的基本信息
crawl_basic_at_date(date)
except:
print("抓取股票基本信息出错，日期：%s" %date,flush=True)
def crawl_basic_at_date(date):
"""
从Tushare抓取指定日期的股票基本信息
:param date: 日期
"""
#从tushare获取基本信息，index是股票代码列表
df_basics = ts.get_stock_basics(date)
#如果当日没有基本信息，不做操作
if df_basics is None:
return
#初始化更新列表
update_requests=[]
codes = set(df_basics.index)
#按照股票代码提取所有数据
for code in codes:
#获取一只股票的数据
doc = df_basics.loc[code]
try:
#将上市日期，19971113转换成199-11-13
time_to_market = datetime.strptime(str(doc['timeToMarket']),"%Y%m%d").strftime('%Y-%m-%d')
#将总股本和流通股本转为数字
totals = float(doc['totals'])
outstanding = float(doc['outstanding'])
#组合基本信息文档
doc.update(
{
#股票代码
'code':code,
#日期
'date':date,
#上市时间
'timeToMarket':time_to_market,
#流通股本
'outstanding':outstanding,
#总股本
'totals':totals
}
)
#生成更新请求，按照code、date创建索引
update_requests.append(
UpdateOne(
{'code':code,'date':date},
{'$set':doc},
upsert=True
)
)
except:
print('发生异常，股票代码：%s，日期：%s' %(code,date),flush=True)
print(doc,flush=True)
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)
执行上述程序后，会将股票基本信息（包括，股票代码，日期，上市时间，流通股本和总股本）保存在MongoDB中名为basic的集合（表）中
填充停牌日的行情数据
代码实现：
ef fill_daily_k_at_suspension_days(begin_date=None,end_date=None):
"""
填充指定日期范围内，股票停牌日的行情数据
填充时，停牌的开盘价、最高价、最低价和收盘价都为最近一个交易日的收盘价，成交量为0
is_trading为False
"""
#当前日期的前一天
before = datetime.now() - timedelta(days=1)
#找到据当前最近一个交易日的所有股票的基本信息
basics =[]
while 1:
#转化成str
last_trading_date = before.strftime("%Y-%m-%d")
#因为Tushare的基本信息从2016-08-09开始，如果早于这个时间就结束查找
if last_trading_date < '2016-08-09':
break
#找当日的基本信息
basic_cursor = DB_CONN['basic'].find(
{'date':last_trading_date},
#填充时需要用到两个子段：股票代码code和上市日期timeToMarket，上市日期用来判断是否上市
projection={'code':True,'timeToMarket':True,'_id':False},
#一次性返回5000条数据，可以降低网络IO开销，提高速度
batch_size=5000
)
#将数据放到basics列表中
basics = [basic for basic in basic_cursor]
#如果查到数据，跳出循环
if len(basics)>0:
break
#没有找到数据，则继续向前一天
before = before-timedelta(days=1)
#获取指定日期范围内所有交易日列表
all_dates = get_trading_date(begin_date,end_date)
#填充daily数据集中的停牌日数据
fill_daily_k_at_suspension_days_at_date_at_one_collections(basics,all_dates,'daily')
#填充daily_hfq数据集中的停牌日数据
fill_daily_k_at_suspension_days_at_date_at_one_collections(basics,all_dates,'daily_hfq')
def fill_daily_k_at_suspension_days_at_date_at_one_collections(basics,all_dates,collection):
"""
更新单个数据集的单个日期的数据
:param basic:基本信息
:param all_dates:日期列表
:param collection:集合名
"""
code_last_trading_daily_dict = dict()
for date in all_dates:
update_requests = []
last_daily_code_set = set(code_last_trading_daily_dict.keys())
for basic in basics:
code = basic['code']
#如果循环日期小于上市日期
if date < basic['timeToMarket']:
print('日期：%s,%s还没上市，上市日期为%s'%(date,code,basic['timeToMarket']),flush=True)
else:
#找到当日数据
daily = DB_CONN[collection].find(
{'code':code,'date':date}
)
if daily is not None:
code_last_trading_daily_dict[code] = daily
last_daily_code_set.add(code)
else:
if code in last_daily_code_set:
last_trading_daily = code_last_trading_daily_dict[code]
suspension_daily_doc = {
'code':code,
'date':date,
'close':last_trading_daily['close'],
'open':last_trading_daily['close'],
'high':last_trading_daily['close'],
'low':last_trading_daily['close'],
'volume':0,
'is_trading':False
}
update_requests.append(
UpdateOne(
{'code':code,'date':date},
{'$set':suspension_daily_doc},
upsert=True
)
)
if len(update_requests)>0:
update_result = DB_CONN[collection].bulk_write(update_requests,ordered=False)
print('填充停牌数据，日期：%s,数据集：%s，插入：%4d条，更新：%4d条'%(date,collection,update_result.upserted_count,update_result.modified_count),flush=True)