dagster用于AI量化投资的数据流编排

无论是量化投资还是算法模型，当然这二者会统一到“AI量化”这个大命题下边——都需要一个pipeline的编排调度。

无论是量化还是算法，我们花了大量的时间，在数据整理、准备下。策略或者模型反倒花的时间很少。这本身是不对的，若是这个过程不标准，很多工作需要重复去做，浪费时间，容易出错，还没有积累就不应该了。

所以，反复调研了airflow, prefect，dagster。

目前看dagster还是比较契合。尽管这次引入asset的概念搞得有点云里雾里，但看它的案例即有量化，也为AI，也我们的场景契合。

python更多以函数为计算单元。

下面这个函数是官方的一个tutorial，这个“资产”其实就是从远程拉一个“谷物”信息的csv，解析成python的list[dict]。

import csv
import requests
from dagster import asset


@asset
def cereals():
    response = requests.get("https://docs.dagster.io/assets/cereal.csv")
    lines = response.text.split("\n")
    cereal_rows = [row for row in csv.DictReader(lines)]

    return cereal_rows

if __name__ == '__main__':
    rows = cereals()
    print(rows)

原始表格的信息如上。

这个函数如何直接使用，就是一个函数，本身可以直接运行，所以debug是很方便的。——这是我们要求的低侵入性，就是业务代码不需要有太多的改动，可以在框架里呈现。

而当我们使用 dagster dev -f cereal.py时，可以启动UI:

只要文件里定义了asset，就可以“物化“，其实就是运行这个”函数“。

我们看下”物化“具体做了什么，就是把这个数组，保存在本地storage/cereal文件里。

使用dagit -f也可以运行，但没有启动deamon等服务，dagster dev方便开发调试。

资产可以形成图，上下游依赖。有时候获得相当的资源会失败，那只需要重建这一部分即可。这里的思想挺好的。

从工作流到数据流。

如上都是基于内存计算。

我们看一下远程下载zip文件

@asset
def cereal_ratings_zip() -> None:
    urllib.request.urlretrieve(
        "https://dagster-git-tutorial-nothing-elementl.vercel.app/assets/cereal-ratings.csv.zip",
        "cereal-ratings.csv.zip",
    )

这里的一个逻辑上，从远程下载csv，并对数据进行筛选和排序后，与远程下载zip文件解压后的结果合并计算。