signal-platform/app/services/earnings_alignment.py

"""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