fix(analytics): de-inflate domrf_kn snapshot counts (8.49×) + backfill UPSERT columns (#2464) (#2470)

backend/app/services/analytics_queries.py

@@ -504,6 +504,15 @@ def developer_history(
 
 
 def developer_portfolio(db: Session, developer_id: str) -> list[dict[str, Any]]:
+    # #2464 cluster E: без дедупа каждый ЖК возвращался ×N раз (одна строка на
+    # snapshot_date, ретенции нет) — портфель девелопера раздувался в разы (прод-замер
+    # ~7.6-9.0× per-developer). DISTINCT ON (obj_id) + snapshot_date DESC NULLS LAST —
+    # latest-snapshot-per-obj_id (тот же паттерн, что cmp_rows/latest_obj CTE в
+    # recommend_mix ниже в этом файле). dev_id — стабильный идентификатор
+    # (не меняется между снапшотами одного ЖК) → фильтруется ДО DISTINCT ON, как
+    # region_cd/district_name в сиблингах. Итоговый ORDER BY ready_dt вынесен во внешний
+    # SELECT: DISTINCT ON требует, чтобы его собственный ORDER BY начинался с ключа
+    # дедупа (obj_id, snapshot_date), не с ready_dt.
     rows = (
         db.execute(
             text(
@@ -511,8 +520,15 @@ def developer_portfolio(db: Session, developer_id: str) -> list[dict[str, Any]]:
             SELECT obj_id, comm_name, addr, region_cd, flat_count,
                    square_living, ready_dt, obj_class, escrow,
                    problem_flag, latitude, longitude, is_ekb
-              FROM domrf_kn_objects
-             WHERE dev_id = :dev
+              FROM (
+                  SELECT DISTINCT ON (obj_id)
+                         obj_id, comm_name, addr, region_cd, flat_count,
+                         square_living, ready_dt, obj_class, escrow,
+                         problem_flag, latitude, longitude, is_ekb
+                    FROM domrf_kn_objects
+                   WHERE dev_id = :dev
+                   ORDER BY obj_id, snapshot_date DESC NULLS LAST
+              ) latest
              ORDER BY ready_dt DESC NULLS LAST
             """
             ),
@@ -1797,12 +1813,38 @@ def _active_competitors_count(
     Возвращает (count, scope_used). Min 1 чтобы не делить на 0."""
 
     def _q(where_extras: str, params: dict[str, Any]) -> int:
+        # #2464 cluster E: domrf_kn_objects хранит МНОЖЕСТВО snapshot_date на obj_id
+        # (UNIQUE(obj_id, snapshot_date), ретенции нет) — прод-замер 4090 строк / 482
+        # distinct obj_id для site_status='Строящиеся' (8.49× инфляция). Наивный
+        # COUNT(*) считал каждый исторический снапшот отдельно. Fix: DISTINCT ON
+        # (obj_id) + snapshot_date DESC NULLS LAST даёт latest-snapshot-per-obj_id
+        # (сиблинг-паттерн: _L3_FUTURE_SQL #1212 в site_finder/supply_layers.py).
+        #
+        # ВСЕ волатильные предикаты применяются ПОСЛЕ DISTINCT ON (во внешнем WHERE),
+        # а не внутри CTE — иначе DISTINCT ON вернул бы «последний снапшот, ПРОШЕДШИЙ
+        # фильтр», а не истинно последний снапшот объекта. Прод-замер per-obj_id
+        # variance across snapshots (authoritative, coordinator 2026-07-08):
+        #   dev_id=0, region_cd=0 (СТАБИЛЬНЫ) · district_name=1, obj_class=29,
+        #   site_status=11 (ВОЛАТИЛЬНЫ).
+        # → В CTE остаётся ТОЛЬКО region_cd (стабилен + ограничивает стоимость дедупа
+        #   как partition-scope). {where_extras} (district_name/obj_class) и
+        #   site_status='Строящиеся' — во внешнем WHERE, на истинно-последней строке.
+        #   Пример бага, который это чинит: ЖК сменил класс Комфорт→Бизнес; при
+        #   фильтре obj_class внутри CTE взяли бы старый Комфорт-снапшот и посчитали
+        #   его активным Комфортом, хотя его актуальный класс уже Бизнес.
         n = db.execute(
             text(
                 f"""
-                SELECT COUNT(*) FROM domrf_kn_objects
-                 WHERE region_cd = :rc
-                   AND site_status = 'Строящиеся'
+                WITH latest AS (
+                    SELECT DISTINCT ON (obj_id)
+                           obj_id, site_status, district_name,
+                           obj_class, obj_class_fallback
+                      FROM domrf_kn_objects
+                     WHERE region_cd = :rc
+                     ORDER BY obj_id, snapshot_date DESC NULLS LAST
+                )
+                SELECT COUNT(*) FROM latest
+                 WHERE site_status = 'Строящиеся'
                    {where_extras}
                 """
             ),
@@ -2233,14 +2275,28 @@ def _competitors_two_dim(
         db.execute(
             text(
                 f"""
-                WITH active AS (
-                    SELECT DISTINCT ON (obj_id) obj_id, latitude, longitude, district_name
+                WITH latest AS (
+                    -- #2464 cluster E: сначала истинно-последний снапшот на obj_id,
+                    -- ТОЛЬКО по стабильному region_cd (variance=0). Волатильные
+                    -- предикаты (site_status=11, district_name=1, obj_class=29 per
+                    -- прод-замер) — НЕ внутри DISTINCT ON, иначе взяли бы «последний
+                    -- снапшот, прошедший фильтр», а не истинно последний (ЖК,
+                    -- сменивший класс/район/статус, засчитался бы по устаревшему
+                    -- снапшоту). Зеркалит _q() в _active_competitors_count выше.
+                    SELECT DISTINCT ON (obj_id)
+                           obj_id, latitude, longitude, district_name,
+                           site_status, obj_class, obj_class_fallback
                       FROM domrf_kn_objects
                      WHERE region_cd = :rc
-                       AND site_status = 'Строящиеся'
+                     ORDER BY obj_id, snapshot_date DESC NULLS LAST
+                ),
+                active AS (
+                    -- Волатильные фильтры на истинно-последней строке.
+                    SELECT obj_id, latitude, longitude, district_name
+                      FROM latest
+                     WHERE site_status = 'Строящиеся'
                        AND district_name = :dn
                        {class_filter}
-                     ORDER BY obj_id, snapshot_date DESC NULLS LAST
                 ),
                 centroid AS (
                     SELECT ST_SetSRID(ST_GeomFromText(:centroid), 4326)::geography AS pt

backend/app/services/scrapers/domrf_kn.py

@@ -535,14 +535,29 @@ UPSERT_OBJECT_SQL = text(
         :snapshot_date
     )
     ON CONFLICT (obj_id, snapshot_date) DO UPDATE SET
+        -- #2464 cluster E: hobj_id/short_addr/dev_inn/region_cd/floor_min/floor_max/
+        -- problem_flag/green_house были в INSERT (populated every call by
+        -- _norm_object) но пропущены в DO UPDATE SET → re-run скрапера для
+        -- уже существующего (obj_id, snapshot_date) оставлял эти 8 колонок stale.
+        -- Все 8 — direct EXCLUDED (не COALESCE): freshly computed каждый вызов,
+        -- в т.ч. problem_flag должен ОБНУЛЯТЬСЯ, если проблема на объекте снята
+        -- (COALESCE зафиксировал бы устаревший "проблемный" статус навсегда).
+        hobj_id = EXCLUDED.hobj_id,
         comm_name = EXCLUDED.comm_name,
         addr = EXCLUDED.addr,
+        short_addr = EXCLUDED.short_addr,
+        region_cd = EXCLUDED.region_cd,
         dev_id = EXCLUDED.dev_id,
         dev_name = EXCLUDED.dev_name,
+        dev_inn = EXCLUDED.dev_inn,
+        floor_min = EXCLUDED.floor_min,
+        floor_max = EXCLUDED.floor_max,
         flat_count = EXCLUDED.flat_count,
         square_living = EXCLUDED.square_living,
         ready_dt = EXCLUDED.ready_dt,
+        problem_flag = EXCLUDED.problem_flag,
         site_status = EXCLUDED.site_status,
+        green_house = EXCLUDED.green_house,
         escrow = EXCLUDED.escrow,
         obj_class = COALESCE(EXCLUDED.obj_class, domrf_kn_objects.obj_class),
         wall_type = COALESCE(EXCLUDED.wall_type, domrf_kn_objects.wall_type),
@@ -921,6 +936,9 @@ UPSERT_PHOTO_SQL = text(
         :local_path, :thumb_path, :downloaded_at
     )
     ON CONFLICT (obj_id, obj_file_id) DO UPDATE SET
+        -- #2464 cluster F: build_type в INSERT column list (populated every call)
+        -- но пропущен здесь → re-run скрапера оставлял build_type stale.
+        build_type = EXCLUDED.build_type,
         ord_num = EXCLUDED.ord_num,
         photo_url = EXCLUDED.photo_url,
         photo_dttm = EXCLUDED.photo_dttm,

backend/tests/services/scrapers/test_domrf_kn_upsert_sql.py

@@ -0,0 +1,107 @@
+"""#2464 cluster E + F — UPSERT DO UPDATE SET column-omission regression tests.
+
+Both UPSERT_OBJECT_SQL and UPSERT_PHOTO_SQL had columns populated on every
+INSERT (fresh scrape values, see _norm_object / catalog scraper) but silently
+omitted from `ON CONFLICT ... DO UPDATE SET`. Re-running the scraper for an
+already-seen (obj_id, snapshot_date) / (obj_id, obj_file_id) row left those
+columns stale forever (INSERT wins once, DO UPDATE never refreshes them).
+
+SQL-string assertions against the module-level `text(...)` constants — no live
+DB needed; the INSERT column list is the source of truth for "must this column
+also be kept fresh on conflict", so each test also cross-checks against the
+INSERT list to prevent the same class of omission from creeping back in for a
+newly-added column.
+"""
+
+from __future__ import annotations
+
+import re
+
+from app.services.scrapers.domrf_kn import UPSERT_OBJECT_SQL, UPSERT_PHOTO_SQL
+
+_OBJECT_PREVIOUSLY_OMITTED = [
+    "problem_flag",
+    "green_house",
+    "floor_min",
+    "floor_max",
+    "hobj_id",
+    "short_addr",
+    "dev_inn",
+    "region_cd",
+]
+
+
+def _insert_columns(sql: str, table: str) -> list[str]:
+    """Extract the `INSERT INTO <table> (col1, col2, ...)` column list."""
+    m = re.search(rf"INSERT INTO {table}\s*\(([^)]+)\)", sql)
+    assert m, f"could not locate INSERT INTO {table} (...) column list"
+    return [c.strip() for c in m.group(1).split(",")]
+
+
+def _do_update_set_columns(sql: str) -> set[str]:
+    """Extract every `col = ` LHS from the `DO UPDATE SET` clause."""
+    m = re.search(r"DO UPDATE SET(.*)", sql, re.DOTALL)
+    assert m, "could not locate ON CONFLICT ... DO UPDATE SET clause"
+    body = m.group(1)
+    return set(re.findall(r"(?:^|,)\s*(\w+)\s*=", body, re.MULTILINE))
+
+
+class TestUpsertObjectSqlDoUpdateSet:
+    def test_previously_omitted_columns_now_present(self) -> None:
+        sql = str(UPSERT_OBJECT_SQL)
+        updated = _do_update_set_columns(sql)
+        missing = [c for c in _OBJECT_PREVIOUSLY_OMITTED if c not in updated]
+        assert not missing, f"still missing from DO UPDATE SET: {missing}"
+
+    def test_previously_omitted_columns_use_direct_excluded_not_coalesce(self) -> None:
+        # These 8 are freshly computed every scrape call (per _norm_object) --
+        # COALESCE would permanently pin a stale value once ever set (most
+        # visibly wrong for problem_flag: a resolved problem must be able to
+        # clear back to NULL, not get stuck "still problematic" forever).
+        sql = " ".join(str(UPSERT_OBJECT_SQL).split())
+        for col in _OBJECT_PREVIOUSLY_OMITTED:
+            assert (
+                f"{col} = EXCLUDED.{col}" in sql
+            ), f"{col} should be direct `= EXCLUDED.{col}` (not COALESCE):\n{sql}"
+            # guard against a COALESCE(EXCLUDED.col, ...) formulation, which would
+            # permanently pin a stale value once ever set instead of refreshing it.
+            assert (
+                f"COALESCE(EXCLUDED.{col}," not in sql
+            ), f"{col} must not be wrapped in COALESCE (needs to be able to clear to NULL)"
+
+    def test_all_insert_columns_covered_by_do_update_set_or_conflict_target(self) -> None:
+        # Locks in the class of bug (INSERT populates a column, DO UPDATE SET
+        # silently drops it) so it can't recur for a column added later.
+        sql = str(UPSERT_OBJECT_SQL)
+        insert_cols = set(_insert_columns(sql, "domrf_kn_objects"))
+        updated_cols = _do_update_set_columns(sql)
+        conflict_target = {"obj_id", "snapshot_date"}  # immutable conflict key, not updated
+        missing = insert_cols - updated_cols - conflict_target
+        assert not missing, f"INSERT columns missing from DO UPDATE SET: {sorted(missing)}"
+
+    def test_no_double_colon_cast(self) -> None:
+        sql = str(UPSERT_OBJECT_SQL)
+        assert re.search(r":[a-z_]+::[a-z]", sql) is None
+
+
+class TestUpsertPhotoSqlDoUpdateSet:
+    def test_build_type_now_present(self) -> None:
+        sql = str(UPSERT_PHOTO_SQL)
+        updated = _do_update_set_columns(sql)
+        assert "build_type" in updated
+
+    def test_build_type_uses_direct_excluded(self) -> None:
+        sql = " ".join(str(UPSERT_PHOTO_SQL).split())
+        assert "build_type = EXCLUDED.build_type" in sql
+
+    def test_all_insert_columns_covered_by_do_update_set_or_conflict_target(self) -> None:
+        sql = str(UPSERT_PHOTO_SQL)
+        insert_cols = set(_insert_columns(sql, "domrf_kn_photos"))
+        updated_cols = _do_update_set_columns(sql)
+        conflict_target = {"obj_id", "obj_file_id"}
+        missing = insert_cols - updated_cols - conflict_target
+        assert not missing, f"INSERT columns missing from DO UPDATE SET: {sorted(missing)}"
+
+    def test_no_double_colon_cast(self) -> None:
+        sql = str(UPSERT_PHOTO_SQL)
+        assert re.search(r":[a-z_]+::[a-z]", sql) is None

backend/tests/services/test_analytics_queries_domrf_dedup.py

@@ -0,0 +1,380 @@
+"""#2464 cluster E — domrf_kn_objects snapshot dedup regression tests.
+
+domrf_kn_objects retains MULTIPLE historical rows per obj_id (one per
+snapshot_date, UNIQUE(obj_id, snapshot_date), no retention). Prod measurement:
+4090 rows / 482 distinct obj_id for site_status='Строящиеся' (8.49x inflation).
+
+Three read-side queries counted/returned every retained snapshot instead of the
+latest one per obj_id:
+  - _active_competitors_count (via its _q closure)
+  - developer_portfolio
+  - _competitors_two_dim (its `active` CTE)
+
+All fixed with the established repo dedup pattern: DISTINCT ON (obj_id) ...
+ORDER BY obj_id, snapshot_date DESC NULLS LAST (sibling: _L3_FUTURE_SQL #1212 in
+app/services/site_finder/supply_layers.py, cmp_rows/latest_obj CTE in
+analytics_queries.recommend_mix).
+
+VOLATILE-FILTER PLACEMENT (authoritative prod measurement, coordinator
+2026-07-08 — per-obj_id variance across snapshots): dev_id=0, region_cd=0
+(STABLE) vs district_name=1, obj_class=29, site_status=11 (VOLATILE). Volatile
+predicates MUST be applied to the true-latest row (outer/downstream WHERE), NOT
+pre-filtered inside the DISTINCT ON CTE — else DISTINCT ON returns "the latest
+snapshot that still MATCHED the filter", not the object's true latest snapshot
+(a ЖК that sold out / changed class / moved districts would be miscounted by a
+stale snapshot). Only the stable region_cd stays inside the CTE as the dedup
+partition-scope. dev_id (developer_portfolio) is likewise stable → kept in-CTE.
+
+Mock-based — mirrors the SQL-shape assertion convention used in
+tests/services/site_finder/test_supply_layers.py (TestLayer2Hidden.
+test_sql_dedups_latest_snapshot) since there's no SQLite stand-in for
+Postgres-only `DISTINCT ON` syntax and no throwaway-schema DML fixture in this
+repo's integration-test conventions (tests/integration/conftest.py is
+EXPLAIN-only / read-only against a real tunnel DB — safe for plan-checks, not
+for INSERT fixtures). A second class (`TestDedupAlgorithmSpec`) hand-replicates
+the exact dedup+filter semantics the SQL implements in pure Python and asserts
+the two behaviors the epic calls out explicitly: multi-snapshot obj_id counted
+once, and a sold-out object with only an OLD 'Строящиеся' snapshot excluded.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock
+
+from app.services.analytics_queries import _active_competitors_count, developer_portfolio
+
+# ── shared helpers (mirrors test_supply_layers.py _mock_db/_executed_sql) ─────
+
+
+def _mock_scalar_db(value: int) -> MagicMock:
+    """Session stand-in: every db.execute(...).scalar() returns `value`."""
+    db = MagicMock()
+    db.execute.return_value.scalar.return_value = value
+    return db
+
+
+def _mock_mapping_db(rows: list[dict[str, Any]]) -> MagicMock:
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.all.return_value = rows
+    return db
+
+
+def _executed_sql(db: MagicMock, call_index: int = 0) -> str:
+    args, _kwargs = db.execute.call_args_list[call_index]
+    return str(args[0])
+
+
+def _executed_params(db: MagicMock, call_index: int = 0) -> dict:
+    args, _kwargs = db.execute.call_args_list[call_index]
+    return args[1]
+
+
+def _split_latest_cte(sql: str) -> tuple[str, str]:
+    """Split whitespace-normalized SQL into (CTE body, outer body).
+
+    CTE body = `WITH latest AS ( ... )` up to the `)` that closes the CTE,
+    which is the first `)` after the ORDER BY feeding DISTINCT ON. Outer body =
+    everything after. Lets tests assert WHICH side a predicate lives on.
+    """
+    norm = " ".join(sql.split())
+    cte_open = norm.index("WITH latest AS (")
+    order_by_idx = norm.index("ORDER BY obj_id, snapshot_date DESC NULLS LAST")
+    cte_close = norm.index(")", order_by_idx)
+    return norm[cte_open:cte_close], norm[cte_close:]
+
+
+# ── _active_competitors_count SQL shape ────────────────────────────────────
+
+
+class TestActiveCompetitorsCountSqlShape:
+    def test_dedups_via_distinct_on_obj_id_latest_snapshot(self) -> None:
+        # n never reaches the >=2 early-return threshold with scalar()==0, so
+        # all 3 cascade tiers execute — call_index 0 is tier1 (district+class).
+        db = _mock_scalar_db(0)
+        _active_competitors_count(
+            db, region_code=66, district_name="Ленинский", target_class="Комфорт"
+        )
+        sql = _executed_sql(db, 0)
+        norm = " ".join(sql.split())
+        assert "DISTINCT ON (obj_id)" in norm
+        assert "ORDER BY obj_id, snapshot_date DESC NULLS LAST" in norm
+
+    def test_site_status_filter_is_volatile_applied_after_distinct_on(self) -> None:
+        # Regression guard for the exact bug shape #2464 warns about: a
+        # sold-out object whose ONLY 'Строящиеся' row is an old snapshot must
+        # not be counted. That requires site_status to be filtered on the
+        # OUTER (post-DISTINCT-ON) query, never inside the `latest` CTE's
+        # WHERE — else DISTINCT ON would pick "the latest snapshot that still
+        # matched site_status='Строящиеся'" instead of the object's true
+        # latest snapshot. (Prod per-obj_id variance: site_status=11.)
+        db = _mock_scalar_db(0)
+        _active_competitors_count(
+            db, region_code=66, district_name="Ленинский", target_class="Комфорт"
+        )
+        cte_body, outer_body = _split_latest_cte(_executed_sql(db, 0))
+        # site_status legitimately appears in the CTE's SELECT list (it's the
+        # column DISTINCT ON needs to expose) -- what must NOT appear is a
+        # filter predicate on it inside the CTE's WHERE.
+        assert (
+            "site_status = 'Строящиеся'" not in cte_body
+        ), f"site_status must not pre-filter the DISTINCT ON CTE (volatile field):\n{cte_body}"
+        assert "site_status = 'Строящиеся'" in outer_body
+
+    def test_district_and_class_are_volatile_applied_after_distinct_on(self) -> None:
+        # Companion to the site_status guard: authoritative prod measurement
+        # (coordinator 2026-07-08) — per-obj_id variance across snapshots is
+        # district_name=1 and obj_class=29 (NON-zero → VOLATILE), unlike
+        # dev_id/region_cd (0 → stable). So district_name/obj_class predicates
+        # must ALSO be applied on the true-latest row (outer WHERE), never
+        # pre-filtered inside the DISTINCT ON CTE — else a ЖК that changed
+        # class/district gets counted by a stale snapshot. Tier-1
+        # (district+class) is call_index 0 when target_class is given.
+        db = _mock_scalar_db(0)
+        _active_competitors_count(
+            db, region_code=66, district_name="Ленинский", target_class="Комфорт"
+        )
+        cte_body, outer_body = _split_latest_cte(_executed_sql(db, 0))
+        # CTE WHERE must scope on ONLY the stable region_cd — no volatile predicate.
+        assert (
+            "district_name = :dn" not in cte_body
+        ), f"district_name (volatile) must not pre-filter the DISTINCT ON CTE:\n{cte_body}"
+        assert (
+            "COALESCE(obj_class, obj_class_fallback) = :cls" not in cte_body
+        ), f"obj_class (volatile) must not pre-filter the DISTINCT ON CTE:\n{cte_body}"
+        # Both live in the outer WHERE, applied to the deduped true-latest row.
+        assert "district_name = :dn" in outer_body
+        assert "COALESCE(obj_class, obj_class_fallback) = :cls" in outer_body
+
+    def test_cte_where_scopes_only_stable_region_cd(self) -> None:
+        # Belt-and-suspenders: the DISTINCT ON CTE's WHERE partition-scope is
+        # exactly `region_cd = :rc` and nothing else (all volatile predicates
+        # relocated). Guards against a future re-introduction of a volatile
+        # pre-filter that would silently pick a stale-but-matching snapshot.
+        db = _mock_scalar_db(0)
+        _active_competitors_count(
+            db, region_code=66, district_name="Ленинский", target_class="Комфорт"
+        )
+        cte_body, _outer = _split_latest_cte(_executed_sql(db, 0))
+        # extract the CTE's WHERE clause (between WHERE and the feeding ORDER BY,
+        # which _split_latest_cte leaves at the tail of the CTE body).
+        where_start = cte_body.index("WHERE ") + len("WHERE ")
+        where_end = cte_body.index("ORDER BY", where_start)
+        where_clause = cte_body[where_start:where_end].strip()
+        assert (
+            where_clause == "region_cd = :rc"
+        ), f"CTE WHERE must scope on ONLY stable region_cd, got: {where_clause!r}"
+
+    def test_no_double_colon_cast(self) -> None:
+        import re
+
+        db = _mock_scalar_db(0)
+        _active_competitors_count(db, region_code=66, district_name="Ленинский", target_class=None)
+        sql = _executed_sql(db, 0)
+        assert re.search(r":[a-z_]+::[a-z]", sql) is None
+
+    def test_cascade_still_stops_early_when_tier_hits_threshold(self) -> None:
+        # Sanity: fallback cascade logic (n>=2 -> stop) is untouched by the dedup fix.
+        db = _mock_scalar_db(5)
+        n, scope = _active_competitors_count(
+            db, region_code=66, district_name="Ленинский", target_class="Комфорт"
+        )
+        assert n == 5
+        assert scope == "district+class"
+        assert db.execute.call_count == 1
+
+
+# ── developer_portfolio SQL shape ──────────────────────────────────────────
+
+
+class TestDeveloperPortfolioSqlShape:
+    def test_dedups_via_distinct_on_obj_id_latest_snapshot(self) -> None:
+        db = _mock_mapping_db([])
+        developer_portfolio(db, "6208_0")
+        sql = _executed_sql(db, 0)
+        norm = " ".join(sql.split())
+        assert "DISTINCT ON (obj_id)" in norm
+        assert "ORDER BY obj_id, snapshot_date DESC NULLS LAST" in norm
+
+    def test_preserves_final_ready_dt_ordering(self) -> None:
+        # DISTINCT ON's own ORDER BY must start with (obj_id, snapshot_date) —
+        # the caller-facing ready_dt ordering has to live in the OUTER SELECT.
+        db = _mock_mapping_db([])
+        developer_portfolio(db, "6208_0")
+        sql = _executed_sql(db, 0)
+        norm = " ".join(sql.split())
+        assert norm.rstrip().endswith("ORDER BY ready_dt DESC NULLS LAST")
+
+    def test_dev_id_filter_unchanged(self) -> None:
+        db = _mock_mapping_db([])
+        developer_portfolio(db, "6208_0")
+        params = _executed_params(db, 0)
+        assert params == {"dev": "6208_0"}
+
+    def test_returned_columns_unchanged(self) -> None:
+        # Fix must not add/drop columns from the caller-facing shape.
+        rows = [
+            {
+                "obj_id": 1,
+                "comm_name": "ЖК Тест",
+                "addr": "ул. Тест, 1",
+                "region_cd": 66,
+                "flat_count": 100,
+                "square_living": 5000.0,
+                "ready_dt": None,
+                "obj_class": "Комфорт",
+                "escrow": True,
+                "problem_flag": None,
+                "latitude": 56.8,
+                "longitude": 60.6,
+                "is_ekb": True,
+            }
+        ]
+        db = _mock_mapping_db(rows)
+        out = developer_portfolio(db, "6208_0")
+        assert len(out) == 1
+        assert set(out[0].keys()) == {
+            "obj_id",
+            "comm_name",
+            "addr",
+            "region_cd",
+            "flat_count",
+            "square_living",
+            "ready_dt",
+            "obj_class",
+            "escrow",
+            "problem_flag",
+            "lat",
+            "lon",
+            "is_ekb",
+        }
+
+
+# ── _competitors_two_dim `active`/`latest` CTE shape ───────────────────────
+
+
+def _mock_two_dim_db(radius_n: int, district_only_n: int) -> MagicMock:
+    """Session stand-in for _competitors_two_dim: call 0 = centroid lookup
+    (returns a non-null WKT so the main query runs), call 1 = the radius/
+    district aggregate. radius_n/district_only_n chosen so total_weighted >= 1
+    keeps execution off the single-dim fallback path (which would add calls)."""
+    db = MagicMock()
+    centroid_res = MagicMock()
+    centroid_res.mappings.return_value.first.return_value = {"centroid_wkt": "POINT(60.6 56.8)"}
+    main_res = MagicMock()
+    main_res.mappings.return_value.first.return_value = {
+        "radius_n": radius_n,
+        "district_only_n": district_only_n,
+    }
+    db.execute.side_effect = [centroid_res, main_res]
+    return db
+
+
+class TestCompetitorsTwoDimActiveCte:
+    """#2464 sibling: _competitors_two_dim's `active` CTE had site_status /
+    district_name / obj_class ALL inside its DISTINCT ON — same latest-matching-
+    not-latest bug. Fix dedups to true-latest per obj_id first (region_cd-only
+    CTE scope), then applies the volatile predicates."""
+
+    def _main_sql(self, db: MagicMock) -> str:
+        # call_index 1 is the radius/district aggregate (0 is the centroid lookup).
+        return " ".join(_executed_sql(db, 1).split())
+
+    def test_dedups_via_distinct_on_before_volatile_filters(self) -> None:
+        from app.services.analytics_queries import _competitors_two_dim
+
+        db = _mock_two_dim_db(radius_n=3, district_only_n=0)
+        _competitors_two_dim(db, region_code=66, district_name="Ленинский", target_class="Комфорт")
+        sql = self._main_sql(db)
+        assert "DISTINCT ON (obj_id)" in sql
+        assert "ORDER BY obj_id, snapshot_date DESC NULLS LAST" in sql
+
+    def test_volatile_filters_are_post_distinct_on(self) -> None:
+        from app.services.analytics_queries import _competitors_two_dim
+
+        db = _mock_two_dim_db(radius_n=3, district_only_n=0)
+        _competitors_two_dim(db, region_code=66, district_name="Ленинский", target_class="Комфорт")
+        sql = self._main_sql(db)
+        # Split the DISTINCT ON CTE (`latest`) from everything downstream. Its
+        # WHERE must scope on ONLY stable region_cd; site_status/district_name/
+        # obj_class predicates live in the downstream `active` CTE.
+        latest_open = sql.index("WITH latest AS (")
+        order_by_idx = sql.index("ORDER BY obj_id, snapshot_date DESC NULLS LAST")
+        latest_close = sql.index(")", order_by_idx)
+        latest_body = sql[latest_open:latest_close]
+        downstream = sql[latest_close:]
+
+        assert "site_status = 'Строящиеся'" not in latest_body
+        assert "district_name = :dn" not in latest_body
+        assert "COALESCE(obj_class, obj_class_fallback) = :cls" not in latest_body
+
+        assert "site_status = 'Строящиеся'" in downstream
+        assert "district_name = :dn" in downstream
+        assert "COALESCE(obj_class, obj_class_fallback) = :cls" in downstream
+
+    def test_output_shape_preserved(self) -> None:
+        from app.services.analytics_queries import _competitors_two_dim
+
+        db = _mock_two_dim_db(radius_n=3, district_only_n=2)
+        radius_n, district_only_n, total_weighted, scope = _competitors_two_dim(
+            db, region_code=66, district_name="Ленинский", target_class="Комфорт"
+        )
+        assert radius_n == 3
+        assert district_only_n == 2
+        # 3*1.0 + 2*0.6 = 4.2
+        assert total_weighted == 4.2
+        assert scope == "district_2d"
+
+
+# ── Pure-Python spec replica: exact dedup + volatile-filter algorithm ─────
+#
+# MagicMock can't execute real SQL (and DISTINCT ON has no SQLite equivalent),
+# so this class locks in the ALGORITHM the SQL is required to implement:
+# "take the row with MAX(snapshot_date) per obj_id, THEN filter on that row's
+# site_status" — as opposed to the buggy "filter rows on site_status, then
+# take MAX(snapshot_date) among survivors". It's a spec/regression guard, not
+# a live-DB execution test (see module docstring for why).
+
+
+def _latest_snapshot_active_count(rows: list[dict[str, Any]]) -> int:
+    """Reference implementation of what the fixed `_q` SQL computes."""
+    latest_by_obj: dict[int, dict[str, Any]] = {}
+    for row in rows:
+        obj_id = row["obj_id"]
+        cur = latest_by_obj.get(obj_id)
+        if cur is None or (row["snapshot_date"] or "") > (cur["snapshot_date"] or ""):
+            latest_by_obj[obj_id] = row
+    return sum(1 for r in latest_by_obj.values() if r["site_status"] == "Строящиеся")
+
+
+class TestDedupAlgorithmSpec:
+    def test_multi_snapshot_same_obj_id_counted_once(self) -> None:
+        rows = [
+            {"obj_id": 1, "snapshot_date": "2026-04-27", "site_status": "Строящиеся"},
+            {"obj_id": 1, "snapshot_date": "2026-05-25", "site_status": "Строящиеся"},
+            {"obj_id": 1, "snapshot_date": "2026-06-28", "site_status": "Строящиеся"},
+        ]
+        assert _latest_snapshot_active_count(rows) == 1
+
+    def test_sold_out_object_with_old_active_snapshot_not_counted(self) -> None:
+        # obj_id=2's LATEST snapshot says "Реализован" (sold out) even though
+        # older snapshots said "Строящиеся" — must NOT count as active.
+        rows = [
+            {"obj_id": 2, "snapshot_date": "2026-04-27", "site_status": "Строящиеся"},
+            {"obj_id": 2, "snapshot_date": "2026-05-25", "site_status": "Строящиеся"},
+            {"obj_id": 2, "snapshot_date": "2026-06-28", "site_status": "Реализован"},
+        ]
+        assert _latest_snapshot_active_count(rows) == 0
+
+    def test_mixed_objects_only_latest_active_ones_counted(self) -> None:
+        rows = [
+            # obj 1: still active across all 3 snapshots -> counts once.
+            {"obj_id": 1, "snapshot_date": "2026-04-27", "site_status": "Строящиеся"},
+            {"obj_id": 1, "snapshot_date": "2026-06-28", "site_status": "Строящиеся"},
+            # obj 2: sold out on the latest snapshot -> excluded.
+            {"obj_id": 2, "snapshot_date": "2026-04-27", "site_status": "Строящиеся"},
+            {"obj_id": 2, "snapshot_date": "2026-06-28", "site_status": "Реализован"},
+            # obj 3: single snapshot, active -> counts once.
+            {"obj_id": 3, "snapshot_date": "2026-06-28", "site_status": "Строящиеся"},
+        ]
+        assert _latest_snapshot_active_count(rows) == 2