如果您遇到了以下场景：

我的框架有了，可是我如何将我自己的数据源接入这个框架呢？

我发现我的框架可能需要多个数据源，一部分是我自己买的，可是还有一部分需要从另一个供应商获取，我该怎么办？

Datasource模块就是用来解决您上述的问题的。他的主要作用是获取外部第三方数据的模块，可以同时支持多个数据源。模块主要使用的是配置驱动+基类复写的方式完成扩展的。

用户自定义位置：userspace/extensions/data_source/

在看到更多用例之前，我们来明确一些基本概念和术语，这样能更好地帮助您使用这个模块：

Provider：数据供应商，比如新浪财经，东方财富，Akshare，Tushare系统默认自带了一些例子，其他的供应商或API需要用户自己定义

Schema：数据格式规范。我们从数据源（可以是多个）获得数据后，需要把多个API的数据整合成我们最后需要存储或使用的一个标准格式，这个格式就是由Schema定义的。这里我们说的schema都定义在我们数据库表的定义里，比如core/tables/macro/cpi/schema.py

Handler: 处理与第三方数据源对接，获取数据，重组数据，最后入库的一个处理器。我们作为用户主要定义的就是这个handler以及其对应的配置config

Mapping：位置：userspace/data_source/mapping.py 这里是定义我们的每个数据源是通过哪个handler从哪个provider里获取数据的关系表

注意：

一个data source只能对应一个schema（数据表），多个获取数据的API，其中这些API可以来自不同的Provider。

一个data source最终只能由一个handler来完成，但这个data source可以有多个handler。（可以把handler看成获取数据的手段，手段可以是多个，但最终落地执行的方案只能是一个）

一个完整的案例：接入龙虎榜数据

假设你想在策略里使用龙虎榜数据，NTQ 内置的数据源不提供这个，需要自己接入。

第一步：建目录

在 userspace 下创建 data source 的目录结构：

userspace/extensions/data_source/
├── mapping.py                    # 注册数据源
├── providers/                     # API 提供者
│   └── tushare/
│       └── provider.py
└── handlers/                      # 数据处理器
    └── dragon_typer/
        ├── config.py              # 数据源配置
        └── handler.py             # 处理逻辑

第二步：写 mapping.py

mapping.py 是入口，NTQ 启动时自动发现并加载它。在里面注册你的数据源：

DATA_SOURCES = {
    "dragon_typer": {
        "handler": "DragonTyperHandler",
        "is_enabled": True,
    },
}

• key（dragon_typer）：数据源的唯一标识，后续 CLI 命令用这个名字指定
• handler：Handler 类名。可以写简写（只写类名），系统会自动补全为 userspace.extensions.data_source.handlers.dragon_typer.DragonTyperHandler；也可以写完整路径
• is_enabled：是否启用。禁用的数据源不会被 renew 批量更新

第三步：写 Provider

Provider 封装外部 API 的调用逻辑。一个 Provider 可以服务多个数据源（比如 tushare 同时提供行情和龙虎榜）。

# providers/tushare/provider.py
from core.modules.data_source.core.base_class.base_provider import BaseProvider

class TushareProvider(BaseProvider):
    provider_name = "tushare"
    requires_auth = True
    auth_type = "token"

    # 限流配置：每个 API 方法每分钟最多调用多少次
    api_limits = {
        "top_list": 100,     # 龙虎榜接口，每分钟 100 次
        "daily": 500,        # 日线行情，每分钟 500 次
    }
    default_rate_limit = 60  # 没有列出的 API 默认每分钟 60 次

关键变化：限流不再在 config 里写 max_per_minute，而是直接在 Provider 类的 api_limits 字典里声明。这样多个数据源共用同一个 Provider 时，限流逻辑只维护一份。

第四步：写 config.py

config 定义数据源的拉取和存储规则：

# handlers/dragon_typer/config.py
CONFIG = {
    # 绑定哪张数据库表
    "table": "dragon_typer",

    # 存储模式：upsert = 有则更新，无则插入
    "save_mode": "upsert",
    "save_batch_size": 1000,

    # renew 配置：增量更新
    "renew": {
        "type": "incremental",          # 增量模式
        "last_update_info": {
            "date_field": "trade_date",  # 用哪个字段判断最后更新日期
            "date_format": "%Y%m%d",
        },
    },

    # API 调用配置
    "apis": {
        "fetch_daily": {
            "api_name": "top_list",
            "provider_name": "tushare",
            "method": "top_list",
            "params_mapping": {
                "trade_date": "{date}",
            },
            "result_mapping": {
                "code": "ts_code",
                "name": "name",
                "amount": "amount",
                "trade_date": "trade_date",
            },
            "params": {},
        },
    },

    # 忽略 API 返回但不需要存储的字段
    "ignore_fields": [],

    # 默认日期范围（年）
    "default_date_range_years": 3,
}

关键字段说明：

第五步：写 handler.py

Handler 控制拉取流程——按日期循环调用 API、合并结果、写入数据库：

# handlers/dragon_typer/handler.py
from core.modules.data_source.core.base_class.base_handler import BaseHandler

class DragonTyperHandler(BaseHandler):
    """龙虎榜数据处理器"""

    def fetch_data(self, date_range):
        """按日期范围拉取数据"""
        for date in date_range:
            result = self.call_api("fetch_daily", date=date)
            if result:
                self.save_batch(result)

BaseHandler 的构造函数已经帮你注入了 schema、config、providers，你只需要关注业务逻辑。

第六步：执行更新

# 更新龙虎榜数据
python cli.py renew dragon_typer

# 更新所有已启用的数据源
python cli.py renew

# 列出所有数据源
python cli.py renew list

# 强制更新（忽略增量判断，全量重刷）
python cli.py renew dragon_typer -f

其他可用配置

renew 模式对比

rolling 配置示例：

"renew": {
    "type": "rolling",
    "rolling": {
        "unit": "day",
        "length": 90,
    },
},

result_mapping

result_mapping 把 API 返回的字段名映射成数据库表的字段名。比如 tushare 返回 ts_code，你的表字段叫 code：

"result_mapping": {
    "code": "ts_code",       # 表字段: API 字段
    "name": "name",
    "amount": "amount",
    "trade_date": "trade_date",
}

params_mapping

params_mapping 定义调用 API 时参数怎么填充。{date} 会被替换成当前迭代的日期：

"params_mapping": {
    "trade_date": "{date}",
    "limit": "100",          # 固定值
}

多数据源共用一个 Provider

在 mapping.py 里注册多个数据源，它们都指向同一个 tushare Provider：

DATA_SOURCES = {
    "daily_price": {
        "handler": "DailyPriceHandler",
    },
    "dragon_typer": {
        "handler": "DragonTyperHandler",
    },
}

两个 handler 的 config 里都写 "provider_name": "tushare"，限流由 Provider 的 api_limits 统一管理，不会因为两个数据源各自计数导致超限。

常见问题

数据源没被发现

检查 mapping.py 里是否定义了 DATA_SOURCES 字典，handler 路径是否正确。is_enabled 是否为 True。

API 调用被限流

限流由 Provider 的 api_limits 控制。如果调用频率太高，调高对应 API 的限流值，或者在 config 的 params 里减小每次请求的数据量。

增量更新没生效

检查 renew.last_update_info 的 date_field 是否和数据库表里的字段名一致。系统用这个字段查询最后一条数据的日期，从那天开始增量拉取。

想切换 renew 模式

直接改 config 里的 renew.type，下次 renew 命令就用新模式。不需要改代码或重新建表。从 incremental 切到 refresh 会全量重刷；切到 rolling 会清理超期旧数据。