import streamlit as stimport akshare as akimport pandas as pdimport seaborn as snsimport matplotlib.pyplot as pltimport matplotlib.colors as mcolorsfrom datetime import datetimeplt.rcParams['font.sans-serif'] = ['STHeiti'] plt.rcParams['axes.unicode_minus'] = Falsedual_colors = ['#4CB050', '#FF3333'] cmap = mcolors.ListedColormap(dual_colors)bounds = [-100, 0, 100]norm = mcolors.BoundaryNorm(bounds, cmap.N)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()