A股主要指数涨跌幅趋势分析简单例子

import streamlit as st
import tushare as ts
import pandas as pd
import plotly.graph_objects as go
from datetime import datetime, timedelta
# 初始化Tushare（替换为你的token）
ts.set_token("替换为你的token")
pro = ts.pro_api()
# 指数代码映射表
INDEX_MAP = {
    "上证指数": "000001.SH",
    "深证成指": "399001.SZ",
    "创业板指": "399006.SZ",
    "科创50": "000688.SH",
    "沪深300": "000300.SH",
    "中证500": "000905.SH",
    "上证50": "000016.SH"
}
# 获取指数数据（带缓存）
@st.cache_data(ttl=3600)
def get_index_data(ts_code, days=180):
    end_date = datetime.now().strftime("%Y%m%d")
    start_date = (datetime.now() - timedelta(days=days)).strftime("%Y%m%d")
    df = pro.index_daily(ts_code=ts_code, start_date=start_date, end_date=end_date)
    # 数据预处理
    if not df.empty:
        df["trade_date"] = pd.to_datetime(df["trade_date"])
        df = df.sort_values("trade_date").set_index("trade_date")
        # 计算移动平均线（修复MA5错误）
        df["MA5"] = df["close"].rolling(window=5, min_periods=1).mean()
        df["MA20"] = df["close"].rolling(window=20, min_periods=1).mean()
    return df
# 绘制K线图函数
def plot_kline(df, title):
    fig = go.Figure(data=[go.Candlestick(
        x=df.index,
        open=df["open"],
        high=df["high"],
        low=df["low"],
        close=df["close"],
        increasing_line_color="red",
        decreasing_line_color="green",
        name="K线"
    )])
    # 添加均线
    fig.add_trace(go.Scatter(
        x=df.index, y=df["MA5"],
        line=dict(color="orange", width=1.5),
        name="5日均线"
    ))
    fig.add_trace(go.Scatter(
        x=df.index, y=df["MA20"],
        line=dict(color="purple", width=1.5),
        name="20日均线"
    ))
    # 设置布局
    fig.update_layout(
        title=f"{title} K线图 (最新价: {df['close'].iloc[-1]})",
        xaxis_title="日期",
        yaxis_title="价格",
        template="plotly_white",
        hovermode="x unified",
        height=600
    )
    return fig
# 趋势分析函数
def analyze_trend(df):
    if len(df) < 5:
        return "➡ 数据不足", "gray"
    # 计算均线状态
    ma5_today = df["MA5"].iloc[-1]
    ma20_today = df["MA20"].iloc[-1]
    # 判断趋势
    if ma5_today > ma20_today:
        return "↑ 上升趋势", "red"
    elif ma5_today < ma20_today:
        return "↓ 下降趋势", "green"
    else:
        return "➡ 震荡趋势", "gray"
# -------------------- 页面布局 --------------------
st.set_page_config(page_title="指数监控", layout="wide")
st.title("📊 A股主要指数监控面板")
# 1. 实时涨跌幅仪表盘
st.subheader("当日涨跌幅概览")
cols = st.columns(len(INDEX_MAP))
for i, (name, code) in enumerate(INDEX_MAP.items()):
    try:
        # 获取最新数据
        df = get_index_data(code, days=5)
        if not df.empty and len(df) >= 2:
            last_close = df["close"].iloc[-1]
            prev_close = df["close"].iloc[-2]
            pct_chg = (last_close - prev_close) / prev_close * 100
            # 设置颜色
            color = "red" if pct_chg > 0 else "green"
            with cols[i]:
                st.metric(
                    label=name,
                    value=f"{last_close:.2f}",
                    # delta=f"{pct_chg:.2f}%",
                    delta_color="normal"
                )
                st.markdown(f"<span style='color:{color};font-weight:bold'>"
                            f"{'↑' if pct_chg > 0 else '↓'} {pct_chg:.2f}%</span>",
                            unsafe_allow_html=True)
        else:
            cols[i].warning("数据获取失败")
    except Exception as e:
        cols[i].error(f"错误: {str(e)}")
# 分隔线
st.divider()
# 2. K线图展示区
st.subheader("指数K线趋势分析")
# 侧边栏控制
with st.sidebar:
    st.header("参数设置")
    selected_index = st.selectbox("选择指数", list(INDEX_MAP.keys()))
    analysis_days = st.slider("K线图周期(天)", 30, 365, 180)
    show_volume = st.checkbox("显示成交量", True)
# 获取并展示选定指数数据
try:
    ts_code = INDEX_MAP[selected_index]
    df = get_index_data(ts_code, days=analysis_days)
    if not df.empty:
        # 趋势分析结论
        trend, color = analyze_trend(df)
        st.markdown(f"**趋势方向**: <span style='color:{color};font-weight:bold'>{trend}</span>",
                    unsafe_allow_html=True)
        # 绘制K线图
        st.plotly_chart(plot_kline(df, selected_index), use_container_width=True)
        # 显示成交量
        if show_volume:
            vol_fig = go.Figure()
            vol_fig.add_trace(go.Bar(
                x=df.index,
                y=df["vol"],
                name="成交量",
                marker_color="rgba(100, 149, 237, 0.6)"
            ))
            vol_fig.update_layout(
                title=f"{selected_index} 成交量",
                xaxis_title="日期",
                yaxis_title="成交量(手)",
                height=300
            )
            st.plotly_chart(vol_fig, use_container_width=True)
except Exception as e:
    st.error(f"数据加载失败: {str(e)}")