taiyidata 1.0.1

太一数据官方 Python SDK，提供 A 股、基金、指数和金融搜索数据接口

taiyidata

太一数据（OneShare）官方 Python SDK v1。它为太一数据开放平台的股票、基金、指数、公司信息和金融搜索接口提供稳定的统一客户端，并默认将响应转换为适合分析的 pandas.DataFrame。

• 24 个公开接口全部覆盖
• 太一数据专属的 set_token() / taiyi_api() 使用方式
• 默认返回 DataFrame，同时保留完整原始响应
• 内置总截止时间、弱网重试、指数退避、抖动、限速和熔断器
• 线程安全的 Session 池与有界批量队列，避免大并发压垮客户端或服务端
• API Key 仅从参数、当前进程或环境变量读取，不写入磁盘

接口文档：太一数据开放平台

安装

python -m pip install taiyidata

支持 Python 3.9 及以上版本。

快速开始

配置 API Key

推荐通过环境变量配置：

export TAIYIDATA_TOKEN="sk-********"

也可以只在当前 Python 进程设置：

import taiyidata as td

td.set_token("sk-********")
api = td.taiyi_api()

还可以把 Token 直接交给客户端，优先级最高：

api = td.taiyi_api(token="sk-********")

获取 DataFrame

import taiyidata as td

api = td.taiyi_api()

df = api.get_stock_history_quotation(
    stock_code="600519.SH",
    indicators="open,high,low,latest",
    startdate="2026-01-01",
    enddate="2026-03-31",
)

print(df.head())

完整 API 响应不会丢失，可以从 DataFrame 属性中读取：

raw_response = df.attrs["raw_response"]

如果不需要 DataFrame 转换，传入 raw=True：

raw = api.get_stock_history_quotation(
    stock_code="600519.SH",
    indicators="open,high,low,latest",
    startdate="2026-01-01",
    enddate="2026-03-31",
    raw=True,
)

通用调用入口

所有显式方法最终都使用 query()。它也能调用未来新增的安全 /openapi/ 接口：

df = api.query("get_stock_code", stock_name="贵州茅台")

为了防止 API Key 被发送到第三方域名，query() 不接受任意绝对 URL。 24 个注册接口默认启用严格参数检查，拼错的参数会在发起网络请求前抛出 ParameterError。如服务端临时增加参数，可在创建客户端时设置 strict_params=False。

弱网络与重试

默认情况下，每个请求失败后最多重试 2 次。连接失败、超时、HTTP 408/429/500/502/503/504、响应体中的同类状态码，以及传输中断造成的临时 JSON 损坏都会进入统一重试流程。退避等待带随机抖动；服务端返回 Retry-After 时优先遵守， 默认最多等待 120 秒，避免异常响应让任务永久停住。429、503 和 Retry-After 会触发 客户端级共享冷却窗口，使同一批次的其他线程一起降速，避免形成重试风暴。 连续瞬时故障达到阈值后，熔断器会暂时拒绝新请求，并在恢复窗口后放行一个探测请求。

import taiyidata as td


def on_retry(event: td.RetryEvent) -> None:
    # event 不含 Token 和请求参数，可安全用于监控。
    print(event.endpoint, event.attempt, event.delay, event.reason)


api = td.taiyi_api(
    timeout=(5, 30),          # 连接超时、读取超时
    total_timeout=120,        # 排队、重试和等待的总时间预算
    retries=3,
    backoff_factor=0.5,
    retry_jitter=0.25,
    max_retry_after=120,
    max_concurrency=6,        # 整个客户端的 HTTP 并发硬上限
    requests_per_second=5,    # 可选；根据账户额度主动限速
    queue_timeout=60,         # 可选；排队过久时明确失败
    circuit_breaker_threshold=5,
    circuit_breaker_timeout=30,
    on_retry=on_retry,
)

如果希望完全接受服务端任意长度的 Retry-After，可设置 max_retry_after=None。 鉴权错误和参数错误不会重试。total_timeout=None 可关闭总截止时间，但不建议用于 无人值守任务。用完客户端后可以调用 api.close()，也可以使用 with td.taiyi_api() as api: 自动释放连接。

api.close(wait=True, timeout=30) 会停止接收新请求并等待活动连接自然结束；返回值表示 活动请求是否在指定时间内全部退出。外部传入的 Session 永远不会由 SDK 关闭。

大批量请求与有界队列

query_many() 使用受控线程池。max_workers 控制并发执行数，max_pending 控制已经 进入内存但尚未返回的最大任务数，因此输入即使是大型生成器，也不会一次性创建全部任务。 每项失败会保存在自己的结果中，不会丢失同批次已经成功的数据。

import taiyidata as td

api = td.taiyi_api(max_concurrency=8, requests_per_second=5)

requests = (
    td.BatchRequest(
        "get_stock_code",
        {"stock_name": name},
        raw=True,
        request_id=name,
    )
    for name in ["贵州茅台", "宁德时代", "中国平安"]
)

results = api.query_many(
    requests,
    max_workers=4,
    max_pending=8,
    ordered=True,
    cancel_event=None,
)

for result in results:
    if result.ok:
        print(
            result.request.request_id,
            result.value,
            result.attempts,
            result.queue_wait,
            result.request_elapsed,
        )
    else:
        print(result.request.request_id, result.error)

超大任务可以使用 iter_query_many() 按完成顺序流式消费，进一步降低内存占用：

for result in api.iter_query_many(requests, max_workers=4, max_pending=8):
    if result.ok:
        save(result.value)

如需在批次结束后统一抛错，使用 raise_on_error=True；抛出的 BatchError.results 仍包含全部成功与失败结果。业务侧应结合太一数据账户的实际频率额度设置 requests_per_second，并发数并不等于服务端允许的每秒请求数。

如需停止继续接收生成器中的新任务，可传入 threading.Event 作为 cancel_event。已经进入 执行阶段的请求会自然结束，尚未从输入迭代器读取的任务不会被消费。失败项可以直接重试：

failed = [result.request for result in results if not result.ok]
retry_results = api.query_many(failed, max_workers=2, max_pending=4)

24 个接口

编号	Python 方法	API 路径
01	get_stock_code	/openapi/get_stock_code/
02	search_securities_code	/openapi/search_securities_code/
03	get_stock_real_time_quotation	/openapi/real_time_quotation/
04	get_stock_history_quotation	/openapi/stock_history_quotation/
05	get_high_frequency_quotes	/openapi/get_high_frequency_quotes/
06	get_intraday_snapshot	/openapi/get_intraday_snapshot/
07	get_date_sequence	/openapi/get_date_sequence/
08	get_fund_realtime_valuation	/openapi/get_fund_realtime_valuation/
09	get_fund_daily_valuation	/openapi/get_fund_daily_valuation/
10	query_trading_dates	/openapi/query_trading_dates/
11	offset_trading_date	/openapi/offset_trading_date/
12	get_stock_basic_info	/openapi/get_stock_basic_info/
13	get_listed_company_info	/openapi/get_listed_company_info/
14	get_stock_equity_shareholder	/openapi/get_stock_equity_shareholder/
15	get_stock_financial_data	/openapi/get_stock_financial_data/
16	smart_stock_picking	/openapi/smart_stock_picking/
17	query_announcement	/openapi/query_announcement/
18	get_stock_daily_quotes_tech	/openapi/get_stock_daily_quotes_tech/
19	get_index_history_quotation	/openapi/get_index_history_quotation/
20	get_index_basic_info	/openapi/get_index_basic_info/
21	get_common_index_codes	/openapi/get_common_index_codes/
22	get_index_margin_trading	/openapi/get_index_margin_trading/
23	get_index_technical_indicators	/openapi/get_index_technical_indicators/
24	web_search	/v1/web-search/

金融搜索使用独立 URL：

results = api.web_search(
    query="英伟达 财报 资本开支",
    domains="sec.gov,nvidia.com",
    freshness="oneweek",
    count=10,
)

各方法的参数名称、类型和默认值与官方接口文档保持一致。

契约提示：日期序列接口线上参数名为 indicators，SDK 同时兼容文档表中的 indicators_json。日行情与技术指标页面当前未列出 indicators，但生产网关仍将其 作为必填参数，空数组只会返回空数据。SDK 以真实服务契约为准；这两个接口都会自动 把用户传入的 Python 指标数组编码为网关所需的 JSON 字符串。

DataFrame 转换

• 行情类 time + table 列式结构会转换为普通行表，多证券结果附加代码列后合并。
• data 指标数组转换为含 code、indicator、description、value 的长表。
• 普通记录数组使用 pandas.json_normalize 扁平化。
• 字典或标量转换为单行或 value 列。
• 空结果返回空 DataFrame，不作为接口失败处理。

任何时候都可使用 raw=True 跳过转换。 如果大型批次不需要在 DataFrame.attrs 中保留完整 JSON，可使用 td.taiyi_api(retain_raw_response=False) 降低内存占用。

异常处理

import taiyidata as td

try:
    df = td.taiyi_api().get_stock_code("贵州茅台")
except td.AuthenticationError:
    print("请检查 API Key")
except td.RateLimitError:
    print("请求过于频繁")
except td.ParameterError as exc:
    print(f"参数错误：{exc}")
except td.NetworkError:
    print("网络连接失败")
except td.TaiyiDataError as exc:
    print(f"接口调用失败：{exc}")

可用异常包括：

• AuthenticationError
• ParameterError
• RateLimitError
• RequestTimeoutError
• TotalTimeoutError
• QueueTimeoutError
• CircuitOpenError
• NetworkError
• ResponseDecodeError
• ServiceError
• HTTPError
• APIError
• BatchError

异常消息和 SDK 日志不会输出 API Key。

开发与验证

python -m pip install -e '.[test,quality,publish]'
python -m ruff format --check src tests
python -m ruff check src tests
python -m mypy
python -m pytest -m 'not live' --cov=taiyidata
python -m build
python -m twine check dist/*

真实接口测试会产生 API 调用，仅在同时配置以下变量时运行：

export TAIYIDATA_TOKEN="sk-********"
export TAIYIDATA_RUN_LIVE=1
python -m pytest -m live tests/test_live_contract.py

License

BSD-3-Clause

版本变更参见 CHANGELOG.md，安全问题报告流程参见 SECURITY.md。