Finctx Python SDK

在 Python 中查询股票、基金、财务和宏观数据，读取主题跟踪资讯，或订阅实时快讯。

1. 安装

需要 Python 3.9 或更高版本。安装下载的 SDK 包：

python -m pip install ./finctx-0.3.0-py3-none-any.whl

也可以直接安装源码包：

python -m pip install ./finctx-0.3.0.tar.gz

需要处理 Arrow 数据或使用本地数据库时，再安装相应扩展：

python -m pip install "finctx[arrow]==0.3.0"     # Arrow 数据转换
python -m pip install "finctx[marketdb]==0.3.0"  # 本地数据库

2. 配置 API Key

在 Finctx 网站创建 API Key，然后在运行 Python 的环境中设置：

export FINCTX_API_KEY="<你的 API Key>"

Windows PowerShell 使用 $env:FINCTX_API_KEY = "<你的 API Key>"。不要将真实 Key 写入代码或提交到 Git。

也可以将 FINCTX_API_KEY=<你的 API Key> 保存到 ~/.finctx/credentials.env，并将文件权限设为仅当前用户可读写（Linux／macOS 使用 chmod 600 ~/.finctx/credentials.env）。SDK 按“传入的 api_key → 环境变量 → 用户凭据文件”的顺序读取 Key，不会自动加载项目的 .env。

3. 第一次查询

下面查询证券简称，并把结果提取为 Python 字典列表：

from finctx import FinctxClient

with FinctxClient() as client:
    result = client.get_s_stockname(code=["600745.SH"])
    rows = client.rows(result)
    print(rows)

接口调用需要相应权限，并按账号的额度和计费规则使用。rows() 适用于包含 data_frame 的表格结果；如果响应格式不符，会抛出错误。

查找接口和参数

这些查询在本地执行，无需 API Key：

from finctx import FinctxClient

print(FinctxClient.endpoints()["daily"]["parameters"])
print(FinctxClient.fields("daily", search="收盘"))
help(FinctxClient.daily)

endpoints() 列出可调用的接口及其参数，fields() 查找返回字段，help() 查看方法签名。调用业务方法时使用关键字参数；具体返回内容取决于所选接口。

处理 Arrow 数据

安装 arrow 扩展后，可以把 Arrow 结果转换成表格：

from finctx import FinctxClient

with FinctxClient() as client:
    raw = client.arrow(
        handler="get_s_stockname",
        params={"code": ["600745.SH"]},
        fields=["code", "sename"],
    )
    table = client.arrow_to_table(raw)
    print(table.to_pylist())

Arrow 接口返回原始 bytes，转换由调用方显式执行。

读取主题跟踪资讯

读取关注范围的内容、搜索资讯，或查看关注的主题：

from finctx import FinctxClient

with FinctxClient() as client:
    feed = client.get_tracer_feed(limit=20)
    print(feed["items"])

    page = client.search_tracer_content(q="美债", scope="following", limit=20)
    print(page["items"])

    topics = client.list_tracer_topics()
    print(topics)

资讯结果使用 items，直接读取，不调用 rows()。新闻搜索返回 has_more=True 时，下一页传入 next_cursor，并保留相同筛选条件及返回的时间范围；Python 参数 from_、to 分别使用 window["from"]、window["to"]。

主题创建使用 create_tracer_topic(query=...)；关注和取消关注使用 follow_tracer_topic(id=...)、unfollow_tracer_topic(id=...)，主题 ID 从主题列表取得。创建回执表示任务已受理，不代表内容已生成完成。

订阅实时快讯

from finctx import FinctxClient

with FinctxClient() as client:
    with client.ws_newsflash(limit=10) as stream:
        print(stream.receive_json())

SDK 不自动重连。持续订阅时，由应用保存已处理消息的游标并处理断线。主题跟踪的 SSE 推送需使用支持流式读取的 HTTP 客户端，SDK 的 request() 不支持持续读取 SSE。

使用本地数据库

安装 marketdb 扩展后，可以将数据保存到本地 DuckDB 文件，并用 SQL 查询：

from finctx.marketdb import DATASETS, MarketDB

db = MarketDB()
db.init()
print(sorted(DATASETS))  # 查看支持同步的数据集
print(db.status())      # 查看本地数据和同步记录

通过 db.sync(数据集名称, **查询参数) 下载数据，db.sql(SQL) 执行只读查询，db.export(表名, 文件路径) 导出 CSV 或 Parquet。

同步后检查返回的 status：complete 表示本次查询范围已完整写入，incomplete 表示未确认完整，不写入数据。单次同步上限为 100,000 行或约 64 MiB；超限时缩小查询范围。validate() 检查本地数据库，不验证远端是否还有更多历史数据。

调整地址和超时

通常只需配置 API Key。需要覆盖默认设置时使用：

设置	默认值	配置方式
HTTP 服务地址	https://api.ultrastiching.com	base_url= 或 FINCTX_BASE_URL
HTTP 超时	30 秒	timeout=
WebSocket 服务地址	wss://ws.ultrastiching.com	websocket_base_url=
WebSocket 超时	30 秒	websocket_timeout=
本地数据库路径	~/.finctx/data/marketdb.duckdb	MarketDB(path=...) 或 FINCTX_DB

显式参数优先于环境变量。服务地址需包含协议，如 https://api.example.com，可带端口，不附带接口路径、查询参数或片段；HTTP 地址设置不会同时修改 WebSocket 地址。

处理错误

from finctx import FinctxApiError, FinctxClient, FinctxTransportError

try:
    with FinctxClient() as client:
        result = client.get_s_stockname(code=["600745.SH"])
except FinctxApiError as error:
    print(error.status_code, error.error_code, error.request_id)
except FinctxTransportError as error:
    print(error.cause)

FinctxApiError 提供 HTTP 状态、服务端错误码、请求编号和原始响应；FinctxTransportError 提供网络错误原因。响应无法解析为 JSON 时抛出 FinctxResponseError，表格结果格式不符时抛出 FinctxResultError。根据具体错误处理，SDK 默认不自动重试。

从旧版升级

0.3.0 的安装包与导入名统一为 finctx，客户端为 FinctxClient，异常类使用 Finctx 前缀。升级时更新导入、异常捕获和环境变量；新版不自动读取旧配置或旧用户目录。已有数据库可通过 FINCTX_DB 或 MarketDB(path=...) 继续打开。

许可证：Apache License 2.0。