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/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/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/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} 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/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/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_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_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_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)) + ] 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} 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`.