QuantEngineByItz/tools/fetch_trade_statistics_motie_v1.py

"""관세청/산업통상부 수출입동향 → 섹터별 수출 추세(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())