Python SDK

概述

Infoway 官方 Python SDK，提供 REST API 和 WebSocket 实时推送接口，覆盖股票、加密货币、外汇等多市场数据。

项目	说明
PyPI	infoway-sdk
Python 版本	3.9+
协议	MIT
GitHub	infoway-api/infoway-sdk-python

安装

pip install infoway-sdk

快速开始

from infoway import InfowayClient, KlineType

client = InfowayClient(api_key="YOUR_API_KEY")

# 实时行情
trades = client.stock.get_trade("AAPL.US")

# 加密货币日K线
klines = client.crypto.get_kline("BTCUSDT", kline_type=KlineType.DAY, count=30)

# 市场温度
temp = client.market.get_temperature(market="HK,US")

# 行业板块排名
plates = client.plate.get_industry("HK", limit=10)

配置

可以直接传入 api_key，也可以通过环境变量设置：

export INFOWAY_API_KEY="YOUR_API_KEY"

# 自动读取环境变量 INFOWAY_API_KEY
client = InfowayClient()

客户端选项

client = InfowayClient(
    api_key="YOUR_API_KEY",
    base_url="https://data.infoway.io",  # 默认值
    timeout=15.0,                         # 请求超时（秒）
    max_retries=3,                        # 失败重试次数
)

REST API 模块

模块	访问方式	说明
Stock	client.stock	港股、美股、A股 — 行情、盘口、K线
Crypto	client.crypto	加密货币 — 行情、盘口、K线
Japan	client.japan	日本市场 — 行情、盘口、K线
India	client.india	印度市场 — 行情、盘口、K线
Common	client.common	跨市场通用 — 行情、盘口、K线
Basic	client.basic	品种列表、交易日、交易时间、复权因子
Market	client.market	市场温度、涨跌统计、全球指数、领涨行业
Plate	client.plate	行业/概念板块、成分股、热力图
Stock Info	client.stock_info	基本面 — 估值、评级、公司概览、全景数据

行情数据方法（stock / crypto / japan / india / common）

方法	说明
get_trade(codes)	获取实时成交数据
get_depth(codes)	获取买卖盘口深度
get_kline(codes, kline_type, count)	获取K线（OHLCV）数据

基础信息方法

client.basic.get_symbols("US")             # 品种列表
client.basic.get_symbol_info("AAPL.US")    # 品种详情
client.basic.get_trading_days("US")        # 交易日历
client.basic.get_trading_hours("US")       # 交易时间
client.basic.get_adjustment_factors("AAPL.US")  # 复权因子

市场概览方法

client.market.get_temperature("HK,US")    # 市场温度/情绪
client.market.get_breadth("US")            # 涨跌统计
client.market.get_indexes()                # 全球主要指数
client.market.get_leaders("US", limit=10)  # 领涨行业
client.market.get_rank_config("US")        # 排行榜配置

板块方法

client.plate.get_industry("HK", limit=200)         # 行业板块列表
client.plate.get_concept("HK", limit=100)           # 概念板块列表
client.plate.get_members("IN20293.HK")              # 板块成分股
client.plate.get_intro("IN20293.HK")                # 行业介绍
client.plate.get_chart("HK", limit=50)              # 板块热力图

个股基本面方法

client.stock_info.get_valuation("AAPL.US")   # 估值数据
client.stock_info.get_ratings("AAPL.US")     # 机构评级
client.stock_info.get_company("AAPL.US")     # 公司概览
client.stock_info.get_panorama("AAPL.US")    # 全景数据
client.stock_info.get_concepts("AAPL.US")    # 概念标签
client.stock_info.get_events("AAPL.US", limit=20)  # 公司大事
client.stock_info.get_drivers("AAPL.US")     # 关键驱动分析

WebSocket 实时推送

import asyncio
from infoway.ws import InfowayWebSocket

async def main():
    ws = InfowayWebSocket(api_key="YOUR_API_KEY", business="stock")

    async def on_trade(msg):
        print(f"成交: {msg}")

    ws.on_trade = on_trade
    await ws.subscribe_trade("AAPL.US,TSLA.US")
    await ws.connect()

asyncio.run(main())

WebSocket 特性

• 自动重连：指数退避策略（1秒 → 30秒上限）

• 心跳保活：每30秒自动发送心跳包

• 自动重订阅：重连后自动恢复之前的订阅

• 事件回调：on_trade、on_depth、on_kline、on_error、on_reconnect、on_disconnect

WebSocket 消息码

客户端 → 服务器（订阅 / 心跳）：

码值	名称	说明
10000	SUB_TRADE	订阅成交数据
10003	SUB_DEPTH	订阅盘口数据
10006	SUB_KLINE	订阅K线数据（data.arr=[{codes, type}]，type 见下方 K线类型）
10010	HEARTBEAT	心跳保活
11000	UNSUB_TRADE	取消订阅成交
11001	UNSUB_DEPTH	取消订阅盘口
11002	UNSUB_KLINE	取消订阅K线

服务器 → 客户端（确认 / 推送）：

码值	名称	说明
10001	SUB_TRADE_ACK	成交订阅确认
10002	PUSH_TRADE	实时成交推送
10004	SUB_DEPTH_ACK	盘口订阅确认
10005	PUSH_DEPTH	实时盘口推送
10007	SUB_KLINE_ACK	K线订阅确认
10008	PUSH_KLINE	实时K线推送
11010	UNSUB_ACK	取消订阅确认

K线类型

枚举值	周期
KlineType.MIN_1 (1)	1分钟
KlineType.MIN_5 (2)	5分钟
KlineType.MIN_15 (3)	15分钟
KlineType.MIN_30 (4)	30分钟
KlineType.HOUR_1 (5)	1小时
KlineType.HOUR_2 (6)	2小时
KlineType.HOUR_4 (7)	4小时
KlineType.DAY (8)	日线
KlineType.WEEK (9)	周线
KlineType.MONTH (10)	月线
KlineType.QUARTER (11)	季线
KlineType.YEAR (12)	年线

错误处理

from infoway import InfowayClient, InfowayAPIError, InfowayAuthError, InfowayTimeoutError

client = InfowayClient(api_key="YOUR_API_KEY")

try:
    trades = client.stock.get_trade("AAPL.US")
except InfowayAuthError:
    print("API Key 无效")
except InfowayTimeoutError:
    print("请求超时")
except InfowayAPIError as e:
    print(f"API 错误 [{e.ret}]: {e.msg}")

---