easyquotes 0.1.1

EasyQuote Python SDK — global market data (A-share, HK, US) in a few lines of code

EasyQuote Python SDK

A股、港股、美股及境外市场行情数据 Python 客户端。基于 EasyQuote HTTP API，支持实时行情、历史 K 线、逐笔成交、资金流向、基本面数据、板块数据及 WebSocket 推送。

安装

pip install easyquotes

可选扩展：

pip install "easyquotes[stream]"        # WebSocket 同步推送
pip install "easyquotes[stream-async]"  # WebSocket 异步推送

Python 版本要求：>= 3.10

快速开始

通过环境变量配置 API Key：

export EASYQUOTE_API_KEY="eq_your_key_here"

SDK 默认连接 https://122.51.7.196。如果需要覆盖服务端点，可以同时设置：

export EASYQUOTE_BASE_URL="https://quotes.example.com"

from easyquote import EasyQuoteClient

client = EasyQuoteClient()  # 自动读取 EASYQUOTE_API_KEY
quote = client.get_quote("600519.SH")
print(f"{quote.name}: {quote.price} ({quote.change_pct:+.2f}%)")

也可以直接传入 api_key：

client = EasyQuoteClient(api_key="eq_your_key_here")

标的代码格式

市场	格式示例	说明
A 股上交	600519.SH	沪市，后缀 .SH
A 股深交	000001.SZ	深市，后缀 .SZ
A 股北交	836239.BJ	北交所，后缀 .BJ
港股	00700.HK	五位数代码，后缀 .HK
美股	TSLA.US	股票代码，后缀 .US
指数	000001.SH	上证指数等

API 参考

行情 (Quotes)

方法	说明
get_quote(symbol)	单只标的实时行情，返回 Quote
get_quotes(symbols)	批量实时行情，返回 list[Quote]
get_snapshots(symbols)	批量行情快照（mac 协议），symbols 为 [{'market':'SH','code':'600519'}]
get_depths(symbols)	批量五档盘口（mac 协议）

K 线 (K-Lines)

方法	说明
get_klines(symbol, period, count, start)	A 股/港股/美股历史 K 线，返回 list[KLine]
get_index_klines(symbol, period, count, start)	指数 K 线
get_global_klines(market, code, period, count, adjust)	境外行情 K 线，adjust='qfq'（前复权）/'hfq'（后复权）

period 可选值："1m" / "5m" / "15m" / "30m" / "60m" / "daily" / "weekly" / "monthly"

分时 (Intraday)

方法	说明
get_minute(symbol)	当日分时数据，返回 list[MinuteBar]
get_minute_history(symbol, date)	历史某日分时数据，date='20260617'

逐笔 (Transactions)

方法	说明
get_transactions(symbol, start, count)	今日逐笔成交明细
get_transaction_history(symbol, date, start, count)	历史逐笔成交，date='20260617'

资金流向 (Fund Flow)

方法	说明
get_fund_flow(symbol)	当日主力资金流向，返回 FundFlow
get_fund_flow_history(symbol, start, count)	历史资金流向列表

基本面 A 股 (A-share Fundamentals)

方法	说明
get_finance(symbol)	A 股财务指标（PE/PB/EPS 等）
get_xdxr(symbol)	除权除息记录
get_announcements(code, count, page)	公司公告列表，code='600519'（不含市场后缀）

基本面 全球 (Global Fundamentals)

方法	说明
get_company(symbol)	公司概况，返回 Company，支持 A 股/港股/美股
get_executives(symbol)	高管信息，返回 list[Executive]
get_financial_reports(symbol, kind)	财务报告，kind='ALL'/'ANNUAL'/'QUARTERLY'

板块 (Sectors)

方法	说明
get_sectors(block_file)	板块数据，block_file 可选 'block_zs.dat'（行业指数）/ 'block_gn.dat'（概念）/ 'block_fg.dat'（风格）
get_board_ranking(board_type, top_n, sort_by, ascending)	板块涨跌排行，board_type='industry'/'concept'/'style'
get_board_members(board_symbol, count)	板块成员列表

市场 (Market)

方法	说明
get_market_stat()	市场统计（涨跌家数、成交额等）
get_security_list(market, start)	证券列表，market='SH'/'SZ'/'BJ'
get_security_count(market)	证券总数，返回 int

境外行情 (Global Quotes)

方法	说明
get_global_quote(market, code)	境外单只行情，market='NASDAQ'/'NYSE'/'HKEX' 等
get_global_klines(market, code, period, count, adjust)	境外 K 线

实时推送 (Streaming)

方法	说明
stream()	同步 WebSocket 推送，需安装 easyquotes[stream]
stream_async()	异步 WebSocket 推送，需安装 easyquotes[stream-async]

使用示例

批量获取行情

quotes = client.get_quotes(["600519.SH", "000001.SZ", "TSLA.US", "00700.HK"])
for q in quotes:
    print(f"{q.symbol}  {q.price:>10.2f}  {q.change_pct:+.2f}%")

K 线数据

klines = client.get_klines("600519.SH", period="daily", count=120)
for k in klines[-5:]:
    print(f"{k.datetime}  O:{k.open}  H:{k.high}  L:{k.low}  C:{k.close}  V:{k.volume}")

历史分时

bars = client.get_minute_history("600519.SH", date="20260617")
for b in bars[:10]:
    print(b.time, b.price, b.volume)

资金流向

ff = client.get_fund_flow("600519.SH")
print(f"主力净流入: {ff.net_main / 1e8:.2f} 亿")

板块排行

ranking = client.get_board_ranking(board_type="industry", top_n=10, sort_by="change_pct")
for b in ranking:
    print(b["name"], b["change_pct"])

同步 WebSocket 推送

with client.stream() as ws:
    ws.subscribe(["600519.SH", "000001.SZ", "TSLA.US"])
    for event in ws:
        if event["op"] == "quotes":
            for q in event["data"]:
                print(q["symbol"], q["last_price"])

异步 WebSocket 推送

import asyncio

async def main():
    async with client.stream_async() as ws:
        await ws.subscribe(["600519.SH", "TSLA.US"])
        async for event in ws:
            print(event)

asyncio.run(main())

逐笔成交

ticks = client.get_transactions("600519.SH", start=0, count=50)
for t in ticks:
    print(t["time"], t["price"], t["volume"], t["direction"])

境外行情

# 获取纳斯达克个股 K 线
klines = client.get_global_klines("NASDAQ", "TSLA", period="daily", count=60)

# 获取港股行情
quote = client.get_global_quote("HKEX", "00700")

错误处理

from easyquote import EasyQuoteClient
from easyquote import AuthError, RateLimitError, APIError, NotFoundError

client = EasyQuoteClient(api_key="eq_your_key")

try:
    quote = client.get_quote("600519.SH")
except AuthError:
    print("API Key 无效或未设置")
except RateLimitError:
    print("请求频率超限，请稍后重试")
except NotFoundError:
    print("标的不存在")
except APIError as e:
    print(f"API 错误 {e.code}: {e.message}")

自托管部署

如果使用自托管的 EasyQuote 后端（基于 easy-tdx 的 FastAPI 服务），推荐通过环境变量指定服务地址：

export EASYQUOTE_BASE_URL="https://quotes.example.com"

客户端会自动读取该配置：

client = EasyQuoteClient(api_key="eq_your_key")

也可以通过 base_url 参数为单个客户端显式指定，参数优先于环境变量：

client = EasyQuoteClient(
    api_key="eq_your_key",
    base_url="https://quotes.example.com",
)

HTTP 和 WebSocket 均会自动切换到指定地址（wss:// 对应 HTTPS，ws:// 对应 HTTP）。

生产环境应使用 HTTPS。通过 HTTP 访问时，API Key 会以明文在网络中传输。

许可证

MIT