API 开发者文档

鉴权说明

API 使用私有密钥进行鉴权。REST 请求需在 Header 中包含 X-API-KEY，WebSocket 需在连接后发送认证 JSON 包。

# REST Header 示例
headers = { "X-API-KEY": "bc_xxxxxxxxxxxx" }

# WebSocket 认证包示例
auth_payload = { "key": "bc_xxxxxxxxxxxx" }

股票接口系列 (Stocks)

股票：历史 K 线 (Bars)

请求参数

响应数据结构 (Bars Schema)

股票：最新 K 线 (Latest Bars)

获取指定股票当前交易日（或最近一个交易日）的最新一根分钟级 K 线数据。

请求参数

股票：历史逐笔成交 (Trades)

请求参数

响应数据结构 (Trades Schema)

股票：最新成交 (Latest Trades)

获取指定股票的全市场最新一笔成交记录。

请求参数

股票：历史逐笔报价 (Quotes)

获取历史全市场最优买卖报价（NBBO）的变动记录。

请求参数

响应数据结构 (Quotes Schema)

股票：最新报价 (Latest Quotes)

获取指定股票全市场实时的最新最优买卖报价（NBBO）。

请求参数

股票：集合竞价 (Auctions)

检索指定股票每日的开盘竞价 (Open Auction) 与 收盘竞价 (Closing Auction) 数据。

请求参数

股票：市场快照 (Snapshots)

一次性获取包含最新成交、最新报价、分钟线、当日线、前一日线在内的全景快照数据。

请求参数

股票：元数据字典 (Meta Conditions & Exchanges)

分页处理 (Next Page Token)

当单次查询结果超过 max_rows 时，响应体中会包含 next_page_token。将此值作为下一次请求的 page_token 参数即可获取后续数据。

Python REST 实例代码

1. 获取股票历史 K 线

import requests
url = "http://bcprivateserver.site/api/v1/market/stocks/bars"
headers = { "X-API-KEY": "YOUR_API_KEY" }
params = {
    "tickers": ["QQQ", "AAPL"], #可以一次请求多个股票代码
    "interval": "1Day",
    "start_time": "2016-01-01T00:00:00Z"
}
response = requests.get(url, headers=headers, params=params)
print(response.json())

WebSocket 接入说明

股票实时 K 线推送结构 (Bars - "b")

股票实时成交推送结构 (Trades - "t")

股票实时报价推送结构 (Quotes - "q")

WebSocket 实时订阅实例代码

import asyncio
import websockets
import json
from datetime import datetime

# ================= 配置区 =================
WS_URL = "ws://bcprivateserver.site/stream"

MY_API_KEY = "bc_xxxxxxxxxxxx"

SUB_CONFIG = {
    "action": "subscribe",
    "stocks": {
        # 订阅股票实时 K 线数据 (Real-time Bars)
        "bars":["MSFT", "TSLA"]
    }
}


async def market_data_stream():
    print(f"正在连接行情网关: {WS_URL}")

    try:
        async with websockets.connect(WS_URL) as websocket:
            print("连接成功，开始身份验证...")

            # 1. 身份验证
            await websocket.send(json.dumps({"key": MY_API_KEY}))

            # 2. 接收欢迎信息
            welcome_msg = await websocket.recv()
            print(f"服务器返回: {welcome_msg}")

            # 3. 发送订阅指令
            await websocket.send(json.dumps(SUB_CONFIG))
            print(f"订阅指令已发送: {SUB_CONFIG}")
            print("-" * 80)

            # 4. 持续接收数据
            while True:
                message = await websocket.recv()
                data_list = json.loads(message)

                for item in data_list:
                    msg_type = item.get("T")
                    symbol = item.get("S")
                    time_str = item.get("t")

                    # 股票 / ETF K线
                    if msg_type == "b":
                        print(
                            f"[K线] {symbol:<6} | "
                            f"开:{item.get('o'):<8} 高:{item.get('h'):<8} "
                            f"低:{item.get('l'):<8} 收:{item.get('c'):<8} "
                            f"量:{item.get('v'):<8} 时间:{time_str}"
                        )

                    
                    elif msg_type == "t":
                        print(
                            f"[成交] {symbol:<20} | "
                            f"价格:{item.get('p'):<8} 数量:{item.get('s'):<6} "
                            f"交易所:{item.get('x'):<4} 时间:{time_str}"
                        )

                    
                    elif msg_type == "q":
                        bid_price = item.get("bp")
                        bid_size = item.get("bs")
                        ask_price = item.get("ap")
                        ask_size = item.get("as")

                        spread = None
                        if bid_price and ask_price:
                            spread = round(ask_price - bid_price, 4)

                        print(
                            f"[报价] {symbol:<20} | "
                            f"Bid:{bid_price} x {bid_size} | "
                            f"Ask:{ask_price} x {ask_size} | "
                            f"Spread:{spread} | 时间:{time_str}"
                        )

                    # 系统通知
                    elif msg_type == "success":
                        print(f"系统通知: {item.get('msg')}")

                    # 错误
                    elif msg_type == "error":
                        print(f"错误提示: {item.get('msg')}")
                        print("服务器可能已断开连接。")

    except websockets.ConnectionClosed:
        print("连接已被服务器关闭。")

    except Exception as e:
        print(f"发生异常: {e}")


if __name__ == "__main__":
    try:
        asyncio.run(market_data_stream())
    except KeyboardInterrupt:
        print("\n已手动停止订阅。")

错误码定义