Blame for engine/segments.py

"""CanonicalSegmentState: reportable segments reconciled to the company before anything uses them.

A filing's XBRL instance can carry one segment under more than one member: the operating-segment
member in the segment note (with operating income) and another member of the same segment in the
revenue-disaggregation note (revenue by product line, no operating income). Naming members by a
display string and treating each as a business counted the same revenue twice. Here:

1. Every segment is identified by its XBRL member QName (``segment_id``), with its axis and period.
2. Members that are the same segment are merged as aliases: the same normalized name, or equal
   revenue for the same period where one member's name contains the other's. The member that carries
   operating income is kept as the operating segment; the others are recorded as aliases.
3. Business-line revenue must sum to consolidated revenue for the same period (from the same XBRL
   instance), within ``TOLERANCE``. Shares are taken of consolidated revenue, not of the row total.
4. Only then are industries assigned, as explicit allocations whose fractions sum to 1 per segment:
   by segment name; else by filed product-line revenue when the product lines reconcile to the
   segment; else the company's own industry.

Any failed invariant leaves the state NOT_RECONCILED, and the business-mix beta, the segment-build
requirement and the segment tables are disabled (the company's single-industry figures apply).
"""
from __future__ import annotations

import re

TOLERANCE = 0.02                 # Σ segment revenue vs consolidated revenue, same period
ALIAS_REVENUE_TOLERANCE = 0.001  # two members with the same revenue to 0.1% describe the same segment

RECONCILED, NOT_RECONCILED, UNAVAILABLE, SINGLE = "RECONCILED", "NOT_RECONCILED", "UNAVAILABLE", "SINGLE_SEGMENT"
_IGNORE = re.compile(r"\b(corporate|eliminations?|unallocated|reconcil)", re.I)


def _key(name):
    s = re.sub(r"(segments?|members?)$", "", re.sub(r"[^a-z0-9]", "", str(name or "").lower()))
    return s


def _local(member):
    return str(member or "").split(":")[-1]


def _same_revenue(a, b):
    return a and b and abs(a / b - 1.0) <= ALIAS_REVENUE_TOLERANCE


def canonicalize(mix, companyfacts=None, *, tolerance=TOLERANCE):
    """CanonicalSegmentState (a dict) from ``canonical.parse_segment_mix`` output."""
    mix = mix or {}
    rows = [dict(g) for g in mix.get("segments") or []]
    base = {"period_start": mix.get("period_start"), "period_end": mix.get("period_end"), "source": mix.get("source"),
            "form": mix.get("form"), "filed": mix.get("filed"), "url": mix.get("url"),
            "axis": mix.get("axis") or "us-gaap:StatementBusinessSegmentsAxis",
            "consolidated_revenue": mix.get("consolidated_revenue"),
            "consolidated_source": mix.get("consolidated_tag"), "segments": [], "geographic": [], "aliases": [],
            "excluded": [], "invariants": []}
    if not rows:
        return dict(base, status=UNAVAILABLE, reason=(mix.get("error") or "no segment facts in the latest periodic report"),
                    business_mix_allowed=False)

    for g in rows:
        g["segment_id"] = g.get("member")
    # every member on the segment axis is an operating segment, including one named by a region (it is the
    # company's own business there); corporate / elimination rows are kept out of the list but not out of
    # the revenue reconciliation
    geo = [g.get("segment_id") for g in rows if g.get("kind") == "geographic"]
    excl = [g for g in rows if _IGNORE.search(str(g.get("name") or ""))]
    bl = [g for g in rows if g not in excl]

    # ── aliases: one segment under several members ──
    kept, aliases, conflicts = [], [], []
    for g in sorted(bl, key=lambda x: (x.get("operating_income") is None, -(x.get("revenue") or 0))):
        k = _key(g.get("name"))
        twin = None
        for h in kept:
            hk = _key(h.get("name"))
            if k == hk:
                twin = h
                break
            if _same_revenue(g.get("revenue"), h.get("revenue")) and k and hk and (k in hk or hk in k):
                twin = h
                break
        if twin is None:
            kept.append(dict(g, aliases=[], product_lines=list(g.get("product_lines") or [])))
            continue
        if not _same_revenue(g.get("revenue"), twin.get("revenue")):
            conflicts.append(f"{g['segment_id']} and {twin['segment_id']} share the name '{g.get('name')}' but report "
                             f"different revenue ({g.get('revenue'):,.0f} vs {twin.get('revenue'):,.0f})")
            kept.append(dict(g, aliases=[], product_lines=list(g.get("product_lines") or [])))
            continue
        twin["aliases"].append(g["segment_id"])
        have = {p.get("name") for p in twin["product_lines"]}
        twin["product_lines"] += [p for p in g.get("product_lines") or [] if p.get("name") not in have]
        for f in ("operating_income", "margin", "capex", "capex_to_revenue", "revenue_prior", "growth"):
            if twin.get(f) is None and g.get(f) is not None:
                twin[f] = g[f]
        aliases.append({"kept": twin["segment_id"], "merged": g["segment_id"],
                        "reason": "same segment reported under a second member (revenue equal for the same period)"})

    inv = []
    ids = [g["segment_id"] for g in kept]
    inv.append({"name": "unique segment ids", "ok": len(ids) == len(set(ids)) and all(ids), "detail": ", ".join(map(str, ids))})
    names = [str(g.get("name") or "") for g in kept]
    dup = sorted({n for n in names if names.count(n) > 1})
    inv.append({"name": "unique display names", "ok": not dup and not conflicts,
                "detail": "; ".join(conflicts) or (f"duplicate names: {', '.join(dup)}" if dup else "unique")})
    tot = sum(g.get("revenue") or 0 for g in kept if (g.get("revenue") or 0) > 0)
    other = sum(g.get("revenue") or 0 for g in excl)
    cons = mix.get("consolidated_revenue")
    cov = ((tot + other) / cons) if (cons and cons > 0) else None
    inv.append({"name": "Σ segment revenue = consolidated revenue (same period)",
                "ok": cov is not None and abs(cov - 1.0) <= tolerance,
                "detail": (f"Σ segments {tot:,.0f}" + (f" + corporate/eliminations {other:,.0f}" if other else "")
                           + f" vs consolidated {cons:,.0f} ({cov:.1%})" if cov is not None
                           else "consolidated revenue for the segment period not found in the filing")})
    # completeness against the company-facts segment list when it covers the same fiscal period
    cf = companyfacts or {}
    if cf.get("segments") and str(cf.get("fy_end") or "")[:10] == str(mix.get("period_end") or "")[:10]:
        known = set(ids) | {a for g in kept for a in g["aliases"]}
        known_local = {_key(_local(x)) for x in known}
        miss = [g for g in cf["segments"] if g.get("kind") != "geographic" and not _IGNORE.search(str(g.get("name") or ""))
                and g.get("member") not in known and _key(_local(g.get("member"))) not in known_local
                and cons and (g.get("revenue") or 0) > 0.05 * cons]
        inv.append({"name": "segments complete vs company facts", "ok": not miss,
                    "detail": ("missing: " + ", ".join(str(g.get("member")) for g in miss)) if miss else "complete"})

    for g in kept:
        g["revenue_share"] = ((g.get("revenue") or 0) / cons) if (cons and cons > 0) else None
        g["role"] = "operating segment" if g.get("operating_income") is not None else "segment (revenue only)"
    ok = all(i["ok"] for i in inv)
    status = (RECONCILED if ok else NOT_RECONCILED) if len(kept) >= 2 else SINGLE
    kept.sort(key=lambda g: -(g.get("revenue") or 0))
    return dict(base, segments=kept, geographic=geo, excluded=[g.get("segment_id") for g in excl], excluded_revenue=other,
                aliases=aliases,
                invariants=inv, sum_revenue=tot, coverage=cov, status=status,
                business_mix_allowed=(status == RECONCILED),
                reason=("reconciled" if ok else "; ".join(i["name"] + ": " + i["detail"] for i in inv if not i["ok"])))


def allocate(canon, company_industry, industry_of, mapper=None):
    """Add ``allocations`` [{industry, fraction, basis, mapping}] to each canonical segment (fractions sum to 1).

    ``mapper(segment_id, text)`` → SegmentIndustryMapping (damodaran_data.map_segment): the keyword map plus the
    parent-sector compatibility gate and its fallback. Each segment keeps its mapping as ``industry_mapping`` and each
    allocation carries the mapping its industry came through, so nothing downstream reads an industry without its
    compatibility status. ``industry_of(text, (), fallback)`` → (industry, how) is the legacy keyword map, used only
    when no mapper is given."""
    if not canon or canon.get("status") not in (RECONCILED, SINGLE):
        return canon
    if mapper is None:
        def mapper(segment_id, text):                     # legacy callers: the keyword map without the gate
            ind, how = industry_of(text, (), None)
            ok = bool(ind) or how.startswith("regional")
            return {"segment_id": segment_id, "input_text": str(text or ""), "selected_industry": ind or company_industry,
                    "mapping_method": how, "matched_pattern": "legacy" if ind else None, "matched_text": None,
                    "parent_industry": company_industry, "compatibility_status": "PASS", "rejected": [],
                    "fallback_used": None if ind else "COMPANY_INDUSTRY", "eligible_for_value_weight": ok or bool(company_industry),
                    "eligible_for_beta": ok or bool(company_industry)}
    for g in canon.get("segments") or []:
        mp = mapper(g.get("segment_id"), g.get("name"))
        g["industry_mapping"] = mp
        how = str(mp.get("mapping_method") or "")
        if how.startswith("regional"):
            g["allocations"] = [{"industry": company_industry, "fraction": 1.0, "mapping": mp,
                                 "basis": "company industry (a segment named by a region is the company's own business there)"}]
            continue
        if mp.get("matched_pattern"):
            g["allocations"] = [{"industry": mp["selected_industry"], "fraction": 1.0, "basis": f"segment name ({how})",
                                 "mapping": mp}]
            continue
        rejected = "; ".join(f"{r.get('industry')} rejected: {r.get('reason')}" for r in mp.get("rejected") or [])
        pl = [p for p in g.get("product_lines") or [] if (p.get("revenue") or 0) > 0]
        psum = sum(p["revenue"] for p in pl)
        if pl and g.get("revenue") and abs(psum / g["revenue"] - 1.0) <= TOLERANCE:
            acc, via = {}, {}
            for p in pl:
                pm = mapper(g.get("segment_id"), p.get("name"))
                pi = pm["selected_industry"] if pm.get("matched_pattern") else company_industry
                acc[pi] = acc.get(pi, 0.0) + p["revenue"] / psum
                via.setdefault(pi, pm if pm.get("matched_pattern") else mp)
            g["allocations"] = [{"industry": k, "fraction": v, "basis": "filed product-line revenue", "mapping": via[k]}
                                for k, v in sorted(acc.items(), key=lambda kv: -kv[1])]
        else:
            g["allocations"] = [{"industry": mp.get("selected_industry"), "fraction": 1.0, "mapping": mp,
                                 "basis": ("company industry (" + (f"the segment name's keyword match was not compatible: {rejected}"
                                                                   if rejected else "segment name has no industry match")
                                           + ("; product-line revenue does not reconcile to the segment)" if pl else ")"))}]
    bad = [g["segment_id"] for g in canon.get("segments") or []
           if abs(sum(a["fraction"] for a in g.get("allocations") or []) - 1.0) > 1e-9]
    canon["invariants"] = list(canon.get("invariants") or []) + [
        {"name": "industry allocation fractions sum to 1", "ok": not bad, "detail": ", ".join(bad) or "all segments"}]
    if bad:
        canon["status"], canon["business_mix_allowed"] = NOT_RECONCILED, False
        canon["reason"] = "industry allocation fractions do not sum to 1: " + ", ".join(bad)
    return canon


def for_perimeter(canon, excluded_segment_ids, action_id=None):
    """The segment set of a valuation perimeter: the canonical segments less those that belong to a separated
    business (by segment id, resolved once from the corporate action). Shares are re-based to the remaining set."""
    c = dict(canon or {})
    ex = set(excluded_segment_ids or ())
    keep = [dict(g) for g in c.get("segments") or [] if g.get("segment_id") not in ex]
    tot = sum(g.get("revenue") or 0 for g in keep if (g.get("revenue") or 0) > 0)
    for g in keep:
        g["revenue_share"] = ((g.get("revenue") or 0) / tot) if tot else None
    c.update(segments=keep, perimeter="continuing company", perimeter_action_id=action_id,
             excluded_for_perimeter=sorted(ex), reason=(c.get("reason") or "") + (f"; {len(ex)} segment(s) of the separated "
                                                                                    f"business excluded ({action_id})" if ex else ""))
    return c


def check(canon, tol=TOLERANCE):
    """[(name, ok, detail)] — the invariants recomputed from the state alone (used by validation)."""
    out = []
    if not canon or canon.get("status") in (UNAVAILABLE, None):
        return out
    segs = canon.get("segments") or []
    ids = [g.get("segment_id") for g in segs]
    out.append(("unique segment ids", len(ids) == len(set(ids)) and all(ids), ", ".join(map(str, ids))))
    if canon.get("status") == RECONCILED:
        cons = canon.get("consolidated_revenue")
        tot = sum(g.get("revenue") or 0 for g in segs) + (canon.get("excluded_revenue") or 0)
        out.append(("Σ segment revenue = consolidated", bool(cons) and abs(tot / cons - 1) <= tol,
                    f"{tot:,.0f} vs {cons or 0:,.0f}"))
        sh = sum(g.get("revenue_share") or 0 for g in segs) + ((canon.get("excluded_revenue") or 0) / cons if cons else 0)
        out.append(("segment shares sum to 1", abs(sh - 1) <= tol, f"{sh:.4f}"))
        for g in segs:
            fr = sum(a.get("fraction") or 0 for a in g.get("allocations") or [])
            if g.get("allocations") is not None and abs(fr - 1) > 1e-6:
                out.append((f"allocations of {g.get('segment_id')} sum to 1", False, f"{fr:.6f}"))
    return out