Spaces:
Sleeping
Sleeping
| """How much a driver moved a number — computed, never estimated. | |
| The question "what did the supply constraint cost in margin?" has three | |
| possible answers and only two of them are allowed here. | |
| **Computed.** Arithmetic on figures the company disclosed. A product line's | |
| contribution to group growth is exactly ``(revenue_now - revenue_prior) / | |
| group_revenue_prior``; the contributions sum to the group growth rate, which | |
| is the checksum. A margin change splits exactly into a mix effect and a rate | |
| effect. Both are reproducible from the filing and are computed here. | |
| **Disclosed.** The company's own quantification — a guided margin range | |
| against a reported one. The model may transcribe those figures under the | |
| usual citation rules, and ``bridge()`` does the subtraction. It never | |
| supplies a figure of its own. | |
| **Estimated.** "Supply constraints probably cost 150bp." Not produced here, | |
| not accepted from a model, and not renderable. A number with no source is | |
| indistinguishable from a number that is wrong, and this product's whole | |
| claim is that it does not print those. | |
| """ | |
| from __future__ import annotations | |
| from typing import Optional | |
| from ingestion.segments import PRODUCT_AXIS, SEGMENT_AXIS | |
| def growth_contributions(ticker: str, axis: str = PRODUCT_AXIS) -> dict: | |
| """Each member's contribution, in percentage points, to group growth. | |
| A line growing 29% matters less than one growing 12% if the first is a | |
| tenth of the second's size. Contribution is the figure that settles it, | |
| and the contributions add up to the group growth rate -- returned as | |
| ``total_pp`` so the arithmetic can be checked on screen. | |
| """ | |
| from analytics.segments import breakdown | |
| view = breakdown(ticker, axis) | |
| lines = [ | |
| line for line in (view.get("lines") or []) | |
| if line.get("revenue") and line.get("prior_revenue") | |
| ] | |
| if not lines: | |
| return {} | |
| prior_total = sum(line["prior_revenue"] for line in lines) | |
| if not prior_total: | |
| return {} | |
| contributions = [ | |
| { | |
| "label": line["label"], | |
| "delta": line["revenue"] - line["prior_revenue"], | |
| "contribution_pp": (line["revenue"] - line["prior_revenue"]) / prior_total * 100, | |
| "yoy_pct": line.get("yoy_pct"), | |
| "revenue": line["revenue"], | |
| } | |
| for line in lines | |
| ] | |
| contributions.sort(key=lambda c: c["contribution_pp"], reverse=True) | |
| return { | |
| "axis": axis, | |
| "period_end": view.get("period_end"), | |
| "prior_period_end": view.get("prior_period_end"), | |
| "lines": contributions, | |
| "total_pp": sum(c["contribution_pp"] for c in contributions), | |
| "source_url": view.get("source_url") or "", | |
| } | |
| def margin_decomposition(ticker: str) -> dict: | |
| """Split the segment margin change into mix and rate, exactly. | |
| Two ways a group margin can move: the business earns more per dollar in | |
| the places it already sells (rate), or the sales shift towards places | |
| that already earned more (mix). Standard two-factor split -- | |
| mix = sum over members of (weight_now - weight_prior) * margin_prior | |
| rate = sum over members of weight_now * (margin_now - margin_prior) | |
| -- and the two add to the total change, which is returned so the identity | |
| can be shown rather than asserted. Only the reportable-segment axis | |
| carries operating income, so this is a regional read. | |
| """ | |
| from analytics.segments import breakdown | |
| from storage import segments_db | |
| view = breakdown(ticker, SEGMENT_AXIS) | |
| period_end = view.get("period_end") | |
| prior_end = view.get("prior_period_end") | |
| if not period_end or not prior_end: | |
| return {} | |
| prior_rows = { | |
| row["member"]: row | |
| for row in segments_db.read_segments(ticker, SEGMENT_AXIS, prior_end) | |
| } | |
| pairs = [] | |
| for line in view.get("lines") or []: | |
| was = prior_rows.get(line.get("member")) | |
| if not was or not was.get("revenue") or not line.get("revenue"): | |
| continue | |
| if was.get("operating_income") is None or line.get("operating_income") is None: | |
| continue | |
| pairs.append((line, was)) | |
| if len(pairs) < 2: | |
| return {} | |
| total_now = sum(line["revenue"] for line, _was in pairs) | |
| total_prior = sum(was["revenue"] for _line, was in pairs) | |
| oi_now = sum(line["operating_income"] for line, _was in pairs) | |
| oi_prior = sum(was["operating_income"] for _line, was in pairs) | |
| if not total_now or not total_prior: | |
| return {} | |
| margin_now = oi_now / total_now * 100 | |
| margin_prior = oi_prior / total_prior * 100 | |
| members = [] | |
| mix_total = rate_total = 0.0 | |
| for line, was in pairs: | |
| weight_now = line["revenue"] / total_now | |
| weight_prior = was["revenue"] / total_prior | |
| rate_prior = was["operating_income"] / was["revenue"] * 100 | |
| rate_now = line["operating_income"] / line["revenue"] * 100 | |
| mix = (weight_now - weight_prior) * rate_prior | |
| rate = weight_now * (rate_now - rate_prior) | |
| mix_total += mix | |
| rate_total += rate | |
| members.append({ | |
| "label": line["label"], | |
| "margin_now_pct": rate_now, | |
| "margin_prior_pct": rate_prior, | |
| "mix_pp": mix, | |
| "rate_pp": rate, | |
| "total_pp": mix + rate, | |
| }) | |
| members.sort(key=lambda m: m["total_pp"], reverse=True) | |
| return { | |
| "period_end": period_end, | |
| "prior_period_end": prior_end, | |
| "margin_now_pct": margin_now, | |
| "margin_prior_pct": margin_prior, | |
| "change_pp": margin_now - margin_prior, | |
| "mix_pp": mix_total, | |
| "rate_pp": rate_total, | |
| "members": members, | |
| "source_url": view.get("source_url") or "", | |
| } | |
| def bridge(impact: dict) -> Optional[dict]: | |
| """Turn one disclosed driver->metric claim into a computed delta. | |
| *impact* carries only figures the company stated, transcribed by the | |
| model under the citation rules: where the metric is now, and where the | |
| company says it is going. The subtraction happens here so that no model | |
| ever reports a delta it worked out itself. | |
| Returns None when the claim is not arithmetically usable, which keeps a | |
| half-filled impact off the page rather than rendering "from 50.1% to | |
| None". | |
| """ | |
| if not isinstance(impact, dict): | |
| return None | |
| try: | |
| current = float(impact["from_value"]) | |
| low = float(impact["to_low"]) | |
| high = float(impact.get("to_high", impact["to_low"])) | |
| except (KeyError, TypeError, ValueError): | |
| return None | |
| if low > high: | |
| low, high = high, low | |
| # Basis points read better than "percentage points of a percentage" for | |
| # a margin; anything else keeps the metric's own unit. | |
| unit = str(impact.get("unit") or "%") | |
| delta_low, delta_high = low - current, high - current | |
| return { | |
| "driver": str(impact.get("driver") or "").strip(), | |
| "metric": str(impact.get("metric") or "").strip(), | |
| "basis": str(impact.get("basis") or "disclosed"), | |
| "unit": unit, | |
| "from_value": current, | |
| "to_low": low, | |
| "to_high": high, | |
| "delta_low_pp": delta_low, | |
| "delta_high_pp": delta_high, | |
| "delta_low_bp": delta_low * 100, | |
| "delta_high_bp": delta_high * 100, | |
| "is_range": abs(high - low) > 1e-9, | |
| "fact_ids": [str(i) for i in (impact.get("fact_ids") or [])], | |
| } | |