lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

feat(tradein): СберИндекс monthly city-level price index pull (#887, data-layer) #890

Merged
bot-reviewer merged 1 commit from feat/887-sberindex-pull into main 2026-05-31 12:44:57 +00:00
Conversation 0 Commits 1 Files changed 6 +992
6 changed files with 992 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

405
tradein-mvp/backend/app/services/sber_index.py Normal file
View file

@ -0,0 +1,405 @@
"""СберИндекс 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:
643 = РФ (confirmed from live session, vault research note 2026-05-31-0845).
# TODO (#887 follow-up): resolve EKB (Свердловская обл) + Moscow REF_AREA codes.
# The dynpage config endpoint (GET /api/dynpage/ru/dashboards/<dashboard>) returns
# a base64-encoded React config with region options, but requires a working TLS chain
# from the deployment environment. Candidate codes: Свердловская обл ≈ 66,
# Москва ≈ 77 (ОКАТО-derived). Confirm by inspecting the dynpage form field
# «REF_AREA» values in a browser session or after fixing TLS in prod.
# Once confirmed, add them to SBER_REF_AREAS below and remove this TODO.
"""
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.
# 643 = РФ is confirmed (live reverse-engineering, vault 2026-05-31-0845).
# EKB and Moscow codes are pending TLS resolution — see TODO in module docstring.
SBER_REF_AREAS: dict[str, str] = {
"643": "Россия",
# TODO (#887 follow-up): uncomment + verify EKB / Moscow REF_AREA codes.
# "66": "Свердловская область", # candidate — confirm via dynpage form
# "77": "Москва", # candidate — confirm via dynpage form
}
# 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

42
tradein-mvp/backend/app/services/scheduler.py
View file

@ -28,6 +28,10 @@ Sources:
(tasks/deactivate_stale_avito.py, #759; marks avito listings
with last_seen_at > TTL days as is_active=false — no DELETE,
no external calls, enabled by default; window after avito sweep)
- sber_index_pull → run_sber_index_pull
(tasks/sber_index_pull.py, #887; monthly pull of СберИндекс
city-level secondary-housing price index — pure HTTP+DB, no
anti-bot/proxy; window 05:00-06:00 UTC, ~28-day cadence)
"""
from __future__ import annotations
@ -596,6 +600,42 @@ async def trigger_deactivate_stale_avito_run(
return run_id
async def trigger_sber_index_pull_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
"""Создать scrape_runs + launch run_sber_index_pull в asyncio.create_task.
Monthly pull (#887): fetches city-level secondary-housing price index from
sberindex.ru/api/sowa across three dashboards (real_estate_deals,
residential_real_estate_prices, dinamika-tsen-obyavlenii) and upserts into
sber_price_index. Pure HTTP+DB, no scraper proxy, no anti-bot.
Mirrors trigger_yandex_address_backfill_run: claim run → create_task → mark_done/failed
delegated to tasks/sber_index_pull.py.
Returns run_id или None (skip — already running).
"""
run_id = _claim_run(db, schedule_row)
if run_id is None:
return None
params = schedule_row.get("default_params") or {}
async def _run() -> None:
run_db = SessionLocal()
try:
from app.tasks.sber_index_pull import run_sber_index_pull
await run_sber_index_pull(run_db, run_id=run_id, params=params)
except Exception:
logger.exception("scheduler: run_sber_index_pull crashed run_id=%d", run_id)
finally:
run_db.close()
task = asyncio.create_task(_run())
task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
logger.info("scheduler: triggered sber_index_pull run_id=%d", run_id)
return run_id
async def trigger_asking_to_sold_ratio_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
"""Создать scrape_runs + launch recompute_asking_to_sold_ratios в executor (sync DB-only task).
@ -916,6 +956,8 @@ async def scheduler_loop() -> None:
await trigger_yandex_address_backfill_run(db, sch)
elif source == "deactivate_stale_avito":
await trigger_deactivate_stale_avito_run(db, sch)
elif source == "sber_index_pull":
await trigger_sber_index_pull_run(db, sch)
else:
logger.warning("scheduler: unknown source=%s, skip", source)
finally:

94
tradein-mvp/backend/app/tasks/sber_index_pull.py Normal file
View file

@ -0,0 +1,94 @@
"""Scheduled task: pull СберИндекс city-level secondary-housing price index (#887).
Delegates HTTP fetch + upsert to app.services.sber_index.pull_sber_indices().
Wired into the in-app scheduler as source='sber_index_pull'. Triggered monthly via
the scrape_schedules seed (migration 093).
Three dashboards fetched per configured city:
- real_estate_deals (hedonic sold-price index)
- residential_real_estate_prices (repeat-sales hedonic)
- dinamika-tsen-obyavlenii (asking-price benchmark)
NOTE: estimator time-adjust wire (step 4, #887) is OUT OF SCOPE for this module.
That is a client-visible change requiring a backtest gate — see issue #887 step 4
and the follow-up note in app/services/sber_index.py.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from sqlalchemy.orm import Session
from app.services import scrape_runs as runs_mod
from app.services.sber_index import pull_sber_indices
logger = logging.getLogger(__name__)
__all__ = [
"SberIndexPullResult",
"run_sber_index_pull",
]
@dataclass
class SberIndexPullResult:
"""Result counters from a scheduled СберИндекс pull run."""
upserted: int = 0
skipped: int = 0
errors: int = 0
duration_sec: float = field(default=0.0)
async def run_sber_index_pull(
db: Session,
*,
run_id: int,
params: dict, # type: ignore[type-arg]
) -> SberIndexPullResult:
"""Execute СберИндекс pull with run lifecycle management.
Wraps pull_sber_indices(), emitting heartbeat before the fetch and marking
the scrape_run as done/failed on completion.
Params (from default_params jsonb in scrape_schedules):
interval_days: int — informational only (scheduler uses next_run_at arithmetic).
"""
import time
counters: dict[str, int] = {
"upserted": 0,
"skipped": 0,
"errors": 0,
}
started_at = time.monotonic()
try:
runs_mod.update_heartbeat(db, run_id, counters)
result_counters = await pull_sber_indices(db)
counters = result_counters
duration_sec = time.monotonic() - started_at
runs_mod.mark_done(db, run_id, {**counters, "duration_sec": int(duration_sec)})
logger.info(
"sber_index_pull run_id=%d done — upserted=%d skipped=%d errors=%d %.1fs",
run_id,
counters["upserted"],
counters["skipped"],
counters["errors"],
duration_sec,
)
return SberIndexPullResult(
upserted=counters["upserted"],
skipped=counters["skipped"],
errors=counters["errors"],
duration_sec=duration_sec,
)
except Exception as exc:
logger.exception("sber_index_pull run_id=%d failed", run_id)
runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
raise

70
tradein-mvp/backend/data/sql/092_sber_price_index.sql Normal file
View file

@ -0,0 +1,70 @@
-- 092_sber_price_index.sql
-- Issue #887 — city-level secondary-housing price index from СберИндекс (sberindex.ru).
--
-- Context:
-- СберИндекс publishes monthly city/region-level secondary-housing price statistics
-- across three dashboards:
-- * real_estate_deals — hedonic sold-price index (руб/м²)
-- * residential_real_estate_prices — repeat-sales / asking-adjusted hedonic
-- * dinamika-tsen-obyavlenii — asking-price benchmark (объявления, руб/м²)
--
-- The pull service (app/services/sber_index.py) fetches all three series for configured
-- cities (initially РФ-643, with EKB/MSK codes TBD — see TODO in sber_index.py) via the
-- /api/sowa POST endpoint (reverse-engineered in #794 research note) and upserts rows here.
--
-- The table feeds the estimator time-adjust coefficient (step 4, #887 out-of-scope —
-- backtest-gated, separate follow-up).
--
-- Schema notes:
-- * city — REF_AREA label as returned by the API (e.g. «Россия», «Екатеринбург»).
-- * period_month — first day of the reported month (normalised from API timestamp).
-- * segment — realty type label from API (e.g. «Вторичный»).
-- * dashboard — slug used in the route (e.g. 'real_estate_deals').
-- * PRIMARY KEY (city, period_month, dashboard) — one value per city×month×series;
-- upsert on conflict keeps the freshest fetched_at.
--
-- Idempotent: CREATE TABLE IF NOT EXISTS — safe to re-apply. ADDITIVE only, no drops.
--
-- Dependencies: none (standalone table).
--
-- Deploy order: apply before deploying app/services/sber_index.py +
-- app/tasks/sber_index_pull.py + scheduler.py changes.
BEGIN;
CREATE TABLE IF NOT EXISTS sber_price_index (
city text NOT NULL,
period_month date NOT NULL,
segment text NOT NULL DEFAULT '',
dashboard text NOT NULL,
index_value_rub_m2 double precision NOT NULL,
source text NOT NULL DEFAULT 'sberindex',
fetched_at timestamptz NOT NULL DEFAULT now(),
PRIMARY KEY (city, period_month, dashboard)
);
COMMENT ON TABLE sber_price_index IS
'Monthly city-level secondary-housing price index from СберИндекс (sberindex.ru). '
'Three series: real_estate_deals (hedonic sold), residential_real_estate_prices '
'(repeat-sales hedonic), dinamika-tsen-obyavlenii (asking benchmark). '
'Populated by app/tasks/sber_index_pull.py via in-app scheduler (#887). '
'Estimator time-adjust wire is a separate backtest-gated follow-up (step 4, #887).';
COMMENT ON COLUMN sber_price_index.city IS
'REF_AREA label from the API response (e.g. «Россия», «Екатеринбург»).';
COMMENT ON COLUMN sber_price_index.period_month IS
'First day of the reported month (normalised from API ISO timestamp).';
COMMENT ON COLUMN sber_price_index.segment IS
'Realty type label from API (e.g. «Вторичный»). Empty string when absent.';
COMMENT ON COLUMN sber_price_index.dashboard IS
'Dataset slug used in the /dataset/v1/<dashboard> route.';
COMMENT ON COLUMN sber_price_index.index_value_rub_m2 IS
'Price index in rubles per square metre.';
COMMENT ON COLUMN sber_price_index.fetched_at IS
'Wall-clock timestamp when this row was last written/updated.';
-- Index for time-series queries (latest N months per city × dashboard)
CREATE INDEX IF NOT EXISTS idx_sber_price_index_city_dashboard_period
ON sber_price_index (city, dashboard, period_month DESC);
COMMIT;

73
tradein-mvp/backend/data/sql/093_scrape_schedules_seed_sber_index_pull.sql Normal file
View file

@ -0,0 +1,73 @@
-- 093_scrape_schedules_seed_sber_index_pull.sql
-- Issue #887 — seed scrape_schedules row for monthly СберИндекс price-index pull.
--
-- Context:
-- app/services/sber_index.py fetches city-level secondary-housing price statistics
-- from sberindex.ru/api/sowa (three dashboards: real_estate_deals,
-- residential_real_estate_prices, dinamika-tsen-obyavlenii) and upserts rows into
-- sber_price_index (created in migration 092).
--
-- The pull is a pure HTTP + DB operation — no scraper proxy, no anti-bot, no cookies.
-- The /api/sowa endpoint requires only the rquid header (32-hex UUID) and a
-- content-type: application/json header. Rate is low (3 series × 3 cities = 9 requests
-- per run, monthly data since 2017 — ~100 rows per series per city at most).
--
-- Fix:
-- 1. services/sber_index.py — fetch_sber_index() + pull_sber_indices() service.
-- 2. tasks/sber_index_pull.py — run_sber_index_pull() task wrapper.
-- 3. scheduler.py — trigger_sber_index_pull_run() + dispatch branch
-- `elif source == "sber_index_pull"`.
-- 4. This migration — seeds the scrape_schedules row (enabled=true, monthly window).
--
-- Schedule window 05:00-06:00 UTC:
-- Monthly cadence — runs roughly once a month. Placed after rosreestr_dkp_import
-- (04:00-06:00 UTC window) so both windows overlap in the early-morning batch slot.
-- The index is updated by СберИндекс once a month; daily pulls would be wasteful.
-- default_params.interval_days=28 mirrors the avito / yandex monthly approach
-- (scheduler re-computes next_run_at after each fire; 28 days ≈ monthly).
--
-- next_run_at bootstrapped to tomorrow 05:00 UTC — scheduler will not fire immediately
-- on deploy (same pattern as 078/079/082/088/091).
--
-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply.
--
-- Dependencies:
-- 052_scrape_schedules.sql (table + UNIQUE(source)).
-- 092_sber_price_index.sql (sber_price_index table must exist).
-- scheduler.py change (trigger_sber_index_pull_run must be deployed).
-- services/sber_index.py + tasks/sber_index_pull.py must be deployed.
--
-- Deploy order:
-- Apply after deploying the scheduler.py + task + service changes so the scheduler
-- can dispatch 'sber_index_pull' correctly on first fire.
BEGIN;
INSERT INTO scrape_schedules (
source,
enabled,
window_start_hour,
window_end_hour,
next_run_at,
default_params
)
VALUES
(
'sber_index_pull',
true, -- SAFE: pure HTTP+DB, no anti-bot, low rate
5, -- window 05:00-06:00 UTC
6,
((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 5)) AT TIME ZONE 'UTC',
'{"interval_days": 28}'::jsonb -- monthly cadence (~28 days between runs)
)
ON CONFLICT (source) DO NOTHING;
COMMENT ON TABLE scrape_schedules IS
'In-app scheduler config (заменяет cron-script setup). '
'Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), '
'cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), '
'asking_to_sold_ratio_refresh (#648), refresh_search_matview (#769), '
'yandex_address_backfill (#855, EKB pilot), '
'sber_index_pull (#887, monthly СберИндекс city-level price index).';
COMMIT;

308
tradein-mvp/backend/tests/test_sber_index.py Normal file
View file

@ -0,0 +1,308 @@
"""Tests for sber_index service — pure unit tests, no network/DB.
Coverage:
(a) build_sber_route: base64 route builder produces correct encoded string.
(b) decode_sber_response: decodes a sample base64 payload into correct rows.
(c) pull_sber_indices upsert SQL is idempotent (ON CONFLICT shape assertion via
fake-db roundtrip — verifies upsert is called with correct SQL and params).
"""
from __future__ import annotations
import base64
import json
import os
import sys
from datetime import date
from unittest.mock import MagicMock, patch
from urllib.parse import unquote
# DATABASE_URL required by config before any app import.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
# WeasyPrint stub — not installed in CI without GTK.
_wp_mock = MagicMock()
sys.modules.setdefault("weasyprint", _wp_mock)
import pytest # noqa: E402
from app.services.sber_index import ( # noqa: E402
_parse_period_month,
build_sber_route,
decode_sber_response,
)
# ---------------------------------------------------------------------------
# (a) Route builder
# ---------------------------------------------------------------------------
def test_build_sber_route_rф_real_estate_deals() -> None:
"""Route for РФ (643) + real_estate_deals should decode to expected path."""
encoded = build_sber_route("real_estate_deals", "643")
decoded = base64.b64decode(encoded).decode("utf-8")
assert decoded.startswith("/dataset/v1/real_estate_deals?filter=")
assert "limit=1000" in decoded
assert "offset=0" in decoded
# Extract filter JSON from decoded path
filter_part = decoded.split("filter=")[1].split("&")[0]
filter_json = json.loads(unquote(filter_part))
assert filter_json["REF_AREA"] == "643"
assert filter_json["FREQ"] == "M"
assert filter_json["SOURCE"] == "SI"
assert filter_json["REALTY"] == "2"
def test_build_sber_route_different_dashboard() -> None:
"""Route slug is correctly embedded in the decoded path."""
encoded = build_sber_route("dinamika-tsen-obyavlenii", "643")
decoded = base64.b64decode(encoded).decode("utf-8")
assert "/dataset/v1/dinamika-tsen-obyavlenii?" in decoded
def test_build_sber_route_is_valid_base64() -> None:
"""Encoded route must be decodeable ASCII base64."""
encoded = build_sber_route("residential_real_estate_prices", "77")
# Should not raise
decoded_bytes = base64.b64decode(encoded)
assert len(decoded_bytes) > 0
def test_build_sber_route_ref_area_embedded() -> None:
"""REF_AREA value is preserved exactly in the filter JSON."""
encoded = build_sber_route("real_estate_deals", "66")
decoded = base64.b64decode(encoded).decode("utf-8")
filter_part = decoded.split("filter=")[1].split("&")[0]
filter_json = json.loads(unquote(filter_part))
assert filter_json["REF_AREA"] == "66"
# ---------------------------------------------------------------------------
# (b) Response decoder
# ---------------------------------------------------------------------------
def _make_b64_cell(value: str) -> str:
return "__string__" + base64.b64encode(value.encode("utf-8")).decode("ascii")
def _make_num_cell(value: float) -> str:
return f"__number__{value}"
def _make_sowa_response(rows: list[list[str]]) -> dict: # type: ignore[type-arg]
"""Build a minimal SOWA response dict with a fixed field list."""
fields = [
"indicator_id",
"kpi_id",
"period",
"value",
"obs_status",
"source",
"ref_area",
"realty",
"freq",
"unit_measure",
"unit_mult",
]
encoded_fields = [_make_b64_cell(f) for f in fields]
return {
"SOWA": {
"method": "GET",
"route": "test_route",
"data": {
"type": "object",
"value": [
{"key": "fields", "value": encoded_fields},
{"key": "data", "value": rows},
],
},
}
}
def test_decode_sber_response_basic() -> None:
"""Decode a two-row response and check period + value extraction."""
row1 = [
_make_b64_cell("1"), # indicator_id
_make_num_cell(643032001), # kpi_id
_make_b64_cell("2017-01-30T21:00:00Z"), # period
_make_num_cell(48549), # value
_make_b64_cell("A"), # obs_status
_make_b64_cell("Данные СберИндекса"), # source
_make_b64_cell("Россия"), # ref_area
_make_b64_cell("Вторичный"), # realty
_make_b64_cell("Месяц"), # freq
_make_b64_cell("руб"), # unit_measure
_make_num_cell(0), # unit_mult
]
row2 = [
_make_b64_cell("1"),
_make_num_cell(643032001),
_make_b64_cell("2024-06-30T21:00:00Z"),
_make_num_cell(120000),
_make_b64_cell("A"),
_make_b64_cell("Данные СберИндекса"),
_make_b64_cell("Россия"),
_make_b64_cell("Вторичный"),
_make_b64_cell("Месяц"),
_make_b64_cell("руб"),
_make_num_cell(0),
]
payload = _make_sowa_response([row1, row2])
decoded = decode_sber_response(payload)
assert len(decoded) == 2
assert decoded[0]["ref_area"] == "Россия"
assert decoded[0]["value"] == 48549.0
assert decoded[0]["period"] == "2017-01-30T21:00:00Z"
assert decoded[1]["value"] == 120000.0
def test_decode_sber_response_period_normalises_to_first_of_month() -> None:
"""_parse_period_month always returns the first day of the month."""
result = _parse_period_month("2017-01-30T21:00:00Z")
assert result == date(2017, 1, 1)
result2 = _parse_period_month("2024-06-30T21:00:00Z")
assert result2 == date(2024, 6, 1)
def test_decode_sber_response_empty_data() -> None:
"""Empty data section returns empty list."""
payload = _make_sowa_response([])
decoded = decode_sber_response(payload)
assert decoded == []
def test_decode_sber_response_missing_fields_key_raises() -> None:
"""Response without 'fields' key raises ValueError."""
bad_payload = {
"SOWA": {
"method": "GET",
"route": "x",
"data": {"type": "object", "value": [{"key": "data", "value": []}]},
}
}
with pytest.raises(ValueError, match="fields"):
decode_sber_response(bad_payload)
# ---------------------------------------------------------------------------
# (c) Upsert SQL idempotency — fake-db roundtrip
# ---------------------------------------------------------------------------
class _ExecuteCapture:
"""Captures execute() calls and their params; simulate no-op DB."""
def __init__(self) -> None:
self.calls: list[tuple[str, dict]] = [] # type: ignore[type-arg]
def execute(self, stmt: object, params: dict | None = None) -> MagicMock: # type: ignore[return-value]
sql_str = str(stmt)
self.calls.append((sql_str, params or {}))
return MagicMock(fetchone=lambda: None, fetchall=lambda: [])
def commit(self) -> None:
pass
@pytest.mark.asyncio
async def test_pull_sber_indices_upsert_on_conflict_idempotent() -> None:
"""pull_sber_indices emits ON CONFLICT ... DO UPDATE upsert SQL (idempotent)."""
from app.services.sber_index import pull_sber_indices
fake_db = _ExecuteCapture()
fake_row = ("Россия", date(2024, 1, 1), "Вторичный", "real_estate_deals", 95000.0)
# Patch fetch_sber_index to yield one deterministic row per call
async def _mock_fetch(client: object, *, dashboard: str, ref_area: str): # type: ignore[override]
yield fake_row
with patch("app.services.sber_index.fetch_sber_index", side_effect=_mock_fetch):
# Only pull one dashboard × one city to keep assertion simple
result = await pull_sber_indices(
fake_db, # type: ignore[arg-type]
cities={"643": "Россия"},
dashboards=["real_estate_deals"],
)
assert result["upserted"] == 1
assert result["errors"] == 0
# Find the INSERT call
insert_calls = [sql for sql, _ in fake_db.calls if "INSERT INTO sber_price_index" in sql]
assert insert_calls, "Expected at least one INSERT INTO sber_price_index call"
upsert_sql = insert_calls[0]
# Must contain ON CONFLICT clause with DO UPDATE SET
assert "ON CONFLICT" in upsert_sql
assert "DO UPDATE SET" in upsert_sql
assert "index_value_rub_m2" in upsert_sql
assert "fetched_at" in upsert_sql
# Must NOT contain :: type casts (psycopg v3 rule)
import re
assert not re.search(
r":[a-z_]+::[a-z]", upsert_sql
), "SQL must not contain ::type casts — use CAST(... AS type) instead"
@pytest.mark.asyncio
async def test_pull_sber_indices_error_per_series_continues() -> None:
"""A failing series increments errors but does not abort other series."""
from app.services.sber_index import pull_sber_indices
fake_db = _ExecuteCapture()
call_count = 0
async def _mock_fetch_with_error(client: object, *, dashboard: str, ref_area: str): # type: ignore[override]
nonlocal call_count
call_count += 1
if dashboard == "real_estate_deals":
raise Exception("Simulated network failure")
yield ("Россия", date(2024, 1, 1), "Вторичный", dashboard, 90000.0)
with patch("app.services.sber_index.fetch_sber_index", side_effect=_mock_fetch_with_error):
result = await pull_sber_indices(
fake_db, # type: ignore[arg-type]
cities={"643": "Россия"},
dashboards=["real_estate_deals", "residential_real_estate_prices"],
)
# First series failed, second succeeded
assert result["errors"] == 1
assert result["upserted"] == 1
@pytest.mark.asyncio
async def test_pull_sber_indices_asking_benchmark_logged(caplog: pytest.LogCaptureFixture) -> None:
"""After upserting dinamika-tsen-obyavlenii, a benchmark INFO line is logged."""
import logging
from app.services.sber_index import pull_sber_indices
fake_db = _ExecuteCapture()
async def _mock_fetch(client: object, *, dashboard: str, ref_area: str): # type: ignore[override]
yield ("Россия", date(2024, 6, 1), "Вторичный", dashboard, 115000.0)
with patch("app.services.sber_index.fetch_sber_index", side_effect=_mock_fetch):
with caplog.at_level(logging.INFO, logger="app.services.sber_index"):
await pull_sber_indices(
fake_db, # type: ignore[arg-type]
cities={"643": "Россия"},
dashboards=["dinamika-tsen-obyavlenii"],
)
assert any(
"benchmark" in record.message for record in caplog.records
), "Expected a benchmark log line after upserting dinamika-tsen-obyavlenii"
assert any(
"115000" in record.message for record in caplog.records
), "Benchmark log should include the latest asking index value"