【原】AI量化：如何设计一下可扩展的databuilder?

AI量化实验室 2024-05-13 发布于北京

展开全文

原创文章第535篇，专注“AI量化投资、世界运行的规律、个人成长与财富自由"。

本周开始，专栏第2章节，金融量化数据准备。

第1章已在星球专栏更新完毕：

金融领域是数据最为丰富和规范的行业之一。

数据获取是比较便利的，但要确保数据源的质量、稳定，以及更新数据的及时，可扩展还是需要花费一番功夫。

对于有预算的同学，商业金融数据库会是不错的选择，就需要按相应的文档执行就好，缺点就是相对个人略贵。

好在目前有不少免费开源或者少量付费就可以获取数据的开发包，比如akshare, baostock和tushare。

从数据类型来看，主要分成量价数据（日频为主）、基本面数据（财务报表），另类数据（比如北向资金或者一些行业相关数据）。

2.2.4 金融数据写入mongo库

在python生态里，数据通过pandas的Dataframe来传递非常方便，下面的代码片段直接把一个dataframe写入到一个mongo表中：

def write_df(tb_name, df, db='datalake', drop_tb_if_exist=False):

    db = get_db()

    if drop_tb_if_exist:

        if tb_name in db.list_collection_names():

            db.drop_collection(tb_name)

    try:

        if len(df) == 0:

            print('长度为零')

        else:

            docs = df.to_dict(orient='records')

            get_db()[tb_name].insert_many(docs, ordered=False)

    except pymongo.errors.BulkWriteError as e:

        print(e)

熟悉mysql的同学可以看出，mongo的明显优势：

不需要预定义schema。无论你的数据有多少表，或者原先表里有什么数据，它会自动匹配列名以及列的数据类型。

根据_id字段自动排重，跳过已经存在的行。

在金融量化领域，最典型的场景是我们要持续更新量价日线数据。上一次更新到2024-05-01，你本次从2025-04-01开始更新也没有问题，只要你的_id字段是唯一的，比如使用日期作_id的key即可。

Insert_many是批量写入json数据，这样的效率比一条条写入要高很多。

这一个函数基本就满足金融价量数据入库的需求。

另一个函数update_dict，用于更新基础数据，比如股票，基金的基本信息。

这里没有使用insert_one而是使用update_one的好处在于，我们希望不存在就创建，存在就更新的统一操作，不需要判断原本数据存在与否。

def update_dict(tb_name, dict_data, key='symbol', upsert=True):

    db = get_db()

    db[tb_name].update_one({key: dict_data[key]}, {'$set': dict_data}, upsert=upsert)

2.3 databuilder：可扩展的数据更新器

量化本身就是一种“数据分析”。

量化数据在持续增加，日频数据我们需要盘后更新，基本面数据是季度更新。还有一些量化指标，我们需要在基础数据更新之后进行“预计算”。

由于波及的数据类型、数据表很多，很容易搞混淆，因为需要一个良好的设计模式。

2.3.1 DataService:数据获取服务

DataService负责从数据源获取数据，我们抽象出基类：

get_basic_items，返回列表。

Get_quotes_df，传入标的代码symbol以及起始日期，默认值为20050101。

class DataService:

    def get_basic_items(self):

        return []



    def get_quotes_df(self, symbol, start_date='20050101'):

        pass

2.3.1.1 BondDataService

使用ak下载可转债更表的BondDataService

class BondDataService(DataService):

    def get_basic_items(self):

        return api_ak.get_basic_cov_bonds()



    def get_quotes_df(self, symbol, start_date='20050101'):

        pass

2.3.2 基础信息构建与查询：Basic

class Basic:

    def __init__(self, tb_name, service: DataService):

        self.tb_name = tb_name

        self.service = service



    def build(self):

        items = self.service.get_basic_items()

        for item in tqdm(items):

            # 更新 mongo 对应表的某一行

            mongo_utils.update_dict(self.tb_name, item, key='symbol', upsert=True)



    def query(self, filter):

        pass

Basic接受两个参数：tb_name是表名，service是数据数据。

Build函数把service返回的列表，更新到tb_name表中。

Query函数支持对tb_name进行查询。

2.3.2.1 BondBasic

class BondBasic(Basic):

    def __init__(self, service: DataService, tb_name='basic_bond'):

        super(BondBasic, self).__init__(tb_name, service)