refactor: enforce market data quality contracts

backend/data/__init__.py

@@ -1,4 +1,14 @@
 from .gateway import DataGateway, build_data_gateway
 from .policy import DataPolicyError, DataSourcePolicy
+from .quality import DataQualityError, DataQualityGate, QualityEvidence, QualityReport
 
-__all__ = ["DataGateway", "DataPolicyError", "DataSourcePolicy", "build_data_gateway"]
+__all__ = [
+    "DataGateway",
+    "DataPolicyError",
+    "DataQualityError",
+    "DataQualityGate",
+    "DataSourcePolicy",
+    "QualityEvidence",
+    "QualityReport",
+    "build_data_gateway",
+]

backend/data/gateway.py

@@ -2,10 +2,12 @@ from __future__ import annotations
 
 from collections.abc import Callable
 from dataclasses import dataclass
+from datetime import datetime
 
 from backend.data.contracts import DataUsage
 from backend.data.policy import DataSourcePolicy
 from backend.data.providers import IfindProvider, TushareProvider
+from backend.data.quality import DataQualityGate, QualityEvidence, QualityReport
 from chart_data_provider import EastmoneyChartClient, MarketChartClient
 from ifind_client import IfindHttpClient
 from realtime_aggregator import WebRealtimeAggregator
@@ -15,6 +17,7 @@ from tushare_client import TushareClient
 @dataclass(frozen=True)
 class DataGateway:
     policy: DataSourcePolicy
+    quality: DataQualityGate
     tushare_provider: TushareProvider
     ifind_provider: IfindProvider
     chart_data: MarketChartClient
@@ -36,6 +39,27 @@ class DataGateway:
     def assert_source(self, dataset_id: str, provider_id: str, usage: DataUsage) -> None:
         self.policy.assert_allowed(dataset_id, provider_id, usage)
 
+    def provider_chain(self, dataset_id: str, usage: DataUsage) -> tuple[str, ...]:
+        dataset = self.policy.dataset(dataset_id)
+        allowed = []
+        for provider_id in dataset.providers:
+            try:
+                self.policy.assert_allowed(dataset_id, provider_id, usage)
+            except Exception:
+                continue
+            allowed.append(provider_id)
+        if not allowed:
+            raise RuntimeError(f"No permitted provider for {dataset_id} ({usage})")
+        return tuple(allowed)
+
+    def require_quality(
+        self,
+        evidence: QualityEvidence,
+        usage: DataUsage,
+        as_of: str | datetime | None = None,
+    ) -> QualityReport:
+        return self.quality.require(evidence, usage, as_of)
+
 
 def build_data_gateway(
     credentials: dict[str, object],
@@ -48,8 +72,10 @@ def build_data_gateway(
     token_supplier = tushare_token_supplier or (
         lambda: str(credentials.get("tushare_token") or "")
     )
+    policy = DataSourcePolicy.load()
     return DataGateway(
-        policy=DataSourcePolicy.load(),
+        policy=policy,
+        quality=DataQualityGate.load(policy),
         tushare_provider=TushareProvider(token_supplier),
         ifind_provider=IfindProvider(ifind),
         chart_data=MarketChartClient(ifind, EastmoneyChartClient()),

backend/data/quality.py

@@ -0,0 +1,202 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from datetime import date, datetime, time, timedelta, timezone
+from pathlib import Path
+from typing import Any
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
+
+from app_config import APP_DIR
+from backend.data.contracts import DataUsage
+from backend.data.policy import DataPolicyError, DataSourcePolicy
+
+
+class DataQualityError(RuntimeError):
+    pass
+
+
+def market_timezone(name: str = "Asia/Shanghai"):
+    try:
+        return ZoneInfo(name)
+    except ZoneInfoNotFoundError:
+        if name != "Asia/Shanghai":
+            raise
+        return timezone(timedelta(hours=8), name)
+
+
+@dataclass(frozen=True)
+class QualityEvidence:
+    dataset_id: str
+    provider_id: str
+    data_time: str | datetime
+    observed_at: str | datetime
+    actual_count: int | None = None
+    expected_count: int | None = None
+    units: dict[str, str] | None = None
+    adjustment: str = ""
+    available_at: str | datetime | None = None
+
+
+@dataclass(frozen=True)
+class QualityReport:
+    accepted: bool
+    dataset_id: str
+    provider_id: str
+    usage: DataUsage
+    coverage_ratio: float | None
+    age_seconds: float
+    issues: tuple[str, ...]
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "accepted": self.accepted,
+            "dataset_id": self.dataset_id,
+            "provider_id": self.provider_id,
+            "usage": self.usage,
+            "coverage_ratio": self.coverage_ratio,
+            "age_seconds": round(self.age_seconds, 3),
+            "issues": list(self.issues),
+        }
+
+
+class DataQualityGate:
+    def __init__(
+        self,
+        source_policy: DataSourcePolicy,
+        payload: dict[str, Any],
+    ) -> None:
+        self.source_policy = source_policy
+        self.timezone = market_timezone(
+            str(payload.get("timezone") or "Asia/Shanghai")
+        )
+        self.defaults = dict(payload.get("defaults") or {})
+        self.unit_profiles = dict(payload.get("unit_profiles") or {})
+        self.rules = dict(payload.get("datasets") or {})
+
+    @classmethod
+    def load(
+        cls,
+        source_policy: DataSourcePolicy,
+        path: Path | None = None,
+    ) -> "DataQualityGate":
+        config_path = path or APP_DIR / "config" / "data-quality.config.json"
+        payload = json.loads(config_path.read_text(encoding="utf-8"))
+        return cls(source_policy, payload)
+
+    def evaluate(
+        self,
+        evidence: QualityEvidence,
+        usage: DataUsage,
+        as_of: str | datetime | None = None,
+    ) -> QualityReport:
+        issues: list[str] = []
+        try:
+            self.source_policy.assert_allowed(
+                evidence.dataset_id, evidence.provider_id, usage
+            )
+        except DataPolicyError as exc:
+            issues.append(str(exc))
+
+        rule = self.rules.get(evidence.dataset_id)
+        if rule is None:
+            issues.append(f"Missing quality rule: {evidence.dataset_id}")
+            rule = {}
+        if rule.get("blocked"):
+            issues.append(f"Dataset quality is blocked: {evidence.dataset_id}")
+
+        reference = self._datetime(as_of or datetime.now(self.timezone))
+        data_time = self._datetime(evidence.data_time)
+        observed_at = self._datetime(evidence.observed_at)
+        tolerance = float(
+            (self.defaults.get(usage) or {}).get("future_tolerance_seconds") or 0
+        )
+        if data_time > reference + timedelta(seconds=tolerance):
+            issues.append("Data time is later than the evaluation time")
+        if observed_at > reference + timedelta(seconds=tolerance):
+            issues.append("Observation time is later than the evaluation time")
+        if observed_at < data_time:
+            issues.append("Observation time precedes data time")
+
+        age_seconds = max(0.0, (reference - data_time).total_seconds())
+        freshness = rule.get("freshness_seconds")
+        if freshness is not None and age_seconds > float(freshness):
+            issues.append(
+                f"Data is stale: {age_seconds:.1f}s exceeds {float(freshness):.1f}s"
+            )
+
+        coverage_ratio: float | None = None
+        if evidence.expected_count is not None:
+            if evidence.expected_count <= 0:
+                issues.append("Expected count must be positive")
+            elif evidence.actual_count is None or evidence.actual_count < 0:
+                issues.append("Actual count is missing or invalid")
+            else:
+                coverage_ratio = min(1.0, evidence.actual_count / evidence.expected_count)
+                minimum = float(rule.get("min_coverage_ratio") or 0)
+                if coverage_ratio < minimum:
+                    issues.append(
+                        f"Coverage {coverage_ratio:.3f} is below {minimum:.3f}"
+                    )
+
+        required_adjustment = str(rule.get("adjustment") or "")
+        if required_adjustment and evidence.adjustment != required_adjustment:
+            issues.append(
+                f"Adjustment {evidence.adjustment or 'missing'} does not match {required_adjustment}"
+            )
+
+        profile_id = str(rule.get("unit_profile") or "none")
+        required_units = dict(self.unit_profiles.get(profile_id) or {})
+        supplied_units = evidence.units or {}
+        for field, expected_unit in required_units.items():
+            actual_unit = supplied_units.get(field)
+            if actual_unit != expected_unit:
+                issues.append(
+                    f"Unit for {field} is {actual_unit or 'missing'}, expected {expected_unit}"
+                )
+
+        if rule.get("point_in_time") == "announcement_date" and usage == "calculation":
+            if evidence.available_at is None:
+                issues.append("Point-in-time availability is missing")
+            elif self._datetime(evidence.available_at) > reference:
+                issues.append("Point-in-time data was not available at evaluation time")
+
+        return QualityReport(
+            accepted=not issues,
+            dataset_id=evidence.dataset_id,
+            provider_id=evidence.provider_id,
+            usage=usage,
+            coverage_ratio=coverage_ratio,
+            age_seconds=age_seconds,
+            issues=tuple(issues),
+        )
+
+    def require(
+        self,
+        evidence: QualityEvidence,
+        usage: DataUsage,
+        as_of: str | datetime | None = None,
+    ) -> QualityReport:
+        report = self.evaluate(evidence, usage, as_of)
+        if not report.accepted:
+            raise DataQualityError("; ".join(report.issues))
+        return report
+
+    def _datetime(self, value: str | datetime) -> datetime:
+        if isinstance(value, datetime):
+            parsed = value
+        else:
+            text = str(value or "").strip()
+            if not text:
+                raise DataQualityError("Quality evidence timestamp is missing")
+            try:
+                parsed = datetime.fromisoformat(text)
+            except ValueError:
+                try:
+                    day = date.fromisoformat(text)
+                except ValueError as exc:
+                    raise DataQualityError(f"Invalid quality timestamp: {text}") from exc
+                parsed = datetime.combine(day, time.min)
+        if parsed.tzinfo is None:
+            return parsed.replace(tzinfo=self.timezone)
+        return parsed.astimezone(self.timezone)

config/README.md

@@ -9,6 +9,8 @@ These registries describe the approved product surface during architecture migra
   assigned to a feature owner.
 - `data-fields.config.json`: canonical data products, provider eligibility, intended use, and
   known blocked datasets.
+- `data-quality.config.json`: freshness, coverage, units, adjustment, point-in-time, and
+  fail-closed rules for every canonical data product.
 
 During Stage 04 these files are contract inputs, not runtime replacements. Backend access in
 `api_access.py` remains authoritative until the HTTP governance phase switches it atomically.

config/data-quality.config.json

@@ -0,0 +1,45 @@
+{
+  "schema_version": 1,
+  "timezone": "Asia/Shanghai",
+  "defaults": {
+    "calculation": {"missing_policy": "fail_closed", "provenance_required": true, "future_tolerance_seconds": 5},
+    "display": {"missing_policy": "unavailable", "provenance_required": true, "future_tolerance_seconds": 5}
+  },
+  "unit_profiles": {
+    "none": {},
+    "calendar": {"trade_date": "date", "is_open": "boolean", "previous_open_date": "date"},
+    "master": {"list_date": "date"},
+    "daily_ohlcv": {"open": "CNY/share", "high": "CNY/share", "low": "CNY/share", "close": "CNY/share", "pct_chg": "percent", "volume_shares": "share", "amount_yuan": "CNY"},
+    "valuation": {"turnover_rate_pct": "percent", "volume_ratio": "ratio", "total_mv_10k_yuan": "10k CNY", "circ_mv_10k_yuan": "10k CNY", "pe_ttm": "ratio", "pb": "ratio", "ps_ttm": "ratio", "dv_ttm_pct": "percent"},
+    "fundamental": {"roe_pct": "percent", "roa_pct": "percent", "roic_pct": "percent", "gross_margin_pct": "percent", "net_profit_yoy_pct": "percent", "revenue_yoy_pct": "percent", "operating_cashflow_quality": "ratio"},
+    "moneyflow": {"small_net_yuan": "CNY", "medium_net_yuan": "CNY", "large_net_yuan": "CNY", "extra_large_net_yuan": "CNY", "total_net_yuan": "CNY"},
+    "industry": {"pct_chg": "percent", "turnover_rate_pct": "percent"},
+    "limit_event": {"first_time": "datetime", "last_time": "datetime", "open_times": "count", "consecutive_boards": "count"},
+    "auction": {"price": "CNY/share", "volume_shares": "share", "amount_yuan": "CNY", "pre_close": "CNY/share", "turnover_rate_pct": "percent", "volume_ratio": "ratio", "float_share": "share"},
+    "popularity": {"ths_rank": "rank", "dc_rank": "rank", "rank_change": "rank", "dual_source": "boolean"},
+    "dragon_tiger": {"buy_yuan": "CNY", "sell_yuan": "CNY", "net_buy_yuan": "CNY"},
+    "intraday": {"quote_time": "datetime", "open": "CNY/share", "high": "CNY/share", "low": "CNY/share", "close": "CNY/share", "avg_price": "CNY/share", "volume_shares": "share", "amount_yuan": "CNY"},
+    "realtime_index": {"quote_time": "datetime", "price": "CNY", "pct_chg": "percent", "amount_yuan": "CNY"},
+    "sentiment": {"temperature": "score", "confidence": "percent"}
+  },
+  "datasets": {
+    "market.trade_calendar": {"unit_profile": "calendar", "min_coverage_ratio": 1.0},
+    "market.stock_master": {"unit_profile": "master", "min_coverage_ratio": 0.98},
+    "market.stock_daily": {"unit_profile": "daily_ohlcv", "min_coverage_ratio": 0.98, "adjustment": "current-unadjusted"},
+    "market.daily_valuation": {"unit_profile": "valuation", "min_coverage_ratio": 0.95},
+    "market.fundamentals": {"unit_profile": "fundamental", "min_coverage_ratio": 0.90, "point_in_time": "announcement_date"},
+    "market.moneyflow": {"unit_profile": "moneyflow", "min_coverage_ratio": 0.90},
+    "market.industry_sw": {"unit_profile": "industry", "min_coverage_ratio": 0.95},
+    "market.limit_events": {"unit_profile": "limit_event", "min_coverage_ratio": 1.0},
+    "market.auction_close": {"unit_profile": "auction", "min_coverage_ratio": 0.90},
+    "market.auction_dynamic": {"unit_profile": "auction", "min_coverage_ratio": 0.80, "freshness_seconds": 10},
+    "market.popularity": {"unit_profile": "popularity", "min_coverage_ratio": 0.95},
+    "market.dragon_tiger": {"unit_profile": "dragon_tiger", "min_coverage_ratio": 0.95},
+    "chart.stock_daily": {"unit_profile": "daily_ohlcv", "min_coverage_ratio": 1.0, "adjustment": "forward1"},
+    "chart.intraday": {"unit_profile": "intraday", "min_coverage_ratio": 1.0, "freshness_seconds": 30},
+    "observation.realtime_indices": {"unit_profile": "realtime_index", "min_coverage_ratio": 1.0, "freshness_seconds": 90},
+    "derived.sentiment": {"unit_profile": "sentiment", "min_coverage_ratio": 1.0},
+    "research.consensus": {"unit_profile": "none", "min_coverage_ratio": 0.0, "blocked": true},
+    "market.level2": {"unit_profile": "none", "min_coverage_ratio": 0.0, "blocked": true}
+  }
+}

docs/governance/stage-07-data-quality.md

@@ -0,0 +1,32 @@
+# Stage 07: Data Quality and Provenance Gates
+
+Date: 2026-07-29
+
+## Result
+
+- Added a quality rule for every registered data product.
+- Added canonical Asia/Shanghai timestamp handling.
+- Added source and usage verification before quality acceptance.
+- Added maximum-age checks for realtime auction, intraday charts, and realtime indices.
+- Added minimum coverage thresholds for deterministic datasets.
+- Added unit profiles for prices, shares, currency, percentages, ratios, ranks, and timestamps.
+- Added adjustment checks that distinguish current unadjusted deterministic bars from iFinD
+  forward-adjusted display charts.
+- Added announcement-date point-in-time checks for financial data.
+- Added explicit provider chains; a display fallback cannot silently become a calculation
+  source.
+- Analyst consensus and Level-2 remain blocked until a qualified provider is registered.
+
+## Fail-Closed Contract
+
+A calculation evidence envelope is rejected when its source is unauthorized, its dataset is
+blocked, its timestamp is from the future, its realtime data is stale, its coverage is below
+the registered threshold, its units or adjustment mode differ, or its financial record was not
+available at the evaluation time.
+
+## Compatibility
+
+Existing legacy provider response shapes remain unchanged in this stage. New governed feature
+paths must submit quality evidence through `DataGateway.require_quality`. Existing feature
+paths receive envelopes as they migrate behind domain services, avoiding a simultaneous
+rewrite of calculations and provider routing.

tests/test_data_gateway.py

@@ -1,8 +1,16 @@
 from __future__ import annotations
 
 import unittest
+from datetime import datetime, timedelta
 
-from backend.data import DataPolicyError, DataSourcePolicy, build_data_gateway
+from backend.data import (
+    DataPolicyError,
+    DataQualityError,
+    DataSourcePolicy,
+    QualityEvidence,
+    build_data_gateway,
+)
+from backend.data.quality import market_timezone
 
 
 class DataGatewayTests(unittest.TestCase):
@@ -43,6 +51,101 @@ class DataGatewayTests(unittest.TestCase):
         self.assertEqual(source.count("TushareClient(self.token)"), 1)
         self.assertIn("return gateway.tushare()", source)
 
+    def test_quality_gate_accepts_matching_daily_evidence(self) -> None:
+        timezone = market_timezone()
+        now = datetime(2026, 7, 29, 16, 0, tzinfo=timezone)
+        gateway = build_data_gateway({})
+        report = gateway.require_quality(
+            QualityEvidence(
+                dataset_id="market.stock_daily",
+                provider_id="tushare",
+                data_time="2026-07-29",
+                observed_at=now,
+                actual_count=5000,
+                expected_count=5000,
+                adjustment="current-unadjusted",
+                units={
+                    "open": "CNY/share", "high": "CNY/share", "low": "CNY/share",
+                    "close": "CNY/share", "pct_chg": "percent",
+                    "volume_shares": "share", "amount_yuan": "CNY",
+                },
+            ),
+            "calculation",
+            now,
+        )
+        self.assertTrue(report.accepted)
+        self.assertEqual(report.coverage_ratio, 1.0)
+
+    def test_quality_gate_rejects_stale_dynamic_auction(self) -> None:
+        timezone = market_timezone()
+        now = datetime(2026, 7, 29, 9, 24, tzinfo=timezone)
+        gateway = build_data_gateway({})
+        with self.assertRaises(DataQualityError):
+            gateway.require_quality(
+                QualityEvidence(
+                    dataset_id="market.auction_dynamic",
+                    provider_id="ifind",
+                    data_time=now - timedelta(seconds=30),
+                    observed_at=now - timedelta(seconds=29),
+                    units={
+                        "price": "CNY/share", "volume_shares": "share",
+                        "amount_yuan": "CNY", "pre_close": "CNY/share",
+                        "turnover_rate_pct": "percent", "volume_ratio": "ratio",
+                        "float_share": "share",
+                    },
+                ),
+                "calculation",
+                now,
+            )
+
+    def test_quality_gate_rejects_low_coverage_and_wrong_adjustment(self) -> None:
+        timezone = market_timezone()
+        now = datetime(2026, 7, 29, 16, 0, tzinfo=timezone)
+        gateway = build_data_gateway({})
+        report = gateway.quality.evaluate(
+            QualityEvidence(
+                dataset_id="market.stock_daily",
+                provider_id="tushare",
+                data_time="2026-07-29",
+                observed_at=now,
+                actual_count=4000,
+                expected_count=5000,
+                adjustment="forward1",
+            ),
+            "calculation",
+            now,
+        )
+        self.assertFalse(report.accepted)
+        self.assertTrue(any("Coverage" in issue for issue in report.issues))
+        self.assertTrue(any("Adjustment" in issue for issue in report.issues))
+
+    def test_quality_gate_enforces_financial_point_in_time(self) -> None:
+        timezone = market_timezone()
+        now = datetime(2026, 7, 29, 16, 0, tzinfo=timezone)
+        gateway = build_data_gateway({})
+        report = gateway.quality.evaluate(
+            QualityEvidence(
+                dataset_id="market.fundamentals",
+                provider_id="tushare",
+                data_time="2026-06-30",
+                observed_at=now,
+                available_at="2026-08-15",
+            ),
+            "calculation",
+            now,
+        )
+        self.assertFalse(report.accepted)
+        self.assertTrue(any("not available" in issue for issue in report.issues))
+
+    def test_provider_chain_never_silently_promotes_display_fallback(self) -> None:
+        gateway = build_data_gateway({})
+        self.assertEqual(
+            gateway.provider_chain("chart.intraday", "display"),
+            ("ifind", "eastmoney"),
+        )
+        with self.assertRaises(RuntimeError):
+            gateway.provider_chain("chart.intraday", "calculation")
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_governance_registries.py

@@ -22,6 +22,7 @@ class GovernanceRegistryTests(unittest.TestCase):
         self.pages = load("pages.config.json")
         self.api = load("api.config.json")
         self.data = load("data-fields.config.json")
+        self.quality = load("data-quality.config.json")
 
     def test_features_are_unique_and_use_declared_roles(self) -> None:
         roles = set(self.features["roles"])
@@ -82,6 +83,15 @@ class GovernanceRegistryTests(unittest.TestCase):
                 self.assertEqual(dataset["usage"], "blocked")
         self.assertEqual(len(ids), len(set(ids)))
 
+    def test_every_dataset_has_one_quality_rule_and_known_unit_profile(self) -> None:
+        dataset_ids = {item["id"] for item in self.data["datasets"]}
+        rules = self.quality["datasets"]
+        self.assertEqual(set(rules), dataset_ids)
+        profiles = set(self.quality["unit_profiles"])
+        self.assertTrue(
+            all(str(rule.get("unit_profile") or "none") in profiles for rule in rules.values())
+        )
+
 
 if __name__ == "__main__":
     unittest.main()