基于dagster定期更新A股日线后复权数据

原创文章第291篇，专注“个人成长与财富自由、世界运作的逻辑与投资"。

星球后台有星友留言说：“为何要花时间去搞这么多基础建设”，使用现有的平台，做策略研究不就好了。

我原本也是这么想的。

无奈找不了好的工具平台，只能自己来做。而且很多机器模型原本就是黑盒，如果我们再叠加一项平台的黑盒，那如何感大规模应用。

我们做AI量化的逻辑AI量化投资思考与下一步计划，与传统投资软件完全不同。传统就是K线，指标，形态。压力线，支撑线，这里有一些统计的逻辑，但还没有完全量化。我们是从大数据挖掘的角度出发，面对这4000多支股票，全天候、全方位去量化它们，对它们进行排序，与市场环境进行匹配，进而选出当下最大概率适合建仓的股票。

数据是AI量化的基础，从性价比而言，目前还是选择tushare和akshare。

在数据更新上，使用dagster编排系统：

可以直接“物化”：

代码如下：

import pandas as pd
from dagster import asset, get_dagster_logger, materialize

from utils import mongo_utils
from tasks import ts_loader


def format_df(df: pd.DataFrame):
    df = df.rename(columns={'ts_code': 'symbol'})
    df['_id'] = df['symbol']
    return df


@asset(description='A股股票列表', group_name='basic')
def stock_list():
    df = ts_loader.get_stock_list()
    df = format_df(df)
    get_dagster_logger().info(df)
    mongo_utils.write_df('basic_stock', df, drop_tb_if_exist=True)
    return df


if __name__ == '__main__':
    materialize([stock_list])

在mongo里呈现结果：

茅台数据的增量更新：

import akshare as ak
from datetime import datetime
from quant_project.utils import mongo_utils
import pymongo

import pandas as pd


def _get_start_date(tb_name, symbol):
    start_date = ''
    items = mongo_utils.get_db()[tb_name].find({'symbol': symbol}).sort('date', pymongo.DESCENDING).limit(
        1)
    items = list(items)
    if len(items) == 0:
        print('不存在{}记录'.format(symbol))
        start_date = ''
    else:
        start_date = items[0]['date']
        print('{}从{}更新'.format(symbol, start_date))
    return start_date


def load_symbol_quotes_from_start_date(symbol, start_date='20050101'):
    code = symbol[:6]
    end_date = datetime.now().strftime('%Y%m%d')
    df = ak.stock_zh_a_hist(symbol=code, period="daily", start_date=start_date, end_date=end_date,
                            adjust="hfq")

    return df


def _format_df(df: pd.DataFrame):
    df.rename(columns={'日期': 'date', '开盘': 'open', '收盘': 'close', '最高': 'high', '最低': 'low', '成交量': 'volume',
                       '换手率': 'turn_over'}, inplace=True)
    df = df[['date', 'open', 'high', 'low', 'close', 'volume', 'turn_over']]
    df['date'] = df['date'].apply(lambda x: x.strftime('%Y%m%d'))
    return df


def load_symbol_quotes(symbol):
    tb_name = 'stock_quotes'
    start_date = _get_start_date(tb_name, symbol)
    df = load_symbol_quotes_from_start_date(symbol, start_date)
    df = _format_df(df)
    df['symbol'] = symbol
    df['_id'] = df['symbol'] + '_' + df['date']
    mongo_utils.write_df(tb_name, df)


if __name__ == '__main__':
    symbol = '600519.SH'
    load_symbol_quotes('600519.SH')

代码已经上传至的星球中。知识星球与开源项目：万物之中，希望至美

mongo中茅台的历史——后复权的数据，支持增量更新。

这样就有两个任务：

年化29.3%，大类资产ETF动量轮动，添加商品与货币ETF，实盘跟进中（代码+数据下载）

年化25.5%，使用pycaret机器学习自主调参进行大类资产轮动