| import pandas as pd |
| import numpy as np |
| from modules.utils import fmt_p, sig_stars |
|
|
|
|
| def run_panel(df, entity_col, time_col, dep_col, indep_cols): |
| try: |
| from linearmodels.panel import PanelOLS, RandomEffects, PooledOLS |
| except ImportError: |
| return None, "linearmodels ํจํค์ง๊ฐ ํ์ํฉ๋๋ค: pip install linearmodels" |
|
|
| try: |
| data = df[[entity_col, time_col, dep_col] + indep_cols].dropna().copy() |
|
|
| |
| try: |
| data[time_col] = pd.to_datetime(data[time_col]) |
| except Exception: |
| data[time_col] = pd.to_numeric(data[time_col], errors="coerce") |
|
|
| data = data.set_index([entity_col, time_col]) |
| dep = data[dep_col] |
| exog = data[indep_cols] |
|
|
| |
| fe = PanelOLS(dep, exog, entity_effects=True).fit( |
| cov_type="clustered", cluster_entity=True) |
| re = RandomEffects(dep, exog).fit() |
| po = PooledOLS(dep, exog).fit() |
|
|
| def extract(res, label): |
| rows = [] |
| for var in indep_cols: |
| try: |
| rows.append({ |
| "๋ณ์": var, "๋ชจํ": label, |
| "๊ณ์(ฮฒ)": round(float(res.params[var]), 4), |
| "ํ์ค์ค์ฐจ": round(float(res.std_errors[var]), 4), |
| "t/z๊ฐ": round(float(res.tstats[var]), 3), |
| "p๊ฐ": fmt_p(float(res.pvalues[var])), |
| "์ ์์ฑ": sig_stars(float(res.pvalues[var])), |
| }) |
| except Exception: |
| pass |
| return pd.DataFrame(rows) |
|
|
| coef_all = pd.concat([extract(fe,"๊ณ ์ ํจ๊ณผ(FE)"), |
| extract(re,"ํ๋ฅ ํจ๊ณผ(RE)"), |
| extract(po,"Pooled OLS")], ignore_index=True) |
|
|
| |
| try: |
| from linearmodels.panel import hausman |
| h_stat, h_p = hausman(fe, re) |
| hausman_df = pd.DataFrame([{ |
| "๊ฒ์ ํต๊ณ๋": round(float(h_stat), 3), |
| "p๊ฐ": fmt_p(float(h_p)), |
| "๊ฒฐ๋ก ": "๊ณ ์ ํจ๊ณผ ๋ชจํ ์ ํ ๊ถ์ฅ (p<.05)" if float(h_p) < 0.05 |
| else "ํ๋ฅ ํจ๊ณผ ๋ชจํ ์ ํ ๊ฐ๋ฅ (pโฅ.05)" |
| }]) |
| except Exception: |
| hausman_df = pd.DataFrame([{"์๋ด": "Hausman ๊ฒ์ ๋ถ๊ฐ (linearmodels ๋ฒ์ ํ์ธ ํ์)"}]) |
|
|
| |
| def safe_r2(res): |
| try: return round(float(res.rsquared), 3) |
| except: return "-" |
|
|
| fit_df = pd.DataFrame([ |
| {"๋ชจํ": "๊ณ ์ ํจ๊ณผ(FE)", "Rยฒ": safe_r2(fe), "๊ด์ธก์": int(fe.nobs)}, |
| {"๋ชจํ": "ํ๋ฅ ํจ๊ณผ(RE)", "Rยฒ": safe_r2(re), "๊ด์ธก์": int(re.nobs)}, |
| {"๋ชจํ": "Pooled OLS", "Rยฒ": safe_r2(po), "๊ด์ธก์": int(po.nobs)}, |
| ]) |
|
|
| return {"๊ณ์๋น๊ต": coef_all, "๋ชจํ์ ํฉ๋": fit_df, "Hausman๊ฒ์ ": hausman_df}, None |
|
|
| except Exception as e: |
| return None, f"ํจ๋๋ถ์ ์ค๋ฅ: {str(e)}" |
|
|