ever-flow/visualization_modules

1import streamlit as st2import pandas as pd3import numpy as np4import plotly.graph_objects as go5import re6from typing import List7 8 9def show_growth_pattern(DF_RAW):10    st.header("성장 패턴 분석")11    12    def growth_symbol(x):13        return "▲" if x >= 5 else "▼" if x <= -5 else "▬"14    def make_pattern(r):15        return "".join(growth_symbol(r[f"Growth_LTM-{i}"]) for i in (3,2,1))16    pattern_category_map = {17        "▲▲▲":"Good","▬▲▲":"Good","▲▬▲":"Up","▬▬▲":"Up","▼▲▲":"Turn Up",18        "▼▬▲":"Recent Up","▬▼▲":"Recent Up","▼▼▲":"Recent Up","▲▼▲":"Recent Up",19        "▲▲▬":"Up & Flat","▬▲▬":"Up & Flat","▼▲▬":"Up & Flat","▲▬▬":"Up & Flat",20        "▬▬▬":"Flat",21        "▼▬▬":"Down & Flat","▲▼▬":"Down & Flat","▬▼▬":"Down & Flat","▼▼▬":"Down & Flat",22        "▼▲▼":"Recent Down","▲▲▼":"Recent Down","▬▲▼":"Recent Down","▲▬▼":"Recent Down",23        "▲▼▼":"Turn Down","▬▬▼":"Down","▼▬▼":"Down","▬▼▼":"Bad","▼▼▼":"Bad"24    }25    category_order = ["Good","Up","Turn Up","Recent Up","Up & Flat","Flat","Down & Flat","Recent Down","Turn Down","Down","Bad","기타"]26    growth_weights = {"Good":3,"Up":2,"Turn Up":1.5,"Recent Up":1,"Up & Flat":0.5,"Flat":0,"Down & Flat":-0.5,"Recent Down":-1,"Turn Down":-1.5,"Down":-2,"Bad":-3,"기타":0}27    color_map = dict(zip(category_order, [(255,0,0),(255,64,48),(255,128,64),(255,192,128),(192,224,192),(128,128,128),(192,160,224),(224,160,192),(224,128,128),(192,64,64),(0,0,255),(192,192,192)]))28 29    with st.sidebar:30        market_sel = st.selectbox("상장시장", ["한국 전체","KOSPI","KOSDAQ","미국 전체","NASDAQ","일본 전체","Prime (Domestic Stocks)","Standard (Domestic Stocks)","Prime (Foreign Stocks)"], key="market_sel3")31        if market_sel in ["한국 전체","KOSPI","KOSDAQ"]:32            country_filter = "한국"33        elif market_sel in ["미국 전체","NASDAQ"]:34            country_filter = "미국"35        elif market_sel in ["일본 전체","Prime (Domestic Stocks)","Standard (Domestic Stocks)","Prime (Foreign Stocks)"]:36            country_filter = "일본"37        else:38            country_filter = "Unclassified"39        market_filter = None if market_sel.endswith("전체") else market_sel40        class_type = st.radio("분류 체계", ["EMSEC","EMTEC"], horizontal=True, key="class_type3")41        if class_type == "EMSEC":42            sectors = sorted(DF_RAW["Sector"].dropna().unique())43            sector_sel = st.selectbox("Sector", ["전체"] + sectors, key="sector_sel3")44            if sector_sel != "전체":45                industries = sorted(DF_RAW.loc[DF_RAW["Sector"]==sector_sel, "Industry"].dropna().unique())46                industry_sel = st.selectbox("Industry", ["전체"] + industries, key="industry_sel3")47            else:48                industry_sel = "전체"49            row_level = "Sector" if sector_sel=="전체" else "Industry"50        else:51            themes = sorted(DF_RAW["Theme"].dropna().unique())52            theme_sel = st.selectbox("Theme", ["전체"] + themes, key="theme_sel3")53            if theme_sel != "전체":54                techs = sorted(DF_RAW.loc[DF_RAW["Theme"]==theme_sel, "Technology"].dropna().unique())55                tech_sel = st.selectbox("Technology", ["전체"] + techs, key="tech_sel3")56            else:57                tech_sel = "전체"58            row_level = "Theme" if theme_sel=="전체" else "Technology"59        metric_sel = st.selectbox("지표", ["Net Income","EBITDA","EBIT","Revenue"], key="metric_sel3")60 61    def get_cols(df: pd.DataFrame, base: str) -> List[str]:62        pat = re.compile(re.sub(r"\s+","",base) + r"\s*\(LTM(?:-\d)?\)_USD", re.I)63        cols = [c for c in df.columns if pat.search(re.sub(r"\s+","",c))]64        if len(cols) < 4:65            expected_cols = [f"{base} (LTM-{i})_USD" for i in range(0, 4)]66            st.warning(f"{base} 기준 LTM~LTM-3 열이 충분하지 않습니다: {cols}. 기본값으로 대체합니다.")67            return expected_cols[:4]68        def keyfn(c):69            m = re.search(r"LTM-(\d)", c)70            return int(m.group(1)) if m else -171        return sorted(cols, key=keyfn)[:4]72 73    def calc_growth(sub: pd.DataFrame, cols: List[str]) -> pd.DataFrame:74        g = pd.DataFrame({"ticker": sub["ticker"]})75        for i in (1,2,3):76            if cols[i-1] in sub.columns and cols[i] in sub.columns:77                g[f"Growth_LTM-{i}"] = (sub[cols[i-1]] - sub[cols[i]]) / sub[cols[i]].replace(0, np.nan) * 10078            else:79                g[f"Growth_LTM-{i}"] = np.nan80        return g.replace([np.inf,-np.inf], np.nan).dropna()81 82    def build_growth(metric: str, df: pd.DataFrame) -> pd.DataFrame:83        d = df.copy()84        if metric == "EBITDA":85            ecols = get_cols(d, "EBIT")86            dcols = get_cols(d, "Depreciation")87            for i, (e, dcol) in enumerate(zip(ecols, dcols)):88                d[f"EBITDA_{i}"] = d.get(e, 0) + d.get(dcol, 0)89            return calc_growth(d, [f"EBITDA_{i}" for i in range(4)])90        base_map = {"Net Income": "Net Income", "EBIT": "EBIT", "Revenue": "Revenue"}91        cols = get_cols(d, base_map[metric])92        return calc_growth(d, cols)93 94    DF = DF_RAW[(DF_RAW["Country"] == country_filter) & ((market_filter is None) | (DF_RAW["Market"] == market_filter))].copy()95    if class_type == "EMSEC":96        if sector_sel != "전체": DF = DF[DF["Sector"] == sector_sel]97        if industry_sel != "전체": DF = DF[DF["Industry"] == industry_sel]98    else:99        if theme_sel != "전체": DF = DF[DF["Theme"] == theme_sel]100        if tech_sel != "전체": DF = DF[DF["Technology"] == tech_sel]101    if DF.empty:102        st.warning("조건에 맞는 데이터가 없습니다.")103        st.stop()104 105    growth_df = build_growth(metric_sel, DF)106    rows_df = DF[[row_level,"ticker"]].drop_duplicates()107    merged = rows_df.merge(growth_df, on="ticker").dropna()108    if merged.empty:109        st.warning("해당 조건·지표에서 성장률 데이터가 없습니다.")110        st.stop()111 112    merged["패턴"] = merged.apply(make_pattern, axis=1)113    merged["카테고리"] = merged["패턴"].map(pattern_category_map).fillna("기타")114    ratio = (merged.groupby([row_level,"카테고리"])["ticker"].nunique().reset_index(name="count").pipe(lambda x: x.merge(merged.groupby(row_level)["ticker"].nunique().reset_index(name="total"), on=row_level)))115    ratio["rate"] = ratio["count"] / ratio["total"]116    pv = ratio.pivot(index=row_level, columns="카테고리", values="rate").fillna(0)117    pv = pv[[c for c in category_order if c in pv.columns]]118    scores = pv.mul([growth_weights[c] for c in pv.columns], axis=1).sum(axis=1)119    pv = pv.loc[scores.sort_values(ascending=False).index]120 121    bars = [go.Bar(y=pv.index, x=pv[cat], name=cat, orientation="h", marker_color=f"rgb{color_map[cat]}") for cat in pv.columns]122    fig = go.Figure(bars).update_layout(123        barmode="stack",124        height=max(600, 40*len(pv)),125        title=f"{metric_sel} 3년 성장 패턴 ({row_level})",126        xaxis_title="비율",127        yaxis_title=row_level,128        legend_title="카테고리",129        xaxis=dict(tickformat=".0%")130    )131    st.plotly_chart(fig, use_container_width=True)132    st.caption(f"기업 수: {merged['ticker'].nunique():,} | 행 레벨: {row_level}")133    with st.expander("📋 비율 테이블"):134        st.dataframe(pv.style.format("{:.1%}"), use_container_width=True)135 136