Python实现A股实时行情看板：Streamlit + AkShare + 新浪备用接口

说明

本文项目仅用于 Python 编程学习、数据接口调用、Pandas 数据处理和 Streamlit 可视化练习，不构成任何投资建议，也不能作为股票买卖依据。

项目中的实时行情、涨跌幅、换手率、量比和“信号参考”等内容，均来自公开数据接口或程序规则计算。数据可能存在延迟、缺失、错误或接口暂时不可用等情况。股市有风险，投资需谨慎。

一、项目介绍

这次分享的是一个使用 Python 和 Streamlit 开发的 A 股实时行情看板。

用户可以在侧边栏输入多个股票代码，例如：

000001,600519,300750

程序会获取这些股票的实时行情，并展示股票名称、最新价格、涨跌幅、涨跌额、成交量、成交额、最高价、最低价、今开、昨收、换手率、量比、市盈率、市净率等数据。

项目还支持自动刷新。用户可以设置 15 秒到 120 秒的刷新间隔，打开自动刷新后，页面会按照设定的时间重新获取行情。

为了提高稳定性，程序默认使用 AkShare 提供的东方财富实时行情接口。当东方财富接口请求失败时，会自动切换到新浪实时行情接口。

二、项目主要功能

1. 支持输入多个自选股代码

多个代码使用英文逗号分隔，例如：

000001,600519,300750

程序会自动去除空格和小数点，并兼容中文逗号。

2. 获取 A 股实时行情

程序首先调用 AkShare 的 stock_zh_a_spot_em 方法获取东方财富 A 股实时行情。

为了减少偶发网络错误，程序最多尝试请求三次，每次失败后等待 1.5 秒。

3. 新浪实时行情备用接口

当东方财富接口连续失败时，程序会自动请求新浪行情接口。

上海股票代码会转换成 sh 开头的格式，例如：

600519 转换为 sh600519

深圳股票代码会转换成 sz 开头的格式，例如：

000001 转换为 sz000001

4. 自定义自动刷新时间

用户可以通过滑块设置刷新间隔，范围为 15 秒到 120 秒，默认值为 30 秒。

开启自动刷新后，程序使用 streamlit-autorefresh 定时刷新页面。

5. 展示单股实时指标

程序会为每只自选股展示股票名称和代码、最新价格、涨跌幅、成交量、成交额以及简单信号参考。

6. 展示实时行情表格

实时行情表主要包括代码、名称、最新价、涨跌幅、涨跌额、成交量、成交额、振幅、最高价、最低价、今开、昨收、量比、换手率、动态市盈率、市净率和信号参考。

7. 涨跌幅横向对比图

项目使用 Plotly 绘制自选股涨跌幅横向柱状图，并按照涨跌幅从低到高排序。

三、项目使用的技术

Streamlit：构建网页界面，包括侧边栏、输入框、滑块、自动刷新开关、指标卡片、表格和图表。

AkShare：获取东方财富 A 股实时行情数据。

Requests：直接请求新浪实时行情接口，作为备用数据源。

Pandas：处理实时行情数据、筛选自选股、添加信号列和整理展示字段。

Plotly Express：绘制股票涨跌幅横向柱状图。

streamlit-autorefresh：控制 Streamlit 页面自动刷新。

四、项目文件结构

项目目录可以写成：

a_stock_realtime_dashboard

realtime_app.py

requirements.txt

其中 realtime_app.py 是项目主程序，requirements.txt 用于保存依赖库。

五、安装项目依赖

建议先创建虚拟环境：

python -m venv venv

Windows PowerShell 激活虚拟环境：

.\venv\Scripts\Activate.ps1

Windows CMD 激活虚拟环境：

venv\Scripts\activate

安装依赖：

pip install streamlit akshare pandas plotly requests streamlit-autorefresh

也可以使用 requirements.txt：

pip install -r requirements.txt

六、运行项目

在 realtime_app.py 所在目录执行：

streamlit run realtime_app.py

如果提示 streamlit 不是内部或外部命令，可以使用：

python -m streamlit run realtime_app.py

运行成功后，浏览器通常会打开：

http://localhost:8501

七、核心代码说明

1. 设置不经过代理的行情域名

程序将东方财富相关域名加入 NO_PROXY 和 no_proxy 环境变量，减少代理或 VPN 对行情接口的影响。

2. 处理用户输入的股票代码

normalize_codes 函数会把中文逗号替换成英文逗号，然后分割字符串，并去除空格和小数点。

3. 获取东方财富实时行情

load_realtime_quotes 函数使用 AkShare 获取实时行情，并设置 8 秒缓存。当接口失败时，程序会记录异常并重试三次。

4. 构造新浪股票代码

以 6 或 9 开头的代码使用 sh 前缀，其他代码使用 sz 前缀。这是适合学习演示的简化规则。

5. 解析新浪实时行情

程序通过 requests.get 请求新浪行情接口，并设置 Referer 请求头。接口返回 GBK 编码数据，因此需要设置 response.encoding = "gbk"。

程序从返回字段中解析股票名称、开盘价、昨收价、最新价、最高价、最低价、成交量和成交额，并计算涨跌额和涨跌幅。

新浪备用接口中未完整处理量比、换手率、市盈率和市净率，因此这些字段会使用 0 或空值占位。

6. 筛选自选股

filter_watchlist 函数根据用户输入的股票代码筛选实时行情。如果用户没有输入代码，程序默认展示行情数据中的前 30 只股票。

7. 生成简单信号参考

项目根据涨跌幅、换手率和量比进行粗略分类。

涨跌幅大于等于 5%，并且量比大于等于 2 时，显示“短线过热”。

涨跌幅大于等于 2%，并且换手率大于等于 3% 时，显示“活跃偏强”。

涨跌幅小于等于负 5% 时，显示“风险释放”。

涨跌幅绝对值小于 1% 时，显示“波动较小”。

其他情况显示“正常波动”。

这些标签仅用于演示条件判断，不属于专业交易策略，不能作为投资依据。

8. 设置自动刷新

用户可以在侧边栏控制是否自动刷新，并设置刷新间隔。程序把秒数乘以 1000 转换成毫秒后传给自动刷新组件。

9. 显示实时行情指标

程序使用 st.columns 创建多列布局，每行最多显示三只股票。

10. 绘制涨跌幅对比图

程序先按照涨跌幅升序排列数据，然后使用 Plotly Express 绘制横向柱状图。

八、常见问题

1. 页面提示实时行情获取失败

可以检查网络、代理或 VPN、AkShare 版本、接口是否暂时不可用以及股票代码格式是否正确。

更新 AkShare：

pip install -U akshare

2. 提示缺少 streamlit_autorefresh

执行：

pip install streamlit-autorefresh

3. 输入代码后没有数据显示

确认多个代码使用英文逗号隔开，例如：

000001,600519,300750

4. 新浪数据模式下部分字段为 0 或空值

这是因为备用接口主要解析价格、成交量和成交额等字段，其他部分字段没有完整处理。

5. 为什么最短刷新时间设置为 15 秒

第三方公开接口不适合过于频繁地请求。刷新过快可能增加请求失败概率，也可能给接口造成不必要的压力。

6. 为什么不是逐笔成交数据

本项目展示的是公开接口返回的实时或准实时行情，不是交易所级别的逐笔成交数据，不同接口可能存在延迟。

九、可以继续优化的方向

可以增加股票名称搜索、自选股本地保存、历史分时图、指数行情、数据导出、异常日志、涨跌颜色设置，以及更完整的市场代码识别。

十、项目完整代码

import os import time import akshare as ak import pandas as pd import plotly.express as px import requests import streamlit as st from streamlit_autorefresh import st_autorefresh def bypass_proxy_for_market_data() -> None: hosts = [ "eastmoney.com", "push2his.eastmoney.com", "push2.eastmoney.com", "quote.eastmoney.com", ] existing = os.environ.get("NO_PROXY") or os.environ.get("no_proxy") or "" current_hosts = [item.strip() for item in existing.split(",") if item.strip()] for host in hosts: if host not in current_hosts: current_hosts.append(host) os.environ["NO_PROXY"] = ",".join(current_hosts) os.environ["no_proxy"] = ",".join(current_hosts) bypass_proxy_for_market_data() st.set_page_config(page_title="A股实时行情看板", layout="wide") def normalize_codes(raw_codes: str) -> list[str]: codes = [] for item in raw_codes.replace("，", ",").split(","): code = item.strip().replace(" ", "").replace(".", "") if code: codes.append(code) return codes @st.cache_data(ttl=8) def load_realtime_quotes() -> pd.DataFrame: last_error = None for _ in range(3): try: return ak.stock_zh_a_spot_em() except Exception as error: last_error = error time.sleep(1.5) raise RuntimeError(f"实时行情接口连续失败：{last_error}") def sina_symbol(code: str) -> str: if code.startswith(("6", "9")): return f"sh{code}" return f"sz{code}" @st.cache_data(ttl=15) def load_sina_realtime_quotes(codes: list[str]) -> pd.DataFrame: rows = [] symbols = ",".join(sina_symbol(code) for code in codes) if not symbols: return pd.DataFrame() response = requests.get( f"https://hq.sinajs.cn/list={symbols}", headers={"Referer": "https://finance.sina.com.cn/"}, timeout=10, ) response.encoding = "gbk" response.raise_for_status() for line in response.text.splitlines(): if '="' not in line: continue raw_symbol = line.split("var hq_str_", 1)[1].split("=", 1)[0] code = raw_symbol[2:] payload = line.split('="', 1)[1].rstrip('";') fields = payload.split(",") if len(fields) < 32 or not fields[0]: continue name = fields[0] open_price = float(fields[1] or 0) prev_close = float(fields[2] or 0) latest_price = float(fields[3] or 0) high = float(fields[4] or 0) low = float(fields[5] or 0) volume = float(fields[8] or 0) amount = float(fields[9] or 0) change = latest_price - prev_close if prev_close else 0 pct_change = change / prev_close * 100 if prev_close else 0 rows.append( { "代码": code, "名称": name, "最新价": latest_price, "涨跌幅": pct_change, "涨跌额": change, "成交量": volume, "成交额": amount, "振幅": (high - low) / prev_close * 100 if prev_close else 0, "最高": high, "最低": low, "今开": open_price, "昨收": prev_close, "量比": 0, "换手率": 0, "市盈率-动态": None, "市净率": None, } ) return pd.DataFrame(rows) def filter_watchlist(data: pd.DataFrame, codes: list[str]) -> pd.DataFrame: if not codes: return data.head(30) return data[data["代码"].astype(str).isin(codes)].copy() def make_realtime_signal(row: pd.Series) -> str: pct_change = row.get("涨跌幅", 0) turnover = row.get("换手率", 0) volume_ratio = row.get("量比", 0) if pct_change >= 5 and volume_ratio >= 2: return "短线过热" if pct_change >= 2 and turnover >= 3: return "活跃偏强" if pct_change <= -5: return "风险释放" if abs(pct_change) < 1: return "波动较小" return "正常波动" st.title("A股实时行情看板") st.caption("数据来自 AkShare 东方财富实时行情接口，适合学习和观察，不构成投资建议。") with st.sidebar: st.header("实时设置") raw_codes = st.text_area("自选股代码", value="000001,600519,300750", help="多个代码用英文逗号隔开") refresh_seconds = st.slider("刷新间隔（秒）", min_value=15, max_value=120, value=30, step=5) auto_refresh = st.toggle("自动刷新", value=True) if auto_refresh: st_autorefresh(interval=refresh_seconds * 1000, key="realtime_refresh") codes = normalize_codes(raw_codes) try: try: realtime = load_realtime_quotes() watchlist = filter_watchlist(realtime, codes) data_mode = "东方财富实时行情" except Exception as realtime_error: st.warning(f"东方财富实时行情暂时获取失败，正在尝试新浪实时行情：{realtime_error}") watchlist = load_sina_realtime_quotes(codes) data_mode = "新浪实时行情" if watchlist.empty: st.warning("没有获取到真实行情数据。请检查网络、代理/VPN，或稍后再试。") st.stop() watchlist = watchlist.copy() watchlist["信号参考"] = watchlist.apply(make_realtime_signal, axis=1) display_columns = [ "代码", "名称", "最新价", "涨跌幅", "涨跌额", "成交量", "成交额", "振幅", "最高", "最低", "今开", "昨收", "量比", "换手率", "市盈率-动态", "市净率", "信号参考", ] display_columns = [column for column in display_columns if column in watchlist.columns] latest_time = pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S") st.info(f"数据模式：{data_mode}。最近刷新时间：{latest_time}，当前展示 {len(watchlist)} 只股票。") st.subheader("单股实时指标") metric_columns = st.columns(min(len(watchlist), 3)) for index, (_, stock) in enumerate(watchlist.iterrows()): column = metric_columns[index % len(metric_columns)] with column: st.markdown(f"**{stock['名称']}（{stock['代码']}）**") st.metric("最新价", f"{stock['最新价']:.2f}", f"{stock['涨跌幅']:.2f}%") st.write(f"信号参考：{stock['信号参考']}") st.caption(f"成交量：{stock['成交量']:.0f} | 成交额：{stock['成交额']:.0f}") st.subheader("实时行情表") st.dataframe(watchlist[display_columns], use_container_width=True, height=360) st.subheader("涨跌幅对比") chart_data = watchlist.sort_values("涨跌幅", ascending=True) fig = px.bar( chart_data, x="涨跌幅", y="名称", color="涨跌幅", color_continuous_scale=["#16a34a", "#e5e7eb", "#dc2626"], orientation="h", height=420, ) fig.update_layout(margin=dict(l=20, r=20, t=20, b=20)) st.plotly_chart(fig, use_container_width=True) st.subheader("信号说明") st.write( "这个实时信号只根据涨跌幅、换手率、量比做粗略分类。" "它适合提醒你继续观察，不适合作为买卖依据。" ) except Exception as error: st.error(f"实时行情获取失败：{error}")

十一、项目总结

通过这个项目，可以练习 Python 调用公开行情接口、AkShare 实时数据获取、Requests 请求解析、Pandas 表格处理、Streamlit 网页开发、页面自动刷新、Plotly 图表绘制、主备数据源切换和网络异常处理。

这个项目代码量不大，适合作为 Python 数据分析、Streamlit 和金融数据接口的入门练习。

十二、免责声明

本文和项目代码仅供 Python 编程学习、课程实践和技术交流使用。