diff --git a/src/market_signal_sources/artifacts/research_export.py b/src/market_signal_sources/artifacts/research_export.py index bce46ba..86c9eba 100644 --- a/src/market_signal_sources/artifacts/research_export.py +++ b/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), diff --git a/src/market_signal_sources/cli/export_us_equity_price_proxy_research_csv.py b/src/market_signal_sources/cli/export_us_equity_price_proxy_research_csv.py index 30e2399..4d061b2 100644 --- a/src/market_signal_sources/cli/export_us_equity_price_proxy_research_csv.py +++ b/src/market_signal_sources/cli/export_us_equity_price_proxy_research_csv.py @@ -2,6 +2,8 @@ import argparse from collections.abc import Sequence +from hashlib import sha256 +from io import BytesIO import json from pathlib import Path import sys @@ -9,7 +11,6 @@ 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), } diff --git a/tests/test_btc_cycle_bundle.py b/tests/test_btc_cycle_bundle.py index a299bb0..bf92a0d 100644 --- a/tests/test_btc_cycle_bundle.py +++ b/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,