signal-platform/app/services/dolt_earnings_importer.py

"""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, ticker_service
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
# Initial-load gates (when nothing is loaded yet — the ratio gate has no baseline).
# The source publishes a forward calendar; require a real horizon, not one stray
# future row. 21 days is a conservative floor under the ~35d horizon observed on
# the live clone.
MIN_FORWARD_HORIZON_DAYS = 21
# ...and require the symbol join to reach most of the tracked universe, so a
# broken/normalization-dropped join can't seed a hollow calendar.
MIN_INITIAL_COVERAGE = 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:
        timeout = settings.dolt_command_timeout_seconds
        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, timeout=timeout)
        return await self._dolt.current_commit(
            self.repo_dir, binary=self.binary, timeout=timeout
        )

    async def stage(self, db) -> StagedEarnings:
        universe = await self._load_universe(db)  # {normalised symbol: ticker_id}

        timeout = settings.dolt_command_timeout_seconds
        cal_raw = await self._dolt.query_csv(
            self.repo_dir, _CAL_SQL, binary=self.binary, timeout=timeout
        )
        hist_raw = await self._dolt.query_csv(
            self.repo_dir, _HIST_SQL, binary=self.binary, timeout=timeout
        )
        _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:
        # Promote deletes+reinserts the forward calendar, so this gate is
        # fail-closed. The forward calendar is the project's acceptance gate.
        messages: list[str] = []
        current_future = await self._current_future_count(db)
        universe_size = int(staged.stats.get("universe_size", 0) or 0)
        coverage = (
            staged.stats.get("symbols_with_calendar", 0) / universe_size
            if universe_size
            else 0.0
        )
        horizon_days = (
            (staged.max_announce_date - self.today).days if staged.max_announce_date else 0
        )

        if staged.future_count == 0:
            messages.append("no future-dated earnings rows staged")
        elif current_future == 0:
            # Initial load: no baseline for the ratio gate, so require a real
            # forward horizon and broad universe coverage instead of one stray row.
            if horizon_days < MIN_FORWARD_HORIZON_DAYS:
                messages.append(
                    f"forward horizon only {horizon_days}d < {MIN_FORWARD_HORIZON_DAYS}d "
                    "on initial load"
                )
            if coverage < MIN_INITIAL_COVERAGE:
                messages.append(
                    f"initial universe coverage {coverage:.0%} "
                    f"< {MIN_INITIAL_COVERAGE:.0%} — symbol join likely broken"
                )
        elif 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,
            "forward_horizon_days": horizon_days,
            "universe_coverage": round(coverage, 3),
        }
        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(
                ticker_service.active_only(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