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

33次阅读

共计 12449 个字符，预计需要花费 32 分钟才能阅读完成。

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`amount
type1=`SYMBOL`NANOTIMESTAMP`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE
db=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_mv
type2=`SYMBOL`NANOTIMESTAMP`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE`DOUBLE
hushen_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 datetime
import tushare as ts
import pandas as pd
import numpy as np
import dolphindb as ddb
pro=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 dates

for 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_line
5,332,932

select count(*) from hushen_daily_indicator
5,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_code
select * from t where trade_date=2008.11.14
ts_code            trade_date    mvol
000001.SZ    2008.11.14    0.048551
000002.SZ    2008.11.14    0.04565
000004.SZ    2008.11.14    0.030721
000005.SZ    2008.11.14    0.046655
000006.SZ    2008.11.14    0.043092
000008.SZ    2008.11.14    0.035764
000009.SZ    2008.11.14    0.051113
000010.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 desc
ts_code            corr_ts_code    corr
000001.SZ    601166.SH    0.859
000001.SZ    600000.SH    0.8406
000001.SZ    002920.SZ    0.8175
000001.SZ    600015.SH    0.8153
000001.SZ    600036.SH    0.8129
000001.SZ    600016.SH    0.8022
000001.SZ    002142.SZ    0.7956
000001.SZ    601169.SH    0.7882
000001.SZ    601009.SH    0.7778
000001.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.06
trade_date    ts_code        A1
2017.07.06    000001.SZ    252.5
2017.07.06    000002.SZ    1,103.5
2017.07.06    000004.SZ    252.5
2017.07.06    000005.SZ    252.5
2017.07.06    000006.SZ    1,103.5
2017.07.06    000008.SZ    1,972.5
2017.07.06    000009.SZ    1,103.5
2017.07.06    000010.SZ    1,103.5
2017.07.06    000011.SZ    1,103.5
...

select * from alpha98(stock1) where trade_date=2017.07.19
ts_code    trade_date    A98
000001.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    224
000006.SZ    2017.07.19    791
000007.SZ    2017.07.19    (609)
000008.SZ    2017.07.19    444
000009.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.01
endDate=2018.01.01 
holdingDays=21
groups=10
ports = 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_code
lastDays = 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 date
plot(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 插件，还提供了功能强大的集成开发工具，反对图形化数据显示，让试验后果更加直观易于了解。

正文完
dolphindb
发表至： dolphindb
2020-12-21
 0
关于dolphindb:如何使用Window-Join快速估计个股交易成本
关于dolphindb:时序数据库在工业物联网中的应用
关于dolphindb:干货丨如何使用时序数据库快速计算买方或卖方驱动交易
关于dolphindb:干货丨如何高速迁移MySQL数据到时序数据库DolphinDB