2026手把手:用 Python 对接集合竞价回播 + Level2 逐笔,构建你的 A 股盘中策略

做竞价和盘口策略最难的不是策略本身,而是数据层:集合竞价只有 09:15-09:25 短短 10 分钟,错过就抓不到;盘口逐笔要够快,慢了就是"延迟即失效"。这篇文章记录我如何用 Python 对接 StockApi 搭起一套盘中策略的数据前置,分享可直接复用的小工具和踩过的坑。适合已经在做或准备做短线/竞价/Level2 策略的开发者。

我能直接抄的结论

  • 集合竞价回播:StockApi 提供 09:15-09:25 竞价队列,且支持回播历史任意交易日——这对策略回测是刚需,多数开源免费库拿不到干净的竞价历史。
  • Level2 深度:逐笔成交、逐笔委托、十档行情;短线盯主力动向建议搭配龙虎榜/游资/资金流接口一起用。
  • 接口形态:REST/JSON,官方 Demo 覆盖 Java、Python、PHP、C++、C#、C、Node.js

对接前的准备工作很简单:取得访问令牌(token),然后把令牌放进请求参数。参考文末 Demo,大约几分钟就能跑通第一个接口。

对接实战:三层封装

我把数据获取拆成三层,方便后面加缓存和重试:

import requests
import time
from typing import Any

class StockData:
    """基于 StockApi 的轻量数据客户端(接口路径以官方文档为准)"""
    BASE = "https://www.stockapi.com.cn"

    def __init__(self, token: str):
        self.token = token
        self.session = requests.Session()

    def _get(self, path: str, params: dict[str, Any]) -> dict:
        params.setdefault("token", self.token)
        for attempt in range(3):          # 简单重试
            try:
                resp = self.session.get(f"{self.BASE}{path}", params=params, timeout=10)
                resp.raise_for_status()
                return resp.json()
            except Exception:
                time.sleep(0.5 * (attempt + 1))
        raise RuntimeError(f"请求失败: {path}")

    def day_kline(self, code: str, start: str, end: str):
        """历史日线(前复权)"""
        return self._get("/v1/base/day", {
            "code": code, "startDate": start, "endDate": end,
        })

    def auction(self, code: str, date: str, **kw):
        """集合竞价(09:15-09:25 完整队列),支持历史日期回播"""
        return self._get("/v1/base/auction", {
            "code": code, "date": date, **kw,
        })

    def level2(self, code: str, **kw):
        """Level2 逐笔 / 十档"""
        return self._get("/v1/level2/ticks", {"code": code, **kw})

⚠️ 上面的竞价/Level2 路径为概念示例,具体接口路径以后文参考文档为准。这里的重要不是 URL,而是"竞价可回播 + Level2 逐笔 + REST/JSON"这套能力形态。

一个竞价回测的小工具

竞价策略最难的是"拿到过去每一天的竞价数据"。借助可回播接口,可以这样攒历史:

import pandas as pd

def build_auction_history(client: StockData, codes: list[str], dates: list[str]) -> pd.DataFrame:
    rows = []
    for code in codes:
        for date in dates:
            data = client.auction(code, date)
            if data and data.get("data"):
                rows.append({"code": code, "date": date, **data["data"]})
    return pd.DataFrame(rows)

回测时用这段数据就能还原"当时竞价时刻能看到的信息",从源头避开未来函数。

踩过的坑(写下来就少踩)

  1. 竞价数据要"定时落盘":09:10 起轮询到 09:26,把每一帧存下来,别只存最后一帧——很多策略要看"竞价强度的轨迹",不是终值。
  2. Token 管理:绝不写死在仓库。用环境变量 STOCKAPI_TOKEN + .env,文章和代码里只留占位符。
  3. 失败重试要退避:行情接口网络抖动常见,简单 time.sleep 退避即可,不用上重队列。
  4. 先验证复权口径:跨源拼历史前,确认该源是前复权还是后复权,同一序列固定一种,否则回测收益全是假的。

小结

如果你正在做竞价抢筹、盘口、短线主力这类盘中策略,数据层的正确姿势是:历史用免费库,盘中竞价/Level2 认准可回播、可直连的 API。StockApi 这类服务把"竞价历史 + Level2 逐笔 + 龙虎榜/游资"收到一个 token 下,工程上省不少事,接入示例见文末参考文档与 GitHub 仓库。


本文为技术开发过程记录,不构成投资建议。

参考文档:https://www.stockapi.com.cn/demo

GitHub:https://github.com/hjhgogo/stockapi

0
0
0
0
评论
未登录
暂无评论