| """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 |
| ALIAS_REVENUE_TOLERANCE = 0.001 |
|
|
| 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") |
| |
| |
| |
| 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] |
|
|
| |
| 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")}) |
| |
| 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): |
| 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 |
|
|
