Spaces:
Sleeping
Sleeping
| """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 | |