说明本文和项目代码仅用于Python编程、数据分析和可视化学习不构成任何投资建议。程序中出现的“观察偏强”“中性观望”“风险偏高”等结果都是根据固定规则计算得到的学习示例不能作为股票买卖依据。股市有风险投资需谨慎。一、项目简介最近在学习Python数据处理和Streamlit网页开发于是做了一个A股历史行情查询与可视化的小项目。这个项目不需要单独写前端页面。运行app.py之后Streamlit会自动生成一个网页。用户只需要输入股票代码、开始日期和结束日期就可以查询对应股票的历史日线数据。网页中会显示股票名称、最新收盘价、当日涨跌幅、K线图、MA5、MA20、MA60均线、成交量、RSI14、20日年化波动率以及程序根据几个技术指标生成的简单参考说明。项目默认先通过AkShare获取东方财富历史行情。如果这个接口临时请求失败程序会自动切换到BaoStock继续获取数据这样可以减少因为单一接口不可用而导致整个页面无法运行的情况。二、项目用到的工具本项目主要使用了Streamlit、AkShare、BaoStock、Pandas和Plotly。Streamlit负责生成网页界面AkShare负责获取A股行情和股票名称BaoStock作为备用数据源Pandas负责数据清洗和指标计算Plotly负责绘制K线、均线、成交量和RSI图表。项目运行环境为Python 3核心文件是app.py。三、安装项目依赖先在终端中进入项目文件夹然后执行python -m venv venvWindows PowerShell中激活虚拟环境.\venv\Scripts\Activate.ps1Windows CMD中激活虚拟环境venv\Scripts\activate安装依赖pip install streamlit akshare baostock pandas plotly四、运行项目在app.py所在目录执行streamlit run app.py如果提示找不到streamlit命令可以改为python -m streamlit run app.py程序启动后终端一般会显示下面的本地地址http://localhost:8501在浏览器中打开这个地址就可以进入查询页面。五、股票代码处理为了减少输入格式带来的问题程序会先对股票代码进行简单清理。代码如下def normalize_stock_code(raw_code: str) - str: return raw_code.strip().replace( , ).replace(., )这个函数会删除股票代码前后的空格同时删除中间的空格和小数点。例如输入600519、600519.或者600 519程序都会尽可能处理成标准代码。六、获取股票历史行情程序首先通过AkShare中的stock_zh_a_hist接口获取A股日线数据。主要代码如下st.cache_data(ttl60 * 30) def load_stock_history(symbol: str, start_date: str, end_date: str) - pd.DataFrame: data ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustqfq, )symbol表示股票代码period设置为daily表示查询日线数据start_date和end_date分别表示开始日期和结束日期adjust设置为qfq表示使用前复权数据。这里使用了Streamlit的数据缓存功能缓存时间设置为30分钟。这样在短时间内重复查询相同内容时不需要每次都重新请求接口页面响应速度会更快。七、统一数据字段AkShare返回的数据字段是中文名称为了方便后面的计算程序把字段统一改成了英文。data data.rename( columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 涨跌幅: pct_change, 换手率: turnover, } )之后再把date转换成Pandas日期格式并按照日期从早到晚排序。data[date] pd.to_datetime(data[date]) data data.sort_values(date).reset_index(dropTrue)八、加入BaoStock备用数据源网络环境、代理设置或者接口维护都可能造成东方财富行情暂时无法获取。因此程序中加入了BaoStock备用方案。try: data load_stock_history( st.session_state[last_symbol], start_text, end_text ) data_mode 东方财富历史行情 except Exception as history_error: st.warning(f东方财富历史行情暂时获取失败正在尝试BaoStock历史行情{history_error}) data load_baostock_history( st.session_state[last_symbol], start_text, end_text ) data_mode BaoStock历史行情程序会先请求AkShare。如果出现异常就在页面中给出提示然后自动调用BaoStock。BaoStock使用的股票代码格式和普通六位代码不同。上海股票通常需要写成sh.600519深圳股票通常需要写成sz.000001所以程序增加了一个转换函数。def baostock_code(symbol: str) - str: if symbol.startswith(6): return fsh.{symbol} return fsz.{symbol}目前这个判断主要用于普通A股学习示例。如果以后要支持北交所、ETF、指数或者其他证券类型还需要继续补充代码规则。九、计算移动平均线和波动率项目计算了MA5、MA20和MA60三条移动平均线。data[ma5] data[close].rolling(5).mean() data[ma20] data[close].rolling(20).mean() data[ma60] data[close].rolling(60).mean()MA5表示最近5个交易日收盘价的平均值MA20表示最近20个交易日收盘价的平均值MA60表示最近60个交易日收盘价的平均值。每日收益率使用pct_change计算data[daily_return] data[close].pct_change()20日年化波动率的代码为data[volatility20] data[daily_return].rolling(20).std() * (252 ** 0.5)这里用252近似表示一年中的交易日数量。这个数值只能反映历史价格波动程度不能直接预测未来涨跌。十、计算RSI14RSI是相对强弱指标。本项目使用14个交易日作为计算窗口。def calculate_rsi(close: pd.Series, window: int) - pd.Series: delta close.diff() gain delta.clip(lower0) loss -delta.clip(upper0) avg_gain gain.rolling(window).mean() avg_loss loss.rolling(window).mean() rs avg_gain / avg_loss return 100 - (100 / (1 rs))程序先计算相邻交易日收盘价的变化再分别提取上涨和下跌部分最后计算平均上涨幅度、平均下跌幅度和RSI结果。图表中另外画出了70和30两条参考线。程序把RSI高于70视为短线可能偏热把RSI低于30视为可能进入超卖区域。这只是常见技术指标规则的代码演示并不是交易建议。十一、简单信号是怎样生成的项目中设置了一个trend_score变量用来演示怎样把多个条件组合成一个结果。程序主要判断以下几项第一最新收盘价是否高于MA20。第二MA20是否高于MA60。第三RSI是否处于较高或较低位置。第四20日年化波动率是否过高。最后根据分数返回“观察偏强”“中性观望”或者“风险偏高”。if trend_score 2: label 观察偏强 elif trend_score -1: label 风险偏高 else: label 中性观望这个部分的主要作用是展示条件判断和文本输出并不是真正的量化交易策略也没有经过完整回测不能用于真实交易。十二、绘制K线和均线K线图使用Plotly中的Candlestick绘制。fig.add_trace( go.Candlestick( xdata[date], opendata[open], highdata[high], lowdata[low], closedata[close], nameK线, ) )接着把MA5、MA20和MA60添加到同一张图中。fig.add_trace(go.Scatter(xdata[date], ydata[ma5], nameMA5, linedict(width1.5))) fig.add_trace(go.Scatter(xdata[date], ydata[ma20], nameMA20, linedict(width1.5))) fig.add_trace(go.Scatter(xdata[date], ydata[ma60], nameMA60, linedict(width1.5)))这样在查看K线的同时也可以看到不同周期移动平均线的变化。十三、绘制成交量和RSI成交量使用柱状图显示def draw_volume_chart(data: pd.DataFrame) - go.Figure: fig go.Figure() fig.add_trace(go.Bar(xdata[date], ydata[volume], name成交量)) fig.update_layout(height260, margindict(l20, r20, t30, b20)) return figRSI使用折线图显示同时添加70和30两条参考线def draw_rsi_chart(data: pd.DataFrame) - go.Figure: fig go.Figure() fig.add_trace(go.Scatter(xdata[date], ydata[rsi14], nameRSI14)) fig.add_hline(y70, line_dashdash, line_colorred) fig.add_hline(y30, line_dashdash, line_colorgreen) fig.update_layout(height260, margindict(l20, r20, t30, b20), yaxis_range[0, 100]) return fig十四、Streamlit页面部分页面左侧使用sidebar放置股票代码、开始日期、结束日期和查询按钮。with st.sidebar: st.header(查询设置) raw_symbol st.text_input(股票代码, value000001, help示例000001平安银行600519贵州茅台) end st.date_input(结束日期, valuedate.today()) start st.date_input(开始日期, valuedate.today() - timedelta(days365)) run st.button(开始抓取, use_container_widthTrue)默认股票代码是000001默认查询最近一年的数据。查询完成后页面使用五个metric组件显示股票名称、最新收盘价、当日涨跌幅、RSI14和信号参考。col1, col2, col3, col4, col5 st.columns(5) col1.metric(股票名称, stock_name) col2.metric(最新收盘, f{latest[close]:.2f}) col3.metric(当日涨跌幅, f{latest[pct_change]:.2f}%) col4.metric(RSI14, f{latest[rsi14]:.2f}) col5.metric(信号参考, signal)页面下方依次显示K线图、成交量图、RSI图、信号解释和最近30条原始数据。十五、异常处理如果查询结果为空程序会提示用户检查股票代码和日期范围。if data.empty: st.warning(没有抓到数据。请检查股票代码或日期范围。) st.stop()如果程序执行过程中出现其他异常会在页面中显示错误信息。except Exception as error: st.error(f抓取或分析失败{error})这样比直接让程序退出更容易定位问题。十六、完整项目代码app.pyfrom datetime import date, timedelta import os import akshare as ak import pandas as pd import plotly.graph_objects as go import streamlit as st 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_stock_code(raw_code: str) - str: return raw_code.strip().replace( , ).replace(., ) st.cache_data(ttl60 * 30) def load_stock_history(symbol: str, start_date: str, end_date: str) - pd.DataFrame: data ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustqfq, ) if data.empty: return data data data.rename( columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 涨跌幅: pct_change, 换手率: turnover, } ) data[date] pd.to_datetime(data[date]) data data.sort_values(date).reset_index(dropTrue) return add_indicators(data) st.cache_data(ttl60 * 30) def load_stock_name(symbol: str) - str: try: spot_data ak.stock_zh_a_spot_em() except Exception: return load_baostock_name(symbol) matched spot_data[spot_data[代码].astype(str) symbol] if matched.empty: return load_baostock_name(symbol) return matched.iloc[0][名称] def baostock_code(symbol: str) - str: if symbol.startswith(6): return fsh.{symbol} return fsz.{symbol} st.cache_data(ttl60 * 30) def load_baostock_name(symbol: str) - str: import baostock as bs login bs.login() if login.error_code ! 0: return symbol try: result bs.query_stock_basic(codebaostock_code(symbol)) if result.error_code ! 0 or not result.next(): return symbol row dict(zip(result.fields, result.get_row_data())) return row.get(code_name) or symbol finally: bs.logout() st.cache_data(ttl60 * 30) def load_baostock_history(symbol: str, start_date: str, end_date: str) - pd.DataFrame: import baostock as bs login bs.login() if login.error_code ! 0: raise RuntimeError(login.error_msg) try: result bs.query_history_k_data_plus( baostock_code(symbol), date,open,high,low,close,volume,amount,pctChg,turn, start_datef{start_date[:4]}-{start_date[4:6]}-{start_date[6:]}, end_datef{end_date[:4]}-{end_date[4:6]}-{end_date[6:]}, frequencyd, adjustflag2, ) rows [] while result.next(): rows.append(result.get_row_data()) data pd.DataFrame(rows, columnsresult.fields) finally: bs.logout() if data.empty: return data data data.rename( columns{ pctChg: pct_change, turn: turnover, } ) data[date] pd.to_datetime(data[date]) for column in [open, high, low, close, volume, amount, pct_change, turnover]: data[column] pd.to_numeric(data[column], errorscoerce) data data.sort_values(date).reset_index(dropTrue) return add_indicators(data) def add_indicators(data: pd.DataFrame) - pd.DataFrame: data data.copy() data[ma5] data[close].rolling(5).mean() data[ma20] data[close].rolling(20).mean() data[ma60] data[close].rolling(60).mean() data[daily_return] data[close].pct_change() data[volatility20] data[daily_return].rolling(20).std() * (252 ** 0.5) data[rsi14] calculate_rsi(data[close], 14) return data def calculate_rsi(close: pd.Series, window: int) - pd.Series: delta close.diff() gain delta.clip(lower0) loss -delta.clip(upper0) avg_gain gain.rolling(window).mean() avg_loss loss.rolling(window).mean() rs avg_gain / avg_loss return 100 - (100 / (1 rs)) def make_signal(data: pd.DataFrame) - tuple[str, str]: latest data.dropna().iloc[-1] trend_score 0 reasons [] if latest[close] latest[ma20]: trend_score 1 reasons.append(收盘价在 20 日均线上方短期趋势偏强。) else: reasons.append(收盘价在 20 日均线下方短期趋势偏弱。) if latest[ma20] latest[ma60]: trend_score 1 reasons.append(20 日均线高于 60 日均线中期趋势更健康。) else: reasons.append(20 日均线低于 60 日均线中期趋势仍需观察。) if latest[rsi14] 30: trend_score 1 reasons.append(RSI 低于 30可能进入超卖区。) elif latest[rsi14] 70: trend_score - 1 reasons.append(RSI 高于 70短线过热风险较高。) else: reasons.append(RSI 位于常规区间未出现明显极端情绪。) if latest[volatility20] 0.45: trend_score - 1 reasons.append(20 日年化波动率较高价格波动风险较大。) else: reasons.append(20 日年化波动率相对可控。) if trend_score 2: label 观察偏强 elif trend_score -1: label 风险偏高 else: label 中性观望 return label, \n.join(reasons) def draw_price_chart(data: pd.DataFrame, display_name: str) - go.Figure: fig go.Figure() fig.add_trace( go.Candlestick( xdata[date], opendata[open], highdata[high], lowdata[low], closedata[close], nameK线, ) ) fig.add_trace(go.Scatter(xdata[date], ydata[ma5], nameMA5, linedict(width1.5))) fig.add_trace(go.Scatter(xdata[date], ydata[ma20], nameMA20, linedict(width1.5))) fig.add_trace(go.Scatter(xdata[date], ydata[ma60], nameMA60, linedict(width1.5))) fig.update_layout( titlef{display_name} 前复权日线, height560, margindict(l20, r20, t50, b20), xaxis_rangeslider_visibleFalse, ) return fig def draw_volume_chart(data: pd.DataFrame) - go.Figure: fig go.Figure() fig.add_trace(go.Bar(xdata[date], ydata[volume], name成交量)) fig.update_layout(height260, margindict(l20, r20, t30, b20)) return fig def draw_rsi_chart(data: pd.DataFrame) - go.Figure: fig go.Figure() fig.add_trace(go.Scatter(xdata[date], ydata[rsi14], nameRSI14)) fig.add_hline(y70, line_dashdash, line_colorred) fig.add_hline(y30, line_dashdash, line_colorgreen) fig.update_layout(height260, margindict(l20, r20, t30, b20), yaxis_range[0, 100]) return fig st.title(A股市场爬虫与投资信号可视化) st.caption(数据用于学习和研究不构成任何投资建议。真实交易请结合基本面、仓位、风险承受能力和专业意见。) with st.sidebar: st.header(查询设置) raw_symbol st.text_input(股票代码, value000001, help示例000001 平安银行600519 贵州茅台) end st.date_input(结束日期, valuedate.today()) start st.date_input(开始日期, valuedate.today() - timedelta(days365)) run st.button(开始抓取, use_container_widthTrue) symbol normalize_stock_code(raw_symbol) start_text start.strftime(%Y%m%d) end_text end.strftime(%Y%m%d) if run or last_symbol not in st.session_state: st.session_state[last_symbol] symbol try: try: data load_stock_history(st.session_state[last_symbol], start_text, end_text) data_mode 东方财富历史行情 except Exception as history_error: st.warning(f东方财富历史行情暂时获取失败正在尝试 BaoStock 历史行情{history_error}) data load_baostock_history(st.session_state[last_symbol], start_text, end_text) data_mode BaoStock 历史行情 if data.empty: st.warning(没有抓到数据。请检查股票代码或日期范围。) st.stop() stock_name load_stock_name(st.session_state[last_symbol]) display_name f{stock_name}{st.session_state[last_symbol]} latest data.iloc[-1] signal, reason make_signal(data) st.info(f数据模式{data_mode}) st.subheader(display_name) col1, col2, col3, col4, col5 st.columns(5) col1.metric(股票名称, stock_name) col2.metric(最新收盘, f{latest[close]:.2f}) col3.metric(当日涨跌幅, f{latest[pct_change]:.2f}%) col4.metric(RSI14, f{latest[rsi14]:.2f}) col5.metric(信号参考, signal) st.plotly_chart(draw_price_chart(data, display_name), use_container_widthTrue) left, right st.columns(2) left.subheader(成交量) left.plotly_chart(draw_volume_chart(data), use_container_widthTrue) right.subheader(RSI) right.plotly_chart(draw_rsi_chart(data), use_container_widthTrue) st.subheader(信号解释) st.info(reason) st.subheader(原始数据) st.dataframe(data.tail(30), use_container_widthTrue) except Exception as error: st.error(f抓取或分析失败{error})十七、总结这个项目把数据获取、数据清洗、技术指标计算、图表绘制和网页展示放在了一个程序中比较适合用来练习Python数据分析和Streamlit开发。通过这个项目可以熟悉AkShare和BaoStock的基本使用方法理解Pandas滚动计算学习Plotly金融图表并掌握Streamlit输入组件、缓存、页面布局和异常提示的写法。再次说明本文和项目中的所有代码仅用于学习和技术交流程序输出不构成任何投资建议。第三方数据接口可能存在延迟、缺失、错误或临时不可用的情况任何真实投资行为都应由使用者独立判断并自行承担风险。
Python实现A股行情爬取与可视化，一个适合练手的Streamlit项目（附完整代码、详细代码说明）
说明本文和项目代码仅用于Python编程、数据分析和可视化学习不构成任何投资建议。程序中出现的“观察偏强”“中性观望”“风险偏高”等结果都是根据固定规则计算得到的学习示例不能作为股票买卖依据。股市有风险投资需谨慎。一、项目简介最近在学习Python数据处理和Streamlit网页开发于是做了一个A股历史行情查询与可视化的小项目。这个项目不需要单独写前端页面。运行app.py之后Streamlit会自动生成一个网页。用户只需要输入股票代码、开始日期和结束日期就可以查询对应股票的历史日线数据。网页中会显示股票名称、最新收盘价、当日涨跌幅、K线图、MA5、MA20、MA60均线、成交量、RSI14、20日年化波动率以及程序根据几个技术指标生成的简单参考说明。项目默认先通过AkShare获取东方财富历史行情。如果这个接口临时请求失败程序会自动切换到BaoStock继续获取数据这样可以减少因为单一接口不可用而导致整个页面无法运行的情况。二、项目用到的工具本项目主要使用了Streamlit、AkShare、BaoStock、Pandas和Plotly。Streamlit负责生成网页界面AkShare负责获取A股行情和股票名称BaoStock作为备用数据源Pandas负责数据清洗和指标计算Plotly负责绘制K线、均线、成交量和RSI图表。项目运行环境为Python 3核心文件是app.py。三、安装项目依赖先在终端中进入项目文件夹然后执行python -m venv venvWindows PowerShell中激活虚拟环境.\venv\Scripts\Activate.ps1Windows CMD中激活虚拟环境venv\Scripts\activate安装依赖pip install streamlit akshare baostock pandas plotly四、运行项目在app.py所在目录执行streamlit run app.py如果提示找不到streamlit命令可以改为python -m streamlit run app.py程序启动后终端一般会显示下面的本地地址http://localhost:8501在浏览器中打开这个地址就可以进入查询页面。五、股票代码处理为了减少输入格式带来的问题程序会先对股票代码进行简单清理。代码如下def normalize_stock_code(raw_code: str) - str: return raw_code.strip().replace( , ).replace(., )这个函数会删除股票代码前后的空格同时删除中间的空格和小数点。例如输入600519、600519.或者600 519程序都会尽可能处理成标准代码。六、获取股票历史行情程序首先通过AkShare中的stock_zh_a_hist接口获取A股日线数据。主要代码如下st.cache_data(ttl60 * 30) def load_stock_history(symbol: str, start_date: str, end_date: str) - pd.DataFrame: data ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustqfq, )symbol表示股票代码period设置为daily表示查询日线数据start_date和end_date分别表示开始日期和结束日期adjust设置为qfq表示使用前复权数据。这里使用了Streamlit的数据缓存功能缓存时间设置为30分钟。这样在短时间内重复查询相同内容时不需要每次都重新请求接口页面响应速度会更快。七、统一数据字段AkShare返回的数据字段是中文名称为了方便后面的计算程序把字段统一改成了英文。data data.rename( columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 涨跌幅: pct_change, 换手率: turnover, } )之后再把date转换成Pandas日期格式并按照日期从早到晚排序。data[date] pd.to_datetime(data[date]) data data.sort_values(date).reset_index(dropTrue)八、加入BaoStock备用数据源网络环境、代理设置或者接口维护都可能造成东方财富行情暂时无法获取。因此程序中加入了BaoStock备用方案。try: data load_stock_history( st.session_state[last_symbol], start_text, end_text ) data_mode 东方财富历史行情 except Exception as history_error: st.warning(f东方财富历史行情暂时获取失败正在尝试BaoStock历史行情{history_error}) data load_baostock_history( st.session_state[last_symbol], start_text, end_text ) data_mode BaoStock历史行情程序会先请求AkShare。如果出现异常就在页面中给出提示然后自动调用BaoStock。BaoStock使用的股票代码格式和普通六位代码不同。上海股票通常需要写成sh.600519深圳股票通常需要写成sz.000001所以程序增加了一个转换函数。def baostock_code(symbol: str) - str: if symbol.startswith(6): return fsh.{symbol} return fsz.{symbol}目前这个判断主要用于普通A股学习示例。如果以后要支持北交所、ETF、指数或者其他证券类型还需要继续补充代码规则。九、计算移动平均线和波动率项目计算了MA5、MA20和MA60三条移动平均线。data[ma5] data[close].rolling(5).mean() data[ma20] data[close].rolling(20).mean() data[ma60] data[close].rolling(60).mean()MA5表示最近5个交易日收盘价的平均值MA20表示最近20个交易日收盘价的平均值MA60表示最近60个交易日收盘价的平均值。每日收益率使用pct_change计算data[daily_return] data[close].pct_change()20日年化波动率的代码为data[volatility20] data[daily_return].rolling(20).std() * (252 ** 0.5)这里用252近似表示一年中的交易日数量。这个数值只能反映历史价格波动程度不能直接预测未来涨跌。十、计算RSI14RSI是相对强弱指标。本项目使用14个交易日作为计算窗口。def calculate_rsi(close: pd.Series, window: int) - pd.Series: delta close.diff() gain delta.clip(lower0) loss -delta.clip(upper0) avg_gain gain.rolling(window).mean() avg_loss loss.rolling(window).mean() rs avg_gain / avg_loss return 100 - (100 / (1 rs))程序先计算相邻交易日收盘价的变化再分别提取上涨和下跌部分最后计算平均上涨幅度、平均下跌幅度和RSI结果。图表中另外画出了70和30两条参考线。程序把RSI高于70视为短线可能偏热把RSI低于30视为可能进入超卖区域。这只是常见技术指标规则的代码演示并不是交易建议。十一、简单信号是怎样生成的项目中设置了一个trend_score变量用来演示怎样把多个条件组合成一个结果。程序主要判断以下几项第一最新收盘价是否高于MA20。第二MA20是否高于MA60。第三RSI是否处于较高或较低位置。第四20日年化波动率是否过高。最后根据分数返回“观察偏强”“中性观望”或者“风险偏高”。if trend_score 2: label 观察偏强 elif trend_score -1: label 风险偏高 else: label 中性观望这个部分的主要作用是展示条件判断和文本输出并不是真正的量化交易策略也没有经过完整回测不能用于真实交易。十二、绘制K线和均线K线图使用Plotly中的Candlestick绘制。fig.add_trace( go.Candlestick( xdata[date], opendata[open], highdata[high], lowdata[low], closedata[close], nameK线, ) )接着把MA5、MA20和MA60添加到同一张图中。fig.add_trace(go.Scatter(xdata[date], ydata[ma5], nameMA5, linedict(width1.5))) fig.add_trace(go.Scatter(xdata[date], ydata[ma20], nameMA20, linedict(width1.5))) fig.add_trace(go.Scatter(xdata[date], ydata[ma60], nameMA60, linedict(width1.5)))这样在查看K线的同时也可以看到不同周期移动平均线的变化。十三、绘制成交量和RSI成交量使用柱状图显示def draw_volume_chart(data: pd.DataFrame) - go.Figure: fig go.Figure() fig.add_trace(go.Bar(xdata[date], ydata[volume], name成交量)) fig.update_layout(height260, margindict(l20, r20, t30, b20)) return figRSI使用折线图显示同时添加70和30两条参考线def draw_rsi_chart(data: pd.DataFrame) - go.Figure: fig go.Figure() fig.add_trace(go.Scatter(xdata[date], ydata[rsi14], nameRSI14)) fig.add_hline(y70, line_dashdash, line_colorred) fig.add_hline(y30, line_dashdash, line_colorgreen) fig.update_layout(height260, margindict(l20, r20, t30, b20), yaxis_range[0, 100]) return fig十四、Streamlit页面部分页面左侧使用sidebar放置股票代码、开始日期、结束日期和查询按钮。with st.sidebar: st.header(查询设置) raw_symbol st.text_input(股票代码, value000001, help示例000001平安银行600519贵州茅台) end st.date_input(结束日期, valuedate.today()) start st.date_input(开始日期, valuedate.today() - timedelta(days365)) run st.button(开始抓取, use_container_widthTrue)默认股票代码是000001默认查询最近一年的数据。查询完成后页面使用五个metric组件显示股票名称、最新收盘价、当日涨跌幅、RSI14和信号参考。col1, col2, col3, col4, col5 st.columns(5) col1.metric(股票名称, stock_name) col2.metric(最新收盘, f{latest[close]:.2f}) col3.metric(当日涨跌幅, f{latest[pct_change]:.2f}%) col4.metric(RSI14, f{latest[rsi14]:.2f}) col5.metric(信号参考, signal)页面下方依次显示K线图、成交量图、RSI图、信号解释和最近30条原始数据。十五、异常处理如果查询结果为空程序会提示用户检查股票代码和日期范围。if data.empty: st.warning(没有抓到数据。请检查股票代码或日期范围。) st.stop()如果程序执行过程中出现其他异常会在页面中显示错误信息。except Exception as error: st.error(f抓取或分析失败{error})这样比直接让程序退出更容易定位问题。十六、完整项目代码app.pyfrom datetime import date, timedelta import os import akshare as ak import pandas as pd import plotly.graph_objects as go import streamlit as st 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_stock_code(raw_code: str) - str: return raw_code.strip().replace( , ).replace(., ) st.cache_data(ttl60 * 30) def load_stock_history(symbol: str, start_date: str, end_date: str) - pd.DataFrame: data ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustqfq, ) if data.empty: return data data data.rename( columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 涨跌幅: pct_change, 换手率: turnover, } ) data[date] pd.to_datetime(data[date]) data data.sort_values(date).reset_index(dropTrue) return add_indicators(data) st.cache_data(ttl60 * 30) def load_stock_name(symbol: str) - str: try: spot_data ak.stock_zh_a_spot_em() except Exception: return load_baostock_name(symbol) matched spot_data[spot_data[代码].astype(str) symbol] if matched.empty: return load_baostock_name(symbol) return matched.iloc[0][名称] def baostock_code(symbol: str) - str: if symbol.startswith(6): return fsh.{symbol} return fsz.{symbol} st.cache_data(ttl60 * 30) def load_baostock_name(symbol: str) - str: import baostock as bs login bs.login() if login.error_code ! 0: return symbol try: result bs.query_stock_basic(codebaostock_code(symbol)) if result.error_code ! 0 or not result.next(): return symbol row dict(zip(result.fields, result.get_row_data())) return row.get(code_name) or symbol finally: bs.logout() st.cache_data(ttl60 * 30) def load_baostock_history(symbol: str, start_date: str, end_date: str) - pd.DataFrame: import baostock as bs login bs.login() if login.error_code ! 0: raise RuntimeError(login.error_msg) try: result bs.query_history_k_data_plus( baostock_code(symbol), date,open,high,low,close,volume,amount,pctChg,turn, start_datef{start_date[:4]}-{start_date[4:6]}-{start_date[6:]}, end_datef{end_date[:4]}-{end_date[4:6]}-{end_date[6:]}, frequencyd, adjustflag2, ) rows [] while result.next(): rows.append(result.get_row_data()) data pd.DataFrame(rows, columnsresult.fields) finally: bs.logout() if data.empty: return data data data.rename( columns{ pctChg: pct_change, turn: turnover, } ) data[date] pd.to_datetime(data[date]) for column in [open, high, low, close, volume, amount, pct_change, turnover]: data[column] pd.to_numeric(data[column], errorscoerce) data data.sort_values(date).reset_index(dropTrue) return add_indicators(data) def add_indicators(data: pd.DataFrame) - pd.DataFrame: data data.copy() data[ma5] data[close].rolling(5).mean() data[ma20] data[close].rolling(20).mean() data[ma60] data[close].rolling(60).mean() data[daily_return] data[close].pct_change() data[volatility20] data[daily_return].rolling(20).std() * (252 ** 0.5) data[rsi14] calculate_rsi(data[close], 14) return data def calculate_rsi(close: pd.Series, window: int) - pd.Series: delta close.diff() gain delta.clip(lower0) loss -delta.clip(upper0) avg_gain gain.rolling(window).mean() avg_loss loss.rolling(window).mean() rs avg_gain / avg_loss return 100 - (100 / (1 rs)) def make_signal(data: pd.DataFrame) - tuple[str, str]: latest data.dropna().iloc[-1] trend_score 0 reasons [] if latest[close] latest[ma20]: trend_score 1 reasons.append(收盘价在 20 日均线上方短期趋势偏强。) else: reasons.append(收盘价在 20 日均线下方短期趋势偏弱。) if latest[ma20] latest[ma60]: trend_score 1 reasons.append(20 日均线高于 60 日均线中期趋势更健康。) else: reasons.append(20 日均线低于 60 日均线中期趋势仍需观察。) if latest[rsi14] 30: trend_score 1 reasons.append(RSI 低于 30可能进入超卖区。) elif latest[rsi14] 70: trend_score - 1 reasons.append(RSI 高于 70短线过热风险较高。) else: reasons.append(RSI 位于常规区间未出现明显极端情绪。) if latest[volatility20] 0.45: trend_score - 1 reasons.append(20 日年化波动率较高价格波动风险较大。) else: reasons.append(20 日年化波动率相对可控。) if trend_score 2: label 观察偏强 elif trend_score -1: label 风险偏高 else: label 中性观望 return label, \n.join(reasons) def draw_price_chart(data: pd.DataFrame, display_name: str) - go.Figure: fig go.Figure() fig.add_trace( go.Candlestick( xdata[date], opendata[open], highdata[high], lowdata[low], closedata[close], nameK线, ) ) fig.add_trace(go.Scatter(xdata[date], ydata[ma5], nameMA5, linedict(width1.5))) fig.add_trace(go.Scatter(xdata[date], ydata[ma20], nameMA20, linedict(width1.5))) fig.add_trace(go.Scatter(xdata[date], ydata[ma60], nameMA60, linedict(width1.5))) fig.update_layout( titlef{display_name} 前复权日线, height560, margindict(l20, r20, t50, b20), xaxis_rangeslider_visibleFalse, ) return fig def draw_volume_chart(data: pd.DataFrame) - go.Figure: fig go.Figure() fig.add_trace(go.Bar(xdata[date], ydata[volume], name成交量)) fig.update_layout(height260, margindict(l20, r20, t30, b20)) return fig def draw_rsi_chart(data: pd.DataFrame) - go.Figure: fig go.Figure() fig.add_trace(go.Scatter(xdata[date], ydata[rsi14], nameRSI14)) fig.add_hline(y70, line_dashdash, line_colorred) fig.add_hline(y30, line_dashdash, line_colorgreen) fig.update_layout(height260, margindict(l20, r20, t30, b20), yaxis_range[0, 100]) return fig st.title(A股市场爬虫与投资信号可视化) st.caption(数据用于学习和研究不构成任何投资建议。真实交易请结合基本面、仓位、风险承受能力和专业意见。) with st.sidebar: st.header(查询设置) raw_symbol st.text_input(股票代码, value000001, help示例000001 平安银行600519 贵州茅台) end st.date_input(结束日期, valuedate.today()) start st.date_input(开始日期, valuedate.today() - timedelta(days365)) run st.button(开始抓取, use_container_widthTrue) symbol normalize_stock_code(raw_symbol) start_text start.strftime(%Y%m%d) end_text end.strftime(%Y%m%d) if run or last_symbol not in st.session_state: st.session_state[last_symbol] symbol try: try: data load_stock_history(st.session_state[last_symbol], start_text, end_text) data_mode 东方财富历史行情 except Exception as history_error: st.warning(f东方财富历史行情暂时获取失败正在尝试 BaoStock 历史行情{history_error}) data load_baostock_history(st.session_state[last_symbol], start_text, end_text) data_mode BaoStock 历史行情 if data.empty: st.warning(没有抓到数据。请检查股票代码或日期范围。) st.stop() stock_name load_stock_name(st.session_state[last_symbol]) display_name f{stock_name}{st.session_state[last_symbol]} latest data.iloc[-1] signal, reason make_signal(data) st.info(f数据模式{data_mode}) st.subheader(display_name) col1, col2, col3, col4, col5 st.columns(5) col1.metric(股票名称, stock_name) col2.metric(最新收盘, f{latest[close]:.2f}) col3.metric(当日涨跌幅, f{latest[pct_change]:.2f}%) col4.metric(RSI14, f{latest[rsi14]:.2f}) col5.metric(信号参考, signal) st.plotly_chart(draw_price_chart(data, display_name), use_container_widthTrue) left, right st.columns(2) left.subheader(成交量) left.plotly_chart(draw_volume_chart(data), use_container_widthTrue) right.subheader(RSI) right.plotly_chart(draw_rsi_chart(data), use_container_widthTrue) st.subheader(信号解释) st.info(reason) st.subheader(原始数据) st.dataframe(data.tail(30), use_container_widthTrue) except Exception as error: st.error(f抓取或分析失败{error})十七、总结这个项目把数据获取、数据清洗、技术指标计算、图表绘制和网页展示放在了一个程序中比较适合用来练习Python数据分析和Streamlit开发。通过这个项目可以熟悉AkShare和BaoStock的基本使用方法理解Pandas滚动计算学习Plotly金融图表并掌握Streamlit输入组件、缓存、页面布局和异常提示的写法。再次说明本文和项目中的所有代码仅用于学习和技术交流程序输出不构成任何投资建议。第三方数据接口可能存在延迟、缺失、错误或临时不可用的情况任何真实投资行为都应由使用者独立判断并自行承担风险。
