# -*- coding: utf-8 -*- """ 공동주택 공시가격 수집 + 실거래가 결합 ===================================== VWorld NED API(getApartHousingPriceAttr)로 PNU·기준연도별 공동주택 공시가격을 호(단위)별로 받아 캐싱하고, 실거래가와 (PNU + 전용면적)으로 결합해 '실거래가 대비 공시가격' 분석용 데이터를 만든다. - 대상: 아파트, 연립다세대 (둘 다 공동주택가격 대상) - 오피스텔/단독: 공동주택가격 대상 아님 → 제외 - 서비스URL(http://localhost)을 Referer로 전송해야 데이터 API 통과 """ from __future__ import annotations import os import sys import time from pathlib import Path import pandas as pd import requests import realestate as re # get_vworld_key 재사용 try: sys.stdout.reconfigure(encoding="utf-8") except Exception: pass RESULT_DIR = Path(__file__).with_name("결과") CACHE_DIR = RESULT_DIR / "공시가격캐시" APT_PRICE_URL = "https://api.vworld.kr/ned/data/getApartHousingPriceAttr" # 응답에서 보관할 필드 KEEP_FIELDS = [ "pnu", "stdrYear", "aphusNm", "prvuseAr", "pblntfPc", "dongNm", "hoNm", "floorNm", "regstrSeCodeNm", "ldCodeNm", ] def _referer() -> str: re._hydrate() return os.environ.get("VWORLD_REFERER", "http://localhost").strip() def fetch_apt_price_pnu(pnu: str, year: str | int, *, session: requests.Session, key: str, ref: str, num_of_rows: int = 1000, max_pages: int = 20) -> list[dict]: """PNU 1개의 공동주택 공시가격(해당 연도) 호별 목록. totalCount 기준 페이지네이션.""" out: list[dict] = [] page = 1 while page <= max_pages: params = { "key": key, "pnu": pnu, "stdrYear": str(year), "format": "json", "numOfRows": num_of_rows, "pageNo": page, "domain": ref, } r = session.get(APT_PRICE_URL, params=params, headers={"Referer": ref}, timeout=30) r.raise_for_status() try: node = r.json().get("apartHousingPrices") except Exception: break if not node: break fields = node.get("field", []) if isinstance(fields, dict): fields = [fields] for f in fields: out.append({k: f.get(k) for k in KEEP_FIELDS}) total = int(node.get("totalCount") or 0) if page * num_of_rows >= total or not fields: break page += 1 return out def fetch_apt_prices(pnus: list[str], year: str | int, *, cache_name: str, sleep_sec: float = 0.1, save_every: int = 200) -> pd.DataFrame: """여러 PNU의 공시가격을 캐싱하며 수집. 이미 캐시에 있는 PNU는 건너뜀.""" CACHE_DIR.mkdir(parents=True, exist_ok=True) cache_path = CACHE_DIR / f"{cache_name}_{year}.csv" done: set[str] = set() if cache_path.exists(): prev = pd.read_csv(cache_path, dtype=str) done = set(prev["조회PNU"].unique()) print(f" 캐시 {len(prev)}행 / PNU {len(done)}개 로드 (재사용)") else: prev = pd.DataFrame() key = re.get_vworld_key() ref = _referer() session = requests.Session() todo = [p for p in pnus if p not in done] print(f" 수집 대상 PNU: {len(todo)} (전체 {len(pnus)}, 완료 {len(done)})") buf: list[dict] = [] fetched = 0 for i, pnu in enumerate(todo, 1): try: rows = fetch_apt_price_pnu(pnu, year, session=session, key=key, ref=ref) for row in rows: row["조회PNU"] = pnu buf.append(row) fetched += 1 except requests.RequestException as e: buf.append({"조회PNU": pnu, "_error": str(e)[:100]}) if i % 50 == 0: print(f" ...{i}/{len(todo)} (누적 {len(buf)}행)") if i % save_every == 0: _flush(prev, buf, cache_path); prev = pd.read_csv(cache_path, dtype=str); buf = [] time.sleep(sleep_sec) _flush(prev, buf, cache_path) final = pd.read_csv(cache_path, dtype=str) print(f" 완료: {cache_path} ({len(final)}행)") return final def _flush(prev: pd.DataFrame, buf: list[dict], path: Path) -> None: if not buf: return add = pd.DataFrame(buf) combined = pd.concat([prev, add], ignore_index=True) if not prev.empty else add combined.to_csv(path, index=False, encoding="utf-8-sig") # --------------------------------------------------------------------------- # 결합 # --------------------------------------------------------------------------- def _price_table(cache_name: str, years: list[int]) -> pd.DataFrame: """여러 연도 캐시를 합쳐 (PNU, 전용_r, 기준연도) -> 대표 공시가격(중앙값) 표.""" frames = [] for y in years: path = CACHE_DIR / f"{cache_name}_{y}.csv" if not path.exists(): continue p = pd.read_csv(path, dtype=str) p = p[p["pnu"].notna()].copy() if p.empty: continue p["전용_r"] = pd.to_numeric(p["prvuseAr"], errors="coerce").round(0) p["공시가격_원"] = pd.to_numeric(p["pblntfPc"], errors="coerce") p["기준연도"] = str(y) frames.append(p[["pnu", "전용_r", "기준연도", "공시가격_원"]]) if not frames: return pd.DataFrame(columns=["PNU", "전용_r", "기준연도", "공시가격_원"]) allp = pd.concat(frames, ignore_index=True) agg = (allp.groupby(["pnu", "전용_r", "기준연도"])["공시가격_원"] .median().reset_index().rename(columns={"pnu": "PNU"})) return agg def join_price(trade_types: list[str], cache_name: str, years: list[int]) -> pd.DataFrame: """실거래가(매매) ↔ 공시가격 결합: (PNU + 전용면적 + 거래연도==기준연도).""" trades = pd.read_csv(RESULT_DIR / "분석_실거래가_매칭키.csv", encoding="utf-8-sig", dtype=str) trades = trades[trades["자료구분"].isin(trade_types)].dropna(subset=["PNU"]).copy() trades["전용_r"] = pd.to_numeric(trades["전용면적_㎡"], errors="coerce").round(0) trades["기준연도"] = trades["거래연도"].astype(str) agg = _price_table(cache_name, years) merged = trades.merge(agg, on=["PNU", "전용_r", "기준연도"], how="left") merged["거래금액_원"] = pd.to_numeric(merged["거래금액_원"], errors="coerce") # 현실화율 = 공시가격 / 실거래가 (공시가격이 실거래가의 몇 %인지) merged["현실화율"] = (merged["공시가격_원"] / merged["거래금액_원"]).round(4) return merged YEARS = list(range(2019, 2025)) # 2019~2024 JOBS = { "아파트": ["아파트_매매"], "연립다세대": ["연립다세대_매매"], } def main() -> None: trades = pd.read_csv(RESULT_DIR / "분석_실거래가_매칭키.csv", encoding="utf-8-sig", dtype=str) trades["거래연도"] = pd.to_numeric(trades["거래연도"], errors="coerce") # 1) 연도별 공시가격 수집: 각 연도에 거래가 있었던 PNU만 for cache_name, types in JOBS.items(): sub = trades[trades["자료구분"].isin(types)].dropna(subset=["PNU"]) for y in YEARS: pnus = sorted(sub[sub["거래연도"] == y]["PNU"].unique()) if not pnus: continue print(f"\n■ {cache_name} {y}: 대상 PNU {len(pnus)}") fetch_apt_prices(pnus, y, cache_name=cache_name) # 2) 결합 print("\n■ 결합 (거래연도=공시기준연도)") frames = [] for cache_name, types in JOBS.items(): m = join_price(types, cache_name, YEARS) matched = m["공시가격_원"].notna().sum() print(f" {cache_name}: {len(m)}행 중 공시가격 매칭 {matched} ({100*matched/max(len(m),1):.1f}%)") frames.append(m) out = pd.concat(frames, ignore_index=True) out.to_csv(RESULT_DIR / "분석_실거래가_공시가격_결합.csv", index=False, encoding="utf-8-sig") print(f"\n저장: 결과/분석_실거래가_공시가격_결합.csv ({len(out)}행)") if __name__ == "__main__": main()