Dagster搭建AI量化系统底座

Quantlab近期的计划：

1、重构dagster：含etf数据更新定时任务（symbol列表来自django），回测任务来自mongo（需要设计mongo存储json，比当前的toml还要简单，再抽象一层，当前的Toml还是通用结构，精细化为四大模板：大类资产配置+定期再平衡；轮动策略，择时策略，多策略组合）。

2、整合回测系统到dagster，支持定时跑任务，更新结果（结果指标，订单明细，交易明细， equity曲线等，可以一个任务存一个hdf5到mongo）

3、策略集回测结果写回mongo

4、通过GraphQL向dagster集群远程发布命令，执行回测。

我们要解决同学们环境配置复杂，使用策略不易的问题，最好就是尽量SAAS化，能不安装就不安装，能提供在线服务就提供在线服务。比如数据更新，比如策略自动运行。否则后续的tensorflow，pytorch，强化学习，对于新手都太难了。给代码也用不起来。

以Dagster为主，还是Django为主，我还是思考了一下，决定还是Dagster为主。

Django擅长做管理系统，当然它也可以做异步任务与定时任务（通过Celery，但没那么直观和方便），Django最强项是它的管理后台以及ORM。但金融上，更多依赖mongo等NoSQL数据库。

01 Dagster搭建AI量化系统底座

Dagster的文档如下：https://docs.dagster.io/getting-started/install

直接在venv使用pip进行安装：

pip install dagster dagster-webserver

新版本Dagster是1.5.5，多了一个dagster-webserver。

我们使用官方案例——一个ETL的pipline来初始化我们的QuantTask子系统。

dagster project from-example –name quanttask –example quickstart_etl

然后cd进入quanttask目录，使用命令：dagster dev即可以打开webui。

import pandas as pd
from dagster import get_dagster_logger, asset

from quanttask.utils import loader_ak, mongo_utils
from quanttask.utils.tools import _get_start_date


@asset(description='更新场内基金日线数据', group_name='etf')
def update_etf_quotes():
    symbol = '518880.SH'
    table_name = 'etf_quotes'
    df = loader_ak.load_etf_hist_quotes(symbol, _get_start_date(table_name, symbol))
    mongo_utils.write_df(table_name, df)