From 11115c15d9010522a354e4c41f517365935372c7 Mon Sep 17 00:00:00 2001 From: Maria Samos Rivas Date: Tue, 13 Oct 2026 14:18:54 -0400 Subject: [PATCH 1/4] feat: batch upsert_rows under the Postgres parameter limit Long backfills sent every row in one statement, which fails past 32,767 bind parameters. Rows are now written in batches sized to stay under it. Co-authored-by: Vichu Selvaraju --- backend/foresight/repository/__init__.py | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/backend/foresight/repository/__init__.py b/backend/foresight/repository/__init__.py index 594e78c..4580b0d 100644 --- a/backend/foresight/repository/__init__.py +++ b/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} From 44f48d3fb5dcf8cc3f5f6df6c3b79a175e562c44 Mon Sep 17 00:00:00 2001 From: Maria Samos Rivas Date: Tue, 13 Oct 2026 14:19:56 -0400 Subject: [PATCH 2/4] feat: retry and pace AeroDataBox and chunk weather archive requests AeroDataBox requests are paced, retry 429/5xx/network errors with backoff, and raise RateLimitedError once the quota is spent; delayed flights from the previous day are skipped. Weather archive requests go a year at a time so a failure only redoes that year. Co-authored-by: Vichu Selvaraju --- backend/foresight/ingest/weather.py | 39 +++++++--- .../foresight/services/flight_data_service.py | 59 ++++++++++++-- backend/tests/test_flight_data_service.py | 78 +++++++++++++++++++ backend/tests/test_weather_ingest.py | 16 +++- 4 files changed, 171 insertions(+), 21 deletions(-) create mode 100644 backend/tests/test_flight_data_service.py diff --git a/backend/foresight/ingest/weather.py b/backend/foresight/ingest/weather.py index 4af6485..66d9003 100644 --- a/backend/foresight/ingest/weather.py +++ b/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) diff --git a/backend/foresight/services/flight_data_service.py b/backend/foresight/services/flight_data_service.py index 1c2379a..4bd52c9 100644 --- a/backend/foresight/services/flight_data_service.py +++ b/backend/foresight/services/flight_data_service.py @@ -15,6 +15,8 @@ import io from datetime import date, datetime, time, timedelta from pathlib import Path +from time import sleep +from typing import Any from zoneinfo import ZoneInfo import httpx @@ -29,6 +31,16 @@ OPENSKY_TOKEN_URL = "https://auth.opensky-network.org/auth/realms/opensky-network/protocol/openid-connect/token" AERODATABOX_HOST = "aerodatabox.p.rapidapi.com" DUBLIN = ZoneInfo("Europe/Dublin") +# RapidAPI's basic plan allows about one request a second. +AERODATABOX_PACE_SECONDS = 1.1 +# A 429 is often just "too fast": wait 2s, 4s, 8s before treating it as the quota. +AERODATABOX_MAX_RETRIES = 3 +AERODATABOX_BACKOFF_SECONDS = 2.0 +_RETRY_STATUS = frozenset({429, 500, 502, 503, 504}) + + +class RateLimitedError(Exception): + """AeroDataBox said 429: too many requests, or the monthly quota is used up.""" def _days(start: date, end: date) -> list[date]: @@ -140,6 +152,31 @@ def fetch_opensky(start: date, end: date, airport: str = "EIDW") -> pd.DataFrame return df +def _get_aerodatabox(url: str, **kwargs: Any) -> httpx.Response: + """GET from AeroDataBox, retrying rate limits, 5xx and network errors. + + Raises: + RateLimitedError: If it still answers 429 after every retry. + httpx.HTTPStatusError: On any other failed status. + """ + for attempt in range(AERODATABOX_MAX_RETRIES): + try: + response = httpx.get(url, **kwargs) + except httpx.TransportError: + pass # network blip or timeout: wait and try again + else: + if response.status_code not in _RETRY_STATUS: + response.raise_for_status() + return response + sleep(AERODATABOX_BACKOFF_SECONDS * 2**attempt) + # Last try: a 429 now most likely means the monthly quota is used up. + response = httpx.get(url, **kwargs) + if response.status_code == 429: + raise RateLimitedError(response.text[:300]) + response.raise_for_status() + return response + + def fetch_aerodatabox(start: date, end: date, airport: str = "DUB") -> pd.DataFrame: """Fetch scheduled flights arriving at an airport from AeroDataBox (via RapidAPI). @@ -151,13 +188,15 @@ def fetch_aerodatabox(start: date, end: date, airport: str = "DUB") -> pd.DataFr airport (str): IATA airport code. Defaults to "DUB" (Dublin). Returns: - pandas.DataFrame: One row per flight, with columns ``flight_number``, - ``airline``, ``aircraft_model``, ``origin_iata``, ``origin_name``, - ``origin_country``, ``status`` and ``arrival_time`` (scheduled, in Dublin - time). Empty if there were no flights. + pandas.DataFrame: One row per flight scheduled in the range, with columns + ``flight_number``, ``airline``, ``aircraft_model``, ``origin_iata``, + ``origin_name``, ``origin_country``, ``status`` and ``arrival_time`` + (scheduled, in Dublin time). Empty if there were no flights. Raises: - httpx.HTTPStatusError: If a request fails. + RateLimitedError: If AeroDataBox still answers 429 after retrying + (usually the monthly quota). + httpx.HTTPStatusError: If a request fails otherwise. """ headers = { "X-RapidAPI-Key": settings.xrapidapi_key, @@ -178,19 +217,23 @@ def fetch_aerodatabox(start: date, end: date, airport: str = "DUB") -> pd.DataFr midnight = datetime.combine(day, time.min) for window_start in (midnight, midnight + timedelta(hours=12)): window_end = window_start + timedelta(hours=11, minutes=59) - response = httpx.get( + sleep(AERODATABOX_PACE_SECONDS) + response = _get_aerodatabox( f"https://{AERODATABOX_HOST}/flights/airports/iata/{airport}" f"/{window_start:%Y-%m-%dT%H:%M}/{window_end:%Y-%m-%dT%H:%M}", params=params, headers=headers, timeout=30, ) - response.raise_for_status() if response.status_code == 204: # no flights in this window continue for item in response.json().get("arrivals", []): move = item.get("movement") or {} origin = move.get("airport") or {} + scheduled = move.get("scheduledTime") or {} + # Skip delayed flights that were scheduled for the day before. + if not scheduled.get("local", "").startswith(day.isoformat()): + continue rows.append( { "flight_number": item.get("number"), @@ -200,7 +243,7 @@ def fetch_aerodatabox(start: date, end: date, airport: str = "DUB") -> pd.DataFr "origin_name": origin.get("name"), "origin_country": origin.get("countryCode"), "status": item.get("status"), - "arrival_time": (move.get("scheduledTime") or {}).get("utc"), + "arrival_time": scheduled.get("utc"), } ) diff --git a/backend/tests/test_flight_data_service.py b/backend/tests/test_flight_data_service.py new file mode 100644 index 0000000..b5793dc --- /dev/null +++ b/backend/tests/test_flight_data_service.py @@ -0,0 +1,78 @@ +from datetime import date +from typing import Any + +import httpx +import pytest + +from foresight.services import flight_data_service +from foresight.services.flight_data_service import ( + RateLimitedError, + fetch_aerodatabox, +) + +JUNE_1 = date(2026, 6, 1) + + +def _arrival(number: str, local: str) -> dict[str, Any]: + return { + "number": number, + "movement": { + "airport": {"iata": "LHR"}, + "scheduledTime": {"local": local, "utc": f"{local[:16]}Z"}, + }, + } + + +def _fake_api(monkeypatch: pytest.MonkeyPatch, *responses: httpx.Response) -> list[str]: + """Answer each request with the next response; returns the URLs asked for.""" + urls, queue = [], list(responses) + + def get(url: str, **kwargs: Any) -> httpx.Response: + urls.append(url) + response = queue.pop(0) + response.request = httpx.Request("GET", url) + return response + + monkeypatch.setattr(flight_data_service.httpx, "get", get) + monkeypatch.setattr(flight_data_service, "sleep", lambda seconds: None) + return urls + + +def test_fetches_both_halves_of_the_day(monkeypatch: pytest.MonkeyPatch) -> None: + urls = _fake_api(monkeypatch, httpx.Response(204), httpx.Response(204)) + + assert fetch_aerodatabox(JUNE_1, JUNE_1).empty + assert [url.split("/DUB/")[1] for url in urls] == [ + "2026-06-01T00:00/2026-06-01T11:59", + "2026-06-01T12:00/2026-06-01T23:59", + ] + + +def test_skips_duplicates_and_the_day_before(monkeypatch: pytest.MonkeyPatch) -> None: + ei1 = _arrival("EI 1", "2026-06-01 10:05+01:00") + late = _arrival("FR 2", "2026-05-31 23:50+01:00") + _fake_api( + monkeypatch, + httpx.Response(200, json={"arrivals": [ei1, late]}), + httpx.Response(200, json={"arrivals": [ei1]}), + ) + + assert list(fetch_aerodatabox(JUNE_1, JUNE_1)["flight_number"]) == ["EI 1"] + + +def test_rate_limit_raises(monkeypatch: pytest.MonkeyPatch) -> None: + quota = [httpx.Response(429, text="exceeded the MONTHLY quota") for _ in range(4)] + urls = _fake_api(monkeypatch, *quota) + + with pytest.raises(RateLimitedError, match="MONTHLY quota"): + fetch_aerodatabox(JUNE_1, JUNE_1) + assert len(urls) == 4 # 1 try + 3 retries + + +def test_retries_a_rate_limit_then_carries_on(monkeypatch: pytest.MonkeyPatch) -> None: + urls = _fake_api( + monkeypatch, httpx.Response(429), httpx.Response(204), httpx.Response(204) + ) + + assert fetch_aerodatabox(JUNE_1, JUNE_1).empty + assert len(urls) == 3 # the morning window was asked twice diff --git a/backend/tests/test_weather_ingest.py b/backend/tests/test_weather_ingest.py index b46db29..1febaf7 100644 --- a/backend/tests/test_weather_ingest.py +++ b/backend/tests/test_weather_ingest.py @@ -7,7 +7,7 @@ import pandas as pd import pytest -from foresight.ingest.weather import _rows +from foresight.ingest.weather import _rows, chunks _DAY = 86400 # temperature_2m_max, temperature_2m_min, precipitation_sum, wind_speed_10m_max, @@ -111,3 +111,17 @@ def test_row_matches_the_table_columns() -> None: "is_forecast": True, "source": "open-meteo", } + + +def test_long_ranges_are_split_a_year_at_a_time() -> None: + spans = list(chunks(date(2024, 1, 1), date(2025, 6, 30))) + assert spans == [ + (date(2024, 1, 1), date(2024, 12, 30)), + (date(2024, 12, 31), date(2025, 6, 30)), + ] + + +def test_a_short_range_is_one_request() -> None: + assert list(chunks(date(2025, 6, 1), date(2025, 6, 1))) == [ + (date(2025, 6, 1), date(2025, 6, 1)) + ] From cde8920f62b4b36dac6c0085e7eca76efaeaf2a8 Mon Sep 17 00:00:00 2001 From: Maria Samos Rivas Date: Tue, 13 Oct 2026 14:26:18 -0400 Subject: [PATCH 3/4] feat: backfill weather, events and flights over a hotel's stay window Services fetch and upsert without committing; events yield after each month and flights after each day, so the scripts in scripts/backfill/ commit as they go and an interrupted run resumes where it stopped. The default window is a hotel's first and last stay_date in daily_performance. ingest/run.py's market helpers and migrate() are public for the scripts. Co-authored-by: narang-an Co-authored-by: Vichu Selvaraju --- backend/foresight/ingest/predicthq.py | 4 +- backend/foresight/ingest/run.py | 12 +- backend/foresight/repository/hotel.py | 17 ++ backend/foresight/repository/weather.py | 17 ++ .../foresight/services/backfill/__init__.py | 0 backend/foresight/services/backfill/events.py | 265 ++++++++++++++++++ .../foresight/services/backfill/flights.py | 62 ++++ .../foresight/services/backfill/weather.py | 32 +++ .../services/backfill/weather_coverage.py | 54 ++++ backend/foresight/services/backfill/window.py | 43 +++ backend/scripts/backfill/backfill_events.py | 106 +++++++ backend/scripts/backfill/backfill_flights.py | 71 +++++ backend/scripts/backfill/backfill_weather.py | 64 +++++ backend/scripts/backfill/weather_coverage.py | 54 ++++ backend/tests/test_backfill_events.py | 227 +++++++++++++++ backend/tests/test_backfill_flights.py | 63 +++++ backend/tests/test_backfill_window.py | 55 ++++ backend/tests/test_hotel_repository.py | 22 ++ backend/tests/test_weather_coverage.py | 56 ++++ backend/tests/test_weather_repository.py | 17 +- 20 files changed, 1232 insertions(+), 9 deletions(-) create mode 100644 backend/foresight/services/backfill/__init__.py create mode 100644 backend/foresight/services/backfill/events.py create mode 100644 backend/foresight/services/backfill/flights.py create mode 100644 backend/foresight/services/backfill/weather.py create mode 100644 backend/foresight/services/backfill/weather_coverage.py create mode 100644 backend/foresight/services/backfill/window.py create mode 100644 backend/scripts/backfill/backfill_events.py create mode 100644 backend/scripts/backfill/backfill_flights.py create mode 100644 backend/scripts/backfill/backfill_weather.py create mode 100644 backend/scripts/backfill/weather_coverage.py create mode 100644 backend/tests/test_backfill_events.py create mode 100644 backend/tests/test_backfill_flights.py create mode 100644 backend/tests/test_backfill_window.py create mode 100644 backend/tests/test_weather_coverage.py diff --git a/backend/foresight/ingest/predicthq.py b/backend/foresight/ingest/predicthq.py index 6ea1013..7522aef 100644 --- a/backend/foresight/ingest/predicthq.py +++ b/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) diff --git a/backend/foresight/ingest/run.py b/backend/foresight/ingest/run.py index a407955..67012bd 100644 --- a/backend/foresight/ingest/run.py +++ b/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)) diff --git a/backend/foresight/repository/hotel.py b/backend/foresight/repository/hotel.py index b23787d..323701b 100644 --- a/backend/foresight/repository/hotel.py +++ b/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) diff --git a/backend/foresight/repository/weather.py b/backend/foresight/repository/weather.py index bc5ee20..79f4290 100644 --- a/backend/foresight/repository/weather.py +++ b/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()} diff --git a/backend/foresight/services/backfill/__init__.py b/backend/foresight/services/backfill/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/foresight/services/backfill/events.py b/backend/foresight/services/backfill/events.py new file mode 100644 index 0000000..8b69464 --- /dev/null +++ b/backend/foresight/services/backfill/events.py @@ -0,0 +1,265 @@ +from __future__ import annotations + +import asyncio +import logging +from collections import Counter +from dataclasses import dataclass, field +from datetime import date, timedelta +from typing import TYPE_CHECKING, Any + +import httpx + +from foresight.ingest.predicthq import ( + API_URL, + PAGE_LIMIT, + STATES, + Rejected, + rejection_reason, + to_normalized, +) +from foresight.repository import Outcome +from foresight.repository.events import upsert_event + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Awaitable, Callable + + from sqlalchemy.ext.asyncio import AsyncSession + + from foresight.ingest.run import Market + +log = logging.getLogger("foresight.ingest") + +MAX_RETRIES = 5 +BACKOFF_SECONDS = 2.0 +# PredictHQ's docs don't name the status for an exhausted plan; treat both. +_OUT_OF_CREDITS = {402, 403} + + +class PredictHQLimitError(Exception): + """PredictHQ refused further requests; the run should stop.""" + + +class OutOfCreditsError(PredictHQLimitError): + """The plan's quota or credits are used up.""" + + +class RetriesExhaustedError(PredictHQLimitError): + """Still rate limited or failing after every retry.""" + + +@dataclass(slots=True) +class BackfillReport: + market: Market + start: date + end: date + chunks_total: int + chunks_done: int = 0 + fetched: int = 0 + outcomes: Counter[Outcome] = field(default_factory=Counter) + rejected: list[Rejected] = field(default_factory=list) + overflow_chunks: list[tuple[date, date]] = field(default_factory=list) + stopped: str | None = None + resume_from: date | None = None + dry_run: bool = False + + def summary(self) -> str: + """One line of counts for the run.""" + counts = ( + "(dry run, nothing written)" + if self.dry_run + else " ".join(f"{o.value}={self.outcomes[o]}" for o in Outcome) + ) + return ( + f"{self.market.city} {self.start}..{self.end} " + f"chunks={self.chunks_done}/{self.chunks_total} " + f"fetched={self.fetched} {counts} rejected={len(self.rejected)}" + ) + + +def month_chunks(start: date, end: date) -> list[tuple[date, date]]: + """Split a date range into calendar-month pieces, oldest first. + + Args: + start (date): First day, inclusive. + end (date): Last day, inclusive. + + Returns: + list[tuple[date, date]]: Inclusive (first, last) day of each piece. + """ + chunks = [] + first = start + while first <= end: + next_month = (first.replace(day=1) + timedelta(days=32)).replace(day=1) + last = min(next_month - timedelta(days=1), end) + chunks.append((first, last)) + first = last + timedelta(days=1) + return chunks + + +def chunk_params(market: Market, start: date, end: date, radius: str) -> dict[str, Any]: + """Query for every event active within a chunk, with a day of slack each side. + + Args: + market (Market): Centre of the search. + start (date): First day of the chunk. + end (date): Last day of the chunk. + radius (str): Search radius, e.g. "10km". + + Returns: + dict[str, Any]: PredictHQ query parameters. + """ + return { + "within": f"{radius}@{market.latitude},{market.longitude}", + "active.gte": (start - timedelta(days=1)).isoformat(), + "active.lte": (end + timedelta(days=1)).isoformat(), + "state": STATES, + "limit": PAGE_LIMIT, + "sort": "start", + } + + +async def _get( + client: httpx.AsyncClient, + url: str, + params: dict[str, Any] | None, + *, + sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, +) -> dict[str, Any]: + """GET one page, waiting and retrying only on 429, 5xx or network errors. + + Args: + client (httpx.AsyncClient): Authenticated client. + url (str): Page URL. + params (dict[str, Any] | None): Query, or None for a `next` link. + sleep (Callable[[float], Awaitable[None]]): Waits between retries. + + Returns: + dict[str, Any]: The page's JSON body. + + Raises: + OutOfCreditsError: On a 402 or 403. + RetriesExhaustedError: If every retry still fails. + httpx.HTTPStatusError: On any other 4xx. + """ + problem, delay = "", 0.0 + for attempt in range(MAX_RETRIES + 1): + if attempt: + await sleep(delay) + try: + response = await client.get(url, params=params) + except httpx.TransportError as exc: + problem = f"network error: {exc}" + delay = BACKOFF_SECONDS * 2**attempt + continue + + status = response.status_code + if status in _OUT_OF_CREDITS: + raise OutOfCreditsError(f"HTTP {status}: {response.text[:300]}") + if status == 429 or status >= 500: + problem = f"HTTP {status}" + delay = _retry_after(response) or BACKOFF_SECONDS * 2**attempt + log.warning("PredictHQ %s, retrying in %.0fs", problem, delay) + continue + response.raise_for_status() + return response.json() + + raise RetriesExhaustedError(f"{problem} after {MAX_RETRIES} retries") + + +def _retry_after(response: httpx.Response) -> float | None: + try: + return float(response.headers["Retry-After"]) + except (KeyError, ValueError): + return None + + +async def backfill( + session: AsyncSession | None, + market: Market, + start: date, + end: date, + *, + token: str, + radius: str = "10km", + transport: httpx.AsyncBaseTransport | None = None, + sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, +) -> AsyncIterator[BackfillReport]: + """Fetch and upsert every event active from `start` to `end`, a month at a time. + + Yields the report after each month so the caller can commit as it goes; + never commits itself. A PredictHQ limit ends the run: the report is yielded + once more with `stopped` and `resume_from` set. + + Args: + session (AsyncSession | None): Where to write; None for a dry run. + market (Market): City and centre of the search. + start (date): First day, inclusive. + end (date): Last day, inclusive. + token (str): PredictHQ access token. + radius (str): Search radius, e.g. "10km". + transport (httpx.AsyncBaseTransport | None): Test transport. + sleep (Callable[[float], Awaitable[None]]): Waits between retries. + + Yields: + BackfillReport: Running counts after each month, or the stopped run. + """ + chunks = month_chunks(start, end) + report = BackfillReport( + market=market, + start=start, + end=end, + chunks_total=len(chunks), + dry_run=session is None, + ) + headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"} + + async with httpx.AsyncClient( + headers=headers, timeout=30.0, transport=transport + ) as client: + for first, last in chunks: + try: + await _backfill_chunk( + client, session, report, first, last, radius, sleep + ) + except PredictHQLimitError as exc: + report.stopped = str(exc) + report.resume_from = first + yield report + return + report.chunks_done += 1 + yield report + + +async def _backfill_chunk( + client: httpx.AsyncClient, + session: AsyncSession | None, + report: BackfillReport, + first: date, + last: date, + radius: str, + sleep: Callable[[float], Awaitable[None]], +) -> None: + url: str | None = API_URL + query: dict[str, Any] | None = chunk_params(report.market, first, last, radius) + while url: + body = await _get(client, url, query, sleep=sleep) + if body.get("overflow") and (first, last) not in report.overflow_chunks: + log.warning( + "PredictHQ overflow for %s..%s: plan truncated results", first, last + ) + report.overflow_chunks.append((first, last)) + + for raw in body.get("results", []): + report.fetched += 1 + try: + event = to_normalized(raw, city=report.market.city) + 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) + report.rejected.append(Rejected(ref, rejection_reason(exc))) + continue + if session is not None: + report.outcomes[await upsert_event(session, event)] += 1 + + # `next` already carries every query parameter. + url, query = body.get("next"), None diff --git a/backend/foresight/services/backfill/flights.py b/backend/foresight/services/backfill/flights.py new file mode 100644 index 0000000..a220190 --- /dev/null +++ b/backend/foresight/services/backfill/flights.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from datetime import timedelta +from typing import TYPE_CHECKING + +from foresight.repository.flights import upsert_flight_arrivals +from foresight.services.flight_data_service import RateLimitedError, fetch_aerodatabox + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Callable + from datetime import date + + import pandas as pd + from sqlalchemy.ext.asyncio import AsyncSession + +SOURCE = "aerodatabox" + + +class FlightBackfillStoppedError(Exception): + """AeroDataBox rate limited us; `day` is the first day not saved.""" + + def __init__(self, day: date, reason: str) -> None: + super().__init__(f"rate limited on {day}: {reason}") + self.day = day + + +async def backfill_flights( + session: AsyncSession, + start: date, + end: date, + airport: str = "DUB", + fetch: Callable[[date, date, str], pd.DataFrame] = fetch_aerodatabox, +) -> AsyncIterator[tuple[date, int]]: + """Fetch and save the arrival count for each day from `start` to `end`. + + Yields (day, count) after each day is written, so the caller can commit + day by day. Does not commit itself. + + Raises: + FlightBackfillStoppedError: If AeroDataBox rate limits us; every day yielded + before it is complete. + """ + day = start + while day <= end: + try: + flights = fetch(day, day, airport) + except RateLimitedError as exc: + raise FlightBackfillStoppedError(day, str(exc)) from exc + await upsert_flight_arrivals( + session, + [ + { + "airport": airport, + "arrival_date": day, + "flight_count": len(flights), + "avg_fare_eur": None, + "source": SOURCE, + } + ], + ) + yield day, len(flights) + day += timedelta(days=1) diff --git a/backend/foresight/services/backfill/weather.py b/backend/foresight/services/backfill/weather.py new file mode 100644 index 0000000..9c74baf --- /dev/null +++ b/backend/foresight/services/backfill/weather.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +from datetime import date +from typing import TYPE_CHECKING + +from foresight.ingest.weather import fetch_actuals +from foresight.repository import weather as weather_repo + +if TYPE_CHECKING: + from sqlalchemy.ext.asyncio import AsyncSession + + from foresight.repository import Outcome + + +async def backfill_weather( + session: AsyncSession, + *, + city: str, + latitude: float, + longitude: float, + start: date, + end: date, +) -> dict[Outcome, int]: + """Fetch recorded weather for every day from `start` to `end` and save it. + + The archive rejects future dates, so `end` is clamped to today. Does not + commit; the caller owns the transaction. + """ + rows = fetch_actuals( + city, latitude, longitude, "auto", start=start, end=min(end, date.today()) + ) + return await weather_repo.upsert_weather(session, rows) diff --git a/backend/foresight/services/backfill/weather_coverage.py b/backend/foresight/services/backfill/weather_coverage.py new file mode 100644 index 0000000..e12755b --- /dev/null +++ b/backend/foresight/services/backfill/weather_coverage.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date, timedelta +from typing import TYPE_CHECKING + +from foresight.repository import weather as weather_repo + +if TYPE_CHECKING: + from collections.abc import Iterable + + from sqlalchemy.ext.asyncio import AsyncSession + + +@dataclass(frozen=True, slots=True) +class WeatherCoverage: + city: str + start: date + end: date + days: int # days in the window + rows: int # days with a row saved + earliest: date | None # first day with values + missing: list[tuple[date, date]] # no row at all + empty: list[tuple[date, date]] # row saved, values missing (archive lag) + + +async def weather_coverage( + session: AsyncSession, *, city: str, start: date, end: date +) -> WeatherCoverage: + """Count saved days and group the gaps into runs of consecutive days.""" + saved = await weather_repo.recorded_days(session, city=city, start=start, end=end) + window = [start + timedelta(days=n) for n in range((end - start).days + 1)] + usable = [d for d, has_values in saved.items() if has_values] + return WeatherCoverage( + city=city, + start=start, + end=end, + days=len(window), + rows=len(saved), + earliest=min(usable, default=None), + missing=_spans(d for d in window if d not in saved), + empty=_spans(sorted(d for d, has_values in saved.items() if not has_values)), + ) + + +def _spans(days: Iterable[date]) -> list[tuple[date, date]]: + """Group sorted days into (first, last) runs of consecutive days.""" + spans: list[tuple[date, date]] = [] + for day in days: + if spans and day == spans[-1][1] + timedelta(days=1): + spans[-1] = (spans[-1][0], day) + else: + spans.append((day, day)) + return spans diff --git a/backend/foresight/services/backfill/window.py b/backend/foresight/services/backfill/window.py new file mode 100644 index 0000000..e1a01cc --- /dev/null +++ b/backend/foresight/services/backfill/window.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from foresight.repository import hotel as hotel_repo + +if TYPE_CHECKING: + from datetime import date + + from sqlalchemy.ext.asyncio import AsyncSession + + +async def hotel_window( + session: AsyncSession, + hotel_id: int | None, + start: date | None, + end: date | None, +) -> tuple[date, date]: + """Fill a missing `start` or `end` from the hotel's stay_date range. + + Every source aims at the same window; one whose API can't reach back that far + just returns nothing for the early days, and the coverage report shows the gap. + + Raises: + ValueError: If a date is missing and the hotel has no nights loaded (or + no hotel was given), or if `start` is after `end`. + """ + if start is None or end is None: + loaded = ( + None + if hotel_id is None + else await hotel_repo.stay_date_range(session, hotel_id) + ) + if loaded is None: + raise ValueError( + "no hotel nights loaded; give --hotel-id of a loaded hotel, " + "or both --start and --end" + ) + start = start or loaded[0] + end = end or loaded[1] + if start > end: + raise ValueError(f"start {start} is after end {end}") + return start, end diff --git a/backend/scripts/backfill/backfill_events.py b/backend/scripts/backfill/backfill_events.py new file mode 100644 index 0000000..ef931e7 --- /dev/null +++ b/backend/scripts/backfill/backfill_events.py @@ -0,0 +1,106 @@ +import argparse +import asyncio +import logging +import sys +from collections.abc import Sequence +from datetime import date + +from foresight.config import settings +from foresight.database.session import async_session_factory, engine +from foresight.ingest.run import market_totals, migrate, resolve_market +from foresight.services.backfill.events import BackfillReport, backfill +from foresight.services.backfill.window import hotel_window + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="scripts/backfill/backfill_events.py", + description="Backfill PredictHQ events for a date range.", + ) + parser.add_argument( + "--start", type=date.fromisoformat, help="YYYY-MM-DD; defaults to the hotel's" + ) + parser.add_argument( + "--end", type=date.fromisoformat, help="YYYY-MM-DD; defaults to the hotel's" + ) + parser.add_argument( + "--hotel-id", type=int, help="use this hotel's city and lat/lon" + ) + parser.add_argument("--lat", type=float, help="market latitude (no --hotel-id)") + parser.add_argument("--lon", type=float, help="market longitude (no --hotel-id)") + parser.add_argument("--city", help="market city (no --hotel-id)") + parser.add_argument("--radius", default="10km", help="e.g. 10km, 5mi") + parser.add_argument( + "--dry-run", + action="store_true", + help="fetch and normalize, without writing to the database", + ) + return parser + + +async def _run(args: argparse.Namespace) -> BackfillReport: + try: + async with async_session_factory() as session: + market = await resolve_market(session, args) + try: + start, end = await hotel_window( + session, args.hotel_id, args.start, args.end + ) + except ValueError as exc: + raise SystemExit(f"error: {exc}") from exc + report = None + async for progress in backfill( + None if args.dry_run else session, + market, + start, + end, + token=settings.predicthq_token, + radius=args.radius, + ): + report = progress + if not args.dry_run: + await session.commit() # keep each finished month + if report is None: + raise SystemExit("error: empty date range") + if not args.dry_run: + by_status, by_source = await market_totals(session, market.city) + print( + f"{market.city}: events by status {by_status}, " + f"observations by source {by_source}" + ) + finally: + await engine.dispose() + return report + + +def main(argv: Sequence[str] | None = None) -> int: + parser = _parser() + args = parser.parse_args(argv) + if args.hotel_id is None and None in (args.lat, args.lon, args.city): + parser.error("give --hotel-id, or all of --lat, --lon and --city") + if not settings.predicthq_token: + parser.error("PREDICTHQ_TOKEN is not set; add it to .env") + + logging.basicConfig(level=logging.WARNING, format="%(levelname)s %(message)s") + if not args.dry_run: + migrate() + engine.echo = False # DEBUG=true would bury the report in SQL + report = asyncio.run(_run(args)) + + print(report.summary()) + for r in report.rejected: + print(f" rejected {r.source_ref:<22} {r.reason}") + if report.overflow_chunks: + print(f" overflow (plan truncated): {report.overflow_chunks}") + if report.stopped: + print(f"stopped: {report.stopped}", file=sys.stderr) + print( + f"saved everything before it; resume with --start {report.resume_from}", + file=sys.stderr, + ) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/scripts/backfill/backfill_flights.py b/backend/scripts/backfill/backfill_flights.py new file mode 100644 index 0000000..b4eed65 --- /dev/null +++ b/backend/scripts/backfill/backfill_flights.py @@ -0,0 +1,71 @@ +import argparse +import asyncio +import logging +import sys +from datetime import date + +from foresight.config import settings +from foresight.database.session import async_session_factory, engine +from foresight.ingest.run import migrate +from foresight.services.backfill.flights import ( + FlightBackfillStoppedError, + backfill_flights, +) +from foresight.services.backfill.window import hotel_window + +log = logging.getLogger("foresight.backfill") + + +async def _run(args: argparse.Namespace) -> date | None: + """Backfill day by day, committing each. Returns the day a limit stopped on.""" + try: + async with async_session_factory() as session: + try: + start, end = await hotel_window( + session, args.hotel_id, args.start, args.end + ) + except ValueError as exc: + raise SystemExit(f"error: {exc}") from exc + try: + async for day, count in backfill_flights( + session, start, end, args.airport + ): + await session.commit() # keep each finished day + log.info("%s: %d arrivals saved", day, count) + except FlightBackfillStoppedError as exc: + log.error("%s", exc) + return exc.day + finally: + await engine.dispose() + return None + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Save the number of flights arriving each day, from AeroDataBox." + ) + parser.add_argument( + "--start", type=date.fromisoformat, help="YYYY-MM-DD; defaults to the hotel's" + ) + parser.add_argument( + "--end", type=date.fromisoformat, help="YYYY-MM-DD; defaults to the hotel's" + ) + parser.add_argument("--hotel-id", type=int, help="default the dates to this hotel") + parser.add_argument("--airport", default="DUB", help="IATA code (default: DUB)") + args = parser.parse_args() + args.airport = args.airport.upper() + if not settings.xrapidapi_key: + parser.error("XRAPIDAPI_KEY is not set; add it to .env") + + logging.basicConfig(level=logging.INFO, format="%(message)s") + migrate() + engine.echo = False + stopped_on = asyncio.run(_run(args)) + if stopped_on: + print(f"resume with --start {stopped_on}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/scripts/backfill/backfill_weather.py b/backend/scripts/backfill/backfill_weather.py new file mode 100644 index 0000000..45a4275 --- /dev/null +++ b/backend/scripts/backfill/backfill_weather.py @@ -0,0 +1,64 @@ +import argparse +import asyncio +from datetime import date + +from foresight.database.session import async_session_factory, engine +from foresight.ingest.run import resolve_market +from foresight.services.backfill.weather import backfill_weather +from foresight.services.backfill.window import hotel_window + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Backfill recorded weather for a city over a date range." + ) + parser.add_argument( + "--start", type=date.fromisoformat, help="YYYY-MM-DD; defaults to the hotel's" + ) + parser.add_argument( + "--end", type=date.fromisoformat, help="YYYY-MM-DD; defaults to the hotel's" + ) + parser.add_argument( + "--hotel-id", type=int, help="use this hotel's city and lat/lon" + ) + parser.add_argument("--city", help="market city (no --hotel-id)") + parser.add_argument("--lat", type=float, help="market latitude (no --hotel-id)") + parser.add_argument("--lon", type=float, help="market longitude (no --hotel-id)") + return parser + + +async def main() -> int: + parser = _parser() + args = parser.parse_args() + if args.hotel_id is None and None in (args.city, args.lat, args.lon): + parser.error("give --hotel-id, or all of --city, --lat and --lon") + + engine.echo = False + try: + async with async_session_factory() as session: + market = await resolve_market(session, args) + try: + start, end = await hotel_window( + session, args.hotel_id, args.start, args.end + ) + except ValueError as exc: + parser.error(str(exc)) + counts = await backfill_weather( + session, + city=market.city, + latitude=market.latitude, + longitude=market.longitude, + start=start, + end=end, + ) + await session.commit() + finally: + await engine.dispose() + + summary = ", ".join(f"{n} {outcome.value}" for outcome, n in counts.items()) + print(f"weather | {market.city} | {start} → {end} | {summary}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main())) diff --git a/backend/scripts/backfill/weather_coverage.py b/backend/scripts/backfill/weather_coverage.py new file mode 100644 index 0000000..e1d04d6 --- /dev/null +++ b/backend/scripts/backfill/weather_coverage.py @@ -0,0 +1,54 @@ +import argparse +import asyncio +from datetime import date + +from foresight.database.session import async_session_factory, engine +from foresight.services.backfill.weather_coverage import weather_coverage +from foresight.services.backfill.window import hotel_window + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Report recorded-weather coverage for a city over a date range." + ) + parser.add_argument("--city", required=True, help="market city, e.g. Dublin") + parser.add_argument("--hotel-id", type=int, help="default the dates to this hotel") + parser.add_argument( + "--start", type=date.fromisoformat, help="YYYY-MM-DD; defaults to the hotel's" + ) + parser.add_argument( + "--end", type=date.fromisoformat, help="YYYY-MM-DD; defaults to the hotel's" + ) + return parser + + +async def main() -> int: + parser = _parser() + args = parser.parse_args() + + engine.echo = False + try: + async with async_session_factory() as session: + try: + start, end = await hotel_window( + session, args.hotel_id, args.start, args.end + ) + except ValueError as exc: + parser.error(str(exc)) + report = await weather_coverage( + session, city=args.city, start=start, end=end + ) + finally: + await engine.dispose() + + print(f"weather | {report.city} | {report.start} → {report.end}") + print(f" rows: {report.rows} of {report.days} days") + print(f" earliest usable: {report.earliest or 'none'}") + for label, spans in (("missing:", report.missing), ("no values:", report.empty)): + gaps = ", ".join(str(a) if a == b else f"{a} → {b}" for a, b in spans) + print(f" {label:<16} {gaps or 'none'}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main())) diff --git a/backend/tests/test_backfill_events.py b/backend/tests/test_backfill_events.py new file mode 100644 index 0000000..1b2541c --- /dev/null +++ b/backend/tests/test_backfill_events.py @@ -0,0 +1,227 @@ +from __future__ import annotations + +from datetime import date +from typing import TYPE_CHECKING, Any + +import httpx +import pytest +from sqlalchemy import delete, func, select +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine + +from foresight.config import settings +from foresight.database.models.external import Event, EventObservation +from foresight.ingest.predicthq import API_URL +from foresight.ingest.run import Market +from foresight.repository import Outcome +from foresight.services.backfill.events import ( + MAX_RETRIES, + OutOfCreditsError, + RetriesExhaustedError, + _get, + backfill, + chunk_params, + month_chunks, +) + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Callable + +DUBLIN = Market(city="Dublin", latitude=53.3498, longitude=-6.2603) + + +class FakeSleep: + def __init__(self) -> None: + self.waits: list[float] = [] + + async def __call__(self, seconds: float) -> None: + self.waits.append(seconds) + + +def _raw(event_id: str, day: str) -> dict[str, Any]: + return { + "id": event_id, + "title": f"Gig {event_id}", + "category": "concerts", + "state": "active", + "start": f"{day}T19:00:00Z", + "end": f"{day}T22:00:00Z", + "timezone": "Europe/Dublin", + "location": [-6.2286, 53.3478], + "country": "IE", + "updated": "2025-01-01T00:00:00Z", + } + + +def _client(handler: Callable[[httpx.Request], httpx.Response]) -> httpx.AsyncClient: + return httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + +async def _drain(*args: Any, **kwargs: Any) -> Any: + """Run a backfill to the end and return its last report.""" + report = None + async for report in backfill(*args, **kwargs): # noqa: B007 + pass + return report + + +def _replies(*responses: httpx.Response) -> Callable[[httpx.Request], httpx.Response]: + queue = list(responses) + return lambda request: queue.pop(0) + + +def test_month_chunks_split_partial_months() -> None: + assert month_chunks(date(2024, 1, 15), date(2024, 3, 10)) == [ + (date(2024, 1, 15), date(2024, 1, 31)), + (date(2024, 2, 1), date(2024, 2, 29)), + (date(2024, 3, 1), date(2024, 3, 10)), + ] + + +def test_month_chunks_single_day() -> None: + day = date(2025, 6, 30) + assert month_chunks(day, day) == [(day, day)] + + +def test_chunk_params_cover_the_chunk_with_slack_and_no_revision_filter() -> None: + params = chunk_params(DUBLIN, date(2025, 3, 1), date(2025, 3, 31), "10km") + + assert params["active.gte"] == "2025-02-28" + assert params["active.lte"] == "2025-04-01" + assert params["within"] == "10km@53.3498,-6.2603" + assert "updated.gte" not in params + + +async def test_success_never_waits() -> None: + sleep = FakeSleep() + async with _client(_replies(httpx.Response(200, json={"results": []}))) as client: + assert await _get(client, API_URL, None, sleep=sleep) == {"results": []} + assert sleep.waits == [] + + +async def test_rate_limit_backs_off_then_succeeds() -> None: + sleep = FakeSleep() + handler = _replies( + httpx.Response(429), httpx.Response(429), httpx.Response(200, json={}) + ) + async with _client(handler) as client: + await _get(client, API_URL, None, sleep=sleep) + assert sleep.waits == [2.0, 4.0] + + +async def test_rate_limit_honours_retry_after() -> None: + sleep = FakeSleep() + handler = _replies( + httpx.Response(429, headers={"Retry-After": "7"}), httpx.Response(200, json={}) + ) + async with _client(handler) as client: + await _get(client, API_URL, None, sleep=sleep) + assert sleep.waits == [7.0] + + +async def test_gives_up_after_every_retry() -> None: + sleep = FakeSleep() + async with _client(lambda request: httpx.Response(429)) as client: + with pytest.raises(RetriesExhaustedError): + await _get(client, API_URL, None, sleep=sleep) + assert len(sleep.waits) == MAX_RETRIES + + +async def test_out_of_credits_stops_without_waiting() -> None: + sleep = FakeSleep() + async with _client(_replies(httpx.Response(402, text="quota exceeded"))) as client: + with pytest.raises(OutOfCreditsError, match="quota exceeded"): + await _get(client, API_URL, None, sleep=sleep) + assert sleep.waits == [] + + +async def test_dry_run_counts_without_a_database() -> None: + page = {"results": [_raw("phq-a", "2025-03-05"), {"id": "phq-bad"}]} + report = await _drain( + None, + DUBLIN, + date(2025, 3, 1), + date(2025, 3, 31), + token="t", + transport=httpx.MockTransport(_replies(httpx.Response(200, json=page))), + sleep=FakeSleep(), + ) + + assert (report.fetched, len(report.rejected), report.chunks_done) == (2, 1, 1) + assert report.stopped is None + + +@pytest.fixture +async def session() -> AsyncIterator[AsyncSession]: + engine = create_async_engine(settings.database_url) + try: + conn = await engine.connect() + except (OSError, ConnectionError) as exc: + await engine.dispose() + pytest.skip(f"Postgres unavailable: {exc}") + + trans = await conn.begin() + async with AsyncSession( + bind=conn, join_transaction_mode="create_savepoint" + ) as session: + await session.execute(delete(EventObservation)) + await session.execute(delete(Event)) + yield session + await trans.rollback() + await conn.close() + await engine.dispose() + + +async def test_out_of_credits_keeps_earlier_pages_and_rerun_adds_no_duplicates( + session: AsyncSession, +) -> None: + page_1 = { + "results": [_raw("phq-a", "2025-03-05"), _raw("phq-b", "2025-03-12")], + "next": f"{API_URL}?page=2", + } + page_2 = {"results": [_raw("phq-c", "2025-03-20")]} + march = (date(2025, 3, 1), date(2025, 3, 31)) + + stopped = await _drain( + session, + DUBLIN, + *march, + token="t", + transport=httpx.MockTransport( + _replies(httpx.Response(200, json=page_1), httpx.Response(402)) + ), + sleep=FakeSleep(), + ) + assert stopped.stopped is not None + assert stopped.resume_from == date(2025, 3, 1) + assert await session.scalar(select(func.count()).select_from(Event)) == 2 + + resumed = await _drain( + session, + DUBLIN, + *march, + token="t", + transport=httpx.MockTransport( + _replies(httpx.Response(200, json=page_1), httpx.Response(200, json=page_2)) + ), + sleep=FakeSleep(), + ) + assert resumed.stopped is None + assert resumed.outcomes[Outcome.NEW] == 1 + assert await session.scalar(select(func.count()).select_from(Event)) == 3 + + +async def test_yields_after_each_month_without_committing() -> None: + empty = httpx.Response(200, json={"results": []}) + reports = [ + report.chunks_done + async for report in backfill( + None, + DUBLIN, + date(2025, 1, 15), + date(2025, 3, 10), + token="t", + transport=httpx.MockTransport(lambda request: empty), + sleep=FakeSleep(), + ) + ] + assert reports == [1, 2, 3] diff --git a/backend/tests/test_backfill_flights.py b/backend/tests/test_backfill_flights.py new file mode 100644 index 0000000..f102130 --- /dev/null +++ b/backend/tests/test_backfill_flights.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +from datetime import date +from typing import TYPE_CHECKING + +import pandas as pd +import pytest +from sqlalchemy import select + +from foresight.database.models.external import FlightArrival +from foresight.services.backfill.flights import ( + FlightBackfillStoppedError, + backfill_flights, +) +from foresight.services.flight_data_service import RateLimitedError + +if TYPE_CHECKING: + from sqlalchemy.ext.asyncio import AsyncSession + +AIRPORT = "TST" + + +async def _saved(session: AsyncSession) -> dict[date, int | None]: + rows = await session.scalars( + select(FlightArrival).where(FlightArrival.airport == AIRPORT) + ) + return {row.arrival_date: row.flight_count for row in rows} + + +async def test_rate_limit_keeps_earlier_days_and_rerun_overwrites( + session: AsyncSession, +) -> None: + def rate_limited_on_june_2(start: date, end: date, airport: str) -> pd.DataFrame: + if start == date(2026, 6, 2): + raise RateLimitedError("Too many requests") + return pd.DataFrame({"flight_number": ["EI 1", "EI 2"]}) + + saved_days = [] + with pytest.raises(FlightBackfillStoppedError) as stopped: + async for day, _ in backfill_flights( + session, + date(2026, 6, 1), + date(2026, 6, 3), + AIRPORT, + fetch=rate_limited_on_june_2, + ): + saved_days.append(day) + assert stopped.value.day == date(2026, 6, 2) + assert saved_days == [date(2026, 6, 1)] + assert await _saved(session) == {date(2026, 6, 1): 2} + + def one_flight(start: date, end: date, airport: str) -> pd.DataFrame: + return pd.DataFrame({"flight_number": ["EI 1"]}) + + async for _ in backfill_flights( + session, date(2026, 6, 1), date(2026, 6, 3), AIRPORT, fetch=one_flight + ): + pass + assert await _saved(session) == { + date(2026, 6, 1): 1, + date(2026, 6, 2): 1, + date(2026, 6, 3): 1, + } diff --git a/backend/tests/test_backfill_window.py b/backend/tests/test_backfill_window.py new file mode 100644 index 0000000..098fda1 --- /dev/null +++ b/backend/tests/test_backfill_window.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +from datetime import date + +import pytest + +from foresight.repository import hotel as hotel_repo +from foresight.services.backfill.window import hotel_window + +HOTEL_RANGE = (date(2022, 3, 1), date(2025, 9, 30)) + + +@pytest.fixture +def hotel_range(monkeypatch: pytest.MonkeyPatch): + """Stand in for the DB: stay_date_range returns whatever the test sets.""" + answer = {"range": HOTEL_RANGE} + + async def fake_stay_date_range(session, hotel_id): + return answer["range"] + + monkeypatch.setattr(hotel_repo, "stay_date_range", fake_stay_date_range) + return answer + + +async def test_defaults_to_the_hotel_range(hotel_range) -> None: + assert await hotel_window(None, 1, None, None) == HOTEL_RANGE + + +async def test_given_dates_win_over_the_hotel_range(hotel_range) -> None: + start, end = date(2024, 1, 1), date(2024, 1, 31) + + assert await hotel_window(None, 1, start, None) == (start, HOTEL_RANGE[1]) + assert await hotel_window(None, 1, start, end) == (start, end) + + +async def test_both_dates_need_no_hotel(hotel_range) -> None: + start, end = date(2024, 1, 1), date(2024, 1, 31) + assert await hotel_window(None, None, start, end) == (start, end) + + +async def test_no_hotel_nights_needs_both_dates(hotel_range) -> None: + hotel_range["range"] = None + + with pytest.raises(ValueError, match="no hotel nights loaded"): + await hotel_window(None, 1, None, None) + + +async def test_missing_date_without_a_hotel_is_rejected(hotel_range) -> None: + with pytest.raises(ValueError, match="no hotel nights loaded"): + await hotel_window(None, None, date(2024, 1, 1), None) + + +async def test_start_after_end_is_rejected(hotel_range) -> None: + with pytest.raises(ValueError, match="is after"): + await hotel_window(None, None, date(2025, 1, 1), date(2024, 1, 1)) diff --git a/backend/tests/test_hotel_repository.py b/backend/tests/test_hotel_repository.py index 5f04036..a630ec5 100644 --- a/backend/tests/test_hotel_repository.py +++ b/backend/tests/test_hotel_repository.py @@ -10,6 +10,7 @@ from foresight.repository import Outcome from foresight.repository.hotel import ( get_hotel, + stay_date_range, upsert_daily_performance, upsert_hotel, upsert_room_type, @@ -162,3 +163,24 @@ async def test_a_reload_without_coordinates_keeps_them( assert hotel is not None assert (hotel.latitude, hotel.longitude) == (53.3498, -6.2603) assert hotel.total_rooms == 45 + + +async def test_stay_date_range_spans_every_room_type(session: AsyncSession) -> None: + hotel_id = await _hotel(session) + king = await _room_type(session, hotel_id) + twin = await _room_type(session, hotel_id, name="Twin") + await upsert_daily_performance( + session, + [ + _night(king, stay_date=date(2026, 6, 3)), + _night(twin, stay_date=date(2026, 5, 30)), + ], + ) + assert await stay_date_range(session, hotel_id) == ( + date(2026, 5, 30), + date(2026, 6, 3), + ) + + +async def test_stay_date_range_is_none_without_nights(session: AsyncSession) -> None: + assert await stay_date_range(session, await _hotel(session)) is None diff --git a/backend/tests/test_weather_coverage.py b/backend/tests/test_weather_coverage.py new file mode 100644 index 0000000..2ef17c8 --- /dev/null +++ b/backend/tests/test_weather_coverage.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +from datetime import date +from typing import TYPE_CHECKING + +from foresight.repository import weather as weather_repo +from foresight.services.backfill.weather_coverage import weather_coverage + +if TYPE_CHECKING: + import pytest + + +def _saved(monkeypatch: pytest.MonkeyPatch, days: dict[date, bool]) -> None: + async def fake_recorded_days(session, *, city, start, end): + return days + + monkeypatch.setattr(weather_repo, "recorded_days", fake_recorded_days) + + +async def test_counts_rows_and_groups_gaps(monkeypatch: pytest.MonkeyPatch) -> None: + _saved( + monkeypatch, + { + date(2024, 3, 1): False, + date(2024, 3, 2): True, + date(2024, 3, 5): True, + date(2024, 3, 6): False, + }, + ) + + report = await weather_coverage( + None, city="Dublin", start=date(2024, 3, 1), end=date(2024, 3, 7) + ) + + assert (report.days, report.rows) == (7, 4) + assert report.earliest == date(2024, 3, 2) + assert report.missing == [ + (date(2024, 3, 3), date(2024, 3, 4)), + (date(2024, 3, 7), date(2024, 3, 7)), + ] + assert report.empty == [ + (date(2024, 3, 1), date(2024, 3, 1)), + (date(2024, 3, 6), date(2024, 3, 6)), + ] + + +async def test_nothing_saved_is_one_big_gap(monkeypatch: pytest.MonkeyPatch) -> None: + _saved(monkeypatch, {}) + + report = await weather_coverage( + None, city="Dublin", start=date(2024, 3, 1), end=date(2024, 3, 31) + ) + + assert report.rows == 0 + assert report.earliest is None + assert report.missing == [(date(2024, 3, 1), date(2024, 3, 31))] diff --git a/backend/tests/test_weather_repository.py b/backend/tests/test_weather_repository.py index ff5280e..4effec3 100644 --- a/backend/tests/test_weather_repository.py +++ b/backend/tests/test_weather_repository.py @@ -7,7 +7,7 @@ from foresight.database.models.external import Weather from foresight.repository import Outcome -from foresight.repository.weather import upsert_weather +from foresight.repository.weather import recorded_days, upsert_weather if TYPE_CHECKING: from sqlalchemy.ext.asyncio import AsyncSession @@ -85,3 +85,18 @@ async def test_an_update_restamps_captured_at(session: AsyncSession) -> None: assert ( await session.scalar(select(Weather.captured_at).where(Weather.city == CITY)) ) > first + + +async def test_recorded_days_flags_days_without_values(session: AsyncSession) -> None: + await upsert_weather( + session, + [ + _row(weather_date=date(2026, 6, 1)), + _row(weather_date=date(2026, 6, 2), temp_max_c=None), + _row(weather_date=date(2026, 6, 3), is_forecast=True), + ], + ) + days = await recorded_days( + session, city=CITY, start=date(2026, 6, 1), end=date(2026, 6, 30) + ) + assert days == {date(2026, 6, 1): True, date(2026, 6, 2): False} From cbcf66c171298eaefb359b9971d8f43869f1e91c Mon Sep 17 00:00:00 2001 From: Maria Samos Rivas Date: Tue, 13 Oct 2026 14:26:18 -0400 Subject: [PATCH 4/4] docs: add backfill coverage report and backfill services Co-authored-by: Vichu Selvaraju --- agents/services.md | 4 +++ docs/backfill-coverage.md | 60 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+) create mode 100644 docs/backfill-coverage.md diff --git a/agents/services.md b/agents/services.md index a891b64..1a96be1 100644 --- a/agents/services.md +++ b/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, diff --git a/docs/backfill-coverage.md b/docs/backfill-coverage.md new file mode 100644 index 0000000..23b5573 --- /dev/null +++ b/docs/backfill-coverage.md @@ -0,0 +1,60 @@ +# Backfill coverage + +What the historical backfill (ticket 37) actually got, per source. Numbers are +from a local Docker Postgres run on 2026-10-12; re-run on Neon once it's set up. + +Hotel data isn't loaded yet, so these runs used explicit `--start`/`--end` +instead of the hotel window. Once a hotel's nights are in `daily_performance`, +pass `--hotel-id` and leave the dates out: every script defaults to that +hotel's first → last `stay_date`. + +## Summary + +| Source | Market | Requested | Earliest usable | Rows | Known gaps | +|---|---|---|---|---|---| +| Weather (Open-Meteo) | Dublin | 2020-01-01 → 2026-10-11 | 2020-01-01 | 2,476 | none | +| Events (PredictHQ) | Dublin, 10 km | 2026-01-01 → 2026-10-11 | 2026-07-01 | 2,666 | everything before 2026-07-01 | +| Flights (AeroDataBox) | DUB | not run | — | — | no API key yet | + +## Weather — Open-Meteo Historical API + +- Recorded actuals (`is_forecast = false`), one row per city and day. +- The archive goes back to 1940, so it covers any hotel window. +- 2,476 of 2,476 days saved; no missing days and no days without values. +- Re-running the full range kept the count at 2,476 (zero duplicates). +- Requests are one year at a time, retrying 429/5xx/network errors (2s, 4s, 8s). +- Free tier is for non-commercial use; a commercial deployment needs a paid plan. + +## Events — PredictHQ + +- 2,666 events: 2,488 active, 178 that PredictHQ marked deleted (cancelled). +- **Trial floor: 2026-07-01.** Requesting from 2026-01-01 returned nothing + before July, so the trial gives about 3½ months of history. Covering an + older hotel window needs a paid plan. +- Re-running September: `new=0 unchanged=987` (zero duplicates). +- Monthly chunks, retries on 429/5xx with `Retry-After`, stops cleanly and + prints where to resume if credits run out. + +## Flights — AeroDataBox (via RapidAPI) + +- **Not run: no `XRAPIDAPI_KEY` yet.** +- Each day costs about 4 API units; the free plan has 400 a month, so roughly + 100 days per month on the free tier. +- History floor not confirmed yet — check the RapidAPI plan and record it here. +- Retries 429/5xx (2s, 4s, 8s); a 429 after that is treated as the monthly + quota, and the run stops with `resume with --start `. + +## Re-running + +From `backend/`, with `DATABASE_URL` pointing at the target database: + + uv run python scripts/backfill/backfill_weather.py --hotel-id 1 + uv run python scripts/backfill/weather_coverage.py --city Dublin --hotel-id 1 + uv run python scripts/backfill/backfill_events.py --hotel-id 1 + uv run python scripts/backfill/backfill_flights.py --hotel-id 1 --airport DUB + +Leave out `--start`/`--end` to use the hotel's window. Without a hotel, give +`--city --lat --lon` (weather, events) and both dates. + +Events commit after each month and flights after each day, so a run stopped by +a quota keeps everything before it; re-run with the printed `--start`.