# -*- coding: utf-8 -*-
"""보고서1 2장(데이터) — 36개 산업 분기 실질부가가치의 62·69개 산업 확장과 연쇄가중 총량 재현.

노트북 01(입수)·02(분석)·03(결과)이 이 모듈의 함수만 호출한다. 경로·상수는 config.py에 둔다.

계산 구조
  1. 입수      : 한국은행 국민계정 36개 산업 분기 실질·명목 부가가치(ECOS 200Y103·200Y104),
                 총부가가치(200Y104 항목 1200), 실질 GDP(200Y108), 산업연관표 2020(271Y120·271Y122),
                 광업제조업동향조사 생산지수(KOSIS 101/DT_1F02001), 발전량(KOSIS 388/TX_38803_A016A, e-나라지표 1339)
  2. 36→62    : KEIT62 ↔ BOK36 ↔ IO 상품중분류(83) 대조표. 여러 62가 한 BOK36을 나누어 가지면
                 2020년 IO 부가가치 비중(상수)으로 배분한다. 62 = BOK36 분기 실질VA × 몫
  3. 시변 몫   : 이차전지 = k·(C282/C28) (k는 IO 소분류 373/37 비중 2020–2023 평균에 맞춘 상수),
                 신재생 = 발전비중(신재생/총발전) × 전력부문 VA몫(IO 45/(45+46))
  4. 62→69    : 반도체·조선·철강·택배(IO 고정 몫), 이차전지·신재생(시변 몫)을 모산업에서 분리하고
                 건설을 건물건설·토목으로 나눈다. 분리 후 모산업은 '(잔여)' 열로 남는다
  5. 총량      : levels_agg = 실질 GDP(계절조정). 62·69의 '총량GDP' 열은 36개 산업 실질VA의 단순합이다
  6. IO 연계   : 중분류 생산유발계수를 62·69 산업으로 VA 가중 집계한 연계강도 행렬(산업모형 사전용)
  7. 연쇄 재현 : 36개 산업 실질VA의 단순합은 연쇄가중 총부가가치와 다르다. 전기 명목 비중으로 가중한
                 성장배율을 2020Q1부터 양방향으로 누적해 총부가가치를 재현한다
"""
from __future__ import annotations

import hashlib
import json
import os
import time
from collections import defaultdict

import numpy as np
import pandas as pd

ANCHOR = "2020Q1"      # 연쇄 재현 기준점(36개 산업 단순합과 총부가가치의 괴리가 가장 작은 분기)


# ── 공통 ──────────────────────────────────────────────────────────────
def quarters(start: str = "2000Q1", end: str = "2025Q4") -> list[str]:
    """'2000Q1'~'2025Q4' 형식의 분기 목록."""
    y, q = int(start[:4]), int(start[-1])
    y1, q1 = int(end[:4]), int(end[-1])
    out = []
    while (y, q) <= (y1, q1):
        out.append(f"{y}Q{q}")
        q += 1
        if q > 4:
            y, q = y + 1, 1
    return out


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()


# ── 1. 입수: 스냅샷 읽기 ──────────────────────────────────────────────
def read_va36(path: str) -> tuple[pd.DataFrame, pd.DataFrame, dict]:
    """36개 산업 분기 장형 CSV → (실질 광형, 명목 광형, 코드→산업명). 열은 ECOS 표시 순서."""
    g = pd.read_csv(path, dtype={"산업코드": str})
    order = list(dict.fromkeys(g["산업코드"]))
    names = dict(zip(g["산업코드"], g["산업명"]))
    real = g.pivot_table(index="분기", columns="산업코드", values="실질_십억원", aggfunc="first")[order]
    nom = g.pivot_table(index="분기", columns="산업코드", values="명목_십억원", aggfunc="first")[order]
    real.index.name = nom.index.name = "분기"
    return real, nom, names


def read_gva_total(path: str) -> pd.Series:
    """총부가가치(기초가격, 실질, 계절조정) 분기 계열."""
    d = pd.read_csv(path)
    return d.set_index("분기")["총부가가치_실질_십억원"]


def read_gdp(path: str) -> pd.DataFrame:
    return pd.read_csv(path, index_col=0)


def read_io_va83(path: str, year: int = 2020) -> pd.Series:
    """IO 상품중분류(83) 부가가치(백만원) — io_code(1~83) → 값."""
    io = pd.read_csv(path)
    io = io[(io["year"] == year) & (io["io_code"] <= 83)]
    return io.set_index("io_code")["va_mil_won"]


def read_crosswalk(path: str) -> dict:
    """KEIT62 ↔ BOK36 ↔ IO83 대조표(JSON) → MAP(io83→(keit,bok)), keit_name, keit_bok."""
    cw = json.load(open(path, encoding="utf-8"))
    return {"MAP": {int(k): tuple(v) for k, v in cw["MAP"].items()},
            "keit_name": {int(k): v for k, v in cw["keit_name"].items()},
            "keit_bok": {int(k): v for k, v in cw["keit_bok"].items()}}


def read_mfg_prod(path: str) -> pd.DataFrame:
    """광업제조업동향조사 생산지수(원지수) 장형 → 월(행) × KSIC(열)."""
    m = pd.read_csv(path)
    return m.pivot_table(index="date", columns="ksic", values="value", aggfunc="first").sort_index()


def snapshot_table(files: dict, expected: dict) -> pd.DataFrame:
    """스냅샷 파일별 크기·행수·SHA-256과 기록값 대조표."""
    rows = []
    for label, p in files.items():
        h = sha256(p)
        n = sum(1 for _ in open(p, encoding="utf-8-sig")) - 1 if p.endswith(".csv") else None
        rows.append({"파일": os.path.basename(p), "내용": label, "바이트": os.path.getsize(p), "행수": n,
                     "SHA-256(앞 16자)": h[:16], "기록값과 일치": h == expected.get(os.path.basename(p))})
    out = pd.DataFrame(rows)
    out["행수"] = out["행수"].astype("Int64")
    return out


# ── 1'. 입수: 원천 재수집(ONLINE=1) ────────────────────────────────────
ECOS_URL = "https://ecos.bok.or.kr/api/StatisticSearch/{key}/json/kr/1/{n}/{table}/{cycle}/{start}/{end}{item}"
KOSIS_URL = "https://kosis.kr/openapi/Param/statisticsParameterData.do"


def _get_json(url: str, params: dict | None = None, retry: int = 3, timeout: int = 120):
    import requests
    for k in range(retry):
        try:
            return requests.get(url, params=params, timeout=timeout).json()
        except Exception:
            if k == retry - 1:
                raise
            time.sleep(2)


def ecos_rows(key: str, table: str, cycle: str, start: str, end: str, item: str = "", n: int = 100000) -> pd.DataFrame:
    """ECOS StatisticSearch → 행 DataFrame(ITEM_CODE1·ITEM_CODE2·TIME·DATA_VALUE …)."""
    url = ECOS_URL.format(key=key, n=n, table=table, cycle=cycle, start=start, end=end,
                          item=f"/{item}" if item else "")
    j = _get_json(url)
    rows = j.get("StatisticSearch", {}).get("row", [])
    if not rows:
        msg = j.get("RESULT", {}).get("MESSAGE", "빈 응답")
        raise RuntimeError(f"ECOS {table}: {msg}")
    d = pd.DataFrame(rows)
    d["DATA_VALUE"] = pd.to_numeric(d["DATA_VALUE"], errors="coerce")
    return d


def fetch_va36_online(key: str, codes: list[str], start="1980Q1", end="2025Q4") -> tuple[pd.DataFrame, pd.DataFrame, pd.Series]:
    """ECOS 200Y104(실질)·200Y103(명목) 전 항목 → 36개 산업 광형 2개 + 총부가가치(1200, 실질)."""
    out = {}
    for table, kind in [("200Y104", "real"), ("200Y103", "nominal")]:
        d = ecos_rows(key, table, "Q", start, end)
        out[kind] = d.pivot_table(index="TIME", columns="ITEM_CODE1", values="DATA_VALUE", aggfunc="first")
    real, nom = out["real"][codes], out["nominal"][codes]
    gva = out["real"]["1200"].rename("총부가가치_실질_십억원")
    return real, nom, gva


def fetch_gdp_online(key: str, start="1980Q1", end="2025Q4") -> pd.Series:
    """ECOS 200Y108(지출, 계절조정, 실질) 항목 10601 = 국내총생산."""
    d = ecos_rows(key, "200Y108", "Q", start, end, item="10601")
    return d.set_index("TIME")["DATA_VALUE"].rename("실질GDP_십억원")


def fetch_io_va83_online(key: str, year: int = 2020) -> pd.Series:
    """ECOS 271Y120(투입산출표 총거래표, 생산자가격, 중분류) — 부가가치계(9690) 행."""
    d = ecos_rows(key, "271Y120", "A", str(year), str(year))
    d = d[(d["ITEM_CODE2"] == "9690") & d["ITEM_CODE1"].str.isdigit()]
    s = d.assign(io=d["ITEM_CODE1"].astype(int)).query("io <= 83").set_index("io")["DATA_VALUE"]
    return s.sort_index().rename("va_mil_won")


def fetch_io_M_online(key: str, year: int = 2020) -> pd.DataFrame:
    """ECOS 271Y122(생산유발계수, 생산자가격, 중분류)."""
    d = ecos_rows(key, "271Y122", "A", str(year), str(year), n=8000)
    d = d[d["ITEM_CODE1"].str.isdigit() & d["ITEM_CODE2"].str.isdigit()]
    return pd.DataFrame({"i": d["ITEM_CODE1"].astype(int), "j": d["ITEM_CODE2"].astype(int),
                         "v": d["DATA_VALUE"].astype(float)}).reset_index(drop=True)


def kosis_rows(key: str, org: str, tbl: str, prd: str, start: str, end: str, **obj) -> pd.DataFrame:
    p = dict(method="getList", apiKey=key, format="json", jsonVD="Y", prdSe=prd,
             startPrdDe=start, endPrdDe=end, orgId=org, tblId=tbl)
    p.update(obj)
    j = _get_json(KOSIS_URL, params=p)
    if isinstance(j, dict):
        raise RuntimeError(f"KOSIS {org}/{tbl}: {j.get('errMsg', j)}")
    d = pd.DataFrame(j)
    d["DT"] = pd.to_numeric(d["DT"], errors="coerce")
    return d


def fetch_mfg_prod_online(key: str, start="200001", end="202604") -> pd.DataFrame:
    """KOSIS 101/DT_1F02001 생산지수(원지수, T10), 전국(00), C28·C282 → 월 × KSIC."""
    frames = []
    for ksic in ("C28", "C282"):
        d = kosis_rows(key, "101", "DT_1F02001", "M", start, end, itmId="T10", objL1="00", objL2=ksic)
        frames.append(pd.DataFrame({"date": d["PRD_DE"].map(lambda t: f"{t[:4]}-{t[4:6]}-01"),
                                    "ksic": d["C2"], "value": d["DT"]}))
    m = pd.concat(frames)
    return m.pivot_table(index="date", columns="ksic", values="value", aggfunc="first").sort_index()


def fetch_kpx_online(key: str, start=2006, end=2016) -> pd.DataFrame:
    """KOSIS 388/TX_38803_A016A 발전량: 신재생 및 기타(A005)·합계(A001), GWh."""
    d = kosis_rows(key, "388", "TX_38803_A016A", "Y", str(start), str(end), itmId="ALL", objL1="ALL")
    w = d[d["C1"].isin(["A001", "A005"])].pivot_table(index="PRD_DE", columns="C1", values="DT", aggfunc="first")
    out = pd.DataFrame({"연도": w.index.astype(int), "신재생및기타_GWh": w["A005"].values, "합계_GWh": w["A001"].values})
    return out.reset_index(drop=True)


def extract_io_small_xlsx(paths: dict) -> pd.DataFrame:
    """한국은행 산업연관표 「투입산출표(생산자가격) 통합소분류」 xlsx에서 37x(전기장비) 부가가치계를 뽑는다.

    paths: {연도: xlsx 경로}. 첫 시트(총거래표)의 5행이 상품코드, 6행이 상품명, '부가가치계' 행이 값이다.
    """
    import openpyxl
    recs = []
    for y, f in sorted(paths.items()):
        ws = openpyxl.load_workbook(f, read_only=True, data_only=True).worksheets[0]
        rows = list(ws.iter_rows(values_only=True))
        codes, names = rows[4], rows[5]
        v = [r for r in rows if r[1] and str(r[1]).replace(" ", "").startswith("부가가치계")][0]
        for j in range(2, len(codes)):
            if codes[j] is not None and str(codes[j]).startswith("37"):
                recs.append({"연도": int(y), "소분류코드": str(codes[j]), "소분류명": names[j],
                             "부가가치_백만원": int(v[j]), "원본파일": os.path.basename(f)})
    return pd.DataFrame(recs)


def diff_summary(label: str, snap, live) -> dict:
    """스냅샷과 재수집값의 공통 관측 비교(수정 공표는 허용 — 차이를 보고만 한다)."""
    a = pd.DataFrame(snap).astype(float)
    b = pd.DataFrame(live).astype(float)
    a.columns = [str(c) for c in a.columns]
    b.columns = [str(c) for c in b.columns]
    a.index = a.index.map(str)
    b.index = b.index.map(str)
    idx, cols = a.index.intersection(b.index), a.columns.intersection(b.columns)
    d = (b.loc[idx, cols] - a.loc[idx, cols])
    rel = (d.abs() / a.loc[idx, cols].abs().replace(0, np.nan)) * 100
    return {"항목": label, "공통 관측": int(d.notna().sum().sum()),
            "차이 건수": int((d.abs() > 1e-9).sum().sum()),
            "최대 절대차": float(d.abs().max().max()) if d.size else np.nan,
            "최대 상대차(%)": float(rel.max().max()) if rel.size else np.nan,
            "스냅샷에만 있는 행": len(a.index.difference(b.index)), "재수집에만 있는 행": len(b.index.difference(a.index))}


# ── 2. 분석: 36→62 ───────────────────────────────────────────────────
def io_va_by_keit(io_va: pd.Series, MAP: dict) -> dict:
    """IO83 부가가치(2020)를 KEIT62 단위로 합한다."""
    out: dict = defaultdict(float)
    for code, v in io_va.items():
        k = MAP.get(int(code), (None, None))[0]
        if k is not None and pd.notna(v):
            out[k] += float(v)
    return dict(out)


def norm_bok(b: str) -> str:
    """대조표의 '또는'(모호 표기)은 마지막 코드로 확정한다. 병합('a;b')은 유지한다."""
    return b.split("또는")[-1].strip() if "또는" in b else b


def build_levels_62(real36: pd.DataFrame, cw: dict, io_va: pd.Series) -> tuple[pd.DataFrame, pd.DataFrame]:
    """BOK36 분기 실질VA × 몫 → 62개 산업 분기 패널과 배분표.

    한 BOK36(또는 병합 묶음)에 KEIT 하나만 대응하면 몫 1, 여럿이면 2020년 IO83 부가가치 비중으로 나눈다.
    '총량GDP' 열은 36개 산업 실질VA의 단순합이다(국내총생산이 아니다).
    """
    W36 = real36[sorted(real36.columns)]
    va_k = io_va_by_keit(io_va, cw["MAP"])
    groups = defaultdict(list)
    for k, b in cw["keit_bok"].items():
        groups[norm_bok(b)].append(k)
    out, rows = {}, []
    for bok_str, ks in groups.items():
        cols = [c.strip() for c in bok_str.split(";") if c.strip() in W36.columns]
        parent = W36[cols].sum(axis=1)
        if len(ks) == 1:
            k = ks[0]
            out[f"{k}.{cw['keit_name'][k]}"] = parent * 1.0
            rows.append((k, cw["keit_name"][k], bok_str, "1:1/병합", 1.0))
        else:
            tot = sum(va_k.get(k, 0.0) for k in ks) or 1.0
            for k in ks:
                sh = va_k.get(k, 0.0) / tot
                out[f"{k}.{cw['keit_name'][k]}"] = parent * sh
                rows.append((k, cw["keit_name"][k], bok_str, f"분할({len(ks)})", round(sh, 4)))
    W62 = pd.DataFrame(out).reindex(W36.index)
    W62 = W62[sorted(W62.columns, key=lambda c: int(c.split(".")[0]))]
    W62.insert(0, "총량GDP", W36.sum(axis=1))
    W62.index.name = "quarter"
    mapping = pd.DataFrame(rows, columns=["KEIT", "산업명", "BOK36부모", "유형", "몫"]).sort_values("KEIT").reset_index(drop=True)
    return W62, mapping


# ── 2. 분석: 시변 몫(이차전지·신재생) ────────────────────────────────
def io_battery_bench(io_small: pd.DataFrame) -> pd.Series:
    """IO 소분류 373(전지) 부가가치 / 37x(전기장비) 합 — 연도별."""
    s = io_small.astype({"소분류코드": str})
    tot = s.groupby("연도")["부가가치_백만원"].sum()
    bat = s[s["소분류코드"] == "373"].set_index("연도")["부가가치_백만원"]
    return (bat / tot).rename("전지/전기장비")


def quarterly_mean(monthly: pd.Series) -> pd.Series:
    d = pd.to_datetime(monthly.index)
    q = d.year.astype(str) + "Q" + d.quarter.astype(str)
    return monthly.groupby(np.asarray(q)).mean()


def battery_share(mfg: pd.DataFrame, qs: list[str], bench: dict) -> tuple[pd.Series, float, pd.Series]:
    """이차전지 몫 = k·(C282/C28). k = 벤치마크 평균 / proxy(2020Q1–2023Q4) 평균(상수)."""
    c282, c28 = quarterly_mean(mfg["C282"].dropna()), quarterly_mean(mfg["C28"].dropna())
    proxy = (c282 / c28).reindex(qs)
    bench_mean = float(np.mean(list(bench.values())))
    k = bench_mean / proxy.loc["2020Q1":"2023Q4"].mean()
    share = (k * proxy).ffill().bfill()
    share.name, share.index.name = "share_battery", "q"
    return share, float(k), proxy


def renew_actual(kpx: pd.DataFrame, kepco: pd.DataFrame) -> pd.DataFrame:
    """연도별 실측 발전비중표: 2006–2015 KPX(신재생및기타/합계), 2016–2025 KEPCO(6개 발전원 합)."""
    rows = []
    for r in kpx.itertuples():
        if 2006 <= r.연도 <= 2015:
            rows.append({"year": int(r.연도), "renewable_GWh": int(r.신재생및기타_GWh), "total_GWh": int(r.합계_GWh),
                         "renew_share": round(r.신재생및기타_GWh / r.합계_GWh, 6), "source": "KPX(388/TX_38803_A016A)"})
    src = ["원자력", "양수", "석탄", "유류", "가스", "신재생"]
    for _, r in kepco.iterrows():
        tot, ren = float(sum(r[c] for c in src)), float(r["신재생"])
        rows.append({"year": int(r["연도"]), "renewable_GWh": int(ren), "total_GWh": int(tot),
                     "renew_share": round(ren / tot, 6), "source": "KEPCO(전력통계/idx1339)"})
    return pd.DataFrame(rows)


def renew_share(act: pd.DataFrame, kpx: pd.DataFrame, elec_share: float, base2000: float,
                qs: list[str]) -> tuple[pd.Series, dict]:
    """신재생 몫 = 발전비중(연, Q1 기준 다음 해 Q1까지 선형보간; 2025 이후 평탄) × 전력부문 VA몫.

    원 코드(build_share_renew.py)와 같이 2016–2025년은 소수 6자리로 반올림한 발전비중을, 2006–2015년은
    반올림하지 않은 비율을 쓴다. 2000–2005년은 2000년 0.9%에서 2006년 실측까지 선형으로 잇는다.
    """
    by_year = {int(r.year): float(r.renew_share) for r in act.itertuples() if int(r.year) >= 2016}
    for r in kpx.itertuples():
        if 2006 <= r.연도 <= 2015:
            by_year[int(r.연도)] = r.신재생및기타_GWh / r.합계_GWh
    s2006 = by_year[2006]
    for y in range(2000, 2006):
        by_year[y] = base2000 + (s2006 - base2000) * (y - 2000) / (2006 - 2000)
    val = {}
    for qk in qs:
        y, q = int(qk[:4]), int(qk[-1])
        a = by_year[y]
        b = by_year.get(y + 1, a)
        val[qk] = a + (b - a) * (q - 1) / 4.0
    sb = pd.Series(val, name="share_renew") * elec_share
    sb.index.name = ""
    return sb, by_year


def io_share(io_va: pd.Series, new: list[int], allc: list[int]) -> float:
    return float(io_va.reindex(new).sum() / io_va.reindex(allc).sum())


# ── 2. 분석: 62→69 분화 ──────────────────────────────────────────────
def _codes(s) -> list[int]:
    return [int(float(x)) for x in str(s).split(";") if str(x).strip() not in ("", "nan")]


def build_levels_69(L62: pd.DataFrame, rules: pd.DataFrame, io_va: pd.Series,
                    share_batt: pd.Series, share_ren: pd.Series) -> tuple[pd.DataFrame, pd.DataFrame]:
    """62 패널에서 분화 규칙표(carveout_rules.csv) 순서대로 신산업을 떼어 69 패널을 만든다."""
    W = L62.copy()
    col = {int(c.split(".")[0]): c for c in W.columns if c[0].isdigit()}
    timevar = {"share_battery": share_batt.reindex(W.index).ffill().bfill(),
               "share_renew": share_ren.reindex(W.index).ffill().bfill()}
    log = []
    for r in rules.sort_values("순서").itertuples():
        pc = col[int(r.모산업_KEIT)]
        if r.방식 == "IO고정":
            share = io_share(io_va, _codes(r.IO분자), _codes(r.IO분모))
        else:
            share = timevar[r.방식]
        W[r.신산업] = W[pc] * share
        rest = pc + "(잔여)" if r.나머지열 == "(잔여)" else r.나머지열
        W[rest] = W[pc] * (1 - share)
        W.drop(columns=[pc], inplace=True)
        log.append({"신산업": r.신산업, "모산업": pc, "나머지 열": rest, "방식": r.방식,
                    "몫(평균)": round(float(np.mean(np.asarray(share, dtype=float))), 4)})
    return W, pd.DataFrame(log)


def build_levels_agg(gdp: pd.DataFrame, start="1980Q1", end="2025Q4") -> pd.DataFrame:
    """총량 패널 = 실질 GDP(계절조정, 지출측). 두 열은 같은 값이다(총량 모형 입력 형식)."""
    s = gdp.loc[start:end, "실질GDP_십억원"]
    out = pd.DataFrame({"총량GDP": s, "총량": s})
    out.index.name = "분기"
    return out


# ── 2. 분석: IO 연계강도 ─────────────────────────────────────────────
def _normalize(W: pd.DataFrame) -> pd.DataFrame:
    A = W.astype(float).values.copy()
    n = A.shape[0]
    off = A[~np.eye(n, dtype=bool)]
    m = off[off > 0].mean() if (off > 0).any() else 1.0
    A = np.clip(A / m, 0.05, None)
    np.fill_diagonal(A, 1.0)
    return pd.DataFrame(A, index=W.index, columns=W.columns)


def _linkage(io2col: dict, va: dict, M: pd.DataFrame, cols: list[str]) -> pd.DataFrame:
    """W[I,J] = Σ_{i∈I} va_i · Σ_{j∈J} M[i,j] / Σ_{i∈I} va_i, 오프대각 평균 1로 정규화(하한 0.05)."""
    Mmat = M.pivot(index="i", columns="j", values="v")
    grp: dict = {}
    for io in [c for c in Mmat.index if c in io2col]:
        grp.setdefault(io2col[io], []).append(io)
    W = pd.DataFrame(0.0, index=cols, columns=cols)
    for I in cols:
        iset = grp.get(I, [])
        if not iset:
            continue
        wsum = sum(va.get(i, 0.0) for i in iset) or 1.0
        for J in cols:
            jset = grp.get(J, [])
            if not jset:
                continue
            val = sum(va.get(i, 0.0) * Mmat.loc[i, [j for j in jset if j in Mmat.columns]].sum() for i in iset)
            W.loc[I, J] = val / wsum
    return _normalize(W)


def io_linkage(L62: pd.DataFrame, L69: pd.DataFrame, cw: dict, io_va: pd.Series, M: pd.DataFrame,
               carve_io: dict) -> tuple[pd.DataFrame, pd.DataFrame, int]:
    """62·69 산업 IO 연계강도 행렬. 이차전지·신재생은 IO83에 별도 코드가 없어 부모 잔여와 같은 연계를 갖지 않는다."""
    MAP = cw["MAP"]
    va = io_va.to_dict()
    cols62 = [c for c in L62.columns if c != "총량GDP"]
    keit2col = {int(c.split(".")[0]): c for c in cols62}
    io2col62 = {io: keit2col[MAP[io][0]] for io in MAP if MAP[io][0] in keit2col}
    W62 = _linkage(io2col62, va, M, cols62)
    cols69 = [c for c in L69.columns if c != "총량GDP"]

    def res_col(k):
        for c in cols69:
            if c.startswith(f"{k}.") and c.endswith("(잔여)"):
                return c
        return None

    io2col69 = {}
    for io in MAP:
        k = MAP[io][0]
        if io in carve_io and carve_io[io] in cols69:
            io2col69[io] = carve_io[io]
        else:
            r, base = res_col(k), keit2col.get(k)
            if base in cols69:
                io2col69[io] = base
            elif r:
                io2col69[io] = r
            elif carve_io.get(io) in cols69:
                io2col69[io] = carve_io[io]
    W69 = _linkage(io2col69, va, M, cols69)
    return W62, W69, len(set(io2col69.values()))


# ── 2. 분석: 연쇄가중 총량 재현 ──────────────────────────────────────
def weighted_growth(nom_base: pd.Series, real_base: pd.Series, real_next: pd.Series) -> float:
    """기준 시점 명목 비중으로 가중한 산업별 실질 성장배율의 합(라스파이레스형 연쇄 고리)."""
    w = nom_base / nom_base.sum()
    return float((w * (real_next / real_base)).sum())


def reconstruct(real: pd.DataFrame, nom: pd.DataFrame, total: pd.Series, anchor: str) -> pd.Series:
    """기준점의 재현값을 공식 총량으로 두고, 앞쪽은 곱하고 뒤쪽은 나누어 양방향으로 잇는다."""
    qs = list(real.index)
    i0 = qs.index(anchor)
    rec = pd.Series(index=qs, dtype=float)
    rec.loc[anchor] = total.loc[anchor]
    for i in range(i0 + 1, len(qs)):
        t, tm1 = qs[i], qs[i - 1]
        rec.loc[t] = rec.loc[tm1] * weighted_growth(nom.loc[tm1], real.loc[tm1], real.loc[t])
    for i in range(i0 - 1, -1, -1):
        t, tp1 = qs[i], qs[i + 1]
        rec.loc[t] = rec.loc[tp1] / weighted_growth(nom.loc[t], real.loc[t], real.loc[tp1])
    return rec


def recon_table(real: pd.DataFrame, nom: pd.DataFrame, gva: pd.Series, anchor: str = ANCHOR) -> pd.DataFrame:
    """분기별 36개 단순합·재현값과 총부가가치 대비 괴리(%)."""
    gva = gva.reindex(real.index)
    rec = reconstruct(real, nom, gva, anchor)
    s = real.sum(axis=1)
    out = pd.DataFrame({"36합": s, "총부가가치": gva, "재현값": rec,
                        "단순합_괴리_pct": 100 * (s / gva - 1), "재현_괴리_pct": 100 * (rec / gva - 1)})
    out.index.name = "분기"
    return out


def recon_annual(real: pd.DataFrame, nom: pd.DataFrame, gva: pd.Series, anchor_year: str = "2020") -> pd.DataFrame:
    """분기 계절조정 계열을 연간 합으로 묶어 같은 방식으로 연 단위 연쇄를 재현한다."""
    yr = [q[:4] for q in real.index]
    RA, NA, GA = real.groupby(yr).sum(), nom.groupby(yr).sum(), gva.reindex(real.index).groupby(yr).sum()
    rec = reconstruct(RA, NA, GA, anchor_year)
    out = pd.DataFrame({"36합": RA.sum(axis=1), "총부가가치": GA, "재현값": rec,
                        "단순합_괴리_pct": 100 * (RA.sum(axis=1) / GA - 1), "재현_괴리_pct": 100 * (rec / GA - 1)})
    out.index.name = "연도"
    return out


# ── 3. 결과: 공표 산출물 대조 ────────────────────────────────────────
def max_abs_diff(new: pd.DataFrame, ref_path: str, index_col=0) -> tuple[float, bool, bool]:
    """(최대 절대차, 행 일치, 열 일치). 공표 CSV와 계산 결과를 같은 형식으로 읽어 비교한다."""
    ref = pd.read_csv(ref_path, index_col=index_col)
    a = new.copy()
    a.index, ref.index = a.index.map(str), ref.index.map(str)
    same_rows, same_cols = list(a.index) == list(ref.index), list(map(str, a.columns)) == list(map(str, ref.columns))
    num = ref.select_dtypes("number").columns
    d = (a[num].astype(float) - ref[num].astype(float)).abs().max().max()
    return float(d), same_rows, same_cols


def panel_summary(real36: pd.DataFrame, L62: pd.DataFrame, L69: pd.DataFrame, show: list[str]) -> pd.DataFrame:
    """산업 해상도별(36·62·69) 산업 수·기간·선택 분기 합계(조원)·36개 단순합 대비 최대 오차(십억원)."""
    s36 = real36.loc[L62.index].sum(axis=1)
    rows = []
    for name, df in [("36(한국은행)", real36.loc[L62.index]), ("62(KEIT)", L62.drop(columns="총량GDP")),
                     ("69(62+분화)", L69.drop(columns="총량GDP"))]:
        s = df.sum(axis=1)
        row = {"분류": name, "산업 수": df.shape[1], "기간": f"{df.index[0]}–{df.index[-1]}"}
        row.update({f"{q} 합계(조원)": s.loc[q] / 1e3 for q in show})
        row["36합 대비 최대오차(십억원)"] = float((s - s36).abs().max())
        rows.append(row)
    return pd.DataFrame(rows)


def mece_gap(L62: pd.DataFrame, L69: pd.DataFrame, rules: pd.DataFrame) -> float:
    """분화 항등식: 신산업 + 나머지 열 = 모산업(62). 모든 분기·규칙의 최대 절대오차(십억원)."""
    col = {int(c.split(".")[0]): c for c in L62.columns if c[0].isdigit()}
    gap = 0.0
    for r in rules.itertuples():
        pc = col[int(r.모산업_KEIT)]
        rest = pc + "(잔여)" if r.나머지열 == "(잔여)" else r.나머지열
        gap = max(gap, float((L69[r.신산업] + L69[rest] - L62[pc]).abs().max()))
    return gap