bot-backend
19085547bc
Data-layer impl of #887 (steps 1-3+5; client-visible estimator time-adjust step 4 deferred to backtest-gated follow-up — estimator.py untouched). - migration 092: sber_price_index table (additive, CREATE TABLE IF NOT EXISTS) - migration 093: scrape_schedules seed (source=sber_index_pull, ON CONFLICT DO NOTHING) - sber_index.py: base64 /api/sowa route builder + decoder + async fetch + idempotent upsert ON CONFLICT (city,period_month,dashboard) - task wrapper + scheduler trigger + dispatch branch (monthly); 3 dashboards (deals/hedonic/asking) REALTY=2; per-series graceful errors; asking benchmark log NOTE: REF_AREA codes EKB/MSK unresolved (sberindex TLS unreachable from CI; 643=РФ only) — TODO in SBER_REF_AREAS. 20 tests pass. Refs #887, #794, #652
94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
"""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
|