feat: scope multisource daily assurance by input field

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:

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