Python实战：基于Streamlit的股票筛选系统，实时K线图+数据缓存优化

基于 Streamlit 构建的股票筛选分析工具，整合了 Tushare 接口获取股票数据，并通过交互式界面实现股票筛选、信息展示和 K 线图分析。以下是深度解读：

一、代码结构概览

1. 依赖库导入

   import streamlit as st
   import tushare as ts
   import pandas as pd
   import plotly.graph_objects as go

   • Streamlit：用于构建 Web 应用界面。
   • Tushare：提供股票市场数据接口。
   • Pandas：数据处理和分析。
   • Plotly：生成交互式图表（如 K 线图）。

2. Tushare API 配置

   ts.set_token("Your_API_Token_Here")
   pro = ts.pro_api()

   • 需替换为实际的 Tushare Token，否则无法获取数据。

二、核心功能模块分析

1. 获取股票基础信息（get_stock_list）

@st.cache_data
def get_stock_list():# 获取股票基本信息（如代码、名称、行业、市值等）df = pro.stock_basic(list_status="L", fields="ts_code,symbol,name,area,industry,list_date")# 填充缺失值df["area"] = df["area"].fillna("Unknown")df["industry"] = df["industry"].fillna("Unknown")# 获取并处理市值数据market_cap_data = pro.daily_basic()market_cap_data["total_mv"] = market_cap_data["total_mv"] / 10000  # 转换为亿元# 合并数据集df = pd.merge(df, market_cap_data[["ts_code", "total_mv"]], on="ts_code", how="left")df = df.rename(columns={"total_mv": "market_cap"})return df

• 功能：获取股票基础信息并合并市值数据。
• 优化点：
  • list_status="L" 表示只获取上市股票（非停牌）。
  • @st.cache_data 缓存数据，避免重复请求。
  • 市值数据转换为亿元（/10000）提高可读性。

2. 获取股票日线数据（get_daily_data）

@st.cache_data
def get_daily_data(ts_code):data = pro.daily(ts_code=ts_code)data["trade_date"] = pd.to_datetime(data["trade_date"])return data.sort_values("trade_date")

• 功能：根据股票代码获取日线数据（如开盘价、收盘价等）。
• 优化点：
  • @st.cache_data 缓存数据，提升性能。
  • 日期格式转换为 datetime 类型，便于后续筛选。

3. 主界面逻辑（main()）

def main():st.title("股票筛选分析工具")# 侧边栏筛选min_market_cap = st.sidebar.number_input("最小市值(亿)", value=100)max_market_cap = st.sidebar.number_input("最大市值(亿)", value=1000)selected_industry = st.sidebar.multiselect("选择行业", stocks["industry"].unique())selected_area = st.sidebar.multiselect("选择地域", stocks["area"].unique())# 应用筛选条件filtered_stocks = stocks[(stocks["market_cap"] >= min_market_cap) & (stocks["market_cap"] <= max_market_cap)]if selected_industry: filtered_stocks = filtered_stocks[filtered_stocks["industry"].isin(selected_industry)]if selected_area: filtered_stocks = filtered_stocks[filtered_stocks["area"].isin(selected_area)]# 展示结果st.dataframe(filtered_stocks.sort_values(by="market_cap", ascending=False))

• 功能：通过侧边栏提供筛选选项（市值、行业、地域），并展示符合条件的股票列表。
• 交互亮点：
  • 动态筛选：实时响应用户输入（无需刷新页面）。
  • 数据排序：按市值从高到低展示，便于用户快速定位。

4. 个股详情分析

if not filtered_stocks.empty:selected_stock = st.selectbox("选择股票查看详情", filtered_stocks["name"])start_date = st.date_input("开始日期", value=today - pd.Timedelta(days=365))end_date = st.date_input("结束日期", value=today)# 获取日线数据并绘制K线图stock_code = filtered_stocks[filtered_stocks["name"] == selected_stock]["ts_code"].values[0]daily_data = get_daily_data(stock_code)daily_data = daily_data[(daily_data["trade_date"] >= start_date) &(daily_data["trade_date"] <= end_date)]# 绘制K线图fig = go.Figure(data=[go.Candlestick(x=daily_data["trade_date"],open=daily_data["open"],high=daily_data["high"],low=daily_data["low"],close=daily_data["close"],increasing_line_color="green",decreasing_line_color="red",)])st.plotly_chart(fig, use_container_width=True)

• 功能：选择某只股票后，展示其历史 K 线图。
• 图表设计：
  • 使用 Plotly 的 Candlestick 组件生成交互式 K 线图。
  • 颜色区分涨跌（涨：绿色，跌：红色）。
  • 用户可缩放/平移查看历史数据。

三、代码特征

1. 交互式筛选与可视化

   • 通过 Streamlit 的组件（如 number_input、multiselect、selectbox）实现动态交互。
   • 实时响应：用户操作后，界面立即更新，无需手动刷新。

2. 数据缓存优化

   • 使用 @st.cache_data 缓存股票列表和日线数据，减少 API 请求次数。
   • 适用于频繁访问的场景（如股票筛选）。

3. 异常处理与用户体验

   • 缺失值处理：填充 area 和 industry 的缺失值为 "Unknown"。
   • 数据缺失提示：若 K 线数据缺失，提示用户并使用前值填充。
   • 空结果提示：若筛选条件不匹配，显示警告信息。

四、完整代码

import streamlit as st
import tushare as ts
import pandas as pd
import plotly.graph_objects as go
import sys
import os# 设置Tushare API（需替换为您的实际Token）
ts.set_token("Your_API_Token_Here")
pro = ts.pro_api()# 缓存股票基础数据
@st.cache_data
def get_stock_list():# 获取股票基本信息df = pro.stock_basic(exchange="", list_status="L", fields="ts_code,symbol,name,area,industry,list_date")# 处理缺失值df["area"] = df["area"].fillna("Unknown")df["industry"] = df["industry"].fillna("Unknown")# 获取并处理市值数据market_cap_data = pro.daily_basic()market_cap_data["total_mv"] = market_cap_data["total_mv"] / 10000  # 转换为亿元# 合并数据集df = pd.merge(df, market_cap_data[["ts_code", "total_mv"]], on="ts_code", how="left")df = df.rename(columns={"total_mv": "market_cap"})return df# 缓存股票日线数据
@st.cache_data
def get_daily_data(ts_code):data = pro.daily(ts_code=ts_code)data["trade_date"] = pd.to_datetime(data["trade_date"])return data.sort_values("trade_date")# 主应用界面
def main():st.title("股票筛选分析工具")# 侧边栏筛选选项st.sidebar.header("筛选条件")# 获取股票数据stocks = get_stock_list()# 市值范围筛选min_market_cap = st.sidebar.number_input("最小市值(亿)", min_value=0, value=100)max_market_cap = st.sidebar.number_input("最大市值(亿)", min_value=0, value=1000)# 行业筛选industry_list = stocks["industry"].unique().tolist()selected_industry = st.sidebar.multiselect("选择行业", industry_list)# 地域筛选area_list = stocks["area"].unique().tolist()selected_area = st.sidebar.multiselect("选择地域", area_list)# 应用筛选条件filtered_stocks = stocks[(stocks["market_cap"] >= min_market_cap) & (stocks["market_cap"] <= max_market_cap)]if selected_industry:filtered_stocks = filtered_stocks[filtered_stocks["industry"].isin(selected_industry)]if selected_area:filtered_stocks = filtered_stocks[filtered_stocks["area"].isin(selected_area)]# 格式化并展示结果display_df = filtered_stocks.copy()display_df["market_cap"] = display_df["market_cap"].round(2)st.dataframe(display_df.sort_values(by="market_cap", ascending=False))# 个股详情部分if not filtered_stocks.empty:st.subheader("个股详情分析")# 股票选择器selected_stock = st.selectbox("选择股票查看详情", filtered_stocks["name"])# 日期范围选择today = pd.Timestamp.today()col1, col2 = st.columns(2)with col1:start_date = st.date_input("开始日期", value=today - pd.Timedelta(days=365), max_value=today)with col2:end_date = st.date_input("结束日期", value=today, min_value=start_date, max_value=today)# 获取选定股票的日线数据stock_code = filtered_stocks[filtered_stocks["name"] == selected_stock]["ts_code"].values[0]daily_data = get_daily_data(stock_code)# 按日期范围筛选daily_data = daily_data[(daily_data["trade_date"] >= pd.Timestamp(start_date)) &(daily_data["trade_date"] <= pd.Timestamp(end_date))]# 处理缺失值if daily_data[["open", "high", "low", "close"]].isnull().values.any():st.warning("部分价格数据缺失 - 使用前值填充")daily_data[["open", "high", "low", "close"]] = daily_data[["open", "high", "low", "close"]].ffill()# 绘制K线图if not daily_data.empty:fig = go.Figure(data=[go.Candlestick(x=daily_data["trade_date"],open=daily_data["open"],high=daily_data["high"],low=daily_data["low"],close=daily_data["close"],increasing_line_color="green",decreasing_line_color="red",)])fig.update_layout(title=f"{selected_stock} K线图",xaxis_title="日期",yaxis_title="价格")st.plotly_chart(fig, use_container_width=True)else:st.warning("所选日期范围内无可用数据")else:st.warning("没有符合筛选条件的股票")if __name__ == "__main__":# 检查依赖是否安装try:import streamlitimport tushareimport pandasimport plotlymain()except ImportError as e:print(f"缺少依赖库: {e}")print("请执行以下命令安装依赖:")print("pip install streamlit tushare pandas plotly")