203 lines
7.2 KiB
Python
203 lines
7.2 KiB
Python
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 backend.bootstrap.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)
|