# -*- coding: utf-8 -*- """보고서1 5장(전망 결과) — 정합화·요약·전달 산출물 계산 함수. 노트북 01(입수)·02(분석)·03(결과)이 이 모듈의 함수만 호출한다. 경로·상수는 config.py에 둔다. 출처(원 프로젝트, 정리 사본): industry_pipeline/reconcile.py reconcile · summary_table (2026-09-03판) industry_pipeline/build_total_series.py total_series (2026-09-01판) scratchpad/fill_2025_2035_va.py fill_template · card_data (2026-09-03판, 통합 번들 industry/code/ 사본과 동일) scratchpad/build_va_detail.py va_detail (같은 판; SHAP 항목은 02 10절 정비판 분해로 채운다) 원 코드는 결과를 원 프로젝트 경로에 덮어썼다. 이 모듈은 DataFrame·dict를 돌려주고 저장은 노트북이 한다. 정합화(02_분석) 계산식 — 시나리오 s, 연도 t, 산업 i g^M_{s,t} = AGG_{s,t} / AGG_{2025} 총량(탑다운) 모형의 성장인자 g^P_t = KEIS_t / KEIS_2025 기존 전망(2024–2034판) 전산업의 성장인자 g_{s,t} = exp[ w·ln g^P_t + (1−w)·ln g^M_{s,t} ] 기존 전망 수축 결합(기하 가중), w = 0.45 T_{s,t} = Σ_i VA^{실적}_{i,2025} · g_{s,t} 통제 총량 VA_{i,s,t} = T_{s,t} · ŷ_{i,s,t} / Σ_j ŷ_{j,s,t} 비례 벤치마킹(바텀업 ŷ의 산업 구성비 유지) 같은 연도·시나리오의 모든 산업에 같은 비례계수 T/Σŷ를 곱하는 결정론적 벤치마킹이다(가중최소자승 사영이 아니다). """ from __future__ import annotations import copy import hashlib import json import numpy as np import openpyxl import pandas as pd from openpyxl.utils import get_column_letter SCN = ["S1", "S2", "S3", "S4", "S5"] VAL = "실질VA_십억원" BASE_YEAR, END_YEAR = 2025, 2035 # ── 공통 ────────────────────────────────────────────────────────────── def sha256(path: str) -> str: h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1 << 20), b""): h.update(chunk) return h.hexdigest() def cagr(x0: float, x1: float, years: int) -> float: """연평균 증가율(%).""" return 100.0 * ((x1 / x0) ** (1.0 / years) - 1.0) def read_long(path: str) -> pd.DataFrame: """장형 전망표(시나리오·연도·산업·실질VA_십억원).""" return pd.read_csv(path, encoding="utf-8-sig") def totals(df: pd.DataFrame) -> pd.DataFrame: """연도 × 시나리오 전산업 합계(십억원).""" return df.groupby(["시나리오", "연도"])[VAL].sum().unstack("시나리오") def describe_long(df: pd.DataFrame) -> dict: return {"행": len(df), "산업": df["산업"].nunique(), "시나리오": ",".join(sorted(df["시나리오"].unique())), "연도": f"{df['연도'].min()}–{df['연도'].max()}"} # ── 입수: 엑셀 VA 시트 추출 ──────────────────────────────────────────── def extract_va_sheet(xlsx_path: str, sheet: str, y0: int = 2000, y1: int = 2035) -> pd.DataFrame: """KEIS 서식 VA 시트(2행=연도, B열=번호, C열=산업명, D열부터 2000년) → 장형(번호·산업·연도·VA_백만원). 산업 행은 3행부터 C열이 '전산업'인 행까지 읽는다. 전산업 행은 번호 0으로 표기한다. 수식 셀은 엑셀이 저장한 계산값을 읽는다(data_only). """ ws = openpyxl.load_workbook(xlsx_path, data_only=True, read_only=True)[sheet] grid = [list(r) for r in ws.iter_rows(min_row=1, max_row=ws.max_row, values_only=True)] header = grid[1] col_of = {int(v): j for j, v in enumerate(header) if isinstance(v, (int, float)) and 1990 < v < 2100} rows = [] for r in grid[2:]: name = r[2] if name is None: continue num = 0 if r[1] is None else int(r[1]) for y in range(y0, y1 + 1): rows.append((num, str(name).strip(), y, float(r[col_of[y]]))) if str(name).strip() == "전산업": break return pd.DataFrame(rows, columns=["번호", "산업", "연도", "VA_백만원"]) # ── 분석: 기존 전망 성장인자 ────────────────────────────────────────── def prior_factors(prev: pd.DataFrame, base: int = BASE_YEAR, end: int = END_YEAR) -> dict[int, float]: """기존 전망(2024–2034판) 전산업 행의 성장인자 g^P_t = VA_t / VA_base (base=1).""" tot = prev[prev["번호"] == 0].set_index("연도")["VA_백만원"] return {y: float(tot.loc[y] / tot.loc[base]) for y in range(base, end + 1)} # ── 분석: 정합화(reconcile.py) ──────────────────────────────────────── def topdown_factors(agg: pd.DataFrame) -> tuple[pd.DataFrame, float]: """총량 모형 장형표 → (연도 × 시나리오 수준, 2025 실적).""" lv = agg.groupby(["시나리오", "연도"])[VAL].sum().unstack("시나리오") a25 = agg[(agg["시나리오"] == "실적") & (agg["연도"] == BASE_YEAR)][VAL].sum() return lv, float(a25) def bokdpm_annual_gdp(lgdp_q: pd.Series) -> pd.Series: """BOKDPM 실질GDP(분기, 100·ln, 과거 평활값 + 기준선 전망) → 연간 수준(분기 수준의 합, 십억 원 단위 아님·비율만 쓴다).""" yr = pd.Series([int(q[:4]) for q in lgdp_q.index], index=lgdp_q.index) return np.exp(lgdp_q / 100.0).groupby(yr).sum() def topdown_bokdpm(lgdp_q: pd.Series, agg_varx: pd.DataFrame, scenarios=SCN) -> pd.DataFrame: """통제 총량의 장형표(2026-10 정비): 기준선 = BOKDPM 실질GDP, 시나리오 편차 = 총량 VARX의 S1 대비 비율. GDP^{s}_t = GDP^{BOKDPM}_t × AGG_{s,t} / AGG_{S1,t}. 3장 외생과 같은 '기준선 + (시나리오 − S1) 편차' 구성이다. 실적(2000–2025)은 BOKDPM 평활값(관측 실질GDP와 같다)이다. """ ga = bokdpm_annual_gdp(lgdp_q) ta = totals(agg_varx) rows = [("실적", int(y), "총량", float(ga[y])) for y in ga.index if y <= BASE_YEAR] for s in scenarios: rows += [(s, int(y), "총량", float(ga[y] * ta.loc[y, s] / ta.loc[y, "S1"])) for y in ga.index if y > BASE_YEAR] return pd.DataFrame(rows, columns=["시나리오", "연도", "산업", VAL]) def reconcile(bu: pd.DataFrame, agg: pd.DataFrame, prior: dict[int, float], w: float, scenarios=SCN) -> tuple[pd.DataFrame, pd.DataFrame]: """바텀업 산업 전망을 통제 총량에 비례 벤치마킹한다(원 reconcile.py와 같은 계산). 반환: (정합화 장형표[소수 셋째 자리 반올림], 연도·시나리오별 인자표) """ lv, agg25 = topdown_factors(agg) ind25 = bu[(bu["시나리오"] == "실적") & (bu["연도"] == BASE_YEAR)][VAL].sum() out = [bu[bu["시나리오"] == "실적"]] fac = [] for s in scenarios: d = bu[bu["시나리오"] == s].copy() for y in sorted(d["연도"].unique()): m = d["연도"] == y yhat = d.loc[m, VAL] gm = lv.loc[y, s] / agg25 # 총량 모형 성장인자 gp = prior.get(int(y), gm) # 기존 전망 성장인자 g = np.exp(w * np.log(gp) + (1 - w) * np.log(gm)) # 결합 성장인자 ctrl = ind25 * g # 통제 총량 d.loc[m, VAL] = (ctrl * (yhat / yhat.sum())).values # 비례 벤치마킹 fac.append({"시나리오": s, "연도": int(y), "바텀업합_십억원": float(yhat.sum()), "총량모형_gM": float(gm), "기존전망_gP": float(gp), "결합_g": float(g), "통제총량_십억원": float(ctrl), "비례계수": float(ctrl / yhat.sum())}) out.append(d) r = pd.concat(out).sort_values(["시나리오", "연도", "산업"]).reset_index(drop=True) r[VAL] = r[VAL].round(3) return r, pd.DataFrame(fac) def summary_table(r: pd.DataFrame, scenarios=SCN) -> pd.DataFrame: """시나리오 × 산업: VA2025·VA2035·연평균(2026–35) — reconcile.py의 요약표와 같은 형식.""" rows = [] a25 = r[(r["시나리오"] == "실적") & (r["연도"] == BASE_YEAR)].set_index("산업")[VAL] for s in scenarios: f35 = r[(r["시나리오"] == s) & (r["연도"] == END_YEAR)].set_index("산업")[VAL] for ind in a25.index: v25 = a25[ind]; v35 = f35.get(ind) if v35 is None: continue cg = ((v35 / v25) ** (1 / 10) - 1) * 100 if v25 > 0 else 0 rows.append([s, ind, round(v25, 2), round(v35, 2), round(cg, 3)]) return pd.DataFrame(rows, columns=["시나리오", "산업", "VA2025", "VA2035", "CAGR_26_35(%)"]) def total_series(r: pd.DataFrame, scenarios=SCN) -> pd.DataFrame: """연도 × [실적, S1–S5] 전산업 총계(조원, 2025년은 모든 시나리오가 실적과 같다) — build_total_series.py.""" tot = totals(r) / 1000.0 W = pd.DataFrame(index=sorted(r["연도"].unique())); W.index.name = "연도" W["실적"] = tot["실적"] anchor = tot["실적"].dropna().index.max() for s in scenarios: col = tot[s].copy() col.loc[anchor] = tot["실적"].loc[anchor] W[s] = col return W.round(1) def base_scenario(r: pd.DataFrame, label: str = "기준(S1)") -> pd.DataFrame: """기준 시나리오(S1) 2025 실적 + 2026–2035 전망 → 보고서2 4장 입력 형식.""" a = r[(r["시나리오"] == "실적") & (r["연도"] == BASE_YEAR)] f = r[(r["시나리오"] == "S1") & (r["연도"] > BASE_YEAR)] b = pd.concat([a, f]).assign(시나리오=label) return b.sort_values(["연도", "산업"])[["시나리오", "연도", "산업", VAL]].reset_index(drop=True) def compare_long(new: pd.DataFrame, pub: pd.DataFrame, keys, cols) -> dict: """두 표를 키로 맞춰 열별 최대 절대차를 구한다(행 수·키 집합이 같아야 한다).""" m = new.merge(pub, on=keys, how="outer", suffixes=("_계산", "_공표"), indicator=True) out = {"행_계산": len(new), "행_공표": len(pub), "키불일치": int((m["_merge"] != "both").sum())} for c in cols: out[c] = float((m[f"{c}_계산"] - m[f"{c}_공표"]).abs().max()) return out # ── 결과: 전달 템플릿 채우기(fill_2025_2035_va.py) ──────────────────── COLY = {y: 4 + (y - 2000) for y in range(2000, 2036)} # 2000년 = D열(4) … 2035년 = AM열(39) def s1_with_actuals(r62: pd.DataFrame) -> pd.DataFrame: """연도 × 산업: 실적(2000–2025) + S1 전망(2026–2035).""" a = r62[r62["시나리오"] == "실적"].pivot(index="연도", columns="산업", values=VAL) f = r62[r62["시나리오"] == "S1"].pivot(index="연도", columns="산업", values=VAL) return pd.concat([a, f]).sort_index() def fill_template(template_xlsx: str, r62: pd.DataFrame, out_xlsx: str, copy_style: bool = True) -> dict: """KEIS 62분류 VA 템플릿을 2035년까지 채운다(VA만, 총산출 시트 없음). - 2024–2035: 산업별로 템플릿 2023년 수준 × (정합화 S1 전망의 t년 / 2023년 실적) - 2035 열(AM) 신규: 헤더, 62개 산업, 전산업(65행 = 3–64행 합), 대분류 소계(68–76행: AL 수식을 AM으로 복제) - copy_style=True: AL열의 셀 서식·열 너비를 AM열에 복제한다(값에는 영향이 없다. 원 스크립트에는 없던 표시용 보완). """ our = s1_with_actuals(r62) num_of = lambda c: int(c.split(".")[0]) fac = {num_of(c): {t: (our.loc[t, c] / our.loc[2023, c] if our.loc[2023, c] else 1.0) for t in range(2024, 2036)} for c in our.columns} wb = openpyxl.load_workbook(template_xlsx) va = wb["62분류_불변VA"] OLD, NEW, NAME = {}, {}, {} for r in range(3, 65): num = va.cell(r, 2).value if num is None: continue num = int(num) NAME[num] = va.cell(r, 3).value OLD[num] = {y: va.cell(r, COLY[y]).value for y in range(2023, 2035)} old_agg = {y: va.cell(65, COLY[y]).value for y in range(2023, 2035)} va.cell(2, COLY[2035]).value = 2035 for r in range(3, 65): num = va.cell(r, 2).value if num is None or int(num) not in fac: continue num = int(num) b23 = va.cell(r, COLY[2023]).value NEW[num] = {2023: b23} for y in range(2024, 2036): v = b23 * fac[num][y] va.cell(r, COLY[y]).value = v NEW[num][y] = v new_agg = {2023: va.cell(65, COLY[2023]).value} for y in range(2024, 2036): s = sum((va.cell(r, COLY[y]).value or 0) for r in range(3, 65)) va.cell(65, COLY[y]).value = s new_agg[y] = s colAL, colAM = get_column_letter(COLY[2034]), get_column_letter(COLY[2035]) for r in range(68, 77): f = va.cell(r, COLY[2034]).value if isinstance(f, str) and f.startswith("="): va.cell(r, COLY[2035]).value = f.replace(colAL, colAM) if copy_style: for r in range(1, va.max_row + 1): src, dst = va.cell(r, COLY[2034]), va.cell(r, COLY[2035]) if src.has_style: dst._style = copy.copy(src._style) va.column_dimensions[colAM].width = va.column_dimensions[colAL].width wb.save(out_xlsx) return {"OLD": OLD, "NEW": NEW, "NAME": NAME, "old_agg": old_agg, "new_agg": new_agg, "fac": fac} def _cagr0(a, b, n): return (b / a) ** (1 / n) - 1 if a else 0 def card_data(fill: dict, names: dict[int, str]) -> dict: """비교 카드 요약(card_2025_2035.json과 같은 구조). names: 번호 → 전망표 산업명.""" OLD, NEW, old_agg, new_agg = fill["OLD"], fill["NEW"], fill["old_agg"], fill["new_agg"] inds = [] for num in sorted(NEW): o, n = OLD[num], NEW[num] inds.append({"num": num, "name": names.get(num, str(num)), "old_cagr": 100 * _cagr0(o[2024], o[2034], 10), "new_cagr": 100 * _cagr0(n[2024], n[2034], 10), "old_2034": o[2034] / 1e6, "new_2034": n[2034] / 1e6, "new_2035": n[2035] / 1e6}) agg = {"old_path": {str(y): old_agg[y] / 1e6 for y in range(2023, 2035)}, "new_path": {str(y): new_agg[y] / 1e6 for y in range(2023, 2036)}, "old_cagr_24_34": 100 * _cagr0(old_agg[2024], old_agg[2034], 10), "new_cagr_24_34": 100 * _cagr0(new_agg[2024], new_agg[2034], 10), "new_cagr_25_35": 100 * _cagr0(new_agg[2025], new_agg[2035], 10), "old_2034": old_agg[2034] / 1e6, "new_2034": new_agg[2034] / 1e6, "new_2035": new_agg[2035] / 1e6} return {"agg": agg, "inds": inds} def va_detail(template_xlsx: str, filled_xlsx: str, shap: dict | None = None) -> dict: """산업별 비교 자료(va_detail_62.json과 같은 구조). 단위 조원, 소수 둘째 자리. hist = 2000–2023 실적(템플릿), old = 2023–2034 기존 전망(템플릿), new = 2023–2035 신규 전망(채운 파일). shap = 02 10절의 성장 요인 분해(shap_62.json: exog·meta·rows). 주면 산업별 shap = {total, drift, dynamics, exog{외생 6종}}과 최상위 exog(외생 이름 목록)·shap_meta를 넣는다(원 build_va_detail.py와 같은 구조). 주지 않으면 shap = None. """ wo = openpyxl.load_workbook(template_xlsx, data_only=True)["62분류_불변VA"] wn = openpyxl.load_workbook(filled_xlsx, data_only=True)["62분류_불변VA"] col = lambda y: 4 + (y - 2000) sh = {int(r["ind"].split(".")[0]): r for r in shap["rows"]} if shap else {} inds = [] for r in range(3, 65): num = wo.cell(r, 2).value if num is None: continue num = int(num) g = lambda ws, y: ws.cell(r, col(y)).value or 0 o24, o34 = g(wo, 2024), g(wo, 2034) n24, n34, n25, n35 = g(wn, 2024), g(wn, 2034), g(wn, 2025), g(wn, 2035) inds.append({"num": num, "name": wn.cell(r, 3).value, "hist": [round(g(wo, y) / 1e6, 2) for y in range(2000, 2024)], "old": [round(g(wo, y) / 1e6, 2) for y in range(2023, 2035)], "new": [round(g(wn, y) / 1e6, 2) for y in range(2023, 2036)], "cagr_old": round(100 * _cagr0(o24, o34, 10), 2) if o24 and o34 else 0, "cagr_new": round(100 * _cagr0(n24, n34, 10), 2) if n24 and n34 else 0, "cagr_new35": round(100 * _cagr0(n25, n35, 10), 2) if n25 and n35 else 0, "shap": None if not shap else {"total": sh[num]["total"], "drift": sh[num]["drift"], "dynamics": sh[num]["dynamics"], "exog": {k: sh[num][k] for k in shap["exog"]}}}) out = {"yh": list(range(2000, 2024)), "yo": list(range(2023, 2035)), "yn": list(range(2023, 2036)), "inds": inds} if shap: out = {"exog": shap["exog"], "shap_meta": shap["meta"], **out} return out def max_abs_json_diff(a, b, skip=("shap", "exog")) -> float: """두 JSON 구조의 수치 최대 절대차(문자열은 같아야 하며, 다르면 inf). skip 키는 비교하지 않는다.""" if isinstance(a, dict) and isinstance(b, dict): ks = [k for k in set(a) | set(b) if k not in skip] return max([max_abs_json_diff(a.get(k), b.get(k), skip) for k in ks] or [0.0]) if isinstance(a, list) and isinstance(b, list): if len(a) != len(b): return float("inf") return max([max_abs_json_diff(x, y, skip) for x, y in zip(a, b)] or [0.0]) if isinstance(a, (int, float)) and isinstance(b, (int, float)): return abs(float(a) - float(b)) return 0.0 if a == b else float("inf") def read_filled_values(xlsx: str, years=range(2000, 2036)) -> pd.DataFrame: """채운 62분류 VA 시트의 3–65행(62개 산업 + 전산업) 값 → 장형(번호·산업·연도·VA_백만원).""" ws = openpyxl.load_workbook(xlsx, data_only=True)["62분류_불변VA"] rows = [] for r in range(3, 66): num = ws.cell(r, 2).value name = ws.cell(r, 3).value for y in years: rows.append((0 if num is None else int(num), str(name).strip(), y, float(ws.cell(r, COLY[y]).value))) return pd.DataFrame(rows, columns=["번호", "산업", "연도", "VA_백만원"]) def json_dump(obj, path: str) -> None: with open(path, "w", encoding="utf-8") as f: json.dump(obj, f, ensure_ascii=False, indent=1)