fix(tradein-sql): dedup cross-source houses in 063 migration

A physical building with both Avito ext_house_id (3 listings) and
cian/yandex listings (5 lots, no ext_id) was getting TWO houses rows
(one avito, one derived). Step 4 then split the 8 listings across them.

Fix:
- Step 2 (derived insert): NOT IN (avito-addresses) — skip if already covered
- Step 4 (fuzzy link): COALESCE prefers avito-house at same addr over derived

Adds sanity check: WARN if any normalized address has multiple houses rows.

tradein-mvp/backend/app/schemas/trade_in.py

@@ -46,6 +46,20 @@ class AnalogLot(BaseModel):
     distance_m: int | None = None      # расстояние до целевой квартиры в метрах
 
 
+class CianValuationSummary(BaseModel):
+    """Cian Valuation Calculator данные для UI.
+
+    Источник: external_valuations table (source='cian_valuation').
+    Заполняется только при successful Cian Calculator call в estimator.py.
+    """
+
+    sale_price_rub: int | None = None
+    rent_price_rub: int | None = None
+    chart: list[dict[str, Any]] = Field(default_factory=list)
+    chart_change_pct: float | None = None
+    chart_change_direction: Literal["increase", "decrease", "neutral"] | None = None
+
+
 class AggregatedEstimate(BaseModel):
     estimate_id: UUID
     median_price_rub: int
@@ -66,6 +80,7 @@ class AggregatedEstimate(BaseModel):
     sources_used: list[str] = Field(default_factory=list)  # ['avito', 'cian', 'rosreestr']
     data_freshness_minutes: int | None = None        # сколько минут назад был самый свежий парсинг
     est_days_on_market: int | None = None            # прогноз срока продажи (медиана по аналогам)
+    cian_valuation: CianValuationSummary | None = None
     # ── Параметры оценённой квартиры — нужны, чтобы восстановить карточку
     #    при открытии оценки по ссылке (?id=), когда формы-инпута уже нет ──
     area_m2: float | None = None

tradein-mvp/backend/app/services/estimator.py

@@ -31,7 +31,12 @@ from uuid import uuid4
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
-from app.schemas.trade_in import AggregatedEstimate, AnalogLot, TradeInEstimateInput
+from app.schemas.trade_in import (
+    AggregatedEstimate,
+    AnalogLot,
+    CianValuationSummary,
+    TradeInEstimateInput,
+)
 from app.services.geocoder import GeocodeResult, geocode
 from app.services.house_metadata import get_house_metadata
 from app.services.scrapers.avito_imv import (
@@ -865,6 +870,17 @@ async def estimate_quality(
         sources_used=sources_used,
         data_freshness_minutes=freshness_min,
         est_days_on_market=_estimate_days_on_market(listings_clean, deals),
+        cian_valuation=(
+            CianValuationSummary(
+                sale_price_rub=int(cian_val.sale_price_rub) if cian_val.sale_price_rub else None,
+                rent_price_rub=int(cian_val.rent_price_rub) if cian_val.rent_price_rub else None,
+                chart=list(cian_val.chart or []),
+                chart_change_pct=cian_val.chart_change_pct,
+                chart_change_direction=cian_val.chart_change_direction,
+            )
+            if cian_val is not None
+            else None
+        ),
         area_m2=payload.area_m2,
         rooms=payload.rooms,
         floor=payload.floor,
@@ -1598,4 +1614,5 @@ def _empty_estimate(
         analogs=[],
         actual_deals=[],
         expires_at=expires_at,
+        cian_valuation=None,
     )

tradein-mvp/backend/data/sql/063_backfill_houses_and_link_listings.sql

@@ -11,10 +11,13 @@
 --   1. CREATE OR REPLACE FUNCTION tradein_normalize_short_addr
 --   2. INSERT Avito/CIAN-sourced houses from listings WHERE house_source+house_ext_id NOT NULL
 --   3. INSERT derived houses for listings WITHOUT ext_house_id (grouped by normalized address)
+--      DEDUP: skips addresses already covered by step 2 (prevents split house rows)
 --   4. UPDATE listings.house_id_fk via source+ext_house_id match (direct path)
 --   5. UPDATE listings.house_id_fk via normalized address match (fuzzy path)
+--      DEDUP: COALESCE prefers avito-sourced house over derived at same address
 --   6. Best-effort relink house_placement_history via raw_payload->>'address' (likely 0 rows)
---   7. RAISE NOTICE with final counters
+--   7. Dedup sanity check: WARN if any normalized address maps to >1 houses rows
+--   8. RAISE NOTICE with final counters
 --
 -- Dependencies:
 --   - pgcrypto extension (for digest()) -- verified present
@@ -117,6 +120,11 @@ ON CONFLICT (source, ext_house_id) DO NOTHING;
 -- Groups listings by normalized address; each group = one derived house.
 -- Synthetic ext_house_id = first 32 hex chars of sha256(normalized_addr).
 -- Uses DISTINCT ON (norm_addr) picking latest scraped_at per group.
+--
+-- DEDUP FIX: Excludes normalized addresses already covered by Step 2's
+-- avito/cian-sourced houses. Prevents the same physical building getting
+-- TWO rows (one avito-sourced, one derived) when it has listings from
+-- both Avito (with ext_house_id) and other sources (without ext_house_id).
 -- -------------------------------------------------------------------------
 INSERT INTO houses (
     source,
@@ -138,14 +146,14 @@ SELECT DISTINCT ON (norm_addr)
     'derived'                                                              AS source,
     substring(encode(digest(norm_addr, 'sha256'), 'hex'), 1, 32)          AS ext_house_id,
     ''                                                                     AS url,
-    l.address,
-    l.lat,
-    l.lon,
-    l.geom,
-    l.year_built,
-    l.house_type,
-    l.total_floors,
-    l.address                                                              AS full_address,
+    src.address,
+    src.lat,
+    src.lon,
+    src.geom,
+    src.year_built,
+    src.house_type,
+    src.total_floors,
+    src.address                                                            AS full_address,
     norm_addr                                                              AS short_address,
     NOW()                                                                  AS first_seen_at,
     NOW()                                                                  AS last_scraped_at
@@ -156,10 +164,15 @@ FROM (
     FROM listings l
     WHERE l.address IS NOT NULL
       AND (l.house_source IS NULL OR l.house_ext_id IS NULL)
-) l
-WHERE l.norm_addr IS NOT NULL
-  AND length(l.norm_addr) >= 5
-ORDER BY norm_addr, l.scraped_at DESC
+) src
+WHERE src.norm_addr IS NOT NULL
+  AND length(src.norm_addr) >= 5
+  AND src.norm_addr NOT IN (
+      SELECT tradein_normalize_short_addr(address)
+      FROM houses
+      WHERE source != 'derived' AND address IS NOT NULL
+  )
+ORDER BY norm_addr, src.scraped_at DESC
 ON CONFLICT (source, ext_house_id) DO NOTHING;
 
 -- -------------------------------------------------------------------------
@@ -178,20 +191,43 @@ WHERE l.house_id_fk IS NULL
 -- -------------------------------------------------------------------------
 -- Step 5: Link listings.house_id_fk via normalized address (fuzzy path)
 -- Covers listings where house_source/house_ext_id is absent.
--- houses.short_address was populated above; joined by equality.
+-- houses.short_address was populated above (derived rows); joined by equality.
+--
+-- DEDUP FIX: For cian/yandex listings at an address that already has an
+-- avito-sourced house (from Step 2), we prefer that avito-sourced house
+-- over any derived row. This consolidates all listings for the same physical
+-- building under a single houses row regardless of scrape source.
+--
+-- COALESCE priority:
+--   1. Non-derived house whose address normalizes to the same value
+--   2. Derived house whose short_address equals the normalized listing address
+--
 -- NOTE: houses_addr_fp_idx is on address_fingerprint; short_address has no
 -- dedicated index. If this migration is re-run on large data, consider:
 --   CREATE INDEX IF NOT EXISTS houses_short_addr_idx ON houses(short_address)
 --   WHERE source = 'derived';
 -- For a one-shot migration on 18k rows this is acceptable (seq scan ~ms).
 -- -------------------------------------------------------------------------
+WITH norm_l AS (
+    SELECT id, tradein_normalize_short_addr(address) AS norm_addr
+    FROM listings
+    WHERE house_id_fk IS NULL AND address IS NOT NULL
+)
 UPDATE listings l
-SET house_id_fk = h.id
-FROM houses h
-WHERE l.house_id_fk IS NULL
-  AND l.address IS NOT NULL
-  AND h.source = 'derived'
-  AND h.short_address = tradein_normalize_short_addr(l.address);
+SET house_id_fk = COALESCE(
+    -- Prefer avito/cian-sourced house at same normalized address (cross-source consolidation)
+    (SELECT h.id FROM houses h
+     WHERE h.source != 'derived'
+       AND tradein_normalize_short_addr(h.address) = nl.norm_addr
+     LIMIT 1),
+    -- Fall back to derived house
+    (SELECT h.id FROM houses h
+     WHERE h.source = 'derived'
+       AND h.short_address = nl.norm_addr
+     LIMIT 1)
+)
+FROM norm_l nl
+WHERE l.id = nl.id AND l.house_id_fk IS NULL;
 
 -- -------------------------------------------------------------------------
 -- Step 6: Best-effort relink orphan house_placement_history
@@ -209,7 +245,30 @@ WHERE hph.house_id IS NULL
   AND h.short_address = tradein_normalize_short_addr(hph.raw_payload ->> 'address');
 
 -- -------------------------------------------------------------------------
--- Step 7: Final counters via RAISE NOTICE
+-- Step 7: Dedup sanity check
+-- Warn if any normalized address maps to more than one houses row.
+-- A non-zero count after the dedup fix above means manual review is needed.
+-- -------------------------------------------------------------------------
+DO $$
+DECLARE
+    dup_count int;
+BEGIN
+    SELECT count(*) INTO dup_count FROM (
+        SELECT tradein_normalize_short_addr(address) AS norm, count(*) AS n
+        FROM houses
+        WHERE address IS NOT NULL
+        GROUP BY tradein_normalize_short_addr(address)
+        HAVING count(*) > 1
+    ) dup;
+    IF dup_count > 0 THEN
+        RAISE WARNING 'backfill 063: % normalized addresses still have multiple houses rows', dup_count;
+    ELSE
+        RAISE NOTICE 'backfill 063: dedup OK — 1 house per normalized address';
+    END IF;
+END $$;
+
+-- -------------------------------------------------------------------------
+-- Step 8: Final counters via RAISE NOTICE
 -- -------------------------------------------------------------------------
 DO $$
 DECLARE
@@ -234,7 +293,7 @@ BEGIN
     SELECT count(*) FILTER (WHERE house_id IS NOT NULL) INTO v_history_linked FROM house_placement_history;
 
     RAISE NOTICE
-        'backfill 063: houses=% (direct=%, derived=%)'
+        'backfill 063 final: houses=% (direct=%, derived=%)'
         ' | listings_linked=%/% (addr_coverage=%)'
         ' | history_linked=%/%',
         v_houses_total, v_houses_direct, v_houses_derived,