面板数据处理

context by 是 DolphinDB 独有的功能，是对标准 SQL 语句的拓展，我们可以通过 context by 子句实现的分组计算功能来简化对数据面板的操作。

SQL 的 group by 子句将数据分成多组，每组产生一个值，也就是一行。因此使用 group by 子句后，行数一般会大大减少。

在对面板数据进行分组后，每一组数据通常是时间序列数据，譬如按股票分组，每一个组内的数据是一个股票的价格序列。处理面板数据时，有时候希望保持每个组的数据行数，也就是为组内的每一行数据生成一个值。例如，根据一个股票的价格序列生成回报序列，或者根据价格序列生成一个移动平均价格序列。其它数据库系统（例如 SQL Server, PostgreSQL），用窗口函数（window function）来解决这个问题。DolpinDB 引入了 context by 子句来处理面板数据。context by 与 group by, pivot by 一起组成了 DolphinDB 分组数据处理系统。它与窗口函数相比，除了语法更简洁以外，表达能力上也更强大，具体表现在：

• 不仅能与 select 配合查询数据，也可以与 update 配合更新数据。

• 绝大多数数据库系统在窗口函数中只能使用表中现有的字段分组。context by 子句可以使用任何现有字段和计算字段。

• 绝大多数数据库系统的窗口函数仅限于少数几个函数。context by 不仅不限制使用的函数，而且可以使用任意表达式，譬如多个函数的组合。

• context by 可以与 having 子句配合使用，以过滤每个组内部的行。

例：

(1) 按股票代码进行分组，应用序列函数计算每一只股票的前后交易量比率，进行对比：

select timestamp, sym, price, ratios(volume) ,volume from t context by sym;

// output
timestamp sym price  ratios_volume volume
--------- --- ------ ------------- ------
09:34:57  C   50.6                  2200
09:34:59  C   50.62  0.86           1900
09:35:01  C   50.63  1.106          2100
09:35:02  C   50.64  1.52           3200
09:35:01  IBM 174.97                2500
09:35:02  IBM 175.02 3.52           8800
09:34:57  MS  29.46                 6800
09:34:59  MS  29.48  0.79           5400
09:35:01  MS  29.5   0.24           1300

(2) 结合滑动窗口函数，计算每只股票在 3 次数据更新中的平均价格：

select *, mavg(price,3) from t context by sym;

// output
sym timestamp price  volume mavg_price
--- --------- ------ ------ -----------
C   09:34:57  50.60  2200
C   09:34:59  50.62  1900
C   09:35:01  50.63  2100   50.62
C   09:35:02  50.64  3200   50.63
IBM 09:35:01  174.97 2500
IBM 09:35:02  175.02 8800
MS  09:34:57  29.46  6800
MS  09:34:59  29.48  5400
MS  09:35:01  29.50  1300   29.48

(3) 结合累计窗口函数，计算每只股票在每一次的数据更新中最大交易量：

select timestamp, sym, price,volume, cummax(volume) from t context by sym;

// output
timestamp sym price  volume cummax_volume
--------- --- ------ ------ -------------
09:34:57  C   50.6   2200   2200
09:34:59  C   50.62  1900   2200
09:35:01  C   50.63  2100   2200
09:35:02  C   50.64  3200   3200
09:35:01  IBM 174.97 2500   2500
09:35:02  IBM 175.02 8800   8800
09:34:57  MS  29.46  6800   6800
09:34:59  MS  29.48  5400   6800
09:35:01  MS  29.5   1300   6800

(4) 应用聚合函数，计算每只股票在每分钟中的最大交易量：

select *, max(volume) from t context by sym, timestamp.minute();

// output
sym timestamp price  volume max_volume
--- --------- ------ ------ ----------
C   09:34:57  50.61  2200   2200
C   09:34:59  50.62  1900   2200
C   09:35:01  50.63  2100   3200
C   09:35:02  50.64  3200   3200
IBM 09:35:01  174.97 2500   8800
IBM 09:35:02  175.02 8800   8800
MS  09:34:57  29.46  6800   6800
MS  09:34:59  29.48  5400   6800
MS  09:35:01  29.5   1300   1300

pivot by 是 DolphinDB 的独有功能，是对标准 SQL 语句的拓展，可将数据表中某列的内容按照两个维度整理，产生数据表或矩阵。

通过应用 pivot by 子句，可以对于数据表 t 进行重新排列整理：每行为一秒钟，每列为一只股票，既能够了解单个股票每个时刻的变化，也可以了解各股票之间的差异。

如：对比同一时间段不同股票的交易价格：

select price from t pivot by timestamp, sym;

// output
timestamp C     IBM    MS
--------- ----- ------ -----
09:34:57  50.6         29.46
09:34:59  50.62        29.48
09:35:01  50.63 174.97 29.5
09:35:02  50.64 175.02

pivot by 还可以与聚合函数一起使用。比如，将数据中每分钟的平均收盘价转换为数据表：

select avg(price) from t where sym in `C`IBM pivot by minute(timestamp) as minute, sym;

// output
minute C      IBM
------ ------ -------
09:34m 50.61
09:35m 50.635 174.995

pivot by 与 select 子句一起使用时返回一个表，而和 exec 语句一起使用时返回一个矩阵：

resM = exec avg(price) from t where sym in `C`IBM pivot by minute(timestamp) as minute, sym;
resM

// output
       C      IBM
       ------ -------
09:34m|50.61
09:35m|50.635 174.995

typestr(resM)

// output
FAST DOUBLE MATRIX

下文例举了矩阵形式面板数据的常用操作。

(1) 通过序列函数，对每个股票的相邻价格进行比较。

price = panel(t.timestamp, t.sym, t.price);
deltas(price);

// output
         C    IBM  MS
         ---- ---- -----
09:34:57|
09:34:59|0.02      0.02
09:35:01|0.01      0.02
09:35:02|0.01 0.05

(2) 结合滑动窗口函数，计算每只股票在每 2 次数据更新中的平均价格。

mavg(price,2);

// output
         C      IBM      MS
         ------ ------ -----------
09:34:57|
09:34:59|50.61         29.47
09:35:01|50.63  174.97 29.49
09:35:02|50.63  175.00 29.50

(3) 结合累计窗口函数，计算每只股票中价格的排序。

cumrank(price);

// output
         C IBM MS
         - --- --
09:34:57|0     0
09:34:59|1     1
09:35:01|2 0   2
09:35:02|3 1

(4) 通过聚合函数, 得到每只股票中的最低价格。

min(price);

// output
[50.60,174.97,29.46]

(5) 通过聚合函数，得到每一个同时间段的最低股票价格。

rowMin(price);

 // output
[29.46,29.48,29.5,50.64]

普通矩阵进行二元运算时，按照对应元素分别进行计算，需要保持维度 (shape) 一致，DolphinDB 提供了矩阵对齐的方法，使得矩阵计算不再受维度的限制。

在 1.30.20/2.00.8 版本前，用户需要通过 indexedMatrix 和 indexedSeries 来支持矩阵的对齐运算，其标签必须是严格递增的。

• indexedMatrix：以行列标签为索引的矩阵。
• indexedSeries：带索引标签的向量。

indexedMatrix 和 indexedSeries 在进行二元运算时，系统会自动以 "outer join" 的方式对齐，然后进行运算。

1.30.20/2.00.8 版本后，DolphinDB 提供了用于矩阵对齐的函数 align，拓展了标签矩阵的对齐功能，使矩阵对齐和运算更加灵活。

(1) indexedSeries 之间的对齐运算

两个 indexedSeries 进行二元操作，会根据 index 进行对齐再做计算。

index1 = 2020.11.01..2020.11.06;
value1 = 1..6;
s1 = indexedSeries(index1, value1);

index2 = 2020.11.04..2020.11.09;
value2 =4..9;
s2 = indexedSeries(index2, value2);

s1+s2;

 // output
           #0
           --
2020.11.01|
2020.11.02|
2020.11.03|
2020.11.04|8
2020.11.05|10
2020.11.06|12
2020.11.07|
2020.11.08|
2020.11.09|

(2) indexedMatrix 之间的对齐运算

两个 indexedMatrix 进行二元操作，其对齐的方法和 indexedSeries 一致。

id1 = 2020.11.01..2020.11.06;
m1 = matrix(1..6, 7..12, 13..18).rename!(id1, `a`b`d)
m1.setIndexedMatrix!()

id2 = 2020.11.04..2020.11.09;
m2 = matrix(4..9, 10..15, 16..21).rename!(id2, `a`b`c)
m2.setIndexedMatrix!()

m1+m2;

 // output
           a  b  c d
           -- -- - -
2020.11.01|         
2020.11.02|         
2020.11.03|         
2020.11.04|8  20    
2020.11.05|10 22    
2020.11.06|12 24    
2020.11.07|         
2020.11.08|         
2020.11.09|

(3) indexedSeries 和 indexedMatrix 之间的对齐运算

indexedSeries 与 indexedMatrix 进行二元操作，会根据行标签进行对齐，indexedSeries 与 indexedMatrix 的每列进行计算。

m1=matrix(1..6, 11..16);
m1.rename!(2020.11.04..2020.11.09, `A`B);
m1.setIndexedMatrix!();
m1;

 // output
           A B
           - --
2020.11.04|1 11
2020.11.05|2 12
2020.11.06|3 13
2020.11.07|4 14
2020.11.08|5 15
2020.11.09|6 16

s1;

 // output
           #0
           --
2020.11.01|1
2020.11.02|2
2020.11.03|3
2020.11.04|4
2020.11.05|5
2020.11.06|6

m1 + s1;

 // output
           A B
           - --
2020.11.01|
2020.11.02|
2020.11.03|
2020.11.04|5 15
2020.11.05|7 17
2020.11.06|9 19
2020.11.07|
2020.11.08|
2020.11.09|

(4) 使用 align 函数进行对齐

x1 = [09:00:00, 09:00:01, 09:00:03]
x2 = [09:00:00, 09:00:03, 09:00:03, 09:00:04]
y1 = `a`a`b
y2 = `a`b`b
m1 = matrix(1 2 3, 2 3 4, 3 4 5).rename!(y1,x1)
m2 = matrix(11 12 13, 12 13 14, 13 14 15, 14 15 16).rename!(y2,x2)
a, b = align(m1, m2, 'ej,aj', false);
a;

// output
  09:00:00 09:00:01 09:00:03
  -------- -------- --------
a|1        2        3       
a|2        3        4       
b|3        4        5      

b;

// output
  09:00:00 09:00:01 09:00:03
  -------- -------- --------
a|11       11       13      
b|12       12       14      
b|13       13       15

DolphinDB 提供了 resample 和 asfreq 函数，用于对有时间类型索引的 indexedSeries 或者 indexedMatrix 进行重采样和频度转换。

其实现目的是为用户提供一个对常规时间序列数据重新采样和频率转换的便捷的方法。

下例分别使用 DolphinDB SQL 语句和矩阵来实现计算 Alpha98 因子。全部 DolphinDB 脚本请参考 DolphinDB 实现 98 号因子脚本。

• DolphinDB SQL 语句实现 Alpha98 因子计算的脚本如下：

def alpha98(stock){
	t = select securityid, tradetime, vwap, open, mavg(vol, 5) as adv5, mavg(vol,15) as adv15 from stock context by securityid
	update t set rank_open = rank(open), rank_adv15 = rank(adv15) context by 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 securityid, tradetime, rank(decay7)-rank(decay8) as A98 from t context by tradetime
}

t = loadTable("dfs://k_day_level","k_day")
timer alpha98(t)

• 以下是在 DolphinDB 中通过向量化函数来计算 98 号因子的脚本：

def myrank(x){
	return rowRank(x)\x.columns()
}

def alphaPanel98(vwap, open, vol){
	return myrank(mavg(mcorr(vwap, msum(mavg(vol, 5), 26), 5), 1..7)) - myrank(mavg(mrank(9 - mimin(mcorr(myrank(open), myrank(mavg(vol, 15)), 21), 9), true, 7), 1..8))
}

t = select * from loadTable("dfs://k_day_level","k_day")
timer vwap, open, vol = panel(t.tradetime, t.securityid, [t.vwap, t.open, t.vol])
timer res = alphaPanel98(vwap, open, vol)

通过两个 Alpha98 因子脚本的对比，可以发现用向量化函数来实现 Alpha98 因子的脚本会更加简洁一点。

因为 Alpha98 因子在计算过程中用到了截面数据，也用到了大量时间序列的计算结果。所以在计算某支股票某一天的因子中，既要用到该股票的历史数据，也要用到当天所有股票的信息，对信息量的要求很大。而矩阵形式的面板数据是截面数据和时间序列数据综合起来的一种数据类型，可以支持股票数据按两个维度进行排列，在实现 Alpha98 因子计算中，不需要多次对中间数据或输出数据进行维度转换，简化了计算逻辑。在实现 Alpha98 因子计算的过程中，进行函数嵌套的同时还需要多次进行分组计算来处理数据。对比使用 SQL 语句执行计算，用 panel 函数来处理面板数据，明显计算效率会更高，代码会更简洁。

在性能测试方面，使用单线程计算，SQL 语句计算 Alpha98 因子耗时 610ms。而 panel 函数生成面板数据耗时 70ms，计算 Alpha98 因子耗时 440ms。两者的耗时差异不大，矩阵方式可能略胜一筹。

但是向量化函数处理面板数据也有局限性, 矩阵的面板数据无法进行再次分组，单值模型格式不够直观，而 SQL 支持多列分组，可以联合查询多个字段的信息，适用于海量数据的并行计算。

在处理面板数据时，客户可根据自身对数据的分析需求，来选择不同的方法处理面板数据。

pandas 实现 alpha98 因子的部分脚本如下，完整脚本请参考 python 中实现 98 号因子脚本：

def myrank(x):
    return ((x.rank(axis=1,method='min'))-1)/x.shape[1]

def imin(x):
    return np.where(x==min(x))[0][0]


def rank(x):
    s = pd.Series(x)
    return (s.rank(ascending=True, method="min")[len(s)-1])-1


def alpha98(vwap, open, vol):
    return myrank(vwap.rolling(5).corr(vol.rolling(5).mean().rolling(26).sum()).rolling(7).apply(lambda x: np.sum(np.arange(1, 8)*x)/np.sum(np.arange(1, 8)))) - myrank((9 - myrank(open).rolling(21).corr(myrank(vol.rolling(15).mean())).rolling(9).apply(imin)).rolling(7).apply(rank).rolling(8).apply(lambda x: np.sum(np.arange(1, 9)*x)/np.sum(np.arange(1, 9))))

start_time = time.time()
re=alpha98(vwap, open, vol)
print("--- %s seconds ---" % (time.time() - start_time))

使用 pandas 计算 Alpha98 的耗时为 520s，而使用矩阵实现计算仅耗时 440ms，性能相差千倍。

DolphinDB 内置了许多与时序数据相关的函数，并进行了优化，性能优于其它系统 1~2 个数量级。例如上面例子中使用到的 mavg, mcorr, mrank, mimin, msum 等计算滑动窗口函数。 尤其在计算测试二元滑动窗口（mcorr）中，DolphinDB 的计算耗时 0.6 秒，pandas 耗时 142 秒，性能相差 200 倍以上。为了避免计算结果的偶然性，我们使用了十年的股市收盘价数据，涉及的原始数据量约为 530 万条，对比结果是连续运行十次的耗时。

整体而言，在 Alpha98 因子的计算中，DolphinDB 出现性能上的断层式优势是有迹可循的。