fix(estimator): sanity-clamp quarter-index factor in _apply_quarter_index (#859)

Follow-up hardening from PR #858 review. The quarter-index factor
(target_index / avg_analog_index) was unbounded — pathological FDW data could
yield an extreme multiplier on a client-visible price. Clamp factor to
[estimate_quarter_index_factor_min, _max] = [0.6, 1.8] (outside the realistic
quarter-to-quarter location spread for EKB вторичка). Bounds in settings;
passed as kwargs so _apply_quarter_index stays pure/testable. Clamp fires
logged at INFO. Formula and #764 guards unchanged.

Refs #859, #764
This commit is contained in:
bot-backend 2026-05-31 10:11:16 +03:00
parent b16a45ef87
commit 1cbd4bd4b2
3 changed files with 150 additions and 5 deletions

View file

@ -107,6 +107,11 @@ class Settings(BaseSettings):
# дают индекс > 2.0 при малой выборке → no-op чтобы избежать регрессию.
estimate_quarter_index_max_for_small_n: float = 2.0
estimate_quarter_index_small_n_threshold: int = 50
# Sanity-clamp на factor = target_index / avg_analog_index (#859).
# Belt-and-suspenders против патологичных FDW-данных. Нормальные квартальные
# индексы РФ лежат в [0.6, 1.8]; за этими порогами — артефакт, а не сигнал.
estimate_quarter_index_factor_min: float = 0.6
estimate_quarter_index_factor_max: float = 1.8
# ── Estimate enrichment time-budgets (#654) ──────────────────────────────
# POST /estimate делает несколько ПОСЛЕДОВАТЕЛЬНЫХ блокирующих сетевых

View file

@ -893,6 +893,8 @@ def _apply_quarter_index(
base_range_high: int,
target_index: float,
avg_analog_index: float,
min_factor: float = 0.6,
max_factor: float = 1.8,
) -> tuple[float, int, int, int, float]:
"""Чистая (testable без БД) gap-correction квартального индекса (#764).
@ -904,10 +906,27 @@ def _apply_quarter_index(
Все ценовые выходы масштабируются одним и тем же factor median/range
остаются геометрически консистентными.
min_factor / max_factor sanity-clamp (#859): belt-and-suspenders против
патологичных FDW-данных. Калибруются через settings и передаются из
вызывающего кода, чтобы хелпер оставался чистым (без импорта settings).
Когда clamp меняет raw factor логируется (см. caller).
Returns (adjusted_ppm2, adjusted_median_price, adjusted_range_low,
adjusted_range_high, factor).
"""
factor = target_index / avg_analog_index
raw_factor = target_index / avg_analog_index
factor = max(min_factor, min(max_factor, raw_factor))
if raw_factor != factor:
logger.info(
"quarter_index: factor clamped raw=%.4f%.4f (bounds [%.2f, %.2f])"
" target_index=%.3f avg_analog_index=%.3f",
raw_factor,
factor,
min_factor,
max_factor,
target_index,
avg_analog_index,
)
adjusted_ppm2 = base_median_ppm2 * factor
adjusted_median_price = round(base_median_price * factor)
adjusted_range_low = round(base_range_low * factor)
@ -2194,6 +2213,8 @@ async def estimate_quality(
base_range_high=range_high,
target_index=target_qi,
avg_analog_index=avg_analog_index,
min_factor=settings.estimate_quarter_index_factor_min,
max_factor=settings.estimate_quarter_index_factor_max,
)
analogs_with_qi = sum(
1 for lq, _lp in analog_quarters if lq in analog_index_map

View file

@ -581,7 +581,11 @@ def test_bimodal_guard_skips_high_index_small_n() -> None:
def test_bimodal_guard_allows_high_index_large_n() -> None:
"""Bimodal guard НЕ срабатывает при price_index>2.0 если n_deals>=50."""
"""Bimodal guard НЕ срабатывает при price_index>2.0 если n_deals>=50.
Коррекция применяется, но raw factor=2.5 зажат #859-clamp до max_factor=1.8.
Медиана меняется (guard не блокирует), но масштабируется на 1.8, не 2.5.
"""
analogs_no_cadnum = [
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
]
@ -589,12 +593,13 @@ def test_bimodal_guard_allows_high_index_large_n() -> None:
est = _run_estimate_qi(
analogs=analogs_no_cadnum,
dadata_cadnum=f"{_TARGET_QUARTER}:350",
qi_lookup_result=(2.5, 60), # index>2.0 но n=60>=50 → применяется
qi_lookup_result=(2.5, 60), # index>2.0 но n=60>=50 → bimodal guard не срабатывает
flag_enabled=True,
)
# factor = 2.5 → медиана должна измениться
# Коррекция применена: медиана != base_median (bimodal guard не заблокировал).
# factor=2.5 > max_factor=1.8 → зажат до 1.8 (#859).
assert est.median_price_rub != base_median
assert est.median_price_rub == round(base_median * 2.5)
assert est.median_price_rub == round(base_median * 1.8)
# ─────────────────────────────────────────────────────────────────────────────
@ -864,5 +869,119 @@ def test_guard1b_imv_anchor_below_blend_threshold_prevents_correction() -> None:
assert "квартал" not in (est.confidence_explanation or "").lower()
# ─────────────────────────────────────────────────────────────────────────────
# #859 — sanity-clamp на factor = target_index / avg_analog_index
# ─────────────────────────────────────────────────────────────────────────────
def test_apply_quarter_index_clamp_extreme_high_raw_factor() -> None:
"""Экстремально низкий avg_analog_index → raw factor >> max → зажат до max_factor.
target_index=1.0, avg_analog_index=0.3 raw=3.33, clamped1.8.
Выходы масштабируются на 1.8, не на 3.33.
"""
max_f = 1.8
ppm2, median, low, high, factor = _apply_quarter_index(
base_median_ppm2=100_000.0,
base_median_price=5_000_000,
base_range_low=4_000_000,
base_range_high=6_000_000,
target_index=1.0,
avg_analog_index=0.3,
min_factor=0.6,
max_factor=max_f,
)
assert abs(factor - max_f) < 1e-9, f"Expected factor={max_f}, got {factor}"
assert abs(ppm2 - 100_000.0 * max_f) < 1.0
assert median == round(5_000_000 * max_f)
assert low == round(4_000_000 * max_f)
assert high == round(6_000_000 * max_f)
# Убеждаемся, что raw factor действительно был бы за пределами clamp
raw = 1.0 / 0.3
assert raw > max_f
def test_apply_quarter_index_clamp_extreme_low_raw_factor() -> None:
"""Экстремально высокий avg_analog_index → raw factor << min → зажат до min_factor.
target_index=0.5, avg_analog_index=2.0 raw=0.25, clamped0.6.
Выходы масштабируются на 0.6, не на 0.25.
"""
min_f = 0.6
ppm2, median, low, high, factor = _apply_quarter_index(
base_median_ppm2=100_000.0,
base_median_price=5_000_000,
base_range_low=4_000_000,
base_range_high=6_000_000,
target_index=0.5,
avg_analog_index=2.0,
min_factor=min_f,
max_factor=1.8,
)
assert abs(factor - min_f) < 1e-9, f"Expected factor={min_f}, got {factor}"
assert abs(ppm2 - 100_000.0 * min_f) < 1.0
assert median == round(5_000_000 * min_f)
assert low == round(4_000_000 * min_f)
assert high == round(6_000_000 * min_f)
# Убеждаемся, что raw factor действительно был бы за пределами clamp
raw = 0.5 / 2.0
assert raw < min_f
def test_apply_quarter_index_clamp_normal_factor_unchanged() -> None:
"""Нормальный factor внутри [0.6, 1.8] → clamp не меняет значение (регрессия).
target_index=1.3, avg_analog_index=1.1 raw=1.182, внутри [0.6,1.8] без изменений.
"""
raw_expected = 1.3 / 1.1
ppm2, median, _low, _high, factor = _apply_quarter_index(
base_median_ppm2=200_000.0,
base_median_price=10_000_000,
base_range_low=8_000_000,
base_range_high=12_000_000,
target_index=1.3,
avg_analog_index=1.1,
min_factor=0.6,
max_factor=1.8,
)
assert abs(factor - raw_expected) < 1e-9, f"Expected {raw_expected}, got {factor}"
assert abs(ppm2 - 200_000.0 * raw_expected) < 1.0
assert median == round(10_000_000 * raw_expected)
def test_apply_quarter_index_clamp_boundary_at_max_exact() -> None:
"""factor точно на верхней границе (=1.8) → clamp не применяется."""
# target=1.8, avg=1.0 → raw=1.8 ровно = max_factor
_ppm2, median, _low, _high, factor = _apply_quarter_index(
base_median_ppm2=100_000.0,
base_median_price=5_000_000,
base_range_low=4_000_000,
base_range_high=6_000_000,
target_index=1.8,
avg_analog_index=1.0,
min_factor=0.6,
max_factor=1.8,
)
assert abs(factor - 1.8) < 1e-9
assert median == round(5_000_000 * 1.8)
def test_apply_quarter_index_clamp_boundary_at_min_exact() -> None:
"""factor точно на нижней границе (=0.6) → clamp не применяется."""
# target=0.6, avg=1.0 → raw=0.6 ровно = min_factor
_ppm2, median, _low, _high, factor = _apply_quarter_index(
base_median_ppm2=100_000.0,
base_median_price=5_000_000,
base_range_low=4_000_000,
base_range_high=6_000_000,
target_index=0.6,
avg_analog_index=1.0,
min_factor=0.6,
max_factor=1.8,
)
assert abs(factor - 0.6) < 1e-9
assert median == round(5_000_000 * 0.6)
if __name__ == "__main__": # pragma: no cover
raise SystemExit(pytest.main([__file__, "-q"]))