feat: ingest events from PredictHQ and AskNews onto a shared normalized form

.env.example

@@ -5,3 +5,8 @@ DEBUG=false
 # Database
 # Format: postgresql+asyncpg://<user>:<password>@<host>:<port>/<db_name>
 DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/dbname
+
+# External event sources
+PREDICTHQ_TOKEN=
+ASKNEWS_CLIENT_ID=
+ASKNEWS_CLIENT_SECRET=

backend/foresight/config.py

@@ -3,7 +3,8 @@
 
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(
-        env_file=".env",
+        # Root .env is the one docker-compose reads; backend/.env overrides it.
+        env_file=("../.env", ".env"),
         env_file_encoding="utf-8",
         extra="ignore",
     )
@@ -17,5 +18,10 @@ class Settings(BaseSettings):
         "postgresql+asyncpg://foresight:foresight@localhost:5432/foresight_db"
     )
 
+    # External event sources
+    predicthq_token: str = ""
+    asknews_client_id: str = ""
+    asknews_client_secret: str = ""
+
 
 settings = Settings()

backend/foresight/database/migrations/versions/b7d41e0c9a52_canonical_events_and_observations.py

@@ -0,0 +1,93 @@
+"""canonical events and observations
+
+Replaces the placeholder `events` table with the canonical event contract plus
+an append-only `event_observations` staging table. Nothing writes to `events`
+yet, so it is dropped rather than migrated.
+
+Revision ID: b7d41e0c9a52
+Revises: 72e850f4fa7b
+Create Date: 2026-10-05 11:12:04.882341
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'b7d41e0c9a52'
+down_revision: Union[str, None] = '72e850f4fa7b'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.drop_table('events')
+    op.create_table('events',
+    sa.Column('id', sa.Integer(), nullable=False),
+    sa.Column('dedupe_key', sa.String(length=64), nullable=False),
+    sa.Column('title', sa.String(length=300), nullable=False),
+    sa.Column('title_norm', sa.String(length=300), nullable=False),
+    sa.Column('category', sa.String(length=40), nullable=False),
+    sa.Column('venue', sa.String(length=200), nullable=True),
+    sa.Column('city', sa.String(length=120), nullable=False),
+    sa.Column('city_slug', sa.String(length=120), nullable=False),
+    sa.Column('country', sa.String(length=2), nullable=True),
+    sa.Column('latitude', sa.Float(), nullable=True),
+    sa.Column('longitude', sa.Float(), nullable=True),
+    sa.Column('start_local_date', sa.Date(), nullable=False),
+    sa.Column('end_local_date', sa.Date(), nullable=False),
+    sa.Column('start_at_utc', sa.DateTime(timezone=True), nullable=True),
+    sa.Column('end_at_utc', sa.DateTime(timezone=True), nullable=True),
+    sa.Column('timezone', sa.String(length=64), nullable=False),
+    sa.Column('date_precision', sa.String(length=20), nullable=False),
+    sa.Column('expected_attendance', sa.Integer(), nullable=True),
+    sa.Column('impact_rank', sa.Integer(), nullable=True),
+    sa.Column('local_rank', sa.Integer(), nullable=True),
+    sa.Column('confidence', sa.Float(), nullable=False),
+    sa.Column('status', sa.String(length=20), nullable=False),
+    sa.Column('url', sa.String(length=600), nullable=True),
+    sa.Column('primary_source', sa.String(length=40), nullable=False),
+    sa.Column('first_seen_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.Column('last_seen_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('dedupe_key')
+    )
+    op.create_index(op.f('ix_events_title_norm'), 'events', ['title_norm'], unique=False)
+    op.create_index('ix_events_city_slug_start', 'events', ['city_slug', 'start_local_date'], unique=False)
+    op.create_table('event_observations',
+    sa.Column('id', sa.Integer(), nullable=False),
+    sa.Column('event_id', sa.Integer(), nullable=True),
+    sa.Column('source', sa.String(length=40), nullable=False),
+    sa.Column('source_ref', sa.String(length=600), nullable=False),
+    sa.Column('source_updated_at', sa.DateTime(timezone=True), nullable=True),
+    sa.Column('payload', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+    sa.Column('captured_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.ForeignKeyConstraint(['event_id'], ['events.id'], ),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('source', 'source_ref')
+    )
+
+
+def downgrade() -> None:
+    op.drop_table('event_observations')
+    op.drop_index('ix_events_city_slug_start', table_name='events')
+    op.drop_index(op.f('ix_events_title_norm'), table_name='events')
+    op.drop_table('events')
+    op.create_table('events',
+    sa.Column('id', sa.Integer(), nullable=False),
+    sa.Column('name', sa.String(length=300), nullable=False),
+    sa.Column('category', sa.String(length=60), nullable=True),
+    sa.Column('venue', sa.String(length=200), nullable=True),
+    sa.Column('city', sa.String(length=120), nullable=False),
+    sa.Column('start_date', sa.Date(), nullable=False),
+    sa.Column('end_date', sa.Date(), nullable=False),
+    sa.Column('expected_attendance', sa.Integer(), nullable=True),
+    sa.Column('latitude', sa.Float(), nullable=True),
+    sa.Column('longitude', sa.Float(), nullable=True),
+    sa.Column('source', sa.String(length=40), nullable=False),
+    sa.Column('captured_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.PrimaryKeyConstraint('id')
+    )

backend/foresight/database/migrations/versions/c3e9a1f4b6d8_observation_status.py

@@ -0,0 +1,42 @@
+"""observation status
+
+Each sighting keeps the status its source reported, so a deleted duplicate
+listing can be told apart from the live listing it shares an event with.
+Existing PredictHQ sightings are backfilled from their payload.
+
+Revision ID: c3e9a1f4b6d8
+Revises: b7d41e0c9a52
+Create Date: 2026-10-06 22:30:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'c3e9a1f4b6d8'
+down_revision: Union[str, None] = 'b7d41e0c9a52'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        'event_observations', sa.Column('status', sa.String(length=20), nullable=True)
+    )
+    op.execute(
+        """
+        UPDATE event_observations SET status = CASE payload->>'state'
+            WHEN 'active' THEN 'ACTIVE'
+            WHEN 'predicted' THEN 'PROVISIONAL'
+            WHEN 'deleted' THEN 'DELETED'
+        END
+        WHERE source = 'PREDICTHQ'
+        """
+    )
+
+
+def downgrade() -> None:
+    op.drop_column('event_observations', 'status')

backend/foresight/database/models/__init__.py

@@ -4,6 +4,7 @@
 from foresight.database.models.external import (
     CompetitorRate,
     Event,
+    EventObservation,
     FlightArrival,
     Weather,
 )
@@ -14,6 +15,7 @@
     "CompetitorRate",
     "DailyPerformance",
     "Event",
+    "EventObservation",
     "FlightArrival",
     "Hotel",
     "RoomType",

backend/foresight/database/models/external.py

@@ -2,11 +2,29 @@
 
 from datetime import date, datetime
 from decimal import Decimal
-
-from sqlalchemy import Date, DateTime, Numeric, String, func
-from sqlalchemy.orm import Mapped, mapped_column
+from typing import Any
+
+from sqlalchemy import (
+    Date,
+    DateTime,
+    ForeignKey,
+    Index,
+    Numeric,
+    String,
+    UniqueConstraint,
+    func,
+)
+from sqlalchemy import Enum as SAEnum
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from foresight.database.base import Base
+from foresight.ingest.normalize import (
+    DatePrecision,
+    EventCategory,
+    EventStatus,
+    Source,
+)
 
 
 class Weather(Base):
@@ -36,29 +54,109 @@ class Weather(Base):
 
 
 class Event(Base):
-    """Something happening nearby that drives demand."""
+    """One real-world event occurrence, merged across sources.
+
+    Identified by `dedupe_key` (normalized title + city + local start date) so
+    the same festival seen by PredictHQ and by AskNews lands on one row.
+    """
 
     __tablename__ = "events"
 
     id: Mapped[int] = mapped_column(primary_key=True)
-    name: Mapped[str] = mapped_column(String(300))
-    category: Mapped[str | None] = mapped_column(String(60))
+    dedupe_key: Mapped[str] = mapped_column(String(64), unique=True)
+
+    title: Mapped[str] = mapped_column(String(300))
+    title_norm: Mapped[str] = mapped_column(String(300), index=True)
+    category: Mapped[EventCategory] = mapped_column(
+        SAEnum(EventCategory, native_enum=False, length=40),
+        default=EventCategory.UNKNOWN,
+    )
+
     venue: Mapped[str | None] = mapped_column(String(200))
     city: Mapped[str] = mapped_column(String(120))
+    city_slug: Mapped[str] = mapped_column(String(120))
+    country: Mapped[str | None] = mapped_column(String(2))  # ISO-3166 alpha-2
+    latitude: Mapped[float | None]
+    longitude: Mapped[float | None]
 
-    start_date: Mapped[date] = mapped_column(Date)
-    end_date: Mapped[date] = mapped_column(Date)
+    # Local dates, not UTC: a stay-night is local to the hotel. Inclusive range.
+    start_local_date: Mapped[date] = mapped_column(Date)
+    end_local_date: Mapped[date] = mapped_column(Date)
+    start_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    end_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    timezone: Mapped[str] = mapped_column(String(64), default="UTC")
+    date_precision: Mapped[DatePrecision] = mapped_column(
+        SAEnum(DatePrecision, native_enum=False, length=20),
+        default=DatePrecision.DAY,
+    )
 
     expected_attendance: Mapped[int | None]
+    impact_rank: Mapped[int | None]  # PredictHQ rank, 0-100
+    local_rank: Mapped[int | None]  # PredictHQ local_rank, 0-100
+    confidence: Mapped[float] = mapped_column(default=1.0)
 
-    latitude: Mapped[float | None]
-    longitude: Mapped[float | None]
+    status: Mapped[EventStatus] = mapped_column(
+        SAEnum(EventStatus, native_enum=False, length=20),
+        default=EventStatus.ACTIVE,
+    )
+    url: Mapped[str | None] = mapped_column(String(600))
+
+    # Which source currently owns these values, by precedence.
+    primary_source: Mapped[Source] = mapped_column(
+        SAEnum(Source, native_enum=False, length=40)
+    )
+
+    first_seen_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )
+    last_seen_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )
+
+    observations: Mapped[list["EventObservation"]] = relationship(
+        back_populates="event"
+    )
+
+    # Serves the core lookup: what is happening near this hotel on this night.
+    __table_args__ = (
+        Index("ix_events_city_slug_start", "city_slug", "start_local_date"),
+    )
+
+
+class EventObservation(Base):
+    """One source's raw sighting of an event, kept verbatim.
+
+    Append-only and keyed on (source, source_ref) -- the PredictHQ event id or
+    the AskNews article URL. This is what makes re-runs idempotent and lets a
+    revised upstream record be recognised rather than duplicated.
+
+    `event_id` is null when a sighting cannot be resolved to a dated event, e.g.
+    an article that announces a festival without saying when it is.
+    """
+
+    __tablename__ = "event_observations"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    event_id: Mapped[int | None] = mapped_column(ForeignKey("events.id"))
+
+    source: Mapped[Source] = mapped_column(SAEnum(Source, native_enum=False, length=40))
+    source_ref: Mapped[str] = mapped_column(String(600))
+    source_updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    # What this sighting reported; null only for rows predating the column.
+    status: Mapped[EventStatus | None] = mapped_column(
+        SAEnum(EventStatus, native_enum=False, length=20)
+    )
+
+    payload: Mapped[dict[str, Any]] = mapped_column(JSONB)
 
-    source: Mapped[str] = mapped_column(String(40))
     captured_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now()
     )
 
+    event: Mapped["Event | None"] = relationship(back_populates="observations")
+
+    __table_args__ = (UniqueConstraint("source", "source_ref"),)
+
 
 class CompetitorRate(Base):
     """What another hotel is charging for a given night."""

backend/foresight/ingest/__init__.py

@@ -0,0 +1 @@
+"""Ingestion of external data sources: events, weather, flights, competitor rates."""