analytics/segments.py

"""Turn stored segment rows into a display-ready breakdown.

Two problems sit between the database and a chart, and both are solved here
rather than in the renderer so they can be tested without Streamlit:

**Aggregates double-count.** A filing tags both the detail members and their
subtotals -- Apple reports iPhone, Mac, iPad and Wearables *and* a "Products"
line equal to their sum. Plotting all five inflates the total by 72%. The
subtotal is identified structurally, by being the sum of other members, so no
per-issuer list of member names is needed.

**Year-on-year needs the comparable period.** It comes free: a filing carries
its prior-year quarter in the same instance document, so the comparison is a
second read against a period end the store already holds.
"""
from __future__ import annotations

from itertools import combinations
from typing import Optional

from ingestion.segments import PRODUCT_AXIS, SEGMENT_AXIS
from storage import segments_db

# How close a member must be to a subtotal of others to count as one. Filers
# round to the million, so exact equality is too strict and anything looser
# starts swallowing real members.
_AGGREGATE_TOLERANCE = 0.005

# Enumerating subsets is exponential; a filer with more members than this is
# reporting at a granularity where subtotals are not the failure mode.
_MAX_MEMBERS_FOR_SUBSET_SCAN = 14


def _is_subtotal(value: float, others: list[float]) -> bool:
    """True when *value* equals the sum of two or more of *others*."""
    if value <= 0 or len(others) > _MAX_MEMBERS_FOR_SUBSET_SCAN:
        return False
    # Every subset is tested. An early break on an over-target sum looks like
    # a free prune but is wrong: itertools.combinations does not yield subsets
    # in ascending-sum order, so breaking skips the matching subset that comes
    # later at the same size -- which is exactly how Apple's "Products" line
    # survived its first pass here.
    for size in range(2, len(others) + 1):
        for combo in combinations(others, size):
            total = sum(combo)
            if total and abs(total - value) / value <= _AGGREGATE_TOLERANCE:
                return True
    return False


def _smallest_subset_summing_to(rows: list[dict], target: float) -> list[dict]:
    """The fewest rows whose revenue sums to *target*, or []."""
    priced = [r for r in rows if r.get("revenue")]
    if len(priced) > _MAX_MEMBERS_FOR_SUBSET_SCAN:
        return []
    for size in range(1, len(priced) + 1):
        for combo in combinations(priced, size):
            total = sum(float(r["revenue"]) for r in combo)
            if abs(total - target) / target <= _AGGREGATE_TOLERANCE:
                return list(combo)
    return []


def drop_aggregate_members(rows: list[dict], total: Optional[float] = None) -> list[dict]:
    """Remove members that restate the sum of other members.

    With *total* -- the period's consolidated revenue -- the rule is exact:
    whatever the members overshoot the company's own top line by is the
    subtotal, so the rows summing to that excess are the ones to drop. This is
    the path that runs in production.

    Without it the fallback asks whether each member equals a sum of others.
    That is enough to spot a subtotal but it can also fire on a genuine member
    whose value happens to match a combination of its peers, so it is only
    used when no consolidated figure is available to anchor against.
    """
    priced = [(r, float(r["revenue"])) for r in rows if r.get("revenue")]
    if not priced:
        return list(rows)

    if total:
        excess = sum(v for _r, v in priced) - float(total)
        if excess / float(total) <= _AGGREGATE_TOLERANCE:
            return list(rows)
        drop = {id(r) for r in _smallest_subset_summing_to(rows, excess)}
        if drop:
            return [r for r in rows if id(r) not in drop]
        # The overshoot matches no combination of members: the mix and the
        # top line disagree for a reason this code cannot name. Show every
        # member rather than guessing which one to hide.
        return list(rows)

    keep = []
    for row in rows:
        value = row.get("revenue")
        if not value:
            keep.append(row)
            continue
        others = [v for r, v in priced if r is not row]
        if not _is_subtotal(float(value), others):
            keep.append(row)
    return keep


def _consolidated_revenue(ticker: str, period_end: str) -> Optional[float]:
    """The period's top-line revenue from the metrics table, if stored.

    ``metrics.report_date`` holds a period END despite its name, which is what
    makes it join cleanly against a segment period_end.
    """
    try:
        from storage import metrics_db
        for row in metrics_db.get_all_metrics(ticker.upper()):
            if row.get("report_date") == period_end and row.get("revenue"):
                return float(row["revenue"])
    except Exception:
        return None
    return None


def breakdown(ticker: str, axis: str) -> dict:
    """The latest quarter's lines for *axis*, with year-on-year and margin.

    Returns ``{}`` when nothing is stored, so a caller can treat "no segment
    data" and "no segment block" as the same case.
    """
    period_end = segments_db.latest_period_end(ticker, axis)
    if not period_end:
        return {}

    current = drop_aggregate_members(
        segments_db.read_segments(ticker, axis, period_end),
        total=_consolidated_revenue(ticker, period_end),
    )
    if not current:
        return {}

    prior_end = segments_db.prior_year_period_end(ticker, axis, period_end)
    prior = {
        r["member"]: r
        for r in (segments_db.read_segments(ticker, axis, prior_end) if prior_end else [])
    }

    lines = []
    for row in current:
        revenue = row.get("revenue")
        was = (prior.get(row["member"]) or {}).get("revenue")
        operating_income = row.get("operating_income")
        lines.append({
            "label": row.get("label") or row.get("member"),
            "member": row.get("member"),
            "revenue": revenue,
            "prior_revenue": was,
            "yoy_pct": ((revenue - was) / was * 100) if revenue and was else None,
            "operating_income": operating_income,
            "margin_pct": (
                operating_income / revenue * 100
                if operating_income is not None and revenue else None
            ),
            "source_url": row.get("source_url") or "",
        })

    total = sum(line["revenue"] for line in lines if line["revenue"])
    for line in lines:
        line["share_pct"] = (line["revenue"] / total * 100) if total and line["revenue"] else None

    return {
        "axis": axis,
        "period_end": period_end,
        "prior_period_end": prior_end,
        "total_revenue": total or None,
        "lines": lines,
        "source_url": next((line["source_url"] for line in lines if line["source_url"]), ""),
    }


def products(ticker: str) -> dict:
    return breakdown(ticker, PRODUCT_AXIS)


def regions(ticker: str) -> dict:
    return breakdown(ticker, SEGMENT_AXIS)


def outliers(view: dict) -> dict:
    """The two lines worth a sentence: fastest and slowest year-on-year.

    Returned as ``{"fastest": line, "slowest": line}`` with either key absent
    when there is no year-on-year to rank by. This is what makes the block
    interpretable rather than merely accurate -- a reader should not have to
    scan five percentages to find the one that diverges.
    """
    ranked = [line for line in view.get("lines") or [] if line.get("yoy_pct") is not None]
    if len(ranked) < 2:
        return {}
    ranked.sort(key=lambda line: line["yoy_pct"])
    return {"slowest": ranked[0], "fastest": ranked[-1]}


def margin_spread(view: dict) -> Optional[dict]:
    """The widest gap in segment margin, when the axis carries margins.

    Apple's Greater China runs six points below every other region; that gap
    is a fact about the business the consolidated margin cannot show.
    """
    ranked = [line for line in view.get("lines") or [] if line.get("margin_pct") is not None]
    if len(ranked) < 2:
        return None
    ranked.sort(key=lambda line: line["margin_pct"])
    return {
        "lowest": ranked[0],
        "highest": ranked[-1],
        "spread_pct": ranked[-1]["margin_pct"] - ranked[0]["margin_pct"],
    }


def _money(value: float) -> str:
    for cutoff, unit in ((1e12, "T"), (1e9, "B"), (1e6, "M")):
        if abs(value) >= cutoff:
            return f"${value / cutoff:,.2f}{unit}"
    return f"${value:,.0f}"


def digest_lines(ticker: str) -> list[dict]:
    """Segment results as sentences the narrative digest can cite.

    These are the company's own tagged figures, so they enter the article's
    evidence pool on the same footing as a verified filing fact -- and they
    carry the product and region granularity nothing else in the brief has.
    Every number the sentence states is in the sentence, which is what lets
    the article quote it and still pass the number check.
    """
    lines: list[dict] = []
    for view, origin in ((products(ticker), "product"), (regions(ticker), "region")):
        period_end = view.get("period_end") or ""
        for line in view.get("lines") or []:
            revenue = line.get("revenue")
            if not revenue:
                continue
            parts = [f"{line['label']} revenue {_money(revenue)} in the quarter ended {period_end}"]
            if line.get("yoy_pct") is not None:
                parts.append(f"{line['yoy_pct']:+.1f}% year-on-year")
            if line.get("share_pct") is not None:
                parts.append(f"{line['share_pct']:.0f}% of total revenue")
            if line.get("margin_pct") is not None:
                parts.append(f"segment operating margin {line['margin_pct']:.1f}%")
            lines.append({
                "text": ", ".join(parts) + ".",
                "origin": f"{origin} mix",
                "source_url": line.get("source_url") or view.get("source_url") or None,
            })
    return lines