diff --git a/tradein-mvp/backend/app/api/v1/trade_in.py b/tradein-mvp/backend/app/api/v1/trade_in.py
index 0119af6e..dd9928f6 100644
--- a/tradein-mvp/backend/app/api/v1/trade_in.py
+++ b/tradein-mvp/backend/app/api/v1/trade_in.py
@@ -99,7 +99,10 @@ def get_estimate(
                    area_m2, rooms, floor, total_floors,
                    year_built, house_type, repair_state, has_balcony,
                    canonical_address, house_cadnum, house_fias_id,
-                   dadata_qc_geo, dadata_metro
+                   dadata_qc_geo, dadata_metro,
+                   expected_sold_price, expected_sold_range_low,
+                   expected_sold_range_high, expected_sold_per_m2,
+                   asking_to_sold_ratio, ratio_basis
             FROM trade_in_estimates
             WHERE id = CAST(:id AS uuid)
               AND expires_at > NOW()
@@ -140,6 +143,16 @@ def get_estimate(
         target_lon=row.lon,
         sources_used=row.sources_used or [],
         data_freshness_minutes=row.data_freshness_minutes,
+        # #648 Stage 3 — sold-correction columns rehydrated for shared-link reopen.
+        # (numeric ratio → float for the Pydantic field.)
+        expected_sold_price_rub=row.expected_sold_price,
+        expected_sold_range_low_rub=row.expected_sold_range_low,
+        expected_sold_range_high_rub=row.expected_sold_range_high,
+        expected_sold_per_m2=row.expected_sold_per_m2,
+        asking_to_sold_ratio=(
+            float(row.asking_to_sold_ratio) if row.asking_to_sold_ratio is not None else None
+        ),
+        ratio_basis=row.ratio_basis,
         area_m2=row.area_m2,
         rooms=row.rooms,
         floor=row.floor,
@@ -178,7 +191,10 @@ def estimate_pdf(
                    address, lat, lon, area_m2, rooms, floor, total_floors,
                    year_built, house_type, repair_state, has_balcony,
                    canonical_address, house_cadnum, house_fias_id,
-                   dadata_qc_geo, dadata_metro
+                   dadata_qc_geo, dadata_metro,
+                   expected_sold_price, expected_sold_range_low,
+                   expected_sold_range_high, expected_sold_per_m2,
+                   asking_to_sold_ratio, ratio_basis
             FROM trade_in_estimates
             WHERE id = CAST(:id AS uuid)
             """
@@ -215,6 +231,15 @@ def estimate_pdf(
         target_lon=row.lon,
         sources_used=row.sources_used or [],
         data_freshness_minutes=row.data_freshness_minutes,
+        # #648 Stage 3 — sold-correction columns rehydrated so the PDF carries them.
+        expected_sold_price_rub=row.expected_sold_price,
+        expected_sold_range_low_rub=row.expected_sold_range_low,
+        expected_sold_range_high_rub=row.expected_sold_range_high,
+        expected_sold_per_m2=row.expected_sold_per_m2,
+        asking_to_sold_ratio=(
+            float(row.asking_to_sold_ratio) if row.asking_to_sold_ratio is not None else None
+        ),
+        ratio_basis=row.ratio_basis,
         canonical_address=row.canonical_address,
         house_cadnum=row.house_cadnum,
         house_fias_id=row.house_fias_id,
diff --git a/tradein-mvp/backend/app/schemas/trade_in.py b/tradein-mvp/backend/app/schemas/trade_in.py
index 54c0ddfd..a2b20b1f 100644
--- a/tradein-mvp/backend/app/schemas/trade_in.py
+++ b/tradein-mvp/backend/app/schemas/trade_in.py
@@ -94,6 +94,18 @@ class AggregatedEstimate(BaseModel):
     data_freshness_minutes: int | None = None        # сколько минут назад был самый свежий парсинг
     est_days_on_market: int | None = None            # прогноз срока продажи (медиана по аналогам)
     cian_valuation: CianValuationSummary | None = None
+    # ── Asking→sold correction (#648 Stage 3) — PURELY ADDITIVE ──
+    # Headline median_price_rub/range_*/median_price_per_m2 остаются ASKING (активные
+    # объявления). Эти параллельные expected_sold_* = asking × per-rooms ratio
+    # (asking_to_sold_ratios, migration 080) — релевантная для выкупа цена сделки.
+    # Backtest (#648 Stage 1) показал, что коррекция убирает bias asking-медианы
+    # +20% → −4% на held-out ДКП. None если ratio-таблицы нет / бакет пуст (graceful).
+    expected_sold_price_rub: int | None = None
+    expected_sold_range_low_rub: int | None = None
+    expected_sold_range_high_rub: int | None = None
+    expected_sold_per_m2: int | None = None
+    asking_to_sold_ratio: float | None = None       # =sold/asking, ~0.72–0.93
+    ratio_basis: str | None = None                   # 'per_rooms' | 'global_fallback'
     # ── DaData enrichment (PR Q1) — on-demand для target адреса ──
     # canonical_address — DaData-нормализованная форма (с улицей в short form).
     # house_cadnum     — кадастровый номер ДОМА (для будущего matching Росреестра).
diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py
index 309dc90b..3fc96d3e 100644
--- a/tradein-mvp/backend/app/services/estimator.py
+++ b/tradein-mvp/backend/app/services/estimator.py
@@ -24,6 +24,7 @@ import json
 import logging
 import math
 import re
+import time
 from datetime import UTC, datetime, timedelta
 from typing import Any
 from uuid import uuid4
@@ -158,6 +159,81 @@ def _repair_coefficient(repair_state: str | None) -> float:
     return _REPAIR_COEF.get(repair_state, 1.0)
 
 
+# ── Asking→sold correction ratio lookup (#648 Stage 3) ──────────────────────
+# Таблица asking_to_sold_ratios (migration 080) хранит per-rooms коэффициент
+# ratio = median(SOLD ppm²) / median(ASKING ppm²) (~0.72–0.93). Estimator
+# домножает ASKING-медиану на этот ratio, получая параллельную expected_sold
+# цену (релевантную для выкупа). Headline asking-медиана НЕ меняется.
+#
+# Кэш: tiny in-process dict {bucket: (ratio, basis, fetched_monotonic)} с TTL.
+# Ratio дрейфует медленно (refresh-задача раз в сутки, Stage 4), поэтому 300с
+# TTL более чем достаточно и снимает по SELECT'у с каждой оценки. Single-worker
+# uvicorn/scheduler — GIL делает dict-доступ atomic enough (без явного lock).
+_ASKING_SOLD_RATIO_CACHE_TTL_S = 300.0
+_asking_sold_ratio_cache: dict[int, tuple[float | None, str | None, float]] = {}
+
+
+def _get_asking_sold_ratio(db: Session, rooms: int | None) -> tuple[float | None, str | None]:
+    """Возвращает (ratio, basis) asking→sold для бакета комнат.
+
+    bucket = min(max(rooms or 0, 0), 4). Сначала ищем per-rooms строку
+    (district=''), при отсутствии — global fallback (rooms_bucket=-1). Если
+    таблицы нет / пуста / любая ошибка → (None, None), НЕ raise (graceful:
+    estimator продолжает без sold-коррекции, headline asking-медиана отдаётся).
+
+    Кэшируется на бакет с TTL _ASKING_SOLD_RATIO_CACHE_TTL_S.
+    """
+    bucket = min(max(rooms or 0, 0), 4)
+
+    cached = _asking_sold_ratio_cache.get(bucket)
+    if cached is not None:
+        ratio, basis, fetched = cached
+        if (time.monotonic() - fetched) < _ASKING_SOLD_RATIO_CACHE_TTL_S:
+            return ratio, basis
+
+    ratio: float | None = None
+    basis: str | None = None
+    try:
+        row = db.execute(
+            text(
+                """
+                SELECT ratio, basis FROM asking_to_sold_ratios
+                WHERE rooms_bucket = CAST(:b AS int) AND district = ''
+                """
+            ),
+            {"b": bucket},
+        ).fetchone()
+        if row is None:
+            # Бакет тонкий (n<30 при seed'е) или отсутствует → global fallback (-1).
+            row = db.execute(
+                text(
+                    """
+                    SELECT ratio, basis FROM asking_to_sold_ratios
+                    WHERE rooms_bucket = -1 AND district = ''
+                    """
+                ),
+            ).fetchone()
+        if row is not None and row.ratio is not None:
+            ratio = float(row.ratio)
+            basis = row.basis
+    except Exception as exc:
+        # Таблицы может не быть на свежей/старой БД (миграция 080 не применена),
+        # либо транзакция в сбойном состоянии — graceful: без sold-коррекции.
+        # ОБЯЗАТЕЛЬНО rollback (как в sibling-helper'ах _get_or_fetch_*): неудачный
+        # SELECT помечает транзакцию InFailedSqlTransaction, и без отката следующий
+        # statement (_fetch_deals) упал бы → 500. Откат держит shared session чистой
+        # для последующего INSERT. rollback тоже guard'им (соединение могло умереть).
+        logger.debug("asking_to_sold_ratio lookup skipped (graceful): %s", exc)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+        ratio, basis = None, None
+
+    _asking_sold_ratio_cache[bucket] = (ratio, basis, time.monotonic())
+    return ratio, basis
+
+
 # ── Avito IMV cache lookup (Stage 3) ────────────────────────────────────────
 IMV_CACHE_TTL_HOURS = 24
 
@@ -775,6 +851,28 @@ async def estimate_quality(payload: TradeInEstimateInput, db: Session) -> Aggreg
             f"({_REPAIR_LABEL.get(payload.repair_state, '')} {pct:+d}%)."
         )
 
+    # 4c. Asking→sold коррекция (#648 Stage 3) — PURELY ADDITIVE. Headline
+    # median_price/range_*/median_ppm2 (ASKING активных объявлений) НЕ трогаем;
+    # вычисляем ПАРАЛЛЕЛЬНУЮ expected_sold цену = asking × per-rooms ratio
+    # (asking_to_sold_ratios, migration 080). Это релевантная для выкупа цена
+    # сделки (backtest #648 S1: bias asking-медианы +20% → −4% на held-out ДКП).
+    # NOTE: actual_deals (#564) остаётся ИНФОРМАЦИОННЫМ и НЕ подмешивается в
+    # headline — sold-коррекция здесь единственный sold-сигнал (без double-count).
+    asking_to_sold_ratio, ratio_basis = _get_asking_sold_ratio(db, payload.rooms)
+    if asking_to_sold_ratio is not None and listings_clean:
+        expected_sold_per_m2: int | None = round(median_ppm2 * asking_to_sold_ratio)
+        expected_sold_price: int | None = round(median_price * asking_to_sold_ratio)
+        expected_sold_range_low: int | None = round(range_low * asking_to_sold_ratio)
+        expected_sold_range_high: int | None = round(range_high * asking_to_sold_ratio)
+    else:
+        expected_sold_per_m2 = None
+        expected_sold_price = None
+        expected_sold_range_low = None
+        expected_sold_range_high = None
+        # Не было ratio (нет таблицы/бакета) — не вводим в заблуждение пустым basis.
+        if asking_to_sold_ratio is None:
+            ratio_basis = None
+
     confidence, explanation = _compute_confidence(
         n_analogs,
         median_ppm2,
@@ -924,6 +1022,9 @@ async def estimate_quality(payload: TradeInEstimateInput, db: Session) -> Aggreg
                 sources_used, data_freshness_minutes,
                 canonical_address, house_cadnum, house_fias_id,
                 dadata_qc_geo, dadata_qc_house, dadata_metro,
+                expected_sold_price, expected_sold_range_low,
+                expected_sold_range_high, expected_sold_per_m2,
+                asking_to_sold_ratio, ratio_basis,
                 expires_at
             ) VALUES (
                 CAST(:id AS uuid),
@@ -940,6 +1041,9 @@ async def estimate_quality(payload: TradeInEstimateInput, db: Session) -> Aggreg
                 :canonical_address, :house_cadnum, :house_fias_id,
                 :dadata_qc_geo, :dadata_qc_house,
                 CAST(:dadata_metro_json AS jsonb),
+                :expected_sold_price, :expected_sold_range_low,
+                :expected_sold_range_high, :expected_sold_per_m2,
+                :asking_to_sold_ratio, :ratio_basis,
                 :expires_at
             )
             """
@@ -982,6 +1086,12 @@ async def estimate_quality(payload: TradeInEstimateInput, db: Session) -> Aggreg
             "dadata_qc_geo": dadata.qc_geo if dadata else None,
             "dadata_qc_house": dadata.qc_house if dadata else None,
             "dadata_metro_json": dadata_metro_json,
+            "expected_sold_price": expected_sold_price,
+            "expected_sold_range_low": expected_sold_range_low,
+            "expected_sold_range_high": expected_sold_range_high,
+            "expected_sold_per_m2": expected_sold_per_m2,
+            "asking_to_sold_ratio": asking_to_sold_ratio,
+            "ratio_basis": ratio_basis,
             "expires_at": expires_at,
         },
     )
@@ -1068,6 +1178,12 @@ async def estimate_quality(payload: TradeInEstimateInput, db: Session) -> Aggreg
             if cian_val is not None
             else None
         ),
+        expected_sold_price_rub=expected_sold_price,
+        expected_sold_range_low_rub=expected_sold_range_low,
+        expected_sold_range_high_rub=expected_sold_range_high,
+        expected_sold_per_m2=expected_sold_per_m2,
+        asking_to_sold_ratio=asking_to_sold_ratio,
+        ratio_basis=ratio_basis,
         area_m2=payload.area_m2,
         rooms=payload.rooms,
         floor=payload.floor,
diff --git a/tradein-mvp/backend/data/sql/081_trade_in_estimates_expected_sold.sql b/tradein-mvp/backend/data/sql/081_trade_in_estimates_expected_sold.sql
new file mode 100644
index 00000000..e93b96b9
--- /dev/null
+++ b/tradein-mvp/backend/data/sql/081_trade_in_estimates_expected_sold.sql
@@ -0,0 +1,52 @@
+-- 081_trade_in_estimates_expected_sold.sql
+-- #648 Stage 3 — persist the asking→sold corrected price on each estimate.
+--
+-- ПРОБЛЕМА (#648). Estimator (estimator.py) отдаёт headline-медиану по АКТИВНЫМ
+--   asking-объявлениям. Backtest (#648 Stage 1) показал систематический bias
+--   asking-медианы +20% над реальной ДКП-ценой. Stage 2 (migration 080) сохранил
+--   per-rooms коэффициент asking_to_sold_ratios (ratio = sold/asking, ~0.72–0.93).
+--   Stage 3 (estimator) ДОПОЛНИТЕЛЬНО (purely additive — headline НЕ меняется)
+--   вычисляет expected_sold = asking × ratio. Эту параллельную цену надо
+--   ПЕРСИСТИТЬ на trade_in_estimates, чтобы shared-link reopen (GET /estimate/{id})
+--   и PDF (GET /estimate/{id}/pdf) её отдавали (re-SELECT строит AggregatedEstimate
+--   из явных колонок — без персиста значения терялись бы при reopen).
+--
+-- ДОБАВЛЯЕТ (все nullable — backward-safe; старые строки = NULL = «нет коррекции»):
+--   expected_sold_price        bigint   — asking median_price × ratio
+--   expected_sold_range_low    bigint   — asking range_low × ratio
+--   expected_sold_range_high   bigint   — asking range_high × ratio
+--   expected_sold_per_m2       int      — asking median ₽/м² × ratio
+--   asking_to_sold_ratio       numeric  — применённый ratio (=sold/asking), диагностика
+--   ratio_basis                text     — 'per_rooms' | 'global_fallback'
+--
+-- GRACEFUL: если asking_to_sold_ratios пуста / бакета нет — estimator пишет NULL
+--   в эти колонки (sold-коррекция недоступна), оценка возвращается штатно.
+--
+-- ЗАВИСИМОСТИ: trade_in_estimates (existing), asking_to_sold_ratios (080).
+--   Idempotent (ADD COLUMN IF NOT EXISTS) — безопасно ре-apply / dry-run.
+-- Apply after: 080_asking_to_sold_ratios.sql
+
+BEGIN;
+
+ALTER TABLE trade_in_estimates
+    ADD COLUMN IF NOT EXISTS expected_sold_price      bigint,
+    ADD COLUMN IF NOT EXISTS expected_sold_range_low  bigint,
+    ADD COLUMN IF NOT EXISTS expected_sold_range_high bigint,
+    ADD COLUMN IF NOT EXISTS expected_sold_per_m2     int,
+    ADD COLUMN IF NOT EXISTS asking_to_sold_ratio     numeric,
+    ADD COLUMN IF NOT EXISTS ratio_basis              text;
+
+COMMENT ON COLUMN trade_in_estimates.expected_sold_price IS
+    '#648 S3: asking median_price × per-rooms sold/asking ratio. NULL = коррекция недоступна.';
+COMMENT ON COLUMN trade_in_estimates.expected_sold_range_low IS
+    '#648 S3: asking range_low × ratio.';
+COMMENT ON COLUMN trade_in_estimates.expected_sold_range_high IS
+    '#648 S3: asking range_high × ratio.';
+COMMENT ON COLUMN trade_in_estimates.expected_sold_per_m2 IS
+    '#648 S3: asking median ₽/м² × ratio.';
+COMMENT ON COLUMN trade_in_estimates.asking_to_sold_ratio IS
+    '#648 S3: применённый ratio (=sold/asking, ~0.72–0.93) из asking_to_sold_ratios (080).';
+COMMENT ON COLUMN trade_in_estimates.ratio_basis IS
+    '#648 S3: источник ratio — ''per_rooms'' (свой бакет) | ''global_fallback'' (rooms_bucket=-1).';
+
+COMMIT;
diff --git a/tradein-mvp/backend/tests/test_estimator_cian_integration.py b/tradein-mvp/backend/tests/test_estimator_cian_integration.py
index df165a85..bcdc95ec 100644
--- a/tradein-mvp/backend/tests/test_estimator_cian_integration.py
+++ b/tradein-mvp/backend/tests/test_estimator_cian_integration.py
@@ -116,6 +116,11 @@ def _common_patches(cian_mock):
             "app.services.estimator.estimate_via_cian_valuation",
             new=cian_mock,
         ),
+        # #648 S3: stub asking→sold lookup off (isolates cian-source assertions).
+        patch(
+            "app.services.estimator._get_asking_sold_ratio",
+            return_value=(None, None),
+        ),
     ]
 
 
@@ -142,6 +147,8 @@ def test_estimator_includes_cian_valuation_when_available() -> None:
                   new=AsyncMock(return_value=None)),
             patch("app.services.estimator.estimate_via_cian_valuation",
                   new=cian_mock),
+            patch("app.services.estimator._get_asking_sold_ratio",
+                  return_value=(None, None)),
         ):
             result = await estimate_quality(payload, db)
 
@@ -172,6 +179,8 @@ def test_estimator_graceful_when_cian_returns_none() -> None:
                   new=AsyncMock(return_value=None)),
             patch("app.services.estimator.estimate_via_cian_valuation",
                   new=cian_mock),
+            patch("app.services.estimator._get_asking_sold_ratio",
+                  return_value=(None, None)),
         ):
             result = await estimate_quality(payload, db)
 
@@ -203,6 +212,8 @@ def test_estimator_graceful_when_cian_raises() -> None:
                   new=AsyncMock(return_value=None)),
             patch("app.services.estimator.estimate_via_cian_valuation",
                   new=cian_mock),
+            patch("app.services.estimator._get_asking_sold_ratio",
+                  return_value=(None, None)),
         ):
             result = await estimate_quality(payload, db)
 
@@ -236,6 +247,8 @@ def test_estimator_cian_result_no_sale_price_not_added() -> None:
                   new=AsyncMock(return_value=None)),
             patch("app.services.estimator.estimate_via_cian_valuation",
                   new=cian_mock),
+            patch("app.services.estimator._get_asking_sold_ratio",
+                  return_value=(None, None)),
         ):
             result = await estimate_quality(payload, db)
 
diff --git a/tradein-mvp/backend/tests/test_estimator_expected_sold.py b/tradein-mvp/backend/tests/test_estimator_expected_sold.py
new file mode 100644
index 00000000..8e2791bb
--- /dev/null
+++ b/tradein-mvp/backend/tests/test_estimator_expected_sold.py
@@ -0,0 +1,264 @@
+"""Tests for the asking→sold correction (#648 Stage 3).
+
+Two layers:
+
+1. `_get_asking_sold_ratio` lookup helper (DB mocked):
+   - bucket = min(max(rooms or 0, 0), 4) — clamping + None handling
+   - per-rooms row hit returns (ratio, basis)
+   - per-rooms miss falls back to the global rooms_bucket=-1 row
+   - empty / missing table → (None, None), never raises (graceful)
+   - the in-process TTL cache memoises per bucket
+
+2. The apply-logic in `estimate_quality` (every I/O stubbed, no DB, no network —
+   same isolation harness as test_estimator_repair_coef.py):
+   - when a ratio exists: expected_sold_price_rub ≈ round(median_price_rub * ratio)
+     and expected_sold_per_m2 ≈ round(median_price_per_m2 * ratio)
+   - the headline median_price_rub / ranges / per_m2 stay UNCHANGED (additive)
+   - graceful path: lookup returns (None, None) → all expected_sold_* are None,
+     no crash, headline still returned
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import UTC, datetime
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import anyio
+
+# Settings requires DATABASE_URL at init time. Set dummy DSN before any app import.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+
+# ── Layer 1: _get_asking_sold_ratio lookup helper ───────────────────────────
+
+
+class _FakeRow:
+    """Stand-in for a SQLAlchemy Row (attribute access .ratio / .basis)."""
+
+    def __init__(self, ratio: float | None, basis: str | None) -> None:
+        self.ratio = ratio
+        self.basis = basis
+
+
+def _clear_ratio_cache() -> None:
+    from app.services import estimator
+
+    estimator._asking_sold_ratio_cache.clear()
+
+
+def _db_returning(rows: list[_FakeRow | None]) -> MagicMock:
+    """MagicMock db whose successive .execute(...).fetchone() yield `rows`."""
+    db = MagicMock()
+    db.execute.return_value.fetchone.side_effect = rows
+    return db
+
+
+def test_bucket_clamping() -> None:
+    """rooms None→0, negative→0, >4→4; 0..4 pass through."""
+    from app.services.estimator import _get_asking_sold_ratio
+
+    cases = {None: 0, -3: 0, 0: 0, 1: 1, 2: 2, 3: 3, 4: 4, 5: 4, 99: 4}
+    for rooms, expected_bucket in cases.items():
+        _clear_ratio_cache()
+        db = _db_returning([_FakeRow(0.8, "per_rooms")])
+        _get_asking_sold_ratio(db, rooms)
+        # First positional bind param is {"b": bucket}.
+        bind = db.execute.call_args_list[0].args[1]
+        assert bind["b"] == expected_bucket, (
+            f"rooms={rooms} → bucket {bind['b']} != {expected_bucket}"
+        )
+
+
+def test_per_rooms_hit_returns_ratio_basis() -> None:
+    """Per-rooms row present → returned directly, no fallback query."""
+    from app.services.estimator import _get_asking_sold_ratio
+
+    _clear_ratio_cache()
+    db = _db_returning([_FakeRow(0.74, "per_rooms")])
+    ratio, basis = _get_asking_sold_ratio(db, 1)
+    assert ratio == 0.74
+    assert basis == "per_rooms"
+    assert db.execute.call_count == 1  # no global fallback needed
+
+
+def test_global_fallback_when_per_rooms_missing() -> None:
+    """Per-rooms miss (None) → second query for rooms_bucket=-1 global row."""
+    from app.services.estimator import _get_asking_sold_ratio
+
+    _clear_ratio_cache()
+    db = _db_returning([None, _FakeRow(0.79, "global_fallback")])
+    ratio, basis = _get_asking_sold_ratio(db, 3)
+    assert ratio == 0.79
+    assert basis == "global_fallback"
+    assert db.execute.call_count == 2
+
+
+def test_empty_table_returns_none_none() -> None:
+    """Both queries miss (empty table) → (None, None), no raise."""
+    from app.services.estimator import _get_asking_sold_ratio
+
+    _clear_ratio_cache()
+    db = _db_returning([None, None])
+    assert _get_asking_sold_ratio(db, 2) == (None, None)
+
+
+def test_missing_table_is_graceful() -> None:
+    """db.execute raises (relation does not exist) → (None, None), swallowed."""
+    from app.services.estimator import _get_asking_sold_ratio
+
+    _clear_ratio_cache()
+    db = MagicMock()
+    db.execute.side_effect = RuntimeError("relation asking_to_sold_ratios does not exist")
+    assert _get_asking_sold_ratio(db, 1) == (None, None)
+
+
+def test_cache_memoises_per_bucket() -> None:
+    """Second call for same bucket within TTL → no extra DB query."""
+    from app.services.estimator import _get_asking_sold_ratio
+
+    _clear_ratio_cache()
+    db = _db_returning([_FakeRow(0.8, "per_rooms")])
+    first = _get_asking_sold_ratio(db, 2)
+    second = _get_asking_sold_ratio(db, 2)
+    assert first == second == (0.8, "per_rooms")
+    assert db.execute.call_count == 1  # second served from cache
+
+
+# ── Layer 2: apply-logic inside estimate_quality (all I/O stubbed) ──────────
+
+
+def _make_listing(*, price_per_m2: float, area_m2: float = 40.0) -> dict[str, Any]:
+    price_rub = price_per_m2 * area_m2
+    return {
+        "source": "cian",
+        "source_url": "https://cian.ru/offer/1",
+        "address": "ЕКБ, ул. Учителей, 18",
+        "lat": 56.838,
+        "lon": 60.595,
+        "rooms": 1,
+        "area_m2": area_m2,
+        "floor": 4,
+        "total_floors": 16,
+        "price_rub": price_rub,
+        "price_per_m2": price_per_m2,
+        "listing_date": datetime(2026, 5, 1),
+        "days_on_market": 10,
+        "photo_urls": [],
+        "scraped_at": datetime(2026, 5, 20, tzinfo=UTC),
+        "distance_m": 100.0,
+        "relevance_score": 0.1,
+    }
+
+
+# Three fixed analogs → deterministic median ppm2 = 150_000 (< 5 ⇒ no outlier drop).
+_ANALOGS: list[dict[str, Any]] = [
+    _make_listing(price_per_m2=140_000.0),
+    _make_listing(price_per_m2=150_000.0),
+    _make_listing(price_per_m2=160_000.0),
+]
+
+
+def _make_fake_geo():
+    from app.services.geocoder import GeocodeResult
+
+    return GeocodeResult(
+        lat=56.838,
+        lon=60.595,
+        full_address="Свердловская обл., Екатеринбург, ул. Учителей, 18",
+        provider="nominatim",
+    )
+
+
+def _make_payload():
+    from app.schemas.trade_in import TradeInEstimateInput
+
+    return TradeInEstimateInput(
+        address="ЕКБ, ул. Учителей, 18",
+        area_m2=40.0,
+        rooms=1,
+        floor=4,
+        total_floors=16,
+    )
+
+
+def _run_estimate(ratio_tuple: tuple[float | None, str | None]):
+    """estimate_quality with all deps stubbed; _get_asking_sold_ratio forced."""
+    from app.services.estimator import estimate_quality
+
+    db = MagicMock()
+    payload = _make_payload()
+
+    async def _run():
+        with (
+            patch("app.services.estimator.geocode", new=AsyncMock(return_value=_make_fake_geo())),
+            patch("app.services.estimator.dadata_clean_address",
+                  new=AsyncMock(return_value=None)),
+            patch("app.services.estimator.match_house_readonly", return_value=None),
+            patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)),
+            patch("app.services.estimator._fetch_analogs",
+                  return_value=(list(_ANALOGS), False, "S")),
+            patch("app.services.estimator._fetch_deals", return_value=[]),
+            patch("app.services.estimator._get_or_fetch_imv_cached",
+                  new=AsyncMock(return_value=None)),
+            patch("app.services.estimator._get_or_fetch_yandex_valuation_cached",
+                  new=AsyncMock(return_value=None)),
+            patch("app.services.estimator.estimate_via_cian_valuation",
+                  new=AsyncMock(return_value=None)),
+            patch("app.services.estimator._get_asking_sold_ratio", return_value=ratio_tuple),
+        ):
+            return await estimate_quality(payload, db)
+
+    return anyio.run(_run)
+
+
+def test_expected_sold_applied_when_ratio_present() -> None:
+    """ratio present → expected_sold_* ≈ asking × ratio; headline UNCHANGED."""
+    ratio = 0.74
+    est = _run_estimate((ratio, "per_rooms"))
+
+    # Headline asking-median is the unadjusted market median (additive invariant).
+    expected_headline = int(150_000.0 * 40.0)  # 6_000_000
+    assert est.median_price_rub == expected_headline
+    assert est.median_price_per_m2 == 150_000
+
+    # Parallel expected_sold = asking × ratio (round()).
+    assert est.expected_sold_price_rub == round(est.median_price_rub * ratio)
+    assert est.expected_sold_per_m2 == round(est.median_price_per_m2 * ratio)
+    assert est.expected_sold_range_low_rub == round(est.range_low_rub * ratio)
+    assert est.expected_sold_range_high_rub == round(est.range_high_rub * ratio)
+    assert est.asking_to_sold_ratio == ratio
+    assert est.ratio_basis == "per_rooms"
+
+
+def test_headline_unchanged_vs_no_ratio() -> None:
+    """median_price_rub / ranges / per_m2 identical with and without a ratio."""
+    with_ratio = _run_estimate((0.8, "global_fallback"))
+    without_ratio = _run_estimate((None, None))
+
+    assert with_ratio.median_price_rub == without_ratio.median_price_rub
+    assert with_ratio.range_low_rub == without_ratio.range_low_rub
+    assert with_ratio.range_high_rub == without_ratio.range_high_rub
+    assert with_ratio.median_price_per_m2 == without_ratio.median_price_per_m2
+
+
+def test_graceful_when_ratio_none() -> None:
+    """Lookup returns (None, None) → all expected_sold_* None, no crash."""
+    est = _run_estimate((None, None))
+
+    assert est.expected_sold_price_rub is None
+    assert est.expected_sold_range_low_rub is None
+    assert est.expected_sold_range_high_rub is None
+    assert est.expected_sold_per_m2 is None
+    assert est.asking_to_sold_ratio is None
+    assert est.ratio_basis is None
+    # Headline still produced normally.
+    assert est.median_price_rub == int(150_000.0 * 40.0)
+
+
+def test_global_fallback_basis_carried_through() -> None:
+    """basis='global_fallback' propagates to the returned estimate."""
+    est = _run_estimate((0.79, "global_fallback"))
+    assert est.ratio_basis == "global_fallback"
+    assert est.asking_to_sold_ratio == 0.79
diff --git a/tradein-mvp/backend/tests/test_estimator_repair_coef.py b/tradein-mvp/backend/tests/test_estimator_repair_coef.py
index ba3760e8..24ba4b32 100644
--- a/tradein-mvp/backend/tests/test_estimator_repair_coef.py
+++ b/tradein-mvp/backend/tests/test_estimator_repair_coef.py
@@ -109,6 +109,10 @@ def _run_estimate(repair_state: str | None):
                   new=AsyncMock(return_value=None)),
             patch("app.services.estimator.estimate_via_cian_valuation",
                   new=AsyncMock(return_value=None)),
+            # #648 S3: stub asking→sold lookup off so this test isolates the
+            # repair coefficient (no sold-correction, no DB).
+            patch("app.services.estimator._get_asking_sold_ratio",
+                  return_value=(None, None)),
         ):
             return await estimate_quality(payload, db)
 
diff --git a/tradein-mvp/backend/tests/test_estimator_source_quota.py b/tradein-mvp/backend/tests/test_estimator_source_quota.py
index 79bca71f..0b1e562a 100644
--- a/tradein-mvp/backend/tests/test_estimator_source_quota.py
+++ b/tradein-mvp/backend/tests/test_estimator_source_quota.py
@@ -77,7 +77,7 @@ def test_address_cap_limits_per_address_listings() -> None:
     ]
     db = _make_db_mock(sql_rows)
 
-    result, fallback_used = _fetch_analogs(
+    result, fallback_used, _tier = _fetch_analogs(
         db, lat=56.838, lon=60.595, rooms=1, area=38.0, radius_m=1000
     )
 
@@ -116,7 +116,7 @@ def test_source_quota_prevents_cian_starvation() -> None:
     sql_rows = avito_rows + cian_rows
     db = _make_db_mock(sql_rows)
 
-    result, _ = _fetch_analogs(
+    result, _, _ = _fetch_analogs(
         db, lat=56.838, lon=60.595, rooms=1, area=38.0, radius_m=1000
     )
 
@@ -150,7 +150,7 @@ def test_source_quota_includes_all_when_supply_below_min() -> None:
     sql_rows = avito_rows + cian_rows
     db = _make_db_mock(sql_rows)
 
-    result, _ = _fetch_analogs(
+    result, _, _ = _fetch_analogs(
         db, lat=56.838, lon=60.595, rooms=1, area=38.0, radius_m=1000
     )
 
@@ -177,13 +177,13 @@ def test_fallback_signal_reflects_radius() -> None:
     ]
 
     db_default = _make_db_mock(rows)
-    _, fallback_default = _fetch_analogs(
+    _, fallback_default, _ = _fetch_analogs(
         db_default, lat=56.838, lon=60.595, rooms=1, area=38.0, radius_m=DEFAULT_RADIUS_M
     )
     assert fallback_default is False, "radius == DEFAULT should produce fallback_used=False"
 
     db_fallback = _make_db_mock(rows)
-    _, fallback_wide = _fetch_analogs(
+    _, fallback_wide, _ = _fetch_analogs(
         db_fallback, lat=56.838, lon=60.595, rooms=1, area=38.0, radius_m=FALLBACK_RADIUS_M
     )
     assert fallback_wide is True, "radius == FALLBACK should produce fallback_used=True"