.env.example

@@ -27,6 +27,17 @@ FINNHUB_API_KEY=
 # Fundamentals Provider — Alpha Vantage (optional fallback)
 ALPHA_VANTAGE_API_KEY=
 
+# Dolt bulk data — local clone of post-no-preference/earnings (workstream A).
+# DOLT_BINARY: path to the dolt CLI (set the full path in dev if it's not on PATH,
+#   e.g. Windows: C:\Program Files\Dolt\bin\dolt.exe). DOLT_DATA_DIR holds the
+#   clones; in PRODUCTION it MUST be outside the deploy tree (deploy is
+#   rsync --delete) — e.g. /var/lib/signal-platform/dolt. The earnings clone lives
+#   at <DOLT_DATA_DIR>/<DOLT_EARNINGS_SUBDIR>. Set up: dolt clone post-no-preference/earnings <dir>.
+DOLT_BINARY=dolt
+DOLT_DATA_DIR=dolt-data
+DOLT_EARNINGS_SUBDIR=earnings
+DOLT_MIN_FREE_DISK_GB=2.0
+
 # Regime Monitor — FRED (VIX + HY credit spreads). Free key: https://fred.stlouisfed.org/docs/api/api_key.html
 # Optional: without it the volatility (V1) and credit (C1) pillars show as n/a.
 FRED_API_KEY=

NOTICE

@@ -0,0 +1,24 @@
+Third-party data attribution
+============================
+
+Earnings calendar and EPS history
+---------------------------------
+This application ingests the earnings calendar and EPS surprise history from the
+public DoltHub repository:
+
+    post-no-preference/earnings
+    https://www.dolthub.com/repositories/post-no-preference/earnings
+
+Licensed under Creative Commons Attribution-ShareAlike 4.0 International
+(CC BY-SA 4.0): https://creativecommons.org/licenses/by-sa/4.0/
+
+Use in this project: private, internal ingestion only. The data is normalized
+into PostgreSQL (`earnings_events`) — the announcement calendar is aligned to the
+EPS history via a minimum-cost monotonic pairing, symbols are normalized, and the
+session field is mapped to bmo/amc/unknown. No public API, bulk export, or
+redistribution of the data is provided. This attribution and the upstream license
+are preserved per the CC BY-SA 4.0 terms. Re-review licensing before any public
+or commercial access.
+
+The post-no-preference/stocks repository (workstream B) is not used at this time
+and would be reviewed separately.

app/config.py

@@ -37,6 +37,16 @@ class Settings(BaseSettings):
     # Fundamentals Provider — Alpha Vantage (optional fallback)
     alpha_vantage_api_key: str = ""
 
+    # Dolt bulk-data — local clone of post-no-preference/earnings (workstream A).
+    # dolt_binary: full path when not on PATH (dev/Windows install). dolt_data_dir
+    # holds the clones; in production it MUST be outside the deploy tree (deploy is
+    # rsync --delete) — set DOLT_DATA_DIR to a persistent path. The earnings clone
+    # lives at <dolt_data_dir>/<dolt_earnings_subdir>.
+    dolt_binary: str = "dolt"
+    dolt_data_dir: str = "dolt-data"
+    dolt_earnings_subdir: str = "earnings"
+    dolt_min_free_disk_gb: float = 2.0
+
     # Regime Monitor — FRED (VIX level + HY credit spreads). Optional: without it
     # the volatility (P5) and credit-spread (F2) signals are reported as n/a.
     fred_api_key: str = ""

app/services/dolt_client.py

@@ -0,0 +1,83 @@
+"""Minimal async client for a local Dolt clone.
+
+The application never runs a long-lived Dolt sql-server; it shells out to the
+`dolt` CLI against a persistent clone and reads results as CSV. Every call goes
+through ``asyncio.create_subprocess_exec`` because the scheduler shares one event
+loop with the API (`app/scheduler.py:73`) — a blocking `subprocess.run` here
+would stall request handling.
+
+Production keeps the clone in ``DOLT_DATA_DIR`` outside the deploy tree; the
+binary path and data dir are configured (see ``app/config.py``). Read via
+``dolt sql -r csv``; refresh with ``pull`` and record the resulting commit hash
+as the import revision.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import csv
+import io
+import logging
+import shutil
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+
+class DoltError(RuntimeError):
+    """A dolt subprocess exited non-zero."""
+
+
+async def _run(binary: str, args: list[str], *, cwd: Path) -> str:
+    proc = await asyncio.create_subprocess_exec(
+        binary,
+        *args,
+        cwd=str(cwd),
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    stdout, stderr = await proc.communicate()
+    if proc.returncode != 0:
+        raise DoltError(
+            f"dolt {' '.join(args)} failed ({proc.returncode}): "
+            f"{stderr.decode('utf-8', 'replace').strip()[:500]}"
+        )
+    return stdout.decode("utf-8", "replace")
+
+
+def ensure_free_disk(path: Path, min_free_gb: float) -> None:
+    """Raise if free space at ``path`` is below the threshold (checked before a
+    pull that could grow the clone). Uses the nearest existing ancestor so it
+    works before the clone dir exists."""
+    probe = path
+    while not probe.exists() and probe.parent != probe:
+        probe = probe.parent
+    free_gb = shutil.disk_usage(probe).free / (1024**3)
+    if free_gb < min_free_gb:
+        raise DoltError(
+            f"insufficient disk for dolt at {path}: {free_gb:.1f} GB free "
+            f"< {min_free_gb:.1f} GB required"
+        )
+
+
+async def pull(repo_dir: Path, *, binary: str) -> None:
+    """`dolt pull` the persistent clone to the latest upstream revision."""
+    await _run(binary, ["pull"], cwd=repo_dir)
+
+
+async def current_commit(repo_dir: Path, *, binary: str) -> str:
+    """The HEAD commit hash of the clone — used as the import revision."""
+    rows = await query_csv(
+        repo_dir, "SELECT commit_hash FROM dolt_log ORDER BY date DESC LIMIT 1", binary=binary
+    )
+    if not rows or not rows[0].get("commit_hash"):
+        raise DoltError("could not read HEAD commit hash from dolt_log")
+    return rows[0]["commit_hash"]
+
+
+async def query_csv(repo_dir: Path, sql: str, *, binary: str) -> list[dict[str, str]]:
+    """Run a read query and parse the CSV result into a list of dict rows."""
+    out = await _run(binary, ["sql", "-q", sql, "-r", "csv"], cwd=repo_dir)
+    if not out.strip():
+        return []
+    return list(csv.DictReader(io.StringIO(out)))

app/services/dolt_earnings_importer.py

@@ -0,0 +1,316 @@
+"""Production importer for the DoltHub post-no-preference/earnings calendar.
+
+A ``SourceImporter`` (see ``app/services/data_import.py``) that pulls the local
+Dolt clone, aligns the announcement calendar to the EPS history with the pure DP
+in ``earnings_alignment`` (reused from the research script, not extending it),
+and writes ``earnings_events`` for the tracked universe.
+
+Shadow by construction: nothing reads ``earnings_events`` until the API/panel
+lands (A4), so writing it does not touch production behavior.
+
+**Promotion is destructive** — future-dated rows for this source are deleted and
+re-inserted every run so reschedules/cancellations never linger. The forward
+calendar is the project's acceptance gate, so ``validate`` is fail-closed: it
+blocks promotion when the staged future set is empty or has collapsed relative
+to what's already loaded.
+
+Attribution: the earnings data is CC BY-SA 4.0 from post-no-preference/earnings.
+See the repo ``NOTICE``. Internal use only — no redistribution.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from datetime import date, datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import case, delete, func, select
+
+from app.config import settings
+from app.database import insert_for_session
+from app.models.earnings_event import EarningsEvent
+from app.models.ticker import Ticker
+from app.services import dolt_client, earnings_alignment
+from app.services.data_import import ValidationResult
+
+logger = logging.getLogger(__name__)
+
+SOURCE = "dolt_earnings"
+
+# Earliest announcement date to import (matches the research backfill window).
+WINDOW_START = date(2020, 1, 22)
+# Alignment tolerances (research defaults): an announcement may lead its period
+# end by up to 14 days or lag it by up to 90.
+MAX_LAG_DAYS = 90
+MAX_LEAD_DAYS = 14
+# Fail promotion if the staged forward calendar drops below this fraction of the
+# currently-loaded forward calendar (guards the destructive re-insert against a
+# partial parse / symbol-mapping regression).
+MIN_FUTURE_RATIO = 0.5
+
+_CAL_SQL = (
+    "SELECT act_symbol, `date`, `when` FROM earnings_calendar "
+    f"WHERE `date` >= '{WINDOW_START.isoformat()}'"
+)
+_HIST_SQL = (
+    "SELECT act_symbol, period_end_date, reported, estimate FROM eps_history "
+    f"WHERE period_end_date >= '{(WINDOW_START.replace(year=WINDOW_START.year - 1)).isoformat()}'"
+)
+
+
+@dataclass
+class StagedEarnings:
+    rows: list[dict[str, Any]]
+    stats: dict[str, Any] = field(default_factory=dict)
+    future_count: int = 0
+    max_announce_date: date | None = None
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class DoltEarningsImporter:
+    source = SOURCE
+
+    def __init__(
+        self,
+        *,
+        repo_dir: Path | str | None = None,
+        binary: str | None = None,
+        today: date | None = None,
+        do_pull: bool = True,
+        dolt: Any = dolt_client,
+    ) -> None:
+        self.repo_dir = Path(
+            repo_dir
+            or (Path(settings.dolt_data_dir) / settings.dolt_earnings_subdir)
+        )
+        self.binary = binary or settings.dolt_binary
+        self.today = today or _now().date()
+        self.do_pull = do_pull
+        self._dolt = dolt  # injectable for tests
+
+    # -- SourceImporter protocol -------------------------------------------
+
+    async def detect_revision(self, db) -> str | None:
+        if self.do_pull:
+            dolt_client.ensure_free_disk(self.repo_dir, settings.dolt_min_free_disk_gb)
+            await self._dolt.pull(self.repo_dir, binary=self.binary)
+        return await self._dolt.current_commit(self.repo_dir, binary=self.binary)
+
+    async def stage(self, db) -> StagedEarnings:
+        universe = await self._load_universe(db)  # {normalised symbol: ticker_id}
+
+        cal_raw = await self._dolt.query_csv(self.repo_dir, _CAL_SQL, binary=self.binary)
+        hist_raw = await self._dolt.query_csv(self.repo_dir, _HIST_SQL, binary=self.binary)
+        _require_columns(cal_raw, {"act_symbol", "date", "when"}, "earnings_calendar")
+        _require_columns(
+            hist_raw, {"act_symbol", "period_end_date", "reported", "estimate"}, "eps_history"
+        )
+
+        cal_parsed = _parse_calendar(cal_raw, universe)
+        hist_parsed = _parse_history(hist_raw, universe)
+        calendar, cal_stats = earnings_alignment.dedup_calendar(cal_parsed)
+        history, hist_stats = earnings_alignment.dedup_history(hist_parsed)
+
+        period_lower = WINDOW_START.replace(year=WINDOW_START.year - 1)
+        rows: list[dict[str, Any]] = []
+        matched = unmatched = 0
+        for symbol, events in calendar.items():
+            ticker_id = universe[symbol]
+            periods = [
+                p for p in history.get(symbol, []) if p["period_end_date"] >= period_lower
+            ]
+            matches, unmatched_events, _ = earnings_alignment.align_symbol(
+                events, periods, max_lag_days=MAX_LAG_DAYS, max_lead_days=MAX_LEAD_DAYS
+            )
+            matched += len(matches)
+            unmatched += len(unmatched_events)
+            matched_by_event = {e: p for e, p in matches}
+            for e_idx, event in enumerate(events):
+                p_idx = matched_by_event.get(e_idx)
+                period = periods[p_idx] if p_idx is not None else None
+                rows.append(
+                    {
+                        "ticker_id": ticker_id,
+                        "symbol": symbol,
+                        "announce_date": event["announce_date"],
+                        "session": event["session"],
+                        "period_end": period["period_end_date"] if period else None,
+                        "eps_estimate": period["eps_estimate"] if period else None,
+                        "eps_actual": period["eps_actual"] if period else None,
+                    }
+                )
+
+        future_rows = [r for r in rows if r["announce_date"] > self.today]
+        tickers_with_future = {r["ticker_id"] for r in future_rows}
+        stats = {
+            "calendar": cal_stats,
+            "eps_history": hist_stats,
+            "universe_size": len(universe),
+            "symbols_with_calendar": len(calendar),
+            "matched_events": matched,
+            "unmatched_events": unmatched,
+            "tracked_tickers_with_future_date": len(tickers_with_future),
+        }
+        return StagedEarnings(
+            rows=rows,
+            stats=stats,
+            future_count=len(future_rows),
+            max_announce_date=max((r["announce_date"] for r in rows), default=None),
+        )
+
+    async def validate(self, db, staged: StagedEarnings) -> ValidationResult:
+        messages: list[str] = []
+
+        # Fail-closed forward-calendar protection (promote deletes+reinserts it).
+        if staged.future_count == 0:
+            messages.append("no future-dated earnings rows staged")
+        current_future = await self._current_future_count(db)
+        if current_future > 0 and staged.future_count < current_future * MIN_FUTURE_RATIO:
+            messages.append(
+                f"forward calendar collapsed: staged {staged.future_count} future rows "
+                f"< {MIN_FUTURE_RATIO:.0%} of current {current_future}"
+            )
+
+        keys = [(r["ticker_id"], r["announce_date"]) for r in staged.rows]
+        if len(keys) != len(set(keys)):
+            messages.append("duplicate (ticker_id, announce_date) in staged set")
+
+        summary = {
+            **staged.stats,
+            "staged_rows": len(staged.rows),
+            "future_rows": staged.future_count,
+            "current_future_rows": current_future,
+        }
+        return ValidationResult(
+            ok=not messages,
+            summary=summary,
+            source_max_date=staged.max_announce_date,
+            messages=messages,
+        )
+
+    async def promote(self, db, staged: StagedEarnings, run_id: int) -> dict[str, int]:
+        # Rescheduling: drop this source's future rows, then upsert the staged
+        # set. Past rows (results) are never deleted; moved/cancelled future
+        # dates simply don't reappear.
+        deleted = (
+            await db.execute(
+                delete(EarningsEvent).where(
+                    EarningsEvent.source == SOURCE,
+                    EarningsEvent.announce_date > self.today,
+                )
+            )
+        ).rowcount or 0
+
+        now = _now()
+        for r in staged.rows:
+            stmt = insert_for_session(db, EarningsEvent).values(
+                ticker_id=r["ticker_id"],
+                announce_date=r["announce_date"],
+                session=r["session"],
+                period_end=r["period_end"],
+                eps_estimate=r["eps_estimate"],
+                eps_actual=r["eps_actual"],
+                source=SOURCE,
+                import_run_id=run_id,
+                created_at=now,
+            )
+            # Preserve a non-null prior EPS/period-end if a re-pairing comes back
+            # null; prefer a known session over 'unknown'.
+            stmt = stmt.on_conflict_do_update(
+                index_elements=["ticker_id", "announce_date"],
+                set_={
+                    "session": case(
+                        (stmt.excluded.session != "unknown", stmt.excluded.session),
+                        else_=EarningsEvent.session,
+                    ),
+                    "period_end": func.coalesce(
+                        stmt.excluded.period_end, EarningsEvent.period_end
+                    ),
+                    "eps_estimate": func.coalesce(
+                        stmt.excluded.eps_estimate, EarningsEvent.eps_estimate
+                    ),
+                    "eps_actual": func.coalesce(
+                        stmt.excluded.eps_actual, EarningsEvent.eps_actual
+                    ),
+                    "source": stmt.excluded.source,
+                    "import_run_id": stmt.excluded.import_run_id,
+                },
+            )
+            await db.execute(stmt)
+
+        return {"deleted_future": int(deleted), "upserted": len(staged.rows)}
+
+    # -- helpers -----------------------------------------------------------
+
+    async def _load_universe(self, db) -> dict[str, int]:
+        rows = (await db.execute(select(Ticker.id, Ticker.symbol))).all()
+        return {
+            earnings_alignment.normalise_symbol(symbol): tid
+            for tid, symbol in rows
+            if symbol
+        }
+
+    async def _current_future_count(self, db) -> int:
+        return (
+            await db.execute(
+                select(func.count())
+                .select_from(EarningsEvent)
+                .where(
+                    EarningsEvent.source == SOURCE,
+                    EarningsEvent.announce_date > self.today,
+                )
+            )
+        ).scalar_one()
+
+
+def _require_columns(rows: list[dict[str, str]], required: set[str], table: str) -> None:
+    """Upstream schema-change gate: a missing column stops the run (→ failed)."""
+    if not rows:
+        return
+    present = set(rows[0].keys())
+    missing = required - present
+    if missing:
+        raise ValueError(f"{table}: upstream schema change, missing columns {sorted(missing)}")
+
+
+def _parse_calendar(raw: list[dict[str, str]], universe: dict[str, int]) -> list[dict[str, Any]]:
+    out: list[dict[str, Any]] = []
+    for row in raw:
+        symbol = earnings_alignment.normalise_symbol(row.get("act_symbol"))
+        raw_date = str(row.get("date") or "")[:10]
+        if symbol not in universe or not raw_date:
+            continue
+        announce_date = date.fromisoformat(raw_date)
+        if announce_date < WINDOW_START:
+            continue
+        out.append(
+            {
+                "symbol": symbol,
+                "announce_date": announce_date,
+                "session": earnings_alignment.normalise_session(row.get("when")),
+            }
+        )
+    return out
+
+
+def _parse_history(raw: list[dict[str, str]], universe: dict[str, int]) -> list[dict[str, Any]]:
+    out: list[dict[str, Any]] = []
+    for row in raw:
+        symbol = earnings_alignment.normalise_symbol(row.get("act_symbol"))
+        raw_date = str(row.get("period_end_date") or "")[:10]
+        if symbol not in universe or not raw_date:
+            continue
+        out.append(
+            {
+                "symbol": symbol,
+                "period_end_date": date.fromisoformat(raw_date),
+                "eps_actual": earnings_alignment.safe_number(row.get("reported")),
+                "eps_estimate": earnings_alignment.safe_number(row.get("estimate")),
+            }
+        )
+    return out

app/services/earnings_alignment.py

@@ -0,0 +1,207 @@
+"""Pure calendar<->EPS-history alignment for the DoltHub earnings source.
+
+The earnings repo keeps the announcement calendar (`earnings_calendar`) and the
+reported/estimate EPS history (`eps_history`) in separate tables with no shared
+key — the calendar has announce dates, the history has period-end dates. This
+module reproduces the research importer's **minimum-cost monotonic alignment**
+(`scripts/import_dolthub_earnings.py`) as pure, DB-free, unit-testable functions
+so the production importer can reuse it without extending that one-off script.
+
+Constants and cost function are kept identical to the research script; the DP is
+what pairs each announcement with the quarter it reported, tolerating gaps on
+either side. Do not tune these without re-validating surprise-history pairing.
+"""
+
+from __future__ import annotations
+
+import math
+from collections import defaultdict
+from datetime import date
+from typing import Any
+
+# Alignment costs — identical to scripts/import_dolthub_earnings.py.
+SKIP_EVENT_COST = 45.0
+SKIP_PERIOD_COST = 45.0
+_TYPICAL_ANNOUNCE_LAG_DAYS = 30  # announcements land ~a month after period end
+_MISSING_SESSION_PENALTY = 3.0
+
+# Session normalization → the three values the schema/API promise.
+_SESSION_ALIASES = {
+    "before market open": "bmo",
+    "before open": "bmo",
+    "bmo": "bmo",
+    "after market close": "amc",
+    "after close": "amc",
+    "amc": "amc",
+}
+
+
+def normalise_symbol(value: Any) -> str:
+    """Upper-case, trim, and map dots to dashes so the DoltHub `act_symbol`
+    (`BF.B`) and the app's `tickers.symbol` join after the same normalization."""
+    return str(value or "").strip().upper().replace(".", "-")
+
+
+def normalise_session(value: Any) -> str:
+    """Map the source `when` text to bmo | amc | unknown. Anything not clearly a
+    pre-open or post-close session (including 'during market hours' and blanks)
+    collapses to 'unknown' — the schema/API only promise those three."""
+    cleaned = str(value or "").strip().lower().replace("_", " ").replace("-", " ")
+    return _SESSION_ALIASES.get(cleaned, "unknown")
+
+
+def safe_number(value: Any) -> float | None:
+    if value is None or str(value).strip() == "":
+        return None
+    try:
+        result = float(value)
+    except (TypeError, ValueError):
+        return None
+    return result if math.isfinite(result) else None
+
+
+def dedup_calendar(
+    rows: list[dict[str, Any]],
+) -> tuple[dict[str, list[dict[str, Any]]], dict[str, int]]:
+    """Collapse to one row per (symbol, announce_date), preferring a known
+    session over 'unknown'. Rows must be pre-parsed:
+    {symbol, announce_date: date, session}. Returns {symbol: [events sorted by
+    date]} and dedup stats."""
+    by_key: dict[tuple[str, date], dict[str, Any]] = {}
+    duplicate_rows = 0
+    restated_rows = 0
+    for row in rows:
+        key = (row["symbol"], row["announce_date"])
+        previous = by_key.get(key)
+        if previous is None:
+            by_key[key] = row
+            continue
+        duplicate_rows += 1
+        prev_known = previous["session"] != "unknown"
+        new_known = row["session"] != "unknown"
+        if prev_known and new_known and previous["session"] != row["session"]:
+            restated_rows += 1
+        # Prefer a row that carries a known session.
+        if new_known:
+            by_key[key] = row
+    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for row in by_key.values():
+        grouped[row["symbol"]].append(row)
+    for events in grouped.values():
+        events.sort(key=lambda item: item["announce_date"])
+    return grouped, {
+        "deduped_rows": len(by_key),
+        "duplicate_rows": duplicate_rows,
+        "restated_rows": restated_rows,
+    }
+
+
+def dedup_history(
+    rows: list[dict[str, Any]],
+) -> tuple[dict[str, list[dict[str, Any]]], dict[str, int]]:
+    """Collapse to one row per (symbol, period_end_date), preferring the row with
+    more non-null EPS fields. Rows must be pre-parsed:
+    {symbol, period_end_date: date, eps_actual, eps_estimate}."""
+    fields = ("eps_actual", "eps_estimate")
+    by_key: dict[tuple[str, date], dict[str, Any]] = {}
+    duplicate_rows = 0
+    restated_rows = 0
+    for row in rows:
+        key = (row["symbol"], row["period_end_date"])
+        previous = by_key.get(key)
+        if previous is None:
+            by_key[key] = row
+            continue
+        duplicate_rows += 1
+        if any(
+            previous.get(f) is not None
+            and row.get(f) is not None
+            and previous[f] != row[f]
+            for f in fields
+        ):
+            restated_rows += 1
+        prev_score = sum(previous.get(f) is not None for f in fields)
+        new_score = sum(row.get(f) is not None for f in fields)
+        if new_score >= prev_score:
+            by_key[key] = row
+    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for row in by_key.values():
+        grouped[row["symbol"]].append(row)
+    for periods in grouped.values():
+        periods.sort(key=lambda item: item["period_end_date"])
+    return grouped, {
+        "deduped_rows": len(by_key),
+        "duplicate_rows": duplicate_rows,
+        "restated_rows": restated_rows,
+    }
+
+
+def match_cost(event: dict[str, Any], period: dict[str, Any]) -> float:
+    delta = (event["announce_date"] - period["period_end_date"]).days
+    penalty = _MISSING_SESSION_PENALTY if event.get("session") == "unknown" else 0.0
+    return float(abs(delta - _TYPICAL_ANNOUNCE_LAG_DAYS)) + penalty
+
+
+def align_symbol(
+    events: list[dict[str, Any]],
+    periods: list[dict[str, Any]],
+    *,
+    max_lag_days: int,
+    max_lead_days: int,
+) -> tuple[list[tuple[int, int]], list[int], list[int]]:
+    """Minimum-cost monotonic calendar-to-period alignment for one symbol.
+
+    Both lists must be sorted ascending (by announce_date / period_end_date). A
+    match is allowed only when ``-max_lead_days <= announce_date - period_end <=
+    max_lag_days``. Returns (matches, unmatched_event_indices,
+    unmatched_period_indices).
+    """
+    n_events = len(events)
+    n_periods = len(periods)
+    scores = [[0.0] * (n_periods + 1) for _ in range(n_events + 1)]
+    choices = [[""] * (n_periods + 1) for _ in range(n_events + 1)]
+    for e in range(n_events - 1, -1, -1):
+        scores[e][n_periods] = scores[e + 1][n_periods] + SKIP_EVENT_COST
+        choices[e][n_periods] = "event"
+    for p in range(n_periods - 1, -1, -1):
+        scores[n_events][p] = scores[n_events][p + 1] + SKIP_PERIOD_COST
+        choices[n_events][p] = "period"
+
+    for e in range(n_events - 1, -1, -1):
+        for p in range(n_periods - 1, -1, -1):
+            options = [
+                (scores[e + 1][p] + SKIP_EVENT_COST, 2, "event"),
+                (scores[e][p + 1] + SKIP_PERIOD_COST, 1, "period"),
+            ]
+            delta = (events[e]["announce_date"] - periods[p]["period_end_date"]).days
+            if -max_lead_days <= delta <= max_lag_days:
+                options.append(
+                    (scores[e + 1][p + 1] + match_cost(events[e], periods[p]), 0, "match")
+                )
+            score, _, choice = min(options)
+            scores[e][p] = score
+            choices[e][p] = choice
+
+    matches: list[tuple[int, int]] = []
+    unmatched_events: list[int] = []
+    unmatched_periods: list[int] = []
+    e = p = 0
+    while e < n_events or p < n_periods:
+        if e >= n_events:
+            unmatched_periods.extend(range(p, n_periods))
+            break
+        if p >= n_periods:
+            unmatched_events.extend(range(e, n_events))
+            break
+        choice = choices[e][p]
+        if choice == "match":
+            matches.append((e, p))
+            e += 1
+            p += 1
+        elif choice == "period":
+            unmatched_periods.append(p)
+            p += 1
+        else:
+            unmatched_events.append(e)
+            e += 1
+    return matches, unmatched_events, unmatched_periods

tests/unit/test_dolt_earnings_importer.py

@@ -0,0 +1,262 @@
+"""Integration tests for the DoltHub earnings importer, driven through the real
+import framework with a fake dolt client (no subprocess, no clone).
+
+Covers the load-bearing behaviors: symbol-normalized join to the tracked
+universe, calendar<->history pairing (matched → EPS, unmatched → null), the
+destructive-but-safe reschedule/cancel promotion, past rows never deleted, and
+the fail-closed forward-calendar gates that guard the destructive promote.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import tempfile
+from datetime import date
+from pathlib import Path
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from app.database import Base
+import app.models  # noqa: F401
+from app.models.earnings_event import EarningsEvent
+from app.models.ticker import Ticker
+from app.services.data_import import STATUS_FAILED, STATUS_PROMOTED, run_import
+from app.services.dolt_earnings_importer import DoltEarningsImporter
+
+TODAY = date(2026, 7, 22)
+
+
+@pytest.fixture
+async def engine():
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    eng = create_async_engine(f"sqlite+aiosqlite:///{path}")
+    async with eng.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    try:
+        yield eng
+    finally:
+        await eng.dispose()
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+
+
+def _factory(engine):
+    return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+
+
+class FakeDolt:
+    def __init__(self, calendar, history, commit="c1"):
+        self.calendar = calendar
+        self.history = history
+        self.commit = commit
+        self.pulled = False
+
+    async def pull(self, repo_dir, *, binary):
+        self.pulled = True
+
+    async def current_commit(self, repo_dir, *, binary):
+        return self.commit
+
+    async def query_csv(self, repo_dir, sql, *, binary):
+        if "earnings_calendar" in sql:
+            return self.calendar
+        if "eps_history" in sql:
+            return self.history
+        return []
+
+
+def _cal(symbol, d, when="After market close"):
+    return {"act_symbol": symbol, "date": d, "when": when}
+
+
+def _hist(symbol, pe, reported, estimate):
+    return {"act_symbol": symbol, "period_end_date": pe, "reported": str(reported), "estimate": str(estimate)}
+
+
+def _importer(fake, commit=None):
+    if commit:
+        fake.commit = commit
+    return DoltEarningsImporter(
+        repo_dir="unused", binary="unused", today=TODAY, do_pull=False, dolt=fake
+    )
+
+
+async def _seed_tickers(factory, symbols):
+    async with factory() as s:
+        for sym in symbols:
+            s.add(Ticker(symbol=sym))
+        await s.commit()
+    async with factory() as s:
+        return {sym: tid for tid, sym in (await s.execute(select(Ticker.id, Ticker.symbol))).all()}
+
+
+async def _events(factory):
+    async with factory() as s:
+        rows = (
+            await s.execute(select(EarningsEvent).order_by(EarningsEvent.announce_date))
+        ).scalars().all()
+        return list(rows)
+
+
+# ---------------------------------------------------------------------------
+
+
+async def test_stage_and_promote_basic(engine):
+    factory = _factory(engine)
+    await _seed_tickers(factory, ["AAPL"])
+    fake = FakeDolt(
+        calendar=[_cal("AAPL", "2026-05-01"), _cal("AAPL", "2026-08-01")],
+        history=[_hist("AAPL", "2026-03-31", 1.5, 1.4)],  # only the reported quarter
+    )
+    run = await run_import(_importer(fake), engine=engine)
+
+    assert run.status == STATUS_PROMOTED
+    assert run.source_max_date == date(2026, 8, 1)
+
+    events = await _events(factory)
+    assert len(events) == 2
+    past = next(e for e in events if e.announce_date == date(2026, 5, 1))
+    future = next(e for e in events if e.announce_date == date(2026, 8, 1))
+    # past announcement paired to the reported quarter
+    assert past.eps_actual == 1.5 and past.eps_estimate == 1.4
+    assert past.period_end == date(2026, 3, 31) and past.session == "amc"
+    assert past.import_run_id == run.id
+    # future announcement has no results yet → null EPS/period, session kept
+    assert future.eps_actual is None and future.period_end is None
+    assert future.session == "amc"
+
+
+async def test_symbol_normalisation_join(engine):
+    factory = _factory(engine)
+    ids = await _seed_tickers(factory, ["AAPL", "BRK.B"])
+    fake = FakeDolt(
+        calendar=[_cal("AAPL", "2026-08-01"), _cal("BRK.B", "2026-08-05")],  # dotted source symbol
+        history=[],
+    )
+    run = await run_import(_importer(fake), engine=engine)
+    assert run.status == STATUS_PROMOTED
+
+    events = await _events(factory)
+    mapped = {e.ticker_id for e in events}
+    assert mapped == {ids["AAPL"], ids["BRK.B"]}  # dotted BRK.B joined via normalization
+
+
+async def test_reschedule_moves_future_row(engine):
+    factory = _factory(engine)
+    await _seed_tickers(factory, ["AAPL"])
+
+    fake1 = FakeDolt(calendar=[_cal("AAPL", "2026-08-01")], history=[], commit="c1")
+    await run_import(_importer(fake1), engine=engine)
+
+    fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-08")], history=[], commit="c2")  # moved
+    run2 = await run_import(_importer(fake2), engine=engine)
+    assert run2.status == STATUS_PROMOTED
+
+    dates = {e.announce_date for e in await _events(factory)}
+    assert dates == {date(2026, 8, 8)}  # old future date gone, new one present
+
+
+async def test_cancellation_removes_future_row(engine):
+    factory = _factory(engine)
+    await _seed_tickers(factory, ["AAPL"])
+
+    fake1 = FakeDolt(
+        calendar=[_cal("AAPL", "2026-08-01"), _cal("AAPL", "2026-08-15")], history=[], commit="c1"
+    )
+    await run_import(_importer(fake1), engine=engine)
+
+    fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-15")], history=[], commit="c2")  # 08-01 cancelled
+    await run_import(_importer(fake2), engine=engine)
+
+    dates = {e.announce_date for e in await _events(factory)}
+    assert dates == {date(2026, 8, 15)}
+
+
+async def test_past_row_never_deleted(engine):
+    factory = _factory(engine)
+    await _seed_tickers(factory, ["AAPL"])
+
+    fake1 = FakeDolt(
+        calendar=[_cal("AAPL", "2026-05-01"), _cal("AAPL", "2026-08-01")], history=[], commit="c1"
+    )
+    await run_import(_importer(fake1), engine=engine)
+
+    # Second import's calendar omits the past date but keeps a future one.
+    fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-01")], history=[], commit="c2")
+    await run_import(_importer(fake2), engine=engine)
+
+    dates = {e.announce_date for e in await _events(factory)}
+    assert date(2026, 5, 1) in dates  # past result survived
+    assert date(2026, 8, 1) in dates
+
+
+async def test_validate_fails_when_no_future(engine):
+    factory = _factory(engine)
+    await _seed_tickers(factory, ["AAPL"])
+    fake = FakeDolt(calendar=[_cal("AAPL", "2026-05-01")], history=[])  # only past
+
+    run = await run_import(_importer(fake), engine=engine)
+
+    assert run.status == STATUS_FAILED
+    assert "future" in (run.error_details or "")
+    assert len(await _events(factory)) == 0  # nothing promoted
+
+
+async def test_validate_fails_on_forward_collapse(engine):
+    factory = _factory(engine)
+    await _seed_tickers(factory, ["AAPL", "MSFT", "NVDA", "AMZN"])
+
+    fake1 = FakeDolt(
+        calendar=[
+            _cal("AAPL", "2026-08-01"),
+            _cal("MSFT", "2026-08-02"),
+            _cal("NVDA", "2026-08-03"),
+            _cal("AMZN", "2026-08-04"),
+        ],
+        history=[],
+        commit="c1",
+    )
+    await run_import(_importer(fake1), engine=engine)
+    assert len([e for e in await _events(factory) if e.announce_date > TODAY]) == 4
+
+    # Only one future row now → 1 < 50% of 4 → fail-closed, no destructive wipe.
+    fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-01")], history=[], commit="c2")
+    run2 = await run_import(_importer(fake2), engine=engine)
+
+    assert run2.status == STATUS_FAILED
+    assert "collapsed" in (run2.error_details or "")
+    assert len([e for e in await _events(factory) if e.announce_date > TODAY]) == 4  # preserved
+
+
+# --- Real-clone smoke test: exercises the actual dolt subprocess + parse + align
+# against the local clone. Skips in CI / anywhere the binary or clone is absent.
+
+_DOLT_BIN = os.environ.get("DOLT_BINARY") or shutil.which("dolt") or r"C:\Program Files\Dolt\bin\dolt.exe"
+_CLONE_DIR = Path("dolt-data/earnings")
+
+
+@pytest.mark.skipif(
+    not (Path(_DOLT_BIN).exists() and _CLONE_DIR.exists()),
+    reason="real dolt binary / earnings clone not available",
+)
+async def test_real_clone_smoke(engine):
+    from app.services import dolt_client
+
+    factory = _factory(engine)
+    await _seed_tickers(factory, ["AAPL", "MSFT", "JPM"])
+    imp = DoltEarningsImporter(
+        repo_dir=_CLONE_DIR, binary=_DOLT_BIN, today=date.today(), do_pull=False, dolt=dolt_client
+    )
+    run = await run_import(imp, engine=engine)
+
+    assert run.status == STATUS_PROMOTED
+    events = await _events(factory)
+    assert events, "no earnings parsed from the real clone"
+    assert any(e.announce_date > date.today() for e in events), "no forward calendar"
+    assert any(e.eps_actual is not None for e in events), "no calendar<->history pairing"

tests/unit/test_earnings_alignment.py

@@ -0,0 +1,104 @@
+"""Unit tests for the pure calendar<->EPS-history alignment.
+
+Anchored on the research script's exact constants (SKIP costs = 45, typical lag
+= 30, session penalty = 3, windows 90/14) so a silently changed constant fails
+here rather than quietly corrupting surprise-history pairing.
+"""
+
+from __future__ import annotations
+
+from datetime import date
+
+from app.services import earnings_alignment as ea
+
+
+def test_normalise_symbol():
+    assert ea.normalise_symbol("bf.b ") == "BF-B"
+    assert ea.normalise_symbol(" aapl") == "AAPL"
+    assert ea.normalise_symbol(None) == ""
+
+
+def test_normalise_session():
+    assert ea.normalise_session("Before market open") == "bmo"
+    assert ea.normalise_session("After market close") == "amc"
+    assert ea.normalise_session("During market hours") == "unknown"
+    assert ea.normalise_session(None) == "unknown"
+    assert ea.normalise_session("") == "unknown"
+
+
+def test_safe_number():
+    assert ea.safe_number("1.5") == 1.5
+    assert ea.safe_number("") is None
+    assert ea.safe_number("not-a-number") is None
+    assert ea.safe_number("nan") is None  # non-finite rejected
+
+
+def test_constants_pinned():
+    assert ea.SKIP_EVENT_COST == 45.0
+    assert ea.SKIP_PERIOD_COST == 45.0
+    assert ea._TYPICAL_ANNOUNCE_LAG_DAYS == 30
+    assert ea._MISSING_SESSION_PENALTY == 3.0
+
+
+def test_match_cost_uses_pinned_lag_and_penalty():
+    period = {"period_end_date": date(2026, 3, 31)}
+    # delta == 30 (typical lag) → base cost 0; known session → no penalty
+    e_known = {"announce_date": date(2026, 4, 30), "session": "amc"}
+    assert ea.match_cost(e_known, period) == 0.0
+    # unknown session adds the penalty
+    e_unknown = {"announce_date": date(2026, 4, 30), "session": "unknown"}
+    assert ea.match_cost(e_unknown, period) == 3.0
+    # delta 45 → |45-30| == 15
+    e_far = {"announce_date": date(2026, 5, 15), "session": "amc"}
+    assert ea.match_cost(e_far, period) == 15.0
+
+
+def test_dedup_calendar_prefers_known_session():
+    rows = [
+        {"symbol": "AAPL", "announce_date": date(2026, 5, 1), "session": "unknown"},
+        {"symbol": "AAPL", "announce_date": date(2026, 5, 1), "session": "amc"},
+    ]
+    grouped, stats = ea.dedup_calendar(rows)
+    assert stats["duplicate_rows"] == 1
+    assert grouped["AAPL"][0]["session"] == "amc"
+
+
+def test_dedup_history_prefers_fuller_row():
+    rows = [
+        {"symbol": "AAPL", "period_end_date": date(2026, 3, 31), "eps_actual": 2.0, "eps_estimate": None},
+        {"symbol": "AAPL", "period_end_date": date(2026, 3, 31), "eps_actual": 2.0, "eps_estimate": 1.9},
+    ]
+    grouped, stats = ea.dedup_history(rows)
+    assert stats["duplicate_rows"] == 1
+    kept = grouped["AAPL"][0]
+    assert kept["eps_actual"] == 2.0 and kept["eps_estimate"] == 1.9
+
+
+def _events(*days):
+    return [{"announce_date": d, "session": "amc"} for d in days]
+
+
+def _periods(*days):
+    return [{"period_end_date": d, "eps_actual": 1.0, "eps_estimate": 0.9} for d in days]
+
+
+def test_align_matches_monotonic_pairs():
+    # two announcements ~30d after two quarter ends
+    events = _events(date(2026, 4, 30), date(2026, 7, 30))
+    periods = _periods(date(2026, 3, 31), date(2026, 6, 30))
+    matches, um_events, um_periods = ea.align_symbol(
+        events, periods, max_lag_days=90, max_lead_days=14
+    )
+    assert matches == [(0, 0), (1, 1)]
+    assert um_events == [] and um_periods == []
+
+
+def test_align_leaves_out_of_window_unmatched():
+    # announcement 200 days after the only period end → outside the 90d window
+    events = _events(date(2026, 10, 17))
+    periods = _periods(date(2026, 3, 31))
+    matches, um_events, um_periods = ea.align_symbol(
+        events, periods, max_lag_days=90, max_lead_days=14
+    )
+    assert matches == []
+    assert um_events == [0] and um_periods == [0]