接口参考

创建客户端

from zlt_finorm_client import ZltFinOrmClient

client = ZltFinOrmClient(
    endpoint="http://finorm.zltquant.com:7761",  # 服务地址，客服提供
    token="fk_你的token",                         # 访问凭据，客服提供
    timeout=30.0,                                 # 单次请求超时秒数
    on_truncate="warn",                           # 结果超上限被截断时：warn 告警，raise 抛异常
)

先搞清楚有哪些数据

下面五个函数用于发现可用数据和查询规则，都不消耗查询次数。表和字段编码是客户端内部的数据目录标识，通常由客户端或 AI 根据中文业务需求自动确认，不建议手工猜测。

想快速浏览的话，数据范围那页已经把 15 个库、476 张表和全部字段的中文名列好了，搜中文更快。

capabilities() 服务能力

client.capabilities()

{
  "deploymentMode": "hybrid",
  "serviceVersion": "0.1.0",
  "supportsJoin": True,             # 支持联表
  "supportsArrow": True,
  "supportsHotReload": True,
  "catalogVersion": "catalog-20260904095632448-bab541f5"   # 数据目录版本
}

平时用不到，排查问题时可以拿来确认服务端版本。

categories() 有哪些分类

client.categories()

{"categories": [
  {"category_code": "brd", "category_name": "板块", "table_count": 7},
  {"category_code": "bsc", "category_name": "基础", "table_count": 31},
  {"category_code": "fs",  "category_name": "财报", "table_count": 11},
  ...
]}

分类是跨数据库共用的，比如「基础」分类在 A股库和基金库里都有，table_count 是所有库加起来的数量。

tables() 有哪些表

client.tables()

{"tables": [
  {"category_code": "bsc",
   "table_code": "cneqa_bsc_stk_bsc_info",
   "table_name": "股票基本信息",
   "dataset": "cneqa/bsc/cneqa_bsc_stk_bsc_info"},
  ...
]}

dataset 的第一段就是这张表属于哪个数据库（上面这条属于 cneqa，A股库）。

这个接口会返回完整的表清单。它也接受 category 和 keyword 参数，但当前版本的筛选还没生效，传了也会返回全部表，需要自己在结果里过滤：

all_tables = client.tables()["tables"]
fs_tables = [t for t in all_tables if t["category_code"] == "fs"]        # 财报分类
name_hit = [t for t in all_tables if "股东" in t["table_name"]]           # 按中文名找

describe() 表里有哪些字段

client.describe("cneqa_bsc_stk_bsc_info")

{
  "table_code": "cneqa_bsc_stk_bsc_info",
  "table_name": "股票基本信息",
  "category_code": "bsc",
  "dataset": "cneqa/bsc/cneqa_bsc_stk_bsc_info",
  "fields": [
    {"field_code": "sec_cd",     "field_name": "证券代码", "data_type": "varchar",
     "length": 50, "is_audit_field": False},
    {"field_code": "co_full_nm", "field_name": "公司全称", "data_type": "varchar",
     "length": 50, "is_audit_field": False},
    ...
  ]
}

is_audit_field 为 True 的是数据同步用的内部字段（记录 ID、更新时间之类），查询时不会返回，忽略即可。

只想看字段名和中文名对照：

for f in client.describe("cneqa_bsc_stk_bsc_info")["fields"]:
    if not f["is_audit_field"]:
        print(f["field_code"], f["field_name"])

query_spec() 这张表该怎么查

describe 回答「有哪些字段」，query_spec 回答「这张表能怎么查」。写复杂查询之前先看一眼，能省掉很多试错。

client.query_spec("cneqa_mkt_sh_sz_bj_ab_dly_dat")

{
  "table_code": "cneqa_mkt_sh_sz_bj_ab_dly_dat",
  "table_name": "沪深京AB股日线数据表(含停牌)",

  "symbols_available": True,                      # 现在能不能用 symbols 参数
  "entity_key_format": "不含市场后缀的证券代码，如 600519",
  "entity_locator": {"mode": "SYMBOLS", "column": "stk_cd"},

  "time_axis_field": "mod_time",                  # start_date/end_date 实际过滤这一列
  "time_axis_semantics": "MODIFICATION",          # 这一列的含义：数据更新时间
  "business_date_fields": [],                     # 可用来按业务日期过滤的字段

  "default_fields": ["mkt", "stk_cd", "txn_dt", "cls_prc", ...],
  "filterable_fields": ["stk_cd", "txn_dt", "cls_prc", "trd_val", ...],
  "sortable_fields": ["txn_dt", "cls_prc", ...],

  "constraints": {
    "max_rows": 100000,          # 单次最多返回行数
    "max_symbols": 500,          # 一次最多查几只证券
    "max_date_range_days": 1825, # 时间跨度上限
    "requires_time_range": False
  }
}

几个字段值得特别留意：

symbols_available 告诉你此刻能不能用 symbols 参数。为 False 时改用 filters 按 entity_locator.column 那一列过滤，取值格式看 entity_key_format。

filterable_fields 和 sortable_fields 是能过滤、能排序的字段白名单。不在名单里的字段会被拒绝，报 QUERY_FIELD_UNKNOWN，基本就是名字拼错了。

time_axis_semantics 见下面「时间范围要当心」一节。

查询数据

query()

client.query(
    table,               # 表编码
    symbols=None,        # 证券代码列表
    start_date=None,     # 开始日期，格式 "2026-01-01"
    end_date=None,       # 结束日期
    fields=None,         # 要哪几列，不传返回默认字段
    filters=None,        # 简单过滤，如 {"stk_cd": "600519"}，见下
    filter_expr=None,    # 复杂过滤（区间、大小比较、or），见下
    order_by=None,       # 排序
    limit=None,          # 最多返回几行
    alias=None,          # 给主表起的短名，order_by 和联表要用
    joins=None,          # 联表
    format="json",       # "json"、"pandas"、"npz" 或 "arrow"
)

format="json" 返回一个字典，里面有 columns（列信息）、rows（数据行）、returned_rows（行数）、truncated（是否被截断）。format="pandas" 返回 DataFrame，截断标记放在 df.attrs 里。

format="npz" 返回 NpzResult，里面是 numpy 数组：

res = client.query(table_code, format="npz", **params)

close = res["cls_prc"]              # ndarray，可以直接算
ret = close[1:] / close[:-1] - 1

res.attrs["returned_rows"]          # 元数据，字段与 df.attrs 一致
df = res.to_pandas()                # 需要 DataFrame 时转一下

什么时候用 npz：行数上万，而且取回来要做数值计算（算因子、跑回测、做统计）。 它的价值在于给的就是 numpy 数组，可以直接喂给 numba 这类 JIT 编译器，不用先转 DataFrame。 一百万行解析实测 1.3 秒降到 0.03 秒；60 万行的分组滚动计算，npz + numba 比 pandas groupby 快 9 倍。

只是查出来看看、导出，或者接下来全是 groupby/merge 的话，用 format="pandas" 更省心—— npz 要多装 numpy，还有几处与 json 不同的语义，不做计算就换不回成本。 一千行以下体积反而比 json 大，小结果直接用 json。

用之前有三件事要知道，详见 npz 格式说明：超高精度的金额与因子列会以字符串返回（需要自己转 Decimal）、 证券代码这类重复度高的列按需展开、可空列另有一个空值标记数组。

format="arrow" 返回 ArrowResult，是跨语言的列式格式（需要先装 pyarrow）：

res = client.query(table_code, format="arrow", **params)

res.columns              # 列名
res["cls_prc"]           # 取一列（pyarrow 数组）
res.to_numpy("cls_prc")  # 转 numpy
res.to_pandas()          # 转 DataFrame
res.table                # 底层 pyarrow.Table
res.attrs                # 元数据，同 df.attrs

arrow 和 npz 只有一处不同：DECIMAL。 npz 会把声明精度超过 15 位的列降级成字符串， 而库里 58% 的 decimal 列（成交额、涨跌幅、总资产这些）都超过 15 位；arrow 用原生 decimal128 原样承载。其余列类型两者逐一对应。

按这个选：纯 Python 且后续喂 numba 做数值计算，用 npz（依赖更轻）； 要跨语言消费（C++/Rust/CUDA），或不想让高精度列变字符串，用 arrow。 详见 arrow 格式说明。

format="parquet" 还没实现，传了会报错。

validate_query()

参数和 query() 完全一样，只校验不执行，不消耗查询次数：

check = client.validate_query(table, **params)
check.valid        # True / False
check.errors       # 哪里写错了
check.suggestion   # 服务端给的修正建议

写长参数（尤其是联表）时先校验一遍，比直接查再看报错省事。

简单过滤 filters

filters 是最省事的写法：键是字段名，值是要匹配的内容。

result = client.query(
    table="cneqa_mkt_sh_sz_bj_ab_dly_dat",
    alias="d",
    filters={"stk_cd": "600519"},                  # 值是单个 → 等于
    fields=["stk_cd", "txn_dt", "cls_prc", "trd_val"],
    order_by=[{"source": "d", "field": "txn_dt", "direction": "desc"}],
    limit=5,
)

值写成列表，就是「在这几个里面」，相当于 SQL 的 IN：

filters={"stk_cd": ["600519", "000858", "600036"]}

多个键写在一起是「并且」，几个条件要同时满足：

# 等价于 stk_cd = '600519' 并且 cls_prc = 1348.86
filters={"stk_cd": "600519", "cls_prc": 1348.86}

三条规则记住就够用：

规则	说明
键必须能过滤	字段要在这张表 query_spec 的 filterable_fields 里，不在名单里报 QUERY_FIELD_UNKNOWN
只认主表的列	联表时 filters 只能过滤主表；要过滤被联表的列，用下面 filter_expr 的 source
只能写等于和列表	区间、大于小于、「或者」都表达不了，这些交给 filter_expr

filters 和 filter_expr 不要同时传。两个都给的时候只有 filter_expr 生效，filters 会被整个丢掉、不做合并，两边条件不一致时结果会和你想的差很远。

复杂过滤 filter_expr

一个条件是 field、op、value 三件套：

result = client.query(
    table="cneqa_evnt_eqty_pldg_det",                # 股权质押明细
    alias="p",
    symbols=["600635.SH"],
    filter_expr={"field": "annc_dt", "op": "between",
                 "value": ["2025-01-01", "2026-12-31"]},
    fields=["sec_cd", "sh_nm", "pldg_shr", "annc_dt"],
    order_by=[{"source": "p", "field": "annc_dt", "direction": "desc"}],
    limit=5,
)

op 一共七个：

op	含义	value 怎么写
eq	等于	单个值
in	在列表里	列表。传空列表不报错，但一行都查不到
between	闭区间，含两端	正好两个元素的列表，多一个少一个都会被拒
gt / gte	大于 / 大于等于	单个值
lt / lte	小于 / 小于等于	单个值

多个条件用 and 或 or 包起来，子条件放进 args，还能继续往下嵌套：

# 这两只股票里，收盘价超过 1000 或者成交额超过 50 亿的交易日
filter_expr={
    "op": "and",
    "args": [
        {"field": "stk_cd", "op": "in", "value": ["600519", "000858"]},
        {"op": "or", "args": [
            {"field": "cls_prc", "op": "gt", "value": 1000},
            {"field": "trd_val", "op": "gt", "value": 5000000000},
        ]},
    ],
}

联表时给条件加一个 source，说明这个字段来自哪张表；不写 source 就默认算主表：

client.query(
    "cneqa_mkt_sh_sz_bj_ab_dly_dat",
    alias="q",
    joins=[{"type": "inner", "table": "cneqa_bsc_stk_bsc_info", "alias": "b",
            "on": [{"left":  {"source": "q", "field": "stk_cd"},
                    "right": {"source": "b", "field": "sec_cd"}}]}],
    filter_expr={"op": "and", "args": [
        {"source": "q", "field": "cls_prc", "op": "gt",  "value": 1000},        # 行情表的列
        {"source": "b", "field": "lstg_dt", "op": "gte", "value": "20100101"},  # 基本信息表的列
    ]},
    fields=[
        {"source": "q", "field": "stk_cd",     "alias": "stk_cd"},
        {"source": "q", "field": "cls_prc",    "alias": "cls_prc"},
        {"source": "b", "field": "co_full_nm", "alias": "co_full_nm"},
    ],
    limit=20,
)

条件全是等值的时候，filter_expr 也接受和 filters 一样的简写，服务端会把它当成一串「等于」用 and 连起来：

filter_expr={"stk_cd": "600519", "cls_prc": 1348.86}

收到 QUERY_FILTER_MALFORMED，说明表达式的结构写坏了（and 少了 args、op 拼错之类），不是字段名的问题，不用去核对字段——字段名不对报的是 QUERY_FIELD_UNKNOWN。

证券代码怎么写

symbols 接受这几种写法，大小写都行：

写法	说明
600519.SH	代码加市场后缀，最稳妥
SH600519	市场前缀加代码
600519	只写代码。多个市场都有这个代码时会报 QUERY_SYMBOL_AMBIGUOUS，补上后缀重查

证券简称（比如「贵州茅台」）当前解析不了，请用代码。

时间范围要当心

start_date 和 end_date 过滤的是这张表的时间轴字段，而这个字段不一定是你以为的业务日期。

当前多数表的时间轴是 mod_time，也就是这条记录什么时候被更新，query_spec 里会标成 "time_axis_semantics": "MODIFICATION"。上游做一次历史数据重刷，所有记录的 mod_time 都会变成重刷那天，这时候按年份筛就可能什么都查不到，而返回的列名、格式一切正常，看不出哪里不对。

要按业务日期筛，用 filter_expr 点名具体字段：

# 筛的是「记录什么时候被更新」，不是公告日
client.query(table="cneqa_evnt_eqty_pldg_det", symbols=["600635.SH"],
             start_date="2025-01-01", end_date="2026-12-31")

# 筛的是公告日，这才是你要的
client.query(table="cneqa_evnt_eqty_pldg_det", symbols=["600635.SH"],
             filter_expr={"field": "annc_dt", "op": "between",
                          "value": ["2025-01-01", "2026-12-31"]})

具体该用哪个字段，看 query_spec 的 business_date_fields，或者在 filterable_fields 里找日期类字段（财报和事件类表常见的是 annc_dt 公告日、ddln 报告期截止日）。日期值的格式跟着字段本身走，有的表是 2026-01-01、有的是 20260101，先 describe 看一眼类型，或者先查一行看看返回长什么样。

数据鲜活性与更新时间

不同数据类型的更新节奏不同：行情数据通常按交易日或更高频率更新；财务和股东数据通常在上市公司披露后，经上游采集、清洗和同步后陆续可查。因此“数据已入库”不等于“刚刚发生”，也不保证所有表在同一时刻完成更新。

查询结果中的业务日期（如交易日、报告期、公告日）描述数据对应的业务事件；mod_time 等更新时间字段描述记录何时被同步或修改。两者含义不同，不能用更新时间替代业务日期判断数据新旧。数据是否最新，应结合表的 query_spec、返回元数据和服务公告判断。

联表

联表也走 query()，加上 alias、joins 和对象形式的 fields：

client.query(
    "cneqa_mkt_sh_sz_bj_ab_dly_dat",
    alias="q",
    joins=[{
        "type": "inner",                          # inner 或 left
        "table": "cneqa_bsc_stk_bsc_info",
        "alias": "b",
        "on": [{"left":  {"source": "q", "field": "stk_cd"},
                "right": {"source": "b", "field": "sec_cd"}}],
    }],
    symbols=["600519.SH"],
    fields=[
        {"source": "q", "field": "cls_prc",    "alias": "cls_prc"},
        {"source": "b", "field": "co_full_nm", "alias": "co_full_nm"},
    ],
)

连接条件支持多个字段同时对应（on 数组里写多条）。加上主表最多 5 张表，每一张表都需要有权限，缺一张整个查询就会被拒。

完整例子见示例。

错误码

所有异常都继承自 FinORMError，带 .code、.message、.details、.status、.retry_after。

异常类	错误码	什么意思	怎么办
AuthError	AUTH_INVALID_TOKEN	token 无效或过期	检查 token，或联系客服
AuthError	AUTH_TABLE_DENIED	没有这张表的权限	重试无用，联系客服开通
RateLimitedError	RATE_LIMITED	查询太频繁	等 retry_after 秒再试
QueryTooLargeError	QUERY_TOO_LARGE	时间跨度或证券数超限	缩小范围，分批查
EngineBusyError	ENGINE_BUSY	服务端忙	稍等重试
SymbolAmbiguousError	QUERY_SYMBOL_AMBIGUOUS	代码多义	补上 .SH / .SZ 后缀
SymbolInvalidError	QUERY_SYMBOL_INVALID	代码格式不对	检查写法
TableNotReadyError	DATA_TABLE_NOT_READY	表暂时不可查	稍后重试
TruncatedResultError	无	结果被截断且设了 on_truncate="raise"	缩小范围或调大 limit