免费领取证书

← 返回博客列表

Python实时盯盘与预警

【Python实时盯盘与预警 #13】ETF溢价率异动不会看Python实时监控ETF与跟踪指数偏离

2026年10月15日 08:19 · 麦蕊智数 MyData

上一篇我们监控了宽基指数，但指数只能看方向，不能直接交易。真正在二级市场买卖的是 ETF：沪深 300ETF、创业板 ETF、科创 50ETF……

ETF 的盘中价格会围绕跟踪指数波动。如果 ETF 涨得比指数快，说明资金在抢筹；如果 ETF 明显弱于指数，可能是溢价收窄或资金流出。这个「偏离度」就是盘中很重要的信号。

这篇文章用 Python 实时监控 ETF 与跟踪指数的涨跌幅偏离，帮你识别 ETF 异动。

一、本文你将得到什么

  1. 用麦蕊 fund_real_time 接口拉取 ETF 实时行情。
  2. 用 index_real_time 拉取对应跟踪指数行情。
  3. 计算「ETF - 指数」涨跌幅偏离度。
  4. 设定阈值，自动打印异动 ETF。
  5. 一个可扩展的 ETF 监控脚本，替换代码就能加更多 ETF。

想先在线试试接口效果？打开 API Playground 即可直接调用测试： https://mairuiapi.com/playground

二、环境准备

本文代码使用 mairui SDK 获取 ETF 与指数数据，安装方法如下：

pip install mairui

SDK 的完整接口文档与使用说明请查阅官网：https://www.mairuiapi.com/sdk

ETF 与指数接口的详细参数说明请查阅官网 API 文档：

  • 基金实时行情：https://mairuiapi.com/hsdata
  • 指数实时行情：https://mairuiapi.com/hszsdata

运行环境：

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

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

证书直接写在代码里（文中使用演示证书，请替换成你自己的正式证书）。

import sys, os

三、ETF 与对应指数

下面选了 8 只代表性 ETF，其中 6 只宽基 ETF 有明确跟踪指数，另外 2 只行业 ETF 作为扩展示例：

ETF 代码 ETF 名称 跟踪指数代码 跟踪指数
510050 上证50ETF 000016.SH 上证50
510300 沪深300ETF 000300.SH 沪深300
510500 中证500ETF 000905.SH 中证500
512100 中证1000ETF 000852.SH 中证1000
159915 创业板ETF 399006.SZ 创业板指
588000 科创50ETF 000688.SH 科创50
512690 酒ETF — —
515030 新能源车ETF — —

fund_real_time(code="510050") 返回 ETF 实时行情：p（现价）、pc（涨跌幅%）、zf（振幅%）、cje（成交额元）等。

index_real_time(code="000016.SH") 返回对应指数行情，字段与 ETF 一致。

四、拉取 ETF 实时行情

from mairui import Client
import pandas as pd
import time

ETF_WATCHLIST = [
    {"etf_code": "510050", "etf_name": "上证50ETF", "idx_code": "000016.SH", "idx_name": "上证50"},
    {"etf_code": "510300", "etf_name": "沪深300ETF", "idx_code": "000300.SH", "idx_name": "沪深300"},
    {"etf_code": "510500", "etf_name": "中证500ETF", "idx_code": "000905.SH", "idx_name": "中证500"},
    {"etf_code": "512100", "etf_name": "中证1000ETF", "idx_code": "000852.SH", "idx_name": "中证1000"},
    {"etf_code": "159915", "etf_name": "创业板ETF", "idx_code": "399006.SZ", "idx_name": "创业板指"},
    {"etf_code": "588000", "etf_name": "科创50ETF", "idx_code": "000688.SH", "idx_name": "科创50"},
]

def fetch_etf_snapshot():
    api = Client(licence="<你的证书>")  # 演示证书，请替换成你自己的正式证书
    rows = []
    for item in ETF_WATCHLIST:
        r = api.fund_real_time(code=item["etf_code"])
        rows.append({
            "ETF代码": item["etf_code"],
            "ETF名称": item["etf_name"],
            "现价": r["p"],
            "涨跌幅%": r["pc"],
            "振幅%": r["zf"],
            "成交额亿": r["cje"] / 1e8,
            "跟踪指数代码": item["idx_code"],
        })
        time.sleep(0.05)
    return pd.DataFrame(rows)

etf_df = fetch_etf_snapshot()
print(etf_df.to_string(index=False))

关键解释：

  • fund_real_time 的 code 参数接受纯数字 ETF 代码，不需要加 .SH 后缀。
  • 8 只 ETF 循环调用，约 0.5~1.0 秒完成。
  • cje 是元为单位，除以 1e8 转成「亿」更直观。

五、拉取跟踪指数并计算偏离度

ETF 的涨跌幅应该和跟踪指数接近。如果 ETF 涨跌幅明显偏离指数，说明盘中资金情绪有变化：

def fetch_index_snapshot(idx_codes):
    api = Client(licence="<你的证书>")  # 演示证书，请替换成你自己的正式证书
    result = {}
    for code in set(idx_codes):
        r = api.index_real_time(code=code)
        result[code] = {"指数涨跌幅%": r["pc"]}
        time.sleep(0.05)
    return result

idx_map = fetch_index_snapshot(etf_df["跟踪指数代码"].tolist())

# 合并并计算偏离度
merged = etf_df.copy()
merged["指数涨跌幅%"] = merged["跟踪指数代码"].map(lambda c: idx_map.get(c, {}).get("指数涨跌幅%"))
merged["ETF-指数偏离%"] = merged["涨跌幅%"] - merged["指数涨跌幅%"]
print(merged[["ETF名称", "涨跌幅%", "指数涨跌幅%", "ETF-指数偏离%"]].to_string(index=False))

关键解释：

  • 偏离度 = ETF 涨跌幅 - 跟踪指数涨跌幅。
  • 偏离为正：ETF 强于指数，可能资金抢筹或溢价扩大。
  • 偏离为负：ETF 弱于指数，可能资金流出或溢价收窄。
  • 这本质上是折溢价/资金情绪的代理指标；更精确的溢价率需要 ETF 净值，可在收盘后用净值数据再校准。

六、异动提醒

设定阈值，当 ETF 涨跌幅绝对值超过 1%，或与指数偏离超过 0.5% 时打印提醒：

PC_ALERT = 1.0       # ETF 自身涨跌幅异动阈值
DEVIATION_ALERT = 0.5  # 与指数偏离阈值

abnormal = merged[
    (merged["涨跌幅%"].abs() >= PC_ALERT) |
    (merged["ETF-指数偏离%"].abs() >= DEVIATION_ALERT)
]

for _, row in abnormal.iterrows():
    direction = " " if row["涨跌幅%"] > 0 else " "
    dev_dir = "强于" if row["ETF-指数偏离%"] > 0 else "弱于"
    print(
        f"{direction} {row['ETF名称']}({row['ETF代码']}) "
        f"涨跌幅 {row['涨跌幅%']:+.2f}% "
        f"相对指数{dev_dir}{abs(row['ETF-指数偏离%']):.2f}%"
    )

阈值可按交易风格调整：

  • 日内短线：偏离 0.3%
  • 波段观察：偏离 0.5%~1%
  • 套利盯盘：配合净值计算真实溢价率

七、完整可运行示例

from __future__ import annotations
import os, sys, time


import pandas as pd
from mairui import Client

ETF_WATCHLIST = [
    {"etf_code": "510050", "etf_name": "上证50ETF", "idx_code": "000016.SH", "idx_name": "上证50"},
    {"etf_code": "510300", "etf_name": "沪深300ETF", "idx_code": "000300.SH", "idx_name": "沪深300"},
    {"etf_code": "510500", "etf_name": "中证500ETF", "idx_code": "000905.SH", "idx_name": "中证500"},
    {"etf_code": "512100", "etf_name": "中证1000ETF", "idx_code": "000852.SH", "idx_name": "中证1000"},
    {"etf_code": "159915", "etf_name": "创业板ETF", "idx_code": "399006.SZ", "idx_name": "创业板指"},
    {"etf_code": "588000", "etf_name": "科创50ETF", "idx_code": "000688.SH", "idx_name": "科创50"},
    {"etf_code": "512690", "etf_name": "酒ETF", "idx_code": "", "idx_name": "—"},
    {"etf_code": "515030", "etf_name": "新能源车ETF", "idx_code": "", "idx_name": "—"},
]

PC_ALERT = 1.0
DEVIATION_ALERT = 0.5


def fetch_etf_snapshot():
    api = Client(licence="<你的证书>")  # 演示证书，请替换成你自己的正式证书
    rows = []
    for item in ETF_WATCHLIST:
        r = api.fund_real_time(code=item["etf_code"])
        rows.append({
            "ETF代码": item["etf_code"], "ETF名称": item["etf_name"],
            "现价": r["p"], "涨跌幅%": r["pc"], "振幅%": r["zf"],
            "成交额亿": r["cje"] / 1e8, "跟踪指数代码": item["idx_code"],
        })
        time.sleep(0.05)
    return pd.DataFrame(rows)


def fetch_index_snapshot(idx_codes):
    api = Client(licence="<你的证书>")
    result = {}
    for code in {c for c in idx_codes if c}:
        r = api.index_real_time(code=code)
        result[code] = {"指数涨跌幅%": r["pc"]}
        time.sleep(0.05)
    return result


if __name__ == "__main__":
    etf_df = fetch_etf_snapshot()
    idx_map = fetch_index_snapshot(etf_df["跟踪指数代码"].tolist())

    merged = etf_df.copy()
    merged["指数涨跌幅%"] = merged["跟踪指数代码"].map(
        lambda c: idx_map.get(c, {}).get("指数涨跌幅%")
    )
    merged["ETF-指数偏离%"] = merged["涨跌幅%"] - merged["指数涨跌幅%"]

    print(merged[["ETF名称", "现价", "涨跌幅%", "指数涨跌幅%", "ETF-指数偏离%"]].to_string(index=False))

    print("\n【异动提醒】")
    abnormal = merged[
        (merged["涨跌幅%"].abs() >= PC_ALERT) |
        (merged["ETF-指数偏离%"].abs() >= DEVIATION_ALERT)
    ]
    for _, row in abnormal.iterrows():
        direction = " " if row["涨跌幅%"] > 0 else " "
        if pd.notna(row["ETF-指数偏离%"]):
            dev_dir = "强于" if row["ETF-指数偏离%"] > 0 else "弱于"
            print(f"{direction} {row['ETF名称']} 涨跌幅 {row['涨跌幅%']:+.2f}% "
                  f"相对指数{dev_dir}{abs(row['ETF-指数偏离%']):.2f}%")
        else:
            print(f"{direction} {row['ETF名称']} 涨跌幅 {row['涨跌幅%']:+.2f}%")

运行结果示例：

 ETF名称   现价  涨跌幅% 指数涨跌幅% ETF-指数偏离%
上证50ETF 3.07  +1.22   +1.38     -0.16
沪深300ETF 4.75  +0.89   +0.93     -0.04
中证500ETF 7.99  +1.98   +1.93     +0.05
中证1000ETF 3.11  +2.03   +1.98     +0.05
创业板ETF 3.59  +1.30   +1.35     -0.05
科创50ETF 1.84  +2.56   +2.51     +0.05

【异动提醒】
  上证50ETF 涨跌幅 +1.22% 相对指数弱于0.16%
  中证500ETF 涨跌幅 +1.98% 相对指数强于0.05%

八、常见坑

1. fund_real_time 参数是 code，不是 symbol

正确：api.fund_real_time(code="510050") 错误：api.fund_real_time(symbol="510050") 会报 unexpected keyword argument。

2. ETF 代码不需要交易所后缀

fund_real_time 接受 "510050"，不需要 "510050.SH"。

3. 指数代码必须带交易所后缀

index_real_time(code="000016.SH") 必须带 .SH 或 .SZ，否则返回 404。

4. 行业 ETF 没有对应指数时，偏离度为空

像酒 ETF、新能源车 ETF，若找不到一一对应的单一指数，可以只监控其自身涨跌幅，或手动指定对应行业指数。

5. 盘中「偏离度」不等于「溢价率」

溢价率需要 ETF 净值：(价格 - 净值) / 净值。本文用「涨跌幅偏离」作为资金异动的代理指标，更适合盘中快速扫描；收盘后再用净值校准溢价率。

九、小结

本文把 ETF 实时行情与跟踪指数放在一起比较，计算偏离度，识别盘中异动：

  1. 用 fund_real_time 拉 ETF 实时行情。
  2. 用 index_real_time 拉对应指数行情。
  3. 计算 ETF 涨跌幅与指数涨跌幅的偏离。
  4. 设定阈值，输出异动 ETF。

这个脚本适合挂在盘中关键时点跑一下：开盘看方向、午休看异动、尾盘决定是否过夜。

下一篇预告：

14 我们把视角从单只 ETF 提升到「概念板块」——用 Python 批量拉取板块成分股，计算板块平均涨跌幅，输出盘中概念板块异动排行榜。

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

延伸阅读：

  • 在线体验更多接口：https://mairuiapi.com/playground
  • 查看完整 API 文档：https://mairuiapi.com/hsdata
  • SDK 文档与源码：https://www.mairuiapi.com/sdk
标签： Python量化

相关阅读

QQ 客服 3826425416 咨询时请提供证书号或订单号，便于快速处理
咨询