sem / modules /panel.py
cyj-26's picture
Upload 25 files
26c3195 verified
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)
# Hausman ๊ฒ€์ •
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)}"