ingestion/analyst.py

"""ingestion/analyst.py — analyst expectations + post-earnings price reaction.

All data fetched via yfinance (free, public). Cached in storage/earnings_cache.py
with a 6h TTL to avoid hammering the API.

Returns (data, error) tuples consistent with ingestion/alphavantage.py.
"""
from __future__ import annotations

import hashlib
import json
import re
import unicodedata
from datetime import datetime, timedelta, timezone
from typing import Mapping, Optional
from urllib.parse import urlparse
from zoneinfo import ZoneInfo


_TRUSTED_EVENT_SOURCE_TYPES = {"company_ir", "sec_8-k"}
_EVENT_HASH_RE = re.compile(r"^[0-9a-f]{64}$")
_MARKET_TIMEZONE = "America/New_York"


def _canonical_event_text(value: str) -> str:
    value = unicodedata.normalize("NFKC", str(value or ""))
    value = value.replace("\r\n", "\n").replace("\r", "\n")
    return "\n".join(line.rstrip() for line in value.split("\n")).strip()


def _event_content_hash(source_excerpt: str) -> str:
    return hashlib.sha256(_canonical_event_text(source_excerpt).encode("utf-8")).hexdigest()


def _event_provenance_id(identity: Mapping[str, str]) -> str:
    payload = json.dumps(dict(identity), ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return "evt_" + hashlib.sha256(payload.encode("utf-8")).hexdigest()[:24]


def _derived_market_timing(published_at: str) -> tuple[Optional[str], Optional[str], Optional[str]]:
    """Return (event_date, timing, error) derived from an offset-aware timestamp."""
    try:
        parsed = datetime.fromisoformat(str(published_at).replace("Z", "+00:00"))
    except (TypeError, ValueError):
        return None, None, "invalid_published_at"
    if parsed.tzinfo is None or parsed.utcoffset() is None:
        return None, None, "published_at_requires_timezone"
    try:
        local = parsed.astimezone(ZoneInfo(_MARKET_TIMEZONE))
    except Exception:
        return None, None, "market_timezone_unavailable"
    minutes = local.hour * 60 + local.minute
    if minutes < 9 * 60 + 30:
        timing = "before_open"
    elif minutes >= 16 * 60:
        timing = "after_close"
    else:
        return local.date().isoformat(), None, "publication_during_market_hours"
    return local.date().isoformat(), timing, None


def make_earnings_event_provenance(
    *,
    ticker: str,
    published_at: str,
    source_type: str,
    source_url: str,
    source_excerpt: str,
) -> dict[str, str]:
    """Create a content-addressed provenance record for an earnings release.

    This helper guarantees internal integrity only. ``fetch_price_reaction`` still
    validates the source type, URL, timestamp-derived date/timing, ticker, hash,
    and ID before enabling any return comparison.
    """
    if not str(ticker).strip():
        raise ValueError("ticker is required")
    if source_type not in _TRUSTED_EVENT_SOURCE_TYPES:
        raise ValueError(f"unsupported event source_type: {source_type!r}")
    parsed_url = urlparse(source_url)
    if parsed_url.scheme != "https" or not parsed_url.hostname:
        raise ValueError("source_url must be an absolute HTTPS URL")
    if (
        source_type == "sec_8-k"
        and not (parsed_url.hostname == "sec.gov" or parsed_url.hostname.endswith(".sec.gov"))
    ):
        raise ValueError("sec_8-k provenance must use a sec.gov URL")
    event_date, event_timing, timing_error = _derived_market_timing(published_at)
    if timing_error or not event_date or not event_timing:
        raise ValueError(f"Cannot derive an earnings-event window: {timing_error}")
    excerpt = _canonical_event_text(source_excerpt)
    if not excerpt:
        raise ValueError("source_excerpt is required")
    digest = _event_content_hash(excerpt)
    identity = {
        "ticker": ticker.upper(),
        "event_date": event_date,
        "event_timing": event_timing,
        "source_type": source_type,
        "source_url": source_url,
        "published_at": published_at,
        "content_hash": digest,
    }
    return {
        **identity,
        "source_excerpt": excerpt,
        "evidence_id": _event_provenance_id(identity),
    }


def _validate_earnings_event_provenance(
    provenance: Optional[Mapping[str, object]],
    *,
    ticker: str,
    event_date: str,
    event_timing: str,
) -> tuple[Optional[dict[str, str]], str]:
    """Validate event provenance without trusting caller-supplied status flags."""
    if not isinstance(provenance, Mapping):
        return None, "missing_event_provenance"

    required = {
        "ticker", "event_date", "event_timing", "source_type", "source_url",
        "published_at", "source_excerpt", "content_hash", "evidence_id",
    }
    if any(not provenance.get(field) for field in required):
        return None, "incomplete_event_provenance"

    normalized = {field: str(provenance[field]) for field in required}
    if normalized["ticker"].upper() != ticker.upper():
        return None, "event_ticker_mismatch"
    if normalized["event_date"] != event_date:
        return None, "event_date_mismatch"
    if normalized["event_timing"] != event_timing:
        return None, "event_timing_mismatch"
    if normalized["source_type"] not in _TRUSTED_EVENT_SOURCE_TYPES:
        return None, "untrusted_event_source_type"

    parsed_url = urlparse(normalized["source_url"])
    if parsed_url.scheme != "https" or not parsed_url.hostname:
        return None, "event_source_url_must_be_https"
    if (
        normalized["source_type"] == "sec_8-k"
        and not (parsed_url.hostname == "sec.gov" or parsed_url.hostname.endswith(".sec.gov"))
    ):
        return None, "sec_event_source_must_use_sec_gov"

    derived_date, derived_timing, timing_error = _derived_market_timing(normalized["published_at"])
    if timing_error:
        return None, timing_error
    if derived_date != normalized["event_date"]:
        return None, "published_at_event_date_mismatch"
    if derived_timing != normalized["event_timing"]:
        return None, "published_at_event_timing_mismatch"

    digest = _event_content_hash(normalized["source_excerpt"])
    if not _EVENT_HASH_RE.fullmatch(normalized["content_hash"]):
        return None, "invalid_event_content_hash"
    if digest != normalized["content_hash"]:
        return None, "event_content_hash_mismatch"

    identity = {
        "ticker": normalized["ticker"].upper(),
        "event_date": normalized["event_date"],
        "event_timing": normalized["event_timing"],
        "source_type": normalized["source_type"],
        "source_url": normalized["source_url"],
        "published_at": normalized["published_at"],
        "content_hash": normalized["content_hash"],
    }
    if normalized["evidence_id"] != _event_provenance_id(identity):
        return None, "event_evidence_id_mismatch"
    return normalized, "verified"


def _safe_float(v) -> Optional[float]:
    try:
        f = float(v)
        return None if f != f else f
    except (TypeError, ValueError):
        return None


def _pick_period_row(df, prefer: list[str]):
    """Return the first row from df whose index matches one of the preferred period codes."""
    if df is None or getattr(df, "empty", True):
        return None
    for code in prefer:
        if code in df.index:
            return df.loc[code]
    return None


def _pick_period_row_with_code(df, prefer: list[str]):
    if df is None or getattr(df, "empty", True):
        return None, None
    for code in prefer:
        if code in df.index:
            return df.loc[code], code
    return None, None


def fetch_analyst_estimates(
    ticker: str,
    target_period: Optional[str] = None,
) -> tuple[Optional[dict], Optional[str]]:
    """Fetch analyst consensus EPS, revenue estimates, and 30-day revision %.

    Returns dict with keys:
      consensus_eps_est, consensus_rev_est, estimate_revision_30d_pct
    Each may be None if yfinance does not expose it for this ticker.
    """
    from storage.earnings_cache import get as cache_get, set as cache_set
    cache_key = f"ANALYST:v2:{ticker.upper()}:{target_period or 'unspecified'}"
    cached = cache_get(cache_key, ttl_hours=6)
    if cached is not None:
        return cached, None

    try:
        import yfinance as yf
    except ImportError as e:
        return None, f"yfinance not installed: {e}"

    try:
        t = yf.Ticker(ticker.upper())

        eps_row, eps_code = _pick_period_row_with_code(
            getattr(t, "earnings_estimate", None), ["0q", "+1q", "0y"]
        )
        rev_row, rev_code = _pick_period_row_with_code(
            getattr(t, "revenue_estimate", None), ["0q", "+1q", "0y"]
        )
        trend_row, trend_code = _pick_period_row_with_code(
            getattr(t, "eps_trend", None), ["0q", "+1q", "0y"]
        )

        consensus_eps = _safe_float(eps_row.get("avg")) if eps_row is not None else None
        consensus_rev = _safe_float(rev_row.get("avg")) if rev_row is not None else None

        revision_30d_pct: Optional[float] = None
        if trend_row is not None:
            current = _safe_float(trend_row.get("current"))
            ago30 = _safe_float(trend_row.get("30daysAgo"))
            if current is not None and ago30 is not None and ago30 != 0:
                revision_30d_pct = round((current - ago30) / abs(ago30) * 100, 2)

        result = {
            "consensus_eps_est": consensus_eps,
            "consensus_rev_est": consensus_rev,  # USD raw
            "estimate_revision_30d_pct": revision_30d_pct,
            "provider_period_codes": {
                "eps": eps_code, "revenue": rev_code, "trend": trend_code,
            },
            "target_period": target_period,
            "as_of": datetime.now(timezone.utc).isoformat(),
            # yfinance exposes relative forward periods, not the historical
            # pre-release snapshot required for an actual-vs-consensus claim.
            "period_aligned": False,
            "comparison_allowed": False,
            "alignment_status": "UNVERIFIED_CURRENT_CONSENSUS",
        }
        cache_set(cache_key, result)
        return result, None
    except Exception as e:
        return None, f"yfinance analyst fetch failed: {e}"


def fetch_price_reaction(
    ticker: str,
    filing_date: str,
    *,
    event_kind: str = "filing_date_proxy",
    event_timing: str = "unknown",
    event_provenance: Optional[Mapping[str, object]] = None,
) -> tuple[Optional[dict], Optional[str]]:
    """Compute correctly-windowed returns for a verified earnings event.

    Strings supplied via ``event_kind``/``event_timing`` are never sufficient to
    mark an event aligned. A content-addressed ``event_provenance`` record is
    required. Without one, returns are null and the result is explicitly
    fail-closed.

    Returns {'d1_pct': float|None, 'd5_pct': float|None, 'since_release_pct': float}.
    """
    from storage.earnings_cache import get as cache_get, set as cache_set

    try:
        d = datetime.strptime(filing_date[:10], "%Y-%m-%d")
    except (TypeError, ValueError, IndexError):
        return None, f"invalid filing_date format: {filing_date!r}"

    provenance = None
    provenance_reason = "event_kind_or_timing_not_earnings_release"
    if event_kind == "earnings_release" and event_timing in {"before_open", "after_close"}:
        provenance, provenance_reason = _validate_earnings_event_provenance(
            event_provenance,
            ticker=ticker,
            event_date=filing_date[:10],
            event_timing=event_timing,
        )

    if provenance is None:
        return {
            "d1_pct": None,
            "d5_pct": None,
            "since_release_pct": None,
            "event_date": filing_date[:10],
            "event_kind": event_kind,
            "event_timing": event_timing,
            "event_aligned": False,
            "comparison_allowed": False,
            "alignment_status": "UNVERIFIED_EVENT_PROVENANCE",
            "alignment_reason": provenance_reason,
            "provenance_verified": False,
            "event_provenance_id": None,
            "event_source_type": None,
            "event_source_url": None,
            "baseline_trading_date": None,
            "d1_trading_date": None,
            "d5_trading_date": None,
        }, None

    cache_key = (
        f"PRICEREACT:v3:{ticker.upper()}:{filing_date}:"
        f"{event_timing}:{provenance['evidence_id']}"
    )
    cached = cache_get(cache_key, ttl_hours=24)
    if cached is not None:
        return cached, None

    try:
        import yfinance as yf
        import pandas as pd
    except ImportError as e:
        return None, f"yfinance/pandas not installed: {e}"

    # Ten calendar days guarantee room for a prior trading-session baseline.
    start = (d - timedelta(days=10)).strftime("%Y-%m-%d")
    end = (datetime.today() + timedelta(days=1)).strftime("%Y-%m-%d")

    try:
        hist = yf.download(ticker.upper(), start=start, end=end, progress=False, auto_adjust=True)
        if hist.empty:
            return None, "no price history returned"

        if isinstance(hist.columns, pd.MultiIndex):
            hist.columns = hist.columns.get_level_values(0)

        closes = hist["Close"].dropna()
        if closes.empty:
            return None, "no close prices in range"

        event_positions = [i for i, timestamp in enumerate(closes.index) if timestamp.date() == d.date()]
        if not event_positions:
            result = {
                "d1_pct": None,
                "d5_pct": None,
                "since_release_pct": None,
                "event_date": filing_date[:10],
                "event_kind": event_kind,
                "event_timing": event_timing,
                "event_aligned": False,
                "comparison_allowed": False,
                "alignment_status": "VERIFIED_EVENT_NON_TRADING_DATE",
                "alignment_reason": "event_date_not_a_trading_session",
                "provenance_verified": True,
                "event_provenance_id": provenance["evidence_id"],
                "event_source_type": provenance["source_type"],
                "event_source_url": provenance["source_url"],
                "baseline_trading_date": None,
                "d1_trading_date": None,
                "d5_trading_date": None,
            }
            cache_set(cache_key, result)
            return result, None

        event_idx = event_positions[0]
        if event_timing == "before_open":
            baseline_idx = event_idx - 1
            d1_idx = event_idx
            d5_idx = event_idx + 4
        else:  # after_close
            baseline_idx = event_idx
            d1_idx = event_idx + 1
            d5_idx = event_idx + 5

        if baseline_idx < 0:
            return None, "insufficient price history for event baseline"

        baseline = float(closes.iloc[baseline_idx])
        d1 = (
            (float(closes.iloc[d1_idx]) / baseline - 1) * 100
            if d1_idx < len(closes) else None
        )
        d5 = (
            (float(closes.iloc[d5_idx]) / baseline - 1) * 100
            if d5_idx < len(closes) else None
        )
        since = (float(closes.iloc[-1]) / baseline - 1) * 100

        result = {
            "d1_pct": round(d1, 2) if d1 is not None else None,
            "d5_pct": round(d5, 2) if d5 is not None else None,
            "since_release_pct": round(since, 2),
            "event_date": filing_date[:10],
            "event_kind": event_kind,
            "event_timing": event_timing,
            "event_aligned": True,
            "comparison_allowed": d1 is not None,
            "alignment_status": "VERIFIED_EARNINGS_EVENT",
            "alignment_reason": "verified_provenance_and_trading_window",
            "provenance_verified": True,
            "event_provenance_id": provenance["evidence_id"],
            "event_source_type": provenance["source_type"],
            "event_source_url": provenance["source_url"],
            "baseline_trading_date": closes.index[baseline_idx].date().isoformat(),
            "d1_trading_date": (
                closes.index[d1_idx].date().isoformat() if d1_idx < len(closes) else None
            ),
            "d5_trading_date": (
                closes.index[d5_idx].date().isoformat() if d5_idx < len(closes) else None
            ),
        }
        cache_set(cache_key, result)
        return result, None
    except Exception as e:
        return None, f"price reaction fetch failed: {e}"


def post_earnings_returns_batch(
    ticker: str,
    dates: list[str],
    *,
    event_provenance_by_date: Optional[Mapping[str, Mapping[str, object]]] = None,
) -> dict[str, dict]:
    """Compute d1/d5 returns only for provenance-verified earnings dates.

    Used by dashboard/financials.py to render the surprise history.
    Returns {date_str: {'d1': float|None, 'd5': float|None}}. Returns are
    expressed as decimals (NOT percentages — the caller multiplies by 100).
    Missing/invalid provenance produces no row; filing dates are never treated
    as earnings dates implicitly.
    """
    if not dates or not isinstance(event_provenance_by_date, Mapping):
        return {}

    verified: dict[str, dict[str, str]] = {}
    for date_str in dates:
        candidate = event_provenance_by_date.get(date_str)
        timing = str(candidate.get("event_timing", "")) if isinstance(candidate, Mapping) else ""
        if timing not in {"before_open", "after_close"}:
            continue
        provenance, _ = _validate_earnings_event_provenance(
            candidate,
            ticker=ticker,
            event_date=date_str[:10],
            event_timing=timing,
        )
        if provenance is not None:
            verified[date_str] = provenance
    if not verified:
        return {}

    try:
        import yfinance as yf
        import pandas as pd
    except ImportError:
        return {}

    try:
        parsed = [datetime.strptime(date_str[:10], "%Y-%m-%d") for date_str in verified]
        earliest = min(parsed)
        latest = max(parsed)
        start = (earliest - timedelta(days=10)).strftime("%Y-%m-%d")
        end = (latest + timedelta(days=14)).strftime("%Y-%m-%d")

        hist = yf.download(ticker.upper(), start=start, end=end, progress=False, auto_adjust=True)
        if hist.empty:
            return {}
        if isinstance(hist.columns, pd.MultiIndex):
            hist.columns = hist.columns.get_level_values(0)
        closes = hist["Close"].dropna()
        if closes.empty:
            return {}

        result: dict[str, dict] = {}
        for date_str, provenance in verified.items():
            try:
                d = datetime.strptime(date_str[:10], "%Y-%m-%d")
                event_positions = [
                    i for i, timestamp in enumerate(closes.index)
                    if timestamp.date() == d.date()
                ]
                if not event_positions:
                    continue
                event_idx = event_positions[0]
                if provenance["event_timing"] == "before_open":
                    baseline_idx, d1_idx, d5_idx = event_idx - 1, event_idx, event_idx + 4
                else:
                    baseline_idx, d1_idx, d5_idx = event_idx, event_idx + 1, event_idx + 5
                if baseline_idx < 0:
                    continue
                baseline = float(closes.iloc[baseline_idx])
                d1 = float(closes.iloc[d1_idx]) / baseline - 1 if d1_idx < len(closes) else None
                d5 = float(closes.iloc[d5_idx]) / baseline - 1 if d5_idx < len(closes) else None
                result[date_str] = {
                    "d1": d1,
                    "d5": d5,
                    "event_aligned": True,
                    "event_timing": provenance["event_timing"],
                    "event_provenance_id": provenance["evidence_id"],
                }
            except Exception:
                continue
        return result
    except Exception:
        return {}