ever-flow/visualization_modules

1import streamlit as st2import pandas as pd3import numpy as np4import plotly.graph_objects as go5from typing import List6 7 8def show_scale_heatmap(DF_RAW):9    st.header("규모 변수 Heatmap")10    11    with st.sidebar:12        year_sel = st.selectbox("기준 연도", ["LTM","LTM-1","LTM-2","LTM-3"], key="year_sel4")13        market_sel = st.selectbox("상장시장", ["한국 전체","KOSPI","KOSDAQ","미국 전체","NASDAQ","일본 전체","Prime (Domestic Stocks)","Standard (Domestic Stocks)","Prime (Foreign Stocks)"], key="market_sel4")14        country_filter = market_filter = None15        if "전체" in market_sel:16            country_filter = market_sel.split()[0]17        else:18            country_filter = "한국" if market_sel in ("KOSPI","KOSDAQ") else "미국" if market_sel == "NASDAQ" else "일본"19            market_filter = market_sel20        class_type = st.radio("분류 체계", ["EMSEC","EMTEC"], horizontal=True, key="class_type4")21        if class_type == "EMSEC":22            sectors = sorted([s for s in DF_RAW.Sector.dropna().unique() if s != 'Unclassified'])23            sector_sel = st.selectbox("Sector", ["전체"] + sectors, key="sector_sel4")24            if sector_sel != "전체":25                indus = sorted(DF_RAW.loc[DF_RAW.Sector == sector_sel, "Industry"].dropna().unique())26                industry_sel = st.selectbox("Industry", ["전체"] + indus, key="industry_sel4")27            else:28                industry_sel = "전체"29            row_level = "Sector" if sector_sel == "전체" else "Industry" if industry_sel == "전체" else "Sub_industry"30        else:31            themes = sorted([t for t in DF_RAW.Theme.dropna().unique() if t != 'Unclassified'])32            theme_sel = st.selectbox("Theme", ["전체"] + themes, key="theme_sel4")33            if theme_sel != "전체":34                techs = sorted(DF_RAW.loc[DF_RAW.Theme == theme_sel, "Technology"].dropna().unique())35                tech_sel = st.selectbox("Technology", ["전체"] + techs, key="tech_sel4")36            else:37                tech_sel = "전체"38            row_level = "Theme" if theme_sel == "전체" else "Technology" if tech_sel == "전체" else "Sub_Technology"39        metric_base = {"시가총액": "Market Cap (2024-12-31)_USD", "자산총계": "Assets", "매출액": "Sales"}40        metric_name = st.selectbox("계측값", list(metric_base.keys()), key="metric_name4")41        metric_col = metric_base[metric_name]42 43    DF = DF_RAW[DF_RAW.Year == year_sel].copy()44    if country_filter: DF = DF[DF.Country == country_filter]45    if market_filter: DF = DF[DF.Market == market_filter]46    if class_type == "EMSEC":47        if sector_sel != "전체": DF = DF[DF.Sector == sector_sel]48        if industry_sel != "전체": DF = DF[DF.Industry == industry_sel]49    else:50        if theme_sel != "전체": DF = DF[DF.Theme == theme_sel]51        if tech_sel != "전체": DF = DF[DF.Technology == tech_sel]52    if metric_col not in DF.columns:53        st.error(f"'{metric_name}' 열이 없습니다. 데이터 파일을 확인해주세요.")54        st.stop()55    DF["metric_bil"] = DF[metric_col] / 1e956    DF = DF[DF["metric_bil"].notna()]57    DF = DF[DF["metric_bil"] >= 0]58    if not DF.empty:59        max_th = DF["metric_bil"].quantile(0.999)60        DF = DF[DF["metric_bil"] <= max_th]61    if DF.empty:62        st.warning("조건에 맞는 데이터가 없습니다.")63        st.stop()64 65    country = DF['Country'].unique()[0] if len(DF['Country'].unique()) == 1 else 'Unclassified'66    currency = {'한국': 'KRW', '미국': 'USD', '일본': 'JPY', 'Unclassified': 'USD'}.get(country, 'USD')67 68    valid_vals = DF["metric_bil"]69    vl_max = valid_vals.max() if not valid_vals.empty else 070    def make_edges(max_val: float) -> List[float]:71        base = [10,30,60,100,300,600]72        edges = [0]73        if max_val <= 0:74            edges += base[:1]75        else:76            exp = 077            while True:78                factor = 10 ** exp79                for b in base:80                    edge = b * factor81                    if edge > max_val:82                        edges = sorted(set(edges))83                        return edges + [np.inf]84                    edges.append(edge)85                exp += 186        edges = sorted(set(edges))87        return edges + [np.inf]88    bin_edges = make_edges(vl_max)89    bin_labels = ["0~"] + [f"{int(e):,}~" for e in bin_edges[1:-1]]90    DF["metric_bin"] = pd.cut(DF["metric_bil"], bins=bin_edges, labels=bin_labels, right=False)91 92    pivot = DF.groupby([row_level,"metric_bin"])["Company"].nunique().unstack(fill_value=0).reindex(columns=bin_labels, fill_value=0).astype(int)93    if pivot.empty:94        st.warning("조건에 맞는 데이터가 없어 집계표를 생성할 수 없습니다.")95        st.stop()96    subtotal = pd.DataFrame(pivot.sum()).T97    subtotal.index = ["Subtotal"]98    pivot_full = pd.concat([subtotal, pivot])99 100    rows = pivot_full.index.tolist()101    cols = pivot_full.columns.tolist()102    z_data = pivot_full.values103    z_max = np.nanmax(z_data)104 105    fig = go.Figure()106    fig.add_trace(go.Heatmap(107        z=z_data,108        x=cols,109        y=rows,110        colorscale="Greens",111        colorbar=dict(title="기업 수"),112        hovertemplate="%{y}/%{x}<br>기업수: %{z:,}<extra></extra>",113        xgap=1,114        ygap=1115    ))116    annotations = []117    for i, row in enumerate(rows):118        for j, col in enumerate(cols):119            val = z_data[i, j]120            if val > 0:121                color = "white" if val > z_max * 0.5 else "black"122                annotations.append(123                    dict(124                        x=col,125                        y=row,126                        text=f"{int(val):,}",127                        showarrow=False,128                        font=dict(color=color, size=12)129                    )130                )131    fig.update_layout(132        annotations=annotations,133        height=max(600, 40*len(rows)),134        title=f"{metric_name} 분포 ({row_level}, {currency} billion)",135        xaxis_title="규모 범위",136        yaxis_title=row_level,137        xaxis=dict(side="top"),138        yaxis=dict(autorange="reversed")139    )140    st.plotly_chart(fig, use_container_width=True)141    st.caption(f"기업 수: {DF['Company'].nunique():,} | 통화: {currency}")142    with st.expander("📋 원본 집계표 보기", False):143        st.dataframe(pivot_full, use_container_width=True)144