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)

期权：历史逐笔成交 (Trades)

请求参数

期权：最新成交 (Latest Trades)

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

请求参数

期权：最新报价 (Latest Quotes)

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

请求参数

响应数据结构 (Quotes Schema)

期权：市场快照 (Snapshots)

一次性获取特定期权合约的最新成交、最新报价、隐含波动率 (IV) 及希腊字母 (Greeks) 等组合快照。

请求参数

响应数据结构补充

期权：期权链快照 (Option Chain)

获取指定底层资产 (如股票标的) 所有衍生期权合约的完整快照数据集合（期权链）。

请求参数

期权：元数据字典 (Meta Conditions & Exchanges)

/meta/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())

2. 获取新闻资讯

import requests

url = "http://bcprivateserver.site/api/v1/market/news"

headers = {"X-API-KEY": "YOUR_API_KEY"}

params = {
    "tickers": ["AAPL", "TSLA"],   # 可一次查询多个标的
    "max_rows": 5,                 # 返回5条新闻
    "order": "desc",               # 按最新排序
    "include_content": "true"      # 返回正文内容
}

response = requests.get(url, headers=headers, params=params)

print(response.json())

WebSocket 接入说明

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

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

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

期权实时成交结构 (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"]
    },
    "options": {
        # 订阅期权实时成交数据 (Real-time Trades)
        "trades":["QQQ260224C00605000"],
        # 订阅期权实时逐笔报价 (Real-time Quotes)
        "quotes":["QQQ260224C00605000"]
    }
}


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}"
                        )

                    # 期权报价 Quote
                    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已手动停止订阅。")

错误码定义