fix(research): bind price proxy exports to consumed sources

src/market_signal_sources/artifacts/research_export.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import Iterable, Mapping
+import hashlib
 import json
 from os import PathLike
 from pathlib import Path
@@ -28,6 +29,7 @@ def write_research_export_manifest(
     input_sources: Iterable[Mapping[str, Any]] | None = None,
     transform_parameters: Mapping[str, Any] | None = None,
     quality_report_path: str | PathLike[str] | None = None,
+    input_csv_bytes: bytes | None = None,
 ) -> dict[str, Any]:
     """Write a non-runtime manifest for an offline research CSV export."""
 
@@ -46,8 +48,16 @@ def write_research_export_manifest(
         "columns": [str(column) for column in output_frame.columns],
         "input_csv": {
             "path": str(input_path),
-            "sha256": sha256_file(input_path),
-            "size_bytes": input_path.stat().st_size,
+            "sha256": (
+                hashlib.sha256(input_csv_bytes).hexdigest()
+                if input_csv_bytes is not None
+                else sha256_file(input_path)
+            ),
+            "size_bytes": (
+                len(input_csv_bytes)
+                if input_csv_bytes is not None
+                else input_path.stat().st_size
+            ),
         },
         "output_csv": {
             "path": str(output_path),

src/market_signal_sources/cli/export_us_equity_price_proxy_research_csv.py

@@ -2,14 +2,15 @@
 
 import argparse
 from collections.abc import Sequence
+from hashlib import sha256
+from io import BytesIO
 import json
 from pathlib import Path
 import sys
 
 import pandas as pd
 
 from market_signal_sources.artifacts.research_export import write_research_export_manifest
-from market_signal_sources.artifacts.signal_bundle import sha256_file
 from market_signal_sources.derived.us_equity import (
     NASDAQ_SP500_PRICE_PROXY_ARTIFACT_TYPE,
     NASDAQ_SP500_PRICE_PROXY_TRANSFORM,
@@ -22,8 +23,24 @@ def main(argv: Sequence[str] | None = None) -> int:
     args = parser.parse_args(argv)
 
     try:
-        nasdaq100_frame = pd.read_csv(args.fred_nasdaq100_csv)
-        sp500_frame = pd.read_csv(args.fred_sp500_csv)
+        input_paths = (
+            (args.fred_nasdaq100_csv, "fred.nasdaq100"),
+            (args.fred_sp500_csv, "fred.sp500"),
+        )
+        snapshots: dict[Path, bytes] = {}
+        snapshot_paths: list[tuple[Path, Path, bytes, str]] = []
+        for path, source_id in input_paths:
+            resolved_path = path.resolve(strict=True)
+            if resolved_path not in snapshots:
+                snapshot = path.read_bytes()
+                if path.resolve(strict=True) != resolved_path:
+                    raise ValueError(f"input source path changed while reading: {path}")
+                snapshots[resolved_path] = snapshot
+            snapshot_paths.append(
+                (path, resolved_path, snapshots[resolved_path], source_id)
+            )
+        nasdaq100_frame = pd.read_csv(BytesIO(snapshot_paths[0][2]))
+        sp500_frame = pd.read_csv(BytesIO(snapshot_paths[1][2]))
         output = build_nasdaq_sp500_price_proxy_frame(
             fred_nasdaq100_frame=nasdaq100_frame,
             fred_sp500_frame=sp500_frame,
@@ -35,15 +52,22 @@ def main(argv: Sequence[str] | None = None) -> int:
             provider_timestamp=args.provider_timestamp,
             min_history=args.min_history,
         )
-        args.output_csv.parent.mkdir(parents=True, exist_ok=True)
-        output.to_csv(args.output_csv, index=False)
+        for path, resolved_path, _, _ in snapshot_paths:
+            if path.resolve(strict=True) != resolved_path:
+                raise ValueError(f"input source changed during export: {path}")
+        for resolved_path, snapshot in snapshots.items():
+            if resolved_path.read_bytes() != snapshot:
+                raise ValueError(f"input source changed during export: {resolved_path}")
+
         manifest_path = args.manifest_path or args.output_csv.with_suffix(
             ".manifest.json"
         )
-        input_sources = (
-            _input_source_record(args.fred_nasdaq100_csv, source_id="fred.nasdaq100"),
-            _input_source_record(args.fred_sp500_csv, source_id="fred.sp500"),
+        input_sources = tuple(
+            _input_source_record(path, source_id=source_id, input_bytes=snapshot)
+            for path, _, snapshot, source_id in snapshot_paths
         )
+        args.output_csv.parent.mkdir(parents=True, exist_ok=True)
+        output.to_csv(args.output_csv, index=False)
         manifest = write_research_export_manifest(
             manifest_path,
             output_csv_path=args.output_csv,
@@ -55,6 +79,7 @@ def main(argv: Sequence[str] | None = None) -> int:
             as_of=args.as_of,
             min_history=args.min_history,
             input_sources=input_sources,
+            input_csv_bytes=snapshot_paths[0][2],
             transform_parameters={
                 "nasdaq100_date_column": args.nasdaq100_date_column,
                 "nasdaq100_value_column": args.nasdaq100_value_column,
@@ -117,12 +142,17 @@ def _build_parser() -> argparse.ArgumentParser:
     return parser
 
 
-def _input_source_record(path: Path, *, source_id: str) -> dict[str, object]:
+def _input_source_record(
+    path: Path,
+    *,
+    source_id: str,
+    input_bytes: bytes,
+) -> dict[str, object]:
     return {
         "source_id": source_id,
         "path": str(path),
-        "sha256": sha256_file(path),
-        "size_bytes": path.stat().st_size,
+        "sha256": sha256(input_bytes).hexdigest(),
+        "size_bytes": len(input_bytes),
     }
 
 

tests/test_btc_cycle_bundle.py

@@ -105,6 +105,10 @@
 from market_signal_sources.cli.export_us_equity_price_proxy_research_csv import (
     main as export_us_equity_price_proxy_main,
 )
+import market_signal_sources.cli.export_us_equity_price_proxy_research_csv as price_proxy_cli
+from market_signal_sources.artifacts.research_export import (
+    write_research_export_manifest,
+)
 from market_signal_sources.cli.export_us_equity_public_context_research_csv import (
     main as export_us_equity_public_context_main,
 )
@@ -3554,6 +3558,187 @@ def test_cli_exports_us_equity_price_proxy_research_csv(
     assert validation_summary["output_csv_sha256"] == _sha256(output_csv)
 
 
+def test_price_proxy_export_rejects_source_replacement_before_writing(
+    tmp_path,
+    capsys,
+    monkeypatch,
+) -> None:
+    nasdaq100_csv = tmp_path / "fred_nasdaq100.csv"
+    sp500_csv = tmp_path / "fred_sp500.csv"
+    output_csv = tmp_path / "research" / "proxy.csv"
+    manifest_path = tmp_path / "research" / "proxy.manifest.json"
+    nasdaq100_csv.write_text("DATE,NASDAQ100\n2025-01-02,100\n2025-01-03,101\n")
+    sp500_csv.write_text("DATE,SP500\n2025-01-02,50\n2025-01-03,51\n")
+
+    original_builder = price_proxy_cli.build_nasdaq_sp500_price_proxy_frame
+
+    def replace_source_after_parse(**kwargs):
+        nasdaq100_csv.write_text("DATE,NASDAQ100\n2025-01-02,100\n2025-01-03,999\n")
+        return original_builder(**kwargs)
+
+    monkeypatch.setattr(
+        price_proxy_cli,
+        "build_nasdaq_sp500_price_proxy_frame",
+        replace_source_after_parse,
+    )
+    result = export_us_equity_price_proxy_main(
+        [
+            "--fred-nasdaq100-csv",
+            str(nasdaq100_csv),
+            "--fred-sp500-csv",
+            str(sp500_csv),
+            "--output-csv",
+            str(output_csv),
+            "--manifest-path",
+            str(manifest_path),
+        ]
+    )
+
+    assert result == 2
+    assert "changed" in capsys.readouterr().err.lower()
+    assert not output_csv.exists()
+    assert not manifest_path.exists()
+
+
+def test_price_proxy_export_reads_canonical_same_file_once_and_keeps_path_spelling(
+    tmp_path,
+    capsys,
+    monkeypatch,
+) -> None:
+    source_dir = tmp_path / "source"
+    source_dir.mkdir()
+    source_csv = source_dir / "combined.csv"
+    source_csv.write_text(
+        "DATE,NASDAQ100,SP500\n2025-01-02,100,50\n2025-01-03,101,51\n"
+    )
+    source_nested = source_dir / "nested"
+    source_nested.mkdir()
+    source_link = tmp_path / "combined-link.csv"
+    source_link.symlink_to(source_csv)
+    original_spelling = tmp_path / "source" / "nested" / ".." / "combined.csv"
+    output_csv = tmp_path / "research" / "proxy.csv"
+    manifest_path = tmp_path / "research" / "proxy.manifest.json"
+    read_bytes = Path.read_bytes
+    source_reads = 0
+
+    def count_read_bytes(path):
+        nonlocal source_reads
+        if path.resolve() == source_csv.resolve():
+            source_reads += 1
+        return read_bytes(path)
+
+    monkeypatch.setattr(Path, "read_bytes", count_read_bytes)
+    result = export_us_equity_price_proxy_main(
+        [
+            "--fred-nasdaq100-csv",
+            str(original_spelling),
+            "--fred-sp500-csv",
+            str(source_link),
+            "--output-csv",
+            str(output_csv),
+            "--manifest-path",
+            str(manifest_path),
+        ]
+    )
+
+    assert result == 0
+    capsys.readouterr()
+    assert source_reads == 2
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    assert manifest["input_csv"]["path"] == str(original_spelling)
+    assert [record["source_id"] for record in manifest["input_sources"]] == [
+        "fred.nasdaq100",
+        "fred.sp500",
+    ]
+    assert all(record["sha256"] == _sha256(source_csv) for record in manifest["input_sources"])
+
+
+def test_price_proxy_export_rejects_symlink_retarget_before_writing(
+    tmp_path,
+    capsys,
+    monkeypatch,
+) -> None:
+    first_source = tmp_path / "first.csv"
+    replacement_source = tmp_path / "replacement.csv"
+    sp500_csv = tmp_path / "sp500.csv"
+    source_link = tmp_path / "nasdaq.csv"
+    source_bytes = b"DATE,NASDAQ100\n2025-01-02,100\n2025-01-03,101\n"
+    first_source.write_bytes(source_bytes)
+    replacement_source.write_bytes(source_bytes)
+    sp500_csv.write_text("DATE,SP500\n2025-01-02,50\n2025-01-03,51\n")
+    source_link.symlink_to(first_source)
+    output_csv = tmp_path / "research" / "proxy.csv"
+    manifest_path = tmp_path / "research" / "proxy.manifest.json"
+    original_builder = price_proxy_cli.build_nasdaq_sp500_price_proxy_frame
+
+    def retarget_source_after_parse(**kwargs):
+        source_link.unlink()
+        source_link.symlink_to(replacement_source)
+        return original_builder(**kwargs)
+
+    monkeypatch.setattr(
+        price_proxy_cli,
+        "build_nasdaq_sp500_price_proxy_frame",
+        retarget_source_after_parse,
+    )
+    result = export_us_equity_price_proxy_main(
+        [
+            "--fred-nasdaq100-csv",
+            str(source_link),
+            "--fred-sp500-csv",
+            str(sp500_csv),
+            "--output-csv",
+            str(output_csv),
+            "--manifest-path",
+            str(manifest_path),
+        ]
+    )
+
+    assert result == 2
+    assert "changed" in capsys.readouterr().err.lower()
+    assert not output_csv.exists()
+    assert not manifest_path.exists()
+
+
+def test_research_export_manifest_can_bind_consumed_input_bytes(tmp_path) -> None:
+    input_csv = tmp_path / "input.csv"
+    input_csv.write_bytes(b"later file contents")
+    output_csv = tmp_path / "output.csv"
+    output_csv.write_text("date,value\n2025-01-02,1\n")
+    output_frame = pd.DataFrame({"date": ["2025-01-02"], "value": [1]})
+    consumed_bytes = b"bytes actually parsed"
+
+    default_manifest = write_research_export_manifest(
+        tmp_path / "default-manifest.json",
+        output_csv_path=output_csv,
+        output_frame=output_frame,
+        input_csv_path=input_csv,
+        artifact_type="us_equity_price_proxy_research_csv",
+        transform="us_equity.nasdaq_sp500.price_proxy.v1",
+        source_version="test",
+        as_of=None,
+        min_history=1,
+    )
+    assert default_manifest["input_csv"]["sha256"] == _sha256(input_csv)
+    assert default_manifest["input_csv"]["size_bytes"] == input_csv.stat().st_size
+
+    manifest = write_research_export_manifest(
+        tmp_path / "manifest.json",
+        output_csv_path=output_csv,
+        output_frame=output_frame,
+        input_csv_path=input_csv,
+        artifact_type="us_equity_price_proxy_research_csv",
+        transform="us_equity.nasdaq_sp500.price_proxy.v1",
+        source_version="test",
+        as_of=None,
+        min_history=1,
+        input_csv_bytes=consumed_bytes,
+    )
+
+    assert manifest["input_csv"]["sha256"] == hashlib.sha256(consumed_bytes).hexdigest()
+    assert manifest["input_csv"]["size_bytes"] == len(consumed_bytes)
+
+
 def test_public_context_quality_report_rejects_stale_cape_source(
     tmp_path,
     capsys,