feat(forecasting): §9.x→§22 orchestrator + fix supply-side district resolution (3a)

Add build_site_finder_report (orchestrator.py): computes the §9.x layers (market
metrics, supply layers, future-supply pressure, demand/supply forecast, scenarios,
score card, special indices, recommendation overlay) with their heterogeneous
signatures and feeds the PURE assemble_report → §22 SiteFinderReport. Default segment
= modal competitor class; each §9.x call _safe_call-wrapped (graceful). Standalone —
no endpoint/Celery/persistence (that is 3b).

Prod ground-truth of the orchestrator surfaced a false-BUY bug: future_supply
(compute_future_supply_pressure) read the mixed-vocab persisted view
v_supply_layers_latest by a SCALAR admin district_name, missing all Layer-1
micro-keyed rows → admin parcel (Кировский) got supply=0 → false +1.0 deficit →
'Строить: недонасыщен' headline despite ~45k competing units. Fix: resolve
admin→micros, filter district_name = ANY(CAST(:names AS text[])) where names =
micros (L1) + admin (L2/L3), with :has_district EKB-wide guard (extends PR #1054's
resolver to the persisted-view path it missed). future_supply is the only
v_supply_layers_latest consumer on the forecast path (verified).

Prod after: Кировский supply 0→~42953, deficit +1.0→−1.0 (honest oversupply),
MOI 0→116.6, false-BUY headline gone, overall 0.734→0.42. 80 module tests pass
(signature-trap + resolver-regression guards genuine); ruff clean. Refs #961 #969.

backend/app/services/forecasting/orchestrator.py

@@ -0,0 +1,340 @@
+"""§9.x → §22 оркестратор: посчитать форсайт-слои и собрать `SiteFinderReport`.
+
+Step 3a проводки форсайта (955-A2). §22-сборщик `assemble_report` (#988) ЧИСТЫЙ — он
+лишь РАСКЛАДЫВАЕТ уже-посчитанные advisory-входы по восьми секциям, в БД не ходит и
+§9.x-сервисы не зовёт. ВЫЗЫВАЮЩИЙ (этот модуль) обязан посчитать §9.x-слои и скормить
+их сборщику. Здесь — этот вызывающий: ОРКЕСТРАЦИОННЫЙ ШОВ, НЕ новая §9.x-математика.
+
+Что делает `build_site_finder_report`:
+  1. Дефолт сегмента (audit Q1): если `obj_class` не задан — выводим МОДАЛЬНЫЙ (самый
+     частый) класс конкурентов из `analyze["competitors"]`. Нет конкурентов / нет
+     классов → obj_class=None (ЕКБ-wide all-class).
+  2. Считает §9.x-слои, КАЖДЫЙ через `_safe_call` (зеркало product_scoring._safe_call):
+     сбой/тонкие данные одного слоя → этот слой None, отчёт всё равно собирается
+     (graceful: тонкий/сбойный слой ДЕГРАДИРУЕТ отчёт, НЕ роняет его целиком). Сами
+     §9.x-модули уже graceful внутри — это belt-and-suspenders на оркестрационном шве.
+  3. Раскладывает слои в `assemble_report` (каждый в свой слот, с ПРАВИЛЬНОЙ —
+     ГЕТЕРОГЕННОЙ! — сигнатурой: forecasts/scenarios/special_indices берут `horizons=`,
+     score_card/future_supply берут `horizon_months=`, overlay берёт `target_class=`).
+
+ADVISORY: весь форсайт-стек советующий (cap ≤ 'medium', #990) — отчёт лишь СОБИРАЕТ
+advisory-выводы, не основание для инвест-решения. `advisory=True` прокидывается всюду.
+
+ГЕТЕРОГЕННЫЕ сигнатуры §9.x (audit поймал ловушки — НЕ унифицированы):
+  • compute_market_metrics(db, *, district, premise_kind)           — §9.2
+  • compute_all_layers(db, *, district, premise_kind)               — §9.3 (список строк)
+  • compute_future_supply_pressure(db, *, district, horizon_months) — §9.3 давление
+  • compute_demand_supply_forecast(db, *, spec, district, cad_num, horizons)   — §9.8
+  • compute_scenarios(db, *, spec, district, cad_num, horizons)      — §11
+  • compute_score_card(db, *, spec, district, cad_num, horizon_months)  — §14.2
+  • compute_special_indices(db, *, spec, district, cad_num, horizons)   — §25
+  • build_forecast_overlay(db, *, district, cad_num, horizon_months, target_class) — §10
+
+МАКРО-СВЕЖЕСТЬ (audit MEDIUM, отложено в 3b): domrf-месячный макро ~123 дня устаревает
+(key_rate свежий). У §22-схемы v1.0 (frozen `SiteFinderReport`/`ReportMeta`) НЕТ чистого
+слота под caveat без смены контракта (новое поле + bump schema_version) — что вне 3a.
+Поэтому здесь только ЛОГИРУЕМ `macro_as_of` (последний наблюдаемый месяц макро), чтобы
+сигнал не терялся; проводка его в тело отчёта — follow-up 3b (вместе с endpoint/Celery).
+"""
+
+from __future__ import annotations
+
+import logging
+from collections import Counter
+from collections.abc import Sequence
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from app.services.forecasting.demand_supply_forecast import compute_demand_supply_forecast
+from app.services.forecasting.macro_series import get_monthly_macro
+from app.services.forecasting.product_scoring import compute_score_card
+from app.services.forecasting.recommendation import build_forecast_overlay
+from app.services.forecasting.report import SiteFinderReport
+from app.services.forecasting.report_assembler import assemble_report
+from app.services.forecasting.sales_series import SegmentSpec
+from app.services.forecasting.scenarios import compute_scenarios
+from app.services.forecasting.special_indices import compute_special_indices
+from app.services.site_finder.future_supply import compute_future_supply_pressure
+from app.services.site_finder.market_metrics import compute_market_metrics
+from app.services.site_finder.supply_layers import SupplyLayerRow, compute_all_layers
+
+logger = logging.getLogger(__name__)
+
+# Горизонты прогноза по умолчанию (мес) — зеркало §9.x default grid (#952/#984/#986).
+_DEFAULT_HORIZONS: tuple[int, ...] = (6, 12, 18, 24)
+
+# Основной горизонт для single-горизонтных сервисов (score_card / overlay / future_supply).
+# Берём 12 (типовой средне-срочный продуктовый горизонт) если он в сетке, иначе первый.
+_PREFERRED_PRIMARY_HORIZON: int = 12
+
+# Тип помещения — единый дефолт всего §9.x-стека (жилой сегмент ЕКБ).
+_PREMISE_KIND: str = "квартира"
+
+# Весь отчёт советующий (зеркало advisory-cap стека) — прокидывается в #990 cap.
+_ADVISORY: bool = True
+
+# Маппинг layer-номера → ключ агрегата supply_layers-сводки (#950, §9.3).
+_LAYER_UNIT_KEY: dict[int, str] = {1: "open_units", 2: "hidden_units", 3: "future_units"}
+
+
+def _modal_obj_class(competitors: Any) -> str | None:
+    """Вывести МОДАЛЬНЫЙ (самый частый) класс конкурентов как дефолт сегмента. PURE.
+
+    audit Q1: user-дефолт сегмента = модальный класс конкуренции вокруг участка. Читаем
+    `obj_class` (или legacy-ключ `class`) каждого dict'а-конкурента, считаем Counter,
+    берём самый частый. Ничьи разрешаются детерминированно (по первому встреченному —
+    стабильный порядок Counter.most_common для равных частот).
+
+    Graceful: competitors не список / пуст / ни у одного нет класса → None
+    (ЕКБ-wide all-class сегмент — честно «класс не уточнён»).
+
+    Args:
+        competitors: `analyze["competitors"]` (loosely-typed: ожидаем список dict'ов).
+
+    Returns:
+        Модальный класс (str) или None.
+    """
+    if not isinstance(competitors, list):
+        return None
+    classes: list[str] = []
+    for comp in competitors:
+        if not isinstance(comp, dict):
+            continue
+        value = comp.get("obj_class")
+        if not (isinstance(value, str) and value):
+            value = comp.get("class")
+        if isinstance(value, str) and value:
+            classes.append(value)
+    if not classes:
+        return None
+    # most_common(1) — детерминированно: при равенстве частот сохраняется порядок вставки.
+    return Counter(classes).most_common(1)[0][0]
+
+
+def _primary_horizon(horizons: Sequence[int]) -> int:
+    """Основной горизонт для single-горизонтных сервисов. PURE.
+
+    _PREFERRED_PRIMARY_HORIZON (12 мес) если он в сетке, иначе первый горизонт сетки.
+    Пустая сетка не ожидается (вызывающий даёт дефолт), но на всякий случай → 12.
+    """
+    if _PREFERRED_PRIMARY_HORIZON in horizons:
+        return _PREFERRED_PRIMARY_HORIZON
+    return horizons[0] if horizons else _PREFERRED_PRIMARY_HORIZON
+
+
+def _safe_call(label: str, fn: Any) -> Any:
+    """Вызвать §9.x-сервис graceful: сбой → None + logger.exception (не crash отчёта).
+
+    Зеркало product_scoring._safe_call: любой §9.x-слой может бросить (тонкие данные / нет
+    геометрии cad_num / сбой SQL) — логируем и возвращаем None, чтобы соответствующая
+    секция отчёта осталась пустой/частичной, а остальной отчёт собрался. Намеренно ловим
+    широкий Exception (изоляция одного слоя от отчёта) с ОБЯЗАТЕЛЬНЫМ logger.exception —
+    НЕ молчаливое глотание. §9.x уже graceful внутри; это belt-and-suspenders на шве.
+
+    Args:
+        label: имя слоя для лога (диагностика какой §9.x-вызов деградировал).
+        fn: нулевой-аргумент thunk вокруг §9.x-вызова.
+
+    Returns:
+        Результат `fn()` или None при исключении.
+    """
+    try:
+        return fn()
+    except Exception:
+        logger.exception("orchestrator: §9.x layer %s failed → section degraded", label)
+        return None
+
+
+def _summarize_supply_layers(rows: Sequence[SupplyLayerRow] | None) -> dict[str, Any] | None:
+    """Свернуть строки §9.3-слоёв (#950) в сводку-dict для слота `supply_layers`. PURE.
+
+    `compute_all_layers` отдаёт ПЛОСКИЙ список SupplyLayerRow (L1+L2+L3). Слот
+    `supply_layers` сборщика (#988 `_build_market_now`/`_domrf_coverage`) ждёт ОДИН dict,
+    не список — поэтому агрегируем: Σ units по слою (open/hidden/future_units) + сырой
+    список строк (`rows`) для прозрачности. domrf_coverage здесь НЕ выводим (нет дешёвого
+    продьюсера в сервисах) — сборщик возьмёт fallback из `analyze.market_data_coverage_pct`.
+
+    Graceful: None/пустой список → None (слот останется пустым — секция частична).
+
+    Args:
+        rows: вывод `compute_all_layers` (или None при сбое слоя).
+
+    Returns:
+        Сводка-dict (open_units/hidden_units/future_units/n_rows/rows) или None.
+    """
+    if not rows:
+        return None
+    units_by_key: dict[str, int] = {"open_units": 0, "hidden_units": 0, "future_units": 0}
+    row_dicts: list[dict[str, Any]] = []
+    for row in rows:
+        row_dicts.append(row.as_dict())
+        key = _LAYER_UNIT_KEY.get(int(row.layer))
+        if key is not None and row.units_estimate is not None:
+            units_by_key[key] += int(row.units_estimate)
+    return {
+        "open_units": units_by_key["open_units"],
+        "hidden_units": units_by_key["hidden_units"],
+        "future_units": units_by_key["future_units"],
+        "n_rows": len(row_dicts),
+        "rows": row_dicts,
+    }
+
+
+def _macro_as_of(macro: Any) -> str | None:
+    """Последний наблюдаемый месяц макро-ряда (ISO) — для caveat-лога свежести. PURE.
+
+    `get_monthly_macro` отдаёт MonthlyMacro по ВОЗРАСТАНИЮ month по непрерывной сетке;
+    последний элемент — самый поздний месяц с (carry-forward) данными. Используем только
+    для ЛОГА свежести (audit MEDIUM) — проводка в тело отчёта отложена в 3b.
+
+    Graceful: пустой/None ряд → None.
+    """
+    if not macro:
+        return None
+    last = macro[-1]
+    month = getattr(last, "month", None)
+    isoformat = getattr(month, "isoformat", None)
+    if callable(isoformat):
+        return isoformat()
+    return str(month) if month is not None else None
+
+
+def build_site_finder_report(
+    db: Session,
+    *,
+    analyze: dict[str, Any],
+    cad_num: str,
+    district: str | None,
+    obj_class: str | None = None,
+    horizons: Sequence[int] = _DEFAULT_HORIZONS,
+) -> SiteFinderReport:
+    """Посчитать §9.x-слои и собрать §22 `SiteFinderReport` для участка (ОРКЕСТРАЦИЯ).
+
+    ШОВ между §9.x-сервисами (которые ходят в БД) и ЧИСТЫМ §22-сборщиком (#988
+    `assemble_report`, который только раскладывает уже-посчитанные входы). НЕ новая
+    §9.x-математика — только проводка: считаем слои → раскладываем в сборщик.
+
+    Поток:
+      1. Дефолт сегмента: obj_class=None → модальный класс конкурентов из `analyze`
+         (audit Q1). Строим `spec = SegmentSpec(obj_class=…, district=…)`.
+      2. Считаем восемь §9.x-слоёв, КАЖДЫЙ через `_safe_call` (graceful: сбой слоя →
+         None → его секция деградирует, отчёт собирается). Сигнатуры ГЕТЕРОГЕННЫ
+         (см. module docstring): forecasts/scenarios/special_indices ← `horizons=`;
+         score_card/future_supply ← `horizon_months=`; overlay ← `target_class=`.
+      3. Раскладываем в `assemble_report` (каждый слой в свой слот; supply-строки
+         сворачиваем в сводку-dict; overlay идёт в `recommendation_overlay=`).
+
+    МАКРО-СВЕЖЕСТЬ (audit MEDIUM): логируем `macro_as_of`; проводка в отчёт — 3b
+    (нет чистого слота в схеме v1.0 без смены контракта). См. module docstring.
+
+    Args:
+        db: SQLAlchemy sync Session (read-only по факту — §9.x только читают).
+        analyze: dict вывода analyze_parcel (loosely-typed; читаем defensive `.get()`).
+        cad_num: кадастровый номер участка (вход §9.7 конкурентов + центроид POI).
+        district: район участка (None → ЕКБ-wide; §9.x резолвят админ-имя → микро).
+        obj_class: целевой класс; None → выводим модальный класс конкурентов (Q1).
+        horizons: горизонты прогноза (мес; по умолчанию _DEFAULT_HORIZONS).
+
+    Returns:
+        SiteFinderReport — восемь секций §22, заполнены настолько, насколько хватило
+        §9.x-слоёв (частичный отчёт валиден); advisory=True; `as_dict()` JSON-safe.
+    """
+    horizon_list = list(horizons)
+    primary = _primary_horizon(horizon_list)
+
+    # ── 1. Дефолт сегмента: модальный класс конкурентов (audit Q1) ──────────────
+    resolved_class = (
+        obj_class if obj_class is not None else _modal_obj_class(analyze.get("competitors"))
+    )
+    spec = SegmentSpec(obj_class=resolved_class, district=district)
+
+    logger.info(
+        "build_site_finder_report: cad_num=%s district=%s obj_class=%s (given=%s) "
+        "horizons=%s primary=%d",
+        cad_num,
+        district,
+        resolved_class,
+        obj_class,
+        horizon_list,
+        primary,
+    )
+
+    # ── 2. §9.x-слои, каждый graceful через _safe_call (ГЕТЕРОГЕННЫЕ сигнатуры) ──
+    market_metrics = _safe_call(
+        "market_metrics",
+        lambda: compute_market_metrics(db, district=district, premise_kind=_PREMISE_KIND),
+    )
+    supply_rows = _safe_call(
+        "supply_layers",
+        lambda: compute_all_layers(db, district=district, premise_kind=_PREMISE_KIND),
+    )
+    supply_layers = _summarize_supply_layers(supply_rows)
+    future_supply = _safe_call(
+        "future_supply",
+        lambda: compute_future_supply_pressure(
+            db, district=district, horizon_months=primary, premise_kind=_PREMISE_KIND
+        ),
+    )
+    forecasts = _safe_call(
+        "demand_supply_forecast",
+        lambda: compute_demand_supply_forecast(
+            db, spec=spec, district=district, cad_num=cad_num, horizons=horizon_list
+        ),
+    )
+    scenarios = _safe_call(
+        "scenarios",
+        lambda: compute_scenarios(
+            db, spec=spec, district=district, cad_num=cad_num, horizons=horizon_list
+        ),
+    )
+    product_scores = _safe_call(
+        "score_card",
+        lambda: compute_score_card(
+            db, spec=spec, district=district, cad_num=cad_num, horizon_months=primary
+        ),
+    )
+    special_indices = _safe_call(
+        "special_indices",
+        lambda: compute_special_indices(
+            db, spec=spec, district=district, cad_num=cad_num, horizons=horizon_list
+        ),
+    )
+    recommendation_overlay = _safe_call(
+        "forecast_overlay",
+        lambda: build_forecast_overlay(
+            db,
+            district=district,
+            cad_num=cad_num,
+            horizon_months=primary,
+            target_class=resolved_class,
+        ),
+    )
+
+    # ── Макро-свежесть (audit MEDIUM): только лог; проводка в отчёт — 3b ─────────
+    macro = _safe_call("monthly_macro", lambda: get_monthly_macro(db))
+    macro_as_of = _macro_as_of(macro)
+    if macro_as_of is not None:
+        logger.info(
+            "build_site_finder_report: macro_as_of=%s (domrf monthly may be stale; "
+            "surfacing into report deferred to 3b)",
+            macro_as_of,
+        )
+
+    # ── 3. Раскладка в ЧИСТЫЙ §22-сборщик (#988) — каждый слой в свой слот ───────
+    return assemble_report(
+        analyze,
+        market_metrics=market_metrics,
+        supply_layers=supply_layers,
+        forecasts=forecasts,
+        future_supply=future_supply,
+        scenarios=scenarios,
+        recommendation_overlay=recommendation_overlay,
+        product_scores=product_scores,
+        special_indices=special_indices,
+        segment=spec.as_dict(),
+        cad_num=cad_num,
+        district=district,
+        horizons=horizon_list,
+        advisory=_ADVISORY,
+    )

backend/app/services/site_finder/future_supply.py

@@ -52,6 +52,7 @@ from typing import Any, Literal
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
+from app.services.site_finder.district_resolver import resolve_objective_districts
 from app.services.site_finder.market_metrics import compute_market_metrics
 
 logger = logging.getLogger(__name__)
@@ -266,6 +267,15 @@ def _min_confidence(values: Sequence[Confidence | None]) -> Confidence:
 # / insider_manual живут только в таблице и невыводимы пересчётом. Берём построчно
 # (layer / units / expected_online_date / confidence) — взвешивание L3 по горизонту
 # делаем в Python (per-row дата vs today+horizon), агрегацию L1/L2 — тоже в Python.
+#
+# District-фильтр — по НАБОРУ имён (:names), не по скаляру: view МИКС-вокабулярный.
+# Layer1 (open) строки ключуются неформальным МИКРО-районом (objective_lots.district,
+# напр. «Втузгородок»/«ЖБИ»), а Layer2/3 (hidden/future) — АДМИН-районом (domrf
+# district_name, напр. «Кировский»). Скалярный фильтр по админ-имени промахивался бы
+# мимо ВСЕХ L1 микро-строк → open=0 → ложный дефицит. Резолвер разворачивает админ →
+# микро, а сам админ добавляем в набор для L2/3 (см. compute_future_supply_pressure).
+# Bug: парцель 66:41:0702048:27 (Кировский) давал projected_supply=0 → ложный BUY.
+# psycopg v3: ANY(CAST(:names AS text[])) — массивный bind, НИКОГДА :x::type.
 _SUPPLY_LAYERS_SQL = text(
     """
     SELECT
@@ -275,8 +285,8 @@ _SUPPLY_LAYERS_SQL = text(
         confidence
       FROM v_supply_layers_latest
      WHERE (
-            CAST(:district AS text) IS NULL
+            CAST(:has_district AS boolean) IS FALSE
-            OR district_name = CAST(:district AS text)
+            OR district_name = ANY(CAST(:names AS text[]))
            )
     """
 )
@@ -292,6 +302,9 @@ def compute_future_supply_pressure(
     """Вычислить индекс будущего давления предложения ТЗ §9.3 для района (на лету).
 
     Шаги:
+      0. Резолвить district (админ-имя ЕКБ) → набор имён для МИКС-вокабулярного view:
+         resolve_objective_districts даёт МИКРО-районы (ключ L1-строк), к ним добавляем
+         сам админ (ключ L2/3-строк). None → EKB-wide (без фильтра).
       1. Прочитать ВСЕ свежие supply-строки района из VIEW `v_supply_layers_latest`
          (включает невыводимые L3 graddoc_stub/insider_manual).
       2. Агрегировать: open=Σ L1, hidden=Σ L2, future=Σ L3·горизонт-вес (NULL-дата →
@@ -304,8 +317,27 @@ def compute_future_supply_pressure(
     Возвращает FutureSupplyPressure ВСЕГДА (даже на пустых данных): тогда index=None,
     months_of_pressure=None, confidence='low'. Никогда не бросает, никогда деления на 0.
     """
+    # ── 0. District → набор имён для МИКС-вокабулярного view ───────────────────
+    # view ключует L1-строки МИКРО-районом (objective_lots.district), а L2/3 —
+    # АДМИН-районом (domrf district_name). resolve_objective_districts разворачивает
+    # админ-имя в чистые микро (для L1); сам админ добавляем в набор (для L2/3). Для
+    # raw-микро входа резолвер вернёт [micro] → names=[micro] (legacy сохранён). None
+    # (вне 8 полигонов / 'не определён') → has_district=False → EKB-wide без фильтра.
+    # micros is None ⇔ EKB-wide (вход None/'не определён' ИЛИ админ без чистых алиасов);
+    # зеркало market_metrics has_district. Иначе micros — непустой набор микро-районов.
+    micros = resolve_objective_districts(db, district)
+    names: list[str] = []
+    if micros is not None and district is not None:
+        # dedup сохраняя порядок: микро (L1) + сам админ-район (L2/3). Инлайн-проверка
+        # district is not None (а не через has_district-флаг) сужает тип для mypy —
+        # resolver вернул бы None на None-входе, так что эта ветка ⇔ has_district.
+        names = list(dict.fromkeys([*micros, district]))
+    has_district = bool(names)
+
     # ── 1. Supply-строки района из VIEW (graceful: сбой/пусто → []) ────────────
-    rows = _query_supply_rows(db, {"district": district})
+    rows = _query_supply_rows(
+        db, {"has_district": has_district, "names": names, "district": district}
+    )
 
     # ── 2. Агрегация по слоям + сбор компонентных confidence ──────────────────
     today = date.today()

backend/tests/services/forecasting/test_orchestrator.py

@@ -0,0 +1,332 @@
+"""Unit-тесты §9.x → §22 оркестратора (`build_site_finder_report`, step 3a, 955-A2).
+
+Оркестратор — ШОВ: считает §9.x-слои и раскладывает их в ЧИСТЫЙ §22-сборщик
+(`assemble_report`). Тесты пинят ТРИ контракта шва, БЕЗ живой БД (db = sentinel,
+§9.x-вызовы запатчены):
+
+  • МОДАЛЬНЫЙ дефолт класса (audit Q1): competitors [Комфорт, Комфорт, Бизнес] →
+    obj_class="Комфорт"; пустые competitors → None (ЕКБ-wide). Явный obj_class имеет
+    приоритет над выводом.
+  • ПРОВОДКА/СИГНАТУРЫ (ловушки audit): патчим §9.x на sentinel'ы, ассертим что
+    `assemble_report` получил КАЖДЫЙ вход в правильном слоте, а §9.x вызваны с
+    ПРАВИЛЬНОЙ ГЕТЕРОГЕННОЙ сигнатурой (forecasts/scenarios/special_indices ← `horizons=`;
+    score_card/future_supply ← `horizon_months=`; overlay ← `target_class=`). Регрессия
+    (не тот kwarg) → тест падает.
+  • GRACEFUL: один §9.x-слой бросает → отчёт всё равно собирается (его слот None),
+    исключение НЕ всплывает.
+
+Детерминированно, без сети, без LLM. DATABASE_URL до импорта app-модулей (зеркало
+соседних тестов — на случай side-effect'ов импорта пакета forecasting).
+"""
+
+from __future__ import annotations
+
+import contextlib
+import os
+from collections.abc import Iterator, Mapping
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.services.forecasting.orchestrator import (
+    _modal_obj_class,
+    _primary_horizon,
+    _summarize_supply_layers,
+    build_site_finder_report,
+)
+from app.services.forecasting.report import SiteFinderReport
+
+# Полностью-квалифицированный путь к каждому §9.x-символу как он ИМПОРТИРОВАН в
+# orchestrator (патчим там, где используется, не там, где определён).
+_MOD = "app.services.forecasting.orchestrator"
+
+
+def _analyze(competitors: list[dict[str, Any]] | None = None) -> dict[str, Any]:
+    """Минимальный analyze-dict (только то, что читает оркестратор/сборщик)."""
+    return {
+        "cad_num": "66:41:0702048:27",
+        "district": {"district_name": "Ленинский"},
+        "competitors": competitors if competitors is not None else [],
+    }
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Модальный дефолт класса (audit Q1).
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class TestModalObjClass:
+    def test_modal_picks_most_common(self) -> None:
+        comps = [{"obj_class": "Комфорт"}, {"obj_class": "Комфорт"}, {"obj_class": "Бизнес"}]
+        assert _modal_obj_class(comps) == "Комфорт"
+
+    def test_empty_competitors_none(self) -> None:
+        assert _modal_obj_class([]) is None
+
+    def test_competitors_without_class_none(self) -> None:
+        # есть конкуренты, но ни у одного нет класса → None (честно «не уточнён»).
+        assert _modal_obj_class([{"obj_id": 1}, {"comm_name": "ЖК Альфа"}]) is None
+
+    def test_legacy_class_key_supported(self) -> None:
+        # legacy-ключ `class` (а не `obj_class`) тоже читается.
+        assert _modal_obj_class([{"class": "Стандарт"}, {"class": "Стандарт"}]) == "Стандарт"
+
+    def test_obj_class_preferred_over_legacy(self) -> None:
+        comps = [{"obj_class": "Бизнес", "class": "Комфорт"}]
+        assert _modal_obj_class(comps) == "Бизнес"
+
+    def test_non_list_none(self) -> None:
+        assert _modal_obj_class(None) is None
+        assert _modal_obj_class("Комфорт") is None  # type: ignore[arg-type]
+
+
+class TestPrimaryHorizon:
+    def test_prefers_12_when_present(self) -> None:
+        assert _primary_horizon([6, 12, 18, 24]) == 12
+
+    def test_falls_back_to_first_when_no_12(self) -> None:
+        assert _primary_horizon([6, 18, 24]) == 6
+
+    def test_empty_defaults_to_12(self) -> None:
+        assert _primary_horizon([]) == 12
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Хелпер сводки supply-слоёв (#950 → слот supply_layers сборщика).
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class TestSummarizeSupplyLayers:
+    def test_none_or_empty_none(self) -> None:
+        assert _summarize_supply_layers(None) is None
+        assert _summarize_supply_layers([]) is None
+
+    def test_aggregates_units_by_layer(self) -> None:
+        rows = [
+            MagicMock(layer=1, units_estimate=1200, as_dict=lambda: {"layer": 1}),
+            MagicMock(layer=2, units_estimate=800, as_dict=lambda: {"layer": 2}),
+            MagicMock(layer=3, units_estimate=500, as_dict=lambda: {"layer": 3}),
+            MagicMock(layer=1, units_estimate=None, as_dict=lambda: {"layer": 1}),  # None skip
+        ]
+        summary = _summarize_supply_layers(rows)
+        assert summary is not None
+        assert summary["open_units"] == 1200
+        assert summary["hidden_units"] == 800
+        assert summary["future_units"] == 500
+        assert summary["n_rows"] == 4
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# ПРОВОДКА + СИГНАТУРНЫЕ ловушки (audit): §9.x → sentinel, ассерт слот+kwarg.
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def _patched_layers() -> dict[str, Any]:
+    """Контекст-менеджеры-патчи всех §9.x на распознаваемые sentinel-возвраты."""
+    # supply_layers возвращает СПИСОК строк (оркестратор свернёт в сводку-dict).
+    supply_row = MagicMock(layer=1, units_estimate=10, as_dict=lambda: {"layer": 1})
+    return {
+        "compute_market_metrics": MagicMock(return_value="MM"),
+        "compute_all_layers": MagicMock(return_value=[supply_row]),
+        "compute_future_supply_pressure": MagicMock(return_value="FSP"),
+        "compute_demand_supply_forecast": MagicMock(return_value=["F6", "F12"]),
+        "compute_scenarios": MagicMock(return_value=["SC"]),
+        "compute_score_card": MagicMock(return_value="PSC"),
+        "compute_special_indices": MagicMock(return_value="SI"),
+        "build_forecast_overlay": MagicMock(return_value="OVL"),
+        "get_monthly_macro": MagicMock(return_value=[]),
+    }
+
+
+@contextlib.contextmanager
+def _patch_layers(mocks: Mapping[str, Any], assemble: Any) -> Iterator[None]:
+    """Запатчить ВСЕ §9.x-символы + assemble_report (по `mocks`-словарю). Компактно.
+
+    Патчим там, где символы ИМПОРТИРОВАНЫ в orchestrator (`_MOD`), не там, где определены.
+    """
+    with contextlib.ExitStack() as stack:
+        for name, mock in mocks.items():
+            stack.enter_context(patch(f"{_MOD}.{name}", mock))
+        stack.enter_context(patch(f"{_MOD}.assemble_report", side_effect=assemble))
+        yield
+
+
+class TestWiringAndSignatures:
+    """Патчим §9.x на sentinel'ы + перехватываем assemble_report → ассертим слоты+kwargs."""
+
+    def _run(
+        self, *, obj_class: str | None = None, competitors: list[dict[str, Any]] | None = None
+    ) -> tuple[SiteFinderReport, dict[str, Any], dict[str, Any]]:
+        mocks = _patched_layers()
+        captured: dict[str, Any] = {}
+
+        def _fake_assemble(analyze: dict[str, Any], **kwargs: Any) -> SiteFinderReport:
+            captured["analyze"] = analyze
+            captured.update(kwargs)
+            return SiteFinderReport()
+
+        with _patch_layers(mocks, _fake_assemble):
+            report = build_site_finder_report(
+                MagicMock(name="db"),
+                analyze=_analyze(competitors),
+                cad_num="66:41:0702048:27",
+                district="Ленинский",
+                obj_class=obj_class,
+                horizons=[6, 12, 18],
+            )
+        return report, captured, mocks
+
+    def test_each_layer_lands_in_correct_slot(self) -> None:
+        _report, captured, _mocks = self._run(obj_class="Комфорт")
+        # assemble_report вызван (captured заполнен) — каждый sentinel в СВОЁМ слоте.
+        assert captured["market_metrics"] == "MM"
+        assert captured["future_supply"] == "FSP"
+        assert captured["forecasts"] == ["F6", "F12"]
+        assert captured["scenarios"] == ["SC"]
+        assert captured["product_scores"] == "PSC"
+        assert captured["special_indices"] == "SI"
+        # overlay идёт именно в recommendation_overlay (НЕ product_tz/product_scores).
+        assert captured["recommendation_overlay"] == "OVL"
+        # supply_layers свёрнут в сводку-dict (не сырой список строк).
+        assert isinstance(captured["supply_layers"], dict)
+        assert captured["supply_layers"]["open_units"] == 10
+        # meta-проводка.
+        assert captured["cad_num"] == "66:41:0702048:27"
+        assert captured["district"] == "Ленинский"
+        assert captured["horizons"] == [6, 12, 18]
+        assert captured["advisory"] is True
+        assert captured["segment"] == {
+            "obj_class": "Комфорт",
+            "room_bucket": None,
+            "district": "Ленинский",
+            "price_bucket": None,
+        }
+
+    def test_horizons_services_get_horizons_kwarg(self) -> None:
+        """forecasts/scenarios/special_indices ДОЛЖНЫ получить `horizons=` (Sequence)."""
+        _report, _captured, mocks = self._run(obj_class="Комфорт")
+        horizons_services = (
+            "compute_demand_supply_forecast",
+            "compute_scenarios",
+            "compute_special_indices",
+        )
+        for name in horizons_services:
+            _args, kwargs = mocks[name].call_args
+            assert kwargs.get("horizons") == [6, 12, 18], f"{name} must take horizons="
+            assert "horizon_months" not in kwargs, f"{name} must NOT take horizon_months"
+            assert kwargs.get("cad_num") == "66:41:0702048:27"
+            assert kwargs.get("district") == "Ленинский"
+            # spec прокинут как SegmentSpec.
+            assert kwargs["spec"].obj_class == "Комфорт"
+
+    def test_horizon_months_services_get_int_kwarg(self) -> None:
+        """score_card/future_supply ДОЛЖНЫ получить `horizon_months=` (int=primary=12)."""
+        _report, _captured, mocks = self._run(obj_class="Комфорт")
+        _args, sc_kwargs = mocks["compute_score_card"].call_args
+        assert sc_kwargs.get("horizon_months") == 12  # 12 в сетке [6,12,18] → primary
+        assert "horizons" not in sc_kwargs
+        assert sc_kwargs["spec"].obj_class == "Комфорт"
+        _args, fsp_kwargs = mocks["compute_future_supply_pressure"].call_args
+        assert fsp_kwargs.get("horizon_months") == 12
+        assert "horizons" not in fsp_kwargs
+
+    def test_overlay_gets_target_class_not_spec(self) -> None:
+        """overlay ДОЛЖЕН получить `target_class=` (str) + `horizon_months=`, НЕ spec."""
+        _report, _captured, mocks = self._run(obj_class="Бизнес")
+        _args, kwargs = mocks["build_forecast_overlay"].call_args
+        assert kwargs.get("target_class") == "Бизнес"
+        assert kwargs.get("horizon_months") == 12
+        assert "spec" not in kwargs
+        assert "horizons" not in kwargs
+
+    def test_market_metrics_and_supply_use_premise_kind(self) -> None:
+        """§9.2/§9.3 ходят по premise_kind, без spec/horizons."""
+        _report, _captured, mocks = self._run(obj_class="Комфорт")
+        _args, mm_kwargs = mocks["compute_market_metrics"].call_args
+        assert mm_kwargs.get("premise_kind") == "квартира"
+        assert mm_kwargs.get("district") == "Ленинский"
+        _args, al_kwargs = mocks["compute_all_layers"].call_args
+        assert al_kwargs.get("premise_kind") == "квартира"
+
+    def test_modal_class_default_flows_into_spec_and_overlay(self) -> None:
+        """obj_class=None + конкуренты [Комфорт×2, Бизнес] → модальный 'Комфорт' всюду."""
+        comps = [{"obj_class": "Комфорт"}, {"obj_class": "Комфорт"}, {"obj_class": "Бизнес"}]
+        _report, captured, mocks = self._run(obj_class=None, competitors=comps)
+        assert captured["segment"]["obj_class"] == "Комфорт"
+        _args, ovl_kwargs = mocks["build_forecast_overlay"].call_args
+        assert ovl_kwargs.get("target_class") == "Комфорт"
+        _args, f_kwargs = mocks["compute_demand_supply_forecast"].call_args
+        assert f_kwargs["spec"].obj_class == "Комфорт"
+
+    def test_empty_competitors_default_class_none(self) -> None:
+        """obj_class=None + пустые конкуренты → None (ЕКБ-wide all-class)."""
+        _report, captured, mocks = self._run(obj_class=None, competitors=[])
+        assert captured["segment"]["obj_class"] is None
+        _args, ovl_kwargs = mocks["build_forecast_overlay"].call_args
+        assert ovl_kwargs.get("target_class") is None
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# GRACEFUL: один §9.x-слой бросает → отчёт всё равно собирается, слот None.
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class TestGracefulDegradation:
+    def test_one_layer_raises_report_still_assembles(self) -> None:
+        mocks = _patched_layers()
+        # forecasts-слой бросает — отчёт ДОЛЖЕН собраться с forecasts=None.
+        mocks["compute_demand_supply_forecast"] = MagicMock(side_effect=RuntimeError("boom"))
+        captured: dict[str, Any] = {}
+
+        def _fake_assemble(analyze: dict[str, Any], **kwargs: Any) -> SiteFinderReport:
+            captured.update(kwargs)
+            return SiteFinderReport()
+
+        with _patch_layers(mocks, _fake_assemble):
+            # НЕ должно бросить, несмотря на падение слоя.
+            report = build_site_finder_report(
+                MagicMock(name="db"),
+                analyze=_analyze([{"obj_class": "Комфорт"}]),
+                cad_num="66:41:0702048:27",
+                district="Ленинский",
+                horizons=[6, 12, 18],
+            )
+        assert isinstance(report, SiteFinderReport)
+        # упавший слой → None в слоте; остальные сохранены.
+        assert captured["forecasts"] is None
+        assert captured["market_metrics"] == "MM"
+        assert captured["scenarios"] == ["SC"]
+
+    def test_all_layers_raise_report_still_assembles(self) -> None:
+        """Даже если ВСЕ §9.x-слои бросают — отчёт собирается (все слоты None)."""
+        boom = MagicMock(side_effect=RuntimeError("boom"))
+        # Каждый §9.x-символ → бросающий mock (имена зеркалят _patched_layers).
+        mocks = {name: boom for name in _patched_layers()}
+        captured: dict[str, Any] = {}
+
+        def _fake_assemble(analyze: dict[str, Any], **kwargs: Any) -> SiteFinderReport:
+            captured.update(kwargs)
+            return SiteFinderReport()
+
+        with _patch_layers(mocks, _fake_assemble):
+            report = build_site_finder_report(
+                MagicMock(name="db"),
+                analyze=_analyze([{"obj_class": "Комфорт"}]),
+                cad_num="66:41:0702048:27",
+                district="Ленинский",
+            )
+        assert isinstance(report, SiteFinderReport)
+        for slot in (
+            "market_metrics",
+            "supply_layers",
+            "future_supply",
+            "forecasts",
+            "scenarios",
+            "product_scores",
+            "special_indices",
+            "recommendation_overlay",
+        ):
+            assert captured[slot] is None, f"{slot} should degrade to None"
+        # сегмент всё равно проброшен (он PURE, не от §9.x).
+        assert captured["segment"]["obj_class"] == "Комфорт"

backend/tests/services/site_finder/test_future_supply.py

@@ -5,17 +5,22 @@
     _months_between, _months_of_pressure (guard div-by-zero → None), _saturating_index
     (монотонность + clamp [0,1] + named-saturation), _monthly_absorption_units (оба
     пути из market_metrics), _min_confidence (худшая тянет вниз, whitelisted).
-  • compute_future_supply_pressure через MagicMock-сессию + mock compute_market_metrics:
+  • compute_future_supply_pressure через MagicMock-сессию + mock compute_market_metrics
-    форма SQL (читает VIEW v_supply_layers_latest, НЕ базовую таблицу / recompute;
+    + mock resolve_objective_districts: форма SQL (читает VIEW v_supply_layers_latest,
-    CAST(:x AS type) не :x::type), агрегация по слоям, горизонт-взвешивание L3,
+    НЕ базовую таблицу / recompute; district-фильтр через ANY(CAST(:names AS text[])),
+    НЕ скаляр :district::type), resolver-wiring (админ → микро + сам админ, дедуп;
+    None → EKB-wide без фильтра), агрегация по слоям, горизонт-взвешивание L3,
     graceful empty/thin → index None + confidence 'low', confidence=MIN.
 
-Детерминированно, без LLM. Мокаем compute_market_metrics + db (нет живой БД).
+Детерминированно, без LLM. Мокаем compute_market_metrics + resolve_objective_districts
++ db (нет живой БД).
 """
 
 from __future__ import annotations
 
 import os
+from collections.abc import Iterator
+from contextlib import contextmanager
 from datetime import date
 from unittest.mock import MagicMock, patch
 
@@ -40,6 +45,21 @@ from app.services.site_finder.future_supply import (
 # Путь патча reused-сервиса (импортирован в модуль future_supply).
 _MARKET = "app.services.site_finder.future_supply.compute_market_metrics"
 
+# Путь патча district-резолвера (импортирован в модуль future_supply). Резолвер бьёт
+# живую БД (ekb_districts / ekb_district_alias) — в unit-тестах всегда мокаем.
+_RESOLVE = "app.services.site_finder.future_supply.resolve_objective_districts"
+
+
+def _resolve_passthrough(_db: object, district: str | None) -> list[str] | None:
+    """Дефолтная заглушка резолвера: raw-микро pass-through (зеркало legacy-ветки).
+
+    district — raw-микро («X»/«Автовокзал» в большинстве тестов) → `[district]` (фильтр
+    по самому имени, как было до фикса); None → None (EKB-wide). Тесты, проверяющие
+    admin→micro разворот, передают свой side_effect через `_patched(resolve=…)`.
+    """
+    return [district] if district is not None else None
+
+
 # Разрешённый vocab confidence (зеркало CHECK м.125 / market_metrics).
 _ALLOWED_CONFIDENCE = {"high", "medium", "low"}
 
@@ -313,6 +333,25 @@ def _metrics_stub(
     return m
 
 
+@contextmanager
+def _patched(
+    *, metrics: MagicMock | None = None, resolve: object = _resolve_passthrough
+) -> Iterator[tuple[MagicMock, MagicMock]]:
+    """Патчит compute_market_metrics + resolve_objective_districts разом.
+
+    `resolve`: либо callable (side_effect — резолвит вход в набор микро), либо значение
+    (return_value). По умолчанию pass-through `[district]` — legacy фильтр-по-микро, под
+    который написано большинство compute-тестов (district там — raw-микро «X»/«Автовокзал»).
+    """
+    m = metrics if metrics is not None else _metrics_stub()
+    with patch(_MARKET, return_value=m) as mm, patch(_RESOLVE) as rs:
+        if callable(resolve):
+            rs.side_effect = resolve
+        else:
+            rs.return_value = resolve
+        yield mm, rs
+
+
 # ── compute_future_supply_pressure: SQL форма (читает VIEW) ────────────────────
 
 
@@ -322,7 +361,7 @@ class TestComputeReadsView:
         # включает невыводимые L3 graddoc_stub(#956)/insider_manual. НЕ базовую
         # таблицу supply_layers напрямую и НЕ пересчёт compute_all_layers.
         db = _mock_db([])
-        with patch(_MARKET, return_value=_metrics_stub()):
+        with _patched():
             compute_future_supply_pressure(db, district="Автовокзал")
         sql = _executed_sql(db).lower()
         assert "v_supply_layers_latest" in sql
@@ -331,25 +370,93 @@ class TestComputeReadsView:
         assert "objective_lots" not in sql
         assert "domrf_kn_objects" not in sql
 
-    def test_sql_uses_cast_not_double_colon(self) -> None:
+    def test_sql_uses_array_bind_not_double_colon(self) -> None:
+        # psycopg v3: district-фильтр через ANY(CAST(:names AS text[])) — массивный bind,
+        # НЕ :x::type (SQLAlchemy его молча роняет → syntax error).
         db = _mock_db([])
-        with patch(_MARKET, return_value=_metrics_stub()):
+        with _patched():
             compute_future_supply_pressure(db, district="Автовокзал")
         sql = _executed_sql(db)
-        assert "CAST(:district AS text)" in sql
+        assert "ANY(CAST(:names AS text[]))" in sql
+        assert "CAST(:has_district AS boolean)" in sql
         _assert_no_double_colon_cast(sql)
 
-    def test_params_pass_district(self) -> None:
+    def test_district_not_a_bare_scalar_filter(self) -> None:
+        # Regression guard: МИКС-вокабулярный view нельзя фильтровать скаляром по одному
+        # имени (промахивается мимо L1 микро ИЛИ L2/3 админ строк → ложный дефицит, баг
+        # парцеля 66:41:0702048:27 / Кировский). Фильтр обязан идти по НАБОРУ имён.
         db = _mock_db([])
-        with patch(_MARKET, return_value=_metrics_stub()) as mm:
+        with _patched():
+            compute_future_supply_pressure(db, district="Автовокзал")
+        sql = _executed_sql(db)
+        # ни один из старых скалярных вариантов фильтра по district не должен вернуться.
+        assert "district_name = CAST(:district AS text)" not in sql
+        assert "district_name = :district" not in sql
+
+    def test_params_pass_district_to_market_metrics(self) -> None:
+        # district (админ-имя) пробрасывается как есть в reused market_metrics — там свой
+        # резолвер развернёт его в микро (единый район для absorption).
+        db = _mock_db([])
+        with _patched() as (mm, _rs):
             compute_future_supply_pressure(
                 db, district="Академический", horizon_months=24, premise_kind="квартира"
             )
-        assert _executed_params(db)["district"] == "Академический"
-        # district пробрасывается и в reused market_metrics (единый район).
         assert mm.call_args.kwargs["district"] == "Академический"
 
 
+# ── compute_future_supply_pressure: resolver-wiring (admin→micro+admin) ────────
+
+
+class TestComputeResolverWiring:
+    def test_admin_resolves_to_micros_plus_admin(self) -> None:
+        # Баг-репро: «Кировский» (АДМИН) → резолвер даёт микро [Втузгородок, ЖБИ]. Фильтр
+        # :names обязан быть микро + сам админ (для L2/3 admin-named строк) и дедупнут.
+        db = _mock_db([])
+        with _patched(resolve=lambda _db, _d: ["Втузгородок", "ЖБИ"]) as (_mm, rs):
+            compute_future_supply_pressure(db, district="Кировский")
+        params = _executed_params(db)
+        assert params["names"] == ["Втузгородок", "ЖБИ", "Кировский"]
+        assert params["has_district"] is True
+        # резолвер вызван именно с входным district.
+        assert rs.call_args.args[1] == "Кировский"
+
+    def test_admin_micros_already_containing_admin_deduped(self) -> None:
+        # Если резолвер вернул набор, уже содержащий сам админ, — не дублируем имя.
+        db = _mock_db([])
+        with _patched(resolve=lambda _db, _d: ["Кировский", "Втузгородок"]):
+            compute_future_supply_pressure(db, district="Кировский")
+        assert _executed_params(db)["names"] == ["Кировский", "Втузгородок"]
+
+    def test_raw_micro_input_filters_by_itself(self) -> None:
+        # raw-микро вход (legacy внутренний вызов): резолвер вернул [micro] → names=[micro]
+        # (сам админ == вход → дедуп оставляет одно имя). Legacy фильтр-по-микро сохранён.
+        db = _mock_db([])
+        with _patched(resolve=lambda _db, _d: ["Втузгородок"]):
+            compute_future_supply_pressure(db, district="Втузгородок")
+        assert _executed_params(db)["names"] == ["Втузгородок"]
+        assert _executed_params(db)["has_district"] is True
+
+    def test_none_district_no_filter(self) -> None:
+        # district None → резолвер None → has_district False, names=[] (EKB-wide, без
+        # district-фильтра). Зеркало market_metrics EKB-wide ветки.
+        db = _mock_db([])
+        with _patched(resolve=lambda _db, _d: None):
+            compute_future_supply_pressure(db, district=None)
+        params = _executed_params(db)
+        assert params["has_district"] is False
+        assert params["names"] == []
+
+    def test_admin_without_aliases_ekb_wide(self) -> None:
+        # Админ-имя, но резолвер вернул None (дыра в словаре) → EKB-wide fallback (НЕ
+        # пустой фильтр-«0 строк гарантированно»). has_district False.
+        db = _mock_db([])
+        with _patched(resolve=lambda _db, _d: None):
+            compute_future_supply_pressure(db, district="Кировский")
+        params = _executed_params(db)
+        assert params["has_district"] is False
+        assert params["names"] == []
+
+
 # ── compute_future_supply_pressure: агрегация + композит ───────────────────────
 
 
@@ -381,7 +488,7 @@ class TestComputeAggregation:
         ]
         db = _mock_db(rows)
         # absorption: available 120, mos 6 → 20 ед/мес.
-        with patch(_MARKET, return_value=_metrics_stub(months_of_supply=6.0, n_available=120)):
+        with _patched(metrics=_metrics_stub(months_of_supply=6.0, n_available=120)):
             res = compute_future_supply_pressure(db, district="X", horizon_months=12)
         assert res.open_units == 300
         assert res.hidden_units == 200
@@ -408,7 +515,7 @@ class TestComputeAggregation:
             {"layer": 2, "units_estimate": 100, "expected_online_date": None, "confidence": "high"},
         ]
         db = _mock_db(rows)
-        with patch(_MARKET, return_value=_metrics_stub(months_of_supply=5.0, n_available=100)):
+        with _patched(metrics=_metrics_stub(months_of_supply=5.0, n_available=100)):
             res = compute_future_supply_pressure(db, district="X")
         # absorption 100/5=20; давление только от hidden 100 → 100/20=5 мес.
         assert res.monthly_absorption_units == pytest.approx(20.0)
@@ -421,7 +528,7 @@ class TestComputeAggregation:
             {"layer": 2, "units_estimate": 50, "expected_online_date": None, "confidence": "high"},
         ]
         db = _mock_db(rows)
-        with patch(_MARKET, return_value=_metrics_stub(confidence="low")):
+        with _patched(metrics=_metrics_stub(confidence="low")):
             res = compute_future_supply_pressure(db, district="X")
         assert res.confidence == "low"
 
@@ -430,7 +537,7 @@ class TestComputeAggregation:
             {"layer": 2, "units_estimate": 50, "expected_online_date": None, "confidence": "high"},
         ]
         db = _mock_db(rows)
-        with patch(_MARKET, return_value=_metrics_stub(confidence="high")):
+        with _patched(metrics=_metrics_stub(confidence="high")):
             res = compute_future_supply_pressure(db, district="X")
         assert res.confidence == "high"
         assert res.confidence in _ALLOWED_CONFIDENCE
@@ -444,11 +551,10 @@ class TestComputeGraceful:
         # Пустой склад (worker ещё не наполнил м.125 на prod) + нет поглощения →
         # index None, months_of_pressure None, confidence low. НЕ crash.
         db = _mock_db([])
-        with patch(
+        with _patched(
-            _MARKET,
+            metrics=_metrics_stub(
-            return_value=_metrics_stub(
                 absorption_rate=None, months_of_supply=None, n_available=0, confidence="low"
-            ),
+            )
         ):
             res = compute_future_supply_pressure(db, district="Пусто")
         assert res.index is None
@@ -463,7 +569,7 @@ class TestComputeGraceful:
         # Склад пуст, НО рынок измерим → очереди нет → давление 0.0, index 0.0
         # (валидное «нет будущего давления», НЕ None).
         db = _mock_db([])
-        with patch(_MARKET, return_value=_metrics_stub(months_of_supply=6.0, n_available=120)):
+        with _patched(metrics=_metrics_stub(months_of_supply=6.0, n_available=120)):
             res = compute_future_supply_pressure(db, district="X")
         assert res.months_of_pressure == 0.0
         assert res.index == 0.0
@@ -480,11 +586,10 @@ class TestComputeGraceful:
             },
         ]
         db = _mock_db(rows)
-        with patch(
+        with _patched(
-            _MARKET,
+            metrics=_metrics_stub(
-            return_value=_metrics_stub(
                 absorption_rate=None, months_of_supply=None, n_available=0, confidence="medium"
-            ),
+            )
         ):
             res = compute_future_supply_pressure(db, district="X")
         assert res.hidden_units == 500
@@ -496,9 +601,11 @@ class TestComputeGraceful:
 
     def test_view_query_exception_graceful(self) -> None:
         # Сбой чтения view → [] (graceful), index по absorption (нет очереди → 0.0).
+        # Резолвер мокаем отдельно (иначе RuntimeError-side_effect db.execute пробил бы
+        # и резолвер) — изолируем именно сбой view-запроса.
         db = MagicMock()
         db.execute.side_effect = RuntimeError("view gone")
-        with patch(_MARKET, return_value=_metrics_stub(months_of_supply=6.0, n_available=120)):
+        with _patched(metrics=_metrics_stub(months_of_supply=6.0, n_available=120)):
             res = compute_future_supply_pressure(db, district="Сбой")
         assert res.hidden_units == 0
         assert res.future_units_by_horizon == 0.0
@@ -517,14 +624,14 @@ class TestComputeGraceful:
             },
         ]
         db = _mock_db(rows)
-        with patch(_MARKET, return_value=_metrics_stub(months_of_supply=6.0, n_available=120)):
+        with _patched(metrics=_metrics_stub(months_of_supply=6.0, n_available=120)):
             res = compute_future_supply_pressure(db, district="X")
         assert res.future_units_by_horizon == 0.0
         assert res.months_of_pressure == 0.0
 
     def test_returns_dataclass_always(self) -> None:
         db = _mock_db([])
-        with patch(_MARKET, return_value=_metrics_stub()):
+        with _patched(resolve=lambda _db, _d: None):
             res = compute_future_supply_pressure(db, district=None)
         assert isinstance(res, FutureSupplyPressure)
         assert res.district is None