feat(tradein): refine same-building anchor for within-building heterogeneity #681
3 changed files with 182 additions and 14 deletions
|
|
@ -78,6 +78,13 @@ class Settings(BaseSettings):
|
||||||
estimate_sb_min_comps: int = 2 # стоп на первом тире с ≥ N активных комплов
|
estimate_sb_min_comps: int = 2 # стоп на первом тире с ≥ N активных комплов
|
||||||
estimate_sb_area_sigma: float = 0.18 # σ log-нормального area-веса (Gaussian)
|
estimate_sb_area_sigma: float = 0.18 # σ log-нормального area-веса (Gaussian)
|
||||||
estimate_sb_rooms_match_boost: float = 1.6 # ×вес если rooms компла == target
|
estimate_sb_rooms_match_boost: float = 1.6 # ×вес если rooms компла == target
|
||||||
|
# #680-WB within-building heterogeneity refine: floor-similarity Gaussian по
|
||||||
|
# ОТНОСИТЕЛЬНОЙ вертикальной позиции (floor/total_floors). Прижимает якорь к
|
||||||
|
# комплам с похожим этажом — мелкокомнатный/нижний юнит во флагман-доме больше
|
||||||
|
# не наследует цену видового топ-этажа. 0.0 → выключено (точно старое поведение).
|
||||||
|
# Откалибровано на 55 golden (offline): σ_f=0.25 даёт лучший medAPE без потери
|
||||||
|
# покрытия; Хохрякова 3к/153 overshoot 64%→1.5%, флагман 4к 17.5%→5.4%.
|
||||||
|
estimate_sb_floor_sigma: float = 0.25
|
||||||
estimate_sb_guardrail_tol: float = 0.05 # hard floor: est ≥ min(comp ppm²)×(1−tol)
|
estimate_sb_guardrail_tol: float = 0.05 # hard floor: est ≥ min(comp ppm²)×(1−tol)
|
||||||
asking_to_sold_haircut: float = 0.05 # дефолтная asking→sold скидка (banded по ppm²)
|
asking_to_sold_haircut: float = 0.05 # дефолтная asking→sold скидка (banded по ppm²)
|
||||||
estimate_fsd_k: float = 1.65 # множитель FSD → полуширина диапазона
|
estimate_fsd_k: float = 1.65 # множитель FSD → полуширина диапазона
|
||||||
|
|
|
||||||
|
|
@ -1009,7 +1009,8 @@ def _fetch_anchor_comps(
|
||||||
Excludes lots без price_per_m2. is_active=true. Best-effort: ([], None) на ошибке.
|
Excludes lots без price_per_m2. is_active=true. Best-effort: ([], None) на ошибке.
|
||||||
|
|
||||||
Returns (comps, tier) где tier ∈ {'A','C', None}. comps — list dict с
|
Returns (comps, tier) где tier ∈ {'A','C', None}. comps — list dict с
|
||||||
ключами price_per_m2 (int>0), area_m2 (float|None), rooms (int|None).
|
ключами price_per_m2 (int>0), area_m2 (float|None), rooms (int|None),
|
||||||
|
floor (int|None), total_floors (int|None) — последние два для floor-веса (#680-WB).
|
||||||
"""
|
"""
|
||||||
min_comps = settings.estimate_sb_min_comps
|
min_comps = settings.estimate_sb_min_comps
|
||||||
|
|
||||||
|
|
@ -1028,7 +1029,7 @@ def _fetch_anchor_comps(
|
||||||
db.execute(
|
db.execute(
|
||||||
text(
|
text(
|
||||||
"""
|
"""
|
||||||
SELECT price_per_m2, area_m2, rooms
|
SELECT price_per_m2, area_m2, rooms, floor, total_floors
|
||||||
FROM listings
|
FROM listings
|
||||||
WHERE is_active = true
|
WHERE is_active = true
|
||||||
AND price_per_m2 > 0
|
AND price_per_m2 > 0
|
||||||
|
|
@ -1056,6 +1057,8 @@ def _fetch_anchor_comps(
|
||||||
"price_per_m2": int(r["price_per_m2"]),
|
"price_per_m2": int(r["price_per_m2"]),
|
||||||
"area_m2": float(r["area_m2"]) if r["area_m2"] is not None else None,
|
"area_m2": float(r["area_m2"]) if r["area_m2"] is not None else None,
|
||||||
"rooms": int(r["rooms"]) if r["rooms"] is not None else None,
|
"rooms": int(r["rooms"]) if r["rooms"] is not None else None,
|
||||||
|
"floor": int(r["floor"]) if r["floor"] is not None else None,
|
||||||
|
"total_floors": int(r["total_floors"]) if r["total_floors"] is not None else None,
|
||||||
}
|
}
|
||||||
for r in rows
|
for r in rows
|
||||||
if r["price_per_m2"]
|
if r["price_per_m2"]
|
||||||
|
|
@ -1077,7 +1080,7 @@ def _fetch_anchor_comps(
|
||||||
db.execute(
|
db.execute(
|
||||||
text(
|
text(
|
||||||
"""
|
"""
|
||||||
SELECT price_per_m2, area_m2, rooms
|
SELECT price_per_m2, area_m2, rooms, floor, total_floors
|
||||||
FROM listings
|
FROM listings
|
||||||
WHERE is_active = true
|
WHERE is_active = true
|
||||||
AND price_per_m2 > 0
|
AND price_per_m2 > 0
|
||||||
|
|
@ -1120,6 +1123,8 @@ def _fetch_anchor_comps(
|
||||||
"price_per_m2": int(r["price_per_m2"]),
|
"price_per_m2": int(r["price_per_m2"]),
|
||||||
"area_m2": float(r["area_m2"]) if r["area_m2"] is not None else None,
|
"area_m2": float(r["area_m2"]) if r["area_m2"] is not None else None,
|
||||||
"rooms": int(r["rooms"]) if r["rooms"] is not None else None,
|
"rooms": int(r["rooms"]) if r["rooms"] is not None else None,
|
||||||
|
"floor": int(r["floor"]) if r["floor"] is not None else None,
|
||||||
|
"total_floors": int(r["total_floors"]) if r["total_floors"] is not None else None,
|
||||||
}
|
}
|
||||||
for r in rows
|
for r in rows
|
||||||
if r["price_per_m2"]
|
if r["price_per_m2"]
|
||||||
|
|
@ -1153,15 +1158,24 @@ def _compute_same_building_anchor(
|
||||||
tier: str,
|
tier: str,
|
||||||
sigma: float,
|
sigma: float,
|
||||||
rooms_boost: float,
|
rooms_boost: float,
|
||||||
|
floor_target: int | None = None,
|
||||||
|
total_floors_target: int | None = None,
|
||||||
|
floor_sigma: float = 0.0,
|
||||||
) -> dict[str, Any] | None:
|
) -> dict[str, Any] | None:
|
||||||
"""Чистая (testable без БД) свёртка комплов в anchor-оценку.
|
"""Чистая (testable без БД) свёртка комплов в anchor-оценку.
|
||||||
|
|
||||||
1. similarity-weighted mean ppm²: w_i = exp(−(ln(area_i/area_target))²/(2σ²))
|
1. similarity-weighted mean ppm²: w_i = exp(−(ln(area_i/area_target))²/(2σ²))
|
||||||
× (rooms_boost если rooms_i==rooms_target). area_i=None → area-вес 1.0
|
× (rooms_boost если rooms_i==rooms_target) × FLOOR-вес. area_i=None →
|
||||||
(нейтрально: Yandex-листинги без площади судим только по комнатам).
|
area-вес 1.0 (нейтрально). FLOOR-вес (#680-WB) — Gaussian по ОТНОСИТЕЛЬНОЙ
|
||||||
2. PREMIUM uplift (class-free): если target — топ-юнит ДОМА (area_target ≥ p66
|
вертикальной позиции pos=floor/total_floors: exp(−(pos_i−pos_t)²/(2σ_f²)),
|
||||||
площадей комплов) И tier == 'A' → берём weighted ~p70 верхний квантиль comp
|
σ_f=floor_sigma. Прижимает якорь к комплам с похожим этажом (1-й/последний
|
||||||
ppm² вместо mean (консервативно, только вверх).
|
и видовые этажи реально отличаются по цене). floor_sigma=0 ИЛИ нет floor у
|
||||||
|
target/компла → вес 1.0 (выключено / нейтрально — no regress).
|
||||||
|
2. PREMIUM uplift (class-free): target — топ-юнит ДОМА (area ≥ p66 площадей
|
||||||
|
комплов) И rooms_target ≥ медианы комнат комплов И tier == 'A' → берём
|
||||||
|
weighted ~p70 верхний квантиль ppm² (консервативно, только вверх). Условие
|
||||||
|
по комнатам (#680-WB) не даёт мелкокомнатному юниту во флагман-доме унаследовать
|
||||||
|
headline-премию флагмана (3к/153 в доме с 4к-флагманом ≠ цена флагмана).
|
||||||
3. haircut asking→sold (banded по anchor ppm²): anchor_sold = anchor×(1−haircut).
|
3. haircut asking→sold (banded по anchor ppm²): anchor_sold = anchor×(1−haircut).
|
||||||
4. FSD = 0.07 + 0.25·CV(comp ppm²) + tier_penalty + n_penalty; range полуширина
|
4. FSD = 0.07 + 0.25·CV(comp ppm²) + tier_penalty + n_penalty; range полуширина
|
||||||
= k·fsd. confidence-банд по fsd.
|
= k·fsd. confidence-банд по fsd.
|
||||||
|
|
@ -1176,7 +1190,12 @@ def _compute_same_building_anchor(
|
||||||
return None
|
return None
|
||||||
n = len(ppm2)
|
n = len(ppm2)
|
||||||
|
|
||||||
# 1. similarity weights
|
# target relative vertical position (None → floor-вес отключён/нейтрален).
|
||||||
|
target_pos: float | None = None
|
||||||
|
if floor_sigma > 0 and floor_target and total_floors_target and total_floors_target > 0:
|
||||||
|
target_pos = floor_target / total_floors_target
|
||||||
|
|
||||||
|
# 1. similarity weights (area × rooms × floor-position)
|
||||||
weights: list[float] = []
|
weights: list[float] = []
|
||||||
for c in comps:
|
for c in comps:
|
||||||
a = c.get("area_m2")
|
a = c.get("area_m2")
|
||||||
|
|
@ -1186,19 +1205,35 @@ def _compute_same_building_anchor(
|
||||||
area_w = 1.0 # площадь неизвестна → нейтральный area-вес
|
area_w = 1.0 # площадь неизвестна → нейтральный area-вес
|
||||||
rooms_match = rooms_target is not None and c.get("rooms") == rooms_target
|
rooms_match = rooms_target is not None and c.get("rooms") == rooms_target
|
||||||
rooms_w = rooms_boost if rooms_match else 1.0
|
rooms_w = rooms_boost if rooms_match else 1.0
|
||||||
weights.append(area_w * rooms_w)
|
floor_w = 1.0
|
||||||
|
if target_pos is not None:
|
||||||
|
cf = c.get("floor")
|
||||||
|
ctf = c.get("total_floors")
|
||||||
|
if cf and ctf and ctf > 0:
|
||||||
|
comp_pos = cf / ctf
|
||||||
|
floor_w = math.exp(
|
||||||
|
-((comp_pos - target_pos) ** 2) / (2.0 * floor_sigma * floor_sigma)
|
||||||
|
)
|
||||||
|
# компл без floor/total_floors → нейтральный floor-вес 1.0
|
||||||
|
weights.append(area_w * rooms_w * floor_w)
|
||||||
wsum = sum(weights)
|
wsum = sum(weights)
|
||||||
if wsum > 0:
|
if wsum > 0:
|
||||||
anchor = sum(w * p for w, p in zip(weights, ppm2, strict=True)) / wsum
|
anchor = sum(w * p for w, p in zip(weights, ppm2, strict=True)) / wsum
|
||||||
else:
|
else:
|
||||||
anchor = _percentile(sorted(ppm2), 0.5)
|
anchor = _percentile(sorted(ppm2), 0.5)
|
||||||
|
|
||||||
# 2. premium uplift — топ-юнит дома по площади (≥p66) И Tier A → weighted p70.
|
# 2. premium uplift — топ-юнит дома (площадь ≥ p66 И комнаты ≥ медианы) И Tier A
|
||||||
|
# → weighted p70. Условие по комнатам отсекает мелкие юниты во флагман-домах.
|
||||||
used_uplift = False
|
used_uplift = False
|
||||||
areas = [c.get("area_m2") for c in comps if c.get("area_m2")]
|
areas = [c.get("area_m2") for c in comps if c.get("area_m2")]
|
||||||
|
comp_rooms = [c.get("rooms") for c in comps if c.get("rooms") is not None]
|
||||||
if tier == "A" and areas and area_target > 0:
|
if tier == "A" and areas and area_target > 0:
|
||||||
p66_area = _percentile(sorted(areas), 0.66)
|
p66_area = _percentile(sorted(areas), 0.66)
|
||||||
if area_target >= p66_area:
|
rooms_ok = True
|
||||||
|
if rooms_target is not None and comp_rooms:
|
||||||
|
median_rooms = _percentile(sorted(comp_rooms), 0.5)
|
||||||
|
rooms_ok = rooms_target >= median_rooms
|
||||||
|
if area_target >= p66_area and rooms_ok:
|
||||||
p70 = _percentile(sorted(ppm2), 0.70)
|
p70 = _percentile(sorted(ppm2), 0.70)
|
||||||
if p70 > anchor:
|
if p70 > anchor:
|
||||||
anchor = p70
|
anchor = p70
|
||||||
|
|
@ -1643,6 +1678,9 @@ async def estimate_quality(
|
||||||
tier=anchor_tier or "",
|
tier=anchor_tier or "",
|
||||||
sigma=settings.estimate_sb_area_sigma,
|
sigma=settings.estimate_sb_area_sigma,
|
||||||
rooms_boost=settings.estimate_sb_rooms_match_boost,
|
rooms_boost=settings.estimate_sb_rooms_match_boost,
|
||||||
|
floor_target=payload.floor,
|
||||||
|
total_floors_target=payload.total_floors,
|
||||||
|
floor_sigma=settings.estimate_sb_floor_sigma,
|
||||||
)
|
)
|
||||||
if anchor is not None:
|
if anchor is not None:
|
||||||
est_ppm2 = anchor["anchor_sold_ppm2"]
|
est_ppm2 = anchor["anchor_sold_ppm2"]
|
||||||
|
|
|
||||||
|
|
@ -168,8 +168,20 @@ def test_normalize_no_house_number_returns_none_base() -> None:
|
||||||
# ── Pure anchor compute ─────────────────────────────────────────────────────
|
# ── Pure anchor compute ─────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
def _comp(ppm2: int, area: float | None = None, rooms: int | None = None) -> dict[str, Any]:
|
def _comp(
|
||||||
return {"price_per_m2": ppm2, "area_m2": area, "rooms": rooms}
|
ppm2: int,
|
||||||
|
area: float | None = None,
|
||||||
|
rooms: int | None = None,
|
||||||
|
floor: int | None = None,
|
||||||
|
total_floors: int | None = None,
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
return {
|
||||||
|
"price_per_m2": ppm2,
|
||||||
|
"area_m2": area,
|
||||||
|
"rooms": rooms,
|
||||||
|
"floor": floor,
|
||||||
|
"total_floors": total_floors,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
def test_anchor_premium_lift_hohryakova48() -> None:
|
def test_anchor_premium_lift_hohryakova48() -> None:
|
||||||
|
|
@ -493,3 +505,114 @@ def test_anchor_exposes_comp_max() -> None:
|
||||||
assert res is not None
|
assert res is not None
|
||||||
assert res["comp_max_ppm2"] == 500_000
|
assert res["comp_max_ppm2"] == 500_000
|
||||||
assert res["comp_min_ppm2"] == 300_000
|
assert res["comp_min_ppm2"] == 300_000
|
||||||
|
|
||||||
|
|
||||||
|
# ── #680-WB: within-building heterogeneity refine (uplift-gate + floor weight) ──
|
||||||
|
|
||||||
|
|
||||||
|
def test_anchor_uplift_gated_by_rooms_median_hohryakova_3k() -> None:
|
||||||
|
"""Зеркало Хохрякова 48 3к/153: дом с 4-комн. флагманом (комплы rooms=4 кроме
|
||||||
|
одного 3к). target 3-комн. 153м² (топ по площади, ≥p66) — БЕЗ refine получил бы
|
||||||
|
p70-uplift и унаследовал цену флагмана. С rooms-gate (3 < медианы=4) uplift НЕ
|
||||||
|
срабатывает → anchor падает к similar-комплам, не к headline-флагману."""
|
||||||
|
comps = [
|
||||||
|
_comp(399_477, area=153.2, rooms=4, floor=8, total_floors=8),
|
||||||
|
_comp(683_995, area=146.2, rooms=4, floor=5, total_floors=14),
|
||||||
|
_comp(502_846, area=105.4, rooms=3, floor=6, total_floors=9),
|
||||||
|
_comp(683_994, area=146.2, rooms=4, floor=5, total_floors=15),
|
||||||
|
]
|
||||||
|
res = _compute_same_building_anchor(
|
||||||
|
comps,
|
||||||
|
area_target=153.2,
|
||||||
|
rooms_target=3,
|
||||||
|
tier="A",
|
||||||
|
sigma=0.18,
|
||||||
|
rooms_boost=1.6,
|
||||||
|
floor_target=8,
|
||||||
|
total_floors_target=8,
|
||||||
|
floor_sigma=0.25,
|
||||||
|
)
|
||||||
|
assert res is not None
|
||||||
|
# rooms_target 3 < median comp rooms 4 → uplift gated off.
|
||||||
|
assert res["used_uplift"] is False
|
||||||
|
# anchor НЕ должен быть на уровне флагмана (684k); прижат к похожим юнитам.
|
||||||
|
assert res["anchor_ppm2"] < 550_000
|
||||||
|
|
||||||
|
|
||||||
|
def test_anchor_uplift_still_fires_for_genuine_top_unit() -> None:
|
||||||
|
"""Топ-юнит дома (rooms ≥ медианы И площадь ≥p66) в Tier A → uplift ВСЁ ЕЩЁ
|
||||||
|
срабатывает, когда дорогой компл area-несимметричен и weighted mean занижает
|
||||||
|
(do-no-harm для настоящих флагманов). Зеркало test_anchor_premium_uplift_when_*
|
||||||
|
с rooms_target == медиане комнат комплов (gate проходит)."""
|
||||||
|
comps = [
|
||||||
|
_comp(900_000, area=60.0, rooms=3), # дорогой, малая площадь → area-вес низкий
|
||||||
|
_comp(300_000, area=200.0, rooms=3),
|
||||||
|
_comp(320_000, area=210.0, rooms=3),
|
||||||
|
]
|
||||||
|
res = _compute_same_building_anchor(
|
||||||
|
comps,
|
||||||
|
area_target=205.0, # топ по площади (≥p66)
|
||||||
|
rooms_target=3, # == median comp rooms 3 → rooms-gate проходит
|
||||||
|
tier="A",
|
||||||
|
sigma=0.18,
|
||||||
|
rooms_boost=1.6,
|
||||||
|
)
|
||||||
|
assert res is not None
|
||||||
|
# mean занижен дил. дорогим малометражным → uplift подтягивает к p70.
|
||||||
|
assert res["used_uplift"] is True
|
||||||
|
assert res["anchor_ppm2"] >= 500_000
|
||||||
|
|
||||||
|
|
||||||
|
def test_anchor_floor_weight_tilts_toward_similar_floor() -> None:
|
||||||
|
"""floor_sigma>0 → комплы с похожей вертикальной позицией (floor/total_floors)
|
||||||
|
весят больше. Дорогой видовой топ-этаж и дешёвый средний этаж той же площади;
|
||||||
|
target на среднем этаже → anchor тянется к среднеэтажному компла (не к видовому).
|
||||||
|
target — НЕ топ-юнит по площади (есть компл крупнее) → uplift не маскирует floor."""
|
||||||
|
comps = [
|
||||||
|
_comp(300_000, area=60.0, rooms=2, floor=5, total_floors=20), # mid, как target
|
||||||
|
_comp(600_000, area=60.0, rooms=2, floor=20, total_floors=20), # видовой топ
|
||||||
|
_comp(310_000, area=120.0, rooms=3, floor=4, total_floors=20), # крупнее target
|
||||||
|
]
|
||||||
|
target_kw = dict(
|
||||||
|
area_target=60.0,
|
||||||
|
rooms_target=2,
|
||||||
|
tier="A",
|
||||||
|
sigma=0.18,
|
||||||
|
rooms_boost=1.6,
|
||||||
|
floor_target=5,
|
||||||
|
total_floors_target=20,
|
||||||
|
)
|
||||||
|
off = _compute_same_building_anchor(comps, floor_sigma=0.0, **target_kw)
|
||||||
|
on = _compute_same_building_anchor(comps, floor_sigma=0.25, **target_kw)
|
||||||
|
assert off is not None and on is not None
|
||||||
|
assert off["used_uplift"] is False # target не топ по площади
|
||||||
|
# floor-вес прижимает к среднеэтажному (300k), уводя от видового топа (600k).
|
||||||
|
assert on["anchor_ppm2"] < off["anchor_ppm2"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_anchor_floor_weight_neutral_when_data_missing() -> None:
|
||||||
|
"""Нет floor у компла/target → floor-вес 1.0 (нейтрально), не падает,
|
||||||
|
результат идентичен floor_sigma=0 (no regress на Yandex-листингах без этажа)."""
|
||||||
|
comps = [_comp(300_000, area=60.0, rooms=2), _comp(340_000, area=62.0, rooms=2)]
|
||||||
|
a = _compute_same_building_anchor(
|
||||||
|
comps,
|
||||||
|
area_target=60.0,
|
||||||
|
rooms_target=2,
|
||||||
|
tier="A",
|
||||||
|
sigma=0.18,
|
||||||
|
rooms_boost=1.6,
|
||||||
|
floor_target=None,
|
||||||
|
total_floors_target=None,
|
||||||
|
floor_sigma=0.25,
|
||||||
|
)
|
||||||
|
b = _compute_same_building_anchor(
|
||||||
|
comps,
|
||||||
|
area_target=60.0,
|
||||||
|
rooms_target=2,
|
||||||
|
tier="A",
|
||||||
|
sigma=0.18,
|
||||||
|
rooms_boost=1.6,
|
||||||
|
floor_sigma=0.0,
|
||||||
|
)
|
||||||
|
assert a is not None and b is not None
|
||||||
|
assert abs(a["anchor_ppm2"] - b["anchor_ppm2"]) < 1.0
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue