streamlit股票可视化看板Python代码

使用说明：

1. 安装依赖库：bash
   pip install akshare streamlit plotly pandas numpy
2. 运行应用程序：bash
   streamlit run stock_dashboard.py
3. 功能介绍：

• 侧边栏参数设置：
  • 输入股票代码（如SH600938、SZ000001）
  • 选择数据周期（最近1个月至2年或自定义日期）
  • 选择要显示的技术指标（均线、RSI、MACD、成交量）
  • 调整指标参数（如RSI周期、均线周期）
• 主界面显示：
  • 股票基本统计信息（最新价、涨跌幅、最高价、最低价等）
  • 交互式K线图（支持缩放、平移）
  • 技术指标图表（可选择显示）
  • 股票数据表格（最近50条记录）
  • 数据下载功能（CSV格式）

1. 技术特点：

• 使用akshare获取A股历史数据（前复权）
• 使用plotly绘制交互式图表
• 支持多种技术指标的计算和显示
• 数据缓存机制，提高性能
• 响应式设计，适配不同屏幕尺寸

该应用程序提供了一个功能完整的股票数据可视化看板，用户可以方便地查看股票的K线图和常用技术指标，并进行基本的数据分析。
Pytyhon代码

import pandas as pd
import numpy as np
import akshare as ak
import streamlit as st
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from datetime import datetime, timedelta

# 设置页面配置
st.set_page_config(page_title="股票数据看板", layout="wide", initial_sidebar_state="expanded")

# 设置中文显示
st.markdown("<h1 style='text-align: center;'>股票数据可视化看板</h1>", unsafe_allow_html=True)

# 侧边栏设置
st.sidebar.header("参数设置")

# 股票代码选择
stock_code = st.sidebar.text_input("股票代码", "SH600938", help="例如：SH600938（工商银行）、SZ000001（平安银行）")

# 日期范围选择
date_range = st.sidebar.selectbox(
    "数据周期",
    ["最近1个月", "最近3个月", "最近6个月", "最近1年", "最近2年", "自定义日期"],
    index=2  # 默认选择最近6个月
)

# 根据选择设置日期
if date_range == "自定义日期":
    start_date = st.sidebar.date_input("开始日期", datetime.now() - timedelta(days=180))
    end_date = st.sidebar.date_input("结束日期", datetime.now())
else:
    end_date = datetime.now()
    if date_range == "最近1个月":
        start_date = end_date - timedelta(days=30)
    elif date_range == "最近3个月":
        start_date = end_date - timedelta(days=90)
    elif date_range == "最近6个月":
        start_date = end_date - timedelta(days=180)
    elif date_range == "最近1年":
        start_date = end_date - timedelta(days=365)
    elif date_range == "最近2年":
        start_date = end_date - timedelta(days=730)

# 转换日期格式
start_date_str = start_date.strftime("%Y-%m-%d")
end_date_str = end_date.strftime("%Y-%m-%d")

# 技术指标选择
st.sidebar.header("技术指标")
show_ma = st.sidebar.checkbox("显示均线", True)
ma_periods = st.sidebar.multiselect("均线周期", [5, 10, 20, 60, 120, 250], [5, 10, 20, 60])

show_rsi = st.sidebar.checkbox("显示RSI", True)
rsi_period = st.sidebar.slider("RSI周期", 6, 24, 14)

show_macd = st.sidebar.checkbox("显示MACD", True)

show_volume = st.sidebar.checkbox("显示成交量", True)

# 获取股票数据
@st.cache_data(ttl=3600)  # 缓存1小时
def get_stock_data(stock_code, start_date, end_date):
    """
    使用akshare获取股票历史数据
    """
    try:
        # 转换日期格式为akshare要求的格式
        start_date_ak = start_date.replace("-", "")
        end_date_ak = end_date.replace("-", "")
        
        # 转换股票代码格式为akshare要求的格式
        if stock_code.startswith('SH'):
            stock_code_ak = stock_code[2:]
        elif stock_code.startswith('SZ'):
            stock_code_ak = stock_code[2:]
        else:
            # 假设已经是正确格式
            stock_code_ak = stock_code
        
        # 使用akshare获取股票数据
        stock_df = ak.stock_zh_a_hist(symbol=stock_code_ak, period="daily", 
                                     start_date=start_date_ak, end_date=end_date_ak, adjust="qfq")
        
        stock_df['日期'] = pd.to_datetime(stock_df['日期'])
        stock_df.set_index('日期', inplace=True)
        
        # 重命名列名以便于使用
        stock_df.rename(columns={
            '开盘': 'open',
            '收盘': 'close',
            '最高': 'high',
            '最低': 'low',
            '成交量': 'volume',
            '成交额': 'amount',
            '涨跌幅': 'pct_change'
        }, inplace=True)
        
        return stock_df
    except Exception as e:
        st.error(f"获取数据失败: {e}")
        return None

# 计算技术指标
def calculate_technical_indicators(df):
    """
    计算常用技术指标
    """
    # 计算均线
    for period in ma_periods:
        df[f'MA{period}'] = df['close'].rolling(window=period).mean()
    
    # 计算RSI指标
    if show_rsi:
        delta = df['close'].diff(1)
        gain = delta.where(delta > 0, 0)
        loss = -delta.where(delta < 0, 0)
        
        avg_gain = gain.rolling(window=rsi_period).mean()
        avg_loss = loss.rolling(window=rsi_period).mean()
        
        rs = avg_gain / avg_loss
        df['RSI'] = 100 - (100 / (1 + rs))
    
    # 计算MACD指标
    if show_macd:
        # 计算12日和26日指数移动平均线
        df['EMA12'] = df['close'].ewm(span=12, adjust=False).mean()
        df['EMA26'] = df['close'].ewm(span=26, adjust=False).mean()
        
        # 计算DIF和DEA
        df['DIF'] = df['EMA12'] - df['EMA26']
        df['DEA'] = df['DIF'].ewm(span=9, adjust=False).mean()
        
        # 计算MACD柱状图
        df['MACD'] = 2 * (df['DIF'] - df['DEA'])
    
    return df

# 绘制K线图和技术指标
def plot_stock_chart(df, stock_code):
    """
    使用plotly绘制股票K线图和技术指标
    """
    # 确定需要多少个子图
    rows = 1
    if show_rsi:
        rows += 1
    if show_macd:
        rows += 1
    if show_volume:
        rows += 1
    
    # 创建子图
    fig = make_subplots(
        rows=rows, cols=1,
        shared_xaxes=True,
        vertical_spacing=0.1,
        subplot_titles=(
            f"{stock_code} K线图",
            "RSI指标" if show_rsi else None,
            "MACD指标" if show_macd else None,
            "成交量" if show_volume else None
        )
    )
    
    # 1. 添加K线图
    fig.add_trace(
        go.Candlestick(
            x=df.index,
            open=df['open'],
            high=df['high'],
            low=df['low'],
            close=df['close'],
            name="K线"
        ),
        row=1, col=1
    )
    
    # 添加均线
    if show_ma:
        colors = ['blue', 'green', 'orange', 'red', 'purple', 'brown']
        for i, period in enumerate(ma_periods):
            color = colors[i % len(colors)]
            fig.add_trace(
                go.Scatter(
                    x=df.index,
                    y=df[f'MA{period}'],
                    name=f"MA{period}",
                    line=dict(color=color, width=1)
                ),
                row=1, col=1
            )
    
    # 2. 添加RSI
    current_row = 2
    if show_rsi:
        fig.add_trace(
            go.Scatter(
                x=df.index,
                y=df['RSI'],
                name="RSI",
                line=dict(color='brown', width=1.5)
            ),
            row=current_row, col=1
        )
        # 添加超买超卖线
        fig.add_hline(y=70, line_dash="dash", line_color="red", opacity=0.7, row=current_row, col=1)
        fig.add_hline(y=30, line_dash="dash", line_color="green", opacity=0.7, row=current_row, col=1)
        current_row += 1
    
    # 3. 添加MACD
    if show_macd:
        fig.add_trace(
            go.Scatter(
                x=df.index,
                y=df['DIF'],
                name="DIF",
                line=dict(color='blue', width=1)
            ),
            row=current_row, col=1
        )
        fig.add_trace(
            go.Scatter(
                x=df.index,
                y=df['DEA'],
                name="DEA",
                line=dict(color='red', width=1)
            ),
            row=current_row, col=1
        )
        fig.add_trace(
            go.Bar(
                x=df.index,
                y=df['MACD'],
                name="MACD",
                marker_color=df['MACD'].apply(lambda x: 'red' if x > 0 else 'green')
            ),
            row=current_row, col=1
        )
        current_row += 1
    
    # 4. 添加成交量
    if show_volume:
        # 根据涨跌设置成交量颜色
        colors = df['close'].diff().apply(lambda x: 'green' if x >= 0 else 'red')
        fig.add_trace(
            go.Bar(
                x=df.index,
                y=df['volume'],
                name="成交量",
                marker_color=colors
            ),
            row=current_row, col=1
        )
    
    # 更新布局
    fig.update_layout(
        height=600 + (150 * (rows - 1)),
        width=1200,
        title_x=0.5,
        xaxis_rangeslider_visible=False,
        hovermode='x unified',
        showlegend=True,
        legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
    )
    
    # 更新X轴
    fig.update_xaxes(
        type='category',
        tickformat='%Y-%m-%d',
        tickangle=45,
        showspikes=True,
        spikemode='across'
    )
    
    return fig

# 显示数据表格
def show_data_table(df):
    """
    显示股票数据表格
    """
    st.subheader("股票数据")
    
    # 选择要显示的列
    columns_to_show = ['open', 'close', 'high', 'low', 'volume', 'pct_change']
    
    # 添加选中的均线列
    if show_ma:
        for period in ma_periods:
            columns_to_show.append(f'MA{period}')
    
    # 添加RSI列
    if show_rsi:
        columns_to_show.append('RSI')
    
    # 添加MACD相关列
    if show_macd:
        columns_to_show.extend(['DIF', 'DEA', 'MACD'])
    
    # 显示数据
    st.dataframe(
        df[columns_to_show].tail(50),
        height=400,
        width='stretch'  # 修复：删除重复的width参数，保留width='stretch'
    )

# 计算并显示基本统计信息
def show_statistics(df):
    """
    显示股票数据的基本统计信息
    """
    st.subheader("基本统计信息")
    
    col1, col2, col3, col4 = st.columns(4)
    
    # 计算统计数据
    with col1:
        st.metric("最新价格", f"¥{df['close'].iloc[-1]:.2f}")
        st.metric("最高价", f"¥{df['high'].max():.2f}")
    
    with col2:
        st.metric("开盘价", f"¥{df['open'].iloc[-1]:.2f}")
        st.metric("最低价", f"¥{df['low'].min():.2f}")
    
    with col3:
        st.metric("成交量", f"{df['volume'].iloc[-1]:,.0f}")
        st.metric("平均成交量", f"{df['volume'].mean():,.0f}")
    
    with col4:
        # 计算涨跌幅
        change = df['close'].iloc[-1] - df['close'].iloc[-2]
        pct_change = df['pct_change'].iloc[-1] if not pd.isna(df['pct_change'].iloc[-1]) else (change / df['close'].iloc[-2] * 100)
        st.metric("涨跌额", f"¥{change:.2f}", f"{pct_change:.2f}%")
        
        # 计算总收益率
        total_return = (df['close'].iloc[-1] / df['close'].iloc[0] - 1) * 100
        st.metric("区间收益率", f"{total_return:.2f}%")

# 主程序
if __name__ == "__main__":
    # 获取数据
    st.info(f"正在获取 {stock_code} 从 {start_date_str} 到 {end_date_str} 的数据...")
    stock_data = get_stock_data(stock_code, start_date_str, end_date_str)
    
    if stock_data is not None and not stock_data.empty:
        st.success(f"成功获取 {len(stock_data)} 条数据")
        
        # 计算技术指标
        stock_data_with_indicators = calculate_technical_indicators(stock_data.copy())
        
        # 显示统计信息
        show_statistics(stock_data_with_indicators)
        
        # 绘制图表
        fig = plot_stock_chart(stock_data_with_indicators, stock_code)
        st.plotly_chart(fig, width='stretch')  # 修复：将use_container_width=True替换为width='stretch'
        
        # 显示数据表格
        show_data_table(stock_data_with_indicators)
        
        # 下载数据
        csv = stock_data_with_indicators.to_csv().encode('utf-8')
        st.download_button(
            label="下载数据 (CSV)",
            data=csv,
            file_name=f"{stock_code}_{start_date_str}_{end_date_str}.csv",
            mime="text/csv",
            help="点击下载当前股票数据的CSV文件"
        )
    else:
        st.error("获取数据失败，请检查股票代码和网络连接")