| """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" |
| |
| |
| MORE_DEMANDING, LESS_DEMANDING, MIXED, EQUIVALENT = ("MORE_DEMANDING_COMBINATION", "LESS_DEMANDING_COMBINATION", |
| "MIXED_COMBINATION", "EQUIVALENT_COMBINATION") |
| |
| 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"} |
|
|
| |
| 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 |
| 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 |
| presentation_claim_id: str | None = None |
| years: float | None = None |
| years_claim_id: str | None = None |
| condition: str | None = None |
|
|
| def to_dict(self): |
| return asdict(self) |
|
|
|
|
| @dataclass(frozen=True) |
| class Spec: |
| claim_id: str |
| lhs: str |
| rhs: str |
| comparison_type: str |
| definition: str |
| above: str |
| below: str |
| presentation: str | None = None |
| years: str | None = None |
| lhs_words: str = "" |
| rhs_words: str = "" |
| source: str = "" |
| rhs_noun: str = "" |
| rhs_short: str = "" |
| lhs_subjects: tuple = () |
| rhs_subjects: tuple = () |
|
|
|
|
| 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", |
| |
| |
| 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")), |
| |
| |
| |
| 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"), |
| |
| |
| 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" |
| |
| |
| 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}" |
| |
| 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, |
| |
| "case_variables": {"revenue_case": ["revenue", "sales", "top line", "top-line"], "margin_case": ["margin"]}, |
| |
| |
| "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) |
| |
| 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"]) |
| |
| |
| |
| 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:] + ".", |
| |
| "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": |
| 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 |
|
|
|
|
| |
| |
| |
| 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"), |
| } |
|
|
|
|
| |
| |
| |
| |
| _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*") |
| |
| _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 []))) |
|
|
|
|
| |
| |
| |
| |
| 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"(?<!\w)" + _re.escape(w) + r"(?!\w)", clause) for w in subjects): |
| continue |
| for sign, stems in PERFORMANCE_SURFACE.items(): |
| for st in stems: |
| m = _re.search(r"\b" + _re.escape(st), clause) |
| if m: |
| out.append((sign, m.group(0), "year" in st or "year" in clause[max(0, m.start() - 30):m.end() + 30])) |
| return out |
|
|
|
|
| |
| |
| CONSERVATIVE, AGGRESSIVE, NEUTRAL, NO_HAIRCUT = "CONSERVATIVE", "AGGRESSIVE", "NEUTRAL", "NO_HAIRCUT" |
| ASSESSMENT_WORDS = {CONSERVATIVE: ("conservative", "cautious", "prudent"), |
| AGGRESSIVE: ("aggressive", "generous", "optimistic", "rosy", "heroic")} |
|
|
|
|
| |
| |
| LEVEL_WORDS = {"HIGH": ("high", "elevated", "hefty", "steep", "punitive", "demanding"), "LOW": ("low", "light", "thin", "undemanding")} |
| HIGHER_IS_CONSERVATIVE = ("STORY_WACC", "STORY_WACC_STABLE", "STORY_FAILURE") |
|
|
|
|
| def level_assessment(cid, level): |
| """CONSERVATIVE / AGGRESSIVE that 'high' or 'low' asserts for an assumption claim.""" |
| cons = (level == "HIGH") == (cid in HIGHER_IS_CONSERVATIVE) |
| return CONSERVATIVE if cons else AGGRESSIVE |
|
|
|
|
| def assess_assumptions(state): |
| """{assumption claim: {assessment, benchmark_claim_id, basis}} for the assumptions with a defensible benchmark.""" |
| def v(path): |
| x = state.value(path) |
| return None if x is None else float(x) |
| out = {} |
| fp = v("valuation.story.assumptions.failure_probability") |
| if fp is not None: |
| out["STORY_FAILURE"] = {"assessment": NO_HAIRCUT if fp <= 0 else CONSERVATIVE, "benchmark_claim_id": None, |
| "basis": "0% is no distress haircut at all: the least conservative setting" if fp <= 0 |
| else "a positive failure probability haircuts the value"} |
| pairs = (("STORY_MARGIN", "valuation.story_maturity_margin", "STORY_START_MARGIN", "valuation.story_base.operating_margin", |
| True, "maturity operating margin vs the starting margin"), |
| ("STORY_WACC", "valuation.discount_rate_path.initial_wacc_metric", "WACC_BENCHMARK", |
| "valuation.discount_rate_path.benchmark_wacc_metric", False, "initial cost of capital vs the mature-market benchmark"), |
| ("STORY_WACC_STABLE", "valuation.discount_rate_path.terminal_wacc_metric", "WACC_BENCHMARK", |
| "valuation.discount_rate_path.benchmark_wacc_metric", False, "terminal cost of capital vs the mature-market benchmark")) |
| for cid, path, bid, bpath, higher_is_aggressive, basis in pairs: |
| a, b = v(path), v(bpath) |
| if a is None or b is None: |
| continue |
| if abs(a - b) <= 0.0005: |
| out[cid] = {"assessment": NEUTRAL, "benchmark_claim_id": bid, "basis": basis + " (equal)"} |
| else: |
| agg = (a > b) == higher_is_aggressive |
| out[cid] = {"assessment": AGGRESSIVE if agg else CONSERVATIVE, "benchmark_claim_id": bid, "basis": basis} |
| return out |
|
|
|
|
|
|
| |
| 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']}"] |
|
|
|
|
| |
| 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] |
| |
| 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 = {"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")} |
|
|
|
|
| |
| |
| 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") |
| |
| ALTERNATIVE_MARKERS = ("combination", "alternatively", "either", "trade-off", "tradeoff", "trade off", " or ") |
|
|
|
|
| |
| |
| 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 |
|
|
|
|
| |
| 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") |
|
|
