feat(mcp): cap financial-statement-analysis at the columns a filing presents

robosystems/middleware/mcp/README.md

@@ -124,14 +124,24 @@ Require `roboledger` in `schema_extensions`. These read LadybugDB (OLAP).
 |------|-------------|------------|
 | `get-example-queries` | Query patterns tailored to this graph's schema | — |
 | `resolve-element` | Map a concept ("revenue") to XBRL element qnames | manifest `has_semantic_enrichment=True` |
-| `financial-statement-analysis` | Graph-backed statement read with auto-resolve and dedup | — |
+| `financial-statement-analysis` | Graph-backed statement read with auto-resolve, dedup, and a period cap at the columns a filing presents | — |
 | `live-financial-statement` | Statement from the tenant's live OLTP ledger via CoA→GAAP mapping | tenant graphs only |
 | `build-fact-grid` | Cross-company comparison over canonical concepts | `FACT_GRID_ENABLED` |
 
 `financial-statement-analysis` resolves the latest relevant SEC filing when no
 `report_id` is given (ticker and form-code resolution live in
 `adapters/sec/mcp/report_resolver.py`), and deduplicates facts that appear in
-multiple filings as comparative periods.
+multiple filings as comparative periods. It answers with the columns a filing
+presents, not everything its hypercube holds: an annual form defaults the
+period filter to `annual` (a 10-K's statement hypercube also carries the
+quarterly figures from its notes), `periods` caps the distinct end dates kept
+(two for the balance sheet, three for the flow statements, newest first) and
+the result names the keys it kept and how many it cut, and fact rows carry no
+null fields and a `name` only when it is a label rather than the local part of
+`qname` (SEC rows repeat the qname there; a tenant's rs-gaap rows carry a
+readable label). The graph fetch always takes the query's full row budget so
+that `limit`, applied after the cap, can never hide a period from it; a fetch
+that hits the ceiling is flagged `rows_truncated`.
 
 ### Layer 2b — roboledger OLTP
 

robosystems/middleware/mcp/tools/financial_statement_tools.py

@@ -36,6 +36,98 @@
 
 from .base_tool import BaseTool
 
+# The columns a filing presents: two balance-sheet instants, three years of
+# each flow statement. Everything the hypercube holds beyond that — the
+# quarterly note data a 10-K carries, the equity roll-forward's opening
+# instants — is answered on request through ``periods``, not by default.
+PERIODS_DEFAULT_INSTANT = 2
+PERIODS_DEFAULT_DURATION = 3
+PERIODS_MAX = 100
+
+# The raw-row budget handed to the statement query, independent of the
+# caller's ``limit``. The query fetches newest-first and the period cap runs
+# after it, so a row budget tied to ``limit`` would let a small limit hide
+# older periods from the cap with no ``periods_omitted`` to say so. The
+# query's own ceiling is 1,000; ``limit`` is applied after the cap instead.
+QUERY_ROW_CEILING = 1000
+
+_PERIOD_KEY_FIELDS = ("start_date", "end_date", "period_type", "duration_type")
+
+
+def default_period_type(statement_type: str, form: str | None) -> str | None:
+  """The period filter to apply when the caller gave none.
+
+  A 10-K (or 20-F / 40-F) is filed on an annual cadence, yet its statement
+  hypercube also carries the quarterly figures from its notes — on a FY2024
+  10-K income statement, 8 of 11 period keys and more than half the rows.
+  Nobody asking for "the income statement" wants those, so an annual form
+  defaults to ``annual``. The balance sheet already defaults to instants in
+  the query; a 10-Q's quarter and year-to-date columns share end dates, so
+  the period cap alone bounds it and no filter is forced.
+  """
+  if statement_type == "balance_sheet":
+    return None
+  from robosystems.adapters.sec.mcp import ANNUAL_FORMS
+
+  if form in ANNUAL_FORMS:
+    return "annual"
+  return None
+
+
+def cap_periods(
+  rows: list[dict[str, Any]], periods: int
+) -> tuple[list[dict[str, Any]], list[dict[str, Any]], int]:
+  """Keep the rows whose period ends on one of the ``periods`` newest end dates.
+
+  Returns the kept rows, the distinct period keys they span (newest first),
+  and how many older end dates were cut — so a caller that wants a longer
+  series knows to ask for one.
+  """
+  end_dates = sorted({str(r.get("end_date") or "") for r in rows}, reverse=True)
+  kept_dates = set(end_dates[:periods])
+  kept = [r for r in rows if str(r.get("end_date") or "") in kept_dates]
+  keys: list[dict[str, Any]] = []
+  seen: set[tuple[Any, ...]] = set()
+  for row in kept:
+    key = tuple(row.get(f) for f in _PERIOD_KEY_FIELDS)
+    if key in seen:
+      continue
+    seen.add(key)
+    keys.append({f: row.get(f) for f in _PERIOD_KEY_FIELDS if row.get(f) is not None})
+  keys.sort(
+    key=lambda k: (str(k.get("end_date") or ""), str(k.get("start_date") or "")),
+    reverse=True,
+  )
+  return kept, keys, max(0, len(end_dates) - periods)
+
+
+def compact_fact(row: dict[str, Any]) -> dict[str, Any]:
+  """A fact row as a model reads it: null fields say nothing and are dropped,
+  and ``name`` is carried only when it says more than ``qname`` does.
+
+  On SEC filings ``Element.name`` is the qname's local part
+  (``us-gaap:Assets`` / ``Assets``), pure repetition. On a tenant graph the
+  rs-gaap elements carry a label there (``rs-gaap:NonoperatingIncomeExpense``
+  / ``Nonoperating Income (Expense)``), which reads better than the qname
+  and stays. Nothing in the schema pins either shape, so the row is judged,
+  not the graph.
+  """
+  qname = row.get("qname") or ""
+  name = row.get("name")
+  if name == qname.rsplit(":", 1)[-1]:
+    name = None
+  fact = {
+    "canonical_concept": row.get("canonical_concept"),
+    "qname": qname or None,
+    "name": name,
+    "value": row.get("value"),
+    "start_date": row.get("start_date"),
+    "end_date": row.get("end_date"),
+    "period_type": row.get("period_type"),
+    "duration_type": row.get("duration_type"),
+  }
+  return {k: v for k, v in fact.items() if v is not None}
+
 
 class LiveFinancialStatementTool(BaseTool):
   """MCP tool: generate an ad-hoc OLTP-backed statement for a tenant graph."""
@@ -196,11 +288,19 @@ def get_tool_definition(self) -> dict[str, Any]:
   cash_flow_statement, equity_statement
 - `period_type` — annual (10-K/20-F/40-F, duration facts), quarterly (10-Q
   plus annuals for international filers, duration facts), or instant
-  (point-in-time facts; the balance-sheet default)
+  (point-in-time facts; the balance-sheet default). Unset on an annual form
+  it defaults to annual, so a 10-K's quarterly note figures stay out
+- `periods` — how many period end dates to keep, newest first (default
+  2 for the balance sheet, 3 for the flow statements — the columns a
+  filing presents). Raise it for a longer series
 - `ticker` / `report_id` — which one is required depends on the graph; see NOTES
 
 **RETURNS:**
-- Deduplicated facts (qname, name, value, end_date) ordered by end_date DESC
+- Deduplicated facts (canonical_concept, qname, value, start_date / end_date,
+  period_type, duration_type; null fields omitted; `name` only when it is a
+  label rather than the qname's local part) ordered by end_date DESC
+- `periods` — the period keys the facts span, newest first, and
+  `periods_omitted` when older end dates were cut by the cap
 - resolved_report info when auto-resolution was used
 - Dimensional/segment breakdowns are filtered out (consolidated totals only)
 """,
@@ -229,9 +329,13 @@ def get_tool_definition(self) -> dict[str, Any]:
             "description": "Filter by period type",
             "enum": ["annual", "quarterly", "instant"],
           },
+          "periods": {
+            "type": "integer",
+            "description": "Period end dates to keep, newest first (default 2 for balance_sheet, 3 otherwise; max 100). Raise it for a longer series.",
+          },
           "limit": {
             "type": "integer",
-            "description": "Max fact rows returned (1-1000). Leave at the default for a whole statement; a lower cap cuts rows while subtotals still reflect the full set.",
+            "description": "Max fact rows returned (1-1000), applied after the period cap. Leave at the default for a whole statement; a lower cap cuts rows while subtotals still reflect the full set.",
             "default": 1000,
           },
         },
@@ -259,6 +363,17 @@ async def execute(self, arguments: dict[str, Any]) -> dict[str, Any]:
     fiscal_year = arguments.get("fiscal_year")
     period_type = arguments.get("period_type")
     limit = max(1, min(int(arguments.get("limit", 1000)), 1000))
+    periods_default = (
+      PERIODS_DEFAULT_INSTANT
+      if statement_type == "balance_sheet"
+      else PERIODS_DEFAULT_DURATION
+    )
+    periods_arg = arguments.get("periods")
+    periods = (
+      periods_default
+      if periods_arg is None
+      else max(1, min(int(periods_arg), PERIODS_MAX))
+    )
 
     graph_id = self.client.graph_id
     is_shared = is_shared_repository_or_subgraph(graph_id)
@@ -302,6 +417,11 @@ async def execute(self, arguments: dict[str, Any]) -> dict[str, Any]:
           ),
         }
 
+    if period_type is None:
+      period_type = default_period_type(
+        statement_type, resolved.get("form") if resolved else None
+      )
+
     rows: list[dict] = []
     if report_id or ticker:
       rows = await query_financial_statement(
@@ -310,32 +430,36 @@ async def execute(self, arguments: dict[str, Any]) -> dict[str, Any]:
         report_id=report_id,
         ticker=ticker,
         period_type=period_type,
-        limit=limit,
+        limit=QUERY_ROW_CEILING,
       )
 
-    deduped = deduplicate_facts(rows)[:limit]
-    facts = [
-      {
-        "canonical_concept": row.get("canonical_concept"),
-        "qname": row.get("qname"),
-        "name": row.get("name"),
-        "value": row.get("value"),
-        "start_date": row.get("start_date"),
-        "end_date": row.get("end_date"),
-        "period_type": row.get("period_type"),
-        "duration_type": row.get("duration_type"),
-      }
-      for row in deduped
-    ]
+    kept, period_keys, omitted = cap_periods(deduplicate_facts(rows), periods)
+    facts = [compact_fact(row) for row in kept[:limit]]
 
     result: dict[str, Any] = {
       "graph_id": graph_id,
       "statement_type": statement_type,
       "ticker": ticker,
       "report_id": report_id,
+      "periods": period_keys,
       "facts": facts,
       "fact_count": len(facts),
     }
+    if omitted:
+      result["periods_omitted"] = omitted
+      result["periods_tip"] = (
+        f"{omitted} earlier period end date(s) were cut by the cap of {periods}; "
+        "raise `periods` for a longer series."
+      )
+    if len(rows) >= QUERY_ROW_CEILING:
+      # The graph fetch is newest-first and hit its ceiling, so the oldest
+      # kept period may be incomplete and the cap cannot see past it.
+      result["rows_truncated"] = True
+      result["rows_tip"] = (
+        f"The graph fetch hit its {QUERY_ROW_CEILING}-row ceiling before the "
+        "cap ran; the oldest period kept may be incomplete. Narrow with "
+        "`period_type` or `report_id`."
+      )
 
     if resolved:
       result["resolved_report"] = {