From 88411e53635773f58daf70d35293816cff203379 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 6 Sep 2026 18:17:31 +0800 Subject: [PATCH] fix(freshness): bind local CSV freshness to source available time Stop treating historical local bars as wall-clock fresh; validate provider timestamps against an explicit evaluation/reference clock. Co-Authored-By: Claude --- .../artifacts/quality_report.py | 5 + .../artifacts/validation.py | 97 ++++++++++++- .../providers/local_csv.py | 11 +- tests/test_btc_cycle_bundle.py | 130 ++++++++++++++++++ 4 files changed, 238 insertions(+), 5 deletions(-) diff --git a/src/market_signal_sources/artifacts/quality_report.py b/src/market_signal_sources/artifacts/quality_report.py index a15ca8b..f14d4e6 100644 --- a/src/market_signal_sources/artifacts/quality_report.py +++ b/src/market_signal_sources/artifacts/quality_report.py @@ -207,6 +207,7 @@ def validate_ohlcv_quality_report( "duplicate_date_count", "first_date", "last_date", + "source_available_at", "max_gap_days", "gap_count_above_threshold", ): @@ -297,6 +298,7 @@ def validate_ohlcv_quality_report_file( "dropped_row_count": report["dropped_row_count"], "first_date": report["first_date"], "last_date": report["last_date"], + "source_available_at": report.get("source_available_at", ""), "max_gap_days": report["max_gap_days"], "gap_count_above_threshold": report["gap_count_above_threshold"], } @@ -353,6 +355,9 @@ def _quality_report_payload( "duplicate_date_count": int(duplicate_date_count), "first_date": first_date, "last_date": last_date, + "source_available_at": ( + f"{last_date}T00:00:00Z" if last_date else "" + ), "max_gap_days": int(max_gap_days), "gap_count_above_threshold": int(gap_count_above_threshold), } diff --git a/src/market_signal_sources/artifacts/validation.py b/src/market_signal_sources/artifacts/validation.py index 58df0be..51c638d 100644 --- a/src/market_signal_sources/artifacts/validation.py +++ b/src/market_signal_sources/artifacts/validation.py @@ -1,5 +1,6 @@ from __future__ import annotations +from datetime import datetime, timedelta, timezone import csv from collections.abc import Iterable, Mapping, Sequence import json @@ -61,6 +62,8 @@ def validate_signal_bundle( *, expected_canonical_input: str = CANONICAL_INPUT_DERIVED_INDICATORS, accepted_freshness_statuses: Iterable[str] = (FRESHNESS_FRESH,), + now: str | None = None, + reference_time: str | None = None, ) -> None: """Validate the producer-side market_signal_bundle.v1 contract.""" @@ -82,7 +85,12 @@ def validate_signal_bundle( "signal bundle canonical_input mismatch: " f"expected {expected_canonical_input!r}, got {canonical_input!r}" ) - _validate_freshness(bundle, accepted_freshness_statuses=accepted_freshness_statuses) + _validate_freshness( + bundle, + accepted_freshness_statuses=accepted_freshness_statuses, + now=now, + reference_time=reference_time, + ) _validate_derived_indicators(bundle, canonical_input=canonical_input) _validate_provenance(bundle) @@ -92,6 +100,8 @@ def validate_signal_bundle_manifest( *, expected_canonical_input: str = CANONICAL_INPUT_DERIVED_INDICATORS, accepted_freshness_statuses: Iterable[str] = (FRESHNESS_FRESH,), + now: str | None = None, + reference_time: str | None = None, ) -> dict[str, Any]: """Validate a manifest-referenced signal bundle and return audit summary.""" @@ -116,6 +126,8 @@ def validate_signal_bundle_manifest( bundle, expected_canonical_input=expected_canonical_input, accepted_freshness_statuses=accepted_freshness_statuses, + now=now, + reference_time=reference_time, ) quality_report_summary = _validate_optional_quality_report_reference( manifest, @@ -618,6 +630,17 @@ def _validate_quality_report_bundle_consistency( "signal bundle provenance.raw_artifact_sha256: " f"expected {bundle_raw_sha256}, got {quality_input_sha256}" ) + freshness = bundle.get("freshness") + if not isinstance(freshness, Mapping): + raise SignalBundleValidationError("freshness must be a mapping") + provider_timestamp = str(freshness.get("provider_timestamp", "")).strip() + quality_last_date = str(quality_report.get("last_date", "")).strip() + provider_date = _provider_timestamp_date(provider_timestamp) + if quality_last_date and provider_date and quality_last_date != provider_date: + raise SignalBundleValidationError( + "quality report last_date mismatch with freshness.provider_timestamp: " + f"{quality_last_date!r} != {provider_date!r}" + ) def _validate_quality_report(report: Mapping[str, Any]) -> None: @@ -1051,6 +1074,8 @@ def _validate_freshness( bundle: Mapping[str, Any], *, accepted_freshness_statuses: Iterable[str], + now: str | None = None, + reference_time: str | None = None, ) -> None: freshness = bundle.get("freshness") if not isinstance(freshness, Mapping): @@ -1059,13 +1084,81 @@ def _validate_freshness( if not isinstance(status, str) or not status.strip(): raise SignalBundleValidationError("freshness.status must be a non-empty string") accepted = {str(item).strip().lower() for item in accepted_freshness_statuses} - if status.strip().lower() not in accepted: + normalized_status = status.strip().lower() + if normalized_status not in accepted: raise SignalBundleValidationError(f"unacceptable freshness.status: {status!r}") provider_timestamp = freshness.get("provider_timestamp") if not isinstance(provider_timestamp, str) or not provider_timestamp.strip(): raise SignalBundleValidationError( "freshness.provider_timestamp must be a non-empty string" ) + max_age_hours = freshness.get("max_age_hours") + if not isinstance(max_age_hours, int) or isinstance(max_age_hours, bool) or max_age_hours < 0: + raise SignalBundleValidationError( + "freshness.max_age_hours must be a non-negative integer" + ) + provider_time = _parse_utc_timestamp( + provider_timestamp, + field="freshness.provider_timestamp", + ) + evaluation_time = _resolve_evaluation_time( + bundle, + now=now, + reference_time=reference_time, + ) + if provider_time > evaluation_time: + raise SignalBundleValidationError( + "freshness.provider_timestamp is in the future relative to evaluation time: " + f"{provider_timestamp!r}" + ) + if normalized_status == FRESHNESS_FRESH: + age = evaluation_time - provider_time + if age > timedelta(hours=max_age_hours): + raise SignalBundleValidationError( + "freshness.status claims fresh but provider_timestamp exceeds " + f"max_age_hours={max_age_hours}: age_hours={age.total_seconds() / 3600.0}" + ) + + +def _provider_timestamp_date(provider_timestamp: str) -> str: + if not str(provider_timestamp or "").strip(): + return "" + return _parse_utc_timestamp( + provider_timestamp, + field="freshness.provider_timestamp", + ).date().isoformat() + + +def _resolve_evaluation_time( + bundle: Mapping[str, Any], + *, + now: str | None, + reference_time: str | None, +) -> datetime: + if reference_time is not None: + return _parse_utc_timestamp(reference_time, field="reference_time") + if now is not None: + return _parse_utc_timestamp(now, field="now") + generated_at = bundle.get("generated_at") + if isinstance(generated_at, str) and generated_at.strip(): + return _parse_utc_timestamp(generated_at, field="generated_at") + return datetime.now(timezone.utc) + + +def _parse_utc_timestamp(value: str, *, field: str) -> datetime: + raw = str(value or "").strip() + if not raw: + raise SignalBundleValidationError(f"{field} must be a non-empty timestamp") + normalized = raw.replace("Z", "+00:00") + try: + parsed = datetime.fromisoformat(normalized) + except ValueError as exc: + raise SignalBundleValidationError( + f"{field} must be an ISO-8601 timestamp: {value!r}" + ) from exc + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) def _validate_derived_indicators( diff --git a/src/market_signal_sources/providers/local_csv.py b/src/market_signal_sources/providers/local_csv.py index 21ef57c..50c03c5 100644 --- a/src/market_signal_sources/providers/local_csv.py +++ b/src/market_signal_sources/providers/local_csv.py @@ -26,13 +26,18 @@ def local_csv_provider_metadata( provider_dataset: str = "btc_usd_daily_ohlcv", license_scope: str = "internal_runtime", ) -> LocalCsvProviderMetadata: - """Return auditable provider metadata for a local CSV source artifact.""" + """Return auditable provider metadata for a local CSV source artifact. - normalized_as_of = pd.Timestamp(as_of).normalize().date().isoformat() + ``provider_timestamp`` binds to the actual source end date present in the + CSV (after ``as_of`` filtering), not to the caller-supplied ``as_of`` clock. + """ + + frame = load_ohlcv_csv(path, as_of=as_of) + source_end = pd.Timestamp(frame.iloc[-1]["date"]).normalize().date().isoformat() return LocalCsvProviderMetadata( provider=_non_empty(provider, "provider"), provider_dataset=_non_empty(provider_dataset, "provider_dataset"), - provider_timestamp=f"{normalized_as_of}T00:00:00Z", + provider_timestamp=f"{source_end}T00:00:00Z", raw_artifact_sha256=_sha256_file(Path(path)), license_scope=_non_empty(license_scope, "license_scope"), ) diff --git a/tests/test_btc_cycle_bundle.py b/tests/test_btc_cycle_bundle.py index d155127..43ea9d6 100644 --- a/tests/test_btc_cycle_bundle.py +++ b/tests/test_btc_cycle_bundle.py @@ -3966,3 +3966,133 @@ def test_validator_rejects_index_compatible_profile_mismatch(tmp_path) -> None: with pytest.raises(SignalBundleValidationError, match="compatible_profiles"): validate_signal_bundle_index(paths["index"]) + + +def test_local_csv_provider_timestamp_binds_source_end_not_as_of(tmp_path) -> None: + input_csv = tmp_path / "btc.csv" + frame = _btc_frame(rows=220) + # Source ends well before a later wall-clock as_of. + frame.to_csv(input_csv, index=False) + source_end = pd.Timestamp(frame.iloc[-1]["date"]).date().isoformat() + + metadata = local_csv_provider_metadata( + input_csv, + as_of="2026-09-06", + provider="local_csv", + provider_dataset="btc_usd_daily_ohlcv", + ) + quality = build_ohlcv_quality_report(input_csv, as_of="2026-09-06") + + assert source_end != "2026-09-06" + assert metadata.provider_timestamp == f"{source_end}T00:00:00Z" + assert metadata.provider_timestamp == f"{quality['last_date']}T00:00:00Z" + assert quality["source_available_at"] == f"{source_end}T00:00:00Z" + + +def test_stale_complete_local_csv_cannot_claim_fresh_against_wall_clock(tmp_path) -> None: + input_csv = tmp_path / "btc.csv" + _btc_frame().to_csv(input_csv, index=False) + metadata = local_csv_provider_metadata(input_csv, as_of="2026-09-06") + bundle = build_btc_cycle_signal_bundle( + _btc_frame(), + as_of="2025-09-17", + raw_artifact_sha256=metadata.raw_artifact_sha256, + generated_at="2026-09-06T12:00:00Z", + provider_timestamp=metadata.provider_timestamp, + freshness_status="fresh", + ) + + with pytest.raises(SignalBundleValidationError, match="freshness"): + validate_signal_bundle(bundle, now="2026-09-06T12:00:00Z") + + +def test_fresh_provider_timestamp_within_max_age_is_accepted() -> None: + bundle = build_btc_cycle_signal_bundle( + _btc_frame(), + as_of="2025-09-17", + raw_artifact_sha256="0" * 64, + generated_at="2025-09-17T12:00:00Z", + provider_timestamp="2025-09-17T00:00:00Z", + freshness_status="fresh", + ) + validate_signal_bundle(bundle, now="2025-09-17T12:00:00Z") + + +@pytest.mark.parametrize( + ("now", "should_pass"), + [ + ("2025-09-18T12:00:00Z", True), # exactly 36h + ("2025-09-18T12:00:01Z", False), # just over 36h + ], +) +def test_freshness_max_age_hours_boundary(now: str, should_pass: bool) -> None: + bundle = build_btc_cycle_signal_bundle( + _btc_frame(), + as_of="2025-09-17", + raw_artifact_sha256="0" * 64, + generated_at="2025-09-17T00:15:00Z", + provider_timestamp="2025-09-17T00:00:00Z", + freshness_status="fresh", + ) + if should_pass: + validate_signal_bundle(bundle, now=now) + else: + with pytest.raises(SignalBundleValidationError, match="freshness"): + validate_signal_bundle(bundle, now=now) + + +def test_future_provider_timestamp_is_rejected() -> None: + bundle = build_btc_cycle_signal_bundle( + _btc_frame(), + as_of="2025-09-17", + raw_artifact_sha256="0" * 64, + generated_at="2025-09-17T00:15:00Z", + provider_timestamp="2025-09-18T00:00:00Z", + freshness_status="fresh", + ) + with pytest.raises(SignalBundleValidationError, match="provider_timestamp"): + validate_signal_bundle(bundle, now="2025-09-17T12:00:00Z") + + +def test_provider_timestamp_quality_last_date_conflict_is_rejected(tmp_path) -> None: + input_csv = tmp_path / "btc.csv" + _btc_frame().to_csv(input_csv, index=False) + quality_report_path = tmp_path / "quality_report.json" + write_ohlcv_quality_report( + quality_report_path, + input_csv, + as_of="2025-09-17", + ) + bundle = build_btc_cycle_signal_bundle( + _btc_frame(), + as_of="2025-09-17", + raw_artifact_sha256=_sha256(input_csv), + generated_at="2025-09-17T00:15:00Z", + provider_timestamp="2025-09-16T00:00:00Z", + freshness_status="fresh", + ) + with pytest.raises(SignalBundleValidationError, match="last_date"): + write_signal_bundle_artifacts( + tmp_path, + bundle, + quality_report_path=quality_report_path, + ) + + +def test_historical_reference_replay_requires_reference_time_not_wall_clock() -> None: + bundle = build_btc_cycle_signal_bundle( + _btc_frame(), + as_of="2025-09-17", + raw_artifact_sha256="0" * 64, + generated_at="2025-09-17T00:15:00Z", + provider_timestamp="2025-09-17T00:00:00Z", + freshness_status="fresh", + ) + # Historical reference replay can be fresh relative to the decision clock. + validate_signal_bundle( + bundle, + reference_time="2025-09-17T12:00:00Z", + ) + # The same artifact is not fresh when judged against a later wall clock. + with pytest.raises(SignalBundleValidationError, match="freshness"): + validate_signal_bundle(bundle, now="2026-09-06T12:00:00Z")