Python

requests + pandas 로 Veacon 데이터를 DataFrame 으로 즉시 활용.

Last updated: 2026-04-23

데이터 리서치 팀이 가장 자주 쓰는 조합. 별도 패키지 설치 불필요 (requests + pandas 만 있으면 됨).

설치

bash
pip install requests pandas
# 선택: python-dotenv — .env 파일에서 API Key 로드
pip install python-dotenv

환경변수 로드

python
import os
from dotenv import load_dotenv

load_dotenv()  # .env 에 VEACON_API_KEY=... 저장
API_KEY = os.environ["VEACON_API_KEY"]

1. 최소 예제

python
import requests

r = requests.get(
    "https://veacon.io/api/v1/markets/pulse",
    headers={"X-API-Key": API_KEY},
    params={"region": "강남권", "category": "office", "period": "2026-01"},
    timeout=10,
)
r.raise_for_status()
data = r.json()["data"]
print(data[0]["avg_price"])  # 2577493.75

2. pandas DataFrame

python
import pandas as pd
import requests

def pulse_df(regions, categories, periods, api_key):
    rows = []
    for region in regions:
        for category in categories:
            for period in periods:
                r = requests.get(
                    "https://veacon.io/api/v1/markets/pulse",
                    headers={"X-API-Key": api_key},
                    params={"region": region, "category": category, "period": period},
                    timeout=10,
                )
                if r.status_code == 200:
                    rows.extend(r.json()["data"])
                elif r.status_code == 404:
                    pass  # 해당 조합 데이터 없음 — skip
                elif r.status_code == 402:
                    print(f"Quota exceeded; used={r.headers.get('X-Quota-Used')}")
                    break
                elif r.status_code == 429:
                    retry = int(r.headers.get("Retry-After", "5"))
                    time.sleep(retry)
                else:
                    r.raise_for_status()
    return pd.DataFrame(rows)

df = pulse_df(
    regions=["강남권", "도심권", "서남권"],
    categories=["office"],
    periods=["2026-01", "2026-02", "2026-03"],
    api_key=API_KEY,
)

# 분석 예: 권역별 평균 가격
print(df.groupby("region")["avg_price"].agg(["mean", "median", "count"]))

3. 재시도 + Rate Limit 안전

python
import time
import requests

def pulse(region, category, period, api_key, max_retries=4):
    attempt = 0
    while attempt < max_retries:
        r = requests.get(
            "https://veacon.io/api/v1/markets/pulse",
            headers={"X-API-Key": api_key},
            params={"region": region, "category": category, "period": period},
            timeout=10,
        )

        if r.status_code == 200:
            return r.json()["data"]

        if r.status_code == 429:
            wait = int(r.headers.get("Retry-After", "1"))
            print(f"Rate limited — waiting {wait}s")
            time.sleep(wait)
            attempt += 1
            continue

        if r.status_code >= 500 and attempt < max_retries - 1:
            wait = 2 ** attempt
            time.sleep(wait)
            attempt += 1
            continue

        # 4xx (other than 429) — propagate
        r.raise_for_status()

    raise RuntimeError("Max retries exceeded")

4. Jupyter Notebook: 간단 리포트

python
import matplotlib.pyplot as plt

df = pulse_df(["강남권", "도심권", "서남권"], ["office"],
              ["2026-01", "2026-02", "2026-03"], API_KEY)

# 권역별 월간 중간값 trend
pivot = df.pivot_table(index="period", columns="region", values="median_price")
pivot.plot(kind="line", marker="o", figsize=(10, 5),
           title="서울 상업용 오피스 월간 중간값 추이 (Veacon)")
plt.ylabel("월세 (원)")
plt.grid(alpha=0.3)
plt.tight_layout()
plt.show()

5. 쿼터 관리

python
def check_quota(api_key):
    r = requests.get(
        "https://veacon.io/api/v1/markets/dimensions",
        headers={"X-API-Key": api_key},
    )
    return {
        "limit": int(r.headers.get("X-Quota-Limit", "0")),
        "used": int(r.headers.get("X-Quota-Used", "0")),
        "resets_at": r.headers.get("X-Quota-Resets-At"),
    }

q = check_quota(API_KEY)
pct = 100 * q["used"] / max(q["limit"], 1)
print(f"Quota {q['used']}/{q['limit']} ({pct:.1f}%) — resets at {q['resets_at']}")

6. 데이터 캐싱 (권장)

동일 조합을 반복 호출하지 말고 로컬 캐시:

python
from functools import lru_cache
import json

@lru_cache(maxsize=1024)
def pulse_cached(region, category, period):
    return tuple(json.dumps(row) for row in pulse(region, category, period, API_KEY))

# 같은 파라미터 재호출 시 쿼터 사용 안 됨 (프로세스 생존 기간 동안)

Disk 캐시가 필요하면 requests-cache 또는 diskcache 사용:

python
from requests_cache import CachedSession

session = CachedSession("veacon_cache", expire_after=3600)  # 1시간
r = session.get("https://veacon.io/api/v1/markets/pulse", ...)

참고