"""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