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

import streamlit as stimport akshare as akimport pandas as pdimport seaborn as snsimport matplotlib.pyplot as pltimport matplotlib.colors as mcolorsfrom 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 figdef 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()