analytics/deltas.py

"""analytics/deltas.py — deterministic delta calculations.

Pure Python, no LLM calls, no Streamlit imports.
Computes MetricDelta, EpsSurprise, GuidanceChange, and QuarterSnapshot
from metrics_db rows and alphavantage earnings data.
"""

from dataclasses import dataclass
from typing import Optional

# ---------------------------------------------------------------------------
# Dataclasses
# ---------------------------------------------------------------------------

@dataclass
class MetricDelta:
    label: str               # display label e.g. "Revenue", "EPS", "Op. Margin"
    current: Optional[float]
    prior: Optional[float]
    delta_pct: Optional[float]   # percentage change (or pp change for margins)
    direction: str               # "up", "down", "flat"
    favorable: bool              # True = green, False = red, used by UI
    significant: bool            # |delta_pct| >= threshold
    period_basis: str            # "YoY" or "QoQ"
    unit: str                    # "$B", "$", "%", "pp", "M", etc.


@dataclass
class EpsSurprise:
    latest_beat_pct: float       # surprisePercentage for most recent quarter
    beat_streak: int             # consecutive quarters where surprisePercentage > 0
    avg_4q_surprise: float       # average surprisePercentage over last 4 quarters


@dataclass
class GuidanceChange:
    disclosed_change: str        # "newly_disclosed" | "withdrawn" | "maintained" | "absent"
    latest_verdict: Optional[str]  # from brief["guidance_history"][0]["verdict"] if present
    prior_verdict: Optional[str] = None          # guidance_history[1]["verdict"] — used when latest is "pending"
    prior_actual_result: Optional[str] = None    # guidance_history[1]["actual_result"]


@dataclass
class QuarterSnapshot:
    ticker: str
    period: str              # e.g. "Q12025"
    filing_date: str         # e.g. "2025-01-28"
    metric_deltas: list[MetricDelta]  # 6-8 selected deltas
    eps_surprise: Optional[EpsSurprise]
    guidance_change: Optional[GuidanceChange]
    new_risks_count: int


# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------

def _parse_period(period: str) -> Optional[tuple[int, int]]:
    """Parse "Q12025" -> (1, 2025). Returns None for "FY2024" or unparseable."""
    if not period or not period.startswith("Q"):
        return None
    try:
        # period is "Q<num><4-digit-year>", e.g. "Q12025", "Q42024"
        body = period[1:]  # "12025"
        year = int(body[-4:])
        quarter = int(body[:-4])
        if quarter < 1 or quarter > 4:
            return None
        return (quarter, year)
    except (ValueError, IndexError):
        return None


def _prior_yoy_period(quarter: int, year: int) -> str:
    """Return the period string for the same quarter one year prior."""
    return f"Q{quarter}{year - 1}"


def _previous_quarter_period(period: str) -> Optional[str]:
    parsed = _parse_period(period)
    if parsed is None:
        return None
    quarter, year = parsed
    return f"Q{quarter - 1}{year}" if quarter > 1 else f"Q4{year - 1}"


def _period_sort_key(period: str) -> tuple[int, int]:
    parsed = _parse_period(period)
    if parsed is None:
        return (-1, -1)
    quarter, year = parsed
    return (year, quarter)


_ADDITIVE_Q4_METRICS = (
    "revenue", "free_cash_flow", "capex", "buybacks", "dividends_paid",
)


def _derive_virtual_q4_row(annual: dict, quarters: list[dict]) -> Optional[dict]:
    period = annual.get("period", "")
    if not period.startswith("FY") or not period[2:].isdigit():
        return None
    year = int(period[2:])
    by_period = {q.get("period"): q for q in quarters}
    required = [by_period.get(f"Q{q}{year}") for q in (1, 2, 3)]
    if any(row is None for row in required):
        return None

    q4 = dict(annual)
    q4.update({
        "period": f"Q4{year}",
        "form_type": "DERIVED-Q4",
        "period_basis": "quarter",
        "data_quality_status": "DERIVED",
        "_derived": True,
        "eps": None,
        "shares_diluted": None,
        "effective_tax_rate": None,
        "interest_expense": None,
    })
    for key in _ADDITIVE_Q4_METRICS:
        annual_value = annual.get(key)
        quarter_values = [row.get(key) for row in required]
        q4[key] = (
            annual_value - sum(quarter_values)
            if annual_value is not None and all(v is not None for v in quarter_values)
            else None
        )

    q4_revenue = q4.get("revenue")
    for margin_key in ("gross_margin", "operating_margin"):
        annual_margin = annual.get(margin_key)
        annual_revenue = annual.get("revenue")
        quarter_profits = []
        for row in required:
            rev, margin = row.get("revenue"), row.get(margin_key)
            if rev is None or margin is None:
                quarter_profits = []
                break
            quarter_profits.append(rev * margin)
        if (
            annual_margin is not None and annual_revenue is not None
            and q4_revenue not in (None, 0) and len(quarter_profits) == 3
        ):
            q4[margin_key] = (annual_revenue * annual_margin - sum(quarter_profits)) / q4_revenue
        else:
            q4[margin_key] = None
    q4["revenue_yoy_pct"] = None
    q4["quality_warnings"] = ["virtual_q4:derived_from_fy_minus_q1_q2_q3"]
    return q4


def _quarter_rows_with_virtual_q4(rows: list[dict]) -> list[dict]:
    quarterly = [
        dict(r) for r in rows
        if r.get("form_type") in {"10-Q", "DERIVED-Q4"}
    ]
    existing = {r.get("period") for r in quarterly}
    for annual in (r for r in rows if r.get("form_type") == "10-K"):
        fy = annual.get("period", "")
        q4_period = f"Q4{fy[2:]}" if fy.startswith("FY") else ""
        if not q4_period or q4_period in existing:
            continue
        derived = _derive_virtual_q4_row(annual, quarterly)
        if derived:
            quarterly.append(derived)
            existing.add(q4_period)
    quarterly.sort(key=lambda r: _period_sort_key(r.get("period", "")), reverse=True)
    return quarterly


def _compute_delta(
    label: str,
    current: Optional[float],
    prior: Optional[float],
    is_margin: bool,
    favorable_direction: str,  # "up", "down", or "neutral"
    threshold: float,
    period_basis: str,
    unit: str,
) -> Optional["MetricDelta"]:
    """
    Compute a MetricDelta from raw current/prior values.

    For margins: delta = (current - prior) * 100 (percentage points).
    For other metrics: delta_pct = (current - prior) / abs(prior) * 100.

    Returns None if either value is None, or prior is 0 for non-margin metrics.
    """
    if current is None or prior is None:
        return None
    if not is_margin and prior == 0:
        return None

    delta = (current - prior) * 100 if is_margin else (current - prior) / abs(prior) * 100

    if delta > 0:
        direction = "up"
    elif delta < 0:
        direction = "down"
    else:
        direction = "flat"

    if favorable_direction == "neutral":
        favorable = True
    elif direction == "flat":
        favorable = False
    else:
        favorable = (direction == favorable_direction)

    return MetricDelta(
        label=label,
        current=current,
        prior=prior,
        delta_pct=delta,
        direction=direction,
        favorable=favorable,
        significant=abs(delta) >= threshold,
        period_basis=period_basis,
        unit=unit,
    )


# Metric configuration: (db_key, label, unit, is_margin, favorable_direction, threshold, scale_fn)
# scale_fn is applied to raw value before display (e.g. divide by 1e9 for $B)
_METRIC_CONFIG = [
    # (db_key,         label,            unit,  is_margin, fav_dir,    threshold, divisor)
    ("revenue",         "Revenue",        "$B",  False,     "up",       5.0,       1e9),
    ("eps",             "EPS",            "$",   False,     "up",       5.0,       1.0),
    ("gross_margin",    "Gross Margin",   "pp",  True,      "up",       1.0,       1.0),
    ("operating_margin","Op. Margin",     "pp",  True,      "up",       1.0,       1.0),
    ("free_cash_flow",  "Free Cash Flow", "$B",  False,     "up",       10.0,      1e9),
    ("capex",           "CapEx",          "$B",  False,     "neutral",  10.0,      1e9),
    ("buybacks",        "Buybacks",       "$B",  False,     "up",       20.0,      1e9),
    ("dividends_paid",  "Dividends",      "$B",  False,     "up",       10.0,      1e9),
    ("total_debt",      "Total Debt",     "$B",  False,     "down",     5.0,       1e9),
    ("shares_diluted",  "Shares Out.",    "M",   False,     "down",     1.0,       1e6),
]

# Priority order for selection (lower index = higher priority)
_PRIORITY = [
    "revenue", "eps", "operating_margin", "gross_margin", "free_cash_flow",
    "buybacks", "total_debt", "shares_diluted", "capex", "dividends_paid",
]


# ---------------------------------------------------------------------------
# Public functions
# ---------------------------------------------------------------------------

def compute_metric_deltas(ticker: str) -> list:
    """
    Compute MetricDelta for each tracked metric using metrics_db data.

    Uses YoY comparison as primary, QoQ as fallback.
    Returns up to 8 MetricDelta objects, prioritised by significance then metric priority.
    """
    from storage.metrics_db import get_all_metrics

    try:
        all_rows = get_all_metrics(ticker)
    except Exception:
        return []

    if not all_rows:
        return []

    # SEC 10-Q rows do not include Q4.  Sort by fiscal period rather than filing
    # date so amendments cannot masquerade as the latest operating quarter.
    quarterly = _quarter_rows_with_virtual_q4(all_rows)

    if not quarterly:
        return []

    latest = quarterly[0]
    latest_period = latest.get("period", "")
    parsed = _parse_period(latest_period)

    # Build a lookup by period for quick YoY peer access
    # Iterate newest-first; only add a period if not already present so the
    # newest filing wins when the same period appears more than once.
    period_lookup: dict[str, dict] = {}
    for r in quarterly:
        p = r.get("period", "")
        if p and p not in period_lookup:
            period_lookup[p] = r

    results_yoy: dict[str, MetricDelta] = {}
    results_qoq: dict[str, MetricDelta] = {}

    for db_key, label, unit, is_margin, fav_dir, threshold, divisor in _METRIC_CONFIG:
        current_raw = latest.get(db_key)
        if current_raw is None:
            continue

        current_val = current_raw / divisor if divisor != 1.0 else current_raw

        # --- YoY attempt ---
        if parsed is not None:
            quarter_num, year = parsed
            yoy_period = _prior_yoy_period(quarter_num, year)
            yoy_row = period_lookup.get(yoy_period)
            if yoy_row is not None:
                prior_raw = yoy_row.get(db_key)
                if prior_raw is not None:
                    prior_val = prior_raw / divisor if divisor != 1.0 else prior_raw
                    delta = _compute_delta(
                        label, current_val, prior_val,
                        is_margin, fav_dir, threshold, "YoY", unit,
                    )
                    if delta is not None:
                        results_yoy[db_key] = delta

        # --- QoQ fallback ---
        if db_key not in results_yoy:
            prior_period = _previous_quarter_period(latest_period)
            prior_row = period_lookup.get(prior_period) if prior_period else None
        else:
            prior_row = None
        if prior_row is not None:
            prior_raw = prior_row.get(db_key)
            if prior_raw is not None:
                prior_val = prior_raw / divisor if divisor != 1.0 else prior_raw
                delta = _compute_delta(
                    label, current_val, prior_val,
                    is_margin, fav_dir, threshold, "QoQ", unit,
                )
                if delta is not None:
                    results_qoq[db_key] = delta

    # Merge: YoY takes precedence over QoQ
    all_deltas: dict[str, MetricDelta] = {**results_qoq, **results_yoy}

    # Sort: significant first, then by metric priority
    priority_map = {key: idx for idx, key in enumerate(_PRIORITY)}

    def sort_key(item: tuple[str, MetricDelta]) -> tuple[int, int]:
        db_key, delta = item
        sig_rank = 0 if delta.significant else 1
        prio_rank = priority_map.get(db_key, len(_PRIORITY))
        return (sig_rank, prio_rank)

    sorted_deltas = sorted(all_deltas.items(), key=sort_key)
    return [delta for _, delta in sorted_deltas[:8]]


def compute_eps_surprise(ticker: str) -> Optional[EpsSurprise]:
    """
    Compute EpsSurprise from Alpha Vantage quarterly earnings data.

    Returns None if data is unavailable or parsing fails.
    """
    try:
        from ingestion.alphavantage import fetch_earnings
        data, _err = fetch_earnings(ticker)
        if data is None:
            return None

        quarterly = data.get("quarterlyEarnings")
        if not quarterly:
            return None

        # Parse surprisePercentage values, skipping unparseable entries
        parsed_surprises: list[float] = []
        for item in quarterly:
            raw = item.get("surprisePercentage")
            if raw is None:
                continue
            try:
                parsed_surprises.append(float(raw))
            except (ValueError, TypeError):
                continue

        if not parsed_surprises:
            return None

        latest_beat_pct = parsed_surprises[0]

        # Beat streak: count from the front while surprisePercentage > 0
        beat_streak = 0
        for val in parsed_surprises:
            if val > 0:
                beat_streak += 1
            else:
                break

        # Average of first 4 valid items
        avg_4q_surprise = sum(parsed_surprises[:4]) / min(len(parsed_surprises), 4)

        return EpsSurprise(
            latest_beat_pct=latest_beat_pct,
            beat_streak=beat_streak,
            avg_4q_surprise=avg_4q_surprise,
        )
    except Exception:
        return None


def compute_guidance_change(ticker: str, brief: dict) -> GuidanceChange:
    """
    Determine whether guidance was newly disclosed, withdrawn, maintained, or absent
    by comparing the two most recent 10-Q rows.
    """
    if not isinstance(brief, dict):
        brief = {}

    from storage.metrics_db import get_all_metrics

    try:
        all_rows = get_all_metrics(ticker)
    except Exception:
        all_rows = []

    quarterly = _quarter_rows_with_virtual_q4(all_rows)
    latest_row = quarterly[0] if quarterly else None
    previous_period = _previous_quarter_period(latest_row.get("period", "")) if latest_row else None
    prior_row = next(
        (r for r in quarterly if r.get("period") == previous_period), None
    ) if previous_period else None

    if latest_row is None:
        disclosed_change = "absent"
    elif prior_row is None:
        if latest_row.get("guidance_disclosed"):
            disclosed_change = "newly_disclosed"
        else:
            disclosed_change = "absent"
    elif latest_row.get("guidance_disclosed") and not prior_row.get("guidance_disclosed"):
        disclosed_change = "newly_disclosed"
    elif not latest_row.get("guidance_disclosed") and prior_row.get("guidance_disclosed"):
        disclosed_change = "withdrawn"
    elif latest_row.get("guidance_disclosed") and prior_row.get("guidance_disclosed"):
        disclosed_change = "maintained"
    else:
        disclosed_change = "absent"

    latest_verdict: Optional[str] = None
    prior_verdict: Optional[str] = None
    prior_actual_result: Optional[str] = None
    guidance_history = brief.get("guidance_history")
    if guidance_history and isinstance(guidance_history, list):
        if guidance_history:
            latest_verdict = guidance_history[0].get("verdict")
        if len(guidance_history) >= 2:
            prior_verdict = guidance_history[1].get("verdict")
            prior_actual_result = guidance_history[1].get("actual_result")

    return GuidanceChange(
        disclosed_change=disclosed_change,
        latest_verdict=latest_verdict,
        prior_verdict=prior_verdict,
        prior_actual_result=prior_actual_result,
    )


def compute_risk_diff(brief: dict) -> int:
    """Return the count of risks flagged as new in the current filing."""
    if not isinstance(brief, dict):
        return 0
    return sum(1 for r in brief.get("risks_categorized", []) if r.get("is_new_this_filing"))


def build_quarter_snapshot(ticker: str, brief: dict) -> Optional[QuarterSnapshot]:
    """
    Facade: assemble a QuarterSnapshot from all delta sub-computations.

    Returns None on any exception so callers never crash.
    """
    try:
        from storage.metrics_db import get_all_metrics

        rows = get_all_metrics(ticker)
        if not rows:
            return None

        quarterly = _quarter_rows_with_virtual_q4(rows)
        if not quarterly:
            return None
        latest = quarterly[0]
        metric_deltas = compute_metric_deltas(ticker)
        eps_surprise = compute_eps_surprise(ticker)
        guidance_change = compute_guidance_change(ticker, brief)
        new_risks_count = compute_risk_diff(brief)

        return QuarterSnapshot(
            ticker=ticker,
            period=latest.get("period", ""),
            filing_date=latest.get("filing_date", ""),
            metric_deltas=metric_deltas,
            eps_surprise=eps_surprise,
            guidance_change=guidance_change,
            new_risks_count=new_risks_count,
        )
    except Exception:
        return None