量化数据开发实战系列（第 23 篇）：北交所基础行情：股票列表、实时盘口、历史 K 线数据接入

量化数据开发实战系列（第 23 篇）：北交所基础行情：股票列表、实时盘口、历史 K 线数据接入

前言

第 22 篇我们系统地接入了南北向资金全套接口（每日净流入、历史走势、成分股持仓统计），把「聪明钱」的流向数据沉淀进了本地库。但前面的行情、财务、资金数据主线一直聚焦沪深 A 股，对**北京证券交易所（北交所/BJ）**这条独立市场线还没有落子。

第 22 篇的「下篇预告」已经点明：下一步是北交所全套数据接入实战。本篇作为北交所专题的开篇，先把最基础的行情数据打通——包括北交所股票列表、实时行情、五档盘口、指数实时，以及最重要、最常被下游因子复用的历史 K 线。

这里要特别说明一个关键差异：北交所的接口域名与沪深不同（统一走 bj/ 路径），且代码后缀固定为 .BJ；但接口返回的字段命名、结构与沪深高度一致。本篇所有接口字段均来自官方 docx，逐字照抄、零编造；接口只返回裸字段，涨跌幅、振幅、委比这类指标全部由业务代码自研。

一、接口原始字段梳理

北交所行情接口统一挂在 api.biyingapi.com/bj/ 路径下。下面按「列表 → 实时 → 盘口 → 历史」顺序给出字段表，字段名/类型/说明全部取自官方 docx。

1.1 北交所股票列表 bj/list/all/

• 接口地址：http://api.biyingapi.com/bj/list/all/{LICENCE}
• 接口说明：获取基础的股票代码和名称，用于后续接口的参数传入。数据更新：每日 16:20。
• 返回格式：标准 JSON 数组 [{},…{}]

1.2 北交所指数列表 bj/list/index/

• 接口地址：http://api.biyingapi.com/bj/list/index/{LICENCE}
• 接口说明：获取基础的北交所指数代码和名称（如北证50），用于指数实时接口的参数传入。数据更新：每日 16:20。

1.3 北交所股票实时行情 bj/stock/real/time/{股票代码}/

• 接口地址：http://api.biyingapi.com/bj/stock/real/time/{股票代码(如430017)}/{LICENCE}
• 接口说明：根据《京市股票列表》得到的股票代码获取实时交易数据（可理解为日线的最新数据），券商数据源。盘中实时更新。
• 注意：此接口路径里的代码为纯数字（如 430017，不带 .BJ 后缀）。

1.4 北交所买卖五档盘口 bj/stock/real/five/{股票代码}/

• 接口地址：http://api.biyingapi.com/bj/stock/real/five/{股票代码(如430017)}/{LICENCE}
• 接口说明：根据《京市股票列表》得到的股票代码获取实时买卖五档盘口数据，盘中实时更新。

1.5 北交所指数实时行情 bj/index/real/time/{指数代码}/

• 接口地址：http://api.biyingapi.com/bj/index/real/time/{指数代码(如899050)}/{LICENCE}
• 接口说明：根据《京市指数列表》得到的指数代码获取实时交易数据（日线最新数据），券商数据源。盘中实时更新。
• 注意：路径里的指数代码为纯数字（如 899050，不带 .BJ 后缀）。

1.6 北交所历史 K 线 bj/history/{股票代码.市场}/{分时级别}/{除权方式}/{LICENCE}?st=&et=&lt=

• 接口地址：https://api.biyingapi.com/bj/history/{股票代码.市场（如920547.BJ）}/{分时级别(如d)}/{除权方式}/{LICENCE}?st=开始时间(如20240601)&et=结束时间(如20250430)&lt=最新条数(如100)
• 接口说明：根据《京市股票列表》得到的股票代码和分时级别获取历史交易数据，交易时间升序。
• 分时级别：5 / 15 / 30 / 60 分钟，日线 d，周线 w，月线 m，年线 y。
• 除权方式（日线及以上）：不复权 n、前复权 f、后复权 b、等比前复权 fr、等比后复权 br；分钟级无除权数据，固定用 n。
• 时间格式：st / et 为 YYYYMMDD 或 YYYYMMDDhhmmss；不设置则为全部历史；lt 指定获取最新 N 条。
• 返回格式：JSON / CSV。

⚠️ 字段同名歧义（需要特别注意）：在 1.3 / 1.5 的实时行情里，pc 表示涨跌幅；而在 1.6 的历史 K 线里，pc 表示前收盘价（实时行情里的前收盘价是 yc）。两个接口 pc 含义完全不同，落库时必须分别重命名为 change_pct 与 prev_close，否则下游会算错。本篇代码已做隔离处理。

二、自研衍生指标（接口只给裸字段，以下全部自研）

北交所行情接口同样不返回任何预制统计指标，本篇自研以下衍生量：

接口仅返回原始裸字段，以上统计与盘口力量指标全部基于业务代码计算，文档不提供预制因子。

三、数据表设计

北交所数据与沪深库的建议做法是同库不同表前缀 bj_，既共享一个 quant.db，又能通过前缀清晰隔离。全部表均带联合唯一约束防重。

3.1 bj_stock_list 北交所股票列表

3.2 bj_index_list 北交所指数列表

3.3 bj_realtime 北交所个股实时行情

3.4 bj_five 北交所五档盘口

3.5 bj_index_realtime 北交所指数实时行情

3.6 bj_history 北交所历史 K 线

四、完整可运行代码

代码段开头复用系列统一基础代码块（逐字复制，不改动签名）；下方为本篇新增的北交所行情采集逻辑。演示用 limit 控制规模，避免刷爆配额。

import requests
import logging
import time
import pandas as pd
import numpy as np
import sqlite3
from apscheduler.schedulers.background import BackgroundScheduler

# ========== 全局配置 ==========
LICENCE = "你的licence"
DB_PATH = "quant.db"
LOG_FILE = "quant_collect.log"

# ———-日志初始化———-
logging.basicConfig(
filename=LOG_FILE,
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
filemode="a"
)
logger = logging.getLogger(__name__)

# ———-带重试HTTP请求（复用）———-
def biying_api_get_retry(full_url, timeout=15, max_retry=3):
for attempt in range(1, max_retry + 1):
try:
resp = requests.get(full_url, timeout=timeout)
if resp.status_code == 200:
return resp.json()
logger.warning(f"HTTP状态码异常：{resp.status_code}，第{attempt}次重试")
except Exception as e:
logger.warning(f"网络请求异常，第{attempt}次重试，错误信息：{str(e)}")
time.sleep(2)
logger.error("达到最大重试次数，接口请求失败")
return []

# ———-股票列表（复用，作为代码/交易所参数来源）———-
def fetch_stock_list():
url = f"http://api.biyingapi.com/hslt/list/{LICENCE}"
return biying_api_get_retry(url)

# =====================本篇新增：北交所行情相关=====================
def bj_code_raw(dm):
"""北交所列表返回的 dm 形如 430017.BJ，实时/五档/指数实时接口路径用纯数字代码"""
return str(dm).split(".")[0]

def bj_code_suffix(dm):
"""历史K线/财务接口路径要求带 .BJ 后缀"""
return dm if str(dm).endswith(".BJ") else f"{dm}.BJ"

def biying_api_get_flexible(full_url, timeout=15, max_retry=3):
"""历史K线/财务接口返回 JSON 或 CSV，做一次柔性解析【仅为数据演示，不构成投资建议】"""
import io
for attempt in range(1, max_retry + 1):
try:
resp = requests.get(full_url, timeout=timeout)
if resp.status_code != 200:
logger.warning(f"HTTP状态码异常：{resp.status_code}，第{attempt}次重试")
time.sleep(2)
continue
text = resp.text.strip()
if text.startswith("[") or text.startswith("{"):
return resp.json()
try:
return pd.read_csv(io.StringIO(text)).to_dict(orient="records")
except Exception:
return []
except Exception as e:
logger.warning(f"网络请求异常，第{attempt}次重试，错误信息：{str(e)}")
time.sleep(2)
logger.error("达到最大重试次数，接口请求失败")
return []

def init_bj_market_tables(db_name="quant.db"):
conn = sqlite3.connect(db_name)
cur = conn.cursor()
cur.execute("""
CREATE TABLE IF NOT EXISTS bj_stock_list (
dm TEXT, mc TEXT, jys TEXT, UNIQUE(dm)
)""")
cur.execute("""
CREATE TABLE IF NOT EXISTS bj_index_list (
dm TEXT, mc TEXT, jys TEXT, UNIQUE(dm)
)""")
cur.execute("""
CREATE TABLE IF NOT EXISTS bj_realtime (
code TEXT, price REAL, open REAL, high REAL, low REAL, pre_close REAL,
amount_yi REAL, volume_total REAL, volume_raw REAL, change REAL,
change_pct REAL, amplitude REAL, update_time TEXT, pe REAL,
turnover REAL, pb_ratio REAL, volume REAL,
UNIQUE(code, update_time)
)""")
cur.execute("""
CREATE TABLE IF NOT EXISTS bj_five (
code TEXT, sell_price REAL, buy_price REAL, sell_vol REAL,
buy_vol REAL, update_time TEXT, UNIQUE(code, update_time)
)""")
cur.execute("""
CREATE TABLE IF NOT EXISTS bj_index_realtime (
code TEXT, price REAL, open REAL, high REAL, low REAL, pre_close REAL,
amount_yi REAL, volume_total REAL, volume_raw REAL, change REAL,
change_pct REAL, amplitude REAL, update_time TEXT, pe REAL,
turnover REAL, pb_ratio REAL, volume REAL,
UNIQUE(code, update_time)
)""")
cur.execute("""
CREATE TABLE IF NOT EXISTS bj_history (
code TEXT, trade_date TEXT, open REAL, high REAL, low REAL, close REAL,
volume REAL, amount_yi REAL, prev_close REAL, suspend INTEGER,
UNIQUE(code, trade_date)
)""")
conn.commit()
conn.close()
logger.info("北交所行情数据表初始化完成")

# ———-北交所股票列表 / 指数列表———-
def fetch_bj_stock_list():
url = f"http://api.biyingapi.com/bj/list/all/{LICENCE}"
return biying_api_get_retry(url)

def fetch_bj_index_list():
url = f"http://api.biyingapi.com/bj/list/index/{LICENCE}"
return biying_api_get_retry(url)

def save_bj_stock_list(raw_list, db_name="quant.db"):
if not raw_list:
return
df = pd.DataFrame(raw_list)
df = df.replace([None, "null", ""], np.nan)
df = df.dropna(subset=["dm", "mc"])
write = df[["dm", "mc", "jys"]].copy()
conn = sqlite3.connect(db_name)
write.to_sql("bj_stock_list", conn, if_exists="append", index=False)
conn.close()
logger.info(f"北交所股票列表入库 {len(write)} 条")

def save_bj_index_list(raw_list, db_name="quant.db"):
if not raw_list:
return
df = pd.DataFrame(raw_list)
df = df.replace([None, "null", ""], np.nan)
df = df.dropna(subset=["dm", "mc"])
write = df[["dm", "mc", "jys"]].copy()
conn = sqlite3.connect(db_name)
write.to_sql("bj_index_list", conn, if_exists="append", index=False)
conn.close()
logger.info(f"北交所指数列表入库 {len(write)} 条")

# ———-北交所个股实时行情———-
def fetch_bj_realtime(dm):
code = bj_code_raw(dm)
url = f"http://api.biyingapi.com/bj/stock/real/time/{code}/{LICENCE}"
return biying_api_get_retry(url)

def clean_bj_realtime(raw_list, dm):
if not raw_list:
return pd.DataFrame()
df = pd.DataFrame(raw_list)
keep = ["p", "o", "h", "l", "yc", "cje", "v", "pv", "ud", "pc", "zf", "t", "pe", "tr", "pb_ratio", "tv"]
df = df[keep].copy()
# 重点：实时接口 pc = 涨跌幅；yc = 前收盘价，落库改名隔离
df.columns = ["price", "open", "high", "low", "pre_close", "amount", "volume_total",
"volume_raw", "change", "change_pct", "amplitude", "update_time",
"pe", "turnover", "pb_ratio", "volume"]
df = df.replace([None, "null", ""], np.nan)
num_cols = ["price", "open", "high", "low", "pre_close", "amount", "volume_total",
"volume_raw", "change", "change_pct", "amplitude", "pe", "turnover",
"pb_ratio", "volume"]
for col in num_cols:
df[col] = pd.to_numeric(df[col], errors="coerce")
df["code"] = bj_code_suffix(dm)
df["amount_yi"] = df["amount"] / 1e8
df = df.dropna(subset=["update_time"])
return df[["code", "price", "open", "high", "low", "pre_close", "amount_yi",
"volume_total", "volume_raw", "change", "change_pct", "amplitude",
"update_time", "pe", "turnover", "pb_ratio", "volume"]]

def save_bj_realtime(df, db_name="quant.db"):
if len(df) == 0:
return
conn = sqlite3.connect(db_name)
df.to_sql("bj_realtime", conn, if_exists="append", index=False)
conn.close()
logger.info(f"北交所{dm} 实时行情入库 {len(df)} 条")

# ———-北交所五档盘口 + 自研委比/委差———-
def fetch_bj_five(dm):
code = bj_code_raw(dm)
url = f"http://api.biyingapi.com/bj/stock/real/five/{code}/{LICENCE}"
return biying_api_get_retry(url)

def clean_bj_five(raw_list, dm):
if not raw_list:
return pd.DataFrame(), None
df = pd.DataFrame(raw_list)
keep = ["ps", "pb", "vs", "vb", "t"]
df = df[keep].copy()
df.columns = ["sell_price", "buy_price", "sell_vol", "buy_vol", "update_time"]
df = df.replace([None, "null", ""], np.nan)
for col in ["sell_price", "buy_price", "sell_vol", "buy_vol"]:
df[col] = pd.to_numeric(df[col], errors="coerce")
df["code"] = bj_code_suffix(dm)
df = df.dropna(subset=["update_time"])
# 自研委比 / 委差
vs = df["sell_vol"].fillna(0).sum()
vb = df["buy_vol"].fillna(0).sum()
if (vb + vs) > 0:
weibi = round((vb – vs) / (vb + vs) * 100, 2)
else:
weibi = 0.0
weicha = round(vb – vs, 2)
derived = {"code": bj_code_suffix(dm), "weibi_pct": weibi, "weicha": weicha}
return df[["code", "sell_price", "buy_price", "sell_vol", "buy_vol", "update_time"]], derived

def save_bj_five(df, db_name="quant.db"):
if len(df) == 0:
return
conn = sqlite3.connect(db_name)
df.to_sql("bj_five", conn, if_exists="append", index=False)
conn.close()
logger.info(f"北交所{dm} 五档盘口入库 {len(df)} 条")

# ———-北交所指数实时———-
def fetch_bj_index_realtime(dm):
code = bj_code_raw(dm)
url = f"http://api.biyingapi.com/bj/index/real/time/{code}/{LICENCE}"
return biying_api_get_retry(url)

def clean_bj_index_realtime(raw_list, dm):
if not raw_list:
return pd.DataFrame()
df = pd.DataFrame(raw_list)
keep = ["p", "o", "h", "l", "yc", "cje", "v", "pv", "ud", "pc", "zf", "t", "pe", "tr", "pb_ratio", "tv"]
df = df[keep].copy()
df.columns = ["price", "open", "high", "low", "pre_close", "amount", "volume_total",
"volume_raw", "change", "change_pct", "amplitude", "update_time",
"pe", "turnover", "pb_ratio", "volume"]
df = df.replace([None, "null", ""], np.nan)
num_cols = ["price", "open", "high", "low", "pre_close", "amount", "volume_total",
"volume_raw", "change", "change_pct", "amplitude", "pe", "turnover",
"pb_ratio", "volume"]
for col in num_cols:
df[col] = pd.to_numeric(df[col], errors="coerce")
df["code"] = bj_code_suffix(dm)
df["amount_yi"] = df["amount"] / 1e8
df = df.dropna(subset=["update_time"])
return df[["code", "price", "open", "high", "low", "pre_close", "amount_yi",
"volume_total", "volume_raw", "change", "change_pct", "amplitude",
"update_time", "pe", "turnover", "pb_ratio", "volume"]]

def save_bj_index_realtime(df, db_name="quant.db"):
if len(df) == 0:
return
conn = sqlite3.connect(db_name)
df.to_sql("bj_index_realtime", conn, if_exists="append", index=False)
conn.close()
logger.info(f"北交所指数{df.iloc[0]['code']} 实时行情入库 {len(df)} 条")

# ———-北交所历史 K 线———-
def fetch_bj_history(dm, level="d", fq="n", st="", et="", lt=100):
"""level: d/w/m/y 或 5/15/30/60；fq: n/f/b/fr/br（分钟级固定 n）"""
code = bj_code_suffix(dm)
base = f"https://api.biyingapi.com/bj/history/{code}/{level}/{fq}/{LICENCE}"
params = []
if st:
params.append(f"st={st}")
if et:
params.append(f"et={et}")
if lt:
params.append(f"lt={lt}")
url = base + ("?" + "&".join(params) if params else "")
return biying_api_get_flexible(url)

def clean_bj_history(raw_list, dm):
if not raw_list:
return pd.DataFrame()
df = pd.DataFrame(raw_list)
keep = ["t", "o", "h", "l", "c", "v", "a", "pc", "sf"]
df = df[keep].copy()
# 重点：历史接口 pc = 前收盘价（与实时接口的 pc=涨跌幅 不同）
df.columns = ["trade_date", "open", "high", "low", "close", "volume", "amount", "prev_close", "suspend"]
df = df.replace([None, "null", ""], np.nan)
num_cols = ["open", "high", "low", "close", "volume", "amount", "prev_close", "suspend"]
for col in num_cols:
df[col] = pd.to_numeric(df[col], errors="coerce")
df["code"] = bj_code_suffix(dm)
df["amount_yi"] = df["amount"] / 1e8
df["trade_date"] = pd.to_datetime(df["trade_date"], errors="coerce").dt.strftime("%Y-%m-%d")
df = df.dropna(subset=["trade_date"])
return df[["code", "trade_date", "open", "high", "low", "close", "volume", "amount_yi", "prev_close", "suspend"]]

def save_bj_history(df, db_name="quant.db"):
if len(df) == 0:
return
conn = sqlite3.connect(db_name)
df.to_sql("bj_history", conn, if_exists="append", index=False)
conn.close()
logger.info(f"北交所{dm} 历史K线入库 {len(df)} 条")

# ———-自研：历史K线阶段涨跌幅 / 振幅———-
def calc_bj_kline_metrics(dm, db_name="quant.db"):
"""自研：基于 bj_history 裸字段计算涨跌幅与振幅【仅为数据演示，不构成投资建议】"""
conn = sqlite3.connect(db_name)
df = pd.read_sql(
"SELECT trade_date, open, high, low, close, prev_close FROM bj_history WHERE code=? ORDER BY trade_date ASC",
conn, params=(bj_code_suffix(dm),))
conn.close()
if len(df) < 1:
return df
df["chg_pct"] = round((df["close"] – df["prev_close"]) / df["prev_close"] * 100, 2)
df["amplitude_pct"] = round((df["high"] – df["low"]) / df["prev_close"] * 100, 2)
return df[["trade_date", "close", "prev_close", "chg_pct", "amplitude_pct"]]

# ———-演示入口：采集前 limit 只北交所标的———-
def collect_bj_market_demo(limit=5):
init_bj_market_tables()
# 1) 列表
raw_stocks = fetch_bj_stock_list()
save_bj_stock_list(raw_stocks)
raw_idx = fetch_bj_index_list()
save_bj_index_list(raw_idx)
if not raw_stocks:
logger.error("北交所股票列表为空，终止演示")
return
df_stocks = pd.DataFrame(raw_stocks)
# 2) 逐只采集实时 / 五档 / 历史K线
for _, row in df_stocks.head(limit).iterrows():
dm = row["dm"]
try:
rt = fetch_bj_realtime(dm)
save_bj_realtime(clean_bj_realtime(rt, dm), dm)
five_df, derived = clean_bj_five(fetch_bj_five(dm), dm)
save_bj_five(five_df)
if derived:
logger.info(f"北交所{dm} 委比={derived['weibi_pct']}% 委差={derived['weicha']}")
# 历史日线（不复权），lt=100 控制规模
hist = fetch_bj_history(dm, level="d", fq="n", st="20250101", et="20250825", lt=100)
save_bj_history(clean_bj_history(hist, dm))
except Exception as e:
logger.warning(f"北交所{dm} 行情采集失败：{str(e)}")
time.sleep(0.3)
# 3) 指数实时（取前 limit 个指数）
if raw_idx:
df_idx = pd.DataFrame(raw_idx)
for _, row in df_idx.head(limit).iterrows():
try:
idx_rt = fetch_bj_index_realtime(row["dm"])
save_bj_index_realtime(clean_bj_index_realtime(idx_rt, row["dm"]))
except Exception as e:
logger.warning(f"北交所指数{row['dm']} 实时采集失败：{str(e)}")
time.sleep(0.3)
logger.info("北交所基础行情演示采集完成")

if __name__ == "__main__":
collect_bj_market_demo(limit=5)

五、北交所行情可视化（可选）

复用已入库的 bj_history 绘制个股走势，以及用五档自研委比画盘口力量柱状图，和沪深篇（第 11 篇）一致思路。

import matplotlib.pyplot as plt
plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False

def plot_bj_kline(dm, db_name="quant.db"):
conn = sqlite3.connect(db_name)
df = pd.read_sql(
"SELECT trade_date, close FROM bj_history WHERE code=? ORDER BY trade_date ASC",
conn, params=(dm if dm.endswith(".BJ") else dm + ".BJ",))
conn.close()
if len(df) == 0:
print("暂无北交所K线数据")
return
fig, ax = plt.subplots(figsize=(14, 6))
ax.plot(df["trade_date"], df["close"], color="#e63946", linewidth=2, label=f"{dm} 收盘价")
ax.set_title(f"北交所 {dm} 历史走势")
ax.set_xlabel("交易日")
ax.set_ylabel("收盘价")
ax.legend()
ax.grid(alpha=0.3)
plt.xticks(rotation=45)
plt.tight_layout()
plt.savefig(f"bj_kline_{dm}.png", dpi=200)
plt.show()

# plot_bj_kline("920547.BJ")

六、业务关键点

七、拓展练习

下篇预告

系列第 24 篇：北交所财报、股东、股本数据解析与存储

行情打通后，下一篇进入北交所基本面专题：资产负债表、利润表、现金流量表三张财报的清洗与长表入库，以及每股指标、十大股东、十大流通股东、股东户数、股本表接入。北交所财务接口结构与沪深完全一致，仅域名改为 bj/financial/、代码后缀改 .BJ，我们将复用第 13/14 篇的清洗与长表逻辑，演示北交所标的筛选。

免责申明：文中所有数据处理逻辑仅为编程演示，仅为数据演示，不构成投资建议。市场有风险，投资需谨慎。