gendesign/tradein-mvp/backend/tests/test_2936_unknown_attr_penalty.py

"""Неизвестный year_built / house_type перестаёт считаться идеальным совпадением (#2936).

SQL-формула relevance_score даёт кандидату без year_built штраф 0 — столько же,
сколько точному попаданию, и ЛУЧШЕ, чем известному году с разницей 24 (2.0).
Отсутствие данных выигрывает у знания, и это возвышает источник с худшей
полнотой: avito (год 44 %, тип 0 %) берёт 47 % слотов топ-20 при 21 % доли в пуле.

Флаг, а не правка формулы: влияние на цену измерено дважды — смещения нет, есть
шум (39 % целей сдвигаются >5 %). Какой отбор ТОЧНЕЕ — отвечает только бэктест на
сделках, и включать можно только по его итогу (тот же контракт, что у #2012).

Механизм: NULL получает МЕДИАННЫЙ по пулу штраф того же признака среди кандидатов,
у которых он известен, — не наказание и не награда.
"""

from __future__ import annotations

import os

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

from app.services.estimator import _apply_unknown_attr_penalty


def _pool() -> list[dict]:
    # Штрафы за год при target=2000: 0.0 / 1.0 / 2.0 / 3.0 / 4.0 → медиана 2.0.
    # Типы при target='panel': известные — 3 совпадения (0) и 2 несовпадения (1.5) →
    # медиана 0.0; добавим ещё одно несовпадение, чтобы медиана стала ненулевой.
    return [
        {"id": 1, "year_built": 2000, "house_type": "panel", "relevance_score": 0.5},
        {"id": 2, "year_built": 2012, "house_type": "panel", "relevance_score": 0.5},
        {"id": 3, "year_built": 2024, "house_type": "brick", "relevance_score": 0.5},
        {"id": 4, "year_built": 1964, "house_type": "brick", "relevance_score": 0.5},
        {"id": 5, "year_built": 1952, "house_type": "brick", "relevance_score": 0.5},
        {"id": 6, "year_built": None, "house_type": None, "relevance_score": 0.5},  # NULL
        {"id": 7, "year_built": 2000, "house_type": None, "relevance_score": 0.5},  # NULL тип
    ]


def test_null_year_gets_pool_median_penalty() -> None:
    """Головной: NULL-год получает медиану штрафов известных, а не 0.

    На origin/main функции нет — кандидат 6 остаётся с 0.5 и обгоняет
    кандидатов 3–5 с известным годом.
    """
    pool = _pool()
    _apply_unknown_attr_penalty(pool, target_year=2000, target_house_type=None, min_n=5)
    by = {c["id"]: c["relevance_score"] for c in pool}
    # Известных годов ШЕСТЬ (ид. 1,2,3,4,5 и 7): штрафы 0,1,2,3,4 и 0 → медиана 1.5.
    # (Первая редакция теста считала пятерых и ждала 2.0 — арифметика, не код.)
    assert by[6] == 0.5 + 1.5, (
        f"NULL-год должен получить медианный штраф 1.5, получил {by[6] - 0.5}"
    )
    assert by[1] == 0.5 and by[3] == 0.5, (
        "известный год не должен трогаться — его уже оштрафовал SQL"
    )


def test_null_house_type_gets_pool_median_penalty() -> None:
    """NULL-тип: медиана штрафов известных типов (3×0, 3×1.5 → медиана 0.75)."""
    pool = _pool()
    # Делаем 3 совпадения и 3 несовпадения: 1,2 panel; 3,4,5 brick; добавим ещё panel.
    pool.append({"id": 8, "year_built": 2000, "house_type": "panel", "relevance_score": 0.5})
    _apply_unknown_attr_penalty(pool, target_year=None, target_house_type="panel", min_n=5)
    by = {c["id"]: c["relevance_score"] for c in pool}
    assert by[6] == 0.5 + 0.75 and by[7] == 0.5 + 0.75, f"NULL-тип: {by[6] - 0.5}, {by[7] - 0.5}"
    assert by[3] == 0.5, "известный несовпадающий тип уже оштрафован SQL — не трогаем"


def test_unknown_target_penalizes_nobody() -> None:
    """Симметрия с SQL: если target неизвестен, SQL не штрафует никого — и мы тоже."""
    pool = _pool()
    _apply_unknown_attr_penalty(pool, target_year=None, target_house_type=None, min_n=5)
    assert all(c["relevance_score"] == 0.5 for c in pool)


def test_sparse_pool_is_skipped() -> None:
    """Контроль от шума: меньше min_n известных значений → сигнал пропущен целиком."""
    pool = [*_pool()[:3], {"id": 9, "year_built": None, "house_type": None, "relevance_score": 0.5}]
    _apply_unknown_attr_penalty(pool, target_year=2000, target_house_type="panel", min_n=5)
    assert all(c["relevance_score"] == 0.5 for c in pool), "медиана по трём строкам — не сигнал"


def test_flag_default_off_means_no_call() -> None:
    """Контроль: дефолт флага OFF — поведение прода не меняется этим PR."""
    from app.core.config import settings

    assert settings.estimate_unknown_attr_penalty_enabled is False


def test_penalty_reproduces_sql_terms_not_its_own_idea_of_distance() -> None:
    """Штраф считается по ТЕМ ЖЕ термам, что SQL: abs(Δyear)/12.0 и 1.5 за тип.

    Ловит «починку», где NULL получал бы произвольную константу вместо медианы
    реально начисленных SQL-штрафов.
    """
    pool = [
        {"id": i, "year_built": 2000 + 12 * i, "house_type": "x", "relevance_score": 0.0}
        for i in range(1, 6)
    ] + [{"id": 99, "year_built": None, "house_type": "x", "relevance_score": 0.0}]
    # штрафы 1,2,3,4,5 → медиана 3.0 (= 36 лет / 12)
    _apply_unknown_attr_penalty(pool, target_year=2000, target_house_type=None, min_n=5)
    assert next(c for c in pool if c["id"] == 99)["relevance_score"] == 3.0


def test_tier_h_and_w_outer_selects_carry_year_and_house_type() -> None:
    """Сторож на «мёртвый флаг»: внешние SELECT тиров H и W обязаны выбирать
    year_built и house_type — иначе Python-слой видит None у ВСЕХ и не штрафует
    никого по построению.

    Именно так флаг и был мёртв в первой редакции: _ANALOG_SELECT_COLS дополнили,
    а внешние списки — нет; A/B на 300 целях ЕКБ дал 0 из 5 915 сменённых слотов,
    probe-лог показал pool=30 null_year=30. Проверка по исходнику запроса, а не по
    живой БД: это инвариант текста SQL.
    """
    import inspect
    import re

    from app.services import estimator as m

    src = inspect.getsource(m)
    # Внешний SELECT тира — блок между `\n<отступ>SELECT\n` и `\n<отступ>FROM base` с ТЕМ
    # ЖЕ отступом (иначе rfind цепляет SELECT внутри оконной функции/CASE).
    outer_blocks = []
    for mt in re.finditer(r"\n(\s+)FROM base\n", src):
        i = mt.start()
        j = src.rfind("\n" + mt.group(1) + "SELECT\n", 0, i)
        if j >= 0:
            outer_blocks.append(src[j:i])
    assert len(outer_blocks) >= 3, f"найдено внешних SELECT: {len(outer_blocks)}"
    # Штрафуемые тиры — ровно те, куда #2012 добавил kitchen/ceiling: H и W.
    # Tier S (same building) и anchor-тир штрафов не применяют — их внешние
    # списки kitchen_area_m2 не содержат, и гейт их не трогает.
    penalized = [b for b in outer_blocks if "kitchen_area_m2" in b]
    assert len(penalized) == 2, f"ожидал 2 внешних SELECT тиров H и W, нашёл {len(penalized)}"
    for b in penalized:
        assert "year_built" in b and "house_type" in b, (
            "внешний SELECT тира не выбирает year_built/house_type — флаг #2936 мёртв:\n" + b[-400:]
        )