Qust

支持流式计算的查询引擎，底层基于rust, 应用层用的python

• 流式计算，算子有状态保留，支持流式计算
• 性能高，大多数情况下速度比polars高，内存消耗更少
• 算子丰富，内置丰富的金融算子，比如k线合成、回测、组合优化等等
• 可拓展性强，底层基于rust的datafusion, 拓展到分布式很方便.

git地址

demo地址

安装

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple qust

目的

量化框架的不可能三角：

1. 高性能
2. 易用
3. 实盘回测一致

比如一些事件驱动的框架，优点是实盘回测一致，缺点是速度很慢， 而且不易用，毕竟操作 DataFrame 更加直观

还有一些向量化回测的框架, 优点是回测性能好，但是实盘回测不一致，而且从回测阶段转到实盘阶段比较麻烦

另外就是实盘和回测两套代码这种框架，这个不易用

总结下来就是，要想易用，就得用 DataFrame api 去做策略，要想实盘回测一致，就得用事件驱动

有没有方法能同时兼顾两者？有，用流式计算

底层用rust写就能实现高性能，api 封装成python的 DataFrame api 就能实现易用性，流式计算本身就是事件驱动，实盘回测就一致。qust的目的就是实现这个

使用

import qust as qs
from qust import (col, pms)
import polars as pl
import numpy as np

n = 10
data = pl.DataFrame({
    "factor": np.random.randn(n),
    "code": np.random.choice(["a", "b", "c"], size=n, replace=True),
})
data_next = pl.DataFrame({
    "factor": np.random.randn(n),
    "code": np.random.choice(["a", "b", "c"], size=n, replace=True),
})

df = qs.with_cols(
    col("factor").mean().expanding().alias("cum_mean"),
    col("factor").mean().rolling(3).alias("rolling_mean"),
    col("factor").mean().expanding().over("code").alias("cum_mean_over")
)

print(df.calc_data(data))

shape: (10, 5) ┌───────────┬──────┬───────────┬──────────────┬───────────────┐ │ factor ┆ code ┆ cum_mean ┆ rolling_mean ┆ cum_mean_over │ │ --- ┆ --- ┆ --- ┆ --- ┆ --- │ │ f64 ┆ str ┆ f64 ┆ f64 ┆ f64 │ ╞═══════════╪══════╪═══════════╪══════════════╪═══════════════╡ │ -1.664883 ┆ b ┆ -1.664883 ┆ null ┆ -1.664883 │ │ -0.187855 ┆ a ┆ -0.926369 ┆ null ┆ -0.187855 │ │ -1.11352 ┆ b ┆ -0.988753 ┆ -0.988753 ┆ -1.389201 │ │ -1.212976 ┆ a ┆ -1.044808 ┆ -0.838117 ┆ -0.700416 │ │ 1.305776 ┆ b ┆ -0.574692 ┆ -0.34024 ┆ -0.490876 │ │ -0.418283 ┆ a ┆ -0.548624 ┆ -0.108494 ┆ -0.606371 │ │ -0.521383 ┆ b ┆ -0.544732 ┆ 0.122036 ┆ -0.498503 │ │ -2.068707 ┆ c ┆ -0.735229 ┆ -1.002791 ┆ -2.068707 │ │ -0.46641 ┆ a ┆ -0.70536 ┆ -1.018833 ┆ -0.571381 │ │ 1.223008 ┆ b ┆ -0.512523 ┆ -0.437369 ┆ -0.1542 │ └───────────┴──────┴───────────┴──────────────┴───────────────┘

print(df.calc_data(data_next)) # df 里面的算子都状态保留

shape: (10, 5) ┌───────────┬──────┬───────────┬──────────────┬───────────────┐ │ factor ┆ code ┆ cum_mean ┆ rolling_mean ┆ cum_mean_over │ │ --- ┆ --- ┆ --- ┆ --- ┆ --- │ │ f64 ┆ str ┆ f64 ┆ f64 ┆ f64 │ ╞═══════════╪══════╪═══════════╪══════════════╪═══════════════╡ │ 1.040219 ┆ c ┆ -0.371365 ┆ 0.598939 ┆ -0.514244 │ │ -2.149415 ┆ c ┆ -0.519536 ┆ 0.037937 ┆ -1.059301 │ │ -1.031419 ┆ b ┆ -0.558911 ┆ -0.713538 ┆ -0.300403 │ │ 2.890776 ┆ c ┆ -0.312505 ┆ -0.096686 ┆ -0.071782 │ │ -0.710796 ┆ c ┆ -0.339058 ┆ 0.382854 ┆ -0.199585 │ │ -1.043864 ┆ a ┆ -0.383108 ┆ 0.378705 ┆ -0.665878 │ │ -0.784278 ┆ c ┆ -0.406706 ┆ -0.846313 ┆ -0.297034 │ │ -0.492146 ┆ c ┆ -0.411453 ┆ -0.773429 ┆ -0.324907 │ │ 1.290463 ┆ a ┆ -0.321879 ┆ 0.00468 ┆ -0.339821 │ │ -0.01888 ┆ c ┆ -0.306729 ┆ 0.259812 ┆ -0.286653 │ └───────────┴──────┴───────────┴──────────────┴───────────────┘

与polars语法比较

data = pl.DataFrame({
    "price": range(5),
    "code": ["a", "a", "a", "b", "b"]
})
df = qs.with_cols(
    col("price").sum().expanding().alias("cum_sum_otters"),
    pl.col("price").cum_sum().alias("cum_sum_polars"),
    col("price").sum().expanding().over("code").alias("cum_sum_otters_over"),
    pl.col("price").cum_sum().over("code").alias("cum_sum_polars_over")
)
df.calc_data(data)

shape: (5, 6)

# 广播
col("a") + col("b", "c")
col("a") > col("b", "c")
col("a", "b") + col("c", "d")
col("a", "b") + col("c")
col("a", "b") & col("c")

与polars性能比较

import time
n = 2000000
data = pl.DataFrame({
    "factor": np.random.randn(n),
    "code": np.random.choice(["a", "b"], size=n, replace=True),
})

1. qust单线程 vs polars多线程

s = time.time()
_ = qs.select(
    col("factor").rank().rolling(10).over("code")
).calc_data(data)
print(f"qust: {(time.time() - s) * 1000.0}.ms")

s = time.time()
_ = data.select(
    pl.col("factor").rolling_rank(10).over("code")
)
print(f"polars: {(time.time() - s) * 1000.0}.ms")

qust: 104.89392280578613.ms polars: 194.35358047485352.ms

2. qust多线程 vs polars多线程

s = time.time()
_ = qs.select(
    col(*[col("factor").mean().alias(f"mean_{i}") for i in range(50)]).rolling(10).over("code")
).calc_data(data)
print(f"qust: {(time.time() - s) * 1000.0}.ms")

s = time.time()
_ = data.select(
    [pl.col("factor").rolling_mean(10).over("code").alias(f"mean_{i}") for i in range(50)]
)
print(f"polars: {(time.time() - s) * 1000.0}.ms")

qust: 137.73751258850098.ms polars: 382.92813301086426.ms

3. qust自定义算子 vs polars自定义算子

class MeanUdf(qs.UdfRow):

    def __init__(self):
        self.sum = 0.0
        self.count = 0.0

    def output_schema(self, input_schema):
        return [("mean_res", pl.Float64)]

    def update(self, value):
        self.sum += value
        self.count += 1.0

    def calc(self):
        return [self.sum / self.count]

    def retract(self, value):
        self.sum -= value
        self.count -= 1.0

s = time.time()
_ = qs.select(
    col("factor").udf.row(MeanUdf()).rolling(10).over("code")
).calc_data(data)
print(f"qust: {(time.time() - s)}.s")

s = time.time()
_ = data.select(
    pl.col("factor").rolling_map(lambda x: x.mean(), 10).over("code")
)
print(f"polars: {(time.time() - s)}.s")

qust: 1.3382771015167236.s polars: 51.07790398597717.s

和polars相互使用

data = pl.DataFrame({
    "value": [1, 2, 3, 4, 5]
})
data_next = pl.DataFrame({
    "value": [3, 1, 10]
})

1. 在qust里面使用polars

qs.with_cols(
    (pl.col("value") + 1).alias("value+1"),
    (col("value").pl + 2).alias("value+2"),
    col("value").mean().expanding().select(pl.col("value") - 1).alias("value-1")

).calc_data(data)

shape: (5, 4)

2. 在polars里面使用qust

data.select(
    col("value").mean().rolling(3).alias("value_mean1").pl,
    col("value").pl.rolling_mean(3).alias("value_mean2"),
)

shape: (5, 2)

# 上面的写法没有状态保留, 如果需要状态保留，需要把算子的状态保存到全局变量，使用 `expr.cache(id)`
e = col(
    col("value").mean().alias("mean"),
    col("value").sum().alias("sum"),
).rolling(3)
e_pl = e.cache("unique_id").pl
# 注意这里不能接polars的over，e.cache("unique_id").pl.over("code"), 这种写法会直接报错，
# 可以写成 e.over("code").cache("unique_id").pl, 或者用 data.qs.df
data.select(e_pl)

shape: (5, 1)

data_next.select(e_pl)

shape: (3, 1)

保存到全局的算子状态一直在内存里面，需要清除用:

qs.clear_cache("unique_id") # 单个清除
qs.clear_cache() # 全部清除

由于polars的限制，上面的算子无法多列返回, 所以如果有多列返回，返回的是多列组成的struct

如果需要多列返回，只能这样写:

data.qs.select(e)

shape: (5, 2)

df = qs.select(e)
data.qs.df(df)

shape: (5, 2)

data_next.qs.df(df)

shape: (3, 2)

为什么有polars，还要写qust？

1. 流式计算

写量化策略的时候，一般有下面两种方法

1. 向量化计算
2. 事件驱动

如果策略用向量化计算，在实盘的时候就很慢，因为要重复计算历史数据, 而且很多策略没法向量化

如果策略用的事件驱动，回测的时候就很慢，而且事件驱动写法特别麻烦

流计算就是把算子都写成事件驱动的形式。比如计算移动平均，在算子里面存储两个状态 (sum, count), 每有一个行新数据value过来，更新算子的内部状态:

sum = sum + value

count = count + 1

在需要计算结果的时候就用 sum / count

data = pl.DataFrame({
    "value": [1, 2, 3, 4, 5]
})
data_next = pl.DataFrame({
    "value": [6, 7, 8]
})

df = qs.with_cols(
    col("value").mean().rolling(3).alias("rolling_mean"),
    col("value").std().expanding().alias("cum_std"),
)

print(df.calc_data(data))
shape: (5, 3)
┌───────┬──────────────┬──────────┐
│ value ┆ rolling_mean ┆ cum_std  │
│ ---   ┆ ---          ┆ ---      │
│ i64   ┆ f64          ┆ f64      │
╞═══════╪══════════════╪══════════╡
│ 1     ┆ null         ┆ null     │
│ 2     ┆ null         ┆ 0.707107 │
│ 3     ┆ 2.0          ┆ 1.0      │
│ 4     ┆ 3.0          ┆ 1.290994 │
│ 5     ┆ 4.0          ┆ 1.581139 │
└───────┴──────────────┴──────────┘
print(df.calc_data(data_next))
shape: (3, 3)
┌───────┬──────────────┬──────────┐
│ value ┆ rolling_mean ┆ cum_std  │
│ ---   ┆ ---          ┆ ---      │
│ i64   ┆ f64          ┆ f64      │
╞═══════╪══════════════╪══════════╡
│ 6     ┆ 5.0          ┆ 1.870829 │
│ 7     ┆ 6.0          ┆ 2.160247 │
│ 8     ┆ 7.0          ┆ 2.44949  │
└───────┴──────────────┴──────────┘

在第一个调用df.calc_data(data)的时候，df内部的算子都有状态保留，所以在第二个调用df.calc_data(data_next)时候，没有重新计算

实际情况是，绝大多数算子都有对应的事件驱动形式，少量的算子比如pl.col("a").rank(), 看起来不是事件驱动的形式（当前行的值受到未来行的值的影响），但是其实也可以变换成事件驱动形式，

• 转换成行算子，比如 a 列有a1，a2，a3三个元素，就是col(a1, a2, a3).rank(axis=1)
• 事件驱动形式的批算子，每次计算的时候保证传入的数据完整，比如计算pl.col("a").rank().over("date"), 保证每次计算传入的数据包含整天的所有数据

polars不是也支持streaming吗？我看了polars的底层，觉得polars的streaming不是真正意义上的流式计算，只是为了避免out of memory，而且局限性大(比如over是用的 切割 -> 计算 -> 拼接)。如果polars要实现真正的流式计算，我估计底层得推倒重来改成datafusion的那种框架

2. 表达式解耦

polars的Expr用的enum, 这样就导致每实现一个算子，底层很多代码都要改, 这样就不难理解为什么一个简单的pl.col("a").rolling_rank(10)算子直到最近才实现，而且速度比我一个简单的实现慢一倍。

datafusion聚合算子用的Box<dyn trait>, 然后根据上下文选择不同路径的ExecutionPlan, 这样添加算子很方便，而且优化路径也很清晰，性能还不受影响。

polars这种写法还有个缺点，就是导致同样的逻辑写法割裂，比如求和逻辑有下面写法:

• pl.col("a").sum()
• pl.col("a").cum_sum()
• pl.col("a").rolling_sum(10)
• df.group_by("b").agg([pl.col("a").sum()])

如果说 sum() 和 rolling_sum(10), 都是求和逻辑, 前一个是针对整列，后一个是针对滚动，但是 rank()和rolling_rank(10), 又是两个不想关的算子, 而且并不存在cum_rank()这个算子，这样逻辑就很割裂，为什么能存在cum_sum, 但是不能存在cum_rank, cum_skew, cum_cov?

相反用datafusion的上下文逻辑，写法就比较一致:

• col("a").sum()
• col("a").sum().expanding()
• `col(