[Epic #845][P1] Return-series spreadsheet export

pyproject.toml

@@ -22,6 +22,7 @@ classifiers = [
 ]
 dependencies = [
     "langsmith>=0.4.59",
+    "openpyxl>=3.1.5",
     "stranske-pdf-extract @ git+https://github.com/stranske/Workflows.git@pdf-extract-v0.1.0#subdirectory=packages/stranske_pdf_extract",
 ]
 
@@ -122,4 +123,4 @@ ignore_missing_imports = true
 
 [tool.black]
 line-length = 100
-target-version = ["py312"]
+target-version = ["py312"]

src/inv_man_intake/export/__init__.py

@@ -16,6 +16,7 @@
     build_default_export_service,
     ensure_export_service,
 )
+from inv_man_intake.export.spreadsheet import export_return_series
 
 __all__ = [
     "DefaultExportService",
@@ -30,4 +31,5 @@
     "ManifestSkip",
     "build_default_export_service",
     "ensure_export_service",
+    "export_return_series",
 ]

src/inv_man_intake/export/spreadsheet.py

@@ -0,0 +1,182 @@
+"""No-egress spreadsheet exports for normalized return series."""
+
+from __future__ import annotations
+
+import csv
+import math
+from collections.abc import Sequence
+from io import BytesIO, StringIO
+
+from inv_man_intake.export.manifest import ExportArtifact, ExportItem, ExportManifest
+from inv_man_intake.extraction.providers.base import ExtractedTable
+from inv_man_intake.performance.contracts import (
+    PerformancePayload,
+    PerformancePoint,
+    PerformanceSeries,
+)
+from inv_man_intake.performance.normalize import NormalizedPerformancePayload, normalize_payload
+
+
+def export_return_series(
+    performance: PerformancePayload | NormalizedPerformancePayload | None,
+    *,
+    source_doc_id: str,
+    source_page: int | None = None,
+    method: str = "normalized-performance",
+    tables: Sequence[ExtractedTable] = (),
+) -> ExportManifest:
+    """Export normalized performance rows as XLSX and CSV with source lineage.
+
+    ``None`` is the explicit no-series path for callers that completed extraction
+    without a usable return series.  Concrete payloads are normalized before any
+    cell is written, which also reuses the existing finite-value validation.
+    """
+
+    if performance is None:
+        return ExportManifest.from_items(
+            (ExportItem(item_ref="return-series", skip_reason="no_series_found"),)
+        )
+
+    normalized = (
+        performance
+        if isinstance(performance, NormalizedPerformancePayload)
+        else normalize_payload(performance)
+    )
+    rows = _return_rows(
+        normalized, source_doc_id=source_doc_id, source_page=source_page, method=method
+    )
+    if not rows:
+        return ExportManifest.from_items(
+            (ExportItem(item_ref="return-series", skip_reason="no_series_found"),)
+        )
+
+    provenance_refs = tuple(sorted({row[3] for row in rows}))
+    return ExportManifest.from_items(
+        (
+            ExportItem(
+                item_ref="return-series.csv",
+                artifact=ExportArtifact(
+                    name="return-series.csv",
+                    media_type="text/csv",
+                    content=_build_csv(rows),
+                    provenance_refs=provenance_refs,
+                ),
+            ),
+            ExportItem(
+                item_ref="return-series.xlsx",
+                artifact=ExportArtifact(
+                    name="return-series.xlsx",
+                    media_type=(
+                        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
+                    ),
+                    content=_build_workbook(rows, tables=tables, provenance_refs=provenance_refs),
+                    provenance_refs=provenance_refs,
+                ),
+            ),
+        )
+    )
+
+
+def _return_rows(
+    normalized: NormalizedPerformancePayload,
+    *,
+    source_doc_id: str,
+    source_page: int | None,
+    method: str,
+) -> tuple[tuple[str, float, str, str], ...]:
+    provenance = f"{source_doc_id}:{source_page if source_page is not None else 'unknown'}:{method}"
+    rows: list[tuple[str, float, str, str]] = []
+    for series in (normalized.monthly, normalized.quarterly, normalized.annual):
+        if series is not None:
+            rows.extend(_series_rows(series, provenance=provenance))
+    return tuple(rows)
+
+
+def _series_rows(
+    series: PerformanceSeries,
+    *,
+    provenance: str,
+) -> tuple[tuple[str, float, str, str], ...]:
+    rows: list[tuple[str, float, str, str]] = []
+    for point in series.points:
+        _require_finite(point)
+        rows.append((point.as_of.isoformat(), point.value, series.frequency, provenance))
+    return tuple(rows)
+
+
+def _require_finite(point: PerformancePoint) -> None:
+    if not math.isfinite(point.value):
+        raise ValueError("return series contains a non-finite value")
+
+
+_FORMULA_LEADING_CHARACTERS = ("=", "+", "-", "@")
+
+
+def _safe_cell(value: object) -> object:
+    """Neutralize text a spreadsheet client would evaluate as a formula.
+
+    Extracted document text reaches these cells unfiltered, so a value such as
+    ``=HYPERLINK(...)`` would execute on open.  Numbers are written as numbers
+    and are never rewritten.
+    """
+
+    if isinstance(value, str) and value.startswith(_FORMULA_LEADING_CHARACTERS):
+        return f"'{value}"
+    return value
+
+
+def _safe_row(row: Sequence[object]) -> tuple[object, ...]:
+    return tuple(_safe_cell(cell) for cell in row)
+
+
+def _build_csv(rows: Sequence[tuple[str, float, str, str]]) -> bytes:
+    buffer = StringIO(newline="")
+    writer = csv.writer(buffer, lineterminator="\n")
+    writer.writerow(("period", "value", "frequency", "provenance"))
+    writer.writerows(_safe_row(row) for row in rows)
+    return buffer.getvalue().encode("utf-8")
+
+
+def _build_workbook(
+    rows: Sequence[tuple[str, float, str, str]],
+    *,
+    tables: Sequence[ExtractedTable],
+    provenance_refs: Sequence[str],
+) -> bytes:
+    from openpyxl import Workbook  # type: ignore[import-untyped]
+
+    workbook = Workbook()
+    returns_sheet = workbook.active
+    returns_sheet.title = "Return Series"
+    returns_sheet.append(("period", "value", "frequency", "provenance"))
+    for row in rows:
+        returns_sheet.append(_safe_row(row))
+
+    manifest_sheet = workbook.create_sheet("Manifest")
+    manifest_sheet.append(("kind", "reference"))
+    for reference in provenance_refs:
+        manifest_sheet.append(("provenance", _safe_cell(reference)))
+    for reference in _table_lineage(tables):
+        manifest_sheet.append(("table_cell", _safe_cell(reference)))
+
+    output = BytesIO()
+    workbook.save(output)
+    return output.getvalue()
+
+
+def _table_lineage(tables: Sequence[ExtractedTable]) -> tuple[str, ...]:
+    lineage: list[str] = []
+    for table_index, table in enumerate(tables):
+        location = getattr(table, "location", None)
+        source_doc = getattr(location, "source_doc_id", "unknown")
+        page = getattr(location, "source_page", None)
+        table_id = getattr(table, "table_id", "") or str(table_index)
+        for cell in table.cells:
+            lineage.append(
+                f"{source_doc}:{page if page is not None else 'unknown'}:table:{table_id}:"
+                f"r{cell.row_index}:c{cell.column_index}={cell.value}"
+            )
+    return tuple(lineage)
+
+
+__all__ = ["export_return_series"]

tests/export/test_spreadsheet_export.py

@@ -0,0 +1,124 @@
+"""Return-series spreadsheet export coverage."""
+
+from __future__ import annotations
+
+from csv import reader
+from datetime import date
+from io import BytesIO
+from math import inf, nan
+
+import pytest
+from openpyxl import load_workbook
+
+from inv_man_intake.export import export_return_series
+from inv_man_intake.extraction.providers import ExtractedTable, ExtractedTableCell, SourceLocation
+from inv_man_intake.performance import PerformancePayload, PerformancePoint, PerformanceSeries
+from inv_man_intake.performance.normalize import NormalizedPerformancePayload
+
+
+def test_return_series_roundtrips_with_provenance() -> None:
+    payload = PerformancePayload(
+        monthly=PerformanceSeries(
+            frequency="monthly",
+            points=(
+                PerformancePoint(as_of=date(2026, 1, 31), value=0.125),
+                PerformancePoint(as_of=date(2026, 2, 28), value=-0.01),
+            ),
+        )
+    )
+    manifest = export_return_series(
+        payload,
+        source_doc_id="track-record.pdf",
+        source_page=4,
+        method="performance-normalizer",
+        tables=(
+            ExtractedTable(
+                table_id="returns",
+                location=SourceLocation(source_doc_id="track-record.pdf", source_page=4),
+                cells=(ExtractedTableCell(row_index=0, column_index=0, value="Jan 2026"),),
+            ),
+        ),
+    )
+
+    artifacts = {entry.item_ref: entry.artifact for entry in manifest.artifacts}
+    workbook = load_workbook(BytesIO(artifacts["return-series.xlsx"].content), data_only=True)
+    assert workbook.sheetnames == ["Return Series", "Manifest"]
+    assert list(workbook["Return Series"].iter_rows(values_only=True)) == [
+        ("period", "value", "frequency", "provenance"),
+        ("2026-01-31", 0.125, "monthly", "track-record.pdf:4:performance-normalizer"),
+        ("2026-02-28", -0.01, "monthly", "track-record.pdf:4:performance-normalizer"),
+    ]
+    assert any(
+        row[0] == "table_cell" and "Jan 2026" in row[1]
+        for row in workbook["Manifest"].iter_rows(min_row=2, values_only=True)
+    )
+    assert b"track-record.pdf:4:performance-normalizer" in artifacts["return-series.csv"].content
+
+
+@pytest.mark.parametrize("bad_value", (nan, inf, -inf))
+def test_non_finite_value_is_rejected_not_written(bad_value: float) -> None:
+    payload = PerformancePayload(
+        monthly=PerformanceSeries(
+            frequency="monthly",
+            points=(PerformancePoint(as_of=date(2026, 1, 31), value=bad_value),),
+        )
+    )
+    with pytest.raises(ValueError, match="finite"):
+        export_return_series(payload, source_doc_id="track-record.pdf")
+
+
+def test_missing_series_records_skip() -> None:
+    manifest = export_return_series(None, source_doc_id="track-record.pdf")
+    assert [(entry.item_ref, entry.reason_code) for entry in manifest.skipped] == [
+        ("return-series", "no_series_found")
+    ]
+
+
+def test_empty_normalized_series_records_skip() -> None:
+    payload = NormalizedPerformancePayload(
+        monthly=PerformanceSeries(frequency="monthly", points=()),
+        quarterly=None,
+        annual=None,
+        canonical_months=(),
+        missing_months=(),
+    )
+    manifest = export_return_series(payload, source_doc_id="track-record.pdf")
+    assert [(entry.item_ref, entry.reason_code) for entry in manifest.skipped] == [
+        ("return-series", "no_series_found")
+    ]
+    assert manifest.artifacts == ()
+
+
+def test_formula_leading_provenance_is_escaped_in_both_artifacts() -> None:
+    payload = PerformancePayload(
+        monthly=PerformanceSeries(
+            frequency="monthly",
+            points=(PerformancePoint(as_of=date(2026, 1, 31), value=0.125),),
+        )
+    )
+    manifest = export_return_series(
+        payload,
+        source_doc_id='=HYPERLINK("http://evil.example","click")',
+        source_page=4,
+        method="performance-normalizer",
+        tables=(
+            ExtractedTable(
+                table_id="returns",
+                location=SourceLocation(source_doc_id="@cmd.pdf", source_page=4),
+                cells=(ExtractedTableCell(row_index=0, column_index=0, value="Jan 2026"),),
+            ),
+        ),
+    )
+
+    artifacts = {entry.item_ref: entry.artifact for entry in manifest.artifacts}
+    csv_rows = list(reader(artifacts["return-series.csv"].content.decode("utf-8").splitlines()))
+    assert [row[3] for row in csv_rows[1:]] == [
+        '\'=HYPERLINK("http://evil.example","click"):4:performance-normalizer'
+    ]
+
+    workbook = load_workbook(BytesIO(artifacts["return-series.xlsx"].content), data_only=True)
+    return_rows = list(workbook["Return Series"].iter_rows(min_row=2, values_only=True))
+    assert all(str(row[3]).startswith("'=") for row in return_rows)
+    manifest_rows = list(workbook["Manifest"].iter_rows(min_row=2, values_only=True))
+    assert all(not str(reference).startswith(("=", "+", "@")) for _kind, reference in manifest_rows)
+    assert any(str(reference).startswith("'@cmd.pdf") for _kind, reference in manifest_rows)