关于dolphindb:干货丨如何使用时序数据库处理Tushare金融数据

DolphinDB是新一代的时序数据库，不仅能够作为分布式数据仓库或者内存数据库来应用，而且自带丰盛的计算工具，能够作为钻研工具或钻研平台来应用，十分实用于量化金融、物联网等畛域的海量数据分析。量化金融畛域的不少问题，如交易信号钻研、策略回测、交易成本剖析、股票相关性钻研、市场危险管制等，都能够用DolphinDB来解决。

本教程将介绍如何把Tushare的沪深股票2008年到2017年的日线行情数据和每日指标数据导入到 DolphinDB database，并应用DolphinDB进行金融剖析。Tushare是金融大数据凋谢社区，领有丰盛的金融数据，如股票、基金、期货、数字货币等行情数据，为量化从业人员和金融相干钻研人员收费提供金融数据。

1. 数据详情

Tushare提供的沪深股票日线行情数据蕴含以下字段：

名称            形容ts_code            股票代码trade_date    交易日期open            开盘价high            最高价low            最低价close            收盘价pre_close    昨收价change            涨跌额pct_change    涨跌幅vol            成交量（手）amount            成交额（千元）

每日指标数据蕴含以下字段：

名称            形容ts_code            股票代码trade_date    交易日期close            收盘价turnover_rate    换手率turnover_rate_f    换手率（自在流通股）volume_ratio    量比pe            市盈率（总市值/净利润）pe_ttm            市盈率（TTM）pb            市净率（总市值/净资产）ps            市销率ps_ttm            市销率（TTM）total_share    总股本（万）float_share    流通股本（万）free_share    自在流通股本（万）total_mv    总市值（万元）cric_mv            流通市值（万元）

2. 创立DolphinDB数据库

2.1 装置DolphinDB

从官网下载DolphinDB安装包和DolphinDB GUI.

DolphinDB单节点部署请参考单节点部署。

DolphinDB单服务器集群部署请参考单服务器集群部署。

DolphinDB多物理服务器部署请参考多服务器集群部署。

2.2 创立数据库

咱们能够应用database函数创立分区数据库。

语法：database(directory, [partitionType], [partitionScheme], [locations])

参数

_directory：_数据库保留的目录。DolphinDB有三种类型的数据库，别离是内存数据库、磁盘上的数据库和分布式文件系统上的数据库。创立内存数据库，directory为空；创立本地数据库，directory应该是本地文件系统目录；创立分布式文件系统上的数据库，directory应该以“dfs://”结尾。本教程应用分布式文件系统上的数据库。

_partitionType：_分区形式，有6种形式：程序分区（SEQ），范畴分区（RANGE），哈希分区（HASH），值分区（VALUE），列表分区（LIST），复合分区（COMPO）。

_partitionScheme：_分区计划。各种分区形式对应的分区计划如下：

导入数据前，要做好数据的分区规划，次要思考两个因素：分区字段和分区粒度。

在日常的查问剖析中，依照日期查问的频率最高，所以分区字段为日期trade_date。如果一天一个分区，每个分区的数据量过少，只有3000多条数据，不到1兆大小，而且分区数量十分多。分布式系统在执行查问时，会把查问语句分成多个子工作发送到不同的分区。这样的分区形式会导致子工作数量十分多，而每个工作执行的工夫极短，零碎在治理工作上消耗的工夫反而大于工作自身的执行工夫，显著这样的分区形式是不合理。这种状况下，咱们按日期范畴进行分区，每年的1月1日到次年的1月1日为一个分区，这样既能晋升查问的效率，也不会造成分区粒度过小。

现有数据的时间跨度是2008-2017年，然而为了给将来的数据留出足够的空间，咱们把工夫范畴设置为2008-2030年。执行以下代码：

yearRange=date(2008.01M + 12*0..22)

因为日线行情和每日指标数据的分区计划雷同，因而把它们寄存在同一个数据库dfs://tushare的两个表中，hushen_daily_line用于寄存日线行情数据，hushen_daily_indicator用于寄存每日指标数据。如果须要应用内存数据库，创立数据库时把_directory_设为空；如果须要应用磁盘上的数据库，把_directory_设置为磁盘目录即可。创立数据库的代码如下：

login("admin","123456")dbPath="dfs://tushare"yearRange=date(2008.01M + 12*0..22)if(existsDatabase(dbPath)){    dropDatabase(dbPath)}columns1=`ts_code`trade_date`open`high`low`close`pre_close`change`pct_change`vol`amounttype1=`SYMBOL`NANOTIMESTAMP`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLEdb=database(dbPath,RANGE,yearRange)hushen_daily_line=db.createPartitionedTable(table(100000000:0,columns1,type1),`hushen_daily_line,`trade_date)columns2=`ts_code`trade_date`close`turnover_rate`turnover_rate_f`volume_ratio`pe`pr_ttm`pb`ps`ps_ttm`total_share`float_share`free_share`total_mv`circ_mvtype2=`SYMBOL`NANOTIMESTAMP`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLEhushen_daily_indicator=db.createPartitionedTable(table(100000000:0,columns2,type2),`hushen_daily_indicator,`trade_date)

3. 应用Python API把数据导入到DolphinDB

Tushare提供了两种罕用的数据调取形式：

通过Tushare Python包，返回的是python dataframe类型数据。
通过http协定间接获取，返回的是Json格局数据。

本教程应用了第一种办法调取沪深股票2008年到2017年10年的日线行情数据和每日指标数据。

3.1 下载安装Python3.X和Tushare

具体教程请参考Tushare官网。

3.2 装置DolphinDB的Python3 API

从官网下载Python3 API，把Python3 API的安装包解压至任意目录。在console中进入该目录，执行以下命令：

python setup.py install

应用以下命令更新Python API：

python setup.py install --force

3.3 数据导入

咱们别离应用Tushare Python包的daily和daily_basic接口调取日线行情和每日指标数据，返回的是Python Dataframe类型数据。留神，须要注册Tushare账号能力获取token。接着，通过Python API，连贯到IP为localhost，端口号为8941的DolphinDB数据节点（这里的数据节点IP与端口依据本人集群的状况进行批改），把Tushare返回的Dataframe数据别离追加到之前创立的DolphinDB DFS Table中。

具体的Python代码如下：

import datetimeimport tushare as tsimport pandas as pdimport numpy as npimport dolphindb as ddbpro=ts.pro_api('YOUR_TOKEN')s=ddb.session()s.connect("localhost",8941,"admin","123456")t1=s.loadTable(tableName="hushen_daily_line",dbPath="dfs://tushare")t2=s.loadTable(tableName="hushen_daily_indicator",dbPath="dfs://tushare")def dateRange(beginDate,endDate):    dates=[]    dt=datetime.datetime.strptime(beginDate,"%Y%m%d")    date=beginDate[:]    while date <= endDate:        dates.append(date)        dt=dt + datetime.timedelta(1)        date=dt.strftime("%Y%m%d")    return datesfor dates in dateRange('20080101','20171231'):df=pro.daily(trade_date=dates)df['trade_date']=pd.to_datetime(df['trade_date'])if len(df):    t1.append(s.table(data=df))    print(t1.rows)for dates in dateRange('20080101','20171231'):ds=pro.daily_basic(trade_date=dates)ds['trade_date']=pd.to_datetime(ds['trade_date'])ds['volume_ratio']=np.float64(ds['volume_ratio'])if len(ds):    t2.append(s.table(data=ds))    print(t2.rows)

数据导入胜利后，咱们能够从DolphinDB GUI右下角的变量浏览器中看到两个表的分区状况：

查看数据量：

select count(*) from hushen_daily_line5,332,932select count(*) from hushen_daily_indicator5,333,321

至此，咱们曾经把沪深股票2008年-2017年的日线行情和每日指标数据全副导入到DolphinDB中。

4. 金融剖析

DolphinDB将数据库、编程语言和分布式计算交融在一起，不仅能够用作数据仓库，还能够用作计算和剖析工具。DolphinDB内置了许多通过优化的工夫序列函数，特地实用于投资银行、对冲基金和交易所的定量查问和剖析，能够用于构建基于历史数据的策略测试。上面介绍如何应用Tushare的数据进行金融剖析。

4.1 计算每只股票滚动稳定率

daily_line= loadTable("dfs://daily_line","hushen_daily_line")t=select ts_code,trade_date,mstd(pct_change/100,21) as mvol from daily_line context by ts_codeselect * from t where trade_date=2008.11.14ts_code            trade_date    mvol000001.SZ    2008.11.14    0.048551000002.SZ    2008.11.14    0.04565000004.SZ    2008.11.14    0.030721000005.SZ    2008.11.14    0.046655000006.SZ    2008.11.14    0.043092000008.SZ    2008.11.14    0.035764000009.SZ    2008.11.14    0.051113000010.SZ    2008.11.14    0.027254...

计算每只股票一个月的滚动稳定率，仅需一行代码。DolphinDB自带金融基因，内置了大量与金融相干的函数，能够用简略的代码计算金融指标。

4.2 找到最相干的股票

应用沪深股票日线行情数据，计算股票的两两相关性。首先，生成股票回报矩阵：

retMatrix=exec pct_change/100 as ret from loadTable("dfs://daily_line","hushen_daily_line") pivot by trade_date,ts_code

exec和pivot by是DolphinDB编程语言的特点之一。exec与select的用法雷同，然而select语句生成的是表，exec语句生成的是向量。pivot by用于整顿维度，与exec一起应用时会生成一个矩阵。

接着，生成股票相关性矩阵：

corrMatrix=cross(corr,retMatrix,retMatrix)

下面应用到的cross是DolphinDB中的高阶函数，它以函数和对象作为输出内容，把函数利用到每个对象上。模板函数在简单的批量计算中十分有用。

而后，找到每只股票相关性最高的10只股票：

mostCorrelated = select * from table(corrMatrix).rename!(`ts_code`corr_ts_code`corr) context by ts_code having rank(corr,false) between 1:10

查找与000001.SZ相关性最高的10只股票：

select * from mostCorrelated where ts_code="000001.SZ" order by cor descts_code            corr_ts_code    corr000001.SZ    601166.SH    0.859000001.SZ    600000.SH    0.8406000001.SZ    002920.SZ    0.8175000001.SZ    600015.SH    0.8153000001.SZ    600036.SH    0.8129000001.SZ    600016.SH    0.8022000001.SZ    002142.SZ    0.7956000001.SZ    601169.SH    0.7882000001.SZ    601009.SH    0.7778000001.SZ    601328.SH    0.7736

下面两个示例都比较简单，上面咱们进行简单的计算。

4.3 构建World Quant Alpha #001和#98

WorldQuant LLC发表的论文101 Formulaic Alphas中给出了101个Alpha因子公式。很多集体和机构尝试用不同的语言来实现这101个Alpha因子。本文中，咱们例举了较为简单的Alpha #001和较为简单的Alpha #098两个因子的实现。

Alpha#001公式：rank(Ts_ArgMax(SignedPower((returns<0?stddev(returns,20):close), 2), 5))-0.5

Alpha #001的具体解读能够参考【史上最具体】WorldQuant Alpha 101因子系列#001钻研。

Alpha#98公式：(rank(decay_linear(correlation(vwap, sum(adv5,26.4719), 4.58418), 7.18088))- rank(decay_linear(Ts_Rank(Ts_ArgMin(correlation(rank(open), rank(adv15), 20.8187), 8.62571), 6.95668) ,8.07206)))

这两个因子在计算时候既用到了cross sectional的信息，也用到了大量工夫序列的计算。也即在计算某个股票某一天的因子时，既要用到该股票的历史数据，也要用到当天所有股票的信息，所以计算量很大。

构建这两个因子，须要蕴含以下字段：股票代码、日期、成交量、成交量的加权平均价格、开盘价和收盘价。其中，成交量的加权平均价格能够通过收盘价和成交量计算得出。因而，日线行情的数据能够用于构建这两个因子。

构建因子的代码如下：

def alpha1(stock){    t= select trade_date,ts_code,mimax(pow(iif(ratios(close) < 1.0, mstd(ratios(close) - 1, 20),close), 2.0), 5) as maxIndex from stock context by ts_code    return select trade_date,ts_code,rank(maxIndex) - 0.5 as A1 from t context by trade_date}def alpha98(stock){    t = select ts_code,trade_date, wavg(close,vol) as vwap, open, mavg(vol, 5) as adv5, mavg(vol,15) as adv15 from stock context by ts_code    update t set rank_open = rank(open), rank_adv15 = rank(adv15) context by trade_date    update t set decay7 = mavg(mcorr(vwap, msum(adv5, 26), 5), 1..7), decay8 = mavg(mrank(9 - mimin(mcorr(rank_open, rank_adv15, 21), 9), true, 7), 1..8) context by ts_code    return select ts_code,trade_date, rank(decay7)-rank(decay8) as A98 from t context by trade_date}

构建Alpha #001仅用了2行外围代码，Alpha #98仅用了4行外围代码，并且所有外围代码都是用SQL实现，可读性十分好。SQL中最要害的性能是context by子句实现的分组计算性能。context by是DolphinDB对规范SQL的扩大。与group by每个组产生一行记录不同，context by会输入跟输出雷同行数的记录，所以咱们能够不便的进行多个函数嵌套。cross sectional计算时，咱们用trade_date分组。工夫序列计算时，咱们用ts___code分组。与传统的剖析语言Matlab、SAS不同，DolphinDB脚本语言与分布式数据库和分布式计算严密集成，表达能力强，高性能易扩大，可能满足疾速开发和建模的须要。

查看后果：

select * from alpha1(stock1) where trade_date=2017.07.06trade_date    ts_code        A12017.07.06    000001.SZ    252.52017.07.06    000002.SZ    1,103.52017.07.06    000004.SZ    252.52017.07.06    000005.SZ    252.52017.07.06    000006.SZ    1,103.52017.07.06    000008.SZ    1,972.52017.07.06    000009.SZ    1,103.52017.07.06    000010.SZ    1,103.52017.07.06    000011.SZ    1,103.5...select * from alpha98(stock1) where trade_date=2017.07.19ts_code    trade_date    A98000001.SZ    2017.07.19    (1,073)000002.SZ    2017.07.19    (1,270)000004.SZ    2017.07.19    (1,805)000005.SZ    2017.07.19    224000006.SZ    2017.07.19    791000007.SZ    2017.07.19    (609)000008.SZ    2017.07.19    444000009.SZ    2017.07.19    (1,411)000010.SZ    2017.07.19    (1,333)...

应用单线程计算，Alpha #001耗时仅4秒，简单的Alpha #98耗时仅5秒，性能极佳。

4.4 动量交易策略

动量策略是投资界最风行的策略之一。艰深地讲，动量策略就是“追涨杀跌”，买涨得厉害的，卖跌得厉害的。上面将介绍如何在DolphinDB中测试动量交易策略。

最罕用的动量因素是过来一年扣除最近一个月的收益率。动量策略通常是一个月调整一次，并且持有期也是一个月。本教程中，每天调整1/21的投资组合，并持有新的投资组合21天。

要测试动量交易策略，须要蕴含以下字段的数据：股票代码、日期、每股价格（收盘价格）、流通市值、股票日收益和每日交易量。

显然，只有日线行情的数据是不够的，咱们须要连贯hushen_daily_line和hushen_daily_indicator两个表。

通过equal join，从两个表中抉择须要的字段：

daily_line=loadTable(“dfs://daily_line”,”hushen_daily_line”)daily_indicator=loadTable(“dfs://daily_indicator”,”hushen_daily_indicator”)s=select ts_code,trade_date,close,change,pre_close,vol,amount,turnover_rate,total_share,float_share,free_share,total_mv,circ_mv from ej(daily_line,daily_indicator,`ts_code`trade_date)

（1）对数据进行荡涤和过滤，为每只股票构建过来一年扣除最近一个月收益率的动量信号。

def loadPriceData(inData){    stocks = select ts_code, trade_date, close,change/pre_close as ret, circ_mv from inData where weekday(trade_date) between 1:5, isValid(close), isValid(vol) order by ts_code, trade_date    stocks = select ts_code, trade_date,close,ret,circ_mv, cumprod(1+ret) as cumretIndex from stocks context by ts_code    return select ts_code, trade_date, close, ret, circ_mv, move(cumretIndex,21)move(cumretIndex,252)-1 as signal from stocks context by ts_code }priceData = loadPriceData(s)

（2）生成投资组合

抉择满足以下条件的流通股：动量信号无缺失、当天的交易量为正、市值超过1亿元以及每股价格超过5元。

def genTradables(indata){    return select trade_date, ts_code, circ_mv, signal from indata where close>5, circ_mv>10000, vol>0, isValid(signal) order by trade_date}tradables = genTradables(priceData)

依据每天的动量信号，产生10组流通股票。只保留两个最极其的群体（赢家和输家）。假如在21天内，每天总是多头1元和空头1元，所以咱们每天在赢家组多头1/21，在输家组每天空头1/21。在每组中，咱们能够应用等权重或值权重，来计算投资组合造成日期上每个股票的权重。

//WtScheme=1示意等权重；WtScheme=2示意值权重def formPortfolio(startDate, endDate, tradables, holdingDays, groups, WtScheme){    ports = select date(trade_date) as trade_date, ts_code, circ_mv, rank(signal,,groups) as rank, count(ts_code) as symCount, 0.0 as wt from tradables where date(trade_date) between startDate:endDate context by trade_date having count(ts_code)>=100    if (WtScheme==1){        update ports set wt = -1.0count(ts_code)holdingDays where rank=0 context by trade_date        update ports set wt = 1.0count(ts_code)holdingDays where rank=groups-1 context by trade_date    }    else if (WtScheme==2){        update ports set wt = -circ_mvsum(circ_mv)holdingDays where rank=0 context by trade_date        update ports set wt = circ_mvsum(circ_mv)holdingDays where rank=groups-1 context by trade_date    }    return select ts_code, trade_date as tranche, wt from ports where wt != 0 order by ts_code, trade_date}startDate=2008.01.01endDate=2018.01.01 holdingDays=21groups=10ports = formPortfolio(startDate, endDate, tradables, holdingDays, groups, 2)dailyRtn = select date(trade_date) as trade_date, ts_code, ret as dailyRet from priceData where date(trade_date) between startDate:endDate

（3）计算投资组合中每只股票接下来21天的利润或损失。在投资组合造成后的21天关停投资组合。

def calcStockPnL(ports, dailyRtn, holdingDays, endDate, lastDays){    ages = table(1..holdingDays as age)    dates = sort distinct ports.tranche        dictDateIndex = dict(dates, 1..dates.size())        dictIndexDate = dict(1..dates.size(), dates)    pos = select dictIndexDate[dictDateIndex[tranche]+age] as date, ts_code, tranche, age, take(0.0,size age) as ret, wt as expr, take(0.0,size age) as pnl from cj(ports,ages) where isValid(dictIndexDate[dictDateIndex[tranche]+age]), dictIndexDate[dictDateIndex[tranche]+age]<=min(lastDays[ts_code], endDate)    update pos set ret = dailyRet from ej(pos, dailyRtn,`date`ts_code,`trade_date`ts_code)    update pos set expr = expr*cumprod(1+ret) from pos context by ts_code, tranche    update pos set pnl = expr*ret/(1+ret)    return pos}lastDaysTable = select max(date(trade_date)) as date from priceData group by ts_codelastDays = dict(lastDaysTable.ts_code, lastDaysTable.date)stockPnL = calcStockPnL(ports, dailyRtn, holdingDays, endDate, lastDays)

（4）计算投资组合的利润或损失，并绘制动量策略累计回报走势图。

portPnL = select sum(pnl) as pnl from stockPnL group by date order by dateplot(cumsum(portPnL.pnl) as cumulativeReturn,portPnL.date, "Cumulative Returns of the Momentum Strategy")

上面是沪深股票2008年到2017年的回测后果。回测时，每天产生一个新的tranche，持有21天。应用单线程计算，耗时仅6秒。

如果应用Pandas来解决金融数据，对内存的要求较高，内存应用峰值个别是数据的3-4倍，随着数据的积攒，pandas的内存占用问题会越来越显著。在性能上，pandas在多线程解决方面比拟弱，不能充分利用多核CPU的计算能力，并且pandas不能依据业务字段对数据进行分区，也不反对列式存储，查问数据时必须全表扫描，效率不高。

5. 总结

作为数据库，DolphinDB反对单表PB级存储和灵便的分区形式；作为钻研平台，DolphinDB不仅功能丰富，反对疾速的数据荡涤、高效的数据导入、交互式剖析、库内剖析，流计算框架和离线计算反对生产环境代码重用，而且性能极佳，即便面对宏大数据集，仍能够轻松实现秒级毫秒级的低延时交互剖析。另外，DolphinDB对用户非常敌对，提供了丰盛的编程接口，如Python、C++、Java、C#、R等编程API和Excel的add-in插件、ODBC、JDBC插件，还提供了功能强大的集成开发工具，反对图形化数据显示，让试验后果更加直观易于了解。