403 lines
15 KiB
Python
403 lines
15 KiB
Python
"""СберИндекс 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": "<base64>", "data": {"type": "object", "value": []}}}
|
||
route = base64( /dataset/v1/<dashboard>?filter=<urlencode(json)>&limit=1000&offset=0 )
|
||
filter = {"REF_AREA": "<code>", "FREQ": "M", "SOURCE": "SI", "REALTY": "2"}
|
||
Required headers: content-type, rquid (random 32-hex), x-language, user-agent.
|
||
|
||
Response fields (each cell is __string__<b64> or __number__<raw>):
|
||
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/<dashboard>?filter=<urlencode(json)>&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__<base64>' → decode UTF-8 string
|
||
'__number__<raw>' → 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
|