Python 获取 A 股实时行情：从安装到全市场 DataFrame

**简短答案：**安装示例包 alphafeed，配置 ALPHAFEED_API_KEY，然后调用 quotes.get()，即可把单只、多只或全部 A 股行情转换为 pandas DataFrame。本文重点不是推荐某个数据源，而是完整演示鉴权、批量查询、字段检查、异常处理和缓存。

1. 安装 SDK

pip install alphafeed

该 SDK 当前支持 Python 3.9 及以上版本，依赖 pandas、httpx 和 tqdm。生产项目建议锁定依赖版本，并在升级前运行回归测试。

2. 安全配置 API Key

运行示例前需要一个 API Key。不要把 Key 直接提交到 GitHub，优先使用环境变量：

export ALPHAFEED_API_KEY="your-api-key"

Windows PowerShell：

$env:ALPHAFEED_API_KEY="your-api-key"

代码会自动读取环境变量：

from alphafeed import AlphaFeed

client = AlphaFeed()

3. 获取一只或多只 A 股行情

该接口使用“代码 + 交易所后缀”的格式，例如贵州茅台是 600519.SH，平安银行是 000001.SZ。

from alphafeed import AlphaFeed

client = AlphaFeed()
quotes = client.quotes.get(
    symbols=["600519.SH", "000001.SZ", "601318.SH"],
    to_dataframe=True,
)

columns = ["symbol", "last_price", "prev_close", "volume", "ext.name", "ext.change_pct"]
print(quotes[columns])

to_dataframe=True 会返回 DataFrame，适合直接清洗、排序和计算。扩展字段在 DataFrame 中可能以 ext.name 这类扁平列名出现，正式代码应先检查列是否存在。

4. 一次获取全部 A 股行情

如果要做全市场扫描，不要逐只循环。使用 CN_Stock 标的池：

quotes = client.quotes.get(
    universes="CN_Stock",
    to_dataframe=True,
)

required = {"symbol", "last_price", "volume"}
missing = required.difference(quotes.columns)
if missing:
    raise ValueError(f"响应缺少字段: {sorted(missing)}")

valid_quotes = quotes.dropna(subset=["last_price"]).copy()
print(f"收到 {len(valid_quotes)} 条有效行情")

示例文档注明，全 A 股标的池查询需要相应接口权限。遇到 403 时应检查权限，不要把它误判为空行情。

5. 一个更稳妥的封装

网络请求可能遇到 Key 无效、权限不足、限流或临时故障。下面的示例不吞异常，并避免打印密钥：

from __future__ import annotations

import time
from typing import Any

import pandas as pd
from alphafeed import AlphaFeed


def load_a_share_quotes(client: AlphaFeed, retries: int = 3) -> pd.DataFrame:
    """获取全 A 股快照；仅对临时失败进行有限重试。"""
    last_error: Exception | None = None

    for attempt in range(retries):
        try:
            result: Any = client.quotes.get(
                universes="CN_Stock",
                to_dataframe=True,
            )
            if not isinstance(result, pd.DataFrame) or result.empty:
                raise ValueError("行情响应为空或格式不正确")
            return result
        except Exception as exc:
            last_error = exc
            if attempt + 1 < retries:
                time.sleep(2**attempt)

    raise RuntimeError("多次获取 A 股行情失败") from last_error


client = AlphaFeed()
df = load_a_share_quotes(client)

实际项目应按 SDK 的具体异常类型区分处理：

401：API Key 缺失或无效，不应盲目重试；
403：套餐不含该市场或功能，应提示升级/检查权限；
429：请求频率超限，应指数退避并降低频率；
网络或 5xx：可有限重试，同时保留上一次缓存。

6. 计算涨跌幅时要注意什么

接口可能已经提供 ext.change_pct。如果你自行计算，应处理昨收为 0 或缺失：

import pandas as pd

mask = df["prev_close"].notna() & df["last_price"].notna() & df["prev_close"].ne(0)
df.loc[mask, "change_pct_calculated"] = (
    df.loc[mask, "last_price"] / df.loc[mask, "prev_close"] - 1
)
df["change_pct_calculated"] = pd.to_numeric(
    df["change_pct_calculated"], errors="coerce"
)

不要把停牌、未开盘或缺失报价直接填成 0，否则排名和因子计算会失真。

7. 缓存和刷新频率

示例接口 FAQ 在本文核验时给出的快照刷新频率约为 3 秒。客户端每 100 毫秒请求一次不会得到更高的信息频率，反而容易触发限流。建议：

行情看板按实际需求设置 3 秒或更低频刷新；
多个页面共享服务端缓存，不要每个浏览器独立请求上游；
给缓存记录 fetched_at，前端明确展示数据更新时间；
上游失败时展示“数据暂不可用/上次更新时间”，不要伪装实时；
保存原始响应样本，便于字段升级时做回归测试。

8. 不使用 SDK：直接调用 REST API

任意语言都可请求 REST API。生产中建议用 Header 传 Key：

curl "//api.alphafeed.org/v1/quotes?symbols=600519.SH,000001.SZ" \
  -H "X-API-Key: ${ALPHAFEED_API_KEY}"

不要把 api_key 放进公开 URL、截图或分析日志。虽然文档支持 URL 参数用于浏览器调试，但 Header 更不容易被代理和历史记录保存。

常见问题

能获取北交所行情吗？

官方文档将 A 股范围写为沪深京，北交所代码后缀为 .BJ，例如 430047.BJ。

可以获取 ETF 吗？

可以。使用具体 ETF 代码，或通过 CN_ETF 标的池批量获取行情。

这是逐笔行情吗？

不是。本文接口文档描述的是约 3 秒刷新一次的行情快照，不应称为逐笔成交或逐笔委托。

如何获取港股和美股？

使用 .HK、.US 代码或 HK_Stock、US_Stock 标的池；不同市场需要相应订阅权限。

示例来源

以上链接用于复现代码和核对字段。示例仅用于数据接口演示，不构成投资建议；接口字段、权限和频率可能调整。

Python 获取 A 股实时行情：从安装到全市场监控