"""Гео-согласованный коэффициент asking→sold вне ЕКБ (#3529).

Прод-замер (регион 50): пул даёт 0.891 при 0.73 у Москвы и 0.73 у ЕКБ, потому что обе
стороны фильтровались только по region_code и соединялись только по бакету комнат —
объявления смещены к дальней дешёвой периферии сильнее, чем сделки. Разложение по
кольцам 10 км показывает 0.808 / 0.847 / 0.951 / 0.688 ВНУТРИ колец, причём 76% сделок
лежат в ближних кольцах с 0.81-0.85.

Тесты: (1) арифметика перекоса состава и то, что взвешивание по географии сделок его
снимает; (2) SQL реально соединяет стороны по гео-ключу; (3) путь ЕКБ не тронут;
(4) регион с непересекающейся географией деградирует (строк нет), а не пишет мусор.
"""

import os
import re

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

from app.tasks import asking_to_sold_ratio as ratio_mod

_REGION_SQL = str(ratio_mod._REDERIVE_SQL_REGION.text)
_STATS_SQL = str(ratio_mod._GEO_STATS_SQL_REGION.text)
_EKB_SQL = str(ratio_mod._REDERIVE_SQL.text)


# ── 1. Синтетика: пул завышает, согласованная география — нет ─────────────────

# Ячейки из прод-разложения области 50 по кольцам 10 км (сделки 12 мес vs активные
# объявления): (n_deals, sold_median, n_listings, ask_median).
_PROD_RINGS = [
    (12_189, 175_000, 23_292, 216_666),
    (4_132, 147_026, 10_206, 173_684),
    (1_432, 150_916, 5_243, 158_675),
    (150, 84_390, 2_294, 122_727),
]


def _pooled_ratio(cells: list[tuple[int, int, int, int]]) -> float:
    """СТАРАЯ схема: каждая сторона усредняется по СВОЕЙ популяции, общего ключа нет.

    Ровно то, что делал `FROM deal_side d JOIN ask_side a USING (rooms_bucket)` —
    sold взвешен сделками, ask взвешен объявлениями, география не совпадает.
    """
    sold = sum(n_d * s for n_d, s, _, _ in cells) / sum(n_d for n_d, _, _, _ in cells)
    ask = sum(n_l * a for _, _, n_l, a in cells) / sum(n_l for _, _, n_l, _ in cells)
    return sold / ask


def _geo_matched_ratio(cells: list[tuple[int, int, int, int]]) -> float:
    """НОВАЯ схема: обе стороны взвешены ОДНИМИ весами — числом сделок ячейки."""
    w = [n_d for n_d, _, _, _ in cells]
    sold = sum(n_d * s for n_d, s, _, _ in cells) / sum(w)
    ask = sum(n_d * a for n_d, _, _, a in cells) / sum(w)
    return sold / ask


def test_pool_inflates_and_geo_matching_fixes_it() -> None:
    """Пул завышает коэффициент; та же выборка на согласованной географии — нет.

    Числа взяты из прод-разложения (см. _PROD_RINGS): внутри ячеек 0.69-0.95, а общий
    пул — около 0.89. Перевзвешивание ask-стороны на географию сделок возвращает
    результат в диапазон ближних колец, где лежит 76% сделок.
    """
    pooled = _pooled_ratio(_PROD_RINGS)
    matched = _geo_matched_ratio(_PROD_RINGS)
    per_cell = [s / a for _, s, _, a in _PROD_RINGS]

    # Пул — завышен. NB: 0.858 здесь — НИЖНЯЯ оценка перекоса: пул восстановлен из
    # ячеечных медиан взвешенными средними, тогда как реальный запрос брал МЕДИАНУ
    # пулированной популяции, которая уезжает в дальнюю (объявлений там втрое больше
    # сделок) периферию ещё глубже — прод показывал 0.891 при 0.73 у Москвы и ЕКБ.
    assert pooled > 0.85
    # Согласованная география — внутри диапазона ячеечных отношений, ближе к ближним
    # кольцам (0.81-0.85), где сосредоточена масса сделок.
    assert min(per_cell) <= matched <= max(per_cell)
    assert 0.80 < matched < 0.86
    # И это не косметика: даже по нижней оценке расхождение больше 3 процентных пунктов
    # выкупной цены, по прод-замеру — около 7.
    assert pooled - matched > 0.03


def test_geo_matched_ratio_is_weighted_by_deals_not_listings() -> None:
    """Вес ячейки — число СДЕЛОК: дальняя периферия с горой объявлений не перетягивает.

    Утраиваем объявления в самой дальней (дешёвой) ячейке — согласованный коэффициент
    не двигается вообще, пул уезжает.
    """
    skewed = [
        (n_d, s, n_l * 3 if i == len(_PROD_RINGS) - 1 else n_l, a)
        for i, (n_d, s, n_l, a) in enumerate(_PROD_RINGS)
    ]
    assert _geo_matched_ratio(skewed) == _geo_matched_ratio(_PROD_RINGS)
    assert _pooled_ratio(skewed) != _pooled_ratio(_PROD_RINGS)


def test_sql_implements_the_same_weighted_formula() -> None:
    """Формула из теста выше — ровно то, что считает SQL (иначе двойник разъедется).

    ratio = Σ(w·sold) / Σ(w·ask), w = n_deals; оба медианных столбца — взвешенные
    средние с ТЕМИ ЖЕ весами, поэтому ratio == sold_median/ask_median построчно.
    """
    flat = re.sub(r"\s+", " ", _REGION_SQL)
    ratio_expr = "CAST(SUM(sold_median * n_deals) / SUM(ask_median * n_deals) AS numeric)"
    assert ratio_expr + " AS ratio" in flat
    assert "round(SUM(sold_median * n_deals) / SUM(n_deals))::bigint AS sold_median" in flat
    assert "round(SUM(ask_median * n_deals) / SUM(n_deals))::bigint AS ask_median" in flat
    # Веса — сделки, не объявления.
    assert "SUM(sold_median * n_listings)" not in flat
    assert "SUM(ask_median * n_listings)" not in flat


# ── 2. Гео-ключ есть И в фильтре, И в соединении ──────────────────────────────


def test_region_sql_joins_both_sides_on_the_same_cells() -> None:
    """Старое соединение было ТОЛЬКО по бакету комнат — гео-ключа в нём не было вовсе."""
    flat = re.sub(r"\s+", " ", _REGION_SQL)
    # Ячейка строится ОДИНАКОВО с обеих сторон (одни и те же градусные шаги).
    assert flat.count(f"FLOOR(ST_Y(geom) / {ratio_mod._CELL_LAT_DEG}) AS cell_lat") == 2
    assert flat.count(f"FLOOR(ST_X(geom) / {ratio_mod._CELL_LON_DEG}) AS cell_lon") == 2
    # Соединение сторон несёт гео-ключ.
    assert "JOIN ask_cell a USING (cell_lat, cell_lon)" in flat
    assert "JOIN ask_cell_bucket a USING (cell_lat, cell_lon, rooms_bucket)" in flat
    # Прежних «слепых» соединений не осталось.
    assert "JOIN ask_side a USING (rooms_bucket)" not in flat
    assert "CROSS JOIN ask_global" not in flat
    # Порог ячейки — на ОБЕИХ сторонах.
    assert f"d.n_deals >= {ratio_mod._CELL_MIN_DEALS}" in flat
    assert f"a.n_listings >= {ratio_mod._CELL_MIN_LISTINGS}" in flat
    # Публикуемый барьер 30/30 сохранён (теперь на сумме по удержанным ячейкам).
    assert "HAVING SUM(n_deals) >= 30 AND SUM(n_listings) >= 30" in flat


def test_region_sql_stays_psycopg_v3_safe_and_region_scoped() -> None:
    for sql in (_REGION_SQL, _STATS_SQL):
        assert re.search(r":[a-z_]+::[a-z]", sql) is None
        assert "city ILIKE" not in sql
    # Деривация собирает каждую сторону по разу; stats-запрос дополнительно считает
    # ВСЕ строки региона (включая без geom) — отсюда четыре предиката вместо двух.
    assert _REGION_SQL.count("region_code = CAST(:region_code AS int)") == 2
    assert _STATS_SQL.count("region_code = CAST(:region_code AS int)") == 4
    assert "CAST(:region_code AS int) FROM global_row" in _REGION_SQL
    assert "CAST(:region_code AS int) FROM per_bucket" in _REGION_SQL


def test_district_column_stays_empty() -> None:
    """Идентификатор ячейки в district НЕ пишется — ячейка промежуточная, не публикуемая.

    Потребитель (estimator._get_asking_sold_ratio) читает строго `district = ''`; строка
    на выходе одна на (регион, бакет) — агрегат по всем ячейкам.
    """
    assert _REGION_SQL.count("''::text") == 2
    assert "cell_lat AS district" not in _REGION_SQL
    assert "AS district" in _REGION_SQL


# ── 3. Путь ЕКБ (66) не тронут ────────────────────────────────────────────────


def test_ekb_path_untouched_by_geo_matching() -> None:
    """Регион 66 обязан остаться байт-в-байт прежним — там своя калибровка (#C2/#2583)."""
    assert "cell_lat" not in _EKB_SQL
    assert "ST_X(" not in _EKB_SQL and "ST_Y(" not in _EKB_SQL
    assert _EKB_SQL.count("AND city ILIKE :asking_city") == 2
    assert _EKB_SQL.count("AND (city IS NULL OR city ILIKE :asking_city)") == 2
    assert "FROM deal_side d\n        JOIN ask_side a USING (rooms_bucket)" in _EKB_SQL
    assert "CROSS JOIN ask_global a" in _EKB_SQL
    assert "basis, 66 FROM global_row" in _EKB_SQL
    # Гео-статистика считается только для прочих регионов — 66 её не касается.
    assert "_GEO_STATS_SQL_REGION" in ratio_mod.__dict__ or True
    assert 66 not in ratio_mod._OTHER_REGION_CODES


# ── 4. Гард: непересекающаяся география → деградация, а не мусор ──────────────


def _stats(**over: int) -> dict[str, int]:
    base = {
        "deals_total": 1000,
        "deals_geo": 900,
        "listings_total": 2000,
        "listings_geo": 2000,
        "cells_deal": 20,
        "cells_ask": 25,
        "cells_both_sides": 18,
        "cells_matched": 15,
        "deals_in_cells": 800,
        "listings_in_cells": 1200,
    }
    base.update(over)
    return base


def test_verdict_ok_on_healthy_region() -> None:
    ok, reason = ratio_mod.geo_region_verdict(50, _stats())
    assert ok is True
    assert reason == "ok"


def test_verdict_degrades_on_disjoint_geography() -> None:
    """Сделки и объявления в разных местах — регион не получает НИ ОДНОЙ строки."""
    ok, reason = ratio_mod.geo_region_verdict(50, _stats(cells_matched=1, deals_in_cells=30))
    assert ok is False
    assert "не пересекаются" in reason

    # Пересечение есть, но в него попала меньшая часть сделок — тоже деградация.
    ok, reason = ratio_mod.geo_region_verdict(50, _stats(deals_in_cells=100))
    assert ok is False
    assert "11% сделок" in reason


def test_verdict_degrades_when_a_side_has_no_coordinates() -> None:
    ok, reason = ratio_mod.geo_region_verdict(50, _stats(deals_geo=100))
    assert ok is False
    assert "сделки без geom" in reason
    ok, reason = ratio_mod.geo_region_verdict(50, _stats(listings_geo=10))
    assert ok is False
    assert "объявления без geom" in reason
    ok, reason = ratio_mod.geo_region_verdict(50, None)
    assert ok is False


def test_verdict_degrades_on_empty_region() -> None:
    ok, _ = ratio_mod.geo_region_verdict(50, _stats(deals_total=0, deals_geo=0))
    assert ok is False


def test_counters_expose_composition_and_missing_geom() -> None:
    """Пункты 3 и 5: строки без geom и масса вне пересечения — в счётчиках, не в тишине."""
    c = ratio_mod._geo_region_counters(50, _stats())
    assert c["geo_r50_cells_matched"] == 15
    assert c["geo_r50_cells_dropped"] == 3  # есть обе стороны, но одна тоньше порога
    assert c["geo_r50_deals_no_geom_pct"] == 10  # 100 из 1000 сделок без координат
    assert c["geo_r50_listings_no_geom_pct"] == 0
    assert c["geo_r50_deals_outside_pct"] == 11  # 100 из 900 геокодированных сделок
    assert c["geo_r50_listings_outside_pct"] == 40  # дальняя периферия отброшена намеренно
    assert all(isinstance(v, int) for v in c.values())


def test_skipped_region_still_deletes_stale_rows() -> None:
    """DELETE идёт ДО решения писать/не писать — старый пуловый коэффициент не остаётся."""
    import inspect

    body = inspect.getsource(ratio_mod.recompute_asking_to_sold_ratios)
    del_pos = body.index('db.execute(_DELETE_SQL_REGION, {"region_code": region_code})')
    guard_pos = body.index("if not ok:")
    insert_pos = body.index("db.execute(_REDERIVE_SQL_REGION, params)")
    assert del_pos < guard_pos < insert_pos
    assert "continue" in body[guard_pos:insert_pos]