智量通知识库

Appearance

示例 ​

这一页的例子都可以直接复制运行，把 endpoint 和 token 换成客服给你的值就行。示例结果来自真实数据；行情、财务和股东数据会随着上游发布和同步进度变化，请以实际查询返回为准。

准备 ​

python
from zlt_finorm_client import ZltFinOrmClient

client = ZltFinOrmClient(
    endpoint="http://finorm.zltquant.com:7761",
    token="fk_你的token",
)

用 Python 查数据 ​

查几只股票的基本信息 ​

最简单的一种：指定一张表、几只股票、想要的字段。

python
result = client.query(
    table="cneqa_bsc_stk_bsc_info",
    symbols=["600519.SH", "000858.SZ"],
    fields=["sec_cd", "co_full_nm", "lstg_dt", "iss_prc", "co_web"],
)

for row in result["rows"]:
    print(row)
text
['000858', '宜宾五粮液股份有限公司', '19980427', '14.77', 'www.wuliangye.com.cn']
['600519', '贵州茅台酒股份有限公司', '20010827', '31.39', 'www.moutaichina.com']

symbols 支持 600519、600519.SH、SH600519 几种写法，大小写无所谓。只写 600519 这种裸代码时，如果几个市场都有同样的代码会返回提示，补上 .SH 或 .SZ 再查一次即可。

查一只股票最近几天的行情 ​

行情表数据量大，用 order_by 排序、limit 限制条数，就能拿到最近几天：

python
result = client.query(
    table="cneqa_mkt_sh_sz_bj_ab_dly_dat",         # 沪深京 A/B 股日行情
    alias="d",
    symbols=["600519.SH"],
    fields=["stk_cd", "txn_dt", "opn_prc", "high_prc", "min_prc", "cls_prc", "trd_val"],
    order_by=[{"source": "d", "field": "txn_dt", "direction": "desc"}],
    limit=5,
)

for row in result["rows"]:
    print(row)
text
['600519', 20260825, 1311.89, 1317.0, 1301.11, 1304.0, 2757527040.0]
['600519', 20260824, 1271.01, 1313.8, 1270.33, 1304.66, 6299794432.0]
['600519', 20260821, 1291.5, 1291.5, 1272.01, 1272.83, 4278310912.0]
['600519', 20260820, 1299.8, 1306.88, 1291.0, 1291.5, 3280474112.0]
['600519', 20260819, 1300.0, 1308.88, 1290.5, 1307.88, 4876774912.0]

alias 是给这张表起的短名字，order_by 里要用它指明按哪张表的哪一列排序。单表查询时随便起一个字母就行。

查一只股票的财务指标 ​

财务数据通常按报告期或公告日组织。下面示例查询贵州茅台的营业收入、净利润和每股收益：

python
result = client.query(
    table="cneqa_fs_stk_financial_indicator",
    symbols=["600519.SH"],
    fields=["stk_cd", "rpt_dt", "revenue", "net_profit", "eps"],
    order_by=[{"source": "f", "field": "rpt_dt", "direction": "desc"}],
    alias="f",
    limit=8,
)

具体可用字段以 client.query_spec() 返回结果为准；不同财务表的报告期字段和指标集合可能不同。

查股东信息 ​

下面示例查询贵州茅台的股东名称、持股数量和持股比例：

python
result = client.query(
    table="cneqa_shr_stk_holder",
    symbols=["600519.SH"],
    fields=["stk_cd", "rpt_dt", "holder_nm", "hold_amt", "hold_ratio"],
    order_by=[{"source": "h", "field": "rpt_dt", "direction": "desc"}],
    alias="h",
    limit=20,
)

股东数据的报告期和披露节奏与行情不同，查询时应同时关注返回的报告期或公告日期。

联表：多张财务表一起查 ​

联表就是把多张表按共同业务键对齐，在一次请求中返回来自不同主题的数据。典型做法是：

  1. 选一张主表，例如财务指标表；
  2. 在 joins 中声明要加入的表和连接类型；
  3. 在 on 中写清楚两边用哪些字段对应，通常要同时对齐证券代码和报告期；
  4. 在 fields 中为每一列写明来源 source，避免同名字段混淆。

例如，下面把“通用主要财务指标”“通用资产负债表”和“通用现金流量表”按“机构 + 报告期”对齐，一次返回盈利、资产负债和经营现金流结果：

python
params = {
    "alias": "q",
    "joins": [{
        "type": "inner",
        "table": "cneqa_fs_gen_bs_2007ed",
        "alias": "b",
            "on": [{"left":  {"source": "q", "field": "inst_id"},
                "right": {"source": "b", "field": "inst_id"}},
               {"left":  {"source": "q", "field": "rpt_dt"},
                "right": {"source": "b", "field": "rpt_dt"}}],
    }, {
        "type": "inner",
        "table": "cneqa_fs_gen_cfs_2007ed",
        "alias": "c",
        "on": [{"left":  {"source": "q", "field": "inst_id"},
                "right": {"source": "c", "field": "inst_id"}},
               {"left":  {"source": "q", "field": "rpt_dt"},
                "right": {"source": "c", "field": "rpt_dt"}}],
    }],
    "symbols": ["600519.SH"],
    "fields": [
        {"source": "q", "field": "inst_id", "alias": "inst_id"},
        {"source": "q", "field": "rpt_dt", "alias": "rpt_dt"},
        {"source": "q", "field": "roe", "alias": "roe"},
        {"source": "b", "field": "tot_ast", "alias": "tot_ast"},
        {"source": "b", "field": "tot_liab", "alias": "tot_liab"},
        {"source": "c", "field": "ncfoa", "alias": "ncfoa"},
    ],
    "order_by": [{"source": "q", "field": "txn_dt", "direction": "desc"}],
    "limit": 5,
}

result = client.query("cneqa_fs_gen_key_fin_indc", **params)
for row in result["rows"]:
    print(row)
text
['600519', 20251231, 18.4, ...]

上例展示的是联表结构，实际字段编码和返回数值以当前数据目录为准。主表加上联表最多 5 张；每张参与查询的表都需要有权限。

参数写得比较长时，可以先校验再查，校验不消耗查询次数：

python
check = client.validate_query("cneqa_mkt_sh_sz_bj_ab_dly_dat", **params)
if check.valid:
    result = client.query("cneqa_mkt_sh_sz_bj_ab_dly_dat", **params)
else:
    print(check.errors)

拿到 DataFrame ​

加一个 format="pandas"，返回的就是 DataFrame，可以直接接上你的分析代码：

python
df = client.query(
    table="cneqa_mkt_sh_sz_bj_ab_dly_dat",
    alias="d",
    symbols=["600519.SH"],
    fields=["txn_dt", "cls_prc"],
    order_by=[{"source": "d", "field": "txn_dt", "direction": "desc"}],
    limit=100,
    format="pandas",
)

print(df.head())
print(df["cls_prc"].mean())

大批量取数之后要自己算：用 npz ​

format="npz" 拿到的是 numpy 数组。它的价值不在"取得快"，而在取完之后直接就能算—— 不用先转成 DataFrame，也不用把列再 .values 掏出来。

判断要不要用它，看两个条件，缺一个都不划算：

  1. 行数上万（一千行以下 npz 体积反而比 json 大）；
  2. 取回来要做数值计算——算因子、跑回测、做统计。

如果只是把数据查出来看一眼、导成 Excel，或者接下来全是 groupby/merge 这类 pandas 操作，那就直接用 format="pandas"：npz 要多装一个 numpy，还有几处 和 json 不一样的地方（详见 npz 格式说明）， 不做计算的话这些成本换不回什么。

python
res = client.query(
    table="cneqa_mkt_sh_sz_bj_ab_dly_dat",
    alias="d",
    symbols=["600519.SH", "000001.SZ"],
    start_date="2020-01-01",
    end_date="2024-12-31",
    fields=["stk_cd", "txn_dt", "cls_prc"],
    order_by=[
        {"source": "d", "field": "stk_cd", "direction": "asc"},
        {"source": "d", "field": "txn_dt", "direction": "asc"},
    ],
    format="npz",
)

close = res["cls_prc"]              # ndarray[float64]
print(res.attrs["returned_rows"], res.attrs["truncated"])

配合 numba 做逐行计算 ​

这是 npz 最划算的场景。numba 的 @njit 能把 Python 函数编译成机器码，但它只吃 numpy 数组—— DataFrame 进不去。npz 正好直接给数组，中间一次转换都不用。

还有一处顺手的便利：证券代码这类列在 npz 里是字典编码的， res.codes() 给出的是一串整数下标。@njit 里整数比较很快，而字符串数组基本用不了—— 所以这串下标正好拿来做分组键。

python
import numpy as np
from numba import njit

close = res["cls_prc"]
codes, _ = res.codes("stk_cd")      # int32 分组下标，直接当分组键用

@njit(cache=True)
def daily_return(close, codes):
    """按标的分组算日收益率；跨标的的相邻两行不能相减。"""
    out = np.full(close.size, np.nan)
    for i in range(1, close.size):
        if codes[i] == codes[i - 1]:
            out[i] = close[i] / close[i - 1] - 1.0
    return out

@njit(cache=True)
def rolling_vol(ret, codes, window):
    """按标的分组算滚动波动率。"""
    out = np.full(ret.size, np.nan)
    for i in range(window, ret.size):
        if codes[i] != codes[i - window]:
            continue                        # 窗口跨了标的，跳过
        total = 0.0
        total_sq = 0.0
        count = 0
        for k in range(i - window + 1, i + 1):
            v = ret[k]
            if not np.isnan(v):
                total += v
                total_sq += v * v
                count += 1
        if count > 1:
            mean = total / count
            var = total_sq / count - mean * mean
            out[i] = np.sqrt(var) if var > 0.0 else 0.0
    return out

ret = daily_return(close, codes)
vol = rolling_vol(ret, codes, 20)

查询时记得按 stk_cd, txn_dt 排序（上面的 order_by）。上面两个函数假定同一只 标的的行是连着的、且按日期有序，顺序乱了算出来的结果就是错的。

同样这两步计算，60 万行（500 只标的 × 1200 天）实测：

写法耗时
npz + numba15 毫秒
pandas groupby().pct_change() + rolling().std()142 毫秒

差了 9 倍，而且两者结果逐值一致。行数越多、计算越密集，差距越明显。

numba 要单独装：pip install numba --index-url https://pypi.tuna.tsinghua.edu.cn/simple。 只用 numpy 的向量化写法也比 pandas 快，只是没有 JIT 那么多。

算完再转 DataFrame ​

计算完想接着用 pandas 的话，转一下就行：

python
df = res.to_pandas()
df["ret"] = ret
df["vol_20d"] = vol

高精度金额要精确值：用 arrow ​

成交额、总资产这类列声明精度超过 15 位，npz 会把它们降级成字符串。 要精确数值（或者要把同一份数据交给 C++ / CUDA 那边算）就用 arrow：

python
import pyarrow.compute as pc

res = client.query(
    "cneqa_mkt_sh_sz_bj_ab_dly_dat",
    symbols=["600519.SH"],
    fields=["stk_cd", "txn_dt", "trd_val"],
    format="arrow",
)

# trd_val 是 decimal128，不是字符串
print(res["trd_val"].type)              # decimal128(18, 2)
print(res["trd_val"].to_pylist()[:3])   # [Decimal('...'), ...] 精确值

# 不在乎那几位精度就转 float64
as_float = pc.cast(res["trd_val"], "double").to_numpy()

需要先装 pyarrow。两条通道的完整差异见 arrow 格式说明。

用 AI 助手查数据 ​

装好 skill 和 MCP 之后（见安装），不写代码也能查。

装了 skill：AI 帮你写代码 ​

skill 是给 AI 看的使用说明。装上之后 AI 知道该先查表、再查字段、最后组装查询，而不是凭空猜一个表名。

你说：

text
用 ZltFinOrmClient 查贵州茅台和五粮液的基本信息，
要证券代码、公司全称、上市日期、发行价格。

AI 会做的事： 先调 describe 确认 cneqa_bsc_stk_bsc_info 有哪些字段，然后写出这段代码并运行：

python
client.query(
    table="cneqa_bsc_stk_bsc_info",
    symbols=["600519.SH", "000858.SZ"],
    fields=["sec_cd", "co_full_nm", "lstg_dt", "iss_prc"],
)

你看到：

text
600519  贵州茅台酒股份有限公司  20010827  31.39
000858  宜宾五粮液股份有限公司  19980427  14.77

代码留在你的编辑器里，可以自己改了接着用。

装了 MCP：AI 直接把数据取回来 ​

MCP 让 AI 直接调用数据服务，中间不经过代码。适合临时问一句、看一眼结果的场景。

你说：

text
帮我查贵州茅台最近 5 个交易日的开盘价、收盘价和成交额，按日期倒序。

AI 会做的事： 调用 run_query 工具，参数是：

json
{
  "table": "cneqa_mkt_sh_sz_bj_ab_dly_dat",
  "alias": "d",
  "symbols": ["600519.SH"],
  "fields": ["txn_dt", "opn_prc", "cls_prc", "trd_val"],
  "order_by": [{"source": "d", "field": "txn_dt", "direction": "desc"}],
  "limit": 5
}

你看到：

text
日期        开盘      收盘      成交额
20260825   1311.89   1304.00   27.58 亿
20260824   1271.01   1304.66   63.00 亿
20260821   1291.50   1272.83   42.78 亿
20260820   1299.80   1291.50   32.80 亿
20260819   1300.00   1307.88   48.77 亿

联表也一样，直接说要什么：

text
把茅台最近 5 天的收盘价和公司全称放一起给我，
公司全称在股票基本信息表里。

AI 会自己判断需要联表，组装出 joins 参数并调用 run_query。

MCP 一共提供 6 个工具：list_categories 看分类、list_tables 找表、describe_table 看字段、get_query_spec 看查询规则、validate_query 校验、run_query 查数据。你不用记这些名字，正常提需求就行。

skill 和 MCP 有什么不一样 ​

skill 是教 AI 怎么用这套东西，AI 最终写出的是可以保存复用的 Python 代码。MCP 是让 AI 直接调，快，但结果只在对话里。

两个一起装最省事：需要留代码的时候 AI 写代码，只想问一句的时候直接答。

出错了怎么办 ​

python
from zlt_finorm_client import (
    FinORMError, AuthError, RateLimitedError, SymbolAmbiguousError,
)

try:
    result = client.query("cneqa_bsc_stk_bsc_info", symbols=["600519"])
except AuthError as e:
    print("没有这张表的权限，或者 token 不对：", e.message)
except SymbolAmbiguousError as e:
    print("代码多义，补上市场后缀再试：", e.details)
except RateLimitedError as e:
    print("查得太频繁了，等", e.retry_after, "秒")
except FinORMError as e:
    print(e.code, e.message)

最常碰到的是权限问题：

text
AuthError: [AUTH_TABLE_DENIED] 无该表访问权限

这说明你的账号没有开通这张表，重试没用，联系客服开通即可。完整的错误码清单见接口参考。