diff --git a/realestate-lab/app/listing_collector.py b/realestate-lab/app/listing_collector.py
new file mode 100644
index 0000000..5778599
--- /dev/null
+++ b/realestate-lab/app/listing_collector.py
@@ -0,0 +1,166 @@
+"""매물 수집: 국토부 실거래가(합법, market_deals) + 네이버부동산 호가(회색·폴백, listings)."""
+import os
+import re
+import time
+import logging
+import requests
+from datetime import date
+
+from .lawd_codes import LAWD_CODES
+from .db import (get_listing_criteria, upsert_listing, upsert_market_deal,
+                 save_listing_collect_log)
+
+logger = logging.getLogger("realestate-lab")
+
+MOLIT_KEY = os.getenv("DATA_GO_KR_API_KEY", "")
+NAVER_BASE = "https://new.land.naver.com/api"
+# MOLIT operation ID (스펙 §3, 라이브 확인 필요). house_type → (rent_op, trade_op)
+MOLIT_OPS = {
+    "아파트": ("getRTMSDataSvcAptRent", "getRTMSDataSvcAptTradeDev"),
+    "오피스텔": ("getRTMSDataSvcOffiRent", "getRTMSDataSvcOffiTrade"),
+    "연립다세대": ("getRTMSDataSvcRHRent", "getRTMSDataSvcRHTrade"),
+}
+_HOUSE_TYPE_ALIAS = {"빌라": "연립다세대"}
+
+
+def _won_to_manwon(s) -> int | None:
+    """'43,000' 또는 '2억 9,000' → 만원 정수."""
+    if s is None:
+        return None
+    t = str(s).replace(",", "").strip()
+    if not t:
+        return None
+    m = re.match(r"(?:(\d+)억)?\s*(\d+)?", t)
+    if m and (m.group(1) or m.group(2)):
+        eok = int(m.group(1)) if m.group(1) else 0
+        man = int(m.group(2)) if m.group(2) else 0
+        return eok * 10000 + man
+    try:
+        return int(float(t))
+    except ValueError:
+        return None
+
+
+def _parse_molit_rent(raw: dict, house_type: str) -> dict:
+    deposit = _won_to_manwon(raw.get("deposit"))
+    rent = _won_to_manwon(raw.get("monthlyRent"))
+    return {"house_type": house_type, "complex_name": (raw.get("aptNm") or raw.get("offiNm") or "").strip(),
+            "area": float(raw.get("excluUseAr") or 0) or None, "deal_type": "전세" if not rent else "월세",
+            "deposit": deposit, "monthly_rent": rent, "sale_amount": None,
+            "deal_ym": f"{raw.get('dealYear','')}{str(raw.get('dealMonth','')).zfill(2)}",
+            "floor": str(raw.get("floor") or ""), "source": "molit"}
+
+
+def _parse_molit_trade(raw: dict, house_type: str) -> dict:
+    return {"house_type": house_type, "complex_name": (raw.get("aptNm") or raw.get("offiNm") or "").strip(),
+            "area": float(raw.get("excluUseAr") or 0) or None, "deal_type": "매매",
+            "deposit": None, "monthly_rent": None,
+            "sale_amount": _won_to_manwon(raw.get("dealAmount")),
+            "deal_ym": f"{raw.get('dealYear','')}{str(raw.get('dealMonth','')).zfill(2)}",
+            "floor": str(raw.get("floor") or ""), "source": "molit"}
+
+
+def _parse_naver_article(raw: dict) -> dict:
+    dt = raw.get("tradeTypeName") or ""
+    prc = _won_to_manwon(raw.get("dealOrWarrantPrc"))
+    rent = _won_to_manwon(raw.get("rentPrc"))
+    is_sale = dt == "매매"
+    return {"article_no": str(raw.get("articleNo") or ""), "source": "naver", "deal_type": dt,
+            "deposit": None if is_sale else prc, "monthly_rent": rent,
+            "sale_price": prc if is_sale else None,
+            "area_exclusive": float(raw.get("area2") or raw.get("areaName") or 0) or None,
+            "floor": raw.get("floorInfo"), "complex_name": raw.get("articleName") or raw.get("buildingName"),
+            "dong_code": (raw.get("cortarNo") or "")[:5] or None,
+            "url": f"https://new.land.naver.com/houses?articleNo={raw.get('articleNo')}",
+            "raw_json": None}
+
+
+def _molit_call(op: str, lawd: str, ym: str) -> list:
+    """국토부 실거래 단건 호출 (기존 collector 재시도 관용). 실패 시 []."""
+    if not MOLIT_KEY:
+        return []
+    url = f"https://apis.data.go.kr/1613000/RTMSDataSvc{op[len('getRTMSDataSvc'):]}/{op}"
+    for attempt in range(3):
+        try:
+            resp = requests.get(url, params={"serviceKey": MOLIT_KEY, "LAWD_CD": lawd,
+                                             "DEAL_YMD": ym, "numOfRows": 200, "_type": "json"}, timeout=30)
+            resp.raise_for_status()
+            body = resp.json()
+            items = (((body.get("response") or {}).get("body") or {}).get("items") or {}).get("item") or []
+            return items if isinstance(items, list) else [items]
+        except requests.RequestException as e:
+            if attempt < 2:
+                time.sleep(2 ** attempt)
+            else:
+                logger.error("MOLIT %s 실패: %s", op, e)
+    return []
+
+
+def _fetch_molit_all() -> int:
+    """대상 동 시군구코드 × 유형 × 최근 2개월 전월세·매매 실거래 수집→market_deals. 저장건수 반환."""
+    saved = 0
+    lawds = set(LAWD_CODES.values())
+    today = date.today()
+    yms = [today.strftime("%Y%m"),
+           (today.replace(day=1) - __import__("datetime").timedelta(days=1)).strftime("%Y%m")]
+    for lawd in lawds:
+        for htype, (rent_op, trade_op) in MOLIT_OPS.items():
+            for ym in yms:
+                for op, parser in ((rent_op, _parse_molit_rent), (trade_op, _parse_molit_trade)):
+                    for raw in _molit_call(op, lawd, ym):
+                        try:
+                            d = parser(raw, htype)
+                            d["dong_code"] = lawd
+                            d["dong"] = (raw.get("umdNm") or "").strip() or None
+                            upsert_market_deal(d)
+                            saved += 1
+                        except Exception as e:
+                            logger.warning("MOLIT 파싱 실패: %s", e)
+                    time.sleep(0.3)
+    return saved
+
+
+def _fetch_naver_all() -> tuple:
+    """대상 동 네이버 호가 매물 수집→listings. (new_count, total_count) 반환. 예외는 상위로 전파."""
+    new_count = total = 0
+    crit = get_listing_criteria()
+    for dong in crit.get("dongs", []):
+        code = LAWD_CODES.get(dong)
+        if not code:
+            continue
+        # 저빈도: 동당 1회, UA/Referer 현실값. 응답 articleList 순회.
+        resp = requests.get(f"{NAVER_BASE}/articles",
+                            params={"cortarNo": code, "order": "dateDesc", "page": 1},
+                            headers={"User-Agent": "Mozilla/5.0", "Referer": "https://new.land.naver.com/"},
+                            timeout=20)
+        resp.raise_for_status()
+        for raw in (resp.json().get("articleList") or []):
+            try:
+                d = _parse_naver_article(raw)
+                if not d["article_no"]:
+                    continue
+                d["dong"] = dong
+                _, is_new = upsert_listing(d)
+                total += 1
+                if is_new:
+                    new_count += 1
+            except Exception as e:
+                logger.warning("네이버 파싱 실패: %s", e)
+        time.sleep(1.5)
+    return new_count, total
+
+
+def collect_listings() -> dict:
+    naver_ok = True
+    new_count = total = 0
+    try:
+        _fetch_molit_all()   # 실거래(합법)는 항상 우선 — 안전마진 baseline
+    except Exception as e:
+        logger.error("MOLIT 수집 오류(안전마진 baseline): %s", e)
+    try:
+        new_count, total = _fetch_naver_all()
+    except requests.RequestException as e:
+        naver_ok = False
+        logger.warning("네이버 매물수집 skip(차단/오류) — 안전마진·실거래는 유지: %s", e)
+    save_listing_collect_log(new_count, total, naver_ok)
+    return {"new_count": new_count, "total_count": total, "naver_ok": naver_ok}
diff --git a/realestate-lab/tests/test_listing_collector.py b/realestate-lab/tests/test_listing_collector.py
new file mode 100644
index 0000000..c0bb3dc
--- /dev/null
+++ b/realestate-lab/tests/test_listing_collector.py
@@ -0,0 +1,29 @@
+from unittest.mock import patch
+
+def test_parse_molit_rent_fields():
+    from app.listing_collector import _parse_molit_rent
+    raw = {"aptNm": "OO아파트", "excluUseAr": "42.5", "deposit": "43,000",
+           "monthlyRent": "0", "dealYear": "2026", "dealMonth": "5", "floor": "5"}
+    d = _parse_molit_rent(raw, "아파트")
+    assert d["complex_name"] == "OO아파트" and d["area"] == 42.5
+    assert d["deposit"] == 43000 and d["deal_type"] == "전세" and d["deal_ym"] == "202605"
+
+def test_parse_naver_article_fields():
+    from app.listing_collector import _parse_naver_article
+    raw = {"articleNo": "A123", "tradeTypeName": "전세", "dealOrWarrantPrc": "2억 9,000",
+           "areaName": "42", "area2": "42.5", "floorInfo": "5/15", "articleName": "OO아파트",
+           "buildingName": "OO", "cortarNo": "1159010100"}
+    d = _parse_naver_article(raw)
+    assert d["article_no"] == "A123" and d["deal_type"] == "전세" and d["deposit"] == 29000
+    assert d["source"] == "naver"
+
+def test_collect_naver_block_sets_naver_ok_false(monkeypatch):
+    """네이버 403 → 매물수집 skip + naver_ok=False, 실거래는 계속."""
+    from app import listing_collector as lc
+    monkeypatch.setattr(lc, "_fetch_molit_all", lambda: 2)      # 실거래 2건 저장됨 가정
+    def boom(*a, **k):
+        import requests
+        raise requests.RequestException("403")
+    monkeypatch.setattr(lc, "_fetch_naver_all", boom)
+    r = lc.collect_listings()
+    assert r["naver_ok"] is False   # 안전마진(실거래)은 유지