gahusb
d752675e9d
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EqCYBhvTcdeCTUDX3RhWx9
115 lines
5.0 KiB
Python
115 lines
5.0 KiB
Python
"""매물 조건 매칭 + 임차 안전마진(전세가율)·매매 적정성(호가율) 판정."""
|
|
import logging
|
|
import statistics
|
|
|
|
from .db import (get_listing_criteria, get_market_deals_for, upsert_listing_match,
|
|
get_listings)
|
|
from . import finance_rules
|
|
|
|
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 대칭)."""
|
|
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)
|
|
for l in get_listings(limit=1000):
|
|
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)
|
|
deals = get_market_deals_for(l.get("dong_code"), l.get("complex_name"),
|
|
l.get("area_exclusive"), "전세" if is_rent else "매매")
|
|
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})
|
|
upsert_listing_match(rec)
|
|
logger.info("매물 매칭 완료")
|
|
|
|
|
|
def reg_flags(reg: dict) -> list:
|
|
f = []
|
|
if reg.get("is_toheo"): f.append("토허")
|
|
if reg.get("is_regulated"): f.append("규제지역")
|
|
return f
|