ingestion/guidance_parser.py

"""Parse forward-looking guidance from MD&A and earnings call transcripts.

Uses Claude Haiku to extract structured low/high ranges per metric for the next
reporting period. Returns columns aligned with metrics_db._GUIDANCE_COLS.
"""
from __future__ import annotations
import json
import re
import sys
from typing import Optional

_EMPTY = {
    "guidance_period": None,
    "guidance_revenue_low": None,    "guidance_revenue_high": None,
    "guidance_eps_low": None,        "guidance_eps_high": None,
    "guidance_gross_margin_low": None,     "guidance_gross_margin_high": None,
    "guidance_operating_margin_low": None, "guidance_operating_margin_high": None,
}

MODEL = "claude-haiku-4-5-20251001"
MAX_CHARS = 8000

_GUIDANCE_HINT = re.compile(
    r"(outlook|guidance|expect|anticipate|forecast|project|fiscal\s+(?:first|second|third|fourth|q[1-4]))",
    re.IGNORECASE,
)


def _extract_relevant(text: str, max_chars: int) -> str:
    """Pull paragraphs that look like guidance to keep prompt small."""
    if not text:
        return ""
    paras = re.split(r"\n{2,}|(?<=\.)\s{2,}", text)
    keep = [p for p in paras if _GUIDANCE_HINT.search(p)]
    joined = "\n\n".join(keep) if keep else text
    return joined[:max_chars]


def _extract_json(raw: str) -> str:
    m = re.search(r"```(?:json)?\s*(\{.*\})\s*```", raw, re.DOTALL)
    if m:
        return m.group(1)
    s, e = raw.find("{"), raw.rfind("}")
    return raw[s:e + 1] if (s != -1 and e != -1 and e > s) else raw


_PROMPT = """You extract forward-looking financial guidance from SEC filings and earnings calls.

Current reporting period just disclosed: {current_period}

Input text (MD&A excerpt + transcript excerpt):
---
{text}
---

Extract guidance issued by management for the NEXT reporting period (or full year if only annual is given).
Return STRICT JSON with this exact schema:

{{
  "next_period": "Q12026" | "FY2026" | null,
  "revenue":          {{"low": <USD as number>, "high": <USD as number>}} | null,
  "eps":              {{"low": <number>, "high": <number>}} | null,
  "gross_margin":     {{"low": <decimal 0-1>, "high": <decimal 0-1>}} | null,
  "operating_margin": {{"low": <decimal 0-1>, "high": <decimal 0-1>}} | null
}}

Rules:
- USD values in dollars (e.g. $94.5B → 94500000000).
- Margins as decimals (e.g. "75% gross margin" → 0.75). NEVER as percentages.
- If management gives a point estimate, set low == high.
- If a metric is not mentioned, use null.
- If no forward guidance exists at all, return all-null fields.
- Output JSON only, no commentary."""


def parse_guidance(
    mda_text: str,
    transcript_text: Optional[str],
    current_period: str,
) -> dict:
    """Return a dict with keys matching metrics_db._GUIDANCE_COLS.

    All-None on parse failure or absent guidance. Never raises.
    """
    combined = "\n\n=== TRANSCRIPT ===\n\n".join(
        t for t in (_extract_relevant(mda_text or "", MAX_CHARS // 2),
                    _extract_relevant(transcript_text or "", MAX_CHARS // 2)) if t
    )
    if not combined.strip():
        return dict(_EMPTY)

    try:
        from langchain_anthropic import ChatAnthropic
        from langchain_core.messages import HumanMessage
        llm = ChatAnthropic(model=MODEL, temperature=0)
        resp = llm.invoke([HumanMessage(
            content=_PROMPT.format(current_period=current_period or "unknown", text=combined)
        )])
        raw = resp.content if isinstance(resp.content, str) else str(resp.content)
        data = json.loads(_extract_json(raw))
    except Exception as exc:
        print(f"[guidance_parser] WARNING: parse failed: {exc}", file=sys.stderr)
        return dict(_EMPTY)

    def _pair(obj, key):
        v = obj.get(key) if isinstance(obj, dict) else None
        if not isinstance(v, dict):
            return None, None
        lo, hi = v.get("low"), v.get("high")
        try:
            lo = float(lo) if lo is not None else None
            hi = float(hi) if hi is not None else None
        except (TypeError, ValueError):
            return None, None
        return lo, hi

    rev_lo, rev_hi = _pair(data, "revenue")
    eps_lo, eps_hi = _pair(data, "eps")
    gm_lo, gm_hi = _pair(data, "gross_margin")
    om_lo, om_hi = _pair(data, "operating_margin")
    return {
        "guidance_period": data.get("next_period") if isinstance(data, dict) else None,
        "guidance_revenue_low": rev_lo,    "guidance_revenue_high": rev_hi,
        "guidance_eps_low": eps_lo,        "guidance_eps_high": eps_hi,
        "guidance_gross_margin_low": gm_lo,     "guidance_gross_margin_high": gm_hi,
        "guidance_operating_margin_low": om_lo, "guidance_operating_margin_high": om_hi,
    }