使用场景

NTQ 内置了 20 多种常用数据（K线、财务、指标、宏观等），覆盖了大多数策略场景。但有时候你需要接入自己的数据——比如自己爬的龙虎榜、行业景气度数据、或者第三方另类数据。

这时候就需要用到 data contract（数据契约）。它是 NTQ 里"一种数据"的标准声明方式：告诉系统数据叫什么、长什么样、从哪里加载。策略、扫描、回测引擎都通过统一的接口读取数据，不用关心数据从哪来。

自定义数据放在 userspace/extensions/data_contract/ 目录下，重启 NTQ 后自动发现并注册。

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

假如您手里有一份龙虎榜的 CSV 数据，按日期存了每天上榜的股票和席位信息。您想在策略里用"龙虎榜净买入额"作为过滤条件。

第一步：建目录结构

在 userspace/extensions/data_contract/ 下创建你的数据模块。每种自定义数据一个子目录：

userspace/extensions/data_contract/
├── data_keys.py              # 所有自定义数据的 key 注册在这里
└── dragon_tiger/             # 龙虎榜数据模块（目录名随便起）
    ├── declaration.py        # 声明：元信息 + loader 绑定
    └── loader.py             # 加载器：真正读数据的代码

第二步：注册 data key

在 data_keys.py 里给你的数据起个唯一的 key：

class USER_DATA_KEY:
   DRAGON_TIGER_DAILY = "stock.dragon_tiger.daily"

• 类名必须是 USER_DATA_KEY（固定写法，系统靠这个发现）
• key 的命名约定：{域}.{类别}.{粒度}，比如 stock.dragon_tiger.daily
• 字符串值要全局唯一，不能和内置数据重名

第三步：写数据声明

在 dragon_tiger/declaration.py 里声明数据的元信息：

from .loader import DragonTigerLoader
DRAGON_TIGER_DAILY_DECLARATION = {
   "meta": {
       "key": "stock.dragon_tiger.daily",
       "type": "time_series",
       "scope": "per_entity",
       "list_data_key": "stock.list",
       "display_name": "龙虎榜每日数据",
       "description": "每日龙虎榜上榜股票的净买入额、席位等信息",
       "unique_keys": ["date", "entity_id"],
       "loader": DragonTigerLoader,
   },
   "specific": {},
}

几个关键字段的意思：

变量名必须以 _DECLARATION 结尾，系统靠这个后缀自动发现。

第四步：写加载器

在 dragon_tiger/loader.py 里实现真正的数据读取逻辑。继承 BaseDataContractLoader，实现两个方法：

import pandas as pd
from pathlib import Path
from typing import Any, List, Mapping, Sequence
from core.modules.data_contract.core.base.base_loader import BaseDataContractLoader
CSV_PATH = Path("userspace/extensions/data_contract/dragon_tiger/data.csv")
class DragonTigerLoader(BaseDataContractLoader):
   def load(self, params: Mapping[str, Any]) -> Any:
       """加载单只股票的龙虎榜数据。"""
       entity_id = params["entity_id"]
       start_time = params.get("start_time")
       end_time = params.get("end_time")
       df = pd.read_csv(CSV_PATH)
       df = df[df["entity_id"] == entity_id]
       if start_time:
           df = df[df["date"] >= start_time]
       if end_time:
           df = df[df["date"] <= end_time]
       return df.to_dict("records")
   def load_batch(
       self,
       entity_ids: Sequence[str],
       params: Mapping[str, Any],
   ) -> Mapping[str, Any]:
       """批量加载多只股票的数据（必须用批量查询，不要循环调用 load）。"""
       start_time = params.get("start_time")
       end_time = params.get("end_time")
       df = pd.read_csv(CSV_PATH)
       df = df[df["entity_id"].isin(entity_ids)]
       if start_time:
           df = df[df["date"] >= start_time]
       if end_time:
           df = df[df["date"] <= end_time]
       # 按 entity_id 分组返回
       result = {}
       for entity_id, group in df.groupby("entity_id"):
           result[entity_id] = group.to_dict("records")
       # 没数据的股票也要返回空列表
       for eid in entity_ids:
           if eid not in result:
               result[eid] = []
       return result

两个方法的区别：

• load()：单实体加载。系统在只需要一只股票的数据时调用。
• load_batch()：批量加载。回测时要扫几千只股票，系统会调用这个方法一次性加载，必须用真正的批量查询（比如一条 SQL 查多只股票），不能在里面循环调 load()，否则性能会很差。

第五步：在策略里用

重启 NTQ 后，你的自定义数据就注册好了。在策略的 settings.py 里声明依赖：

data = {
   "base": {
       "universe": "stock.list",
       "calendar": "trade.calendar",
   },
   "required": [
       {
           "data_key": "stock.kline.daily",
           "params": {"adjust": "qfq"},
           "indicators": {"ma20": {}},
       },
       {
           "data_key": "stock.dragon_tiger.daily",
           "params": {},
       },
   ],
}

然后在策略代码里读取：

# 在策略的扫描逻辑中
dragon_tiger_data = ctx.get_data("stock.dragon_tiger.daily", entity_id)
# 返回该股票的龙虎榜数据列表
# [{"date": "20240115", "net_buy": 50000000, ...}, ...]

其他可用模式

上面的案例覆盖了最常见的"per_entity + time_series"场景。以下是其他几种数据类型的用法，供参考。

global 数据（全市场一份）

有些数据不是按股票分的，比如全市场的融资融券总额、行业景气度指数。这种用 scope: "global"：

# declaration.py
MARKET_SENTIMENT_DECLARATION = {
   "meta": {
       "key": "market.sentiment.daily",
       "type": "time_series",
       "scope": "global",
       "display_name": "市场情绪指数",
       "unique_keys": ["date"],
       "loader": MarketSentimentLoader,
   },
   "specific": {},
}

global 数据的 loader 不一样——没有 entity_id 参数：

class MarketSentimentLoader(BaseDataContractLoader):
   def load(self, params: Mapping[str, Any]) -> Any:
       # global 数据没有 entity_id
       start_time = params.get("start_time")
       end_time = params.get("end_time")
       # ... 加载数据
       return data_rows
   def load_batch(self, entity_ids, params):
       # global 数据不需要批量加载，但方法必须实现
       # 直接返回 {"global": self.load(params)} 就行
       return {"global": self.load(params)}

non_time_series 数据（静态表）

有些数据不随时间变化，比如股票的行业分类、IPO 日期。用 type: "non_time_series"：

INDUSTRY_MAPPING_DECLARATION = {
   "meta": {
       "key": "stock.industry.mapping",
       "type": "non_time_series",
       "scope": "per_entity",
       "list_data_key": "stock.list",
       "display_name": "股票行业映射",
       "unique_keys": ["entity_id"],
       "loader": IndustryMappingLoader,
   },
   "specific": {},
}

非时序数据不支持 until() 游标推进，加载后就是全量数据。

自定义 Contract 类（扩展 API）

如果你的数据有特殊的查询需求，可以自定义 Contract 子类，提供方便的方法。比如龙虎榜数据你想加一个 top_net_buy(date, n=10) 方法，返回当天净买入前 10 的股票。

在 dragon_tiger/contract.py 里写：

from core.modules.data_contract.core.base.base_time_series_contract import BaseTimeSeriesContract
class DragonTigerContract(BaseTimeSeriesContract):
   def top_net_buy(self, date: str, n: int = 10) -> list:
       """返回指定日期净买入额前 n 的股票。"""
       if not self.is_loaded:
           return []
       all_rows = []
       for entity_rows in self.data.values():
           all_rows.extend(entity_rows)
       date_rows = [r for r in all_rows if r["date"] == date]
       date_rows.sort(key=lambda x: x["net_buy"], reverse=True)
       return date_rows[:n]

然后在 declaration 里指定：

from .contract import DragonTigerContract
from .loader import DragonTigerLoader
DRAGON_TIGER_DAILY_DECLARATION = {
   "meta": {
       "key": "stock.dragon_tiger.daily",
       "type": "time_series",
       "scope": "per_entity",
       "list_data_key": "stock.list",
       "loader": DragonTigerLoader,
       "contract_class": DragonTigerContract,   # 加上这行
   },
   "specific": {},
}

然后在 declaration 里指定：

from .contract import DragonTigerContract
from .loader import DragonTigerLoader
DRAGON_TIGER_DAILY_DECLARATION = {
   "meta": {
       "key": "stock.dragon_tiger.daily",
       "type": "time_series",
       "scope": "per_entity",
       "list_data_key": "stock.list",
       "loader": DragonTigerLoader,
       "contract_class": DragonTigerContract,   # 加上这行
   },
   "specific": {},
}

使用时：

contract = ContractIssuer.issue("stock.dragon_tiger.daily", fill_in_data=True)
top10 = contract.top_net_buy("20240115", n=10)

编程式注册

如果你不想用目录自动发现，也可以在代码里手动注册：

from core.modules.data_contract import ContractIssuer
issuer = ContractIssuer()
issuer.discover()  # 先加载系统内置的
issuer.register_custom_declaration({
   "meta": {
       "key": "my.custom.data",
       "type": "time_series",
       "scope": "per_entity",
       "list_data_key": "stock.list",
       "loader": MyLoader,
   },
   "specific": {},
})

这种方式适合在测试或脚本里临时用。正式的自定义数据建议还是用目录结构。

内置数据一览

NTQ 内置了以下数据 contract，key 直接用字符串就能引用：

常见问题

Q: 自定义数据会缓存吗？ A: 会。data contract 有运行时指纹（runtime fingerprint），相同参数的数据只会加载一次。

Q: 改了 loader 代码为什么没生效？ A: 需要重启 NTQ。contract 的发现和注册在启动时完成，运行中改代码不会热更新。

Q: 我的数据量很大，加载很慢怎么办？ A: 优化 load_batch() 方法——确保用的是真正的批量查询（比如一条 SQL 查所有实体），而不是循环调 load()。另外可以考虑把数据存到数据库里，用 SQL 查询比读 CSV 快得多。

Q: 自定义数据能参与回测缓存吗？ A: 能。data contract 的 key 和 runtime 参数会参与指纹计算，如果数据没变，回测结果可以复用缓存。