diff --git a/src/quant_platform_kit/data/multisource_assurance.py b/src/quant_platform_kit/data/multisource_assurance.py index 4125fc5..d4ed164 100644 --- a/src/quant_platform_kit/data/multisource_assurance.py +++ b/src/quant_platform_kit/data/multisource_assurance.py @@ -33,6 +33,7 @@ _SOURCE_OBSERVATION_STATUSES = frozenset( {SOURCE_OBSERVATION_READY, SOURCE_OBSERVATION_UNAVAILABLE, SOURCE_OBSERVATION_INVALID} ) +_PRICE_FIELDS = frozenset({"open", "high", "low", "close"}) def _canonical_bytes(value: object) -> bytes: @@ -219,6 +220,8 @@ class MultiSourceDailyBarPolicy: minimum_ready_sources: int = 2 price_relative_tolerance: float = 0.0001 volume_relative_tolerance: float = 0.05 + required_price_fields: tuple[str, ...] = ("open", "high", "low", "close") + compare_volume: bool = True def __post_init__(self) -> None: object.__setattr__(self, "scope_id", _require_identifier(self.scope_id, field_name="scope_id")) @@ -253,6 +256,19 @@ def __post_init__(self) -> None: "volume_relative_tolerance", _require_nonnegative_float(self.volume_relative_tolerance, field_name="volume_relative_tolerance"), ) + try: + price_fields = tuple(self.required_price_fields) + except TypeError as exc: + raise ValueError("required_price_fields must be a non-empty sequence") from exc + if ( + not price_fields + or any(not isinstance(field_name, str) or field_name not in _PRICE_FIELDS for field_name in price_fields) + or len(set(price_fields)) != len(price_fields) + ): + raise ValueError("required_price_fields must contain unique OHLC field names") + if not isinstance(self.compare_volume, bool): + raise ValueError("compare_volume must be a boolean") + object.__setattr__(self, "required_price_fields", price_fields) def to_dict(self) -> dict[str, object]: return { @@ -264,6 +280,8 @@ def to_dict(self) -> dict[str, object]: "minimum_ready_sources": self.minimum_ready_sources, "price_relative_tolerance": self.price_relative_tolerance, "volume_relative_tolerance": self.volume_relative_tolerance, + "required_price_fields": list(self.required_price_fields), + "compare_volume": self.compare_volume, } @property @@ -362,8 +380,9 @@ def assess_multisource_daily_bars( A healthy source is never enough by itself. If a configured source is unavailable, malformed, on a different adjustment basis, or disagrees on - sessions/OHLCV, the result is non-publishable. Callers may still retain a - source-specific artifact for diagnostics or shadow research. + sessions, policy-declared price fields, or policy-required volume, the + result is non-publishable. Callers may still retain a source-specific + artifact for diagnostics or shadow research. """ if not isinstance(policy, MultiSourceDailyBarPolicy): @@ -454,14 +473,15 @@ def _compare_snapshots( if any( _relative_delta(getattr(left, field_name), getattr(right, field_name)) > policy.price_relative_tolerance - for field_name in ("open", "high", "low", "close") + for field_name in policy.required_price_fields ): _append_finding(findings, "daily_bar_price_divergence") break - for left, right in zip(baseline.bars, candidate.bars): - if _relative_delta(left.volume, right.volume) > policy.volume_relative_tolerance: - _append_finding(findings, "daily_bar_volume_divergence") - break + if policy.compare_volume: + for left, right in zip(baseline.bars, candidate.bars): + if _relative_delta(left.volume, right.volume) > policy.volume_relative_tolerance: + _append_finding(findings, "daily_bar_volume_divergence") + break def _append_finding(findings: list[str], finding: str) -> None: diff --git a/tests/test_multisource_assurance.py b/tests/test_multisource_assurance.py index 4f46fa4..c288a55 100644 --- a/tests/test_multisource_assurance.py +++ b/tests/test_multisource_assurance.py @@ -25,7 +25,13 @@ def _policy() -> MultiSourceDailyBarPolicy: ) -def _snapshot(source_id: str, *, close: float = 100.5, adjustment_basis: str = "total_return_adjusted") -> DailyBarSourceSnapshot: +def _snapshot( + source_id: str, + *, + close: float = 100.5, + latest_open: float = 100.0, + adjustment_basis: str = "total_return_adjusted", +) -> DailyBarSourceSnapshot: return DailyBarSourceSnapshot( source_id=source_id, symbol="SOXL", @@ -34,7 +40,7 @@ def _snapshot(source_id: str, *, close: float = 100.5, adjustment_basis: str = " source_artifact_sha256=("a" if source_id == "alpaca_sip" else "b") * 64, bars=( DailyBar("2026-08-20", 99.0, 101.0, 98.0, 100.0, 1_000_000), - DailyBar("2026-08-21", 100.0, 102.0, 99.0, close, 1_100_000), + DailyBar("2026-08-21", latest_open, 102.0, 99.0, close, 1_100_000), ), ) @@ -95,6 +101,65 @@ def test_price_divergence_prevents_a_silent_fallback() -> None: assert not result.can_publish_research_input +def test_field_scoped_policy_can_bind_only_explicit_downstream_inputs() -> None: + observations = ( + _ready("twelve_data_eod", adjustment_basis="split_adjusted"), + _ready("yahoo_chart", adjustment_basis="split_adjusted", latest_open=100.04), + ) + close_only_policy = MultiSourceDailyBarPolicy( + scope_id="us-equity:soxl-close-only", + symbol="SOXL", + date_cutoff="2026-08-21", + adjustment_basis="split_adjusted", + required_source_ids=("twelve_data_eod", "yahoo_chart"), + required_price_fields=("close",), + compare_volume=False, + ) + result = assess_multisource_daily_bars( + close_only_policy, + observations, + ) + all_ohlc_policy = MultiSourceDailyBarPolicy( + scope_id="us-equity:soxl-all-ohlc", + symbol="SOXL", + date_cutoff="2026-08-21", + adjustment_basis="split_adjusted", + required_source_ids=("twelve_data_eod", "yahoo_chart"), + ) + + assert result.status == DATA_ASSURANCE_STATUS_VERIFIED + assert result.can_publish_research_input + assert close_only_policy.to_dict()["required_price_fields"] == ["close"] + assert close_only_policy.to_dict()["compare_volume"] is False + assert assess_multisource_daily_bars(all_ohlc_policy, observations).findings == ( + "daily_bar_price_divergence", + ) + + +@pytest.mark.parametrize( + ("required_price_fields", "compare_volume"), + [ + ((), True), + (("close", "close"), True), + (("adjusted_close",), True), + (("close",), "false"), + ], +) +def test_field_scoped_policy_rejects_ambiguous_or_invalid_scope( + required_price_fields: tuple[str, ...], compare_volume: object +) -> None: + with pytest.raises(ValueError): + MultiSourceDailyBarPolicy( + scope_id="us-equity:soxl-daily", + symbol="SOXL", + date_cutoff="2026-08-21", + adjustment_basis="total_return_adjusted", + required_source_ids=("alpaca_sip", "twelve_data_eod"), + required_price_fields=required_price_fields, + compare_volume=compare_volume, + ) + + def test_adjustment_basis_mismatch_is_not_merged() -> None: result = assess_multisource_daily_bars( _policy(),