"""СберИндекс city-level secondary-housing price index pull (#887). Fetches monthly time-series data from sberindex.ru/api/sowa (reverse-engineered in vault research note 2026-05-31-0845-resolver-794) across three dashboards: - real_estate_deals — hedonic sold-price index - residential_real_estate_prices — repeat-sales hedonic (cleanest signal) - dinamika-tsen-obyavlenii — asking-price benchmark API summary: POST https://sberindex.ru/api/sowa Body: {"SOWA": {"method": "GET", "route": "", "data": {"type": "object", "value": []}}} route = base64( /dataset/v1/?filter=&limit=1000&offset=0 ) filter = {"REF_AREA": "", "FREQ": "M", "SOURCE": "SI", "REALTY": "2"} Required headers: content-type, rquid (random 32-hex), x-language, user-agent. Response fields (each cell is __string__ or __number__): indicator_id, kpi_id, period, value, obs_status, source, ref_area, realty, freq, unit_measure, unit_mult Data-quality benchmark: after upsert of the dinamika-tsen-obyavlenii (asking) series, the latest asking index per city is logged at INFO level as a data-quality reference. Full asking-vs-our-median reconciliation is a separate follow-up (#887 step 5+). NOTE — estimator time-adjust wire (step 4) is OUT OF SCOPE for this PR. It is a client-visible change that requires a backtest gate. Defer to a separate issue. REF_AREA codes (sberindex internal region IDs, not ОКАТО/ISO): 643 = Россия — confirmed live, vault research note 2026-05-31-0845. 66 = Свердловская область — verified 2026-05-31 brute-force against /api/sowa (region name from ref_area field of response); real_estate_deals: 2017-01 51 046 → 2026-04 128 443 руб/м². 77 = Москва — verified 2026-05-31 brute-force against /api/sowa (region name from ref_area field of response); real_estate_deals: 2017-01 148 471 → 2026-04 309 510 руб/м². """ from __future__ import annotations import base64 import json import logging import uuid from collections.abc import AsyncIterator from datetime import UTC, date, datetime from urllib.parse import quote import httpx from sqlalchemy import text from sqlalchemy.orm import Session logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Config — REF_AREA codes # --------------------------------------------------------------------------- # REF_AREA code → city label used as the primary key in sber_price_index. # Codes verified 2026-05-31 by brute-force scan against /api/sowa # (region name taken from ref_area field of API response; not ОКАТО/ISO). SBER_REF_AREAS: dict[str, str] = { "643": "Россия", "66": "Свердловская область", "77": "Москва", } # Dashboards to pull: slug → human label (used as 'segment' placeholder) SBER_DASHBOARDS: list[str] = [ "real_estate_deals", "residential_real_estate_prices", "dinamika-tsen-obyavlenii", ] SBER_API_URL = "https://sberindex.ru/api/sowa" SBER_HTTP_TIMEOUT = 30.0 _CHROME_UA = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/124.0.0.0 Safari/537.36" ) # --------------------------------------------------------------------------- # Route builder # --------------------------------------------------------------------------- def build_sber_route(dashboard: str, ref_area: str) -> str: """Build the base64-encoded route for the /api/sowa payload. Encodes /dataset/v1/?filter=&limit=1000&offset=0 where filter = {"REF_AREA": ref_area, "FREQ": "M", "SOURCE": "SI", "REALTY": "2"}. All filter values are strings per the live-captured payload (vault research note). """ filter_dict = { "REF_AREA": str(ref_area), "FREQ": "M", "SOURCE": "SI", "REALTY": "2", } # urlencode the JSON, then build the path filter_str = quote(json.dumps(filter_dict, separators=(",", ":")), safe="") path = f"/dataset/v1/{dashboard}?filter={filter_str}&limit=1000&offset=0" return base64.b64encode(path.encode("utf-8")).decode("ascii") # --------------------------------------------------------------------------- # Response decoder # --------------------------------------------------------------------------- # Field order returned by the API (confirmed from live session) _FIELD_ORDER = [ "indicator_id", "kpi_id", "period", "value", "obs_status", "source", "ref_area", "realty", "freq", "unit_measure", "unit_mult", ] def _decode_cell(cell: str) -> str | float: """Decode a single response cell. Cells arrive as one of: '__string__' → decode UTF-8 string '__number__' → parse as float """ if cell.startswith("__string__"): encoded = cell[len("__string__") :] return base64.b64decode(encoded).decode("utf-8") if cell.startswith("__number__"): return float(cell[len("__number__") :]) # Fallback: try numeric, else return raw try: return float(cell) except (ValueError, TypeError): return str(cell) def decode_sber_response( response_data: dict, # type: ignore[type-arg] ) -> list[dict[str, str | float]]: """Decode the SOWA response payload into a list of row dicts. Raises ValueError if the response structure is unexpected. """ try: sowa = response_data["SOWA"] values = sowa["data"]["value"] except (KeyError, TypeError) as exc: raise ValueError(f"Unexpected SOWA response structure: {exc}") from exc fields: list[str] | None = None rows_raw: list[list[str]] = [] for entry in values: key = entry.get("key") val = entry.get("value") if key == "fields": fields = [_decode_cell(c) for c in val] # type: ignore[arg-type] elif key == "data": rows_raw = val if fields is None: raise ValueError("No 'fields' key found in SOWA response") result: list[dict[str, str | float]] = [] for raw_row in rows_raw: if len(raw_row) != len(fields): logger.warning( "sber_index: row length mismatch (expected %d, got %d) — skip", len(fields), len(raw_row), ) continue row_dict: dict[str, str | float] = {} for col_name, cell in zip(fields, raw_row, strict=True): row_dict[col_name] = _decode_cell(cell) # type: ignore[arg-type] result.append(row_dict) return result # --------------------------------------------------------------------------- # Period normalisation # --------------------------------------------------------------------------- def _parse_period_month(period_str: str) -> date: """Parse the API period string into the first day of the month. API returns ISO-8601 with UTC offset, e.g. '2017-01-30T21:00:00Z'. We normalise to date(year, month, 1) — the first day of the reported month. The day/time portion is an artefact of timezone shift and is discarded. """ dt = datetime.fromisoformat(period_str.replace("Z", "+00:00")) # Shift to UTC date, then take first day of that month dt_utc = dt.astimezone(UTC) return date(dt_utc.year, dt_utc.month, 1) # --------------------------------------------------------------------------- # Async fetch — single dashboard × city # --------------------------------------------------------------------------- async def fetch_sber_index( client: httpx.AsyncClient, *, dashboard: str, ref_area: str, ) -> AsyncIterator[tuple[str, date, str, str, float]]: """Fetch one dashboard × REF_AREA combination from sberindex.ru/api/sowa. Yields tuples of: (city_label, period_month, segment_label, dashboard, index_value_rub_m2) city_label and segment_label come from the API response fields 'ref_area' and 'realty'. Raises httpx.HTTPStatusError on non-2xx response (caller handles per-series). """ route = build_sber_route(dashboard, ref_area) rquid = uuid.uuid4().hex # 32-hex random, required header payload = { "SOWA": { "method": "GET", "route": route, "data": {"type": "object", "value": []}, } } headers = { "content-type": "application/json", "rquid": rquid, "x-language": "ru", "accept": "application/json, text/plain, */*", "user-agent": _CHROME_UA, } resp = await client.post(SBER_API_URL, json=payload, headers=headers) resp.raise_for_status() rows = decode_sber_response(resp.json()) for row in rows: period_raw = row.get("period", "") value_raw = row.get("value") city_label = str(row.get("ref_area", ref_area)) segment_label = str(row.get("realty", "")) if not period_raw or value_raw is None: continue try: period_month = _parse_period_month(str(period_raw)) index_value = float(value_raw) except (ValueError, TypeError) as exc: logger.warning( "sber_index: skip row — parse error for dashboard=%s ref_area=%s: %s", dashboard, ref_area, exc, ) continue yield (city_label, period_month, segment_label, dashboard, index_value) # --------------------------------------------------------------------------- # Bulk pull + upsert # --------------------------------------------------------------------------- async def pull_sber_indices( db: Session, *, cities: dict[str, str] | None = None, dashboards: list[str] | None = None, ) -> dict[str, int]: """Fetch all configured dashboards × cities and upsert into sber_price_index. Args: db: SQLAlchemy Session (tradein DB). cities: REF_AREA code → city label mapping. Defaults to SBER_REF_AREAS. dashboards: list of dashboard slugs to pull. Defaults to SBER_DASHBOARDS. Returns dict of counters: {upserted: int, skipped: int, errors: int}. Per-series try/except: one failing series logs + continues (does not abort others). After upserting 'dinamika-tsen-obyavlenii', logs the latest asking index per city as a data-quality benchmark reference. """ if cities is None: cities = SBER_REF_AREAS if dashboards is None: dashboards = SBER_DASHBOARDS counters: dict[str, int] = {"upserted": 0, "skipped": 0, "errors": 0} async with httpx.AsyncClient(timeout=SBER_HTTP_TIMEOUT) as client: for ref_area, _city_hint in cities.items(): for dashboard in dashboards: try: rows_to_upsert: list[tuple[str, date, str, str, float]] = [] async for row in fetch_sber_index( client, dashboard=dashboard, ref_area=ref_area ): rows_to_upsert.append(row) if not rows_to_upsert: logger.info( "sber_index: no rows returned for dashboard=%s ref_area=%s", dashboard, ref_area, ) counters["skipped"] += 1 continue # Idempotent upsert — ON CONFLICT(city, period_month, dashboard) for city_label, period_month, segment, dash, value in rows_to_upsert: db.execute( text( """ INSERT INTO sber_price_index (city, period_month, segment, dashboard, index_value_rub_m2, source, fetched_at) VALUES ( CAST(:city AS text), CAST(:period_month AS date), CAST(:segment AS text), CAST(:dashboard AS text), CAST(:index_value AS double precision), 'sberindex', now() ) ON CONFLICT (city, period_month, dashboard) DO UPDATE SET index_value_rub_m2 = EXCLUDED.index_value_rub_m2, segment = EXCLUDED.segment, fetched_at = now() """ ), { "city": city_label, "period_month": period_month.isoformat(), "segment": segment, "dashboard": dash, "index_value": value, }, ) counters["upserted"] += 1 db.commit() logger.info( "sber_index: upserted %d rows for dashboard=%s ref_area=%s", len(rows_to_upsert), dashboard, ref_area, ) # Data-quality benchmark log for asking-price series if dashboard == "dinamika-tsen-obyavlenii" and rows_to_upsert: latest_row = max(rows_to_upsert, key=lambda r: r[1]) city_lbl, latest_month, _, _, latest_val = latest_row logger.info( "sber_index benchmark [asking]: city=%r latest_month=%s " "index_value_rub_m2=%.0f " "(data-quality ref; full asking-vs-our-median is follow-up)", city_lbl, latest_month.isoformat(), latest_val, ) except httpx.HTTPStatusError as exc: logger.error( "sber_index: HTTP %d for dashboard=%s ref_area=%s — skip series", exc.response.status_code, dashboard, ref_area, ) counters["errors"] += 1 except httpx.RequestError as exc: logger.error( "sber_index: network error for dashboard=%s ref_area=%s: %s — skip", dashboard, ref_area, exc, ) counters["errors"] += 1 except Exception: logger.exception( "sber_index: unexpected error for dashboard=%s ref_area=%s — skip", dashboard, ref_area, ) counters["errors"] += 1 return counters