From 831291c8184f2fbaf0c3547ee02525f4ce202576 Mon Sep 17 00:00:00 2001 From: leefer Date: Wed, 29 Jul 2026 17:34:21 +0800 Subject: [PATCH] refactor: enforce market data quality contracts --- backend/data/__init__.py | 12 +- backend/data/gateway.py | 28 +++- backend/data/quality.py | 202 +++++++++++++++++++++++ config/README.md | 2 + config/data-quality.config.json | 45 +++++ docs/governance/stage-07-data-quality.md | 32 ++++ tests/test_data_gateway.py | 105 +++++++++++- tests/test_governance_registries.py | 10 ++ 8 files changed, 433 insertions(+), 3 deletions(-) create mode 100644 backend/data/quality.py create mode 100644 config/data-quality.config.json create mode 100644 docs/governance/stage-07-data-quality.md diff --git a/backend/data/__init__.py b/backend/data/__init__.py index 94a309e..2b06ec9 100644 --- a/backend/data/__init__.py +++ b/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", +] diff --git a/backend/data/gateway.py b/backend/data/gateway.py index f79fdff..933bc93 100644 --- a/backend/data/gateway.py +++ b/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()), diff --git a/backend/data/quality.py b/backend/data/quality.py new file mode 100644 index 0000000..c35d96e --- /dev/null +++ b/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) diff --git a/config/README.md b/config/README.md index b20850c..9296574 100644 --- a/config/README.md +++ b/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. diff --git a/config/data-quality.config.json b/config/data-quality.config.json new file mode 100644 index 0000000..b0ed5db --- /dev/null +++ b/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} + } +} diff --git a/docs/governance/stage-07-data-quality.md b/docs/governance/stage-07-data-quality.md new file mode 100644 index 0000000..8d2f671 --- /dev/null +++ b/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. diff --git a/tests/test_data_gateway.py b/tests/test_data_gateway.py index b409e3d..259131c 100644 --- a/tests/test_data_gateway.py +++ b/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() diff --git a/tests/test_governance_registries.py b/tests/test_governance_registries.py index 3c01fc7..e6e332b 100644 --- a/tests/test_governance_registries.py +++ b/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()