fix(realestate): 시세 표본 광역폴백(M1)+area-NULL upsert skip(M3)+matches JSON파싱(M4)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EqCYBhvTcdeCTUDX3RhWx9

realestate-lab/app/db.py

@@ -10,6 +10,9 @@ logger = logging.getLogger("realestate-lab")
 
 DB_PATH = os.getenv("REALESTATE_DB_PATH", "/app/data/realestate.db")
 
+# listing_matcher.MIN_SAMPLE와 동일 값 유지 — 시세 표본이 이 미만이면 광역 폴백.
+_MARKET_SAMPLE_MIN = 3
+
 
 def _conn():
     c = sqlite3.connect(DB_PATH, timeout=120.0)
@@ -981,6 +984,10 @@ def upsert_listing(data: Dict[str, Any]) -> tuple:
 
 
 def upsert_market_deal(data: Dict[str, Any]) -> None:
+    if data.get("area") is None:
+        # area 없는 실거래는 area BETWEEN 조회에서 절대 매칭 안 돼 median 계산에 무용하고,
+        # dedup 인덱스도 못 걸려(area가 NOT NULL 컬럼이 아니라 조건 매칭 불가) 무한 중복을 유발한다.
+        return
     cols = ("house_type", "dong_code", "dong", "complex_name", "area", "deal_type",
             "deposit", "monthly_rent", "sale_amount", "deal_ym", "floor", "source")
     d = {c: data.get(c) for c in cols}
@@ -992,16 +999,24 @@ def upsert_market_deal(data: Dict[str, Any]) -> None:
 
 def get_market_deals_for(dong_code, complex_name, area, deal_type, months=6) -> List[Dict[str, Any]]:
     with _conn() as conn:
-        rows = conn.execute(
-            """SELECT * FROM market_deals
-               WHERE dong_code=? AND deal_type=?
-                 AND (complex_name=? OR ? IS NULL OR complex_name IS NULL)
-                 AND area BETWEEN ? AND ?
-                 AND deal_ym >= strftime('%Y%m', 'now', ?)""",
-            (dong_code, deal_type, complex_name, complex_name,
-             (area or 0) - 5, (area or 0) + 5, f"-{int(months)} months"),
-        ).fetchall()
-    return [dict(r) for r in rows]
+        def _query(cn):
+            rows = conn.execute(
+                """SELECT * FROM market_deals
+                   WHERE dong_code=? AND deal_type=?
+                     AND (complex_name=? OR ? IS NULL OR complex_name IS NULL)
+                     AND area BETWEEN ? AND ?
+                     AND deal_ym >= strftime('%Y%m', 'now', ?)""",
+                (dong_code, deal_type, cn, cn,
+                 (area or 0) - 5, (area or 0) + 5, f"-{int(months)} months"),
+            ).fetchall()
+            return [dict(r) for r in rows]
+
+        deals = _query(complex_name)
+        if len(deals) < _MARKET_SAMPLE_MIN and complex_name:
+            # 단지 표본이 부족하면 complex 제약을 풀고 같은 동/거래유형/면적±5/최근성으로
+            # 광역 재조회한다. 표본을 늘리기만 하므로 안전(잘못된 판정 유발 X).
+            deals = _query(None)
+    return deals
 
 
 def upsert_listing_match(data: Dict[str, Any]) -> None:
@@ -1043,7 +1058,13 @@ def get_listing_matches() -> List[Dict[str, Any]]:
             "SELECT m.*, l.complex_name, l.dong, l.deal_type, l.deposit, l.sale_price, l.area_exclusive, l.url "
             "FROM listing_matches m JOIN listings l ON l.id=m.listing_id ORDER BY m.created_at DESC"
         ).fetchall()
-    return [dict(r) for r in rows]
+    out = []
+    for r in rows:
+        d = dict(r)
+        d["regulation_flags"] = json.loads(d.get("regulation_flags") or "[]")
+        d["reasons"] = json.loads(d.get("reasons") or "[]")
+        out.append(d)
+    return out
 
 
 _TIER_ORDER = {"위험": 0, "고가": 0, "주의": 1, "시세": 1, "안전": 2, "저평가": 2, "보류": -1}