"""RelationshipClaim: a typed relation between two typed quantities. A relation is not a signed number with a label ("-0.0896, price relative to composite"). It names its left-hand side and right-hand side, the relation that holds between them, the magnitude and the exact derivation of that magnitude. It is created only after ``state.compatible(lhs, rhs, comparison_type)`` passes; an incompatible pair (a margin CHANGE against a margin LEVEL) gets no relation at all. The persona receives a relation as ONE token whose rendering states the complete fact, subject first ("the price ($403.30) is 9.0% below the market composite ($443.00)"). The author may explain it, never re-attach its direction to another subject. A magnitude is bound to the pair it measures: the break-even revenue CAGR is a GROWTH_FROM relation between the starting revenue and the break-even revenue over the story's years, never a gap between two terminal revenue levels. """ from __future__ import annotations from collections.abc import Mapping import re from dataclasses import asdict, dataclass, field ABOVE, BELOW, GREATER_THAN, LESS_THAN, EQUAL = "ABOVE", "BELOW", "GREATER_THAN", "LESS_THAN", "EQUAL" GROWTH_FROM = "GROWTH_FROM" # the two break-even solves are ONE-VARIABLE alternatives (revenue with the margin held; margin with revenue held), so the # relation is a frontier: a revenue/margin COMBINATION more or less demanding than the story's, never "both at once" MORE_DEMANDING, LESS_DEMANDING, MIXED, EQUIVALENT = ("MORE_DEMANDING_COMBINATION", "LESS_DEMANDING_COMBINATION", "MIXED_COMBINATION", "EQUIVALENT_COMBINATION") # the logical form of the proposition a relation states; paraphrase must preserve it SIMPLE, CONJUNCTIVE, ALTERNATIVE_CONDITIONALS, TRADEOFF = "SIMPLE", "CONJUNCTIVE", "ALTERNATIVE_CONDITIONALS", "TRADEOFF" LOGICAL_STRUCTURES = (SIMPLE, CONJUNCTIVE, ALTERNATIVE_CONDITIONALS, TRADEOFF) RELATION_TYPES = (ABOVE, BELOW, GREATER_THAN, LESS_THAN, EQUAL, GROWTH_FROM, MORE_DEMANDING, LESS_DEMANDING, MIXED, EQUIVALENT) _OPERATOR = {ABOVE: ">", GREATER_THAN: ">", BELOW: "<", LESS_THAN: "<", EQUAL: "=", GROWTH_FROM: "growth"} # magnitude definitions: each is a function of (lhs, rhs[, years]) and nothing else RATIO_GAP = "ABS(lhs / rhs - 1)" LEVEL_DIFF = "ABS(lhs - rhs)" CAGR = "(lhs / rhs) ^ (1 / years) - 1" EQUAL_TOLERANCE = {RATIO_GAP: 0.0005, LEVEL_DIFF: 0.0005} @dataclass(frozen=True) class RelationshipClaim: claim_id: str lhs_claim_id: str rhs_claim_id: str lhs_metric_id: str rhs_metric_id: str relation_type: str operator: str magnitude: float | None magnitude_unit: str # PERCENT | PERCENTAGE_POINTS magnitude_definition: str comparison_type: str lhs_value: float | None rhs_value: float | None allowed: bool incompatibility_reason: str | None source: str derivation: str render_mode: str = "full_sentence_clause" rendered: str | None = None signed: float | None = None # lhs/rhs − 1 (or lhs − rhs): the one number the presentation claim shows presentation_claim_id: str | None = None # the report-only claim that shows the magnitude (PRICE_VS_COMPOSITE …) years: float | None = None years_claim_id: str | None = None condition: str | None = None # what the relation holds fixed (e.g. the break-even basis margin) def to_dict(self): return asdict(self) @dataclass(frozen=True) class Spec: claim_id: str lhs: str # claim ids (catalog) rhs: str comparison_type: str definition: str above: str # relation type when lhs > rhs below: str presentation: str | None = None years: str | None = None lhs_words: str = "" rhs_words: str = "" source: str = "" rhs_noun: str = "" # CLAUSE mode: "9.0% below the $443.00 market composite" rhs_short: str = "" # COMPACT mode: "9.0% below market pricing" lhs_subjects: tuple = () # words that name the lhs (a CLAUSE/COMPACT token must follow its subject) rhs_subjects: tuple = () # words that name the rhs (never the subject of a CLAUSE/COMPACT token) def _specs(state): iv_word = state.get("valuation.intrinsic_label") or "intrinsic value" cc = ((state.get("valuation.engine") or {}).get("terms") or {}).get("cross_check") or "cross-check value" PRICE = ("price", "shares", "stock") return ( Spec("REL_PRICE_INTRINSIC", "PRICE", "INTRINSIC_VALUE", "valuation", RATIO_GAP, ABOVE, BELOW, "PRICE_VS_INTRINSIC", lhs_words="the price", rhs_words=f"the {iv_word}", source="market price; selected valuation engine", rhs_noun=iv_word, rhs_short=f"the {iv_word}", lhs_subjects=PRICE, rhs_subjects=(iv_word, "intrinsic", "value")), Spec("REL_PRICE_MECHANICAL", "PRICE", "MECHANICAL_VALUE", "valuation", RATIO_GAP, ABOVE, BELOW, "PRICE_VS_MECHANICAL", lhs_words="the price", rhs_words=f"the {cc} value", source="market price; mechanical cross-check", rhs_noun=f"{cc} value", rhs_short=f"the {cc} value", lhs_subjects=PRICE, rhs_subjects=("dcf", "cross-check", "mechanical")), Spec("REL_PRICE_MARKET_COMPOSITE", "PRICE", "MARKET_COMPOSITE", "valuation", RATIO_GAP, ABOVE, BELOW, "PRICE_VS_COMPOSITE", lhs_words="the price", rhs_words="the market composite", source="market price; market-pricing composite", # the short form names the composite ("the price is 9.0% below market pricing" read as self-contradictory: the # share price IS the market's price) rhs_noun="market composite", rhs_short="the market composite", lhs_subjects=PRICE, rhs_subjects=("composite", "market pricing")), Spec("REL_TARGET_PRICE", "ANALYST_TARGET", "PRICE", "valuation", RATIO_GAP, ABOVE, BELOW, "TARGET_VS_PRICE", lhs_words="the mean analyst target", rhs_words="the price", source="sell-side consensus target; market price", rhs_noun="share price", rhs_short="the price", lhs_subjects=("target",), rhs_subjects=("share price", "the price is")), Spec("REL_FORECAST_EPS_CONSENSUS", "FORECAST_EPS", "CONSENSUS_EPS_NTM", "level", RATIO_GAP, ABOVE, BELOW, "FORECAST_EPS_VS_CONSENSUS", lhs_words="the engine's EPS forecast for the next twelve months", rhs_words="consensus for the same twelve months", source="reconciled forecast; time-weighted consensus", rhs_noun="consensus for the same twelve months", rhs_short="consensus", lhs_subjects=("forecast", "engine"), rhs_subjects=("consensus", "street")), # engine revenue vs consensus revenue for the SAME twelve months (the time-weighted FY1/FY2 consensus): the only # object that may carry "aligned", "above" or "below consensus" for revenue — never a consensus growth RATE or a # calendar-year consensus beside the next-twelve-month forecast Spec("REL_ENGINE_REVENUE_CONSENSUS", "FORECAST_REVENUE", "CONSENSUS_REVENUE_NTM", "level", RATIO_GAP, ABOVE, BELOW, lhs_words="the engine's revenue forecast for the next twelve months", rhs_words="consensus revenue for the same twelve months", source="reconciled forecast; time-weighted consensus", rhs_noun="consensus revenue for the same twelve months", rhs_short="consensus", lhs_subjects=("forecast", "engine"), rhs_subjects=("consensus", "street")), Spec("REL_BREAK_EVEN_REVENUE_STORY", "BE_REVENUE", "STORY_REVENUE_MATURITY", "scenario", RATIO_GAP, GREATER_THAN, LESS_THAN, lhs_words="the revenue today's price needs at maturity", rhs_words="the story's revenue at maturity", source="break-even solve (story DCF); adopted story", rhs_noun="story revenue at maturity", rhs_short="the story's maturity revenue", lhs_subjects=("required revenue", "revenue the price needs", "break-even revenue", "revenue needed"), rhs_subjects=("story's revenue", "story revenue")), Spec("REL_BREAK_EVEN_MARGIN_STORY", "BE_MARGIN", "STORY_MARGIN", "scenario", LEVEL_DIFF, GREATER_THAN, LESS_THAN, lhs_words="the operating margin today's price needs at the story's revenue", rhs_words="the story's maturity operating margin", source="break-even solve (story DCF); adopted story", rhs_noun="story maturity margin", rhs_short="the story's margin", lhs_subjects=("required margin", "margin the price needs", "break-even margin", "margin needed"), rhs_subjects=("story's margin", "story margin")), Spec("REL_BREAK_EVEN_REVENUE_GROWTH", "BE_REVENUE", "STORY_START_REVENUE", "path", CAGR, GROWTH_FROM, GROWTH_FROM, "BE_REVENUE_CAGR", years="STORY_YEARS", lhs_words="the revenue today's price needs at maturity", rhs_words="the story's starting revenue", source="break-even solve (story DCF); StoryBase"), # the company's operating margin against its industry aggregate (the comparison authors kept making untyped): # the continuing-company starting margin when a separation changed the perimeter, else the trailing margin Spec("REL_OPERATING_MARGIN_INDUSTRY", "STORY_START_MARGIN" if ((state.historical.get("perimeter") or {}).get("event")) else "OPERATING_MARGIN_TTM", "DMD_MARGIN", "benchmark", LEVEL_DIFF, ABOVE, BELOW, lhs_words=("the continuing company's operating margin" if ((state.historical.get("perimeter") or {}).get("event")) else "the trailing operating margin"), rhs_words="Damodaran's industry operating margin", source="company margin; industry aggregate", rhs_noun="industry operating margin", rhs_short="the industry margin", lhs_subjects=("operating margin", "margin"), rhs_subjects=("industry",)), ) def _metric(state, cid): """(typed Metric, ClaimDef) for a claim: guidance rows resolve to their typed range view, never to text.""" import catalog as C d = C.claim_def(state, cid) if d is None: return None, None return C._claim_metric(state, cid), d def _pct(v): return f"{abs(v) * 100:.1f}%" def _render(spec, rel_type, mag, lv, rv, lhs_d, rhs_d, years, condition=None): import catalog as C L, R = C.fmt(lhs_d.stype, lv), C.fmt(rhs_d.stype, rv) if spec.definition == CAGR: yrs = f"{int(round(years))}" if years is not None else "the story's" # the sign is part of the claim: a break-even revenue below the starting revenue allows revenue to shrink (PGY: a # -6.1% rate rendered as "requires 6.1% annual revenue growth") if abs(mag) < 0.0005: return f"{spec.lhs_words} ({L}) requires no revenue growth from {spec.rhs_words} ({R}) over {yrs} years" if mag < 0: return (f"{spec.lhs_words} ({L}) allows revenue to shrink {_pct(mag)} a year from {spec.rhs_words} ({R}) " f"over {yrs} years") return (f"{spec.lhs_words} ({L}) requires {_pct(mag)} annual revenue growth from {spec.rhs_words} ({R}) " f"over {yrs} years") if rel_type == EQUAL: return f"{spec.lhs_words} ({L}) is in line with {spec.rhs_words} ({R})" word = {ABOVE: "above", BELOW: "below", GREATER_THAN: "above", LESS_THAN: "below"}[rel_type] size = _pct(mag) if spec.definition == RATIO_GAP else f"{abs(mag) * 100:.1f} percentage points" out = f"{spec.lhs_words} ({L}) is {size} {word} {spec.rhs_words} ({R})" return out + (f", {condition}" if condition else "") def make(state, spec, condition=None): """(RelationshipClaim | None, rejection | None). compatible() decides before any relation exists.""" from state import compatible lm, ld = _metric(state, spec.lhs) rm, rd = _metric(state, spec.rhs) if lm is None or rm is None or lm.value is None or rm.value is None: return None, {"claim_id": spec.claim_id, "reason": "an endpoint is missing or untyped"} ok, why = compatible(lm, rm, spec.comparison_type) if not ok: return None, {"claim_id": spec.claim_id, "lhs": spec.lhs, "rhs": spec.rhs, "comparison_type": spec.comparison_type, "reason": "; ".join(why), "dimensions": list(dict.fromkeys(dimension(w) for w in why))} lv, rv = float(lm.value), float(rm.value) years = None if spec.definition == CAGR: ym, _yd = _metric(state, spec.years) years = float(ym.value) if (ym is not None and ym.value) else None if not (years and lv > 0 and rv > 0): return None, {"claim_id": spec.claim_id, "reason": "growth needs positive levels and a horizon"} mag = (lv / rv) ** (1.0 / years) - 1.0 signed, rel_type = mag, GROWTH_FROM else: signed = (lv / rv - 1.0) if spec.definition == RATIO_GAP else (lv - rv) if spec.definition == RATIO_GAP and not rv: return None, {"claim_id": spec.claim_id, "reason": "zero right-hand side"} mag = abs(signed) rel_type = EQUAL if mag < EQUAL_TOLERANCE[spec.definition] else (spec.above if signed > 0 else spec.below) rendered = _render(spec, rel_type, mag, lv, rv, ld, rd, years, condition) renderings = {"FULL": rendered, "FULL_SENTENCE": rendered[0].upper() + rendered[1:] + "."} if spec.definition != CAGR and spec.rhs_noun: import catalog as C R = C.fmt(rd.stype, rv) size = _pct(mag) if spec.definition == RATIO_GAP else f"{abs(mag) * 100:.1f} percentage points" word = {ABOVE: "above", BELOW: "below", GREATER_THAN: "above", LESS_THAN: "below"}.get(rel_type) if word: renderings["CLAUSE"] = f"{size} {word} the {R} {spec.rhs_noun}" renderings["COMPACT"] = f"{size} {word} {spec.rhs_short}" else: renderings["CLAUSE"] = f"in line with the {R} {spec.rhs_noun}" renderings["COMPACT"] = f"in line with {spec.rhs_short}" # typed grammatical slots: PREDICATE carries its own verb; NOUN_PHRASE is an object ("trades at …") renderings["PREDICATE"] = f"is {renderings['COMPACT']}" if spec.definition == RATIO_GAP and spec.comparison_type in ("valuation", "level"): np_word = {ABOVE: "premium", BELOW: "discount"}.get(rel_type) renderings["NOUN_PHRASE"] = (f"{C._article(size)} {size} {np_word} to {spec.rhs_short}" if np_word else f"parity with {spec.rhs_short}") rel = RelationshipClaim( claim_id=spec.claim_id, lhs_claim_id=spec.lhs, rhs_claim_id=spec.rhs, lhs_metric_id=lm.metric_id, rhs_metric_id=rm.metric_id, relation_type=rel_type, operator=_OPERATOR[rel_type], magnitude=mag, magnitude_unit="PERCENTAGE_POINTS" if spec.definition == LEVEL_DIFF else "PERCENT", magnitude_definition=spec.definition, comparison_type=spec.comparison_type, lhs_value=lv, rhs_value=rv, allowed=True, incompatibility_reason=None, source=spec.source, derivation=f"{spec.definition} with lhs = {spec.lhs} ({lm.metric_id}), rhs = {spec.rhs} ({rm.metric_id})" + (f", years = {spec.years}" if spec.years else ""), rendered=rendered, signed=signed, presentation_claim_id=spec.presentation, years=years, years_claim_id=spec.years, condition=condition) return rel, renderings def attempt(state, lhs, rhs, comparison_type, definition=RATIO_GAP, claim_id="REL_ATTEMPT"): """Try to relate two arbitrary claims (used by tests and diagnostics): the same compatibility gate applies.""" return make(state, Spec(claim_id, lhs, rhs, comparison_type, definition, GREATER_THAN, LESS_THAN, lhs_words=lhs, rhs_words=rhs)) def _expectations(state, items): """ImpliedExpectationsRelationship: the price-implied revenue/margin frontier vs the adopted story, from the two ONE-VARIABLE break-even solves. Each case states what it holds constant; neither implies the other is also required. The polarity is computed, never written by an author.""" rv, mv = items.get("REL_BREAK_EVEN_REVENUE_STORY"), items.get("REL_BREAK_EVEN_MARGIN_STORY") if not (rv or mv): return None parts = [x["relation_type"] for x in (rv, mv) if x] if all(p == GREATER_THAN for p in parts): overall = MORE_DEMANDING elif all(p == LESS_THAN for p in parts): overall = LESS_DEMANDING elif all(p == EQUAL for p in parts): overall = EQUIVALENT else: overall = MIXED be = (state.get("valuation.story") or {}).get("breakeven") or {} bm = be.get("story_margin") sm = state.value("valuation.story_maturity_margin") held_m = float(bm) if bm is not None else (float(sm) if sm is not None else None) import catalog as C money = lambda v: C.fmt(C.CURRENCY_BIG, v) pct = lambda v: C.fmt(C.PCT, v) revenue_case = ({"required_revenue": rv["lhs_value"], "comparison_revenue": rv["rhs_value"], "held_constant_metric": "operating margin at maturity", "held_constant_value": held_m, "basis": "break-even basis" if (bm is not None and sm is not None and abs(float(bm) - float(sm)) > 0.0005) else "adopted story", "relation": rv["relation_type"]} if rv else None) margin_case = ({"required_margin": mv["lhs_value"], "comparison_margin": mv["rhs_value"], "held_constant_metric": "revenue path of the adopted story", "held_constant_value": state.value("valuation.story_revenue_maturity"), "basis": "adopted story", "relation": mv["relation_type"]} if mv else None) head = {MORE_DEMANDING: "today's price requires a more demanding revenue/margin combination than the adopted story", LESS_DEMANDING: "today's price requires a less demanding revenue/margin combination than the adopted story", EQUIVALENT: "today's price requires about the same revenue/margin combination as the adopted story", MIXED: "today's price requires a revenue/margin combination that is more demanding on one variable and less on " "the other than the adopted story"}[overall] cases = [] if revenue_case: basis = (f"holding the operating margin at the break-even basis of {pct(held_m)}" if revenue_case["basis"] == "break-even basis" else f"holding the story's operating margin of {pct(held_m)}") cases.append(f"{basis}, the required maturity revenue is {money(rv['lhs_value'])} versus the story's {money(rv['rhs_value'])}") if margin_case: cases.append(f"holding the story's revenue constant, the required operating margin is {pct(mv['lhs_value'])} versus " f"the story's {pct(mv['rhs_value'])}") rendered = head + (": " + "; alternatively, ".join(cases) if cases else "") return {"claim_id": "REL_EXPECTATIONS_STORY", "lhs_claim_id": "PRICE", "rhs_claim_id": "INTRINSIC_VALUE", "lhs_metric_id": "price_implied_expectations", "rhs_metric_id": "adopted_story", "relation_type": overall, "operator": overall, "magnitude": None, "magnitude_unit": None, "magnitude_definition": "frontier of two one-variable break-even solves (each holding the other variable constant)", "comparison_type": "scenario", "lhs_value": None, "rhs_value": None, "allowed": True, "incompatibility_reason": None, "source": "break-even solve (story DCF); adopted story", "derivation": "overall = MORE if each one-variable case requires more than the story, LESS if each requires less, " "EQUIVALENT if each equals it, else MIXED; the cases are alternatives, not a joint requirement", "render_mode": "full_sentence_clause", "rendered": rendered, "conjunctive": False, "logical_structure": ALTERNATIVE_CONDITIONALS, # the variable each case solves for, as a reader names it (paraphrase of both in one requirement = conjunction) "case_variables": {"revenue_case": ["revenue", "sales", "top line", "top-line"], "margin_case": ["margin"]}, # the claims that state one case each: two of them in one sentence would read as a joint requirement # every claim that belongs to a case: the break-even revenue, its level relation, and the growth that reaches it "alternative_case_claims": {"revenue_case": ["BE_REVENUE", "BE_REVENUE_CAGR", "REL_BREAK_EVEN_REVENUE_STORY", "REL_BREAK_EVEN_REVENUE_GROWTH"], "margin_case": ["BE_MARGIN", "REL_BREAK_EVEN_MARGIN_STORY"]}, "signed": None, "presentation_claim_id": None, "years": None, "years_claim_id": None, "condition": None, "revenue_case": revenue_case, "margin_case": margin_case, "components": {"revenue_relation": rv["relation_type"] if rv else None, "margin_relation": mv["relation_type"] if mv else None, "price_implied_revenue": rv["lhs_value"] if rv else None, "story_revenue": rv["rhs_value"] if rv else None, "price_implied_margin": mv["lhs_value"] if mv else None, "story_margin": mv["rhs_value"] if mv else None}, "overall_requirement": overall} GUIDANCE_CONSENSUS = ("CONSENSUS_EPS_FY1", "CONSENSUS_EPS_FY2", "CONSENSUS_EPS_NTM") def guidance_fiscal_year(period_label): """The fiscal year a guidance row covers, from its period label ("full year 2026" / "fiscal year 2026"); None for a quarter, a half or anything that is not a stated fiscal year.""" m = re.fullmatch(r"\s*(?:full|fiscal)\s+year\s+(20\d\d)\s*", str(period_label or ""), re.I) return int(m.group(1)) if m else None def guidance_compatibility(state, row, cons): """Reasons a guidance row may NOT be compared with a consensus metric (empty = compatible): same fiscal period, a fiscal-year horizon on both sides, the same accounting basis (and, for adjusted figures, a registered definition reconciliation), the same security basis and the continuing perimeter.""" import definitions as D from state import Metric reasons = [] fy = guidance_fiscal_year(row.get("period_label")) ptype = str(getattr(cons, "period_type", "") or "") if ptype == "NTM": reasons.append("horizon: a fiscal-year guide against a next-twelve-month consensus") if fy is None: reasons.append(f"fiscal period: the guidance period '{row.get('period_label')}' is not a stated fiscal year") elif ptype != "NTM": end = str(getattr(cons, "period_end", None) or getattr(cons, "period", None) or "")[:4] if not end.isdigit() or int(end) != fy: reasons.append(f"fiscal period: guidance fiscal {fy} against consensus for the year ending {end or 'unknown'}") gb, cb = row.get("accounting_basis"), getattr(cons, "accounting_basis", None) if gb not in ("GAAP", "ADJUSTED"): reasons.append("accounting basis: the guidance basis is not stated") elif gb != cb: reasons.append(f"accounting basis: {gb} guidance against {cb} consensus") elif gb == "ADJUSTED": g = Metric(metric_id=f"guidance_{row.get('metric_id')}", economic_concept="EPS", value=row.get("midpoint"), period_type="FY", accounting_basis="ADJUSTED", adjustment_definition_id=D.COMPANY_NON_GAAP_EPS, adjustment_provider="company release") ok, why = D.definitions_compatible(g, cons) if not ok: reasons.append(f"accounting definition: {why}") issuer_sb = str((state.get("issuer") or {}).get("security_basis") or "ORDINARY_SHARE") if not row.get("per_share"): reasons.append("security basis: the guidance is not a per-share figure") elif str(getattr(cons, "security_basis", "") or "") != issuer_sb: reasons.append(f"security basis: consensus per {getattr(cons, 'security_basis', None)} against guidance per {issuer_sb}") if str(getattr(cons, "perimeter", "") or "") not in ("continuing company", "consolidated"): reasons.append(f"perimeter: consensus on {getattr(cons, 'perimeter', None)}") return reasons def guidance_eps_consensus(state): """(relation item or None, attempts): the company's EPS guidance range against a consensus EPS for the same fiscal year and basis. Every attempt is recorded with its reasons; a relation exists only when all of them pass.""" import catalog as C g = state.guidance.get("company") or {} if g.get("status") != "AVAILABLE": return None, [] item, attempts = None, [] for i, row in enumerate(g.get("metrics") or [], 1): if row.get("concept") != "EPS": continue gid = f"GUIDANCE_{i}" for cid in GUIDANCE_CONSENSUS: cons = C._claim_metric(state, cid) if cons is None or getattr(cons, "value", None) is None: continue why = guidance_compatibility(state, row, cons) attempts.append({"claim_id": "REL_GUIDANCE_EPS_CONSENSUS", "lhs": gid, "rhs": cid, "comparison_type": "guidance_vs_consensus", "reason": "; ".join(why) if why else None}) if why or item is not None: continue lo, hi, cv = float(row["low"]), float(row["high"]), float(cons.value) rel = BELOW if hi < cv else ABOVE if lo > cv else EQUAL rng = f"{C.fmt(C.CURRENCY, lo)}–{C.fmt(C.CURRENCY, hi)}" if abs(hi - lo) > 1e-9 else C.fmt(C.CURRENCY, lo) basis = {"GAAP": "GAAP", "ADJUSTED": "adjusted"}[row["accounting_basis"]] fy = guidance_fiscal_year(row.get("period_label")) word = {BELOW: "is below", ABOVE: "is above", EQUAL: "brackets"}[rel] clause = (f"the company's fiscal {fy} {basis} EPS guidance ({rng}) {word} the fiscal {fy} consensus " f"({C.fmt(C.CURRENCY, cv)})") item = {"claim_id": "REL_GUIDANCE_EPS_CONSENSUS", "lhs_claim_id": gid, "rhs_claim_id": cid, "guidance_metric_id": row.get("metric_id"), "consensus_metric_id": getattr(cons, "metric_id", None), "fiscal_period": f"FY{fy}", "horizon": "FISCAL_YEAR", "accounting_basis": row["accounting_basis"], "security_basis": str((state.get("issuer") or {}).get("security_basis") or "ORDINARY_SHARE"), "low": lo, "high": hi, "consensus_value": cv, "relation": rel, "relation_type": rel, "magnitude": abs(cv - (hi if rel == BELOW else lo)) / cv if rel != EQUAL else 0.0, "comparison_type": "guidance_vs_consensus", "allowed": True, "rendered": clause, "renderings": {"FULL": clause, "FULL_SENTENCE": clause[0].upper() + clause[1:] + "."}, "logical_structure": SIMPLE, "source": "company guidance (8-K); sell-side consensus"} return item, attempts def verify_guidance_relation(state, item): """GUIDANCE_CONSENSUS_COMPATIBILITY re-derived from state: the published item's row and consensus metric still pass every compatibility check and its fields (period, horizon, basis, range, consensus, relation) are the state's.""" import catalog as C rows = (state.guidance.get("company") or {}).get("metrics") or [] try: row = rows[int(str(item.get("lhs_claim_id")).split("_")[1]) - 1] except (ValueError, IndexError): return ["the guidance row it names does not exist"] cid = item.get("rhs_claim_id") cons = C._claim_metric(state, cid) if cid in GUIDANCE_CONSENSUS else None if cons is None or getattr(cons, "value", None) is None: return [f"consensus metric {cid} is not a consensus EPS in state"] bad = guidance_compatibility(state, row, cons) lo, hi, cv = float(row["low"]), float(row["high"]), float(cons.value) rel = BELOW if hi < cv else ABOVE if lo > cv else EQUAL fy = guidance_fiscal_year(row.get("period_label")) want = {"fiscal_period": f"FY{fy}", "horizon": "FISCAL_YEAR", "accounting_basis": row.get("accounting_basis"), "guidance_metric_id": row.get("metric_id"), "consensus_metric_id": getattr(cons, "metric_id", None), "relation": rel} bad += [f"{k} {item.get(k)!r} ≠ state {v!r}" for k, v in want.items() if item.get(k) != v] bad += [f"{k} {item.get(k)} ≠ state {v}" for k, v in (("low", lo), ("high", hi), ("consensus_value", cv)) if item.get(k) is None or abs(float(item[k]) - v) > 1e-9] return bad def verify_forecast_relation(state, rid, item): """FORECAST_RELATION_REQUIRED re-derived from state: an engine-vs-consensus relation is published only for an engine figure that is not consensus itself, on one horizon, perimeter, currency and definition, with its direction.""" bad = [] key = {"REL_FORECAST_EPS_CONSENSUS": "eps", "REL_ENGINE_REVENUE_CONSENSUS": "revenue"}.get(rid) a = (state.forecast.get("consensus_anchor") or {}).get(key) if key else None if isinstance(a, Mapping) and float(a.get("model_weight", 1.0)) <= 0.0: bad.append(f"the engine's {key} forecast is set from consensus ({a.get('reason')}); no independent comparison") spec = next((s for s in _specs(state) if s.claim_id == rid), None) if spec is None: return bad + [f"{rid} has no relation spec"] rel, extra = make(state, spec) if rel is None: return bad + [f"not compatible in state: {(extra or {}).get('reason')}"] d = rel.to_dict() for k in ("relation_type", "lhs_value", "rhs_value"): if item.get(k) != d.get(k): bad.append(f"{k} {item.get(k)!r} ≠ state {d.get(k)!r}") if rid == "REL_ENGINE_REVENUE_CONSENSUS": lm, _ = _metric(state, "FORECAST_REVENUE") for k, v in (("perimeter", getattr(lm, "perimeter", None)), ("currency", getattr(lm, "currency", None)), ("accounting_basis", getattr(lm, "accounting_basis", None)), ("relation", d.get("relation_type"))): if item.get(k) != v: bad.append(f"{k} {item.get(k)!r} ≠ state {v!r}") return bad def dimension(reason): """The compatibility dimension a compatible()/definitions reason names (its leading field).""" r = str(reason or "") for pre, dim in (("adjust", "DEFINITION"), ("concept", "DEFINITION"), ("quantity", "DEFINITION"), ("accounting basis", "ACCOUNTING_BASIS"), ("period", "PERIOD"), ("forward period", "PERIOD"), ("scope", "PERIMETER"), ("perimeter", "PERIMETER"), ("currency", "CURRENCY"), ("security basis", "SECURITY_BASIS")): if r.startswith(pre): return dim return "PERIOD" if "period" in r else "DEFINITION" def absence(state, rid): """Why an engine-vs-consensus relation is not published, typed: None when it exists; else {kind: ANCHORED, key, anchor} (the engine figure is set from consensus), {kind: INCOMPATIBLE, dimensions} or {kind: MISSING}. Readers' sentences derive from this, never from a fixed reason.""" rs = state.relations.get("state") or {} if rid in (rs.get("items") or {}): return None key = {"REL_FORECAST_EPS_CONSENSUS": "eps", "REL_ENGINE_REVENUE_CONSENSUS": "revenue"}.get(rid) a = ((state.forecast.get("consensus_anchor") or {}).get(key)) if key else None if isinstance(a, Mapping) and float(a.get("model_weight", 1.0)) <= 0.0: return {"kind": "ANCHORED", "key": key, "anchor": a} rej = next((x for x in rs.get("rejected") or [] if x.get("claim_id") == rid), None) if rej and rej.get("dimensions"): return {"kind": "INCOMPATIBLE", "dimensions": list(rej["dimensions"])} return {"kind": "MISSING"} def build(state): """{items: {claim_id: relation dict}, rejected: [...], basis: {...}}: every relation this issuer supports.""" items, rejected = {}, [] be = (state.get("valuation.story") or {}).get("breakeven") or {} cond = None bm = be.get("story_margin") sm = state.value("valuation.story_maturity_margin") if bm is not None and sm is not None and abs(float(bm) - float(sm)) > 0.0005: cond = f"holding the operating margin at {float(bm) * 100:.1f}% (the break-even basis)" for spec in _specs(state): rel, extra = make(state, spec, condition=cond if spec.claim_id == "REL_BREAK_EVEN_REVENUE_STORY" else None) if rel is not None: items[spec.claim_id] = dict(rel.to_dict(), renderings=extra, lhs_subjects=list(spec.lhs_subjects), rhs_subjects=list(spec.rhs_subjects), logical_structure=SIMPLE) elif extra and "missing" not in extra.get("reason", ""): rejected.append(extra) # engine revenue vs consensus: the typed fields the relation is established on er = items.get("REL_ENGINE_REVENUE_CONSENSUS") if er: lm, _ = _metric(state, "FORECAST_REVENUE") rm, _ = _metric(state, "CONSENSUS_REVENUE_NTM") er.update(engine_revenue=er["lhs_value"], consensus_revenue=er["rhs_value"], horizon=(getattr(lm, "period_start", None), getattr(lm, "period_end", None)), perimeter=getattr(lm, "perimeter", None), currency=getattr(lm, "currency", None), accounting_basis=getattr(lm, "accounting_basis", None), difference=er["lhs_value"] - er["rhs_value"], relation=er["relation_type"]) # lineage: an engine-vs-consensus relation compares the engine's own view with consensus; a forecast SET from # consensus (model weight 0: the restatement gate) is consensus under another weighting, so the "gap" is a weighting # artefact and no relation is published (a partial anchor carries its weight) anc = state.forecast.get("consensus_anchor") or {} for rid, key in (("REL_FORECAST_EPS_CONSENSUS", "eps"), ("REL_ENGINE_REVENUE_CONSENSUS", "revenue")): a = anc.get(key) if isinstance(anc, Mapping) else None if rid in items and isinstance(a, Mapping): if float(a.get("model_weight", 1.0)) <= 0.0: items.pop(rid) rejected.append({"claim_id": rid, "reason": f"lineage: the engine's {key} forecast is set from sell-side consensus " f"({str(a.get('reason')).lower()} gate), so it is not an independent view to compare with consensus"}) else: items[rid]["consensus_anchor_weight"] = round(1.0 - float(a["model_weight"]), 2) gi, g_attempts = guidance_eps_consensus(state) if gi: items["REL_GUIDANCE_EPS_CONSENSUS"] = gi rejected += [a for a in g_attempts if a.get("reason")] ex = _expectations(state, items) if ex: _head = ex["rendered"].split(":")[0].strip() items["REL_EXPECTATIONS_STORY"] = dict(ex, renderings={"FULL": ex["rendered"], "FULL_SENTENCE": ex["rendered"][0].upper() + ex["rendered"][1:] + ".", # the head clause alone (the cases are in the valuation section) "COMPACT": _head}) trends, trend_rejected = build_trends(state) items.update(trends) basis = {"breakeven_story_margin": bm, "adopted_story_margin": sm, "breakeven_story_value": be.get("story_value"), "adopted_story_value": state.value("valuation.intrinsic_value")} return {"items": items, "rejected": rejected, "basis": basis, "trends_rejected": trend_rejected, "assessments": assess_assumptions(state)} def pair_index(rel_state): """{frozenset({lhs claim, rhs claim}): relation claim id}.""" out = {} for cid, r in ((rel_state or {}).get("items") or {}).items(): if r.get("lhs_claim_id") and r.get("rhs_claim_id") and cid != "REL_EXPECTATIONS_STORY": out.setdefault(frozenset({r["lhs_claim_id"], r["rhs_claim_id"]}), cid) return out def magnitude_bindings(rel_state): """{presentation / magnitude claim id: (lhs claim, rhs claim)}: a magnitude belongs to the pair it measures.""" return {r["presentation_claim_id"]: (r["lhs_claim_id"], r["rhs_claim_id"]) for r in ((rel_state or {}).get("items") or {}).values() if r.get("presentation_claim_id")} def check(state): """[(check id, problem)]: identity, magnitude and compatibility of every stored relation (REL1–REL3).""" from state import compatible bad = [] rs = state.get("relations.state") or {} specs = {s.claim_id: s for s in _specs(state)} for cid, r in (rs.get("items") or {}).items(): if cid.startswith("TREND_"): bad += [("REL1", p) for p in _check_trend(state, cid, r)] continue if cid == "REL_EXPECTATIONS_STORY": ex = _expectations(state, {k: v for k, v in rs["items"].items() if k != cid and not k.startswith("TREND_")}) if r.get("logical_structure") == "ALTERNATIVE_CONDITIONALS" and r.get("revenue_case") and r.get("margin_case") \ and "alternatively" not in str(r.get("rendered") or ""): bad.append(("REL1", f"{cid}: alternative conditional cases rendered without their alternative structure")) if ex is None or ex["relation_type"] != r.get("relation_type"): bad.append(("REL1", f"{cid}: stored polarity {r.get('relation_type')} ≠ the polarity its components give " f"({ex and ex['relation_type']})")) continue if cid == "REL_GUIDANCE_EPS_CONSENSUS": # typed by guidance_compatibility, not a Spec bad += [("REL3", f"{cid}: {w}") for w in verify_guidance_relation(state, r)] continue spec = specs.get(cid) if spec is None: bad.append(("REL1", f"{cid}: no relation type defines it")) continue if (r.get("lhs_claim_id"), r.get("rhs_claim_id")) != (spec.lhs, spec.rhs): bad.append(("REL1", f"{cid}: lhs/rhs {r.get('lhs_claim_id')}/{r.get('rhs_claim_id')} ≠ {spec.lhs}/{spec.rhs}")) continue lm, _ = _metric(state, spec.lhs) rm, _ = _metric(state, spec.rhs) ok, why = compatible(lm, rm, spec.comparison_type) if (lm is not None and rm is not None) else (False, ["missing"]) if not ok or not r.get("allowed"): bad.append(("REL3", f"{cid}: {spec.lhs} vs {spec.rhs} not compatible ({'; '.join(why)})")) continue if abs(float(lm.value) - float(r["lhs_value"])) > 1e-9 or abs(float(rm.value) - float(r["rhs_value"])) > 1e-9: bad.append(("REL1", f"{cid}: stored endpoint values differ from the state")) fresh, fresh_modes = make(state, spec, condition=r.get("condition")) if fresh is None: continue if r.get("renderings") and r["renderings"] != fresh_modes: bad.append(("REL1", f"{cid}: a render mode does not follow the stored lhs, operator, magnitude and rhs")) if fresh.relation_type != r.get("relation_type") or fresh.operator != r.get("operator"): bad.append(("REL1", f"{cid}: relation {r.get('relation_type')} ≠ {fresh.relation_type} for these values")) if abs(fresh.magnitude - float(r["magnitude"])) > 1e-9 or r.get("magnitude_definition") != spec.definition: bad.append(("REL2", f"{cid}: magnitude {r.get('magnitude')} ≠ {spec.definition} = {fresh.magnitude:.6f}")) if fresh.rendered != r.get("rendered"): bad.append(("REL1", f"{cid}: rendering does not state lhs, relation and rhs as stored")) if spec.presentation: pd = __import__("catalog").claim_def(state, spec.presentation) pv = state.value(pd.path) if pd else None if spec.definition == CAGR: if pv is not None and abs(float(pv) - fresh.magnitude) > 0.002: bad.append(("REL2", f"{cid}: {spec.presentation} {pv} is not {CAGR} of its pair ({fresh.magnitude:.4f})")) elif pv is not None and abs(float(pv) - fresh.signed) > 1e-6: bad.append(("REL2", f"{cid}: {spec.presentation} {pv} ≠ lhs/rhs − 1 = {fresh.signed:.6f}")) return bad # ── the surface a reader takes as asserting expectation polarity ───────────────────────────────────────── # Owned by the ImpliedExpectations relation type: prose may assert a polarity only by citing # [[REL_EXPECTATIONS_STORY]], and only the polarity that relation holds. POLARITY_SURFACE = { MORE_DEMANDING: ("more demanding", "more optimism", "more optimistic", "priced for perfection", "prices in perfection", "demands more than the story", "aggressive relative to the story"), LESS_DEMANDING: ("less demanding", "less optimism", "less optimistic", "conservative relative to", "conservative versus", "conservative compared", "undemanding", "less than the story requires", "below what the story assumes"), } # a comparison between a fundamental and what the price implies is a polarity statement whatever its wording # ("growth is stronger than the market is pricing in" = the price requires LESS than the business delivers). The # subject/predicate of a market-implied requirement and the order words are the relation's own; which side is the # subject of the comparative decides the polarity. _IMPLIED_REF = (r"\b(?:market|price|valuation|stock|shares|break-?even)\b[^.;:]{0,40}?\b(?:prices? in|pricing in|priced in|" r"implies|implied|implying|embeds|embedded|embedding|bakes in|baked in|expects|expecting|requires|requiring|" r"needs|needing|assumes|assuming|discounts|discounting)\b|\bmarket[- ]implied\b|\bimplied by (?:the )?" r"(?:price|market|valuation)\b|\bwhat (?:the )?(?:market|price|stock|shares) (?:is |are )?(?:pric|impl|expect|assum)\w*") # comparative constructions: a comparative word, or "more / less / fewer than" _CMP_UP = (r"\b(?:stronger|higher|faster|better|greater|above|exceed\w*|ahead of|outpac\w*|beyond|beat\w*)\b|" r"\bmore\b(?=[^.;:]{0,30}\bthan\b)") _CMP_DOWN = (r"\b(?:weaker|lower|slower|worse|below|short of|falls? short|lag\w*|behind|trail\w*)\b|" r"\b(?:less|fewer)\b(?=[^.;:]{0,30}\bthan\b)") _FUNDAMENTAL = r"\b(?:growth|revenue|sales|margins?|earnings|eps|profit\w*|top[- ]line|execution|fundamentals)\b" def _structural_polarity(low): import re as _re out = [] for clause in _re.split(r"[;:—–]|,\s*(?:but|while|whereas)\b", low): ref = _re.search(_IMPLIED_REF, clause) if not ref or not _re.search(_FUNDAMENTAL, clause): continue for rx, up in ((_CMP_UP, True), (_CMP_DOWN, False)): m = _re.search(rx, clause) if not m: continue implied_is_subject = ref.start() < m.start() out.append(MORE_DEMANDING if up == implied_is_subject else LESS_DEMANDING) return out def polarity_asserted(sentence, structural=None): """Polarities a sentence asserts: the named surface (structural=False), the comparative against what the price implies (structural=True), or both (None).""" low = str(sentence or "").lower() pols = [pol for pol, phrases in POLARITY_SURFACE.items() if any(p in low for p in phrases)] if structural is not True else [] return list(dict.fromkeys(pols + (_structural_polarity(low) if structural is not False else []))) # ── StockPerformance: what the share price did is a typed, signed return claim ────────────────────────────── # "a weak year for the stock" is a statement about a measured return; it needs the return claim it describes, with # the same sign (and the twelve-month claim when it speaks of a year). Conditionals ("if the shares sell off") state # no performance. NEGATIVE_PERF, POSITIVE_PERF = "NEGATIVE", "POSITIVE" PERFORMANCE_SURFACE = {NEGATIVE_PERF: ("weak year", "weak stretch", "tough year", "sell-off", "selloff", "sold off", "slump", "slid", "beaten down", "beaten-down", "underperform", "tumbl", "plung", "de-rated", "derated", "fell", "dropped", "declined", "lost ground"), POSITIVE_PERF: ("strong year", "rallied", "rally", "outperform", "surged", "run-up", "climbed", "rose", "gained ground")} PERFORMANCE_SUBJECTS = ("stock", "shares", "share price", "the equity") _CONDITIONAL = r"\b(?:if|would|could|may|might|should|risk of|risk that|unless|in case)\b" STOCK_RETURN_CLAIMS = {"RETURN_3M": "3M", "RETURN_12M": "12M", "RETURN_3M_PHRASE": "3M", "RETURN_12M_PHRASE": "12M"} def performance_asserted(sentence, issuer_words=()): """[(sign, matched text, needs a twelve-month claim)] for statements about how the stock has performed. The issuer's own names are subjects too ("S&P Global … after a year of underperformance").""" import re as _re out = [] subjects = tuple(PERFORMANCE_SUBJECTS) + tuple(w for w in issuer_words if w) for clause in _re.split(r"[;:—–.]", str(sentence or "").lower()): if _re.search(_CONDITIONAL, clause) or not any(_re.search(r"(? b) == higher_is_aggressive out[cid] = {"assessment": AGGRESSIVE if agg else CONSERVATIVE, "benchmark_claim_id": bid, "basis": basis} return out # ── TrendRelationship: an earlier and a later value of ONE metric on ONE perimeter ──────────────────────── ACCELERATING, DECELERATING, IMPROVING, DETERIORATING, EXPANDING, CONTRACTING = ( "ACCELERATING", "DECELERATING", "IMPROVING", "DETERIORATING", "EXPANDING", "CONTRACTING") _TREND_SPECS = ( ("TREND_REVENUE_GROWTH", "LIFECYCLE_TRAILING_GROWTH", "LIFECYCLE_FORWARD_GROWTH", "growth", ACCELERATING, DECELERATING, "revenue growth"), ("TREND_REVENUE", "REVENUE_TTM", "FORECAST_REVENUE", "revenue", EXPANDING, CONTRACTING, "revenue"), ("TREND_OPERATING_MARGIN", "OPERATING_MARGIN_TTM", "FORECAST_OPERATING_MARGIN", "margin", IMPROVING, DETERIORATING, "the operating margin"), ("TREND_EPS", "EPS_ADJUSTED_TTM", "FORECAST_EPS", "eps", EXPANDING, CONTRACTING, "adjusted EPS"), ) _TREND_VERB = {ACCELERATING: "accelerates", DECELERATING: "decelerates", IMPROVING: "improves", DETERIORATING: "deteriorates", EXPANDING: "grows", CONTRACTING: "declines"} def _trend(state, spec): from state import _same_perimeter, compatible import catalog as C rid, e_id, l_id, fam, up, down, noun = spec em, ed = _metric(state, e_id) lm, ld = _metric(state, l_id) if em is None or lm is None or em.value is None or lm.value is None: return None, {"relation_id": rid, "reason": "an input is missing"} ok, why = compatible(em, lm, "trend") rec = {"relation_id": rid, "metric_family": fam, "earlier_metric_id": em.metric_id, "later_metric_id": lm.metric_id, "earlier_claim_id": e_id, "later_claim_id": l_id, "earlier_value": float(em.value), "later_value": float(lm.value), "perimeter_compatible": bool(_same_perimeter(em, lm)), "period_compatible": em.period_class == "TRAILING" and lm.period_class == "FORWARD"} if not (ok and rec["period_compatible"]): return None, dict(rec, relation=None, allowed=False, reason="; ".join(why) or "not an earlier → later pair") rel = up if lm.value > em.value else down txt = (f"{noun} {_TREND_VERB[rel]} from {C.fmt(ed.stype, em.value)} (trailing) to {C.fmt(ld.stype, lm.value)} (forward)") return dict(rec, claim_id=rid, relation=rel, relation_type=rel, allowed=True, rendered=txt, renderings={"FULL": txt}, logical_structure=SIMPLE, lhs_claim_id=l_id, rhs_claim_id=e_id, magnitude=None), None def build_trends(state): """({claim_id: TrendRelationship} for compatible pairs only, [rejected pairs with the reason]).""" items, rejected = {}, [] for spec in _TREND_SPECS: t, rej = _trend(state, spec) if t: items[t["claim_id"]] = t elif rej: rejected.append(rej) return items, rejected def _check_trend(state, cid, r): spec = next((x for x in _TREND_SPECS if x[0] == cid), None) if spec is None: return [f"{cid}: no trend type defines it"] fresh, rej = _trend(state, spec) if fresh is None: return [f"{cid}: stored trend has no compatible earlier comparator ({(rej or {}).get('reason')})"] return [] if fresh["relation"] == r.get("relation") and fresh["rendered"] == r.get("rendered") else \ [f"{cid}: stored trend {r.get('relation')} ≠ {fresh['relation']}"] # the surface of a trend CONCLUSION (owned by the TrendRelationship type) and the metric nouns it can be about TREND_SURFACE = {ACCELERATING: ("accelerat", "re-accelerat", "picking up", "picks up", "speeding up"), DECELERATING: ("decelerat", "slowing", "slows", "slowdown", "slowed"), IMPROVING: ("improv",), DETERIORATING: ("deteriorat", "worsen"), EXPANDING: ("expand", "expansion"), CONTRACTING: ("contract", "compress", "shrink")} TREND_SUBJECTS = ("revenue", "sales", "growth", "top line", "top-line", "margin", "earnings", "eps", "profit", "income") def trend_asserted(sentence): """[(trend class, the matched text)] for trend conclusions about a financial metric in a sentence.""" import re as _re low = str(sentence or "").lower() out = [] for cls, stems in TREND_SURFACE.items(): for st in stems: for m in _re.finditer(r"\b" + _re.escape(st), low): win = low[max(0, m.start() - 80):m.end() + 60] # a negated or conditional verb asserts no trend ("even if the multiple does not expand") clause = _re.split(r"[;:,—–.]", low[:m.start()])[-1] if _re.search(r"\b(?:if|unless|would|could|may|might|not|never|no)\b|n't\b", clause[-40:]): continue if any(_re.search(r"\b" + _re.escape(n) + r"\b", win) for n in TREND_SUBJECTS): out.append((cls, m.group(0))) return out # ── valuation anchors: a direction word about an anchor needs the anchor's relation (and must agree with it) ── VALUATION_ANCHORS = {"REL_PRICE_INTRINSIC": ("intrinsic value", "story value", "intrinsic"), "REL_PRICE_MECHANICAL": ("dcf", "cross-check", "mechanical"), "REL_PRICE_MARKET_COMPOSITE": ("composite", "market pricing")} VALUATION_DIRECTION = {ABOVE: ("premium", "above", "expensive", "overvalued", "rich"), BELOW: ("discount", "below", "cheap", "undervalued")} # the predicate of a price-implied requirement ("the price requires / needs / demands / embeds …"): owned by the # ImpliedExpectations relation, used only to recognise a paraphrase of it REQUIREMENT_PREDICATE = ("requir", "need", "demand", "embed", "prices in", "priced in", "implies", "implied", "bakes in", "baked in", "assum", "justif", "break-even", "breakeven") REQUIREMENT_SUBJECT = ("price", "valuation", "break-even", "breakeven", "market", "story", "stock", "shares") # how an ALTERNATIVE_CONDITIONALS proposition marks its alternatives (a combination or trade-off, not a joint requirement) ALTERNATIVE_MARKERS = ("combination", "alternatively", "either", "trade-off", "tradeoff", "trade off", " or ") # ── cross-family fundamental comparisons ("earnings compound faster than revenue"): a relation between two metric # families. No typed relation connects different families, so such a comparison cannot be stated. ───────────── FUNDAMENTAL_FAMILIES = {"EARNINGS": ("earnings power", "earnings", "eps", "profits", "profit", "net income"), "REVENUE": ("revenue", "sales", "top line", "top-line"), "MARGIN": ("margins", "margin"), "CASH_FLOW": ("free cash flow", "fcf", "cash flow")} _FAMILY_CMP = (r"\b(?:faster|slower|more quickly|more slowly|quicker)\s+than\b|\boutpac\w*|\boutgrow\w*|\boutrun\w*|" r"\bgrow\w*\s+ahead\s+of\b|\blag\w*\s+(?:behind\s+)?|\bahead\s+of\b") def family_comparison_asserted(sentence): """[(family A, family B, matched comparative)] for a comparison between two different metric families.""" import re as _re out = [] for clause in _re.split(r"[;:—–.]|\b(?:but|while|whereas|because)\b", str(sentence or "").lower()): m = _re.search(_FAMILY_CMP, clause) if not m: continue def fam(text, last): hits = [(mm.start(), f) for f, words in FUNDAMENTAL_FAMILIES.items() for w in words for mm in _re.finditer(r"\b" + _re.escape(w) + r"\b", text)] if not hits: return None return (max(hits) if last else min(hits))[1] a, b = fam(clause[:m.start()], True), fam(clause[m.end():], False) if a and b and a != b: out.append((a, b, m.group(0).strip())) return out # ── value proximity: "close to fair value" is an EQUAL-within-a-band relation to a named anchor ─────────────── CLOSENESS_BAND = 0.05 PROXIMITY_SURFACE = (r"\b(?:close to|near|around|roughly at|about|in line with|at|approximately)\s+(?:its\s+|their\s+|the\s+)?" r"(?:fair|intrinsic|story|underlying)\s+value\b|\bfairly\s+(?:valued|priced)\b|\bfair[- ]valued\b|" r"\bpriced\s+(?:about\s+)?right\b|\bat\s+fair\s+value\b")