量化数据进阶：多维实战篇（第 3 篇）：沪深实时行情快照：全市场涨跌家数、量比异动与振幅排名

量化数据进阶：多维实战篇（第 3 篇）：沪深实时行情快照：全市场涨跌家数、量比异动与振幅排名

一、前言

上一篇文章《融资融券数据实战：两市概览与个股两融明细》里，我们把「杠杆资金」这条进阶数据线打通了——既抓到了两市融资融券总量（hitc/rzrqzl），也把个股两融明细（hitc/rzrqmx）落进了数据库，并自研了「融资净买入额」「融券净卖出量」等衍生指标。

杠杆资金看的是「谁在借钱买、谁在借券卖」，而本篇要切换到一个更宏观的视角：在同一时刻，全市场几千只股票到底涨多跌少、哪些在放量、哪些在剧烈波动？ 这是我们做板块强弱、情绪温度、异动雷达的地基。

必盈数据提供了一个「一次性拉全市场实时交易数据」的接口 hsrl/real/all，本篇我们就用它把全市场快照接进来，再完全用 Python 手写一批盘面情绪指标。

衔接说明：本篇是「模块二·实时行情与盘口」的第一篇，下一篇（第 4 篇）将下钻到单只股票的实时盘口与五档委买委卖。

二、接口字段解读

接口地址（来源于官方文档，全市场实时快照，仅限白金版/包年版，1 分钟 1 次）：

https://all.biyingapi.com/hsrl/real/all/{您的licence}

返回格式为 JSON 数组 [{}, …{}]，每个元素是一只股票的实时快照。原始字段如下（接口只给裸字段，不含任何衍生指标）：

注意：文档中该接口 dm 仅返回股票代码（不带 .SH/.SZ 后缀），因此「沪深京」市场归类必须用代码前缀自行判断，这正是下面自研指标要做的事。

三、自研衍生指标 / 业务逻辑

接口只吐裸字段，以下指标全部用 Python 计算（接口本身不带这些聚合/筛选字段）：

四、数据表设计

实时快照本质是一张「某个更新时刻的全市场截面」，按 (dm, t) 去重即可多次采集形成时间序列。

CREATE TABLE IF NOT EXISTS market_snapshot (
id INTEGER PRIMARY KEY AUTOINCREMENT,
dm TEXT, — 股票代码
t TEXT, — 更新时间
p REAL, o REAL, h REAL, l REAL, yc REAL,
cje REAL, v REAL, pc REAL, ud REAL, zf REAL, fm REAL,
hs REAL, lb REAL, pe REAL, sz REAL, lt REAL,
sjl REAL, zdf60 REAL, zdfnc REAL, zs REAL,
market TEXT, — 自研：SH/SZ/BJ
UNIQUE(dm, t)
);

五、完整可运行代码

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"
API_BASE = "http://api.biyingapi.com"
ALL_BASE = "https://all.biyingapi.com"

# ———- 日志初始化 ———-
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 classify_market(dm):
dm = str(dm)
if dm.startswith(("60", "68", "9")):
return "SH"
if dm.startswith(("00", "30")):
return "SZ"
if dm.startswith(("8", "4", "92")):
return "BJ"
return "OTHER"

# ———- 全市场实时快照采集 ———-
def fetch_market_snapshot():
url = f"{ALL_BASE}/hsrl/real/all/{LICENCE}"
data = biying_api_get_retry(url)
if not isinstance(data, list) or len(data) == 0:
logger.error("全市场快照返回为空或非列表")
return pd.DataFrame()
df = pd.DataFrame(data)
keep = ["dm", "p", "o", "h", "l", "yc", "cje", "v", "pc", "ud", "zf",
"fm", "hs", "lb", "pe", "sz", "lt", "sjl", "zdf60", "zdfnc", "zs", "t"]
keep = [c for c in keep if c in df.columns]
df = df[keep].copy()
df = df.replace([None, "null", ""], np.nan)
num_cols = ["p", "o", "h", "l", "yc", "cje", "v", "pc", "ud", "zf",
"fm", "hs", "lb", "pe", "sz", "lt", "sjl", "zdf60", "zdfnc", "zs"]
for c in num_cols:
if c in df.columns:
df[c] = pd.to_numeric(df[c], errors="coerce")
df["market"] = df["dm"].apply(classify_market)
return df

# ———- 盘面情绪指标（全部自研） ———-
def compute_market_sentiment(df):
if df.empty:
return {}
valid = df.dropna(subset=["ud"])
up = int((valid["ud"] > 0).sum())
down = int((valid["ud"] < 0).sum())
flat = int((valid["ud"] == 0).sum())
avg_pc = float(valid["pc"].mean())
total = len(valid)
return {
"上涨家数": up,
"下跌家数": down,
"平盘家数": flat,
"全市场平均涨跌幅(%)": round(avg_pc, 4),
"涨跌比(上涨/下跌)": round(up / down, 4) if down else None,
"样本总数": total,
}

def top_by_column(df, col, n=10, ascending=False):
if df.empty or col not in df.columns:
return pd.DataFrame()
sub = df.dropna(subset=[col]).sort_values(col, ascending=ascending).head(n)
return sub[["dm", "market", col, "pc", "p"]]

def volume_anomaly(df, lb_threshold=1.5):
if df.empty or "lb" not in df.columns:
return pd.DataFrame()
sub = df[df["lb"] >= lb_threshold].copy()
sub = sub.sort_values("lb", ascending=False)
return sub[["dm", "market", "lb", "pc", "hs"]]

# ———- 落库 ———-
def save_snapshot(df):
if df.empty:
return
conn = sqlite3.connect(DB_PATH)
try:
cols = ["dm", "t", "p", "o", "h", "l", "yc", "cje", "v", "pc", "ud",
"zf", "fm", "hs", "lb", "pe", "sz", "lt", "sjl", "zdf60",
"zdfnc", "zs", "market"]
cols = [c for c in cols if c in df.columns]
df[cols].to_sql("market_snapshot", conn, if_exists="append",
index=False, method="multi")
conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS uk_snap ON market_snapshot(dm, t)"
)
conn.commit()
logger.info(f"落库全市场快照 {len(df)} 行")
except Exception as e:
logger.error(f"落库失败：{str(e)}")
finally:
conn.close()

# ———- 演示入口 ———-
def demo(limit_n=20):
df = fetch_market_snapshot()
if df.empty:
print("未取到数据，请检查 LICENCE 与网络")
return
sentiment = compute_market_sentiment(df)
print("=== 盘面情绪 ===")
for k, v in sentiment.items():
print(f"{k}: {v}")
print("\\n=== 振幅榜 Top10 ===")
print(top_by_column(df, "zf", n=10))
print("\\n=== 换手榜 Top10 ===")
print(top_by_column(df, "hs", n=10))
print("\\n=== 量比异动（lb>=1.5）Top10 ===")
print(volume_anomaly(df).head(10))
save_snapshot(df)
print(f"\\n【仅为数据演示，不构成投资建议】已落库 {len(df)} 只股票快照")

if __name__ == "__main__":
demo(limit_n=20)

六、业务关键点

七、拓展练习

八、下篇预告

下一篇《实时盘口与五档：买卖盘博弈与委比委差》将下钻到单只股票，用 hsstock/real/time 与 hsstock/real/five 把实时最新价、五档委买委卖价量全部接进来，并自研「委比」「委差」「买卖盘力量」等指标，从全市场温度细化到个股盘口博弈。

九、免责申明

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