本文共 4303 字，大约阅读时间需要 14 分钟。

DolphinDB入门与实践：基于World Quant的金融因子实现

作为一款专为时序数据设计的高性能时序数据库，DolphinDB在金融领域的应用越来越广泛。本文将从新手的视角，介绍DolphinDB的安装、实用操作以及基于World Quant论文《101 Formulaic Alphas》的金融因子实现。

前导知识

在开始使用DolphinDB之前，需要理解一些基本概念：

目标因子介绍

Alpha #001

rank(Ts_ArgMax(SignedPower((returns < 0 ? stddev(returns, 20) : close), 2), 5)) - 0.5

该因子通过计算股票回报率与其波动率的关系，捕捉超额收益的潜在因子。

Alpha #098

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))

该因子结合了成交量、价格和市场流动性等多个因素，用于评估股票的超额收益。

数据准备

数据表结构

以下是用于存储股票日线数据的表结构：

数据导入

使用Tushare API将股票数据导入DolphinDB。以下是导入代码示例：

import tushare as tsimport pandas as pdimport numpy as npimport dolphindb as ddbpro = ts.pro_api('')  # 替换为实际的Pro API密钥s = ddb.session()s.connect("localhost", 8848, "admin", "123456")def dateRange(beginDate, endDate):    dates = []    dt = datetime.datetime.strptime(beginDate, "%Y%m%d")    currentDate = beginDate    while currentDate <= endDate:        dates.append(currentDate)        dt = dt + datetime.timedelta(1)        currentDate = dt.strftime("%Y%m%d")    return datesfor dates in dateRange('20080101', '20100110'):    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', '20100110'):    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中创建数据库，以下是创建磁盘数据库的示例：

dbPath = "D:/Dolphin_test"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')

数据查看

使用DolphinDB查询已保存数据量：

select count(*) from hushen_daily_line

金融因子计算

计算滚动波动率

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.14'

因子计算

以下是Alpha #001和Alpha #098的实现代码：

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_datedef 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

References

本文中的数据库操作示例参考了DolphinDB官方文档，具体操作请见相关文档。