diff --git a/CHANGELOG.md b/CHANGELOG.md index 7074b6e..2626e5c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,75 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed — sense-check adversarial-review findings (PR #143) + +- **EDGAR picked the wrong year.** A company's annual filing repeats the prior + years' revenue for comparison, and the SEC stamps them all with the filing's + year — so the lookup could return a 2-year-old figure labelled as the latest. + Revenue is now selected by the fact's own reporting period end, and the + displayed fiscal year comes from that period, never the filing stamp. +- **The MODELED label is now always shown.** Market shares are always analyst + estimates, so the implied TAM is a modeled cross-check even when every revenue + comes from a filing (Q20) — the label no longer disappears on all-sourced runs. +- **Overlapping shares no longer produce a contradictory range.** When the + players' shares sum past 100%, the sensitivity band, leverage, and residual + are suppressed — only the revenue floor and the overlap warning are shown. +- **EDGAR revenue is flagged as total-company.** An `edgar:` figure is the whole + company's revenue, not one market's — resolution prints a caveat and the + player line is tagged `(filing, total-co)`. +- More revenue tags tried (`…IncludingAssessedTax`, `SalesRevenueNet`, + `SalesRevenueServicesNet`); the no-revenue error names every tag tried; + a bare share token that parses to exactly 100% gets a warning (the `1` → + 100% trap); CSV player files parse directly (a `|` in a company name can no + longer break parsing) with an optional `sourced` column and one shared EDGAR + client across rows. + +### Added — `sense_check` MCP tool + +- The same market-share-implied TAM cross-check, callable by agents: players in, + implied TAM + band + floor + leverage + residual + warnings + assumptions out, + with the MODELED (Q20) marker carried machine-readably (`"modeled": true`) on + every payload. `edgar_ticker` resolution is gated behind + `STRATA_MCP_LIVE_CONNECTORS=1`, like every live connector call. + +### Added — sense-check revenue auto-pull from SEC EDGAR + +- **`--player "Name|edgar:TICKER|share"`** resolves a US-listed company's latest + annual revenue live from SEC EDGAR XBRL (regulator-tier) instead of typing an + estimate — so for public players the revenue is **sourced**, not the analyst's + guess (closing the "private revenue is itself a guess" critique). It picks the + most recent annual fact across revenue concepts (`Revenues` and the ASC 606 + `RevenueFromContractWithCustomerExcludingAssessedTax`), prints the figure + + fiscal year + filer, and counts it as sourced. EDGAR-only (US public cos); + private players stay estimates. Gated by `STRATA_OFFLINE` (refuses live lookups + offline). yfinance is not used — it exposes quotes, not revenue. + +### Fixed — EDGAR fact parsing crashed on present-but-null `fy`/`fp` + +- SEC XBRL `companyconcept` facts frequently carry `"fy": null` / `"fp": null` + (key present, value null). `_parse_fact` guarded on key presence, so + `int(None)` raised `TypeError` on live company-concept data. Now guards on the + value. (Surfaced by the revenue auto-pull exercising the company-concept path.) + +### Added — `strata sense-check`: market-share-implied TAM cross-check + +- **A reusable sense check for any market sizing.** Given the top players in a + market (each with a revenue and an estimated market share), it derives an + implied TAM and cross-checks it against the sourced number: + `implied TAM = sum(player revenue) / sum(market share)`, plus a **revenue + floor** (a TAM can't be below the revenue already inside it) and a + **per-player spread** (each `revenue/share` should imply the same TAM; a wide + spread flags inconsistent share estimates). This is the `COMPARABLE_REVENUE` + method run as an analyst cross-check. +- **Honest by construction (Q20):** market shares are analyst *estimates*, so the + implied TAM is **MODELED** — a cross-check only, never written back as a sourced + primitive. With `--concept ` it compares the implied TAM head-to-head with + the curated market's sourced scenario range and returns `within | above | below` + — the same spirit as the evaluator's `convergence` banner. +- Surfaces: CLI `strata sense-check --player "Name|revenue|share" ... [--concept + ] [--players-file players.csv]` (revenue accepts `31e9`/`3.1b`/`$300m`, + share `0.15` or `15`). `strata size` now prints a one-line pointer so the + cross-check is offered on every sizing. ### Added — `strata portfolio add`: add one company without a CSV - **You can now add a single portfolio company from the CLI** instead of writing diff --git a/backend/src/strata/cli.py b/backend/src/strata/cli.py index 994ae67..96db00a 100644 --- a/backend/src/strata/cli.py +++ b/backend/src/strata/cli.py @@ -4,6 +4,8 @@ import asyncio import base64 +import csv +import os import re from dataclasses import dataclass from datetime import UTC @@ -14,8 +16,9 @@ from dotenv import load_dotenv from strata import __version__ +from strata.connectors.edgar import EDGARClient from strata.evaluator import EvaluatorError, size_concept -from strata.loader import LoaderError +from strata.loader import LoaderError, load_data_dir from strata.models import QuestionCategory, ResolvedAnswer, TAMOutput, Unit from strata.registry import Registry, RegistryError from strata.report import ( @@ -27,7 +30,22 @@ render_premium_html, render_premium_pdf, ) +from strata.report.narrative import usd from strata.sizing.cli import portfolio_app +from strata.sizing.revenue_lookup import RevenueLookupError, resolve_revenue, to_player +from strata.sizing.sense_check import ( + MODELED_LABEL, + SOURCED_TOKENS, + Player, + SenseCheckError, + SenseCheckResult, + normalize_share, + parse_money, + sense_check, + share_token_warning, + split_player_spec, +) +from strata.sizing.service import sourced_scenarios from strata.validator import Severity, ValidationFinding from strata.validator import validate as run_validate @@ -82,6 +100,20 @@ _VERIFY_VALUE_OPT = typer.Option( None, "--value", help="The numeric figure the claim asserts (same unit as the market)." ) +# `sense-check` — market-share-implied (comparable-revenue) cross-check. +_SC_PLAYER_OPT = typer.Option( + [], + "--player", + "-p", + help="A top player as 'Name|revenue|share' (repeatable). revenue: 31e9 / 3.1b / $300m, " + "or 'edgar:TICKER' to pull a SOURCED revenue from SEC EDGAR; share: 0.15 or 15.", +) +_SC_FILE_OPT = typer.Option( + None, "--players-file", help="CSV with columns: name,revenue,market_share." +) +_SC_CONCEPT_OPT = typer.Option( + None, "--concept", help="Curated concept slug to cross-check the implied TAM against." +) # A 64-char hex string is a TAM state hash; anything else is a concept slug. _HASH_RE = re.compile(r"^[0-9a-f]{64}$") @@ -152,6 +184,10 @@ def size( typer.echo(f"ERROR registry: {exc}", err=True) raise typer.Exit(code=1) from exc _render(tam) + typer.echo( + f"\nSense-check (comparable-revenue): strata sense-check --concept {concept_slug} " + f'--player "Name|revenue|share" ...' + ) def _render(tam: TAMOutput) -> None: @@ -397,6 +433,200 @@ def verify_cmd( # ── Database sub-commands ──────────────────────────────────────────── + +def _share_from_token(token: str, ctx: str) -> float: + try: + share = normalize_share(float(token)) + except ValueError as exc: + raise SenseCheckError(f"could not parse share {token!r} in {ctx}") from exc + share_warning = share_token_warning(token, share) + if share_warning: + typer.echo(f" WARN {share_warning}") + return share + + +def _edgar_player(name: str, ticker: str, share: float, client: EDGARClient) -> Player: + """Resolve an ``edgar:TICKER`` revenue token to a SOURCED (but total-company) + revenue via SEC EDGAR (live HTTP).""" + if not ticker: + raise SenseCheckError(f"empty EDGAR ticker for player {name!r}") + if os.environ.get("STRATA_OFFLINE") == "1": + raise SenseCheckError(f"{name!r} needs a live EDGAR lookup but STRATA_OFFLINE=1") + try: + rr = resolve_revenue(ticker, client=client) + except RevenueLookupError as exc: + raise SenseCheckError(str(exc)) from exc + typer.echo( + f" resolved {name} revenue {usd(rr.revenue)} " + f"(EDGAR {rr.concept} FY{rr.fiscal_year}, {rr.entity_name})" + ) + return to_player(name, share, rr) + + +def _player_from_spec(spec: str, client: EDGARClient) -> Player: + """Parse a --player spec. An ``edgar:TICKER`` revenue token resolves to a + sourced total-company revenue; any other token is an analyst estimate + parsed offline.""" + parts = split_player_spec(spec) + share = _share_from_token(parts[2], spec) + if parts[1].lower().startswith("edgar:"): + return _edgar_player(parts[0], parts[1].split(":", 1)[1].strip(), share, client) + sourced = len(parts) == 4 and parts[3].lower() in SOURCED_TOKENS + return Player( + name=parts[0], revenue=parse_money(parts[1]), market_share=share, revenue_sourced=sourced + ) + + +def _collect_players(specs: list[str], players_file: Path | None) -> list[Player]: + # EDGARClient construction is cheap and no-network — one client shared by + # every edgar: row so N public players reuse one HTTP pool + disk cache. + with EDGARClient() as client: + players = [_player_from_spec(s, client) for s in specs] + if players_file is not None: + with players_file.open(newline="", encoding="utf-8") as fh: + # CSV fields go straight into Player — never round-tripped + # through the pipe DSL, so a '|' in a company name can't inject + # extra fields. + for row in csv.DictReader(fh): + name = (row.get("name") or "").strip() + if not name: + continue + rev = (row.get("revenue") or "").strip() + share = _share_from_token( + (row.get("market_share") or "").strip(), f"CSV row {name!r}" + ) + if rev.lower().startswith("edgar:"): + players.append( + _edgar_player(name, rev.split(":", 1)[1].strip(), share, client) + ) + continue + sourced = (row.get("sourced") or "").strip().lower() in SOURCED_TOKENS + players.append( + Player( + name=name, + revenue=parse_money(rev), + market_share=share, + revenue_sourced=sourced, + ) + ) + return players + + +@app.command("sense-check") +def sense_check_cmd( + player: list[str] = _SC_PLAYER_OPT, + players_file: Path | None = _SC_FILE_OPT, + concept: str | None = _SC_CONCEPT_OPT, + data_dir: Path = _DATA_DIR_OPT, +) -> None: + """Market-share-implied TAM cross-check: sum(top-player revenue) / sum(market share). + + A sense check, not a sourced number — market shares are analyst estimates, so + the implied TAM is MODELED and never persisted (Q20). With --concept, the + implied TAM is compared head-to-head with the curated market's sourced + scenario range (within / above / below). + """ + try: + players = _collect_players(player, players_file) + except SenseCheckError as exc: + typer.echo(f"ERROR {exc}", err=True) + raise typer.Exit(code=2) from exc + + scenarios: tuple[float, ...] | None = None + if concept: + try: + reg = Registry.from_loaded(load_data_dir(data_dir)) + scenarios = sourced_scenarios(reg, concept) + except (LoaderError, EvaluatorError, RegistryError) as exc: + typer.echo(f"ERROR could not size concept {concept!r}: {exc}", err=True) + raise typer.Exit(code=1) from exc + + try: + result = sense_check(players, sourced_scenarios=scenarios) + except SenseCheckError as exc: + typer.echo(f"ERROR {exc}", err=True) + raise typer.Exit(code=2) from exc + _render_sense_check(result, concept) + + +def _render_sense_check(r: SenseCheckResult, concept: str | None) -> None: + typer.echo("Market-share-implied TAM (comparable-revenue sense-check)") + typer.echo(f" {MODELED_LABEL}") + # All four band fields are None together (Σshare ≥ 100% suppresses them) — + # collapsed into ONE optional tuple so every suppression gate is the same check. + band = ( + (r.implied_low, r.implied_high, r.leverage, r.residual) + if r.implied_low is not None + and r.implied_high is not None + and r.leverage is not None + and r.residual is not None + else None + ) + if band is not None: + implied_low, implied_high = band[0], band[1] + typer.echo( + f" implied TAM {usd(r.implied_tam)} " + f"(= revenue {usd(r.revenue_floor)} / share {r.combined_share:.0%})" + ) + typer.echo( + f" sensitivity {usd(implied_low)} - {usd(implied_high)} " + f"(shares +/-{r.share_tolerance:.0%})" + ) + typer.echo(f" revenue floor {usd(r.revenue_floor)} (a TAM cannot be below this)") + if band is not None: + leverage, residual = band[2], band[3] + typer.echo( + f" leverage {leverage:.1f}x " + f"(implied TAM is {leverage:.1f}x captured revenue; 1 / {r.combined_share:.0%} share)" + ) + typer.echo( + f" residual {usd(residual)} (revenue attributed to players outside the set)" + ) + typer.echo( + f" per-player spread {r.spread_ratio:.1f}x ({r.n_sourced}/{r.n_players} revenues sourced)" + ) + for pi in r.per_player: + if pi.revenue_note: + tag = f" ({pi.revenue_note})" + elif pi.revenue_sourced: + tag = "" + else: + tag = " (est.)" + typer.echo( + f" {pi.name:24} {usd(pi.revenue)} @ {pi.market_share:.0%}" + f" -> implied {usd(pi.implied_tam)}{tag}" + ) + # With Σshare ≥ 100% the implied TAM is not defensible, so a verdict on it + # would mislead — only the floor + the overlap warning are rendered. The + # sourced trio is always set when verdict is; the None checks just narrow. + if ( + band is not None + and concept + and r.verdict is not None + and r.sourced_low is not None + and r.sourced_high is not None + and r.sourced_mid is not None + ): + typer.echo( + f"\n cross-check vs {concept}: sourced range " + f"{usd(r.sourced_low)} - {usd(r.sourced_high)} (mid {usd(r.sourced_mid)})" + ) + verdict_label = { + "within": "AGREES - implied TAM lands inside the sourced range", + "above": "implied TAM is ABOVE the sourced range", + "below": "implied TAM is BELOW the sourced range", + }[r.verdict] + ratio = f" ({r.vs_mid_ratio:.1f}x the sourced midpoint)" if r.vs_mid_ratio else "" + typer.echo(f" verdict: {verdict_label}{ratio}") + typer.echo("\n Assumptions (the implied TAM holds only if):") + for a in r.assumptions: + typer.echo(f" - {a}") + if r.warnings: + typer.echo("") + for w in r.warnings: + typer.echo(f" WARN {w}") + + db_app = typer.Typer(name="db", help="Database management commands.", no_args_is_help=True) app.add_typer(db_app) diff --git a/backend/src/strata/connectors/edgar.py b/backend/src/strata/connectors/edgar.py index 294327c..ca99688 100644 --- a/backend/src/strata/connectors/edgar.py +++ b/backend/src/strata/connectors/edgar.py @@ -12,7 +12,9 @@ import hashlib import json +import re import time +from collections.abc import Iterable from dataclasses import dataclass from datetime import date from enum import StrEnum @@ -159,6 +161,11 @@ class Fact: frame: str fiscal_year: int | None = None fiscal_period: str | None = None + # The fact's own reporting period. fy/fp describe the FILING, so a FY2025 + # 10-K stamps fy=2025 on its FY2023/24/25 revenue comparatives — only + # start/end identify which year a comparative actually covers. + start: str = "" + end: str = "" @dataclass(frozen=True, slots=True) @@ -172,6 +179,62 @@ class CompanyConcept: units: dict[str, tuple[Fact, ...]] +# ── Period-aware annual fact selection ────────────────────────────── +# Shared with strata.sizing.revenue_lookup. `_pick_fact` (below, used by the +# YAML converters) still uses the older fy/frame heuristic — kept unchanged in +# this pass; only callers that opt into these helpers get period-aware selection. + +# An annual XBRL frame is "CYnnnn" with no quarter suffix (e.g. CY2024); a +# quarterly frame carries a Q (CY2024Q1I). Annual is what a TAM cares about. +_ANNUAL_FRAME = re.compile(r"^CY\d{4}$") +# A fiscal year is ~365 days; this window admits 52/53-week fiscal calendars +# while excluding quarterly and multi-year durations. +_ANNUAL_DAYS = (330, 400) + + +def _is_annual(fact: Fact) -> bool: + """Annual-duration test. When the fact carries its own period (start/end), + trust THAT — fy/fp/form describe the filing, and a 10-K holds quarterly + comparatives too. Frame/form is only a fallback for period-less facts.""" + if fact.start and fact.end: + try: + days = (date.fromisoformat(fact.end) - date.fromisoformat(fact.start)).days + except ValueError: + return bool(fact.frame and _ANNUAL_FRAME.match(fact.frame)) + return _ANNUAL_DAYS[0] <= days <= _ANNUAL_DAYS[1] + return bool((fact.frame and _ANNUAL_FRAME.match(fact.frame)) or fact.form == "10-K") + + +def fiscal_year_from_end(fact: Fact) -> int | None: + """Displayed fiscal year comes from the fact's own period end — never from + ``fy``, which is the filing's year and mislabels comparatives.""" + if fact.end: + try: + return date.fromisoformat(fact.end).year + except ValueError: + return None + return None + + +def fact_recency(fact: Fact) -> tuple[str, str]: + """Sort key: latest period end wins; filed date breaks ties (an amended + re-filing of the same period supersedes the original). ISO date strings + compare correctly as text; a missing end sorts before any real date.""" + return (fact.end, fact.filed) + + +def latest_annual_fact(facts: Iterable[Fact]) -> Fact | None: + """The most recent annual fact, selected by the fact's own period end. A + FY2025 10-K carries FY2023/24/25 comparatives all stamped fy=2025 with one + filed date, so fy/filed cannot distinguish them — only ``end`` can.""" + usable = [f for f in facts if f.value] + if not usable: + return None + annual = [f for f in usable if _is_annual(f)] + pool = annual or usable + return max(pool, key=fact_recency) + + @dataclass(frozen=True, slots=True) class CompanyInfo: cik: str @@ -249,8 +312,12 @@ def _parse_fact(raw: dict[str, Any]) -> Fact: form=str(raw.get("form", "")), filed=str(raw.get("filed", "")), frame=str(raw.get("frame", "")), - fiscal_year=int(raw["fy"]) if "fy" in raw else None, - fiscal_period=str(raw["fp"]) if "fp" in raw else None, + # SEC XBRL facts often carry "fy"/"fp" present-but-null; guard on the + # value, not just key presence, or int(None)/str(None) corrupts the Fact. + fiscal_year=int(raw["fy"]) if raw.get("fy") is not None else None, + fiscal_period=str(raw["fp"]) if raw.get("fp") is not None else None, + start=str(raw.get("start", "") or ""), + end=str(raw.get("end", "") or ""), ) @@ -429,6 +496,9 @@ def __init__( ) default_cache = Path.home() / ".strata" / "cache" / "edgar" self._cache = _DiskCache(cache_dir or default_cache, cache_ttl) + # ticker → CIK map, parsed once per client so resolve_ticker is a dict + # hit after the first call (the SEC list is ~12k rows re-scanned otherwise). + self._cik_by_ticker: dict[str, str] | None = None def close(self) -> None: self._http.close() @@ -523,41 +593,71 @@ def get_company_facts(self, cik: str | int) -> EDGARResponse: facts_raw: dict[str, Any] = raw.get("facts", {}) + cik_str = str(raw.get("cik", "")) + entity_name = str(raw.get("entityName", "")) all_units: dict[str, list[Fact]] = {} - for taxonomy_data in facts_raw.values(): + # Per-tag accumulation alongside the merged view, so one companyfacts + # request can answer per-concept questions (e.g. the revenue lookup) + # without N companyconcept round-trips. + per_tag_units: dict[str, dict[str, list[Fact]]] = {} + per_tag_meta: dict[str, tuple[str, str, str]] = {} # tag → (taxonomy, label, description) + for taxonomy_name in facts_raw: + taxonomy_data: object = facts_raw[taxonomy_name] if not isinstance(taxonomy_data, dict): continue taxonomy_dict: dict[str, Any] = cast(dict[str, Any], taxonomy_data) - for _concept_tag in taxonomy_dict: - concept_data: object = taxonomy_dict[_concept_tag] + for concept_tag in taxonomy_dict: + concept_data: object = taxonomy_dict[concept_tag] if not isinstance(concept_data, dict): continue concept_dict: dict[str, Any] = cast(dict[str, Any], concept_data) units_block: dict[str, Any] = concept_dict.get("units", {}) + per_tag_meta.setdefault( + concept_tag, + ( + str(taxonomy_name), + str(concept_dict.get("label") or ""), + str(concept_dict.get("description") or ""), + ), + ) + tag_units = per_tag_units.setdefault(concept_tag, {}) for unit_label in units_block: fact_list: object = units_block[unit_label] if not isinstance(fact_list, list): continue typed_list: list[dict[str, Any]] = cast(list[dict[str, Any]], fact_list) parsed = [_parse_fact(f) for f in typed_list] - if unit_label in all_units: - all_units[unit_label].extend(parsed) - else: - all_units[unit_label] = parsed + all_units.setdefault(unit_label, []).extend(parsed) + # A tag colliding across taxonomies merges its facts under + # one key (none collide in practice; us-gaap vs dei disjoint). + tag_units.setdefault(unit_label, []).extend(parsed) concept = CompanyConcept( - cik=str(raw.get("cik", "")), - entity_name=str(raw.get("entityName", "")), + cik=cik_str, + entity_name=entity_name, taxonomy="all", tag="ALL", label="All XBRL facts", description="Aggregated company facts across all taxonomies", units={k: tuple(v) for k, v in all_units.items()}, ) + concepts = { + tag: CompanyConcept( + cik=cik_str, + entity_name=entity_name, + taxonomy=per_tag_meta[tag][0], + tag=tag, + label=per_tag_meta[tag][1], + description=per_tag_meta[tag][2], + units={k: tuple(v) for k, v in units.items()}, + ) + for tag, units in per_tag_units.items() + } return EDGARResponse( kind="companyfacts", concept=concept, + concepts=concepts, cached=was_cached, cache_age_seconds=age, ) @@ -800,12 +900,15 @@ def get_company_tickers(self) -> EDGARResponse: def resolve_ticker(self, ticker: str) -> str | None: """Map a ticker symbol to its zero-padded CIK. Returns None if not found.""" - resp = self.get_company_tickers() - target = ticker.upper() - for cik_int, tick, _name in resp.tickers: - if tick.upper() == target: - return str(cik_int).zfill(10) - return None + if self._cik_by_ticker is None: + resp = self.get_company_tickers() + mapping: dict[str, str] = {} + for cik_int, tick, _name in resp.tickers: + # setdefault keeps the FIRST entry per symbol — same winner as + # the previous linear scan when the SEC list carries duplicates. + mapping.setdefault(tick.upper(), str(cik_int).zfill(10)) + self._cik_by_ticker = mapping + return self._cik_by_ticker.get(ticker.upper()) # ── Cache management ───────────────────────────────────────── diff --git a/backend/src/strata/mcp/schemas.py b/backend/src/strata/mcp/schemas.py index 0911aac..d5501c8 100644 --- a/backend/src/strata/mcp/schemas.py +++ b/backend/src/strata/mcp/schemas.py @@ -542,6 +542,58 @@ "additionalProperties": False, } +SENSE_CHECK: dict[str, Any] = { + "type": "object", + "properties": { + "players": { + "type": "array", + "minItems": 1, + "description": ( + "The top players in the market. Each needs a name, a market_share " + "(fraction 0.15 or percent 15), and EITHER a numeric revenue (analyst " + "estimate, offline) OR an edgar_ticker (sourced TOTAL-company revenue " + "from SEC EDGAR — live, gated by STRATA_MCP_LIVE_CONNECTORS=1)." + ), + "items": { + "type": "object", + "properties": { + "name": {"type": "string", "description": "Player / company name."}, + "revenue": { + "type": "number", + "description": ( + "The player's annual revenue IN THIS MARKET (analyst " + "estimate). Mutually exclusive with edgar_ticker." + ), + }, + "edgar_ticker": { + "type": "string", + "description": ( + "US-listed ticker to pull a SOURCED (total-company) revenue " + "from SEC EDGAR. Live call — requires " + "STRATA_MCP_LIVE_CONNECTORS=1. Mutually exclusive with revenue." + ), + }, + "market_share": { + "type": "number", + "description": "Estimated revenue share: fraction (0.15) or percent (15).", + }, + }, + "required": ["name", "market_share"], + "additionalProperties": False, + }, + }, + "concept_slug": { + "type": "string", + "description": ( + "Optional curated concept slug to cross-check the implied TAM against " + "its sourced scenario range (verdict: within / above / below)." + ), + }, + }, + "required": ["players"], + "additionalProperties": False, +} + # ── Output schemas (Phase 0) ──────────────────────────────────────────────── # Advertised in tools/list so a host gets typed structuredContent. Permissive by # design (the payloads are rich + evolving): object with additionalProperties so @@ -622,4 +674,26 @@ "readable_at": {"type": "string"}, }, }, + "sense_check": { + "type": "object", + "additionalProperties": True, + "properties": { + # modeled is ALWAYS true (Q20): shares are analyst estimates. + "modeled": {"type": "boolean", "const": True}, + "modeled_label": {"type": "string"}, + "implied_tam": {"type": "number"}, + "implied_low": {"type": ["number", "null"]}, + "implied_high": {"type": ["number", "null"]}, + "revenue_floor": {"type": "number"}, + "combined_share": {"type": "number"}, + "leverage": {"type": ["number", "null"]}, + "residual": {"type": ["number", "null"]}, + "spread_ratio": {"type": "number"}, + "per_player": {"type": "array", "items": {"type": "object"}}, + "warnings": {"type": "array", "items": {"type": "string"}}, + "assumptions": {"type": "array", "items": {"type": "string"}}, + "sourced_range": {"type": ["object", "null"]}, + "verdict": {"type": ["string", "null"]}, + }, + }, } diff --git a/backend/src/strata/mcp/sense_check_tool.py b/backend/src/strata/mcp/sense_check_tool.py new file mode 100644 index 0000000..ddebaff --- /dev/null +++ b/backend/src/strata/mcp/sense_check_tool.py @@ -0,0 +1,150 @@ +"""MCP `sense_check` tool — the market-share-implied TAM cross-check (agent parity +with `strata sense-check`). + +The payload mirrors the CLI result structurally (implied TAM, band, floor, +leverage, residual, per-player rows, warnings, assumptions) and ALWAYS carries +the machine-readable MODELED marker: market shares are analyst estimates, so the +implied TAM is a modeled cross-check even when every revenue is a sourced filing +(Q20) — it never becomes a sourced number. + +`edgar_ticker` resolution is a live SEC call, gated behind the same +``STRATA_MCP_LIVE_CONNECTORS=1`` env gate as ``get_connector_data`` so the +default MCP surface stays offline-safe. +""" + +from __future__ import annotations + +import os +from typing import Any + +from ..connectors.edgar import EDGARClient +from ..evaluator.engine import EvaluatorError +from ..registry import Registry, RegistryError +from ..sizing.revenue_lookup import RevenueLookupError, resolve_revenue, to_player +from ..sizing.sense_check import ( + MODELED_LABEL, + Player, + SenseCheckError, + normalize_share, +) +from ..sizing.sense_check import ( + sense_check as run_sense_check, +) +from ..sizing.service import sourced_scenarios +from .errors import ( + CODE_CONCEPT_NOT_FOUND, + CODE_EVALUATOR_ERROR, + CODE_INVALID_INPUT, + CODE_LIVE_DISABLED, + ToolError, +) + + +def _player_from_row(row: dict[str, Any], live_enabled: bool, client: EDGARClient) -> Player: + name = str(row.get("name") or "").strip() + if not name: + raise ToolError("each player needs a non-empty 'name'", code=CODE_INVALID_INPUT) + raw_share = row.get("market_share") + if not isinstance(raw_share, (int, float)) or isinstance(raw_share, bool): + raise ToolError( + f"player {name!r}: 'market_share' must be a number (fraction 0.15 or percent 15)", + code=CODE_INVALID_INPUT, + ) + share = normalize_share(float(raw_share)) + + ticker = str(row.get("edgar_ticker") or "").strip() + revenue = row.get("revenue") + if ticker and revenue is not None: + raise ToolError( + f"player {name!r}: provide 'revenue' OR 'edgar_ticker', not both", + code=CODE_INVALID_INPUT, + ) + if ticker: + if not live_enabled: + raise ToolError( + f"player {name!r}: resolving 'edgar_ticker' is a live SEC EDGAR call, " + "which is disabled; set STRATA_MCP_LIVE_CONNECTORS=1 on the server, or " + "pass a numeric 'revenue' instead (offline).", + code=CODE_LIVE_DISABLED, + remediation="set STRATA_MCP_LIVE_CONNECTORS=1 or supply revenue directly", + ) + try: + rr = resolve_revenue(ticker, client=client) + except RevenueLookupError as exc: + raise ToolError(str(exc), code=CODE_INVALID_INPUT, retriable=True) from exc + return to_player(name, share, rr) + if not isinstance(revenue, (int, float)) or isinstance(revenue, bool): + raise ToolError( + f"player {name!r}: provide a numeric 'revenue' or an 'edgar_ticker'", + code=CODE_INVALID_INPUT, + ) + return Player(name=name, revenue=float(revenue), market_share=share) + + +def sense_check_tool( + reg: Registry, *, players: list[dict[str, Any]], concept_slug: str | None = None +) -> dict[str, Any]: + live_enabled = os.getenv("STRATA_MCP_LIVE_CONNECTORS") == "1" + # Construction is cheap and no-network; one client serves every edgar: row. + with EDGARClient() as client: + parsed = [_player_from_row(row, live_enabled, client) for row in players] + + scenarios: tuple[float, ...] | None = None + if concept_slug: + try: + scenarios = sourced_scenarios(reg, concept_slug) + except RegistryError as exc: + raise ToolError( + f"concept {concept_slug!r} not found", + code=CODE_CONCEPT_NOT_FOUND, + remediation="search_concepts to find a matching slug", + ) from exc + except EvaluatorError as exc: + raise ToolError(str(exc), code=CODE_EVALUATOR_ERROR) from exc + + try: + r = run_sense_check(parsed, sourced_scenarios=scenarios) + except SenseCheckError as exc: + raise ToolError(str(exc), code=CODE_INVALID_INPUT) from exc + + sourced_range = None + if r.sourced_low is not None and r.sourced_high is not None: + sourced_range = {"low": r.sourced_low, "mid": r.sourced_mid, "high": r.sourced_high} + return { + # Unconditional Q20 marker — machine-readable, never gated on sourcing. + "modeled": True, + "modeled_label": MODELED_LABEL, + "implied_tam": r.implied_tam, + # Band/leverage/residual are null when Σshare ≥ 100% (a band that would + # exclude its own point estimate is suppressed, never emitted). + "implied_low": r.implied_low, + "implied_high": r.implied_high, + "share_tolerance": r.share_tolerance, + "revenue_floor": r.revenue_floor, + "combined_share": r.combined_share, + "leverage": r.leverage, + "residual": r.residual, + "spread_ratio": r.spread_ratio, + "n_players": r.n_players, + "n_sourced": r.n_sourced, + "all_revenue_sourced": r.all_revenue_sourced, + "per_player": [ + { + "name": p.name, + "revenue": p.revenue, + "market_share": p.market_share, + "implied_tam": p.implied_tam, + "revenue_sourced": p.revenue_sourced, + "revenue_note": p.revenue_note or None, + } + for p in r.per_player + ], + # The total-company caveat now arrives as an engine warning (set by the + # typed marker), so no surface re-derives it from prose. + "warnings": list(r.warnings), + "assumptions": list(r.assumptions), + "concept": concept_slug, + "sourced_range": sourced_range, + "vs_mid_ratio": r.vs_mid_ratio, + "verdict": r.verdict, + } diff --git a/backend/src/strata/mcp/tools.py b/backend/src/strata/mcp/tools.py index 02c0e08..3e92c9e 100644 --- a/backend/src/strata/mcp/tools.py +++ b/backend/src/strata/mcp/tools.py @@ -948,6 +948,16 @@ def _h_record_session_reasoning(reg: Registry, args: dict[str, Any]) -> Awaitabl return record_session_reasoning(reg, payload=dict(args.get("payload") or {})) +def _h_sense_check(reg: Registry, args: dict[str, Any]) -> dict[str, Any]: + from .sense_check_tool import sense_check_tool + + return sense_check_tool( + reg, + players=list(args.get("players", [])), + concept_slug=args.get("concept_slug"), + ) + + def _h_list_sourcing_gaps(reg: Registry, args: dict[str, Any]) -> Awaitable[dict[str, Any]]: from .sourcing_gap_tools import list_sourcing_gaps @@ -1223,6 +1233,21 @@ def _h_list_sourcing_gaps(reg: Registry, args: dict[str, Any]) -> Awaitable[dict "handler": _h_list_sourcing_gaps, "annotations": _READ, }, + "sense_check": { + "description": ( + "[READ] Market-share-implied TAM cross-check: Σ(top-player revenue) / " + "Σ(market share), with sensitivity band, revenue floor, leverage, residual, " + "per-player consistency, and explicit assumptions. ALWAYS MODELED (Q20) — " + "shares are analyst estimates, so the payload carries modeled=true and is " + "never a sourced number. revenue is a number, or edgar_ticker pulls a " + "sourced TOTAL-company revenue from SEC EDGAR (live, gated by " + "STRATA_MCP_LIVE_CONNECTORS=1). Optional concept_slug cross-checks the " + "implied TAM against the curated market's sourced scenario range." + ), + "inputSchema": schemas.SENSE_CHECK, + "handler": _h_sense_check, + "annotations": _READ_NET, + }, "record_session_reasoning": { "description": ( "[WRITE] Attach your reasoning / chain-of-thought / plan for THIS session " diff --git a/backend/src/strata/sizing/revenue_lookup.py b/backend/src/strata/sizing/revenue_lookup.py new file mode 100644 index 0000000..a05d6b5 --- /dev/null +++ b/backend/src/strata/sizing/revenue_lookup.py @@ -0,0 +1,128 @@ +"""EDGAR-backed revenue lookup for the sense-check (public-company revenue). + +Resolves a US-listed company's latest annual revenue from SEC EDGAR XBRL +(regulator-tier, redistributable) so the sense-check can use a SOURCED revenue +figure instead of an analyst estimate — closing the "private revenue is itself a +guess" critique for any public player. + +Scope, stated honestly: +* **EDGAR** is the revenue source: free, regulator-tier, US-listed companies only. +* The figure is **TOTAL company revenue**, not in-market revenue — the caller must + confirm they coincide or substitute a segment figure (the CLI/MCP surfaces say so). +* **yfinance** (the other already-wired finance connector) exposes *quotes*, not + revenue, so it is not used here. +* **Private companies** have no public filing — they stay analyst estimates, or + come from a session-side licensed source (e.g. a Capital IQ / S&P MCP) that is + NEVER committed to the open bank (licensing + tier). +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from ..connectors.edgar import ( + EDGARClient, + EDGARClientError, + Fact, + fact_recency, + fiscal_year_from_end, + latest_annual_fact, +) +from .sense_check import REVENUE_NOTE_FILING_TOTAL_CO, Player + +# us-gaap revenue concepts, in preference order (older filers use Revenues; newer +# ASC 606 filers report the RevenueFromContractWithCustomer tags; some report +# the older SalesRevenue tags). +_REVENUE_CONCEPTS = ( + "Revenues", + "RevenueFromContractWithCustomerExcludingAssessedTax", + "RevenueFromContractWithCustomerIncludingAssessedTax", + "SalesRevenueNet", + "SalesRevenueServicesNet", +) + + +class RevenueLookupError(RuntimeError): + """Ticker not found, or no usable annual revenue concept on file.""" + + +@dataclass(frozen=True, slots=True) +class ResolvedRevenue: + ticker: str + cik: str + entity_name: str + revenue: float + fiscal_year: int | None + concept: str + source_url: str + + +def to_player(name: str, share: float, resolved: ResolvedRevenue) -> Player: + """The ONE way an EDGAR figure becomes a sense-check player: always sourced, + always carrying the total-company marker so the engine emits the caveat — + a sourced-but-total figure is never an unmarked sourced line.""" + return Player( + name=name, + revenue=resolved.revenue, + market_share=share, + revenue_sourced=True, + revenue_note=REVENUE_NOTE_FILING_TOTAL_CO, + ) + + +def resolve_revenue(ticker: str, *, client: EDGARClient | None = None) -> ResolvedRevenue: + """Latest annual revenue for a US-listed ``ticker`` from SEC EDGAR XBRL. + + The figure is the company's TOTAL revenue (whole company, not a market + segment) — the caller owns confirming it equals in-market revenue. + + Makes live HTTP calls (caller decides when — never on an offline path). Pass + ``client`` to inject a configured/fake client (tests do this to stay offline). + Raises :class:`RevenueLookupError` on a miss — never returns a fabricated value. + """ + own = client is None + c = client or EDGARClient() + try: + cik = c.resolve_ticker(ticker) + if cik is None: + raise RevenueLookupError( + f"ticker {ticker!r} not found in the SEC company list (US-listed filers only)" + ) + tried = ", ".join(_REVENUE_CONCEPTS) + miss_msg = ( + f"no usable annual revenue concept on file for {ticker!r} (CIK {cik}); " + f"tried us-gaap tags: {tried}" + ) + # ONE companyfacts request replaces up to 5 per-tag companyconcept calls. + try: + resp = c.get_company_facts(cik) + except EDGARClientError as exc: + raise RevenueLookupError(miss_msg) from exc + by_tag = resp.concepts or {} + # Gather the latest annual fact from EVERY revenue concept, then pick the + # globally most-recent. Filers migrate tags over time (older "Revenues" → + # ASC 606 "RevenueFromContract…"), so stopping at the first non-empty tag + # can return a stale year — take the newest across all of them. + candidates: list[tuple[Fact, str, str]] = [] # (fact, concept_tag, entity_name) + for concept_tag in _REVENUE_CONCEPTS: + concept = by_tag.get(concept_tag) + if concept is None: + continue + fact = latest_annual_fact(concept.units.get("USD", ())) + if fact is not None: + candidates.append((fact, concept_tag, concept.entity_name)) + if not candidates: + raise RevenueLookupError(miss_msg) + fact, concept_tag, entity_name = max(candidates, key=lambda c: fact_recency(c[0])) + return ResolvedRevenue( + ticker=ticker.upper(), + cik=cik, + entity_name=entity_name, + revenue=fact.value, + fiscal_year=fiscal_year_from_end(fact), + concept=concept_tag, + source_url=f"https://www.sec.gov/cgi-bin/browse-edgar?action=getcompany&CIK={cik}", + ) + finally: + if own: + c.close() diff --git a/backend/src/strata/sizing/sense_check.py b/backend/src/strata/sizing/sense_check.py new file mode 100644 index 0000000..b417e40 --- /dev/null +++ b/backend/src/strata/sizing/sense_check.py @@ -0,0 +1,342 @@ +"""Market-share-implied TAM sense-check (the comparable-revenue cross-check). + +Given the top players in a market — each with a revenue and an estimated market +share — derive an implied TAM and (optionally) cross-check it against a sourced +bottoms-up TAM. This is the ``COMPARABLE_REVENUE`` method run as an analyst +sense-check rather than a curated framework: + + implied_tam = Σ player_revenue / Σ player_market_share (the headline) + revenue_floor = Σ player_revenue (a TAM can't be below the revenue + already booked inside it) + per-player = revenue_i / share_i (each should agree; a wide spread + means the share estimates are + internally inconsistent) + residual = implied_tam - revenue_floor (revenue the estimate attributes + to everyone outside the named set) + +It is built to be **defensible**, which means it surfaces its own uncertainty and +assumptions instead of emitting one confident number (see +``docs/methods/sense-check-defensibility.md``): + +* a **sensitivity band** — recompute the implied TAM with shares ±``share_tolerance`` + so the output is a range, the way a TAM expert would sanity-check it; +* a **leverage factor** (``1 / Σshare``) — a small combined share multiplies every + error, so the check warns when the estimate is fragile; +* the load-bearing **assumptions** the caller renders, so the reasoning is explicit. + +**Honesty (Q20).** Market shares are analyst *estimates*, so the implied TAM is a +MODELED cross-check — never written back as a sourced primitive. It exists to +agree or disagree with the sourced number, surfacing a spread the same way the +evaluator's ``convergence`` does for two genuinely sourced method families. +""" + +from __future__ import annotations + +import statistics +from dataclasses import dataclass, field + +# Default agreement band: estimates within this multiple of each other "agree"; +# beyond it, the check flags a disagreement (the whole point of a sense-check). +DEFAULT_AGREE_BAND = 2.0 +# Default relative tolerance applied to shares for the sensitivity band. +DEFAULT_SHARE_TOLERANCE = 0.25 +# Combined share ≤ 25% makes the 1/Σshare multiplier large → fragile estimate +# (the threshold docs/methods/sense-check-defensibility.md states; keep in sync). +LEVERAGE_WARN_SHARE = 0.25 + +# Tokens an optional 4th --player field / CSV `sourced` column may use to flag +# a filing-backed revenue — one list shared by every parsing surface. +SOURCED_TOKENS = ("sourced", "filing", "true", "1", "yes") + +# The unconditional Q20 label every surface renders/ships, machine-readably: +# shares are ALWAYS analyst estimates, so the implied TAM stays modeled even +# when every revenue is a sourced filing. +MODELED_LABEL = ( + "MODELED (Q20): market shares are analyst estimates, so the implied TAM is a " + "modeled cross-check — never a sourced number, never persisted" +) + +# Typed marker for an EDGAR-resolved revenue: a whole-company filing figure, +# not an in-market one. Lives here (not revenue_lookup) so the engine can emit +# the caveat without importing the lookup module (which imports Player from us). +REVENUE_NOTE_FILING_TOTAL_CO = "filing, total-co" +# The caveat sentence, stated ONCE — sense_check() emits it as a per-player +# warning whenever the marker is set, so every surface renders it uniformly. +EDGAR_TOTAL_CO_CAVEAT = ( + "EDGAR figure is TOTAL company revenue — confirm it equals in-market revenue " + "or use a segment figure" +) + +# The load-bearing assumptions the implied TAM rests on. Rendered with every +# result so a reviewer sees exactly what must hold — these are the conditions a +# TAM expert checks first, made explicit rather than buried. +ASSUMPTIONS: tuple[str, ...] = ( + "each revenue is the player's revenue IN THIS MARKET (not total company revenue)", + "market shares are revenue shares OF THIS SAME MARKET (not seats / logos / mindshare)", + "all revenues share one currency, period (e.g. TTM), and geography as the market", + "the market is revenue-defined today (not a pre-revenue / latent market)", + "the named players don't double-count each other's revenue (no channel pass-through)", +) + + +class SenseCheckError(ValueError): + """Bad sense-check input (no players, non-positive revenue, share out of range, + or an unparseable ``--player`` / money / share token).""" + + +@dataclass(frozen=True, slots=True) +class Player: + name: str + revenue: float # currency/year, in the market's unit + market_share: float # fraction in (0, 1] + revenue_sourced: bool = False # True only if revenue is a sourced filing (EDGAR/regulator) + # Display tag for the revenue's basis, e.g. "filing, total-co" for an EDGAR + # figure (whole-company revenue, not in-market) — rendered next to the line + # so a sourced-but-total figure is never an unmarked sourced line. + revenue_note: str = "" + + +@dataclass(frozen=True, slots=True) +class PlayerImplied: + name: str + revenue: float + market_share: float + implied_tam: float # revenue / market_share + revenue_sourced: bool = False + revenue_note: str = "" + + +@dataclass(frozen=True, slots=True) +class SenseCheckResult: + implied_tam: float # combined Σrevenue / Σshare + # Band/leverage/residual are None when Σshare ≥ 100%: a band derived from + # overlapping shares would exclude its own point estimate, leverage < 1 is + # meaningless, and the residual goes negative — suppressed, never emitted. + implied_low: float | None # at shares scaled up by share_tolerance (more share → less TAM) + implied_high: float | None # at shares scaled down by share_tolerance (less share → more TAM) + share_tolerance: float + revenue_floor: float # Σrevenue + combined_share: float # Σshare + leverage: float | None # 1 / Σshare — how much the estimate multiplies the captured revenue + residual: float | None # implied_tam - revenue_floor (revenue attributed outside the set) + per_player: tuple[PlayerImplied, ...] + spread_ratio: float # max per-player implied / min (1.0 == perfect agreement) + n_sourced: int # how many revenues are sourced filings + warnings: tuple[str, ...] = () + # Cross-check against a curated concept's sourced scenario range (when given). + sourced_low: float | None = None + sourced_high: float | None = None + sourced_mid: float | None = None + vs_mid_ratio: float | None = None # implied_tam / sourced_mid + verdict: str | None = None # within | above | below | None (no sourced market) + assumptions: tuple[str, ...] = field(default=ASSUMPTIONS) + + # Derivable state stays derived — per_player is the single source of truth. + @property + def n_players(self) -> int: + return len(self.per_player) + + @property + def all_revenue_sourced(self) -> bool: + return self.n_sourced == self.n_players + + +def normalize_share(raw: float) -> float: + """Accept a share as a fraction (``0.15``) or a percent (``15``). A value > 1 + is read as a percent, so ``15`` → ``0.15``. (A genuine share never exceeds 1, + so this is unambiguous in practice.)""" + return raw / 100.0 if raw > 1.0 else raw + + +def share_token_warning(token: str, share: float) -> str | None: + """The ``1`` trap: a bare integer token parses to a 100% share (fractions ≤ 1 + pass through :func:`normalize_share`), but a user may have meant 1%. Warn when + a token without a decimal point or percent sign lands on exactly 100%.""" + t = token.strip() + if share == 1.0 and "." not in t and "%" not in t: + return ( + f"share token {token!r} parsed as 100% — if you meant 1%, " + f"write '0.01' or '1.0' for an unambiguous fraction" + ) + return None + + +def parse_money(raw: str) -> float: + """Parse a money token: ``31e9`` / ``31000000000`` / ``3.1b`` / ``950m`` / + ``$1.2B`` / ``12,000``. Suffixes k/m/b/t scale by 1e3/1e6/1e9/1e12.""" + s = raw.strip().lower().replace("$", "").replace(",", "").replace("_", "") + if not s: + raise SenseCheckError(f"empty money token: {raw!r}") + mult = 1.0 + suffixes = {"t": 1e12, "b": 1e9, "m": 1e6, "k": 1e3} + if s[-1] in suffixes: + mult = suffixes[s[-1]] + s = s[:-1] + try: + return float(s) * mult + except ValueError as exc: + raise SenseCheckError(f"could not parse money value {raw!r}") from exc + + +def split_player_spec(spec: str) -> list[str]: + """Split a ``--player`` spec into its fields. The field-count and empty-name + guards live HERE only, so every spec branch (plain / edgar) fails the same way.""" + parts = [p.strip() for p in spec.split("|")] + if len(parts) not in (3, 4): + raise SenseCheckError(f"--player must be 'Name|revenue|share[|sourced]', got {spec!r}") + if not parts[0]: + raise SenseCheckError(f"player name is empty in {spec!r}") + return parts + + +def parse_player(spec: str) -> Player: + """Parse a CLI ``--player`` spec ``'Name|revenue|share'`` (optional 4th field + ``sourced`` flags a revenue that comes from a filing). Revenue uses + :func:`parse_money`; share uses :func:`normalize_share`.""" + parts = split_player_spec(spec) + share_s = parts[2] + try: + share = normalize_share(float(share_s)) + except ValueError as exc: + raise SenseCheckError(f"could not parse share {share_s!r} in {spec!r}") from exc + sourced = len(parts) == 4 and parts[3].lower() in SOURCED_TOKENS + return Player( + name=parts[0], revenue=parse_money(parts[1]), market_share=share, revenue_sourced=sourced + ) + + +def sense_check( + players: list[Player], + *, + sourced_scenarios: tuple[float, ...] | None = None, + agree_band: float = DEFAULT_AGREE_BAND, + share_tolerance: float = DEFAULT_SHARE_TOLERANCE, +) -> SenseCheckResult: + """Compute the market-share-implied TAM, its sensitivity band, and (optionally) + compare it to a curated concept's sourced scenario values. Pure and + deterministic — no I/O, no LLM, no persistence.""" + if not players: + raise SenseCheckError("no players provided — add at least one (name, revenue, share)") + if not 0 <= share_tolerance < 1: + raise SenseCheckError(f"share_tolerance must be in [0, 1) (got {share_tolerance})") + + for p in players: + if p.revenue <= 0: + raise SenseCheckError(f"{p.name}: revenue must be > 0 (got {p.revenue})") + if not 0 < p.market_share <= 1: + raise SenseCheckError( + f"{p.name}: market_share must be in (0, 1] (got {p.market_share})" + ) + + warnings: list[str] = [] + total_rev = sum(p.revenue for p in players) + total_share = sum(p.market_share for p in players) + + implied = total_rev / total_share + leverage: float | None + implied_low: float | None + implied_high: float | None + residual: float | None + if total_share >= 1.0: + # Σshare ≥ 100%: capping the upper share at 1.0 would put the band's low + # end ABOVE the point estimate (which itself falls below the revenue + # floor). A band that excludes its own point estimate is never emitted — + # only the floor and the overlap warning survive. + leverage = implied_low = implied_high = residual = None + if total_share > 1.0: + warnings.append( + f"combined market share is {total_share:.0%} (> 100%) — the shares overlap " + f"or are overstated; only the revenue floor ${total_rev:,.0f} is defensible" + ) + else: + leverage = 1.0 / total_share + # Sensitivity band: a share that is x% too low/high moves the TAM the + # other way. Cap the upper share at 1.0 (a combined share can't exceed 100%). + share_hi = min(total_share * (1.0 + share_tolerance), 1.0) + share_lo = total_share * (1.0 - share_tolerance) + implied_low = total_rev / share_hi + implied_high = total_rev / share_lo if share_lo > 0 else float("inf") + residual = implied - total_rev + + if total_share <= LEVERAGE_WARN_SHARE and leverage is not None: + warnings.append( + f"combined share is only {total_share:.0%}, so the implied TAM is {leverage:.0f}x the " + f"captured revenue — it is highly leveraged on the share estimate; widen the player set" + ) + + per = tuple( + PlayerImplied( + p.name, + p.revenue, + p.market_share, + p.revenue / p.market_share, + p.revenue_sourced, + p.revenue_note, + ) + for p in players + ) + implied_vals = [pi.implied_tam for pi in per] + spread = max(implied_vals) / min(implied_vals) + if spread > agree_band: + warnings.append( + f"per-player implied TAMs disagree {spread:.1f}x — the share estimates are " + f"internally inconsistent (each revenue/share should imply the same TAM)" + ) + + n_sourced = sum(1 for p in players if p.revenue_sourced) + all_sourced = n_sourced == len(players) + # The MODELED label is UNCONDITIONAL: market shares are always analyst + # estimates, so the implied TAM stays a modeled cross-check even when every + # revenue is a sourced filing (Q20). + if all_sourced: + warnings.append( + "revenues sourced; shares are analyst estimates — the implied TAM remains " + "MODELED (Q20), a cross-check only, never a sourced number" + ) + else: + warnings.append( + f"{len(players) - n_sourced}/{len(players)} revenues are estimates, not sourced " + f"filings — the implied TAM is MODELED, a cross-check only, never a sourced number (Q20)" + ) + # The engine owns the total-company caveat so every surface (CLI, MCP) + # renders it the same way instead of re-deriving it from the note string. + warnings.extend( + f"{p.name}: {EDGAR_TOTAL_CO_CAVEAT}" + for p in players + if p.revenue_note == REVENUE_NOTE_FILING_TOTAL_CO + ) + + sourced_low = sourced_high = sourced_mid = vs_mid = None + verdict: str | None = None + if sourced_scenarios: + sourced_low = min(sourced_scenarios) + sourced_high = max(sourced_scenarios) + sourced_mid = statistics.median(sourced_scenarios) + if sourced_mid > 0: + vs_mid = implied / sourced_mid + if implied < sourced_low: + verdict = "below" + elif implied > sourced_high: + verdict = "above" + else: + verdict = "within" + + return SenseCheckResult( + implied_tam=implied, + implied_low=implied_low, + implied_high=implied_high, + share_tolerance=share_tolerance, + revenue_floor=total_rev, + combined_share=total_share, + leverage=leverage, + residual=residual, + per_player=per, + spread_ratio=spread, + n_sourced=n_sourced, + warnings=tuple(warnings), + sourced_low=sourced_low, + sourced_high=sourced_high, + sourced_mid=sourced_mid, + vs_mid_ratio=vs_mid, + verdict=verdict, + ) diff --git a/backend/src/strata/sizing/service.py b/backend/src/strata/sizing/service.py index 19cf3b8..93f8b32 100644 --- a/backend/src/strata/sizing/service.py +++ b/backend/src/strata/sizing/service.py @@ -66,6 +66,15 @@ def _offline() -> bool: return os.environ.get("STRATA_OFFLINE") == "1" +def sourced_scenarios(reg: Registry, slug: str) -> tuple[float, ...]: + """A curated concept's sourced scenario values — the comparison side of the + sense-check cross-check, shared by the CLI and the MCP tool. Validates the + slug via ``get_concept`` (RegistryError) before sizing (EvaluatorError) so + each surface can map "unknown slug" and "unsizable" to its own error.""" + reg.get_concept(slug) + return tuple(s.value for s in size_concept(reg, slug).scenarios) + + def _tam_midpoint(tam: TAMOutput) -> float: """Representative TAM value = median of the satisfiable scenario values (no new number invented — pure synthesis of sourced scenarios).""" diff --git a/backend/tests/mcp/test_sense_check_tool.py b/backend/tests/mcp/test_sense_check_tool.py new file mode 100644 index 0000000..42b0719 --- /dev/null +++ b/backend/tests/mcp/test_sense_check_tool.py @@ -0,0 +1,160 @@ +"""MCP `sense_check` tool tests — agent parity with `strata sense-check`. + +All pure / offline: edgar_ticker resolution is env-gated and the live path is +monkeypatched. The payload must ALWAYS carry the machine-readable MODELED (Q20) +marker, mirror the CLI result structurally, and suppress the band when shares +overlap past 100%. +""" + +from __future__ import annotations + +import pytest + +from strata.mcp.errors import CODE_INVALID_INPUT, CODE_LIVE_DISABLED, ToolError +from strata.mcp.sense_check_tool import sense_check_tool +from strata.mcp.server import load_registry +from strata.mcp.tools import TOOLS, call_tool + + +@pytest.fixture +def reg(): + return load_registry() + + +def test_tool_is_registered() -> None: + spec = TOOLS["sense_check"] + assert spec["inputSchema"]["required"] == ["players"] + assert "MODELED" in spec["description"] + + +def test_basic_payload_shape_and_unconditional_modeled(reg) -> None: + out = sense_check_tool( + reg, + players=[ + {"name": "Gong", "revenue": 300e6, "market_share": 0.15}, + {"name": "Clari", "revenue": 200e6, "market_share": 0.10}, + ], + ) + assert out["modeled"] is True # machine-readable Q20 marker + assert "MODELED" in out["modeled_label"] + assert out["implied_tam"] == pytest.approx(2e9) + assert out["revenue_floor"] == pytest.approx(500e6) + assert out["implied_low"] is not None and out["implied_high"] is not None + assert out["leverage"] == pytest.approx(4.0) + assert out["residual"] == pytest.approx(1.5e9) + assert len(out["per_player"]) == 2 + assert out["assumptions"] + assert any("MODELED" in w for w in out["warnings"]) + + +def test_all_sourced_payload_still_modeled(reg, monkeypatch: pytest.MonkeyPatch) -> None: + # Even when every revenue is an EDGAR filing, shares are analyst estimates — + # the modeled marker must not disappear (Q20). + import strata.mcp.sense_check_tool as mod + from strata.sizing.revenue_lookup import ResolvedRevenue + + monkeypatch.setenv("STRATA_MCP_LIVE_CONNECTORS", "1") + + def _fake_resolve(ticker: str, **_kw: object) -> ResolvedRevenue: + return ResolvedRevenue( + ticker=ticker.upper(), + cik="0000000001", + entity_name=f"{ticker} Corp", + revenue=1_000_000_000, + fiscal_year=2024, + concept="Revenues", + source_url="https://www.sec.gov/cgi-bin/browse-edgar?CIK=0000000001", + ) + + monkeypatch.setattr(mod, "resolve_revenue", _fake_resolve) + out = sense_check_tool( + reg, + players=[ + {"name": "Alpha", "edgar_ticker": "AAA", "market_share": 0.10}, + {"name": "Beta", "edgar_ticker": "BBB", "market_share": 0.20}, + ], + ) + assert out["all_revenue_sourced"] is True + assert out["modeled"] is True + assert any("MODELED" in w for w in out["warnings"]) + # EDGAR figures are whole-company: tagged + caveated, never unmarked. + assert all(p["revenue_note"] == "filing, total-co" for p in out["per_player"]) + assert any("TOTAL company revenue" in w for w in out["warnings"]) + + +def test_edgar_ticker_gated_like_live_connectors(reg, monkeypatch: pytest.MonkeyPatch) -> None: + # Same env gate as get_connector_data: no STRATA_MCP_LIVE_CONNECTORS=1, no + # live SEC call — an honest CODE_LIVE_DISABLED error, never a silent fetch. + import strata.mcp.sense_check_tool as mod + + monkeypatch.delenv("STRATA_MCP_LIVE_CONNECTORS", raising=False) + + def _boom(*_a: object, **_k: object) -> object: + raise AssertionError("resolve_revenue must not be called when the live gate is off") + + monkeypatch.setattr(mod, "resolve_revenue", _boom) + with pytest.raises(ToolError) as exc_info: + sense_check_tool( + reg, players=[{"name": "Alpha", "edgar_ticker": "AAA", "market_share": 0.10}] + ) + assert exc_info.value.code == CODE_LIVE_DISABLED + + +def test_overflow_shares_null_band(reg) -> None: + out = sense_check_tool( + reg, + players=[ + {"name": "A", "revenue": 300e6, "market_share": 0.6}, + {"name": "B", "revenue": 200e6, "market_share": 0.6}, + ], + ) + assert out["implied_low"] is None + assert out["implied_high"] is None + assert out["leverage"] is None + assert out["residual"] is None + assert out["revenue_floor"] == pytest.approx(500e6) + assert any("100%" in w for w in out["warnings"]) + + +def test_concept_cross_reference(reg) -> None: + slug = "doc-gen-pc-insurance" # a known sizable curated concept + out = sense_check_tool( + reg, + players=[{"name": "A", "revenue": 2e9, "market_share": 0.5}], + concept_slug=slug, + ) + assert out["concept"] == slug + assert out["sourced_range"] is not None + assert out["verdict"] in {"within", "above", "below"} + + +def test_unknown_concept_is_honest_error(reg) -> None: + with pytest.raises(ToolError): + sense_check_tool( + reg, + players=[{"name": "A", "revenue": 2e9, "market_share": 0.5}], + concept_slug="no-such-concept-zzz", + ) + + +def test_bad_player_rows_raise_invalid_input(reg) -> None: + with pytest.raises(ToolError) as e1: + sense_check_tool(reg, players=[{"name": "A", "market_share": 0.5}]) # no revenue source + assert e1.value.code == CODE_INVALID_INPUT + with pytest.raises(ToolError) as e2: + sense_check_tool( + reg, + players=[{"name": "A", "revenue": 1e9, "edgar_ticker": "AAA", "market_share": 0.5}], + ) + assert e2.value.code == CODE_INVALID_INPUT + + +def test_call_tool_dispatch(reg) -> None: + out = call_tool( + reg, + "sense_check", + {"players": [{"name": "A", "revenue": 4e8, "market_share": 0.2}]}, + ) + assert isinstance(out, dict) + assert out["modeled"] is True + assert out["implied_tam"] == pytest.approx(2e9) diff --git a/backend/tests/sizing/test_revenue_lookup.py b/backend/tests/sizing/test_revenue_lookup.py new file mode 100644 index 0000000..dee65e4 --- /dev/null +++ b/backend/tests/sizing/test_revenue_lookup.py @@ -0,0 +1,281 @@ +"""Tests for EDGAR-backed revenue lookup (mocked client — no network).""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from strata.connectors.edgar import CompanyConcept, EDGARClientError, Fact +from strata.sizing.revenue_lookup import RevenueLookupError, resolve_revenue + + +def _concept(facts: tuple[Fact, ...]) -> CompanyConcept: + return CompanyConcept( + cik="0000320193", + entity_name="Salesforce, Inc.", + taxonomy="us-gaap", + tag="Revenues", + label="Revenues", + description="", + units={"USD": facts}, + ) + + +class _FakeClient: + """Stands in for EDGARClient — no HTTP. ``concept_by_tag`` maps tag -> concept + (the per-tag view one companyfacts request now carries).""" + + def __init__(self, cik: str | None, concept_by_tag: dict[str, CompanyConcept | None]) -> None: + self._cik = cik + self._by_tag = concept_by_tag + self.closed = False + self.facts_calls = 0 + + def resolve_ticker(self, ticker: str) -> str | None: + _ = ticker + return self._cik + + def get_company_facts(self, cik: str) -> SimpleNamespace: + _ = cik + self.facts_calls += 1 + return SimpleNamespace( + concepts={tag: c for tag, c in self._by_tag.items() if c is not None} + ) + + def close(self) -> None: + self.closed = True + + +def test_resolve_picks_latest_annual_revenue() -> None: + facts = ( + Fact( + value=26_000_000_000, + form="10-K", + filed="2022-03-01", + frame="CY2021", + fiscal_year=2021, + start="2021-01-01", + end="2021-12-31", + ), + Fact( + value=34_900_000_000, + form="10-K", + filed="2024-03-01", + frame="CY2023", + fiscal_year=2023, + start="2023-01-01", + end="2023-12-31", + ), + Fact( + value=9_000_000_000, + form="10-Q", + filed="2024-06-01", + frame="CY2024Q1", + fiscal_year=2024, + start="2024-01-01", + end="2024-03-31", + ), + ) + client = _FakeClient("0000320193", {"Revenues": _concept(facts)}) + rr = resolve_revenue("CRM", client=client) + assert rr.revenue == 34_900_000_000 # latest ANNUAL, not the later-period quarterly + assert rr.fiscal_year == 2023 + assert rr.ticker == "CRM" + assert rr.concept == "Revenues" + assert "sec.gov" in rr.source_url + + +def test_comparative_facts_latest_period_end_wins() -> None: + # The regression that mislabeled stale revenue as current: a FY2025 10-K + # carries FY2023/24/25 revenue comparatives ALL stamped fy=2025 with one + # filed date. Only the fact's own period end can tell them apart — the + # latest end must win, and the displayed year must come from end, not fy. + facts = ( + Fact( + value=20_000_000_000, + form="10-K", + filed="2025-03-01", + frame="", + fiscal_year=2025, + start="2022-02-01", + end="2023-01-31", + ), + Fact( + value=25_000_000_000, + form="10-K", + filed="2025-03-01", + frame="", + fiscal_year=2025, + start="2023-02-01", + end="2024-01-31", + ), + Fact( + value=30_000_000_000, + form="10-K", + filed="2025-03-01", + frame="", + fiscal_year=2025, + start="2024-02-01", + end="2025-01-31", + ), + ) + client = _FakeClient("0000320193", {"Revenues": _concept(facts)}) + rr = resolve_revenue("CRM", client=client) + assert rr.revenue == 30_000_000_000 # the latest period end, not the first fy=2025 fact + assert rr.fiscal_year == 2025 # derived from end (2025-01-31), never from fy + + +def test_fiscal_year_label_comes_from_period_end_not_fy() -> None: + # A FY2025-stamped comparative whose own period ends in Jan 2024 is FY2024. + facts = ( + Fact( + value=25_000_000_000, + form="10-K", + filed="2025-03-01", + frame="", + fiscal_year=2025, + start="2023-02-01", + end="2024-01-31", + ), + ) + client = _FakeClient("0000320193", {"Revenues": _concept(facts)}) + rr = resolve_revenue("CRM", client=client) + assert rr.fiscal_year == 2024 + + +def test_zero_or_null_value_facts_never_become_revenue() -> None: + # A null val parses to 0.0 — it must be skipped, not returned as revenue 0. + facts = ( + Fact( + value=34_900_000_000, + form="10-K", + filed="2024-03-01", + frame="", + fiscal_year=2023, + start="2023-02-01", + end="2024-01-31", + ), + Fact( + value=0.0, + form="10-K", + filed="2025-03-01", + frame="", + fiscal_year=2025, + start="2024-02-01", + end="2025-01-31", + ), + ) + client = _FakeClient("0000320193", {"Revenues": _concept(facts)}) + rr = resolve_revenue("CRM", client=client) + assert rr.revenue == 34_900_000_000 + + +def test_null_fy_fact_with_latest_end_still_wins() -> None: + # fy is the filing's year, often null — a null-fy fact must not be demoted. + facts = ( + Fact( + value=26_000_000_000, + form="10-K", + filed="2024-03-01", + frame="", + fiscal_year=2023, + start="2023-01-01", + end="2023-12-31", + ), + Fact( + value=34_900_000_000, + form="10-K", + filed="2025-03-01", + frame="", + fiscal_year=None, + start="2024-01-01", + end="2024-12-31", + ), + ) + client = _FakeClient("0000320193", {"Revenues": _concept(facts)}) + rr = resolve_revenue("CRM", client=client) + assert rr.revenue == 34_900_000_000 + assert rr.fiscal_year == 2024 + + +def test_resolve_falls_back_to_second_concept() -> None: + facts = ( + Fact( + value=5_000_000_000, + form="10-K", + filed="2024-03-01", + frame="CY2023", + fiscal_year=2023, + start="2023-01-01", + end="2023-12-31", + ), + ) + client = _FakeClient( + "0000320193", + {"Revenues": None, "RevenueFromContractWithCustomerExcludingAssessedTax": _concept(facts)}, + ) + rr = resolve_revenue("ABC", client=client) + assert rr.revenue == 5_000_000_000 + assert rr.concept == "RevenueFromContractWithCustomerExcludingAssessedTax" + + +def test_picks_latest_year_across_concepts() -> None: + # Old "Revenues" tag stops at FY2017; ASC 606 tag carries FY2023 → newer wins. + old = ( + Fact( + value=1_500_000_000, + form="10-K", + filed="2018-03-01", + frame="CY2017", + fiscal_year=2017, + start="2017-01-01", + end="2017-12-31", + ), + ) + new = ( + Fact( + value=34_900_000_000, + form="10-K", + filed="2024-03-01", + frame="CY2023", + fiscal_year=2023, + start="2023-01-01", + end="2023-12-31", + ), + ) + client = _FakeClient( + "0000320193", + { + "Revenues": _concept(old), + "RevenueFromContractWithCustomerExcludingAssessedTax": _concept(new), + }, + ) + rr = resolve_revenue("CRM", client=client) + assert rr.revenue == 34_900_000_000 + assert rr.fiscal_year == 2023 + assert rr.concept == "RevenueFromContractWithCustomerExcludingAssessedTax" + assert client.facts_calls == 1 # ONE companyfacts request covers every tag + + +def test_unknown_ticker_raises() -> None: + with pytest.raises(RevenueLookupError): + resolve_revenue("NOPE", client=_FakeClient(None, {})) + + +def test_companyfacts_error_is_honest_lookup_error() -> None: + class _ErrClient(_FakeClient): + def get_company_facts(self, cik: str) -> SimpleNamespace: + raise EDGARClientError("EDGAR 404: resource not found") + + with pytest.raises(RevenueLookupError) as exc_info: + resolve_revenue("ABC", client=_ErrClient("0000320193", {})) + assert "no usable annual revenue concept" in str(exc_info.value) + + +def test_no_revenue_concept_raises_and_names_tried_tags() -> None: + with pytest.raises(RevenueLookupError) as exc_info: + resolve_revenue("ABC", client=_FakeClient("0000320193", {"Revenues": None})) + msg = str(exc_info.value) + assert "Revenues" in msg + assert "SalesRevenueNet" in msg diff --git a/backend/tests/sizing/test_sense_check.py b/backend/tests/sizing/test_sense_check.py new file mode 100644 index 0000000..f200f20 --- /dev/null +++ b/backend/tests/sizing/test_sense_check.py @@ -0,0 +1,312 @@ +"""Unit + CLI tests for the market-share-implied TAM sense-check.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from typer.testing import CliRunner + +from strata.cli import app +from strata.sizing.sense_check import ( + Player, + SenseCheckError, + parse_money, + parse_player, + sense_check, + share_token_warning, +) + +runner = CliRunner() +_DATA_DIR = Path(__file__).resolve().parents[3] / "data" + + +# ── parsing ───────────────────────────────────────────────────────────────── +@pytest.mark.parametrize( + ("token", "expected"), + [ + ("31e9", 31e9), + ("31000000000", 31e9), + ("3.1b", 3.1e9), + ("950m", 950e6), + ("$1.2B", 1.2e9), + ("12,000", 12000.0), + ("2t", 2e12), + ], +) +def test_parse_money(token: str, expected: float) -> None: + assert parse_money(token) == pytest.approx(expected) + + +def test_parse_player_normalizes_share_and_money() -> None: + p = parse_player("Gong|300m|15") # 15 → 0.15 (percent) + assert p.name == "Gong" + assert p.revenue == pytest.approx(300e6) + assert p.market_share == pytest.approx(0.15) + assert parse_player("Clari|200m|0.10").market_share == pytest.approx(0.10) + + +def test_parse_player_bad_spec_raises() -> None: + with pytest.raises(SenseCheckError): + parse_player("only-two|300m") + + +# ── math ───────────────────────────────────────────────────────────────────── +def test_consistent_shares_imply_one_tam() -> None: + # 300m@15% and 200m@10% both imply a $2.0B market → spread 1.0, clean. + r = sense_check([Player("Gong", 300e6, 0.15), Player("Clari", 200e6, 0.10)]) + assert r.implied_tam == pytest.approx(2e9) # 500m / 0.25 + assert r.revenue_floor == pytest.approx(500e6) + assert r.combined_share == pytest.approx(0.25) + assert r.spread_ratio == pytest.approx(1.0) + + +def test_inconsistent_shares_flag_spread() -> None: + # 300m@15% → 2.0B but 200m@4% → 5.0B: a 2.5x disagreement is flagged. + r = sense_check([Player("A", 300e6, 0.15), Player("B", 200e6, 0.04)]) + assert r.spread_ratio == pytest.approx(2.5) + assert any("disagree" in w for w in r.warnings) + + +def test_combined_share_over_100pct_warns_and_suppresses_band() -> None: + # Σshare > 100%: a band would exclude its own point estimate, leverage < 1 + # is meaningless, residual negative — all suppressed; floor + warning remain. + r = sense_check([Player("A", 300e6, 0.6), Player("B", 200e6, 0.6)]) + assert r.combined_share == pytest.approx(1.2) + assert any("100%" in w for w in r.warnings) + assert r.implied_low is None + assert r.implied_high is None + assert r.leverage is None + assert r.residual is None + assert r.revenue_floor == pytest.approx(500e6) + + +def test_modeled_revenue_warns_q20() -> None: + r = sense_check([Player("A", 300e6, 0.15)]) + assert not r.all_revenue_sourced + assert any("MODELED" in w for w in r.warnings) + + +def test_all_sourced_run_is_still_modeled_q20() -> None: + # Shares are ALWAYS analyst estimates, so even an all-sourced-revenue run + # carries the MODELED marker — the label is unconditional (Q20). + r = sense_check( + [ + Player("A", 300e6, 0.15, revenue_sourced=True), + Player("B", 200e6, 0.10, revenue_sourced=True), + ] + ) + assert r.all_revenue_sourced + assert any("MODELED" in w for w in r.warnings) + assert any("analyst estimates" in w for w in r.warnings) + + +def test_share_token_warning_flags_the_1_trap() -> None: + assert share_token_warning("1", 1.0) is not None # bare '1' → 100%: warn + assert share_token_warning("100", 1.0) is not None # bare '100' → 100%: warn + assert share_token_warning("1.0", 1.0) is None # explicit decimal: intended + assert share_token_warning("0.15", 0.15) is None + assert share_token_warning("15", 0.15) is None + + +def test_cross_check_verdict_within_above_below() -> None: + scenarios = (1e9, 2e9, 3e9) # sourced range 1B-3B, mid 2B + assert ( + sense_check([Player("A", 4e8, 0.2)], sourced_scenarios=scenarios).verdict == "within" + ) # 2B + assert ( + sense_check([Player("A", 4e8, 0.05)], sourced_scenarios=scenarios).verdict == "above" + ) # 8B + assert ( + sense_check([Player("A", 4e8, 0.8)], sourced_scenarios=scenarios).verdict == "below" + ) # 0.5B + + +def test_bad_inputs_raise() -> None: + with pytest.raises(SenseCheckError): + sense_check([]) + with pytest.raises(SenseCheckError): + sense_check([Player("A", -1, 0.1)]) + with pytest.raises(SenseCheckError): + sense_check([Player("A", 1e9, 1.2)]) # share > 1 + with pytest.raises(SenseCheckError): + sense_check([Player("A", 1e9, 0.2)], share_tolerance=1.5) # bad tolerance + + +# ── defensibility hardening ─────────────────────────────────────────────────── +def test_sensitivity_band_brackets_the_point_estimate() -> None: + r = sense_check([Player("A", 4e8, 0.2)], share_tolerance=0.25) # 0.4/0.2 = 2.0B + assert r.implied_low < r.implied_tam < r.implied_high + assert r.implied_low == pytest.approx(4e8 / 0.25) # share +25% -> 1.6B + assert r.implied_high == pytest.approx(4e8 / 0.15) # share -25% -> ~2.67B + + +def test_low_combined_share_flags_leverage() -> None: + # 5% combined share → implied TAM is 20x the captured revenue → fragile. + r = sense_check([Player("A", 1e8, 0.05)]) + assert r.leverage == pytest.approx(20.0) + assert any("leveraged" in w for w in r.warnings) + + +def test_residual_is_implied_minus_floor() -> None: + r = sense_check([Player("A", 3e8, 0.15), Player("B", 2e8, 0.10)]) # implied 2.0B, floor 0.5B + assert r.residual == pytest.approx(r.implied_tam - r.revenue_floor) + assert r.residual == pytest.approx(1.5e9) + + +def test_sourced_flag_parsed_and_counted() -> None: + p = parse_player("Salesforce|10b|0.3|sourced") + assert p.revenue_sourced is True + r = sense_check([p, Player("Est", 1e9, 0.1)]) + assert r.n_sourced == 1 + assert not r.all_revenue_sourced + + +def test_result_carries_assumptions() -> None: + r = sense_check([Player("A", 3e8, 0.15)]) + assert len(r.assumptions) >= 4 + assert any("revenue IN THIS MARKET" in a for a in r.assumptions) + + +# ── CLI ─────────────────────────────────────────────────────────────────────── +def test_cli_sense_check_standalone() -> None: + r = runner.invoke( + app, + ["sense-check", "--player", "Gong|300m|0.15", "--player", "Clari|200m|0.10"], + ) + assert r.exit_code == 0, r.stdout + assert "implied TAM" in r.stdout + assert "$2.0B" in r.stdout # money renders via the repo-wide usd() formatter + + +def test_cli_edgar_token_resolves_sourced_revenue(monkeypatch: pytest.MonkeyPatch) -> None: + # Monkeypatch the EDGAR lookup so the CLI path stays offline + deterministic. + import strata.cli as cli + from strata.sizing.revenue_lookup import ResolvedRevenue + + def _fake_resolve(ticker: str, **_kw: object) -> ResolvedRevenue: + return ResolvedRevenue( + ticker=ticker.upper(), + cik="0000320193", + entity_name="Salesforce, Inc.", + revenue=34_900_000_000, + fiscal_year=2023, + concept="Revenues", + source_url="https://www.sec.gov/cgi-bin/browse-edgar?CIK=0000320193", + ) + + monkeypatch.setattr(cli, "resolve_revenue", _fake_resolve) + r = runner.invoke(app, ["sense-check", "--player", "Salesforce|edgar:CRM|0.30"]) + assert r.exit_code == 0, r.stdout + assert "resolved Salesforce revenue" in r.stdout + assert "1/1 revenues sourced" in r.stdout # EDGAR figure counts as sourced + # The figure is whole-company revenue: caveat inline + a marked player line. + assert "TOTAL company revenue" in r.stdout + assert "(filing, total-co)" in r.stdout + + +def test_cli_edgar_token_blocked_offline(monkeypatch: pytest.MonkeyPatch) -> None: + import strata.cli as cli + + def _boom(*_a: object, **_k: object) -> object: + raise AssertionError("resolve_revenue must not be called when STRATA_OFFLINE=1") + + monkeypatch.setattr(cli, "resolve_revenue", _boom) + monkeypatch.setenv("STRATA_OFFLINE", "1") + r = runner.invoke(app, ["sense-check", "--player", "Salesforce|edgar:CRM|0.30"]) + assert r.exit_code == 2 # offline guard fired before any live lookup + + +def test_cli_overflow_share_prints_floor_only() -> None: + # Σshare = 120%: no implied-TAM headline, band, leverage, or residual lines — + # only the revenue floor and the overlap warning are defensible. + r = runner.invoke(app, ["sense-check", "--player", "A|300m|60", "--player", "B|200m|60"]) + assert r.exit_code == 0, r.stdout + assert "revenue floor" in r.stdout + assert "only the revenue floor" in r.stdout # the >100% warning + assert "implied TAM $" not in r.stdout + assert "sensitivity" not in r.stdout + assert "leverage" not in r.stdout + assert "residual" not in r.stdout + + +def test_cli_share_token_one_warns() -> None: + # The '1' trap: a bare integer share token landing on exactly 100% warns. + r = runner.invoke(app, ["sense-check", "--player", "Solo|300m|1"]) + assert r.exit_code == 0, r.stdout + assert "parsed as 100%" in r.stdout + + +def test_cli_edgar_spec_enforces_field_count() -> None: + # The edgar branch must reject malformed specs as loudly as the plain branch. + r = runner.invoke(app, ["sense-check", "--player", "Salesforce|edgar:CRM"]) + assert r.exit_code == 2 + r = runner.invoke(app, ["sense-check", "--player", "A|edgar:CRM|0.3|x|y"]) + assert r.exit_code == 2 + + +def test_cli_csv_players_parse_directly(tmp_path: Path) -> None: + # CSV fields go straight into Player — a '|' in a company name must not + # inject extra pipe-DSL fields, and the optional sourced column is honored. + csv_path = tmp_path / "players.csv" + csv_path.write_text( + "name,revenue,market_share,sourced\nAcme|Pipes Inc,300m,15,\nFiler Co,200m,10,sourced\n", + encoding="utf-8", + ) + r = runner.invoke(app, ["sense-check", "--players-file", str(csv_path)]) + assert r.exit_code == 0, r.stdout + assert "Acme|Pipes Inc" in r.stdout + assert "1/2 revenues sourced" in r.stdout + assert "$2.0B" in r.stdout # 500m / 25%, via the repo-wide usd() formatter + + +def test_cli_csv_edgar_rows_share_one_client( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + import strata.cli as cli + from strata.sizing.revenue_lookup import ResolvedRevenue + + seen_clients: list[object] = [] + + def _fake_resolve(ticker: str, **kw: object) -> ResolvedRevenue: + seen_clients.append(kw.get("client")) + return ResolvedRevenue( + ticker=ticker.upper(), + cik="0000000001", + entity_name=f"{ticker} Corp", + revenue=1_000_000_000, + fiscal_year=2024, + concept="Revenues", + source_url="https://www.sec.gov/cgi-bin/browse-edgar?CIK=0000000001", + ) + + monkeypatch.setattr(cli, "resolve_revenue", _fake_resolve) + csv_path = tmp_path / "players.csv" + csv_path.write_text( + "name,revenue,market_share\nAlpha,edgar:AAA,10\nBeta,edgar:BBB,20\n", + encoding="utf-8", + ) + r = runner.invoke(app, ["sense-check", "--players-file", str(csv_path)]) + assert r.exit_code == 0, r.stdout + assert len(seen_clients) == 2 + assert seen_clients[0] is not None + assert seen_clients[0] is seen_clients[1] # ONE EDGARClient across rows + + +def test_cli_sense_check_against_curated_concept() -> None: + # demo-ai-sales-ops-us has a sourced scenario range; the check should compare. + r = runner.invoke( + app, + [ + "sense-check", + "--concept", + "demo-ai-sales-ops-us", + "--player", + "PlayerOne|2b|0.5", + "-d", + str(_DATA_DIR), + ], + ) + assert r.exit_code == 0, r.stdout + assert "cross-check vs demo-ai-sales-ops-us" in r.stdout + assert "verdict:" in r.stdout diff --git a/backend/tests/test_edgar_parse_fact.py b/backend/tests/test_edgar_parse_fact.py new file mode 100644 index 0000000..be92412 --- /dev/null +++ b/backend/tests/test_edgar_parse_fact.py @@ -0,0 +1,50 @@ +"""Regression: EDGAR _parse_fact must tolerate present-but-null fy/fp. + +Real SEC XBRL companyconcept facts frequently carry "fy": null / "fp": null +(key present, value None). The original guard keyed on `"fy" in raw`, so +int(None) crashed the parse on live data. Guard on the value instead. +""" + +from __future__ import annotations + +from strata.connectors.edgar import _parse_fact + + +def test_parse_fact_tolerates_null_fy_fp() -> None: + fact = _parse_fact( + {"val": 34_900_000_000, "form": "10-K", "filed": "2024-03-01", "fy": None, "fp": None} + ) + assert fact.value == 34_900_000_000 + assert fact.fiscal_year is None + assert fact.fiscal_period is None + + +def test_parse_fact_reads_real_fy_fp() -> None: + fact = _parse_fact({"val": 1.0, "form": "10-K", "filed": "2024-03-01", "fy": 2023, "fp": "FY"}) + assert fact.fiscal_year == 2023 + assert fact.fiscal_period == "FY" + + +def test_parse_fact_carries_period_start_end() -> None: + # fy/fp describe the FILING; only start/end identify the period a + # comparative fact actually covers — they must survive the parse. + fact = _parse_fact( + { + "val": 1.0, + "form": "10-K", + "filed": "2025-03-01", + "fy": 2025, + "start": "2023-02-01", + "end": "2024-01-31", + } + ) + assert fact.start == "2023-02-01" + assert fact.end == "2024-01-31" + + +def test_parse_fact_tolerates_missing_or_null_period() -> None: + fact = _parse_fact({"val": 1.0, "form": "10-K", "filed": "2024-03-01"}) + assert fact.start == "" + assert fact.end == "" + fact = _parse_fact({"val": 1.0, "form": "10-K", "filed": "2024-03-01", "end": None}) + assert fact.end == "" diff --git a/backend/tests/test_mcp_server.py b/backend/tests/test_mcp_server.py index 7e831f3..32aede7 100644 --- a/backend/tests/test_mcp_server.py +++ b/backend/tests/test_mcp_server.py @@ -42,6 +42,7 @@ "get_report_for_tam", "export_report", "verify_market_claim", + "sense_check", "admin_list_feedback", "admin_promote_correction", "report_sourcing_gap", diff --git a/docs/methods/sense-check-defensibility.md b/docs/methods/sense-check-defensibility.md new file mode 100644 index 0000000..59b4d86 --- /dev/null +++ b/docs/methods/sense-check-defensibility.md @@ -0,0 +1,62 @@ +# Sense-check defensibility — surviving the 10 arguments + +The market-share-implied TAM (`strata sense-check`) is a **cross-check**, not a +primary sourced number. This memo is the adversarial review: the ten arguments a +TAM expert raises against the market-share method, and exactly how the tool +answers each — by code, by a surfaced warning, by an explicit assumption, or by +an honest stated limitation. Nothing is hidden; the implied TAM only claims to be +valid if the assumptions hold, and it shows its own uncertainty. + +The method: + + implied TAM = Σ(player revenue) / Σ(player market share) + revenue floor = Σ(player revenue) # a TAM can't be below this + per-player = revenueᵢ / shareᵢ # must all agree + residual = implied − floor # revenue attributed to everyone else + leverage = 1 / Σshare # error multiplier + band = implied at shares ±25% # the estimate is a range, not a point + +--- + +## The 10 arguments + +| # | Expert argument | How the check answers it | +|---|---|---| +| 1 | **"Reported revenue isn't in-market revenue"** — a vendor's total revenue spans adjacent products. | Assumption #1 is stated on every run ("each revenue is the player's revenue IN THIS MARKET"). The tool can't detect this, so it makes the requirement explicit and tags every non-filing revenue `(est.)`. An `edgar:` revenue is tagged `(filing, total-co)` and resolution prints an inline caveat ("EDGAR figure is TOTAL company revenue — confirm it equals in-market revenue or use a segment figure") — a sourced total never passes as an unmarked in-market figure. | +| 2 | **"Share of *what*?"** — revenue share ≠ unit/seat/logo/mindshare. | Assumption #2 ("market shares are revenue shares OF THIS SAME MARKET"). The math is only coherent for revenue shares; the assumption names it so a seats-share input is caught by the reviewer, not silently divided. | +| 3 | **"A small combined share explodes the error."** | The `leverage = 1/Σshare` factor is printed, and a **warning fires when combined share ≤ 25%** (matching `LEVERAGE_WARN_SHARE` with `<=` in code: "the implied TAM is Nx the captured revenue … widen the player set"). The **sensitivity band** (shares ±25%) shows how far the answer moves. | +| 4 | **"Your shares overlap / sum past 100%."** | Detected and warned ("combined market share is N% (> 100%) … only the revenue floor is defensible"). When Σshare ≥ 100% the band, leverage, and residual are **suppressed entirely** — a band derived from overlapping shares would exclude its own point estimate, so only the revenue floor and the warning are emitted. Never rendered as if clean. | +| 5 | **"Your share estimates are internally inconsistent."** | The **per-player spread** (`max(revᵢ/shareᵢ) / min(...)`) is computed; a spread beyond 2× warns. If every player implies a different TAM, the reviewer sees it immediately. | +| 6 | **"Currencies don't match."** | Assumption #3 ("one currency"). The tool is unit-agnostic by design; the assumption is load-bearing and shown. | +| 7 | **"Periods don't match"** — TTM vs FY, mixed years. | Assumption #3 ("one period, e.g. TTM"). Same treatment: explicit, not silent. | +| 8 | **"Geography mismatch"** — global revenue against a US market. | Assumption #3 ("same geography as the market"). This ties to Strata's first-class `MarketScope` (geo_supply/geo_demand); a future increment can bind the players' geo to the concept's declared scope and auto-flag a mismatch. | +| 9 | **"Private-company revenue is itself a guess — this is circular."** | Each revenue carries a `revenue_sourced` flag; the output shows `n/N revenues sourced` and tags estimates `(est.)`. The **MODELED label is unconditional**: market shares are always analyst estimates, so the implied TAM is labelled **MODELED (Q20)** even when every revenue is a sourced filing, and is **never persisted as a sourced primitive** — it can only agree or disagree with the sourced number, never become one. | +| 10 | **"This market isn't revenue-defined yet"** — pre-revenue / latent demand. | Assumption #4 ("the market is revenue-defined today, not pre-revenue / latent"). For a `latent_demand` market the method is the wrong tool, and the assumption says so — the reviewer uses the adoption-curve lens instead. | + +### Bonus: "comparing to one number hides the real spread" +When `--concept` is given, the implied TAM is compared to the curated market's +**full sourced scenario range** (`within / above / below`), not a single point — +honoring Strata's rule that TAMs are scenario trees, never scalars. The verdict +also reports the ratio to the sourced midpoint. + +--- + +## What "defensible" means here + +1. **Every number shows its derivation** — revenue floor, per-player implied, + residual, leverage, and the ±25% band are all printed; nothing is a black box. +2. **The estimate is a range, not a point** — the band is the headline a careful + reviewer trusts. +3. **The assumptions are printed every run** — the five conditions above are the + reasoning a TAM expert audits first; they are explicit, not implied. +4. **It fails honest** — overlap, inconsistency, fragility, and modeled inputs all + produce warnings; a miss is never dressed up as a sourced fact (Q20). + +## Known limitations (stated, not hidden) +- Cannot verify that a revenue is in-market, same-currency, same-period, or + same-geography — it asserts the assumption and tags estimates; the analyst owns + the inputs. +- `MarketScope`-aware geo/period binding (argument 8) is a future increment. +- `edgar:TICKER` pulls a sourced annual revenue from SEC EDGAR (US-listed filers + only), selected by the fact's own period end — but it is TOTAL company revenue, + tagged `(filing, total-co)` with an inline caveat; segment figures stay manual.