"""관세청/산업통상부 수출입동향 → 섹터별 수출 추세(sector_export_trend) 산출기. 실측 결과(2026-06-21 세션): investing.com 직접 스크래핑은 403(Cloudflare)으로 차단되고, 관세청·산업통상부는 실시간 무인증 JSON API를 공개하지 않는다(통계청/관세청 수출입통계는 data.go.kr 공공데이터포털의 서비스키 기반 OpenAPI 또는 매월 발표되는 보도자료 첨부 XLSX/CSV로만 배포). 따라서 이 모듈은 두 경로를 모두 지원한다: 1) API 경로 — data.go.kr 관세청 수출입통계 API. CUSTOMS_API_KEY 환경변수(또는 --api-key) 필요. 키가 없거나 호출 실패 시 추정하지 않고 DATA_MISSING 반환. 2) CSV 경로(권장, 안정적) — 관세청 수출입무역통계(https://unipass.customs.go.kr/ets/) 또는 산업통상부 보도자료에서 사용자가 다운로드한 월별 HS코드별 수출입 CSV를 --csv 인자로 입력. 이 경로가 실패할 일이 없어 1차 권장 경로다. 산출물 sector_export_trend(%, MoM 또는 YoY)는 qualitative_sell_strategy_v1의 fundamental_trajectory 보강 입력 및 compute_satellite_candidate_score의 1차 팩터로 쓰인다. """ from __future__ import annotations import argparse import csv import json import os import sys from collections import defaultdict from pathlib import Path from typing import Any import requests ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) # 섹터 → HS코드 prefix(2~4자리). 위성종목 추천/매도판단에 쓰는 핵심 수출 섹터만 우선 등록. SECTOR_HS_MAP: dict[str, tuple[str, ...]] = { "반도체": ("8541", "8542"), "자동차": ("8701", "8702", "8703", "8704"), "2차전지": ("8507",), "조선": ("8901", "8902", "8905"), "철강": ("72",), "석유화학": ("29", "39"), "디스플레이": ("8524", "9013"), "기계": ("84",), "바이오": ("30",), # universe.Sector 실측 라벨이 "바이오"(헬스 접미사 없음) — 그대로 매칭 "방산": ("93",), # 무기류·탄약(HS Ch.93) — 현대로템 등 보유종목 K-방산 테마 대응 } CUSTOMS_API_BASE = "https://apis.data.go.kr/1220000/nitemtrade/getNitemtradeList" def fetch_customs_trade_api( session: requests.Session, api_key: str | None, hs_code: str, start_ym: str, end_ym: str, ) -> dict[str, Any]: """data.go.kr 관세청 수출입통계 API 호출. 키 없거나 실패 시 DATA_MISSING(추정 금지).""" if not api_key: return {"status": "DATA_MISSING", "note": "CUSTOMS_API_KEY 미설정 — --csv 경로 사용 권장"} try: resp = session.get( CUSTOMS_API_BASE, params={ "serviceKey": api_key, "strtYymm": start_ym, "endYymm": end_ym, "hsSgn": hs_code, "type": "json", }, timeout=15, ) resp.raise_for_status() data = resp.json() except Exception as exc: # noqa: BLE001 — 외부 API 실패는 광범위하게 잡아 DATA_MISSING 처리 return {"status": "API_ERROR", "note": str(exc)} return {"status": "OK", "raw": data, "source_url": CUSTOMS_API_BASE} def load_trade_statistics_csv(path: Path) -> list[dict[str, Any]]: """관세청/산업통상부 배포 CSV. 컬럼: 기간(YYYYMM), HS코드, 수출액(달러), 수입액(달러). 헤더명은 배포처마다 다를 수 있어 한글/영문 별칭을 모두 허용한다. """ alias = { "기간": "period", "year_month": "period", "period": "period", "hs코드": "hs_code", "hs_code": "hs_code", "hscode": "hs_code", "수출액": "export_usd", "export": "export_usd", "export_usd": "export_usd", "수입액": "import_usd", "import": "import_usd", "import_usd": "import_usd", } rows: list[dict[str, Any]] = [] with path.open(encoding="utf-8-sig", newline="") as f: reader = csv.DictReader(f) for raw_row in reader: row: dict[str, Any] = {} for key, value in raw_row.items(): norm_key = alias.get(str(key).strip().lower()) if norm_key: row[norm_key] = value if {"period", "hs_code"}.issubset(row): for money_field in ("export_usd", "import_usd"): if money_field in row: try: row[money_field] = float(str(row[money_field]).replace(",", "")) except ValueError: row[money_field] = 0.0 rows.append(row) return rows def compute_sector_export_trend( rows: list[dict[str, Any]], sector: str, compare: str = "yoy", ) -> dict[str, Any]: """sector_export_trend(%) = 최신월 수출액 / 비교월 수출액 - 1. compare="yoy": 12개월 전 동월 대비. compare="mom": 직전월 대비. 데이터 부족 시 추정하지 않고 DATA_MISSING. """ hs_prefixes = SECTOR_HS_MAP.get(sector) if not hs_prefixes: return {"status": "UNKNOWN_SECTOR", "sector": sector, "known_sectors": list(SECTOR_HS_MAP)} by_period: dict[str, float] = defaultdict(float) for row in rows: hs_code = str(row.get("hs_code") or "") if any(hs_code.startswith(prefix) for prefix in hs_prefixes): period = str(row.get("period") or "") by_period[period] += float(row.get("export_usd") or 0.0) if len(by_period) < 2: return {"status": "DATA_MISSING", "sector": sector, "note": "기간별 수출액 표본 부족"} periods_sorted = sorted(by_period) latest_period = periods_sorted[-1] latest_value = by_period[latest_period] if compare == "mom": compare_period = periods_sorted[-2] else: latest_ym = int(latest_period) target_ym = latest_ym - 100 # YYYYMM에서 12개월 전 = -100 compare_period = str(target_ym) if compare_period not in by_period: return {"status": "DATA_MISSING", "sector": sector, "note": f"YoY 비교월({compare_period}) 데이터 없음 — MoM으로 재시도 권장"} compare_value = by_period.get(compare_period, 0.0) if compare_value <= 0: return {"status": "DATA_MISSING", "sector": sector, "note": "비교월 수출액이 0 이하"} trend_pct = round((latest_value / compare_value - 1.0) * 100.0, 4) return { "status": "OK", "sector": sector, "compare": compare, "latest_period": latest_period, "compare_period": compare_period, "sector_export_trend": trend_pct, } def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--csv", type=Path, help="관세청/산업통상부 배포 수출입 CSV 경로(권장 경로)") ap.add_argument("--sector", default="반도체", choices=list(SECTOR_HS_MAP)) ap.add_argument("--compare", default="yoy", choices=["yoy", "mom"]) ap.add_argument("--api-key", default=os.environ.get("CUSTOMS_API_KEY")) ap.add_argument("--hs-code", default="", help="API 경로 사용 시 HS코드") ap.add_argument("--start-ym", default="") ap.add_argument("--end-ym", default="") args = ap.parse_args() if args.csv: rows = load_trade_statistics_csv(args.csv) result = compute_sector_export_trend(rows, args.sector, args.compare) else: session = requests.Session() result = fetch_customs_trade_api(session, args.api_key, args.hs_code, args.start_ym, args.end_ym) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())