稳定的股票数据源

本项目为个人项目，为个人量化投资分析提供稳定、优雅、便于运维和使用的股票数据源
项目链接：https://github.com/QiYueran-source/market_data.git

1.项目介绍

在投资分析中，稳定的数据来源至关重要！
在制定策略和回测时，只有数据稳定，字段明确，才能保证策略的稳定性和可靠性。 然而，市场上的常见数据源，如TuShare、AkShare等，虽然丰富，但是均有重要的问题， 那就是：

贵

如果我想用TuShare一个源解决所有问题，每年可能需要支付数千元乃至更多的费用， 这对只想研究，还未实盘执行任何策略的投资者来说，可能是一个沉重的负担。

因此，我想到一个方法： 每个数据源，都会提供一些“薅羊毛”的次数，比如TuShare， 其股票日度数源每天可以免费查询6000次，足够覆盖所有日度数据需求， 但是，该免费源仅支持股票日线数据，对于分时、财务、ETF数据，需要额外付费，并且价格不菲。
麦蕊智数则提供实时数据源，一年花费50元左右，可获得每天10000次查询机会， 且包含了实时，ETF，财务等数据，但是该源的字段命名方式较为混乱， 且覆盖的数据类型不多（如宏观、情绪等，均未提供），未来扩展性较差。

为了“花小钱，办大事”，我使用了多个数据源，每个数据源用来获取不同类型的数据， 这样，可以薅到最多的羊毛。

然而，多数据源对接，是一个复杂问题，不同数据源的接口、字段、限流、异常均有不同， 如何设计一个稳定、优雅、便于运维和使用的股票数据源，是一个值得思考的问题。

2.项目设计

2.1 单一事实设计

由于不同源的字段不同，为了保证数据的一致性， 可以事先约定好字段的名称和含义，如open: REAL 开盘价（元）。 不同数据源的开盘价字段可能不同，如open, o, open_price, kpj等， 需要将不同数据源的开盘价字段映射到统一的名称和含义，并储存到数据库中， 最终需要获取数据，只通过数据库查询，而不需要关心具体的数据源。

为了保证数据的一致性，设计静态类TableSchema，用于存储字段的名称和含义，

from models.table_schema import TableSchema, col
import numpy as np

TableSchema(
    database_name='trade_calendar.db',
    table_name='akshare_trade_calendar',
    schema=[
        col('calendar_date', np.str_, 'TEXT', True),
        col('is_open', np.int8, 'INTEGER', False),
    ],
)

该类是数据库中表在python中的映射，保存了库名、表名、字段名、字段在py中类型、字段在db中类型、字段是主键等信息。

对于每一张表，预先定义好其TableSchema； 创建库时，读取所有的Schema，按照其schema属性生成DDL语句，创建表； 在获取数据源数据后，需要按照对应的TableSchema，将字段的名称和类型统一， 提供一个validate(df, schema)函数，用于校验df是否符合Schema预期。

当然，一个表可以对应多个数据源；一个数据源可以为多张表提供数据。 因此，校验数据时，df并不需要有schema的所有字段， df仅需要有schema中定义的主键，且不能有schema中未定义的字段即可， 在更新时，使用upsert，按主键将df更新到数据库中。

2.2 读写分离设计

读写分离设计，主要是减少直接定义DML语句，提高扩展性和可维护性。

为了实现读写分离，设置一个Buffer类，用于缓存数据， 当缓存收集到一定数量后，更新到数据库中。

Buffer(
    buffer_size = 1000,
    schema = XXXTableSchema,
)

一个Buffer类与一个表绑定，并且提供append(df)和flush()函数， 使用append函数，将df添加到buffer，buffer_size增加len(df)条， 在达到buffer_size时，调用flush函数，将buffer中的数据更新到数据库中。

为了保证数据一致性，append后，仍然调用validate(df, schema)函数，校验df是否符合Schema预期。 如果校验通过，则将df添加到buffer，否则，不添加到buffer，并返回错误信息。

2.3 Provider容错设计

Provider层负责对接数据源，由于数据源的异常情况各有不同， 需要进行异常管理。

具体来说，每一个数据源，对应一个provide文件， 提供fetch()函数，用于对接数据源，获取数据。
在fetch()函数中，捕获所有的异常，并且抛出；
然后，定义fetch_and_clean()函数，用于捕获所有fetch()中捕获的异常， 并且提供兜底。 这样，将所有异常在fetch_and_clean()函数中处理。

具体来说，fetch_and_clean()函数中，需要进行以下操作： - 1.捕获异常
- 2.使用logger.exception记录异常
- 3.返回兜底数据

定义一个字典Counter()，用来记录兜底数据的使用次数， 返回给上层，用于邮件汇报。

这样，可以避免将API错误暴露，而是在日志中记录，便于后续排查。

2.4 限流器设计

限流器是数据源的重要限制，每一个数据源都有限制规则， 如麦蕊智数的限流为每分钟300次。

可以设定限流器类，为每一个数据源设置限流：

limiter = APILimiter(
    interval = 60, # 间隔
    max_requests = 300, # 最大请求数
)   

内部，使用滚动窗口的方式： 每一次请求后，都将请求加入进行队列，并且请求完成后，统一停止interval秒， 在请求队列满后，其他请求被阻塞，直到队列中的请求完成。

根据不同的数据源，初始化不同的限流器常量，并且提供一个装饰器，用于装饰fetch函数：

@limiter('mairui')
def fetch():
    pass 

该装饰器内部，每次调用func后，都调用limiter.acquire()， 如果返回True，则继续执行func，否则，阻塞等待，直到队列中的请求完成。

装饰器可以传入数据源名称参数，用于选择不同的限流器。

2.5 Job层

Job层负责调度Provider和Buffer层，实现数据更新， 同时提供一个JobInfo信息，用于记录本次Job的执行情况，交给执行层发送邮件。

Job层还负责异常最终处理，在Job层，对于已经定义的异常，如数据库写入、缓存异常等问题， 已经有了定义好的异常，则使用logger.exception记录异常； 对于为定义异常，则使用try ... except Excpetion as e的方式捕获，并传给JobInfo，用于邮件汇报。

2.6 执行层

执行层是程序的入口，负责启动Job层。 根据业务需求不同，设定了若干个执行脚本，使用crontab定时执行。 在执行完后，根据收集到的JobInfo发送信息到邮箱。

最终的邮件效果如下：