【Python技术】统计各类ETF在历年月份表现

import streamlit as st
import akshare as ak
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt
import matplotlib.colors as mcolors
from datetime import datetime
# 设置全局样式
plt.rcParams['font.sans-serif'] = ['STHeiti']  # 苹果系统字体
plt.rcParams['axes.unicode_minus'] = False
# 自定义双色系
dual_colors = ['#4CB050', '#FF3333']  # 绿跌红涨
cmap = mcolors.ListedColormap(dual_colors)
bounds = [-100, 0, 100]
norm = mcolors.BoundaryNorm(bounds, cmap.N)
# 常见ETF映射表
ETF_MAP = {
    '上证指数':('000001.SH', '上证指数'),
    '银行ETF': ('512800.SH', '华宝中证银行ETF'),
    '人工智能ETF': ('515980.SH', '华富中证人工智能产业ETF'),
    '电力ETF': ('561560.SH', '华泰柏瑞中证电力公用事业ETF'),
    '沪深300ETF': ('510300.SH', '华泰柏瑞沪深300ETF'),
    '黄金ETF': ('518880.SH', '华安黄金易ETF')
}
# 数据获取
@st.cache_data(ttl=3600)
def get_etf_data(symbol):
    try:
        df = ak.fund_etf_fund_info_em(fund=symbol[:6])  # 提取纯数字代码
        df = df.rename(columns={
            '净值日期': 'date',
            '单位净值': 'close',
            '日增长率': 'change'
        })
        df['date'] = pd.to_datetime(df['date'])
        df['year'] = df['date'].dt.year
        df['month'] = df['date'].dt.month
        return df.sort_values('date')
    except Exception as e:
        st.error(f"数据获取失败：{str(e)}")
        return pd.DataFrame()
# 构建月度数据（增加容错机制）
def build_monthly_table(df):
    monthly = []
    for y in range(2015, 2025):
        row = {'年份': y}
        for m in range(1, 13):
            sub = df[(df['year'] == y) & (df['month'] == m)]
            if len(sub) >= 3:
                try:
                    chg = (sub.iloc[-1]['close'] / sub.iloc[0]['close'] - 1) * 100
                    row[f'{m}月'] = round(chg, 2)
                except KeyError:
                    row[f'{m}月'] = None
            else:
                row[f'{m}月'] = None
        monthly.append(row)
    return pd.DataFrame(monthly).set_index('年份')
# 生成热力图
def plot_heatmap(data):
    fig, ax = plt.subplots(figsize=(16, 6))
    sns.heatmap(data,
                cmap=cmap,
                norm=norm,
                annot=True,
                fmt=".1f",
                linewidths=0.5,
                cbar=False,
                annot_kws={'color': 'white', 'weight': 'bold'})
    ax.set_xticklabels(['1月', '2月', '3月', '4月', '5月', '6月',
                        '7月', '8月', '9月', '10月', '11月', '12月'])
    plt.title('月度涨跌幅分布', fontsize=14, pad=20)
    plt.tight_layout()
    return fig
def app():
    # 页面配置
    st.title("📊 行业ETF历史数据分析")
    st.subheader("参数设置")
    selected_name = st.selectbox(
        "选择行业ETF",
        options=list(ETF_MAP.keys()),
        index=0,
        help="选择要分析的行业ETF品种"
    )
    symbol, full_name = ETF_MAP[selected_name]
    # 显示基本信息
    st.markdown(f"""
    **ETF详情**  
    ▸ 代码：`{symbol}`  
    ▸ 全称：{full_name}  
    ▸ 数据范围：2015-2024年
    """)
    # 数据加载与处理
    df = get_etf_data(symbol)
    if not df.empty:
        # 展示关键指标
        col1, col2, col3 = st.columns(3)
        with col1:
            st.metric("最新净值", f"{df.iloc[-1]['close']:.3f}")
        with col2:
            st.metric("历史最高", f"{df['close'].max():.3f}")
        with col3:
            st.metric("数据日期", df['date'].max().strftime("%Y-%m-%d"))
        monthly_df = build_monthly_table(df)
        # 数据表格样式优化
        styled_df = monthly_df.style.format('{:.1f}%', na_rep="-").applymap(
            lambda val: f'background-color: {dual_colors[1] if val >= 0 else dual_colors[0]}; color: white'
        )
        st.dataframe(
            styled_df,
            height=600,
            use_container_width=True,
            column_config={
                "年份": st.column_config.NumberColumn(format="%d")
            }
        )
    else:
        st.warning("未获取到有效数据，请检查ETF代码或网络连接")
if __name__ == "__main__":
    app()