web-page-backend/realestate-lab/app/listing_matcher.py

"""매물 조건 매칭 + 임차 안전마진(전세가율)·매매 적정성(호가율) 판정."""
import logging
import statistics

from .db import (get_listing_criteria, get_market_deals_for, bulk_upsert_listing_matches,
                 _conn)
from . import finance_rules
from .lawd_codes import lawd_code

logger = logging.getLogger("realestate-lab")

# 경계 상수(조정 가능)
JEONSE_SAFE, JEONSE_WARN = 0.70, 0.80
VALUATION_LOW, VALUATION_HIGH = 0.97, 1.05
MIN_SAMPLE = 3
DISCLAIMER = "등기부 선순위 근저당은 자동 확인 불가 — 계약 전 인터넷등기소 열람 필수."


def compute_safety(deposit, deal_type, deals) -> dict:
    vals = [d["deposit"] for d in deals if d.get("deposit")]
    if len(vals) < MIN_SAMPLE:
        return {"jeonse_ratio": None, "tier": "보류", "median": None, "sample": len(vals)}
    median = statistics.median(vals)
    ratio = (deposit or 0) / median if median else None
    if ratio is None:
        tier = "보류"
    elif ratio <= JEONSE_SAFE:
        tier = "안전"
    elif ratio <= JEONSE_WARN:
        tier = "주의"
    else:
        tier = "위험"
    return {"jeonse_ratio": round(ratio, 3) if ratio else None,
            "tier": tier, "median": int(median), "sample": len(vals)}


def compute_valuation(sale_price, deals) -> dict:
    vals = [d["sale_amount"] for d in deals if d.get("sale_amount")]
    if len(vals) < MIN_SAMPLE:
        return {"price_ratio": None, "tier": "보류", "median": None, "sample": len(vals)}
    median = statistics.median(vals)
    ratio = (sale_price or 0) / median if median else None
    if ratio is None:
        tier = "보류"
    elif ratio <= VALUATION_LOW:
        tier = "저평가"
    elif ratio <= VALUATION_HIGH:
        tier = "시세"
    else:
        tier = "고가"
    return {"price_ratio": round(ratio, 3) if ratio else None,
            "tier": tier, "median": int(median), "sample": len(vals)}


def match_listing(criteria, listing, budget) -> dict:
    reasons, passed = [], True
    if criteria.get("dongs") and listing.get("dong") not in criteria["dongs"]:
        passed = False; reasons.append("동 불일치")
    if criteria.get("deal_types") and listing.get("deal_type") not in criteria["deal_types"]:
        passed = False; reasons.append("거래유형 불일치")
    if criteria.get("min_area") and (listing.get("area_exclusive") or 0) < criteria["min_area"]:
        passed = False; reasons.append("면적 미달")
    if listing.get("deal_type") == "매매":
        cap = criteria.get("max_sale_price") or (budget or {}).get("max_price")
        if cap and (listing.get("sale_price") or 0) > cap:
            passed = False; reasons.append("매수가능 상한 초과")
    else:  # 임차
        cap = criteria.get("max_deposit") or (budget or {}).get("max_deposit")
        if cap and (listing.get("deposit") or 0) > cap:
            passed = False; reasons.append("보증금 상한 초과")
    score = 100 if passed else 0
    if passed:
        reasons.append("조건 통과")
    return {"passed": passed, "score": score, "reasons": reasons}


def run_listing_matching() -> None:
    """미평가 포함 전체 매물 재평가 → listing_matches upsert (청약 run_matching 대칭).
    listings 조회 + 매물별 market_deals 조회 + 최종 upsert를 단일 connection으로 처리
    (Task 2 성능개선: 이전엔 매물마다 get_market_deals_for/upsert_listing_match가 각각
    자체 _conn()을 열어 269건×~3conn이 발생했다)."""
    criteria = get_listing_criteria()
    equity = criteria.get("equity") or 0
    income = criteria.get("annual_income") or 0
    jeonse_budget = finance_rules.estimate_jeonse_budget(equity, income)
    with _conn() as conn:
        listings = [dict(r) for r in conn.execute(
            "SELECT * FROM listings ORDER BY created_at DESC LIMIT 1000"
        ).fetchall()]
        recs = []
        for l in listings:
            is_rent = l["deal_type"] != "매매"
            reg = finance_rules.region_regulation(l.get("dong"))
            if is_rent:
                budget = jeonse_budget
            else:
                budget = finance_rules.estimate_purchase_budget(
                    equity, income, reg, criteria.get("is_first_home"), criteria.get("is_homeless"))
            m = match_listing(criteria, l, budget)
            # 네이버 cortarNo 부재 등으로 기존 저장된 listing.dong_code가 None일 수 있음 —
            # dong명에서 5자리 lawd_code를 유도해 즉시 판정(보류 고착 방지).
            dc = l.get("dong_code") or lawd_code(l.get("dong"))
            deals = get_market_deals_for(dc, l.get("complex_name"),
                                         l.get("area_exclusive"), "전세" if is_rent else "매매",
                                         conn=conn)
            rec = {"listing_id": l["id"], "category": "임차" if is_rent else "매매",
                   "passed": 1 if m["passed"] else 0, "match_score": m["score"],
                   "sample_size": 0, "reasons": m["reasons"], "is_new": 1,
                   "regulation_flags": reg_flags(reg)}
            if is_rent:
                s = compute_safety(l.get("deposit"), l["deal_type"], deals)
                rec.update({"jeonse_ratio": s["jeonse_ratio"], "safety_tier": s["tier"],
                            "market_median": s["median"], "sample_size": s["sample"]})
            else:
                v = compute_valuation(l.get("sale_price"), deals)
                rec.update({"price_ratio": v["price_ratio"], "valuation_tier": v["tier"],
                            "market_median": v["median"], "sample_size": v["sample"],
                            "budget_ok": 1 if (l.get("sale_price") or 0) <= (budget.get("max_price") or 0) else 0})
            recs.append(rec)
        bulk_upsert_listing_matches(recs, conn=conn)
    logger.info("매물 매칭 완료(%d건)", len(recs))


def reg_flags(reg: dict) -> list:
    f = []
    if reg.get("is_toheo"): f.append("토허")
    if reg.get("is_regulated"): f.append("규제지역")
    return f