上一篇我们监控了 ETF 与跟踪指数的偏离。但 ETF 只覆盖大类资产和行业，如果你想抓住更细的概念轮动——比如今天半导体、人工智能、低空经济、机器人哪个板块在异动——就需要直接监控概念板块的成分股。

这篇文章用 Python 批量拉取概念板块成分股，计算板块平均涨跌幅，输出盘中概念板块异动排行榜。

一、本文你将得到什么

  1. 用 sectors_list / sector_detail 找到概念板块的成分股。
  2. 用 stock_ssjy_more 批量接口一次拉 20 只成分股行情。
  3. 计算板块平均涨跌幅、最大涨幅、领涨股、总成交额。
  4. 输出板块异动排行榜，定位资金主攻方向。
  5. 一个可扩展的板块监控脚本，替换列表就能监控任意概念。

二、环境准备

本文代码使用 mairui SDK 获取板块与股票数据，安装方法如下：

pip install mairui

运行环境：

  • Python 3.9+
  • mairui SDK 1.0.0
  • pandas（用于表格展示）

本文数据截至 2026-08-08，读者复现时数据可能因行情更新而略有差异。

证书直接写在代码里（文中使用演示证书，演示证书（读者复制即用，无需替换））。

import sys, os

三、概念板块与成分股

麦蕊提供了板块列表和板块明细接口：

from mairui import Client

api = Client(licence="LICENCE-66D8-9F96-0C7F0FBCD073")  # 演示证书，演示证书（读者复制即用，无需替换）

# 获取板块列表
sectors = api.sectors_list()
print(sectors[:3])
# [{'dm': '101075.BKZS', 'mc': 'GNPM2.5', 'jys': 'BK'}, ...]

# 获取某个板块的成分股
detail = api.sector_detail(sector_name="GN半导体")
print(detail["count"])
print(detail["stocks"][:3])
# {'dm': '000016.SZ', 'mc': '*ST康佳A', 'jys': 'SZ'}

关键字段：

  • sectors_list 返回：dm（板块代码）、mc（板块名称）、jys（交易所）。
  • sector_detail 返回：sector_name、count（成分股数量）、stocks（成分股列表）。
  • 成分股代码带交易所后缀，如 000016.SZ，传给 stock_ssjy_more 前要去掉后缀。

四、批量拉取成分股实时行情

用 stock_ssjy_more 批量接口，每次最多拉 20 只：

import time

def fetch_batch_realtime(codes):
    api = Client(licence="LICENCE-66D8-9F96-0C7F0FBCD073")  # 演示证书，演示证书（读者复制即用，无需替换）
    records = []
    for i in range(0, len(codes), 20):
        batch = codes[i:i + 20]
        batch_records = api.stock_ssjy_more(stock_codes=",".join(batch))
        records.extend(batch_records)
        time.sleep(0.05)
    return records

# 示例：取半导体板块前 30 只成分股
stocks = api.sector_detail(sector_name="GN半导体")["stocks"][:30]
codes = [s["dm"].split(".")[0] for s in stocks]  # 去掉 .SZ/.SH
records = fetch_batch_realtime(codes)
print(len(records))

关键解释：

  • stock_ssjy_more 一次最多支持 20 只，逗号分隔。
  • 每个板块 30 只成分股需要 2 次批量调用，耗时约 0.2~0.4 秒。
  • 返回字段：dm（代码）、mc（名称）、p（现价）、pc（涨跌幅%）、zf（振幅%）、cje（成交额元）。

五、计算板块异动分数

把成分股行情汇总成板块维度：

import pandas as pd

def compute_sector_score(sector_name, stocks, records):
    record_map = {r["dm"]: r for r in records}
    rows = []
    for s in stocks:
        r = record_map.get(s["dm"].split(".")[0])
        if r:
            rows.append({
                "代码": r["dm"], "名称": r["mc"],
                "现价": r["p"], "涨跌幅%": r["pc"],
                "振幅%": r["zf"], "成交额亿": r["cje"] / 1e8,
            })
    if not rows:
        return None
    df = pd.DataFrame(rows)
    return {
        "板块": sector_name,
        "成分股数": len(df),
        "上涨家数": int((df["涨跌幅%"] > 0).sum()),
        "下跌家数": int((df["涨跌幅%"] < 0).sum()),
        "平均涨跌幅%": df["涨跌幅%"].mean(),
        "最大涨幅%": df["涨跌幅%"].max(),
        "总成交额亿": df["成交额亿"].sum(),
        "领涨股": df.loc[df["涨跌幅%"].idxmax(), "名称"],
        "领涨股涨幅%": df["涨跌幅%"].max(),
    }

score = compute_sector_score("GN半导体", stocks, records)
print(score)

关键解释：

  • 用板块内成分股的平均涨跌幅代表板块整体强度，比单看几只龙头股更稳。
  • 同时统计上涨家数 / 下跌家数，判断板块是普涨还是个别龙头独秀。
  • 总成交额反映板块活跃度。

六、多板块轮动监控

把多个热门板块放一起比较，输出异动排行榜：

SECTOR_NAMES = ["GN半导体", "GN人工智能", "GN低空经济", "GN机器人", "GN新能源"]

scores = []
for name in SECTOR_NAMES:
    stocks = api.sector_detail(sector_name=name)["stocks"][:30]
    codes = [s["dm"].split(".")[0] for s in stocks]
    records = fetch_batch_realtime(codes)
    score = compute_sector_score(name, stocks, records)
    if score:
        scores.append(score)

score_df = pd.DataFrame(scores).sort_values("平均涨跌幅%", ascending=False)
print(score_df.to_string(index=False))

七、完整可运行示例

from __future__ import annotations
import os, sys, time


import pandas as pd
from mairui import Client

SECTOR_NAMES = ["GN半导体", "GN人工智能", "GN低空经济", "GN机器人", "GN新能源"]
TOP_N_STOCKS = 30
PC_ALERT = 1.0


def normalize_code(dm: str) -> str:
    return dm.split(".")[0]


def fetch_sector_stocks(api: Client, sector_name: str):
    data = api.sector_detail(sector_name=sector_name)
    return data.get("stocks", [])


def fetch_batch_realtime(codes: list[str]):
    api = Client(licence="LICENCE-66D8-9F96-0C7F0FBCD073")  # 演示证书，演示证书（读者复制即用，无需替换）
    records = []
    for i in range(0, len(codes), 20):
        batch = codes[i:i + 20]
        batch_records = api.stock_ssjy_more(stock_codes=",".join(batch))
        if isinstance(batch_records, list):
            records.extend(batch_records)
        time.sleep(0.05)
    return records


def compute_sector_score(sector_name, stocks, records):
    record_map = {r.get("dm"): r for r in records if isinstance(r, dict)}
    rows = []
    for s in stocks:
        r = record_map.get(normalize_code(s["dm"]))
        if r and r.get("p") is not None:
            rows.append({
                "代码": r["dm"], "名称": r.get("mc", s["mc"]),
                "现价": r["p"], "涨跌幅%": r["pc"],
                "振幅%": r.get("zf", 0),
                "成交额亿": r.get("cje", 0) / 1e8,
            })
    if not rows:
        return None
    df = pd.DataFrame(rows)
    return {
        "板块": sector_name,
        "成分股数": len(df),
        "上涨家数": int((df["涨跌幅%"] > 0).sum()),
        "下跌家数": int((df["涨跌幅%"] < 0).sum()),
        "平均涨跌幅%": df["涨跌幅%"].mean(),
        "最大涨幅%": df["涨跌幅%"].max(),
        "平均涨速%": df["涨速%"].mean() if "涨速%" in df.columns else 0,
        "总成交额亿": df["成交额亿"].sum(),
        "领涨股": df.loc[df["涨跌幅%"].idxmax(), "名称"],
        "领涨股涨幅%": df["涨跌幅%"].max(),
    }


if __name__ == "__main__":
    api = Client(licence="LICENCE-66D8-9F96-0C7F0FBCD073")
    scores = []
    for name in SECTOR_NAMES:
        stocks = fetch_sector_stocks(api, name)[:TOP_N_STOCKS]
        codes = [normalize_code(s["dm"]) for s in stocks]
        records = fetch_batch_realtime(codes)
        score = compute_sector_score(name, stocks, records)
        if score:
            scores.append(score)

    score_df = pd.DataFrame(scores).sort_values("平均涨跌幅%", ascending=False)
    print(score_df[["板块", "成分股数", "上涨家数", "下跌家数",
                    "平均涨跌幅%", "最大涨幅%", "总成交额亿", "领涨股", "领涨股涨幅%"]].to_string(index=False))

    print("\n【异动板块】")
    abnormal = score_df[score_df["平均涨跌幅%"].abs() >= PC_ALERT]
    if abnormal.empty:
        print("  当前样本板块平均涨跌幅均未触发阈值。")
    else:
        for _, row in abnormal.iterrows():
            direction = " " if row["平均涨跌幅%"] > 0 else " "
            print(
                f"{direction} {row['板块']} 平均 {row['平均涨跌幅%']:+.2f}% "
                f"({row['上涨家数']}/{row['成分股数']} 只上涨) "
                f"领涨: {row['领涨股']} {row['领涨股涨幅%']:+.2f}%"
            )

运行结果示例：

    板块  成分股数  上涨家数  下跌家数  平均涨跌幅%  最大涨幅%  总成交额亿   领涨股  领涨股涨幅%
 GN半导体    30    16    13     +0.60    +3.88  685.00  *ST皇庭    +3.88
GN低空经济    30    10    18     -0.17    +5.83  281.52   万泽股份    +5.83
 GN机器人    30     8    20     -0.44    +4.89  367.08   冰轮环境    +4.89
GN人工智能    30     8    20     -0.56    +2.73  217.00   特发信息    +2.73
 GN新能源    30     6    23     -0.59    +4.57   63.19   豫能控股    +4.57

八、常见坑

1. sector_detail 成分股代码带后缀，批量接口不要后缀

sector_detail 返回 000016.SZ，但 stock_ssjy_more 要 000016。用 .split(".")[0] 切掉。

2. stock_ssjy_more 没有 zs（涨速）字段

批量接口返回的字段比单股接口少。如果需要「5 分钟涨幅榜」，可以：

  • 方案 A：用本文的「当前涨跌幅」代替，盘中每隔 5 分钟跑一次，计算两次快照差。
  • 方案 B：对领涨股再用 stock_ssjy 单股接口补拉 zs 字段。

3. 板块成分股数量差异大

有的板块几百只，有的几十只。演示时取前 30 只活跃成分股即可；全量拉取调用次数多、慢，不适合盘中高频监控。

4. 非交易时间板块数据为空或异常

如果收盘后跑脚本，部分接口可能返回空列表或 404。建议盘中 09:30–15:00 运行。

九、进阶：做成 5 分钟涨幅榜

把上面的脚本包进一个循环，每 5 分钟跑一次，就能实现真正的「板块 5 分钟涨幅榜」：

import time

while True:
    run_once()  # 跑一次板块监控
    time.sleep(300)  # 等 5 分钟

如果想更省额度，可以把成分股列表缓存到本地，只拉行情，不重复调 sector_detail。

十、小结

本文把概念板块的成分股批量拉取、聚合计算，输出板块异动排行榜：

  1. 用 sector_detail 获取板块成分股。
  2. 用 stock_ssjy_more 批量拉取成分股行情。
  3. 计算板块平均涨跌幅、上涨家数、领涨股、成交额。
  4. 输出异动板块排行榜。

这个脚本适合盘中快速判断「今天资金在攻哪些方向」。配合 #13 的 ETF 监控，你可以同时看清「大盘→ETF→概念板块」三层结构。

下一篇预告：

#15 我们进入「盘后复盘」——用 Python 把一天的盯盘结果生成一页 PDF 报告，包含指数、ETF、异动板块和自选股汇总。

本文为技术演示，不构成投资建议。文中示例数据仅供接口用法参考，不代表实时行情。

# python # 开发语言 # 人工智能 # 机器学习 # java
Logo
DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

cover

成功率不到 70%，Kinova双臂做遮挡抓取究竟差在哪?

avatar DAMO开发者矩阵

鱼香ros2（一）在ubuntu中安装ros2

ros包含大量搭建机器人所需要的软件和工具，是目前应用最广发的机器人操作系统，未来搭载ros的机器人将不断出现在生产和生活中。

avatar DAMO开发者矩阵
cover

Franka 赋能模型应用新研究，Zero-WAM：基于人类视频上下文的世界-动作模型与零样本任务泛化

avatar DAMO开发者矩阵