Python實戰：基于Streamlit的股票篩選系統，實時K線圖+數據緩存優化

基于?Streamlit?構建的股票篩選分析工具，整合了?Tushare?接口獲取股票數據，并通過交互式界面實現股票篩選、信息展示和 K 線圖分析。以下是深度解讀：

一、代碼結構概覽

依賴庫導入
```
import streamlit as st
import tushare as ts
import pandas as pd
import plotly.graph_objects as go
```
- Streamlit：用于構建 Web 應用界面。
- Tushare：提供股票市場數據接口。
- Pandas：數據處理和分析。
- Plotly：生成交互式圖表（如 K 線圖）。
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 線圖。
- 顏色區分漲跌（漲：綠色，跌：紅色）。
- 用戶可縮放/平移查看歷史數據。

三、代碼特征

交互式篩選與可視化
- 通過?Streamlit?的組件（如?number_input、multiselect、selectbox）實現動態交互。
- 實時響應：用戶操作后，界面立即更新，無需手動刷新。
數據緩存優化
- 使用?@st.cache_data?緩存股票列表和日線數據，減少 API 請求次數。
- 適用于頻繁訪問的場景（如股票篩選）。
異常處理與用戶體驗
- 缺失值處理：填充?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")