feat(data-processing): add per-contract ingestion lag metrics

apps/data-processing/.env.example

@@ -65,4 +65,23 @@ ROUND_MIN_CONTRIBUTORS=
 # Time window in hours for timing cluster analysis
 ROUND_TIMING_WINDOW_HOURS=1
 # Maximum allowed ratio in time window (mainnet: 0.8, testnet: 0.96)
-ROUND_TIMING_THRESHOLD=
+ROUND_TIMING_THRESHOLD=
+
+# Per-contract ingestion lag metrics
+# Contract IDs for each domain (testnet defaults are built-in; override here for mainnet)
+CONTRACT_ID_REGISTRY=
+CONTRACT_ID_VAULT=
+CONTRACT_ID_MATCHING_POOL=
+CONTRACT_ID_TREASURY=
+CONTRACT_ID_VESTING=
+# Per-domain lag alert thresholds (seconds). Defaults: WARNING=300, CRITICAL=900
+CONTRACT_LAG_WARNING_SECONDS_REGISTRY=300
+CONTRACT_LAG_CRITICAL_SECONDS_REGISTRY=900
+CONTRACT_LAG_WARNING_SECONDS_VAULT=300
+CONTRACT_LAG_CRITICAL_SECONDS_VAULT=900
+CONTRACT_LAG_WARNING_SECONDS_MATCHING_POOL=300
+CONTRACT_LAG_CRITICAL_SECONDS_MATCHING_POOL=900
+CONTRACT_LAG_WARNING_SECONDS_TREASURY=300
+CONTRACT_LAG_CRITICAL_SECONDS_TREASURY=900
+CONTRACT_LAG_WARNING_SECONDS_VESTING=300
+CONTRACT_LAG_CRITICAL_SECONDS_VESTING=900

apps/data-processing/src/api/ingestion_quality_routes.py

@@ -18,6 +18,49 @@
 router = APIRouter()
 
 
+# ---------------------------------------------------------------------------
+# Per-contract ingestion lag
+# ---------------------------------------------------------------------------
+
+class ContractLagSnapshotResponse(BaseModel):
+    domain: str
+    contract_id: Optional[str] = None
+    lag_seconds: Optional[float] = None
+    latest_onchain_ts: Optional[str] = None
+    latest_processed_ts: Optional[str] = None
+    severity: str
+    warning_threshold_seconds: float
+    critical_threshold_seconds: float
+    details: Dict[str, Any] = {}
+
+
+class ContractLagResponse(BaseModel):
+    checked_at: str
+    snapshots: list[ContractLagSnapshotResponse] = []
+    lag_alerts: list[Dict[str, Any]] = []
+    healthy: bool = True
+
+
+@router.get("/ingestion/contract-lag", response_model=ContractLagResponse)
+async def get_contract_lag() -> ContractLagResponse:
+    """Return current per-contract ingestion lag for all five domains."""
+    from src.ingestion.contract_lag_metrics import run_contract_lag_cycle
+
+    result = run_contract_lag_cycle()
+    return ContractLagResponse(**result)
+
+
+@router.post("/ingestion/contract-lag/run", response_model=ContractLagResponse)
+async def run_contract_lag() -> ContractLagResponse:
+    """Trigger an immediate per-contract lag measurement cycle."""
+    from src.ingestion.contract_lag_metrics import run_contract_lag_cycle
+
+    result = run_contract_lag_cycle()
+    return ContractLagResponse(**result)
+
+
+# ---------------------------------------------------------------------------
+
 class IngestionQualityRunRequest(BaseModel):
     network: str = "testnet"  # "testnet" only in MVP
     asset: str = "XLM"