feat: backfill external signals

agents/services.md

@@ -14,6 +14,10 @@ domain rules live — they're called by routes and call the repository.
 - `backend/foresight/ingest/` — external-source adapters (events, weather,
   flights). Part of the services layer: they fetch and normalize, then write
   through `repository/`.
+- `backend/foresight/services/backfill/` — fetch an external source over a date
+  range and save it. Long runs yield after each unit of work (a month of events,
+  a day of flights) so the CLIs in `backend/scripts/backfill/` can commit as
+  they go.
 
 ## Patterns & conventions
 - Pure-ish domain functions/classes; take plain inputs or `*Request` schemas,

backend/foresight/ingest/predicthq.py

@@ -98,7 +98,7 @@ async def fetch_events(
                 except (KeyError, TypeError, ValueError) as exc:
                     ref = str(raw.get("id", "?")) if isinstance(raw, dict) else "?"
                     log.warning("rejected PredictHQ event %s: %s", ref, exc)
-                    result.rejected.append(Rejected(ref, _reason(exc)))
+                    result.rejected.append(Rejected(ref, rejection_reason(exc)))
 
             # `next` already carries every query parameter.
             url, query = body.get("next"), None
@@ -156,7 +156,7 @@ def _venue(entities: list[Mapping[str, Any]]) -> str | None:
     return next((e.get("name") for e in entities if e.get("type") == "venue"), None)
 
 
-def _reason(exc: Exception) -> str:
+def rejection_reason(exc: Exception) -> str:
     if isinstance(exc, KeyError):
         return f"missing field {exc.args[0]!r}"
     return str(exc)

backend/foresight/ingest/run.py

@@ -160,7 +160,7 @@ def _parser() -> argparse.ArgumentParser:
     return parser
 
 
-async def _resolve_market(session: AsyncSession, args: argparse.Namespace) -> Market:
+async def resolve_market(session: AsyncSession, args: argparse.Namespace) -> Market:
     if args.hotel_id is None:
         return Market(city=args.city, latitude=args.lat, longitude=args.lon)
 
@@ -172,7 +172,7 @@ async def _resolve_market(session: AsyncSession, args: argparse.Namespace) -> Ma
     return Market(city=hotel.city, latitude=hotel.latitude, longitude=hotel.longitude)
 
 
-async def _market_totals(
+async def market_totals(
     session: AsyncSession, city: str
 ) -> tuple[dict[str, int], dict[str, int]]:
     slug = slugify_city(city)
@@ -195,7 +195,7 @@ async def _market_totals(
 
 async def run(args: argparse.Namespace) -> RunReport:
     async with async_session_factory() as session:
-        market = await _resolve_market(session, args)
+        market = await resolve_market(session, args)
         report = RunReport(
             source=Source(args.source),
             market=market,
@@ -223,13 +223,13 @@ async def run(args: argparse.Namespace) -> RunReport:
             report.outcomes[outcome] = report.outcomes.get(outcome, 0) + 1
         await session.commit()
 
-        report.events_by_status, report.observations_by_source = await _market_totals(
+        report.events_by_status, report.observations_by_source = await market_totals(
             session, market.city
         )
     return report
 
 
-def _migrate() -> None:
+def migrate() -> None:
     """Bring the schema up to date, so a fresh pull needs no manual step.
 
     A no-op when already at head. No ini file is passed, so Alembic leaves this
@@ -258,7 +258,7 @@ def main(argv: Sequence[str] | None = None) -> int:
 
     logging.basicConfig(level=logging.WARNING, format="%(levelname)s %(message)s")
     if not (args.dry_run and args.hotel_id is None):
-        _migrate()
+        migrate()
     # DEBUG=true turns on SQL echo, which would bury the report.
     engine.echo = False
     asyncio.run(_main(args))

backend/foresight/ingest/weather.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from datetime import timedelta
 from typing import TYPE_CHECKING, Any
 
 import openmeteo_requests
@@ -8,11 +9,14 @@
 from retry_requests import retry
 
 if TYPE_CHECKING:
+    from collections.abc import Iterator
     from datetime import date
 
 FORECAST_URL = "https://api.open-meteo.com/v1/forecast"
 ARCHIVE_URL = "https://archive-api.open-meteo.com/v1/archive"
 SOURCE = "open-meteo"
+# A year per archive request keeps each call small; a failure only redoes that year.
+CHUNK_DAYS = 365
 
 # Order matters: the response exposes variables by index, not by name.
 _DAILY = (
@@ -141,15 +145,26 @@ def fetch_actuals(
     outright, so a caller backfilling a date range has to clamp `end` to
     today rather than relying on empty rows coming back.
     """
-    response = _get_client().weather_api(
-        ARCHIVE_URL,
-        params={
-            "latitude": latitude,
-            "longitude": longitude,
-            "daily": list(_DAILY),
-            "timezone": timezone,
-            "start_date": start.isoformat(),
-            "end_date": end.isoformat(),
-        },
-    )[0]
-    return _rows(response, city, is_forecast=False)
+    rows: list[dict[str, Any]] = []
+    for chunk_start, chunk_end in chunks(start, end):
+        response = _get_client().weather_api(
+            ARCHIVE_URL,
+            params={
+                "latitude": latitude,
+                "longitude": longitude,
+                "daily": list(_DAILY),
+                "timezone": timezone,
+                "start_date": chunk_start.isoformat(),
+                "end_date": chunk_end.isoformat(),
+            },
+        )[0]
+        rows += _rows(response, city, is_forecast=False)
+    return rows
+
+
+def chunks(start: date, end: date) -> Iterator[tuple[date, date]]:
+    """Split `start`..`end` into back-to-back spans of at most CHUNK_DAYS days."""
+    while start <= end:
+        chunk_end = min(start + timedelta(days=CHUNK_DAYS - 1), end)
+        yield start, chunk_end
+        start = chunk_end + timedelta(days=1)

backend/foresight/repository/__init__.py

@@ -19,6 +19,9 @@
     from sqlalchemy import Table
     from sqlalchemy.ext.asyncio import AsyncSession
 
+# Postgres allows at most 32,767 bind parameters in one statement.
+_MAX_PARAMS = 32_767
+
 
 class Outcome(StrEnum):
     NEW = "new"
@@ -46,6 +49,26 @@ async def upsert_rows(
     if not rows:
         return counts
 
+    # One statement per batch, sized to stay under the parameter limit.
+    batch = max(1, _MAX_PARAMS // len(rows[0]))
+    for i in range(0, len(rows), batch):
+        for outcome, n in (
+            await _upsert_batch(
+                session, table, constraint, mutable, rows[i : i + batch]
+            )
+        ).items():
+            counts[outcome] += n
+    return counts
+
+
+async def _upsert_batch(
+    session: AsyncSession,
+    table: Table,
+    constraint: str,
+    mutable: Sequence[str],
+    rows: list[Mapping[str, Any]],
+) -> dict[Outcome, int]:
+    counts = dict.fromkeys(Outcome, 0)
     stmt = insert(table).values(rows)
     excluded = stmt.excluded
     set_: dict[str, Any] = {c: excluded[c] for c in mutable}

backend/foresight/repository/hotel.py

@@ -174,3 +174,20 @@ async def list_performance(
         stmt = stmt.where(DailyPerformance.stay_date <= end)
     stmt = stmt.order_by(DailyPerformance.stay_date, DailyPerformance.room_type_id)
     return await fetch_page(session, stmt, limit=limit, offset=offset)
+
+
+async def stay_date_range(
+    session: AsyncSession, hotel_id: int
+) -> tuple[date, date] | None:
+    """First and last stay_date loaded for a hotel, or None if it has no nights."""
+    first, last = (
+        await session.execute(
+            select(
+                func.min(DailyPerformance.stay_date),
+                func.max(DailyPerformance.stay_date),
+            )
+            .join(RoomType, DailyPerformance.room_type_id == RoomType.id)
+            .where(RoomType.hotel_id == hotel_id)
+        )
+    ).one()
+    return None if first is None else (first, last)

backend/foresight/repository/weather.py

@@ -2,11 +2,14 @@
 
 from typing import TYPE_CHECKING, Any, cast
 
+from sqlalchemy import select
+
 from foresight.database.models.external import Weather
 from foresight.repository import Outcome, upsert_rows
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Mapping
+    from datetime import date
 
     from sqlalchemy import Table
     from sqlalchemy.ext.asyncio import AsyncSession
@@ -27,3 +30,17 @@ async def upsert_weather(
         mutable=_MEASURES,
         rows=rows,
     )
+
+
+async def recorded_days(
+    session: AsyncSession, *, city: str, start: date, end: date
+) -> dict[date, bool]:
+    """Days with recorded weather saved, mapped to whether it has a temperature."""
+    result = await session.execute(
+        select(Weather.weather_date, Weather.temp_max_c.is_not(None)).where(
+            Weather.city == city,
+            Weather.is_forecast.is_(False),
+            Weather.weather_date.between(start, end),
+        )
+    )
+    return {day: has_values for day, has_values in result.tuples().all()}