表建好之后，怎么往里写数据、怎么查、怎么删？NTQ 提供了 DbBaseModel 作为统一的操作入口——不用手写 SQL，用方法调用就行。

一个完整的案例：往龙虎榜表里写入和查询数据

上一篇文档我们建了一张 dragon_tiger 表，也写了 DragonTigerModel。现在用这个 Model 完成一次完整的数据读写流程。

写入数据：upsert

你有 200 条龙虎榜数据要写入。用 upsert_many，按 entity_id + date 去重——同一天同一只股票的数据有则更新、无则插入：

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"},
    {"entity_id": "600519.SH", "date": "20240115", "net_buy_amount": 80000000, "rank": 2, "reason": "日振幅值达15%", "last_update": "2024-01-15 18:00:00"},
])
# 返回写入行数：3

再跑一次同样的数据也不会报错，重复记录会被覆盖。这就是 upsert 的好处——不用先查再决定插入还是更新。

查询数据：load

查某天所有龙虎榜数据：

rows = model.load(
    condition="date = %s",
    params=("20240115",),
    order_by="net_buy_amount DESC",
)
# 按净买入额降序排列返回
# [{"entity_id": "600519.SH", "net_buy_amount": 80000000, ...}, ...]

查某只股票的龙虎榜历史：

rows = model.load(
    condition="entity_id = %s AND date >= %s",
    params=("600000.SH", "20240101"),
    order_by="date ASC",
)

只取净买入额最高的那条：

row = model.load_one(
    condition="entity_id = %s",
    params=("600000.SH",),
    order_by="net_buy_amount DESC",
)

load 方法支持 order_by、limit、offset 参数，可以组合出分页查询：

# 第 1 页，每页 50 条
page = model.load_paginated(page=1, page_size=50)
# page["data"] 是当前页数据
# page["total"] 是总记录数
# page["total_pages"] 是总页数

统计和检查

看看表里有多少条数据：

total = model.count()
# 200

total_0115 = model.count(condition="date = %s", params=("20240115",))
# 3

检查某条记录是否存在：

exists = model.is_exists(
    condition="entity_id = %s AND date = %s",
    params=("600000.SH", "20240115"),
)
# True

删除数据

删掉某天的数据：

deleted = model.delete(
    condition="date = %s",
    params=("20240115",),
)
# 返回删除行数：3

清空整张表：

model.clear_table()

更新数据

NTQ 没有 update 方法。要更新已有记录，用 upsert —— 传入相同唯一键的新数据就会覆盖旧值：

# 600000.SH 在 20240115 的数据原来 net_buy_amount 是 50000000
# 现在修正为 55000000
model.upsert_one(
    row={"entity_id": "600000.SH", "date": "20240115", "net_buy_amount": 55000000, "rank": 1, "reason": "日涨幅偏离值达7%", "last_update": "2024-01-15 19:00:00"},
    unique_keys=["entity_id", "date"],
)

unique_keys 告诉系统按哪几个字段判断"同一条记录"。和建表时 schema.py 里的唯一索引对应。

其他可用模式

大批量数据用异步写入

如果要一次写入几万条数据（比如导入历史龙虎榜），同步写会很慢。用异步写入——数据先入队列，后台批量落库：

# 异步写入 50000 条
model.upsert_many_async(big_batch, unique_keys=["entity_id", "date"])

# 做完其他事后，等写入完成
model.wait_for_writes()

异步写入的批次大小和刷新间隔在 common.json 的 batch_write 里配置：

{
  "batch_write": {
    "enable": true,
    "batch_size": 1000,
    "flush_interval": 5.0
  }
}

• batch_size：队列攒满 1000 条就刷一次
• flush_interval：最多等 5 秒，不管攒没攒满都刷

写入完成后记得调 wait_for_writes()，否则程序退出时可能还有数据在队列里没落库。

直接写 SQL

Model 封装的方法覆盖了增删改查，但复杂查询还是 SQL 直接。DbBaseModel 提供了原始 SQL 接口：

# 查询：返回 List[Dict]
rows = model.execute_raw_query(
    "SELECT entity_id, SUM(net_buy_amount) as total FROM dragon_tiger GROUP BY entity_id ORDER BY total DESC LIMIT 10"
)
# [{"entity_id": "600519.SH", "total": 240000000}, ...]

# 写操作：返回影响行数
affected = model.execute_raw_update(
    "UPDATE dragon_tiger SET reason = %s WHERE date = %s",
    ("涨幅偏离值达标", "20240115"),
)

也可以通过 DatabaseManager 直接执行跨表查询：

from core.infra.db import Db

db = Db.manager.get_default()
rows = db.execute_sync_query(
    "SELECT a.entity_id, a.close, b.net_buy_amount FROM sys_stock_klines a JOIN dragon_tiger b ON a.id = b.entity_id AND a.date = b.date WHERE a.date = %s",
    ("20240115",),
)

事务

多张表同时写入时，需要保证要么全成功要么全失败。用 DatabaseManager 的事务上下文：

from core.infra.db import Db

db = Db.manager.get_default()

with db.transaction() as cursor:
    cursor.execute("INSERT INTO dragon_tiger (entity_id, date, net_buy_amount) VALUES ('600000.SH', '20240116', 60000000)")
    cursor.execute("UPDATE sys_meta_info SET value = '20240116' WHERE key = 'dragon_tiger_last_update'")
    # 两条都成功才提交，任一失败自动回滚

时间相关查询

时序数据有专门的便捷方法，不用手写 MAX(date) 或 MIN(date)：

# 表里最新一条记录
latest = model.load_latest(date_field="date")
# {"entity_id": "600000.SH", "date": "20240120", ...}

# 表里最早一条记录
first = model.load_first(date_field="date")
# {"entity_id": "000001.SZ", "date": "20240101", ...}

# 最新日期值（字符串）
last_date = model.load_latest_date(date_field="date")
# "20240120"

# 每只股票的最新记录
latests = model.load_latests(date_field="date", group_fields=["entity_id"])
# [{"entity_id": "600000.SH", "date": "20240118", ...}, ...]

这些方法在 K线、财务等时序数据上特别好用——比如查"每只股票的最新收盘价"只需一行代码。

不写 Model 直接操作

偶尔需要临时操作一张表，不想写 Model 类。可以直接用 Db 门面：

from core.infra.db import Db

db = Db.manager.get_default()

# 插入
db.execute_sync_query_for_table(
    "dragon_tiger",
    "INSERT INTO dragon_tiger VALUES ('600000.SH', '20240115', 50000000, 1, '涨幅偏离值', '2024-01-15 18:00:00')"
)

# 查询
rows = db.execute_sync_query_for_table(
    "dragon_tiger",
    "SELECT * FROM dragon_tiger WHERE date = %s",
    ("20240115",)
)

这种方式不推荐日常用——没有类型提示，容易写错列名。正式代码还是建议写 Model 类。