feat(geo): ЖК → district mapping через PostGIS-полигоны OSM (Limit_District_Geometry closed)

- 56_fetch_ekb_districts.py: Overpass query (admin_level=9 в bbox ЕКБ),
  shapely-ассемблинг рваных OSM-веев в MultiPolygon, кэш в data/raw/.
- 56_schema_ekb_districts_geom.sql: CREATE TABLE с GIST.
- 57_kn_objects_district.sql: ALTER + backfill UPDATE через ST_Contains.
  3032 ЖК Свердл проставили district_name (208 в Академическом, 158 в Чкаловском,
  и т.д.), 454 за bbox ЕКБ остались NULL.
- domrf_kn.py: _backfill_district_names() после Phase A — каждый sweep
  автоматически проставляет district_name свежезаписанным ЖК.
- analytics_queries.py: 4 места addr ILIKE → district_name = :dn
  (_velocity_baseline, _elasticity_coef, _active_competitors_count, comparable).
- analytics_queries.py: graceful obj_class-degrade — если в районе все
  obj_class=NULL (kn-API quirk), фильтр по target_class опускается на уровне
  geo-запросов с warning. class_multiplier на ценах продолжает работать.

Smoke (Академический, Comfort, 50K м²): velocity_source=sale_graph (104 ЖК /
1872 точек), competitors=208 в районе, headline «3400 млн ₽ · 112 мес ·
ликвидность 21/100», 5 comparable ЖК района — все реалистичные.

backend/app/services/analytics_queries.py

@@ -1056,7 +1056,7 @@ def _velocity_baseline(
                     SELECT o.obj_id
                       FROM domrf_kn_objects o
                      WHERE o.region_cd = :rc
-                       AND o.addr ILIKE '%' || :dn || '%'
+                       AND o.district_name = :dn
                        {where_class}
                 ),
                 sg AS (
@@ -1122,10 +1122,10 @@ def _active_competitors_count(
         ).scalar()
         return int(n or 0)
 
-    # Tier 1: район + класс
+    # Tier 1: район + класс (через PostGIS-полигоны district_name)
     if target_class:
         n = _q(
-            "AND addr ILIKE '%' || :dn || '%' AND obj_class = :cls",
+            "AND district_name = :dn AND obj_class = :cls",
             {"rc": region_code, "dn": district_name, "cls": target_class},
         )
         if n >= 2:
@@ -1133,7 +1133,7 @@ def _active_competitors_count(
 
     # Tier 2: район (без класса — могут быть ЖК где obj_class NULL)
     n = _q(
-        "AND addr ILIKE '%' || :dn || '%'",
+        "AND district_name = :dn",
         {"rc": region_code, "dn": district_name},
     )
     if n >= 2:
@@ -1169,7 +1169,7 @@ def _elasticity_coef(
                     SELECT o.obj_id
                       FROM domrf_kn_objects o
                      WHERE o.region_cd = :rc
-                       AND o.addr ILIKE '%' || :dn || '%'
+                       AND o.district_name = :dn
                        {where_class}
                 ),
                 pts AS (
@@ -1392,11 +1392,36 @@ def recommend_mix(
 
     # 5b) Velocity baseline (apartments/month per ЖК) + price elasticity.
     # Both are required for the live "цена↔темп" calculator on the frontend.
+    # Graceful: kn-API returns obj_class=NULL для всех ЖК Свердл (отдельный
+    # баг скрейпера). Если в районе нет ни одного НЕ-NULL obj_class —
+    # игнорируем target_class фильтр на уровне velocity/elasticity/comparable
+    # запросов, иначе obj_pool пустой и всё падает в fallback.
+    has_class_data = bool(
+        db.execute(
+            text(
+                """
+                SELECT 1 FROM domrf_kn_objects
+                 WHERE region_cd = :rc
+                   AND district_name = :dn
+                   AND obj_class IS NOT NULL
+                 LIMIT 1
+                """
+            ),
+            {"rc": region_code, "dn": district_row["district_name"]},
+        ).scalar()
+    )
+    target_class_for_geo = target_class if has_class_data else None
+    if target_class and not has_class_data:
+        warnings.append(
+            f"obj_class не заполнен для ЖК района {district_row['district_name']}"
+            f" — фильтр по классу '{target_class}' игнорируется в velocity/comparable"
+            " (но class_multiplier из yandex_realty_zk применяется к ценам)."
+        )
     vel = _velocity_baseline(
         db,
         region_code=region_code,
         district_name=district_row["district_name"],
-        target_class=target_class,
+        target_class=target_class_for_geo,
     )
     market_vel_pm = vel["realised_per_month_median"] or vel["realised_per_month_avg"]
     if market_vel_pm is None:
@@ -1415,7 +1440,7 @@ def recommend_mix(
         db,
         region_code=region_code,
         district_name=district_row["district_name"],
-        target_class=target_class,
+        target_class=target_class_for_geo,
     )
     elasticity = elast["elasticity"]
     if elast["source"] == "fallback":
@@ -1431,7 +1456,7 @@ def recommend_mix(
         db,
         region_code=region_code,
         district_name=district_row["district_name"],
-        target_class=target_class,
+        target_class=target_class_for_geo,
     )
     if competitors_scope == "fallback_singleton":
         warnings.append(
@@ -1551,13 +1576,17 @@ def recommend_mix(
                         AND a.snapshot_date = la.snap
                         AND a.type = 'apartments'
                  WHERE o.region_cd = :rc
-                   AND o.addr ILIKE '%' || :dn || '%'
+                   AND o.district_name = :dn
                    AND (CAST(:cls AS TEXT) IS NULL OR o.obj_class = :cls)
                  ORDER BY o.flat_count DESC NULLS LAST
                  LIMIT 5
                 """
             ),
-            {"rc": region_code, "dn": district_row["district_name"], "cls": target_class},
+            {
+                "rc": region_code,
+                "dn": district_row["district_name"],
+                "cls": target_class_for_geo,
+            },
         )
         .mappings()
         .all()

backend/app/services/scrapers/domrf_kn.py

@@ -338,6 +338,44 @@ def _ensure_snapshot(db: Session, snapshot_date: date) -> None:
     db.commit()
 
 
+def _backfill_district_names(db: Session, snapshot_date: date, region_code: int) -> None:
+    """Spatial-join свежезаписанных ЖК с ekb_districts_geom через PostGIS.
+    Best-effort: если ekb_districts_geom пустая (загрузчик не запущен) —
+    UPDATE затронет 0 строк, ничего не ломается.
+
+    Только для региона 66 (ЕКБ) — для других регионов полигонов нет.
+    """
+    if region_code != 66:
+        return
+    try:
+        db.execute(
+            text(
+                """
+                UPDATE domrf_kn_objects o
+                   SET district_name = d.district_name
+                  FROM ekb_districts_geom d
+                 WHERE o.snapshot_date = :snap
+                   AND o.region_cd = :rc
+                   AND o.latitude IS NOT NULL
+                   AND o.longitude IS NOT NULL
+                   AND (o.district_name IS NULL OR o.district_name <> d.district_name)
+                   AND ST_Contains(
+                       d.geom,
+                       ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)
+                   )
+                """
+            ),
+            {"snap": snapshot_date, "rc": region_code},
+        )
+        db.commit()
+    except Exception as e:
+        logger.warning("backfill district_name failed: %s", e)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+
+
 def _insert_raw(db: Session, snapshot_date: date, endpoint_label: str, payload: Any) -> None:
     body = json.dumps(payload, ensure_ascii=False)
     db.execute(
@@ -1224,6 +1262,10 @@ async def run_region_sweep(
                     },
                 )
                 db.commit()
+                # Spatial-join: проставить district_name свежезаписанным ЖК
+                # через ekb_districts_geom (PostGIS-полигоны OSM). Backfill
+                # лёгкий — только новые/изменённые row на этом snapshot_date.
+                _backfill_district_names(db, snapshot_date, region_code)
                 log_progress(
                     db,
                     run_id,

data/sql/56_fetch_ekb_districts.py

@@ -0,0 +1,242 @@
+"""Fetch ЕКБ district polygons from OSM Overpass and load into PostGIS.
+
+Usage:
+    cd backend && uv run python ../data/sql/56_fetch_ekb_districts.py [--force-refetch]
+
+Without --force-refetch использует кэш `data/raw/ekb_districts.geojson` если он
+существует. Иначе дёргает Overpass API один раз и сохраняет.
+
+После Overpass fetch:
+1. Переводит OSM relation members в GeoJSON MultiPolygon (через shapely).
+2. UPSERT в `ekb_districts_geom (district_name, geom, osm_id, fetched_at)`.
+
+OSM-to-our naming:
+    "Верх-Исетский район"     → "Верх-Исетский"
+    "Академический район"     → "Академический"
+    ...
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import sys
+import urllib.parse
+import urllib.request
+from pathlib import Path
+
+# Add backend to path so we can import app.* in standalone CLI mode.
+ROOT = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(ROOT / "backend"))
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+log = logging.getLogger(__name__)
+
+# Cache lives at repo root, committed to git.
+GEOJSON_CACHE = ROOT / "data" / "raw" / "ekb_districts.geojson"
+
+# OSM admin_level=9 in RU = city districts. ЕКБ has 8 of them.
+# Bbox covers all of Ekaterinburg city + adjacent.
+OVERPASS_QUERY = """
+[out:json][timeout:90];
+(
+  relation['boundary'='administrative']['admin_level'='9']['name'~'район']
+    (56.71,60.42,57.00,60.95);
+);
+out geom;
+"""
+
+OVERPASS_URL = "https://overpass-api.de/api/interpreter"
+
+# OSM display name → ekb_districts.district_name (exact match required).
+DISTRICT_NAME_MAP = {
+    "Академический район": "Академический",
+    "Верх-Исетский район": "Верх-Исетский",
+    "Железнодорожный район": "Железнодорожный",
+    "Кировский район": "Кировский",
+    "Ленинский район": "Ленинский",
+    "Октябрьский район": "Октябрьский",
+    "Орджоникидзевский район": "Орджоникидзевский",
+    "Чкаловский район": "Чкаловский",
+}
+
+
+def fetch_overpass() -> dict:
+    """Hit Overpass API; return raw JSON response."""
+    log.info("Querying Overpass...")
+    req = urllib.request.Request(
+        OVERPASS_URL,
+        data=urllib.parse.urlencode({"data": OVERPASS_QUERY}).encode(),
+        headers={"User-Agent": "gendesign-recommend/1.0 (district mapping)"},
+    )
+    with urllib.request.urlopen(req, timeout=120) as resp:
+        return json.loads(resp.read().decode())
+
+
+def osm_relation_to_geojson(relation: dict) -> dict | None:
+    """Convert one Overpass relation (way-members + tags) to a GeoJSON
+    MultiPolygon Feature. Outer + inner rings are assembled into polygons.
+
+    OSM relations of type=multipolygon have 'members' which are 'way'-objects
+    each carrying their own 'geometry' (list of {lat, lon}). We build outer
+    rings, inner rings (holes), and assemble Polygon parts → MultiPolygon.
+    """
+    from shapely.geometry import MultiPolygon, Polygon, mapping
+    from shapely.ops import polygonize, unary_union
+
+    members = relation.get("members") or []
+    outer_lines: list[list[tuple[float, float]]] = []
+    inner_lines: list[list[tuple[float, float]]] = []
+    for m in members:
+        if m.get("type") != "way":
+            continue
+        coords = [(pt["lon"], pt["lat"]) for pt in (m.get("geometry") or [])]
+        if len(coords) < 2:
+            continue
+        role = m.get("role") or "outer"
+        (inner_lines if role == "inner" else outer_lines).append(coords)
+
+    if not outer_lines:
+        log.warning("Relation %s has no outer members", relation.get("id"))
+        return None
+
+    # Use shapely.polygonize to stitch open way-segments into closed rings.
+    # This handles real-world OSM cases where a single ring is split across
+    # multiple way-members at admin-boundary intersections.
+    from shapely.geometry import LineString
+
+    outer_polys = list(polygonize([LineString(seg) for seg in outer_lines]))
+    inner_polys = list(polygonize([LineString(seg) for seg in inner_lines]))
+
+    if not outer_polys:
+        log.warning(
+            "Relation %s outer members did not form closed rings", relation.get("id")
+        )
+        return None
+
+    # Subtract holes from outers (best-effort).
+    if inner_polys:
+        holes_union = unary_union(inner_polys)
+        outer_polys = [p.difference(holes_union) for p in outer_polys]
+
+    multipoly = MultiPolygon(
+        [p for poly in outer_polys for p in (poly.geoms if poly.geom_type == "MultiPolygon" else [poly])]
+        if any(p.geom_type == "MultiPolygon" for p in outer_polys)
+        else outer_polys
+    )
+
+    # Ensure all geometry is valid; buffer(0) is the standard fix for
+    # self-intersections that OSM data sometimes carries.
+    if not multipoly.is_valid:
+        multipoly = multipoly.buffer(0)
+        if multipoly.geom_type == "Polygon":
+            multipoly = MultiPolygon([multipoly])
+
+    return {
+        "type": "Feature",
+        "properties": {
+            "name": relation.get("tags", {}).get("name"),
+            "osm_id": relation.get("id"),
+            "admin_level": relation.get("tags", {}).get("admin_level"),
+        },
+        "geometry": mapping(multipoly),
+    }
+
+
+def build_geojson(overpass_data: dict) -> dict:
+    """Build a FeatureCollection from raw Overpass response."""
+    features: list[dict] = []
+    for el in overpass_data.get("elements", []):
+        if el.get("type") != "relation":
+            continue
+        feature = osm_relation_to_geojson(el)
+        if feature:
+            features.append(feature)
+    return {"type": "FeatureCollection", "features": features}
+
+
+def load_or_fetch(force: bool = False) -> dict:
+    if GEOJSON_CACHE.exists() and not force:
+        log.info("Using cached GeoJSON: %s", GEOJSON_CACHE)
+        return json.loads(GEOJSON_CACHE.read_text(encoding="utf-8"))
+    raw = fetch_overpass()
+    geojson = build_geojson(raw)
+    GEOJSON_CACHE.parent.mkdir(parents=True, exist_ok=True)
+    GEOJSON_CACHE.write_text(
+        json.dumps(geojson, ensure_ascii=False, indent=2), encoding="utf-8"
+    )
+    log.info("Cached %d features → %s", len(geojson["features"]), GEOJSON_CACHE)
+    return geojson
+
+
+def upsert_into_db(geojson: dict) -> int:
+    from sqlalchemy import text
+
+    from app.core.db import SessionLocal
+
+    db = SessionLocal()
+    try:
+        loaded = 0
+        for feat in geojson.get("features", []):
+            osm_name = feat["properties"].get("name")
+            our_name = DISTRICT_NAME_MAP.get(osm_name)
+            if not our_name:
+                log.warning("Skipping unmapped OSM district: %s", osm_name)
+                continue
+            geom_text = json.dumps(feat["geometry"])
+            db.execute(
+                text(
+                    """
+                    INSERT INTO ekb_districts_geom (district_name, geom, osm_id, fetched_at)
+                    VALUES (
+                        :dn,
+                        ST_Multi(ST_GeomFromGeoJSON(:gj))::geometry(MultiPolygon, 4326),
+                        :osm,
+                        NOW()
+                    )
+                    ON CONFLICT (district_name) DO UPDATE SET
+                        geom = EXCLUDED.geom,
+                        osm_id = EXCLUDED.osm_id,
+                        fetched_at = NOW()
+                    """
+                ),
+                {
+                    "dn": our_name,
+                    "gj": geom_text,
+                    "osm": feat["properties"].get("osm_id"),
+                },
+            )
+            loaded += 1
+            log.info("  ✓ %s (osm_id=%s)", our_name, feat["properties"].get("osm_id"))
+        db.commit()
+        return loaded
+    finally:
+        db.close()
+
+
+def main() -> int:
+    p = argparse.ArgumentParser(description="Fetch+load EKB district polygons")
+    p.add_argument(
+        "--force-refetch",
+        action="store_true",
+        help="Re-hit Overpass even if cache exists",
+    )
+    p.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Fetch+parse only, skip DB upsert",
+    )
+    args = p.parse_args()
+
+    geojson = load_or_fetch(force=args.force_refetch)
+    log.info("Have %d features in GeoJSON", len(geojson["features"]))
+    if args.dry_run:
+        return 0
+    n = upsert_into_db(geojson)
+    log.info("Loaded %d districts into ekb_districts_geom", n)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

data/sql/56_schema_ekb_districts_geom.sql

@@ -0,0 +1,18 @@
+-- ekb_districts_geom — PostGIS-полигоны 8 районов Екатеринбурга из OSM.
+-- Загрузка через scripts/56_fetch_ekb_districts.py.
+-- Используется для spatial-join (lat,lon) → district в:
+--   1. backfill domrf_kn_objects.district_name
+--   2. recommend_mix (_velocity_baseline / _elasticity_coef / comparable / competitors)
+-- Idempotent.
+
+CREATE EXTENSION IF NOT EXISTS postgis;
+
+CREATE TABLE IF NOT EXISTS ekb_districts_geom (
+    district_name TEXT PRIMARY KEY REFERENCES ekb_districts(district_name),
+    geom geometry(MultiPolygon, 4326) NOT NULL,
+    osm_id BIGINT,
+    fetched_at TIMESTAMPTZ DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS idx_ekb_districts_geom_geom
+    ON ekb_districts_geom USING GIST (geom);

data/sql/57_kn_objects_district.sql

@@ -0,0 +1,21 @@
+-- Привязка ЖК к району Екатеринбурга через PostGIS spatial-join.
+-- Источник полигонов: ekb_districts_geom (см. 56_*).
+-- Idempotent.
+
+ALTER TABLE domrf_kn_objects
+    ADD COLUMN IF NOT EXISTS district_name TEXT;
+
+CREATE INDEX IF NOT EXISTS idx_kn_objects_district
+    ON domrf_kn_objects (district_name)
+    WHERE district_name IS NOT NULL;
+
+-- Backfill: для всех ЖК Свердл (region_cd=66) с координатами — найти район.
+-- ST_Contains даёт точное «точка внутри полигона». ЖК вне всех 8 районов
+-- (Первоуральск, Н.Тагил и т.п.) останутся с district_name=NULL.
+UPDATE domrf_kn_objects o
+   SET district_name = d.district_name
+  FROM ekb_districts_geom d
+ WHERE o.latitude IS NOT NULL AND o.longitude IS NOT NULL
+   AND o.region_cd = 66
+   AND ST_Contains(d.geom,
+                   ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326));