fix(freshness): bind local CSV freshness to source available time

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),
     }

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(

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"),
     )

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")