别错过涨停！Python打造A股实时监控大屏

05811

一口深井

昨天发布

作者：派晓生

来自：量子云智侠

之前有粉丝在后台留言说： 👉 “能不能写一个 涨停监控工具？每天涨停板太多，想随时能看。”

于是我写了这款 A股涨停监控系统（网页版），今天完整开源分享给大家（运行效果图见文中）。

在开始之前，先提醒一下使用约束：

使用说明

这个工具虽然轻量，但也有一些使用条件，大家需要注意：

必须是交易日运行涨停数据接口只在交易日有效，节假日或周末运行会返回空数据。
需要联网工具的数据来源于东方财富（akshare 封装接口），运行时必须保持网络畅通。
获取板块 抓取并缓存板块/行业信息（首次运行或板块调整后使用）。
手动刷新 拉取当日最新涨停池并更新展示数据。按钮有“冷却时间”（限频保护已写在代码里），两个按钮都不能频繁点击，短时间内重复点击会被忽略或提示，避免被数据源限流。建议点击间隔 ≥ 60 秒（以你本地界面提示为准）。
网络与交易时段 非交易日/非交易时段接口可能返回空数据，这不是程序错误。
本地存储抓取到的涨停数据会写入到 limit_up.json 文件，作为网页展示的数据源。
展示模式前端使用 Streamlit 搭建，需要浏览器访问 http://localhost:8501 查看。

满足以上条件，就能顺利运行工具。

🛠️ 功能亮点

实时抓取 A股涨停数据（来源：东方财富）
本地缓存 JSON 文件，方便后续二次开发
网页端展示，随时刷新即可更新数据
轻量开箱即用，只需 Python + pip 安装依赖

⚙️ 环境准备

1.安装 Python 👉 Python 官网下载

2.安装依赖包在项目目录下运行：

pip install -r requirements.txt

依赖清单（requirements.txt）：

pandas
akshare
streamlit
plotly

🚀 如何运行

命令行进入项目目录，执行：

streamlit run streamlit_app.py

然后在浏览器访问： 👉 http://localhost:8501

就能看到实时的 A股涨停监控大屏啦！ 🎉

📜 核心代码解析

1. 数据抓取与缓存（涨停监控.py）

import akshare as ak
import pandas as pd
import time
import json
from datetime import datetime

def get_limit_up_pool():
    try:
        df = ak.stock_zt_pool_em(date=datetime.now().strftime("%Y%m%d"))
        return df
    except Exception as e:
        print(f"获取涨停数据失败: {e}")
        return pd.DataFrame()

def run_monitor():
    whileTrue:
        df = get_limit_up_pool()
        ifnot df.empty:
            data = df.to_dict(orient="records")
            with open("limit_up.json", "w", encoding="utf-8") as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
            print(f"{datetime.now()} 已更新 {len(df)} 条涨停数据")
        else:
            print(f"{datetime.now()} 无数据")
        time.sleep(60)

if __name__ == "__main__":
    run_monitor()

功能解析：

ak.stock_zt_pool_em() → 东方财富涨停数据接口
to_dict + json.dump → 将数据保存为 JSON 文件
while True + time.sleep(60) → 循环，每 60 秒更新一次
容错机制 → 请求失败时返回空 DataFrame，避免程序崩溃

2. 网页端展示（streamlit_app.py）

import streamlit as st
import pandas as pd
import json
from datetime import datetime

st.set_page_config(page_title="A股涨停监控系统", layout="wide")

def load_data():
    try:
        with open("limit_up.json", "r", encoding="utf-8") as f:
            return pd.DataFrame(json.load(f))
    except:
        return pd.DataFrame()

st.title("⚡ A股涨停监控系统")
st.write("数据来源：东方财富 · 每分钟自动更新")

if st.button("手动刷新"):
    st.experimental_rerun()

df = load_data()
ifnot df.empty:
    st.dataframe(df, use_container_width=True)
else:
    st.warning("暂无涨停数据，请等待更新...")

功能解析：