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 转换为 sz0000014. 自定义自动刷新时间用户可以通过滑块设置刷新间隔范围为 15 秒到 120 秒默认值为 30 秒。开启自动刷新后程序使用 streamlit-autorefresh 定时刷新页面。5. 展示单股实时指标程序会为每只自选股展示股票名称和代码、最新价格、涨跌幅、成交量、成交额以及简单信号参考。6. 展示实时行情表格实时行情表主要包括代码、名称、最新价、涨跌幅、涨跌额、成交量、成交额、振幅、最高价、最低价、今开、昨收、量比、换手率、动态市盈率、市净率和信号参考。7. 涨跌幅横向对比图项目使用 Plotly 绘制自选股涨跌幅横向柱状图并按照涨跌幅从低到高排序。三、项目使用的技术Streamlit构建网页界面包括侧边栏、输入框、滑块、自动刷新开关、指标卡片、表格和图表。AkShare获取东方财富 A 股实时行情数据。Requests直接请求新浪实时行情接口作为备用数据源。Pandas处理实时行情数据、筛选自选股、添加信号列和整理展示字段。Plotly Express绘制股票涨跌幅横向柱状图。streamlit-autorefresh控制 Streamlit 页面自动刷新。四、项目文件结构项目目录可以写成a_stock_realtime_dashboardrealtime_app.pyrequirements.txt其中 realtime_app.py 是项目主程序requirements.txt 用于保存依赖库。五、安装项目依赖建议先创建虚拟环境python -m venv venvWindows PowerShell 激活虚拟环境.\venv\Scripts\Activate.ps1Windows CMD 激活虚拟环境venv\Scripts\activate安装依赖pip install streamlit akshare pandas plotly requests streamlit-autorefresh也可以使用 requirements.txtpip 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 版本、接口是否暂时不可用以及股票代码格式是否正确。更新 AkSharepip install -U akshare2. 提示缺少 streamlit_autorefresh执行pip install streamlit-autorefresh3. 输入代码后没有数据显示确认多个代码使用英文逗号隔开例如000001,600519,3007504. 新浪数据模式下部分字段为 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_titleA股实时行情看板, layoutwide) 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(ttl8) 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 fsh{code} return fsz{code} st.cache_data(ttl15) 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( fhttps://hq.sinajs.cn/list{symbols}, headers{Referer: https://finance.sina.com.cn/}, timeout10, ) 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(自选股代码, value000001,600519,300750, help多个代码用英文逗号隔开) refresh_seconds st.slider(刷新间隔秒, min_value15, max_value120, value30, step5) auto_refresh st.toggle(自动刷新, valueTrue) if auto_refresh: st_autorefresh(intervalrefresh_seconds * 1000, keyrealtime_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, axis1) 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_widthTrue, height360) st.subheader(涨跌幅对比) chart_data watchlist.sort_values(涨跌幅, ascendingTrue) fig px.bar( chart_data, x涨跌幅, y名称, color涨跌幅, color_continuous_scale[#16a34a, #e5e7eb, #dc2626], orientationh, height420, ) fig.update_layout(margindict(l20, r20, t20, b20)) st.plotly_chart(fig, use_container_widthTrue) st.subheader(信号说明) st.write( 这个实时信号只根据涨跌幅、换手率、量比做粗略分类。 它适合提醒你继续观察不适合作为买卖依据。 ) except Exception as error: st.error(f实时行情获取失败{error})十一、项目总结通过这个项目可以练习 Python 调用公开行情接口、AkShare 实时数据获取、Requests 请求解析、Pandas 表格处理、Streamlit 网页开发、页面自动刷新、Plotly 图表绘制、主备数据源切换和网络异常处理。这个项目代码量不大适合作为 Python 数据分析、Streamlit 和金融数据接口的入门练习。十二、免责声明本文和项目代码仅供 Python 编程学习、课程实践和技术交流使用。