# 基于DolphinDB的因子计算最佳实践 因子挖掘是量化交易的基础。除传统的基本面因子外，从中高频行情数据中挖掘有价值的因子，并进一步建模和回测以构建交易系统，是一个量化团队的必经之路。金融或者量化金融是一个高度市场化、多方机构高度博弈的领域。因子的有效时间会随着博弈程度的加剧而缩短，如何使用更加高效的工具和流程，更快的找到新的有效的因子，是每一个交易团队必须面对的问题。 近年来，DolphinDB 越来越成为国内乃至国际上大量基金（私募和公募）、资管机构、券商自营团队进行因子挖掘的利器。基于大量客户的反馈，我们特撰写此白皮书，**总结使用 DolphinDB 进行因子挖掘的最佳实践。** - [1. 概述](#1-概述) - [2. 测试数据集](#2-测试数据集) - [2.1 逐笔成交数据](#21-逐笔成交数据) - [2.2 快照数据](#22-快照数据) - [2.3 分钟K数据](#23-分钟k数据) - [3. 投研阶段的因子计算](#3-投研阶段的因子计算) - [3.1 面板数据模式](#31-面板数据模式) - [3.2 SQL模式](#32-sql模式) - [3.3 不同频率的因子开发举例](#33-不同频率的因子开发举例) - [4. 生产环境的流式因子计算](#4-生产环境的流式因子计算) - [4.1 流式增量计算](#41-流式增量计算) - [4.2 数据回放](#42-数据回放) - [4.3 对接交易系统](#43-对接交易系统) - [5. 因子的存储和查询](#5-因子的存储和查询) - [5.1 因子存储](#51-因子存储) - [5.2 因子查询](#52-因子查询) - [5.3 在线获取面板数据](#53-在线获取面板数据) - [6. 因子回测和建模](#6-因子回测和建模) - [6.1 因子回测](#61-因子回测) - [6.2 因子相关性分析](#62-因子相关性分析) - [6.3 多因子建模](#63-多因子建模) - [7. 因子计算的工程化](#7-因子计算的工程化) - [7.1 代码管理](#71-代码管理) - [7.2 单元测试](#72-单元测试) - [7.3 并行计算](#73-并行计算) - [7.4 内存管理](#74-内存管理) - [7.5 权限管理](#75-权限管理) - [7.6 任务管理](#76-任务管理) - [8. 实际案例](#8-实际案例) - [8.1 日频因子](#81-日频因子) - [8.2 分钟频因子](#82-分钟频因子) - [8.3 快照因子](#83-快照因子) - [8.4 逐笔因子](#84-逐笔因子) - [9. 总结](#9-总结) - [附录](#附录) ## 1. 概述 交易团队用于因子挖掘的常见技术栈有几个大的类别: * 使用 python、matlab 等数据分析工具 * 委托第三方开发有图形界面的因子挖掘工具 * 使用 java、c++ 等编程语言自行开发挖掘工具 * 在 DolphinDB 等专业工具上进行二次开发 我们暂且不讨论各个技术栈的优缺点。但不管使用何种技术栈，都必须解决以下几个问题: * 能处理不同频率不同规模的数据集 * 能计算不同风格的因子 * 能处理因子数量不断增长的问题 * 能高效的存取原始数据和因子数据 * 能提升因子挖掘的开发效率 * 能提升因子计算的运行效率（高吞吐，低延时） * 能解决研究的因子用于生产（实盘交易）的问题 * 能解决多个交易员（研究员）或交易团队一起使用时的各种工程问题，如代码管理、单元测试、权限管理、大规模计算等 DolphinDB 作为分布式计算、实时流计算及分布式存储一体化的高性能时序数据库，非常适合因子的存储、计算、建模、回测和实盘交易。通过部署 DolphinDB 单机或集群环境，用户可以快速地处理 GB 级别甚至 PB 级别的海量数据集，日级、分钟级、快照和逐笔委托数据均能高效响应。 DolphinDB 内置了多范式的编程语言（函数式，命令式，向量式、SQL式），可以帮助研发人员高效开发不同风格的因子。此外，DolphinDB 还提供了丰富且性能高效的函数库（超1400个内置函数），尤其是窗口处理方面经过优化的内置算子，大大缩短了因子计算的延时。 DolphinDB 自带的数据回放和流式增量计算引擎可以方便地解决因子挖掘中研发和生产一体化的问题。DolphinDB 的分布式存储和计算框架，天生便于解决工程中的可靠性、扩展性等问题。 本文基于国内 A 股市场各个频率的数据来演示 DolphinDB 计算和规划因子库存储的方案。根据批量因子计算、实时因子计算、多因子建模、因子库存储规划、因子计算工程化等各个场景的实操演练，以及针对不同方案的对比分析，本文总结出了在 DolphinDB 中进行因子计算的最佳实践。 ## 2. 测试数据集 本文的因子计算基于三类国内 A 股行情数据集：逐笔数据、快照数据和 K 线数据（分钟 K 线和日 K 线）。快照数据以两种形式存储：（1）各档数据分别存储为一列；（2）用 [array vector](https://www.dolphindb.cn/cn/help/200/DataTypesandStructures/DataForms/Vector/arrayVector.html) 将所有档位的数据存储为一列。 | 数据集|简称|代码样例中的分区数据库路径|代码样例中的表名|分区机制| | :-----| :-----| ----: | ----: |----: | | 逐笔成交|level2_tick| dfs://tick_SH_L2_TSDB|tick_SH_L2_TSDB| VALUE:每交易日, HASH: [SYMBOL, 20]| | 快照|level2_snapshot| dfs://snapshot_SH_L2_TSDB|snapshot_SH_L2_TSDB|VALUE:每交易日, HASH: [SYMBOL, 20]| | 快照(向量存储)|level2_snapshot| dfs://LEVEL2_Snapshot_ArrayVector|Snap|VALUE:每交易日, HASH: [SYMBOL, 20]| | 分钟K线|k_line| dfs://k_minute_level|k_minute|VALUE:交易月, HASH: [SYMBOL, 3]| | 日K线|k_line| dfs://k_day_level|k_day|VALUE:年| ### 2.1 逐笔成交数据 ![逐笔成交数据](script/factorPractice/images/level2_tick_samples.png?inline=false) 逐笔成交是交易所公布买卖双方具体成交的每一笔数据，每3秒发布一次，每次包含这3秒内的所有成交记录。每一笔成交撮合，都由买方和卖方的一笔具体委托组成。上述数据样例采用字段 BuyNo 和 SellNo 标注买卖双方的委托单号，其它关键字段分别为： SecurityID（标的物代码），TradeTime（成交时刻），TradePrice（成交价格），TradeQty（本笔成交量）和 TradeAmount（本笔成交金额）。 每个交易日的原始数据量在 8 GB 上下。根据上表的分区机制进行建库建表，点击查看对应脚本：[逐笔成交数据建库建表完整代码](script/factorPractice/appendix_2.1_createTickDbAndTable_main.dos)。 ### 2.2 快照数据 ![快照数据](script/factorPractice/images/snapshot_samples.png?inline=false) 股票交易所每3秒发布一次，每次涵盖这3秒结束时的日内累计成交量(TotalVolumeTrade)，日内累计成交金额(TotalValueTrade)，3秒终了时的盘口买卖双方挂单（买方为 Bid，卖方在有些数据源字段为 Offer，在有些数据源字段为 Ask，其余字段以此类推：BidPrice 为买方各档价格，OfferPrice 为卖方各档价格，OrderQty 为买卖双方各档的委托单总量， Orders 为买卖双方委托单数），3秒终了时的最近一笔成交价格（LastPx），全天开盘价（OpenPx），日内截止当下最高价（HighPx），日内截止当下最低价（LowPx）等各字段。其他和逐笔成交一致的字段不再赘述，涵义一致，详情可参见交易所数据说明字典。 每个交易日的原始数据量约在 10G 左右。 在 DolphinDB 2.0版本的 TSDB 存储引擎中，支持 array vector 的存储机制，即可以允许数据表中一个 cell 存储一个向量。在本白皮书的案例中，后面文章会详细介绍 array vector 存储方案和普通存储方案的区别。快照数据的买10档或卖10档在本例中作为一个 vector 存入单个 cell 中，其他各字段和普通快照数据表都相同。 两种存储模式的建库建表可以参考[Snapshot普通及arrayVector形式建库和建表完整代码](script/factorPractice/appendix_2.2_createSnapshotDbAndTable_main.dos) 快照数据的 array_vector 存储形式： ![array_vector快照数据](script/factorPractice/images/snapshot_in_array_vector_samples.png?inline=false) ### 2.3 分钟K数据 ![分钟K数据](script/factorPractice/images/kline_samples.png?inline=false) 包含每只股票，每分钟的开盘价、最高价、最低价、收盘价，四个价格字段，同时记录本分钟的成交量和成交金额。另外，数据 K 线可以依据基本字段计算衍生字段，比如：k 线均价(vwap 价格)。k 线数据是由逐笔成交数据聚合产生，具体代码可以参考[第三章3.3.4基于快照数据的分钟聚合](#334-基于快照数据的分钟聚合)。 日 K 数据，存储形式和字段跟分钟 k 线一致，可以由分钟 k 线或高频数据聚合产生，这里不作赘述。 日 k 数据，分钟数据的建库建表可以参考：[k 线数据建库建表完整代码](script/factorPractice/appendix_2.3_createTableKMinute_main.dos) ## 3. 投研阶段的因子计算 在投研阶段，会通过历史数据批量计算生成因子。通常，推荐研究员将每一种因子的计算都封装成自定义函数。根据因子类型和使用者习惯的不同，DolphinDB 提供了面板和 SQL 两种计算方式。 在面板计算中，自定义函数的参数一般为向量，矩阵或表，输出一般为向量，矩阵或表；在 SQL 模式中，自定义函数的参数一般为向量（列），输出一般为向量。因子函数的粒度尽可能小，只包含计算单个因子的业务逻辑，也不用考虑并行计算加速等问题。这样做的优点包括：（1）易于实现流批一体，（2）便于团队的因子代码提交和管理，（3）方便用统一的框架运行因子计算作业。 ### 3.1 面板数据模式 面板数据(panel data)是以时间为索引，标的为列，指标作为内容的一种数据载体，它非常适用于以标的集合为单位的指标计算，将数据以面板作为载体，可以大大简化脚本的复杂度，通常最后的计算表达式可以从原始的数学公式中一对一的翻译过来。除此之外，可以充分利用DolphinDB矩阵计算的高效能。 在因子计算中，面板数据通常可以通过`panel`函数，或者`exec`搭配`pivot by`得到，具体样例如下表：每一行是一个时间点，每一列是一个股票。 ``` 000001 000002 000003 000004 ... --------- ---------- ------- ------ --- 2020.01.02T09:29:00.000|3066.336 3212.982 257.523 2400.042 ... 2020.01.02T09:30:00.000|3070.247 3217.087 258.696 2402.221 ... 2020.01.02T09:31:00.000|3070.381 3217.170 259.066 2402.029 ... ``` 在面板数据上，由于是以时间为索引，标的为列，因子可以方便地在截面上做各类运算。DolphinDB 包含 row 系列函数以及各类滑动窗口函数，在下面两个因子计算例子中，原本复杂的计算逻辑，在面板数据中，可以用一行代码轻松实现。 * Alpha 1 因子计算中，下例使用了rowRank 函数，可以在面板数据中的每一个时间截面对各标的进行排名；iif 条件运算，可以在标的向量层面直接筛选及计算；mimax 及 mstd 等滑动窗口函数也是在标的层面垂直计算的。因此，在面板计算中合理应用 DolphinDB 的内置函数，可以从不同维度进行计算。 ``` //alpha 1 //Alpha#001公式：rank(Ts_ArgMax(SignedPower((returns<0?stddev(returns,20):close), 2), 5))-0.5 @state def alpha1TS(close){ return mimax(pow(iif(ratios(close) - 1 < 0, mstd(ratios(close) - 1, 20),close), 2.0), 5) } def alpha1Panel(close){ return rowRank(X=alpha1TS(close), percent=true) - 0.5 } input = exec close from loadTable("dfs://k_minute","k_minute") where date(tradetime) between 2020.01.01 : 2020.01.31 pivot by tradetime, securityid res = alpha1Panel(input) ``` * Alpha 98 因子计算中，同时使用了三个面板数据，分别是vwap, open和vol。不仅各矩阵内部运用了rowRank函数横向截面运算以及m系列垂直滑动窗口计算，矩阵之间也进行了二元运算。用一行代码解决了多维度的复杂的嵌套计算逻辑。 ``` //alpha 98 //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))) def prepareDataForDDBPanel(raw_data, start_time, end_time){ t = select tradetime,securityid, vwap,vol,open from raw_data where date(tradetime) between start_time : end_time return dict(`vwap`open`vol, panel(t.tradetime, t.securityid, [t.vwap, t.open, t.vol])) } @state def alpha98Panel(vwap, open, vol){ return rowRank(X = mavg(mcorr(vwap, msum(mavg(vol, 5), 26), 5), 1..7),percent=true) - rowRank(X=mavg(mrank(9 - mimin(mcorr(rowRank(X=open,percent=true), rowRank(X=mavg(vol, 15),percent=true), 21), 9), true, 7), 1..8),percent=true) } raw_data = loadTable("dfs://k_minute","k_day") start_time = 2020.01.01 end_time = 2020.12.31 input = prepareDataForDDBPanel(raw_data, start_time, end_time) timer alpha98DDBPanel = alpha98Panel(input.vwap, input.open, input.vol) ``` 基于面板数据的因子计算，耗时主要在面板数据准备和因子计算两个阶段。在很多场景下，面板数据准备的耗时可能超过因子计算本身。为解决这个问题，DolphinDB的TSDB引擎提供了宽表存储，即把面板数据直接存储在数据库表中（面板中每一个列存储为表中的每一个列），这样通过SQL查询可以直接获取面板数据，而不需要通过转置行列来获取，从而大大缩短准备面板数据的时间。在本文的第5章中，我们有详细的宽表和竖表存储性能的对比。 ### 3.2 SQL模式 DolphinDB在存储和计算框架上都是基于列式结构，表中的一个列可以直接作为一个向量化函数的输入参数。因此如果一个因子的计算逻辑只涉及股票自身的时间序列数据，不涉及多个股票横截面上的信息，可以直接在SQL中按股票分组，然后在select中调用因子函数计算每个股票在一段时间内的因子值。如果数据在数据库中本身是按股票分区存储的，那么可以非常高效地实现数据库内并行计算。 ``` def sum_diff(x, y){ return (x-y)\(x+y) } @state def factorDoubleEMA(price){ ema_2 = ema(price, 2) ema_4 = ema(price, 4) sum_diff_1000 = 1000 * sum_diff(ema_2, ema_4) return ema(sum_diff_1000, 2) - ema(sum_diff_1000, 3) } res = select tradetime, securityid, `doubleEMA as factorname, factorDoubleEMA(close) as val from loadTable("dfs://k_minute","k_minute") where tradetime between 2020.01.01 : 2020.01.31 context by securityid ``` 在上面的例子中，我们定义了一个因子函数 factorDoubleEMA，只需要用到股票的价格序列信息。我们在 SQL 中通过 `context by` 子句按股票代码分组，然后调用factorDoubleEMA函数，计算每个股票的因子序列。值得注意的是，`context by` 是 DolphinDB SQL 对 `group by` 的扩展，是 DolphinDB 特有的 SQL 语句。`group by` 只适用于聚合计算，也就是说输入长度为n，输出长度是1。`context by` 适用于向量计算，输入长度是n，输出长度也是n。另外因子函数 factorDOubleEMA 除了可以接受一个向量作为输入，也可以接受一个面板数据作为输入。这也是我们前面强调的，因子函数的粒度尽可能细，这样可以应用于很多场景。 时间序列的因子函数非常普遍，talib 中的所有技术分析指标都属于此类函数，因此都可以使用上述SQL方式或面板数据模式来调用。但是3.1中提到的 alpha1 和 alpha98 等因子，涉及到时间序列和横截面两个维度的计算，我们称之为截面因子，无法将因子逻辑封装在一个自定义函数中，然后在一个 SQL 语句中被调用。通常面对截面因子，我们建议将表作为自定义因子函数的入参，内部用 SQL 进行操作，函数最后返回一个表。 ``` //alpha1 def alpha1SQL(t){ res = select tradetime, securityid, mimax(pow(iif(ratios(close) - 1 < 0, mstd(ratios(close) - 1, 20), close), 2.0), 5) as val from t by securityid return select tradetime, securityid, rank(val, percent=true) - 0.5 as val from res context by tradetime } input = select tradetime,securityid, close from loadTable("dfs://k_day_level","k_day") where tradetime between 2010.01.01 : 2010.12.31 alpha1DDBSql = alpha1SQL(input) //alpha98 def alpha98SQL(mutable t){ update t set adv5 = mavg(vol, 5), adv15 = mavg(vol, 15) context by securityid update t set rank_open = rank(X = open,percent=true), rank_adv15 =rank(X=adv15,percent=true) context by date(tradetime) 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 securityid return select tradetime,securityid, `alpha98 as factorname, rank(X =decay7,percent=true)-rank(X =decay8,percent=true) as val from t context by date(tradetime) } input = select tradetime,securityid, vwap,vol,open from loadTable("dfs://k_day_level","k_day") where tradetime between 2010.01.01 : 2010.12.31 alpha98DDBSql = alpha98SQL(input) ``` ### 3.3 不同频率的因子开发举例 不同频率数据的因子，有着不同的特点。本章节将分别举例分钟频、日频、快照、逐笔数据的特点因子，阐述不同频率数据计算因子的最佳实践。 #### 3.3.1 分钟级和日级数据 日级数据的计算，通常是涉及多个截面的复杂计算，在[`3.1 面板数据模式`](#31-面板数据模式)中已展现。对于稍简单的计算，则与分钟级数据的因子相像。 针对分钟级数据，下面的例子是日内收益率偏度的因子计算，对于这类只涉及表内字段的计算，通常使用 SQL 模式，配合 `group by` 语句将计算分组： ``` defg dayReturnSkew(close){ return skew(ratios(close)) } minReturn = select `dayReturnSkew as factorname, dayReturnSkew(close) as val from loadTable("dfs://k_minute_level", "k_minute") where date(tradetime) between 2020.01.02 : 2020.01.31 group by date(tradetime) as tradetime, securityid #output tradetime securityid factorname val ---------- ---------- ------------- ------- 2020.01.02 000019 dayReturnSkew 11.8328 2020.01.02 000048 dayReturnSkew 11.0544 2020.01.02 000050 dayReturnSkew 10.6186 ``` #### 3.3.2 基于快照数据的有状态因子计算 有状态的因子，意为因子的计算需要基于之前的计算结果，如一般的滑动窗口计算，聚合计算等，都是有状态的因子计算。 下例flow这个自定义函数中，参数为四个列字段，运用 `mavg` 滑动平均函数以及 `iif` 条件运算函数，可以直接在SQL中得到因子结果： ``` @state def flow(buy_vol, sell_vol, askPrice1, bidPrice1){ buy_vol_ma = round(mavg(buy_vol, 5*60), 5) sell_vol_ma = round(mavg(sell_vol, 5*60), 5) buy_prop = iif(abs(buy_vol_ma+sell_vol_ma) < 0, 0.5 , buy_vol_ma/ (buy_vol_ma+sell_vol_ma)) spd = askPrice1 - bidPrice1 spd = iif(spd < 0, 0, spd) spd_ma = round(mavg(spd, 5*60), 5) return iif(spd_ma == 0, 0, buy_prop / spd_ma) } res_flow = select TradeTime, SecurityID, `flow as factorname, flow(BidOrderQty[1],OfferOrderQty[1], OfferPrice[1], BidPrice[1]) as val from loadTable("dfs://LEVEL2_Snapshot_ArrayVector","Snap") where date(TradeTime) <= 2020.01.30 and date(TradeTime) >= 2020.01.01 context by SecurityID # output sample TradeTime SecurityID factorname val ----------------------- ---------- ---------- ----------------- 2020.01.22T14:46:27.000 110065 flow 3.7587 2020.01.22T14:46:30.000 110065 flow 3.7515 2020.01.22T14:46:33.000 110065 flow 3.7443 ... ``` #### 3.3.3 快照数据的多档赋权无状态因子计算 计算Level 2的多档快照数据，传统的方式是将多档量价数据存储成为多个列, 再将多档挂单或者报价用 matrix 转换与权重做计算。 更推荐的做法是，将多档数据存储为 array vector，仍旧可以用原来的自定义函数，但是资源消耗包括效率都有提升。 下面的例子是计算多档报价的权重偏度因子，使用 array vector 后计算时间从4秒缩短到2秒。 ``` def mathWghtCovar(x, y, w){ v = (x - rowWavg(x, w)) * (y - rowWavg(y, w)) return rowWavg(v, w) } @state def mathWghtSkew(x, w){ x_var = mathWghtCovar(x, x, w) x_std = sqrt(x_var) x_1 = x - rowWavg(x, w) x_2 = x_1*x_1 len = size(w) adj = sqrt((len - 1) * len) \ (len - 2) skew = rowWsum(x_2, x_1) \ (x_var * x_std) * adj \ len return iif(x_std==0, 0, skew) } //weights: w = 10 9 8 7 6 5 4 3 2 1 //权重偏度因子： resWeight = select TradeTime, SecurityID, `mathWghtSkew as factorname, mathWghtSkew(BidPrice, w) as val from loadTable("dfs://LEVEL2_Snapshot_ArrayVector","Snap") where date(TradeTime) = 2020.01.02 map resWeight1 = select TradeTime, SecurityID, `mathWghtSkew as factorname, mathWghtSkew(matrix(BidPrice0,BidPrice1,BidPrice2,BidPrice3,BidPrice4,BidPrice5,BidPrice6,BidPrice7,BidPrice8,BidPrice9), w) as val from loadTable("dfs://snapshot_SH_L2_TSDB", "snapshot_SH_L2_TSDB") where date(TradeTime) = 2020.01.02 map #output TradeTime SecurityID factorname val ----------------------- ---------- ---------- ------ ... 2020.01.02T09:30:09.000 113537 array_1 -0.8828 2020.01.02T09:30:12.000 113537 array_1 0.7371 2020.01.02T09:30:15.000 113537 array_1 0.6041 ... ``` #### 3.3.4 基于快照数据的分钟聚合 投研中经常需要基于快照数据聚合分钟线的 OHLC ，下例就是这一场景中的通用做法： ``` //基于快照因子的分钟聚合OHLC，vwap tick_aggr = select first(LastPx) as open, max(LastPx) as high, min(LastPx) as low, last(LastPx) as close, sum(totalvolumetrade) as vol,sum(lastpx*totalvolumetrade) as val,wavg(lastpx, totalvolumetrade) as vwap from loadTable("dfs://LEVEL2_Snapshot_ArrayVector","Snap") where date(TradeTime) <= 2020.01.30 and date(TradeTime) >= 2020.01.01 group by SecurityID, bar(TradeTime,1m) ``` #### 3.3.5 逐笔数据 逐笔数据量较大，一般会针对成交量等字段进行计算，下面的例子计算了每天主买成交量占全部成交量的比例，同样使用 SQL 模式，发挥库内并行计算的优势，并使用 csort 语句用来对组内数据按照时间顺序排序： ``` @state def buyTradeRatio(buyNo, sellNo, tradeQty){ return cumsum(iif(buyNo>sellNo, tradeQty, 0))\cumsum(tradeQty) } factor = select TradeTime, SecurityID, `buyTradeRatio as factorname, buyTradeRatio(BuyNo, SellNo, TradeQty) as val from loadTable("dfs://tick_SH_L2_TSDB","tick_SH_L2_TSDB") where date(TradeTime)<2020.01.31 and time(TradeTime)>=09:30:00.000 context by SecurityID, date(TradeTime) csort TradeTime #output TradeTime SecurityID factorname val ------------------- ---------- ---------- ------ 2020.01.08T09:30:07 511850 buyTradeRatio 0.0086 2020.01.08T09:30:31 511850 buyTradeRatio 0.0574 2020.01.08T09:30:36 511850 buyTradeRatio 0.0569 ... ``` ## 4. 生产环境的流式因子计算 在生产环境中，DolphinDB 提供了实时流计算框架。在流计算框架下，用户在投研阶段封装好的基于批量数据开发的因子函数，可以无缝投入交易和投资方面的生产程序中，这就是通常所说的批流一体。使用流批一体可以加速用户的开发和部署。同时流计算框架还在算法的路径上，做了极致的优化，在具有高效开发的优势的同时，又兼顾了计算的高效性能。在这一章中，将会基于实际的状态因子案例，展示实时流计算的使用方法。 DolphinDB 流计算解决方案的核心部件是流计算引擎和流数据表。流计算引擎用于时间序列处理、横截面处理、窗口处理、表关联、异常检测等操作。流数据表可以看作是一个简化版的消息中间件，或者说是消息中间件中的一个主题（topic），可以往其发布（publish）数据，也可以从其订阅（subscribe）数据。流计算引擎和流数据表均继承于 DolphinDB 的数据表（table），因此都可以通过 `append!` 函数往其注入数据。流计算引擎的输出也是数据表的形式，因此多个计算引擎可以跟搭积木一样自由组合，形成流式处理的流水线。 ![因子数据流处理流程图](script/factorPractice/images/factor_streams.png?inline=false) ### 4.1 流式增量计算 金融方面的原始数据和计算指标，在时间上通常有延续性的关系。以最简单的五周期移动均线 `mavg(close,5)` 为例，当新一个周期的数据传入模型时，可以将之前最远的第五周期值从 sum 中减出，再把最新一个周期的值加入 sum ，这样就不必每个周期只更新一个值时都重算一遍 sum 。这种增量计算是流计算的核心，可以大大降低实时计算的延时。DolphinDB内置了大量量化金融中需要用到的基本算子，并为这些算子实现了高效的增量算法。不仅如此，DolphinDB还支持自定义函数的增量实现。在前一章节中，部分自定义的因子函数加了修饰符 `@state` ，表示该函数支持增量计算。 #### 4.1.1 主买成交量占比因子的流式处理 [第三章3.3.5](#335-逐笔数据)的逐笔数据因子的例子展示了主买成交量占比因子（buyTradeRatio）的批量实现方式。以下代码演示如何使用响应式状态引擎（reactive state engine）来实现该因子的流式增量计算。 ``` @state def buyTradeRatio(buyNo, sellNo, tradeQty){ return cumsum(iif(buyNo>sellNo, tradeQty, 0))\cumsum(tradeQty) } tickStream = table(1:0, `SecurityID`TradeTime`TradePrice`TradeQty`TradeAmount`BuyNo`SellNo, [SYMBOL,DATETIME,DOUBLE,INT,DOUBLE,LONG,LONG]) result = table(1:0, `SecurityID`TradeTime`Factor, [SYMBOL,DATETIME,DOUBLE]) factors = <[TradeTime, buyTradeRatio(BuyNo, SellNo, TradeQty)]> demoEngine = createReactiveStateEngine(name="demo", metrics=factors, dummyTable=tickStream, outputTable=result, keyColumn="SecurityID") ``` 上述代码创建了一个名为demo的响应式状态引擎，SecurityID作为分组键，输入的消息格式同内存表tickStream。需要计算的指标定义在factors中，其中1个是输入表中的原始字段TradeTime，另一个是我们需要计算的因子的函数表示。输出到内存表result，除了在factors中定义的指标外，输出结果还会添加分组键。请注意，自定义的因子函数跟批计算中的完全一致！创建完引擎之后，我们即可往引擎中插入几条数据，并观察计算结果。 ``` insert into demoEngine values(`000155, 2020.01.01T09:30:00, 30.85, 100, 3085, 4951, 0) insert into demoEngine values(`000155, 2020.01.01T09:30:01, 30.86, 100, 3086, 4951, 1) insert into demoEngine values(`000155, 2020.01.01T09:30:02, 30.80, 200, 6160, 5501, 5600) select * from result SecurityID TradeTime Factor ---------- ------------------- ------ 000155 2020.01.01T09:30:00 1 000155 2020.01.01T09:30:01 1 000155 2020.01.01T09:30:02 0.5 ``` 从这个例子可以看出，在DolphinDB中实现因子的流式增量计算非常简单。如果在投研阶段，已经用我们推荐的方式自定义了一个因子函数，在生产阶段只要程序性的创建一个流式计算引擎即可实现目标。这也是为什么我们一再强调，自定义的因子函数必须使用规范的接口，而且只包含核心的因子逻辑，不用考虑并行计算等问题。 #### 4.1.2 大小单的流式处理 资金流分析是逐笔委托数据的一个重要应用场景。在实时处理逐笔数据时，大小单的统计是资金流分析的一个具体应用。大小单在一定程度上能反映主力、散户的动向。但在实时场景中，大小单的生成有很多难点：（1） 大小单的计算涉及历史状态，如若不能实现增量计算，当计算下午的数据时，可能需要回溯有关这笔订单上午的数据，效率会非常低下。 （2）计算涉及至少两个阶段。在第一阶段需要根据订单分组，根据订单的累计成交量判断大小单，在第二阶段要根据股票来分组，统计每个股票的大小单数量及金额。 大小单是一个动态的概念。一个小单在成交量增加后可能变成一个大单。DolphinDB的两个内置函数`dynamicGroupCumsum`和`dynamicGroupCumcount`用于对动态组的增量计算。完整的代码请参考：[章节附件4.1.2 大小单的流式处理](script/factorPractice/appendix_4.1.2_streamComputationOfSmallInflowRate_main.dos)。 ``` @state def factorSmallOrderNetAmountRatio(tradeAmount, sellCumAmount, sellOrderFlag, prevSellCumAmount, prevSellOrderFlag, buyCumAmount, buyOrderFlag, prevBuyCumAmount, prevBuyOrderFlag){ cumsumTradeAmount = cumsum(tradeAmount) smallSellCumAmount, bigSellCumAmount = dynamicGroupCumsum(sellCumAmount, prevSellCumAmount, sellOrderFlag, prevSellOrderFlag, 2) smallBuyCumAmount, bigBuyCumAmount = dynamicGroupCumsum(buyCumAmount, prevBuyCumAmount, buyOrderFlag, prevBuyOrderFlag, 2) f = (smallBuyCumAmount - smallSellCumAmount) \ cumsumTradeAmount return smallBuyCumAmount, smallSellCumAmount, cumsumTradeAmount, f } def createStreamEngine(result){ tradeSchema = createTradeSchema() result1Schema = createResult1Schema() result2Schema = createResult2Schema() engineNames = ["rse1", "rse2", "res3"] cleanStreamEngines(engineNames) metrics3 = <[TradeTime, factorSmallOrderNetAmountRatio(tradeAmount, sellCumAmount, sellOrderFlag, prevSellCumAmount, prevSellOrderFlag, buyCumAmount, buyOrderFlag, prevBuyCumAmount, prevBuyOrderFlag)]> rse3 = createReactiveStateEngine(name=engineNames[2], metrics=metrics3, dummyTable=result2Schema, outputTable=result, keyColumn="SecurityID") metrics2 = <[BuyNo, SecurityID, TradeTime, TradeAmount, BuyCumAmount, PrevBuyCumAmount, BuyOrderFlag, PrevBuyOrderFlag, factorOrderCumAmount(TradeAmount)]> rse2 = createReactiveStateEngine(name=engineNames[1], metrics=metrics2, dummyTable=result1Schema, outputTable=rse3, keyColumn="SellNo") metrics1 = <[SecurityID, SellNo, TradeTime, TradeAmount, factorOrderCumAmount(TradeAmount)]> return createReactiveStateEngine(name=engineNames[0], metrics=metrics1, dummyTable=tradeSchema, outputTable=rse2, keyColumn="BuyNo") } ``` 自定义函数`factorSmallOrderNetAmountRatio`是一个状态因子函数，用于计算小单的净流入资金占总的交易资金的比例。`createStreamEngine`创建流式计算引擎。我们一共创建了3个级联的响应式状态引擎，后一个作为前一个的输出，因此从最后一个引擎开始创建。前两个计算引擎rse1和rse2分别以买方订单号（BuyNo)和卖方订单号（SellNo）作为分组键，计算每个订单的累计交易量，并以此区分是大单或小单。第三个引擎rse3把股票代码（SecurityID）作为分组键，统计每个股票的小单净流入资金占总交易资金的比例。下面我们输入一些样本数据来观察流计算引擎的运行。 ``` result = createResultTable() rse = createStreamEngine(result) insert into rse values(`000155, 1000, 1001, 2020.01.01T09:30:00, 20000) insert into rse values(`000155, 1000, 1002, 2020.01.01T09:30:01, 40000) insert into rse values(`000155, 1000, 1003, 2020.01.01T09:30:02, 60000) insert into rse values(`000155, 1004, 1003, 2020.01.01T09:30:03, 30000) select * from result SecurityID TradeTime smallBuyOrderAmount smallSellOrderAmount totalOrderAmount factor ---------- ------------------- ------------------- -------------------- ---------------- ------ 000155 2020.01.01T09:30:00 20000 20000 20000 0 000155 2020.01.01T09:30:01 60000 60000 60000 0 000155 2020.01.01T09:30:02 0 120000 120000 -1 000155 2020.01.01T09:30:03 30000 150000 150000 -0.8 ``` #### 4.1.3 复杂因子Alpha #1流式计算的快捷实现 从前一个大小单的例子可以看到，有些因子的流式实现比较复杂，需要创建多个引擎进行流水线处理来完成。完全用手工的方式来创建多个引擎其实是一件耗时的工作。如果输入的指标计算只涉及一个分组键，DolphinDB提供了一个解析引擎[`streamEngineParser`](https://www.dolphindb.cn/cn/help/200/FunctionsandCommands/FunctionReferences/s/streamEngineParser.html)来解决此问题。下面我们以第三章3.1面板数据模式的alpha #1因子为例，展示`streamEngineParser`的使用方法。完整代码参考[Alpha #1流式计算](script/factorPractice/appendix_4.1.3_StreamComputationOfAlpha1Factor_main.dos)。以下为核心代码。 ``` @state def alpha1TS(close){ return mimax(pow(iif(ratios(close) - 1 < 0, mstd(ratios(close) - 1, 20),close), 2.0), 5) } def alpha1Panel(close){ return rowRank(X=alpha1TS(close), percent=true) - 0.5 } inputSchema = table(1:0, ["SecurityID","TradeTime","close"], [SYMBOL,TIMESTAMP,DOUBLE]) result = table(10000:0, ["TradeTime","SecurityID", "factor"], [TIMESTAMP,SYMBOL,DOUBLE]) metrics = <[SecurityID, alpha1Panel(close)]> streamEngine = streamEngineParser(name="alpha1Parser", metrics=metrics, dummyTable=inputSchema, outputTable=result, keyColumn="SecurityID", timeColumn=`tradetime, triggeringPattern='keyCount', triggeringInterval=4000) ``` 因子alpha1实际上包含了时间序列处理和横截面处理，需要响应式状态引擎和横截面引擎串联来处理才能完成。但以上代码仅仅使用了streamEngineParser就创建了全部引擎，大大简化了创建过程。 前面三个例子展示了DolphinDB如何通过流计算引擎实现因子在生产环境中的增量计算。值得注意的是，流式计算时直接使用了投研阶段生成的核心因子代码，这很好的解决了传统金融分析面临的批流一体问题。在传统的研究框架下，用户往往需要对同一个因子计算逻辑写两套代码，一套用于在历史数据上建模、回测，另外一套专门处理盘中传入的实时数据。这是因为数据传入程序的形状(机制)不统一，又甚至是编程语言也无法统一。比如研究分析使用了 python 或者 R，在 python 或 R 的研究程序确定模型和参数后，生产交易的程序必须用 C++ 再实现这套模型，才能保证交易时的执行效率。在两套代码完成后，还要再校验它们计算出来的结果是否一致。这样的业务流程毫无疑问加重了研究员和程序员们的负担，也让基金经理们没法更快地让新交易思路迭代上线。在DolphinDB的流式计算中，实时行情订阅、行情数据收录、交易实时计算、盘后研究建模，全都用同一套代码完成，保证在历史回放和生产交易当中数据完全一致。 除了三个例子中用到的响应式状态引擎(reactive state engine)和横截面引擎（cross sectional engine），DolphinDB 还提供了多种流数据处理引擎包括做流表连接的 asof join engine，equal join engine，lookup join engine，window join engine ，时间序列聚合引擎（time series engine），异常检测引擎(anomaly detection engine)，会话窗口引擎（session window engine）等。 ### 4.2 数据回放 前一节我们介绍了因子计算的批流一体实现方案，简单地说，就是一套代码（自定义的因子函数），两种引擎（批计算引擎和流计算引擎）。事实上，DolphinDB提供一种更为简洁的批流一体实现方案，那就是在历史数据建模时，通过数据回放，也用流引擎来实现计算。 在第三章中介绍了用[SQL语句方式批处理计算factorDoubleEMA因子](#32-sql模式)的例子，这里介绍如何使用流计算的方式回放数据，计算 factorDoubleEMA 的因子值。全部代码参考[章节附件4.2 流计算factorDoubleEMA因子](script/factorPractice/appendix_4.2_streamComputationOfDoubleEmaFactor_main.dos) ``` //创建流引擎，并传入因子算法factorDoubleEMA factors = <[TradeTime, factorDoubleEMA(close)]> demoEngine = createReactiveStateEngine(name=engineName, metrics=factors, dummyTable=inputDummyTable, outputTable=resultTable, keyColumn="SecurityID") //demo_engine订阅snapshotStreamTable流表 subscribeTable(tableName=snapshotSharedTableName, actionName=actionName, handler=append!{demoEngine}, msgAsTable=true) //创建播放数据源供replay函数历史回放；盘中的时候，改为行情数据直接写入snapshotStreamTable流表 inputDS = replayDS(, `securityid, HASH,10) def factorDoubleEMAMap(table){ return select tradetime, securityid, `doubleEMA as factorname, factorDoubleEMA(close) as val from table context by securityid map } res = mr(ds,factorDoubleEMAMap,,unionAll) ``` #### 7.3.3 通过submitJob提交任务 之前的两种并行计算都是在前台执行的，并行度是由参数 `localExecutors` 设置。而有些作业可能很大，或者用户不想影响前台使用，此时可以通过 `submitJob` 提交任务。submitJob的并行度由 `maxBatchJobWorker` 参数设置。由于后台作业之间是独立的，通常不需要返回到前端的任务都推荐用后台提交 `submitJob` 的形式。 仍旧以 dayReturnSkew 因子为例。通常我们是需要将因子写入因子库表的，此时可以将整一个过程提交几个后台作业去执行，而在客户端中，同时可以继续做其他计算。由于此例存入的因子库的分区是按月和因子名VALUE分区，故此时应按照月份去提交作业。这样既可以并行写入不会冲突，又可以将作业提交到后台，不影响前台提交其他任务。 ``` def writeDayReturnSkew(dBegin,dEnd){ dReturn = select `dayReturnSkew as factorname, dayReturnSkew(close) as val from loadTable("dfs://k_minute_level", "k_minute") where date(tradetime) between dBegin : dEnd group by date(tradetime) as tradetime, securityid //写入因子库 loadTable("dfs://K_FACTOR_VERTICAL","factor_k").append!(dReturn) } for (i in 0..11){ dBegin = monthBegin(temporalAdd(2020.01.01,i,"M")) dEnd = monthEnd(temporalAdd(2020.01.01,i,"M")) submitJob("writeDayReturnSkew","writeDayReturnSkew_"+dBegin+"_"+dEnd, writeDayReturnSkew,dBegin,dEnd) } ``` ### 7.4 内存管理 内存管理一直是运维人员和研究人员关注的重中之重，本节将从批和流两个角度简单介绍如何在DolphinDB中高效地使用内存。更多有关内存管理的详细内容，请参阅[`DolphinDB内存管理教程`](memory_management.md)。 在配置 DolphinDB 环境时，计算和事务的内存占用可在单节点的 ”dolphindb.cfg” 或集群的 cluster.cfg 中，通过参数”maxMemSize“配置单节点最大可用内存。 - 批处理的内存管理 如[章节3.2中的例子](#32-sql模式)，若对半年的快照数据做操作，批处理方式的中间变量占用内存达到21GB，如果设置的内存小于21GB，则会报Out of Memory错误。这种情况下可以将作业拆分后再提交。 在调试大任务量的计算完成后，可通过 undef 函数将变量赋值为 NULL，或者关闭 session 来及时释放变量的内存。 - 流计算的内存管理 如[章节4.2中的例子](#42-数据回放)，代码中对中间流表调用了函数 `enableTableShareAndPersistence` 以持久化，指定缓存开始为80万行。当流表数据量超过80万行时，旧的数据会持久化到磁盘上，以空出内存里的空间供新数据写入，这样该流表就可以连续处理远远超过80万行的数据。 ### 7.5 权限管理 因子数据是非常重要的数据，一般来说，用户并不能随意访问所有因子，因此需要对因子数据做好权限管理。DolphinDB database 提供了强大、灵活、安全的权限控制系统，可以满足因子库表级，函数视图级的管理。更多有关权限管理的详细内容，请参考[权限管理教程](ACL_and_Security.md)。 在实际的生产中通常使用以下三种管理方式： - 研发人员是管理员，完全掌握数据库 这种情况可以授予研发组 DB_OWNER 的权限(创建数据库并管理其创建的数据库的权限)，使其可以自行创建数据库、表，并对自己创建的数据、表进行权限管理。 ``` login("admin", "123456"); createUser("user1", "passwd1") grant("user1", DB_OWNER) ``` - 运维人员管理数据库，研发人员只有库表的读写权限 这种情况，数据库管理人员可以将数据表的权限授予给因子研发人员，或者创建一个group组，将权限授予这个组，再将需要权限的人员添加到这个组中统一进行管理。 ``` //以用户的方式进行授权 createUser("user1", "passwd1") grant("user1", TABLE_READ, "dfs://db1/pt1") //以group的方式进行授权 createGroup("group1name", "user1") grant("group1name", TABLE_READ, "dfs://db1/pt1") ``` - 研发人员只可读部分而非全库表数据权限 DolphinDB 本身并不直接支持表内数据级的权限控制，但是通过DolphinDB本身灵活的权限控制，我们可以通过其他方式来实现表内数据级的权限控制。 这里我们可以通过对用户授予functionview 权限 VIEW_EXEC 这种方式来实现表内数据级的权限控制。 完整代码参考：[章节附件7.5.3 因子表权限控制](script/factorPractice/appendix_7.5.3_factorTableControll.dos)。通过这份代码，用户"u1"虽然没有表的读权限，但是可以获得表内factor1因子的数据。 ``` //创建用户u1,我们想授予u1 只能读取因子factor1的权限 createUser("u1", "111111") //定义只取因子的函数 def getFactor1Table(){ t=select * from loadTable("dfs://db1","factor") where factor_name="factor1"; return t; } //将函数保存到系统中 addFunctionView(getFactor1Table) //将该函数权限授予用户u1 grant("u1", VIEW_EXEC, "getFactor1Table"); //注意新授予的权限，用户需要重新登录才能加载 factor1_tab=getFactor1Table() ``` ### 7.6 任务管理 因子计算的任务通常分为全量计算所有因子任务、交互式单因子重算任务、所有因子增量计算任务这三种，本章会对每一种因子计算任务进行详细介绍。 因子任务可以通过以下三种方式执行: - (1) 通过交互的方式执行。 - (2) 通过 [submitJob](https://www.dolphindb.cn/cn/help/FunctionsandCommands/FunctionReferences/s/submitJob.html) 提交一个Job来执行。 - (3) 通过 [scheduleJob](scheduledJob.md) 提交一个定时任务来进行周期性的执行。 #### 7.6.1 全量计算 因子的全量跑批任务，通常是系统初始化因子数据时的一次性任务，或者较长周期进行一次的任务，这类任务可以通过单次触发或者定时任务(scheduleJob)的方式进行管理。 * 单次触发的任务：这种任务可以通过 gui 直接执行，也可以通过 api 来调用命令，最好的方式是通过 `submitJob` 函数提交任务。通过 `submitJob` 提交的任务，会提交到服务器的Job 队列中执行，不再受客户端影响，并且可以通过 [getRecentJobs](https://www.dolphindb.cn/cn/help/FunctionsandCommands/FunctionReferences/g/getRecentJobs.html) 观察到任务是否完成。 ``` //对于跑批的任务封装函数 def bacthExeCute(){} // 通过summitjob进行提交 submitJob("batchTask","batchTask", bacthExeCute) ``` * 周期性任务：如果计算的因子频率较低需要每天盘后或者其他周期定期全部重算一次，那我们可以使用定时任务(ScheduleJob)的方式进行管理。 ``` //设置一段时间每天执行 scheduleJob(jobId=`daily, jobDesc="Daily Job 1", jobFunc=bacthExeCute, scheduleTime=17:23m, startDate=2018.01.01, endDate=2018.12.31, frequency='D') ``` #### 7.6.2 因子运维管理 在因子研发过程中，当碰到因子算法、参数调整的情况，我们会需要对该因子进行重新计算，同时需要将计算的新的因子数据更新到数据库中，对于因子更新的频率通常我们有两种方式： * 因子的数据频率较高，数据量很大 因子的数据频率较高，数据量很大时，我们推荐在因子数据分区时拉长时间维度，以因子名进行VALUE分区。这样可以使每个因子的数据独立的保存在一个分区中，控制分区大小在一个合适的范围。当我们碰到因子重算的情况，便可以用 [`dropPartition`](https://www.dolphindb.cn/cn/help/FunctionsandCommands/CommandsReferences/d/dropPartition.html) 函数先删除这个因子所对应的分区数据，然后直接重算这个因子并保存到数据表中。 * 因子的数据频率较低，因子的总数据量较小 当因子的数据频率较低，因子的总数据量较小时，如若将每个因子划分为独立的分区会使得每个分区特别小，而过小的分区可能会影响写入速度。这种情况下，我们可以按照因子 HASH 分区。使用 `update!` 来进行因子数据更新操作，或使用 `upsert` 来进行插入更新操作。此外，对于 TSDB 引擎，可以设置参数 `keepDuplicates=LAST` , 此时可以直接使用 `append!` 或者 `tableInsert` 插入数据，从而达到效率更高的更新数据的效果。 `update!` , `upsert` 以及 TSDB 引擎特殊设置下的直接 `append!` 覆盖数据，这三种更新操作都建议在数据量较小，且更新不频繁的情况下使用。对于需要大量因子重算的数据更新的场景，我们推荐使用**单因子独立分区**的方式。当因子重算时先用`dropPartition`函数删除因子所在分区，再重算写入新因子入库。 ## 8. 实际案例 ### 8.1 日频因子 日频的数据，一般是由逐笔数据或者其他高频数据聚合而成。日频的数据量不大，在日频数据上经常会计算一些动量因子，或者一些复杂的需要观察长期数据的因子。因此在分区考虑上，建议按年分区即可。在因子计算上，日频因子通常会涉及时间和股票多个维度，因此建议用面板模式计算。当然也可以根据不同存储模式，选择不同的计算模式。 在[章节附件8.1 日频因子全流程代码汇总](script/factorPractice/appendix_8.1_case1_daily.dos)中，模拟了 10 年 4000 只股票的数据，总数据量压缩前大约为 1 GB。代码中会展现上述教程中所涉及日频因子的最佳实践，因子包括 Alpha 1、Alpha 98 ，以及不同计算方式（面板或者SQL模式）写入单值模型、多值模型的最佳实践。 ### 8.2 分钟频因子 分钟频的数据，一般是从逐笔数据或快照数据合成而来。分钟频的数据相比日频的数据较大，在分区设计上建议按月VALUE分区，股票HASH的组合分区。在分钟频的数据上，一般会计算日内的收益率等因子。对于这类因子，建议使用SQL的方式以字段作为参数。很多时候，会将投研的因子，在每日收盘之后，增量做所有因子的计算，此时，也需要对于每日增量的因子做工程化管理。建议将所有此类因子用维度表做一个维护，用定时作业将这些因子批量做计算。 在[章节附件8.2 分钟频因子全流程代码汇总](script/factorPractice/appendix_8.2_case2_minute.dos)中，模拟了一年4000只股票的数据，总数据量压缩前大约20GB。其中，会展现上述教程中所有涉及分钟频率的因子的最佳实践，因子包括日内收益偏度因子，factorDoubleEMA等因子，，后续将因子写入单值模型、多值模型的全过程，以及每日增量计算所有因子的工程化最佳实践。 ### 8.3 快照因子 快照数据，一般指3s一条的多档数据。在实际生产中，往往会根据这样的数据产生实时的因子，或根据多档报价、成交量计算，或根据重要字段做计算。这一类因子，推荐使用字段名作为自定义函数的参数。除此之外，由于快照数据的多档的特殊性，普通存储会占用很大的空间，故在存储模式上，我们也推荐将多档数据存为ArrayVector的形式。如此一来，既能节省磁盘空间，又能使代码简洁，省去选取多个重复字段的困扰。 在[章节附件8.3 快照因子全流程代码汇总](script/factorPractice/appendix_8.3_case3_snapshot.dos)中，模拟数据生成了20天快照数据，并将其存储为了普通快照数据和ArrayVector快照数据两种。代码中也展示了对于有状态因子flow和无状态因子权重偏度的在流批一体中的最佳实践。 ### 8.4 逐笔因子 逐笔成交数据，是交易所提供的最详细的每一笔撮合成交数据。每3秒发布一次，每次提供这3秒内的所有撮合记录。涉及逐笔成交数据的因子都是高频因子，推荐调试建模阶段可以在小数据量上使用批处理计算。一旦模型定型，就可以用批处理中同样的计算代码，迁移到流计算中实时处理(这就是所谓的批流一体)，比批处理方式节省内存，同时实时性也更高，模型迭代也更快。 在[章节附件8.4 逐笔因子全流程代码汇总](script/factorPractice/appendix_8.4_case4_streamTick.dos)中，会展现上述教程中所有涉及逐笔成交数据的因子计算、流计算。 ## 9. 总结 用DolphinDB来进行因子的计算时，可选择面板和SQL两种方式来封装因子的核心逻辑。面板方式使用矩阵来计算因子，实现思路非常简练；而SQL方式要求投研人员使用向量化的思路进行因子开发。无论哪种方式，DolphinDB均支持批流一体的实现。DolphinDB内置了相关性和回归分析等计算工具，可分析因子的有效性，可对多因子建模。 在因子库的规划上，如果追求灵活性，建议采用单值纵表模型。如果追求效率和性能，推荐使用TSDB引擎，启用多值宽表模式，标的（股票代码）作为表的列。 最后，基于大部分团队的IT和投研相对独立的事实，给出了在代码管理上的工程化方案，投研团队通过模块和自定义函数封装核心因子业务逻辑，IT团队则维护框架代码。同时利用权限模块有效隔离各团队之间的数据访问权限。 ## 附录 [章节附件2.1 逐笔数据建库建表](script/factorPractice/appendix_2.1_createTickDbAndTable_main.dos) [章节附件2.2 快照数据建库建表](script/factorPractice/appendix_2.2_createSnapshotDbAndTable_main.dos) [章节附件2.3 k线数据建库建表](script/factorPractice/appendix_2.3_createTableKMinute_main.dos) [章节附件4.1.2 流计算大小单因子](script/factorPractice/appendix_4.1.2_streamComputationOfSmallInflowRate_main.dos) [章节附件4.1.3 Alpha #1流式计算](script/factorPractice/appendix_4.1.3_StreamComputationOfAlpha1Factor_main.dos) [章节附件4.2 流计算doubleEma因子](script/factorPractice/appendix_4.2_streamComputationOfDoubleEmaFactor_main.dos) [章节附件4.3.1 python接口订阅流数据](script/factorPractice/appendix_4.3.1_python_callback_handler_subscribing_stream_main.py) [章节附件4.3.2 通过ZMQ消息队列收取DolphinDB推送来的流数据](script/factorPractice/appendix_4.3.2_zmq_consuming_ddb_stream_main.py) [章节附件4.3.3 流计算因子结果推送到外部ZMQ消息队列](script/factorPractice/appendix_4.3.3_streamComputationOfDoubleEmaFactorPublishingOnZMQ_main.dos) [章节附件5.1 因子存储模拟测试:](script/factorPractice/appendix_5.1_factorDataSimulation.zip) [章节附件5.2, 5.3 因子查询测试脚本 :](script/factorPractice/appendix_5.2_factorQueryTest.dos) [章节附件6.1 因子向量化回测](script/factorPractice/appendix_6.1_vectorisedFactorBacktest_main.dos) [章节附件7.2 单元测试](script/factorPractice/appendix_7.2_doubleEMATest.dos) [章节附件7.5.3 因子表权限控制](script/factorPractice/appendix_7.5.3_factorTableControll.dos) [章节附件8.1 日频因子全流程代码汇总](script/factorPractice/appendix_8.1_case1_daily.dos) [章节附件8.2 分钟频因子全流程代码汇总](script/factorPractice/appendix_8.2_case2_minute.dos) [章节附件8.3 快照因子全流程代码汇总](script/factorPractice/appendix_8.3_case3_snapshot.dos) [章节附件8.4 逐笔因子全流程代码汇总](script/factorPractice/appendix_8.4_case4_streamTick.dos) [所有代码附件目录](script/factorPractice)