NTQ 的所有数据表都用 Python 字典声明——没有 SQL 文件，不用手写 CREATE TABLE。你在 schema.py 里定义表结构，系统启动时自动发现并建表。

内置表放在 core/tables/ 下，用户自定义表放在 userspace/extensions/tables/ 下。

一个完整的案例：创建一张龙虎榜数据表

你接入了龙虎榜的 CSV 数据，现在想把它存进 NTQ 的数据库里，方便策略和扫描时快速查询。你需要创建一张表来存这些数据。

第一步：建目录结构

在 userspace/extensions/tables/ 下创建你的表目录：

userspace/extensions/tables/
└── dragon_tiger/            # 目录名随便起（建议和表名一致）
    ├── schema.py             # 必需：表结构定义
    └── model.py              # 可选：数据操作类

每个表一个子目录，目录里至少有一个 schema.py。系统启动时自动扫描所有含 schema.py 的目录。

第二步：写 schema.py

schema.py 里定义一个 schema 字典变量，描述表的结构：

schema = {
    "storage_domain": "strategy",
    "update_key": "user_dragon_tiger",
    "name": "dragon_tiger",
    "primaryKey": ["entity_id", "date"],
    "fields": [
        {
            "name": "entity_id",
            "type": "varchar",
            "length": 16,
            "isRequired": True,
            "nullable": False,
            "description": "股票代码",
        },
        {
            "name": "date",
            "type": "varchar",
            "length": 8,
            "isRequired": True,
            "nullable": False,
            "description": "日期 YYYYMMDD",
        },
        {
            "name": "net_buy_amount",
            "type": "float",
            "isRequired": True,
            "nullable": True,
            "description": "净买入额（元）",
        },
        {
            "name": "rank",
            "type": "int",
            "isRequired": False,
            "nullable": True,
            "description": "当日排名",
        },
        {
            "name": "reason",
            "type": "varchar",
            "length": 255,
            "isRequired": False,
            "nullable": True,
            "description": "上榜原因",
        },
        {
            "name": "last_update",
            "type": "datetime",
            "isRequired": True,
            "nullable": True,
            "description": "本行最后同步时间",
        },
    ],
    "indexes": [
        {"name": "idx_date", "fields": ["date"]},
        {"name": "idx_entity_date", "fields": ["entity_id", "date"], "unique": True},
    ],
}

关键字段的意思：

每个字段定义里：

支持的字段类型：

第三步：写 model.py（可选但推荐）

schema.py 只是声明了表结构，要往表里读写数据，还需要一个 Model 类。在 model.py 里继承 DbBaseModel：

from typing import Any, Dict, List
from core.infra.db.contracts import DbBaseModel
from .schema import schema as _schema

class DragonTigerModel(DbBaseModel):
    def __init__(self, db=None):
        super().__init__(_schema["name"], db)

    def save_records(self, records: List[Dict[str, Any]]) -> int:
        """批量写入龙虎榜数据（有则更新，无则插入）。"""
        return self.upsert_many(records, unique_keys=["entity_id", "date"])

    def get_by_date(self, date: str) -> List[Dict[str, Any]]:
        """查询某天的龙虎榜数据。"""
        return self.load(where="date = ?", params=[date])

DbBaseModel 提供了常用的数据操作方法，你只需要写业务相关的封装：

第四步：使用

重启 NTQ，系统自动发现 userspace/extensions/tables/dragon_tiger/schema.py，创建 dragon_tiger 表。之后就可以用了：

from userspace.extensions.tables.dragon_tiger.model import DragonTigerModel

model = DragonTigerModel()

# 写入数据
model.save_records([
    {"entity_id": "600000.SH", "date": "20240115", "net_buy_amount": 50000000, "rank": 1, "reason": "日涨幅偏离值达7%", "last_update": "2024-01-15 18:00:00"},
    {"entity_id": "000001.SZ", "date": "20240115", "net_buy_amount": -30000000, "rank": 3, "reason": "日跌幅偏离值达-7%", "last_update": "2024-01-15 18:00:00"},
])

# 查询数据
rows = model.get_by_date("20240115")
# [{"entity_id": "600000.SH", "date": "20240115", "net_buy_amount": 50000000, ...}, ...]

如果你同时写了 data contract 的 loader，可以让 loader 从这张表读数据，而不是读 CSV——查询快得多，还支持时间范围过滤。

其他可用配置

storage_domain 怎么选

用户表如果没写 storage_domain，默认归入 strategy 域。

update_key 的作用

update_key 是表迁移的唯一锚点。NTQ 升级时，迁移工具会比较旧 schema 快照和当前 schema.py 的差异，自动生成 ALTER TABLE 语句（加列、加索引等）。update_key 就是用来追踪"这张表是哪个声明"的。

规则：

• 全局唯一，不能和其他表重复
• 一旦确定就不要改（改了等于声明了一张新表）
• 命名建议：user_{表名}（用户表）或 {域}_{表名}（内置表）

多字段主键 vs 唯一索引

NTQ 的主键和唯一索引是两个不同的约束：

• primaryKey：表级主键约束。DuckDB 下会创建唯一索引。
• indexes 里加 "unique": True：唯一索引。

两者效果类似，但主键不能为 NULL，唯一索引可以。建议：用 primaryKey 声明业务主键，用唯一索引做额外约束。

不写 model.py 行不行

行。schema.py 单独就能建表。只是没有 Model 类的话，你需要直接用数据库操作类来读写数据：

from core.infra.db import Db

db = Db.manager.get_default()
db.insert("dragon_tiger", {"entity_id": "600000.SH", "date": "20240115", "net_buy_amount": 50000000})

但这种方式不推荐——没有类型提示，也没有业务方法封装。写个 Model 类花不了几分钟，用起来方便得多。

内置表的结构

内置表放在 core/tables/ 下，按域分子目录：

core/tables/
├── stock/          # 股票相关表
│   ├── stock_list/       # sys_stock_list（股票列表）
│   ├── stock_klines/     # sys_stock_klines（K线行情）
│   ├── stock_indicators/ # sys_stock_indicators（技术指标）
│   ├── corporate_finance/# sys_corporate_finance（财务数据）
│   └── ...
├── index/          # 指数相关表
├── calendar/        # 日历表
├── macro/           # 宏观数据表
├── tag/             # 标签表
└── system/          # 系统表

每个内置表的 schema.py 格式和用户表完全一样，只是表名带 sys_ 前缀。你可以参考内置表的写法来写自己的表。

常见问题

Q: 建好的表在哪看？ A: DuckDB 模式下，表在 userspace/system/db/ 对应的 .duckdb 文件里。用 DuckDB CLI 或 DBeaver 打开文件即可查看。

Q: 改了 schema.py 之后表结构会自动更新吗？ A: 不会。系统启动时只执行 CREATE TABLE IF NOT EXISTS——表已存在就跳过。改了字段需要手动执行迁移：python -m core.infra.db.core.migrate_manager plan 然后 apply。

Q: 用户表名能和内置表重名吗？ A: 不能。表名是全局唯一的。内置表用 sys_ 前缀就是为了避免和用户表冲突。

Q: 我用 MySQL/PostgreSQL，建表流程一样吗？ A: 一样。schema.py 是数据库无关的，系统会根据你配置的数据库类型自动生成对应的 DDL。唯一区别是 DuckDB 按域分文件，MySQL/PostgreSQL 所有表在同一个库里。