# -*- coding: utf-8 -*- """AI 확산 시나리오 산업별 실질부가가치 전망 — 계산 함수 모음. 노트북 01(입수)·02(분석)·03(결과)이 이 모듈의 함수만 호출한다. 경로·상수는 config.py에 둔다. 계산 구조 1. 설문 신호 : AI 수용도조사 통계표 E4-1(성장)·E5-1(고용), 9점 척도, 21개 업종 × 4시점 2. 산업 배분 : 21→62 크로스워크로 62개 산업에 점수 부여. 중심 방식 A1: 강도 a_i = 점수_i(2040) − 1(척도 최저점 '매우 낮음') (E4-1은 영향의 '정도'를 묻는 척도이므로 최저점만이 '영향 최소'라는 실질적 의미를 가진다. 다른 기준점은 allocation_variants) 3. 시점 램프 : 전체 응답 평균 E4-1 궤적을 2025=0, 2040=1로 정규화한 r(t) 4. 크기 Λ : 2035년 실질부가가치 가중평균 프리미엄이 목표값(중심 0.5%p)이 되도록 배율 k를 보정 5. 산업 경로 : VA_AI(i,t) = VA_AI(i,t−1) × [VA_기준(i,t)/VA_기준(i,t−1)] × (1 + p_i,t/100), p_i,t = k·a_i·r(t) 6. 취업계수 : AI 확산에 따른 취업계수 연변화(%p) = (ε − 1)·p_i,t. 중심은 고용 중립(ε=0, 추가 부가가치가 고용을 늘리지 않음). 설문 E4·E5(2024년 대비 2040년 매출·고용 증감 구간)로 구한 업종별 탄력성 ε_i = ln(1+ΔL)/ln(1+ΔY)는 대부분 0 이하이므로 고용 중립이 고용 측 상한이며, ε_i를 그대로 쓴 값을 하방 민감도로 함께 낸다. 7. 정비 전후 : scenario_summary로 같은 AI 확산 층을 정비판·이전 판 기준선에 각각 적용해 핵심 수치를 비교한다(계산 방식은 1–6과 같다). """ from __future__ import annotations import re import numpy as np import openpyxl import pandas as pd TIME_POINTS = [2025, 2029, 2034, 2040] # E4·E5 상세(18구간) 중간값(%): ①감소 50% 이상 … ⑥감소 0~9%, ⑦변동없음, ⑧증가 0~9% … ⑱증가 100% 이상(110으로 둔다) BRACKET_MID = [-55, -45, -35, -25, -15, -5, 0, 5, 15, 25, 35, 45, 55, 65, 75, 85, 95, 110] MIN_SALES_GROWTH = 5.0 # 기대 매출 증가율이 이보다 작으면 탄력성 비율이 불안정하므로 제조·비제조 평균을 쓴다 NEUTRAL = 5.0 # 9점 척도의 중간값 ⑤ "보통"(①매우 낮음 ~ ⑨매우 높음, 영향의 정도) # ── 1. 입수: 설문 통계표 추출 ──────────────────────────────────────────── def _find_title_row(ws, code: str) -> int: """'E4-1.'처럼 문항 코드를 포함한 표 제목 행 번호를 찾는다.""" for r in range(1, ws.max_row + 1): v = ws.cell(r, 1).value if isinstance(v, str) and f"{code}." in v and "【" in v: return r raise ValueError(f"통계표에서 {code} 표를 찾지 못했다.") def _read_block(ws, title_row: int) -> dict[str, list[float]]: """제목 행 아래 블록에서 전체·제조업·비제조업·21개 업종의 4시점 평균 점수를 읽는다.""" out: dict[str, list[float]] = {} r = title_row + 1 while r <= ws.max_row: c1, c2, c4 = (ws.cell(r, c).value for c in (1, 2, 4)) if isinstance(c1, str) and "【" in c1: # 다음 표 제목에서 멈춘다 break vals = [ws.cell(r, c).value for c in range(6, 10)] if isinstance(c1, str) and c1.replace(" ", "") == "전체": out["전체"] = vals elif c2 in ("제조업", "비제조업") and c4 in (None, c2): out.setdefault(c2, vals) elif isinstance(c4, str) and re.match(r"^\d{2}\.", c4.strip()): out.setdefault(c4.strip(), vals) r += 1 return out def read_survey_signal(xlsx_path: str) -> pd.DataFrame: """통계표(xlsx)에서 E4-1(성장)·E5-1(고용) 4시점 평균 점수를 표로 반환한다. 반환: 행=업종(21개 + 전체·제조업·비제조업), 열=AI성장_2025 … AI고용_2040 """ ws = openpyxl.load_workbook(xlsx_path, data_only=True)["통계표"] g = _read_block(ws, _find_title_row(ws, "E4-1")) e = _read_block(ws, _find_title_row(ws, "E5-1")) rows = [] for key in g: row = {"업종": key} row.update({f"AI성장_{t}": float(v) for t, v in zip(TIME_POINTS, g[key])}) row.update({f"AI고용_{t}": float(v) for t, v in zip(TIME_POINTS, e[key])}) rows.append(row) df = pd.DataFrame(rows).set_index("업종") n_detail = sum(bool(re.match(r"^\d{2}\.", k)) for k in df.index) if n_detail != 21: raise ValueError(f"21개 업종이어야 하는데 {n_detail}개를 읽었다.") return df def _read_bracket_block(ws, title_row: int) -> dict[str, tuple[float, int]]: """E4·E5 상세 표: 업종별 기대 증감률(구간 중간값의 가중평균, %)과 응답 수.""" out: dict[str, tuple[float, int]] = {} r = title_row + 3 while r <= ws.max_row: c1, c2, c4 = (ws.cell(r, c).value for c in (1, 2, 4)) if isinstance(c1, str) and "【" in c1: break freq = [ws.cell(r, 6 + 2 * k).value or 0 for k in range(18)] n = sum(freq) key = None if isinstance(c1, str) and c1.replace(" ", "") == "전체": key = "전체" elif isinstance(c4, str) and re.match(r"^\d{2}\.", c4.strip()): key = c4.strip() elif c2 in ("제조업", "비제조업") and c4 in (None, c2): key = c2 if key and key not in out and n > 0: out[key] = (sum(m * q for m, q in zip(BRACKET_MID, freq)) / n, int(n)) r += 1 return out def read_change_expectations(xlsx_path: str, year: int = 2040) -> pd.DataFrame: """통계표 E4·E5 상세(2024년 대비 {year}년 매출·고용 증감 구간)에서 업종별 기대 증감률(%)을 읽는다.""" ws = openpyxl.load_workbook(xlsx_path, data_only=True)["통계표"] def title(prefix): for r in range(1, ws.max_row + 1): v = ws.cell(r, 1).value if isinstance(v, str) and prefix in v and f"(상세)_{year}년" in v: return r raise ValueError(f"통계표에서 {prefix} 상세 {year}년 표를 찾지 못했다.") y = _read_bracket_block(ws, title("E4. 2024년 대비 매출규모 전망")) l = _read_bracket_block(ws, title("E5. 2024년 대비 고용규모 전망")) rows = [{"업종": k, "응답수": y[k][1], "기대매출증감_%": y[k][0], "기대고용증감_%": l[k][0]} for k in y if k in l] return pd.DataFrame(rows).set_index("업종") def employment_elasticity(exp: pd.DataFrame) -> pd.DataFrame: """업종별 고용 탄력성 ε = ln(1+ΔL)/ln(1+ΔY). 기대 매출 증가율이 MIN_SALES_GROWTH(%) 미만이면 제조·비제조 평균으로 대체한다.""" e = exp.copy() raw = np.log1p(e["기대고용증감_%"] / 100) / np.log1p(e["기대매출증감_%"] / 100) e["탄력성_원값"] = raw.where(e["기대매출증감_%"].abs() > 1e-9) grp = {k: ("제조업" if re.match(r"^0\d\.", k) else "비제조업") for k in e.index if re.match(r"^\d{2}\.", k)} eps, src = [], [] for k in e.index: if k in grp and e.loc[k, "기대매출증감_%"] < MIN_SALES_GROWTH: eps.append(float(e.loc[grp[k], "탄력성_원값"])); src.append(f"{grp[k]} 평균(매출 증가 {MIN_SALES_GROWTH:.0f}% 미만)") else: eps.append(float(e.loc[k, "탄력성_원값"])); src.append("업종 값") e["탄력성_ε"], e["탄력성_출처"] = eps, src return e def read_base(csv_path: str) -> pd.DataFrame: """기준 시나리오(보고서1 5장 산출물) → 행=연도, 열=산업, 값=실질VA(십억원).""" b = pd.read_csv(csv_path) return b.pivot(index="연도", columns="산업", values="실질VA_십억원").sort_index() # ── 2. 분석 ──────────────────────────────────────────────────────────── def apply_crosswalk(signal: pd.DataFrame, cw: pd.DataFrame) -> pd.DataFrame: """21→62 크로스워크로 62개 산업에 설문 점수를 붙인다.""" sig = signal.reset_index().rename(columns={"업종": "설문업종"}) out = cw.merge(sig, on="설문업종", how="left", validate="many_to_one") if out["AI성장_2040"].isna().any(): miss = out.loc[out["AI성장_2040"].isna(), "62산업"].tolist() raise ValueError(f"설문 점수가 없는 산업: {miss}") return out.set_index("62번") def ramp_factory(total_growth_scores: list[float]): """전체 응답 평균 E4-1 궤적으로 시점 램프 r(t)를 만든다(2025=0, 2040=1, 연간 선형보간).""" g0, g1 = total_growth_scores[0], total_growth_scores[-1] def ramp(year: int) -> float: return (float(np.interp(year, TIME_POINTS, total_growth_scores)) - g0) / (g1 - g0) return ramp def parent_of(col: str, carve: dict[str, int]) -> int: """산업 열 이름 → 모산업 62번(62분류는 앞 번호, 69분류 분화산업은 carve-out 표).""" m = re.match(r"\s*(\d+)\.", col) if m: return int(m.group(1)) for k, v in carve.items(): if col.strip().startswith(k): return v raise KeyError(f"모산업을 찾지 못한 산업: {col}") def deviations(cols, idx62: pd.DataFrame, carve: dict[str, int] | None = None): """산업별 성장·고용 편차 d_i = 점수(2040) − 5(척도 중간값 "보통"). 앵커 선택의 민감도는 allocation_variants 참조.""" carve = carve or {} dg = {c: idx62.loc[parent_of(c, carve), "AI성장_2040"] - NEUTRAL for c in cols} de = {c: idx62.loc[parent_of(c, carve), "AI고용_2040"] - NEUTRAL for c in cols} return dg, de def calibrate_lambda(base: pd.DataFrame, dev: dict, ramp, target_pp: float, year: int = 2035) -> float: """목표 연도 실질VA 가중평균 프리미엄 = target_pp(%p)가 되도록 Λ를 정한다.""" w = base.loc[year] / base.loc[year].sum() wdev = float(sum(w[c] * dev[c] for c in base.columns)) return target_pp / (wdev * ramp(year)) def project(base: pd.DataFrame, dev: dict, ramp, lam: float, start: int = 2025, end: int = 2035) -> pd.DataFrame: """AI 확산 시나리오 경로: 기준 성장률에 산업별 프리미엄 Λ·d_i·r(t)를 더해 누적한다.""" ai = base.loc[start:end].copy() for c in base.columns: lvl = base.loc[start, c] for y in range(start + 1, end + 1): lvl = lvl * (base.loc[y, c] / base.loc[y - 1, c]) * (1 + lam * dev[c] * ramp(y) / 100.0) ai.loc[y, c] = lvl return ai def premiums(cols, dev_g: dict, dev_e: dict, ramp, lam: float, year: int = 2035) -> pd.DataFrame: """산업별 부가가치·고용 프리미엄과 취업계수 연변화(%p).""" rows = [] for c in cols: vp = lam * dev_g[c] * ramp(year) ep = lam * dev_e[c] * ramp(year) rows.append({"산업": c, f"VA프리미엄_{year}_%p": vp, f"고용프리미엄_{year}_%p": ep, f"취업계수변화_{year}_%p": ep - vp}) return pd.DataFrame(rows).set_index("산업") def cagr(x0: float, x1: float, years: int) -> float: """연평균 증가율(%).""" return 100.0 * ((x1 / x0) ** (1.0 / years) - 1.0) def allocation_variants(idx62: pd.DataFrame, total_growth_scores: list[float]) -> dict: """산업 배분 방식 비교용 프리미엄 함수(보정 전 크기). 키 → f(산업열, 연도). E4-1은 영향의 '정도'(①매우 낮음 ⑤보통 ⑨매우 높음)를 묻는 문항이라 '영향 없음'에 해당하는 자연스러운 0점이 없다. 따라서 산업 배분은 기준점(앵커) 선택에 좌우된다. A 현행 : (s_i,2040 − 5) × r(t) — 중간값 '보통'을 0으로 둔다 A1 앵커 1 : (s_i,2040 − 1) × r(t) — 척도 최저점 '매우 낮음'을 0으로 둔다 B 증가분 : s_i(t) − s_i(2025) — 2025년 영향은 기준선에 반영되었다고 보고 증가분만 쓴다 C 상대강도: s_i,2040 / s_전체,2040 × r(t) """ ramp = ramp_factory(total_growth_scores) tot40 = total_growth_scores[-1] def num(c): return int(re.match(r"\s*(\d+)\.", c).group(1)) def s_t(c, t): return float(np.interp(t, TIME_POINTS, [idx62.loc[num(c), f"AI성장_{x}"] for x in TIME_POINTS])) return { "A 현행(s−5)": lambda c, t: (idx62.loc[num(c), "AI성장_2040"] - 5.0) * ramp(t), "A1 앵커1(s−1)": lambda c, t: (idx62.loc[num(c), "AI성장_2040"] - 1.0) * ramp(t), "B 증가분": lambda c, t: s_t(c, t) - s_t(c, 2025), "C 상대강도": lambda c, t: idx62.loc[num(c), "AI성장_2040"] / tot40 * ramp(t), } def project_variant(base: pd.DataFrame, prem_fn, target_pp: float, start: int = 2025, end: int = 2035): """임의의 프리미엄 함수로 Λ를 보정(목표 연도 VA 가중평균 = target_pp)하고 경로·2035 프리미엄을 반환한다.""" w = base.loc[end] / base.loc[end].sum() lam = target_pp / float(sum(w[c] * prem_fn(c, end) for c in base.columns)) ai = base.loc[start:end].copy() for c in base.columns: lvl = base.loc[start, c] for y in range(start + 1, end + 1): lvl = lvl * (base.loc[y, c] / base.loc[y - 1, c]) * (1 + lam * prem_fn(c, y) / 100.0) ai.loc[y, c] = lvl prem = pd.Series({c: lam * prem_fn(c, end) for c in base.columns}) return ai, prem def central_premium_fn(idx62: pd.DataFrame, total_growth_scores: list[float], allocation: str): """설정된 배분 방식의 프리미엄 함수(보정 전).""" variants = allocation_variants(idx62, total_growth_scores) key = {"A": "A 현행(s−5)", "A1": "A1 앵커1(s−1)", "B": "B 증가분", "C": "C 상대강도"}[allocation] return variants[key] def labor_coef_change(prem: pd.Series, eps_by_col: dict, mode: str = "neutral") -> pd.DataFrame: """AI 확산에 따른 취업계수 연변화(%p) = (ε − 1) × 부가가치 프리미엄. mode="neutral": 중심, ε=0(고용 중립) → −프리미엄. mode="survey": 설문 탄력성 ε_i 적용. 두 값을 모두 열로 남기고, '취업계수변화_2035_%p'에는 mode에 해당하는 값을 넣는다. """ rows = [] for c, p in prem.items(): eps = eps_by_col[c] neutral, survey = -p, (eps - 1.0) * p rows.append({"산업": c, "VA프리미엄_2035_%p": p, "고용탄력성_ε(설문)": eps, "취업계수변화_2035_%p": neutral if mode == "neutral" else survey, "취업계수변화_고용중립_%p": neutral, "취업계수변화_설문탄력성_%p": survey}) return pd.DataFrame(rows).set_index("산업") # ── 3. 정비 전후 비교: 같은 AI 확산 층을 다른 기준선에 적용한 결과 요약 ───────────── def scenario_summary(base62: pd.DataFrame, base69: pd.DataFrame, fn62, fn69, eps62: dict, eps69: dict, carve_names: list, lambdas: list, central: float, mode: str = "neutral", start: int = 2025, end: int = 2035) -> pd.DataFrame: """기준선(62·69분류)에 AI 확산 층(배분 함수·Λ 보정·취업계수)을 적용한 핵심 수치를 장형 표(구분·항목·값)로 반환한다. 계산은 위 함수(project_variant·labor_coef_change)를 그대로 쓰며, 기준선만 바꾸어 정비 전후를 비교하는 데 쓴다. 총량 항목 이름은 Λ 중심값이면 'AI_', 그 밖이면 'L03_'·'L08_'처럼 Λ×10 두 자리를 붙인다. """ rows = [] b25, b35 = base62.loc[start].sum() / 1e3, base62.loc[end].sum() / 1e3 g_b = cagr(b25, b35, end - start) rows += [("총량", "기준_2025_조", b25), ("총량", "기준_2035_조", b35), ("총량", "기준_CAGR", g_b)] for lam in lambdas: tag = "AI" if abs(lam - central) < 1e-12 else f"L{int(round(lam * 10)):02d}" ai, prem = project_variant(base62, fn62, lam, start, end) a35 = ai.loc[end].sum() / 1e3 rows += [("총량", f"{tag}_2035_조", a35), ("총량", f"{tag}_CAGR", cagr(b25, a35, end - start)), ("총량", f"{tag}_차이_%p", cagr(b25, a35, end - start) - g_b), ("총량", f"{tag}_수준차이_조", a35 - b35), ("총량", f"{tag}_수준차이_%", 100.0 * (a35 / b35 - 1.0))] if tag == "AI": lab = labor_coef_change(prem, eps62, mode) rows += [("62 프리미엄(%p)", c, float(v)) for c, v in prem.items()] rows += [("62 취업계수 중심(%p)", c, float(v)) for c, v in lab["취업계수변화_2035_%p"].items()] rows += [("62 취업계수 설문탄력성(%p)", c, float(v)) for c, v in lab["취업계수변화_설문탄력성_%p"].items()] ai69, prem69 = project_variant(base69, fn69, central, start, end) for c in carve_names: rows += [("69 분화 프리미엄(%p)", c, float(prem69[c])), ("69 분화 기준 2035(조원)", c, float(base69.loc[end, c]) / 1e3), ("69 분화 AI 2035(조원)", c, float(ai69.loc[end, c]) / 1e3)] return pd.DataFrame(rows, columns=["구분", "항목", "값"]) def lookup(summary: pd.DataFrame, col: str, group: str, item) -> float: """요약 표에서 (구분, 항목)의 값을 꺼낸다. item이 정수면 62분류 산업 번호로 찾는다.""" s = summary[summary["구분"] == group] if isinstance(item, (int, np.integer)): s = s[s["항목"].astype(str).str.split(".").str[0].str.strip() == str(int(item))] else: s = s[s["항목"] == item] if len(s) != 1: raise KeyError(f"요약 표에서 ({group}, {item})를 하나로 찾지 못했다({len(s)}건).") return float(s[col].iloc[0])