{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# cass_analyze.py\n", "\n", "오비맥주 카스Fresh 제품단위 분석 — H1 전가(2020 종량세 전환)·H2 탄력성·H3 세수\n", "자료: (2026_09_22) 연구용역 기초 자료 (오비맥주_카스).xlsx\n", " Sheet1 출고자료: 월별 Volume(L), 순매출(세전), 세후매출(B), 출고단가(세후)=B/A (2015~2025)\n", " Sheet2 출고가격: 월별 500ml 세전원가/주세/교육세/부가세/출고가격(세후) (2015~2025)\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# -*- coding: utf-8 -*-\n", "\"\"\"오비맥주 카스Fresh 제품단위 분석 — H1 전가(2020 종량세 전환)·H2 탄력성·H3 세수\n", "자료: (2026_09_22) 연구용역 기초 자료 (오비맥주_카스).xlsx\n", " Sheet1 출고자료: 월별 Volume(L), 순매출(세전), 세후매출(B), 출고단가(세후)=B/A (2015~2025)\n", " Sheet2 출고가격: 월별 500ml 세전원가/주세/교육세/부가세/출고가격(세후) (2015~2025)\n", "\"\"\"\n", "import sys, io, json\n", "sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')\n", "import numpy as np, pandas as pd, openpyxl\n", "import statsmodels.formula.api as smf\n", "\n", "XL = r'C:/Temp/주류세/(2026_09_22) 연구용역 기초 자료 (오비맥주_카스).xlsx'\n", "OUT = r'C:/Temp/주류세/공개데이터/DART' # keep near other product-price outputs\n", "wb = openpyxl.load_workbook(XL, data_only=True)\n", "ws1, ws2 = wb.worksheets\n", "\n", "def parse(ws, metrics):\n", " \"\"\"year blocks: first row of block has year in col A; col B=metric label; C=unit; D..O = 12 months\"\"\"\n", " rows=[]; year=None\n", " for r in ws.iter_rows(min_row=3, values_only=True):\n", " a,b = r[0], r[1]\n", " if isinstance(a,(int,float)) or (isinstance(a,str) and a.strip().endswith('년')):\n", " try: year=int(str(a).replace('년','').strip())\n", " except: pass\n", " if b is None: continue\n", " lab=str(b).strip()\n", " key=None\n", " for k,pats in metrics.items():\n", " if any(p in lab for p in pats): key=k; break\n", " if key is None or year is None: continue\n", " for m in range(12):\n", " v=r[3+m]\n", " if v is None or v=='' : continue\n", " try: v=float(v)\n", " except: continue\n", " rows.append(dict(year=year, month=m+1, metric=key, value=v))\n", " return pd.DataFrame(rows)\n", "\n", "m1={'volume':['Volume'],'net_pretax':['순매출'],'sales_posttax':['세후매출'],'unit_posttax':['출고단가']}\n", "m2={'cost':['세전원가'],'juse':['주세'],'edu':['교육세'],'vat':['부가세'],'price500':['출고가격']}\n", "d1=parse(ws1,m1); d2=parse(ws2,m2)\n", "w1=d1.pivot_table(index=['year','month'],columns='metric',values='value').reset_index()\n", "w2=d2.pivot_table(index=['year','month'],columns='metric',values='value').reset_index()\n", "df=pd.merge(w1,w2,on=['year','month'],how='outer').sort_values(['year','month']).reset_index(drop=True)\n", "df['ym']=df.year*100+df.month\n", "df.to_csv(OUT+'/cass_monthly.csv',index=False,encoding='utf-8-sig')\n", "print('월별 관측:',len(df),' 기간:',int(df.ym.min()),'~',int(df.ym.max()))\n", "\n", "# ---------- 연간 집계 ----------\n", "ann=df.groupby('year').agg(\n", " vol=('volume','sum'),\n", " sales_post=('sales_posttax','sum'),\n", " price_L=('unit_posttax','mean'),\n", " cost500=('cost','mean'), juse500=('juse','mean'),\n", " edu500=('edu','mean'), vat500=('vat','mean'), price500=('price500','mean'),\n", " n=('month','count')).reset_index()\n", "ann['tax500']=ann.juse500+ann.edu500 # 주세+교육세\n", "ann['taxall500']=ann.tax500+ann.vat500\n", "ann['juse_share']=ann.juse500/ann.price500\n", "ann['juse_rev']=ann.juse500*(ann.vol/0.5)/1e8 # 억원(500ml 환산 수량 × 병당 주세)\n", "print('\\n===== 연간 집계 (500ml 기준 세금분해) =====')\n", "print(ann[['year','vol','price_L','cost500','juse500','edu500','vat500','price500','juse_share','n']]\n", " .round(2).to_string(index=False))\n", "\n", "# ---------- H1: 전가 (2020 종량세 전환 & 가격 개정 이벤트) ----------\n", "print('\\n===== H1 전가: 세전원가·세금·출고가 변화 (연평균 500ml) =====')\n", "chg=ann[['year','cost500','juse500','edu500','vat500','price500']].copy()\n", "for c in ['cost500','juse500','edu500','vat500','price500']:\n", " chg[c+'_d']=chg[c].diff()\n", "print(chg.round(2).to_string(index=False))\n", "\n", "# 종량세 전환(2019->2020): 주세 변화분이 출고가로 전가된 비율\n", "r19=ann[ann.year==2019].iloc[0]; r20=ann[ann.year==2020].iloc[0]\n", "d_tax=(r20.juse500+r20.edu500)-(r19.juse500+r19.edu500)\n", "d_price=r20.price500-r19.price500\n", "d_cost=r20.cost500-r19.cost500\n", "print(f'\\n[2020 종량세 전환] 주세+교육세 변화 {d_tax:+.2f}원/500ml, 세전원가 변화 {d_cost:+.2f}, 출고가(세후) 변화 {d_price:+.2f}')\n", "print(f' → 세금변화의 소비자가 전가율 ρ = Δ출고가/Δ세금 = {d_price/d_tax:+.2f} (원가 흡수분 {d_cost:+.2f})')\n", "\n", "# 2022 원가주도 인상: 세전원가 변화가 출고가로 전가된 비율\n", "r21=ann[ann.year==2021].iloc[0]; r22=ann[ann.year==2022].iloc[0]\n", "dc=r22.cost500-r21.cost500; dp=r22.price500-r21.price500; dt=(r22.juse500+r22.edu500)-(r21.juse500+r21.edu500)\n", "print(f'[2022 인상] 세전원가 {dc:+.2f}, 세금 {dt:+.2f}, 출고가 {dp:+.2f} → 원가전가율 {dp/dc:+.2f}')\n", "\n", "# 월별 이벤트 전가: 출고가 변동을 원가·세금 변동으로 회귀 (Δ)\n", "mo=df.dropna(subset=['cost','juse','edu','price500']).copy().sort_values('ym')\n", "mo['tax']=mo.juse+mo.edu\n", "for c in ['cost','tax','price500','vat']:\n", " mo['d_'+c]=mo[c].diff()\n", "mo=mo[mo.year>=2016]\n", "reg=smf.ols('d_price500 ~ d_cost + d_tax', data=mo).fit(cov_type='HC1')\n", "print('\\n[월별 Δ전가 회귀] Δ출고가 ~ Δ세전원가 + Δ(주세+교육세)')\n", "print(f' 원가 전가계수={reg.params[\"d_cost\"]:.3f}(t={reg.tvalues[\"d_cost\"]:.2f}), '\n", " f'세금 전가계수={reg.params[\"d_tax\"]:.3f}(t={reg.tvalues[\"d_tax\"]:.2f}), R2={reg.rsquared:.2f}')\n", "\n", "# ---------- H2: 가격탄력성 ----------\n", "print('\\n===== H2 가격탄력성 (카스 제품단위) =====')\n", "e=df.dropna(subset=['volume','unit_posttax']).copy()\n", "e['ln_q']=np.log(e.volume); e['ln_p']=np.log(e.unit_posttax)\n", "e['t']=(e.year-2015)*12+e.month; e['t2']=e.t**2\n", "e['covid']=e.year.isin([2020,2021]).astype(int)\n", "e['post2020']=(e.year>=2020).astype(int) # 종량세 체제\n", "# (A) 단순\n", "mA=smf.ols('ln_q ~ ln_p', data=e).fit(cov_type='HC1')\n", "# (B) 추세+계절+코로나 통제\n", "mB=smf.ols('ln_q ~ ln_p + t + t2 + C(month) + covid', data=e).fit(cov_type='HC1')\n", "# (C) 연간(계절 제거)\n", "ea=ann.copy(); ea['ln_q']=np.log(ea.vol); ea['ln_p']=np.log(ea.price_L); ea['t']=ea.year-2015\n", "mC=smf.ols('ln_q ~ ln_p + t', data=ea).fit(cov_type='HC1')\n", "print(f'(A) 무통제 ε={mA.params[\"ln_p\"]:+.2f} (t={mA.tvalues[\"ln_p\"]:.2f})')\n", "print(f'(B) 월+추세+계절+코로나 ε={mB.params[\"ln_p\"]:+.2f} (t={mB.tvalues[\"ln_p\"]:.2f})')\n", "print(f'(C) 연간+추세 ε={mC.params[\"ln_p\"]:+.2f} (t={mC.tvalues[\"ln_p\"]:.2f})')\n", "\n", "# ---------- H3: 카스 주세수입 궤적 ----------\n", "print('\\n===== H3 카스 주세수입(억원, 500ml환산 병당주세×수량) =====')\n", "print(ann[['year','vol','juse500','juse_rev']].round(1).to_string(index=False))\n", "\n", "# 저장\n", "ann.round(3).to_csv(OUT+'/cass_annual.csv',index=False,encoding='utf-8-sig')\n", "summary=dict(\n", " period=f\"{int(df.ym.min())}-{int(df.ym.max())}\",\n", " n_month=len(df),\n", " passthrough_2020_taxconv=round(d_price/d_tax,3),\n", " cost_absorb_2020=round(d_cost,2),\n", " passthrough_2022_cost=round(dp/dc,3),\n", " reg_cost_pt=round(reg.params['d_cost'],3), reg_tax_pt=round(reg.params['d_tax'],3),\n", " elast_simple=round(mA.params['ln_p'],3),\n", " elast_ctrl=round(mB.params['ln_p'],3),\n", " elast_annual=round(mC.params['ln_p'],3),\n", ")\n", "json.dump(summary, open(OUT+'/cass_summary.json','w',encoding='utf-8'), ensure_ascii=False, indent=2)\n", "print('\\n저장: cass_monthly.csv, cass_annual.csv, cass_summary.json')\n", "print(json.dumps(summary,ensure_ascii=False))\n" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python", "version": "3.12" } }, "nbformat": 4, "nbformat_minor": 5 }