"""Versioned deterministic metric catalogue. Metric weights and scoring kinds are product policy, not provider choices. Providers supply observations only; changing a definition requires a new score version and never rewrites historical snapshots. """ from __future__ import annotations from dataclasses import dataclass from decimal import Decimal from app.core.exceptions import NotFoundError from app.intelligence.scoring import ( DIMENSION_WEIGHTS, METRIC_RULE_LABELS, SCORE_VERSION, MetricKind, validate_metric_key_shape_is_uniform, validate_metric_weight_totals, ) @dataclass(frozen=True) class MetricDefinition: key: str dimension: str kind: MetricKind weight: Decimal unit: str label_zh: str label_en: str label_ja: str def _metric( key: str, dimension: str, kind: MetricKind, weight: str, unit: str, zh: str, en: str, ja: str, ) -> MetricDefinition: return MetricDefinition( key=key, dimension=dimension, kind=kind, weight=Decimal(weight), unit=unit, label_zh=zh, label_en=en, label_ja=ja, ) METRIC_CATALOG_VERSION = SCORE_VERSION # PR-sized catalogue slice. Later domain providers extend this immutable v1 # list before the product flag opens; tests require every provider observation # to resolve here, so no unreviewed metric can silently affect a score. METRIC_CATALOG: dict[str, MetricDefinition] = { row.key: row for row in ( _metric( "people_org.hris_freshness", "people_org", MetricKind.FRESHNESS, ".25", "missed_cycles", "HRIS 数据新鲜度", "HRIS data freshness", "HRIS データ鮮度", ), _metric( "people_org.org_coverage", "people_org", MetricKind.COVERAGE, ".25", "percent", "组织归属完整度", "Organization coverage", "組織所属の完全性", ), _metric( "people_org.manager_coverage", "people_org", MetricKind.COVERAGE, ".25", "percent", "经理关系完整度", "Manager relationship coverage", "上司関係の完全性", ), _metric( "people_org.prediction_data_coverage", "people_org", MetricKind.COVERAGE, ".25", "percent", "预测字段完整度", "Prediction field coverage", "予測項目の完全性", ), # v2 lifecycle weights (unchanged in v3 — v3 redefined # ``onboarding_sla``'s reading, not the weight split). # ``lifecycle.transfer_recovery`` (.15) was # retired — its signal source never produced a production row, so it # was permanently confidence-0 and its weight only shrank the scoring # denominator. Its share went to the two headline lifecycle outcomes # (day1_ready / leaver_failure_rate) and to owner_continuity, which is # a real and well-populated signal. ``onboarding_sla`` stays at .20: # v3 gave it a distinct bar (hard readiness within the shipped 72h # SLA, vs day1_ready's "by end of hire_date"), but it still reads the # same hire population, so raising it would deepen that overlap. _metric( "lifecycle.day1_ready", "lifecycle", MetricKind.COVERAGE, ".30", "percent", "Day-1 就绪率", "Day-1 readiness", "Day-1 準備率", ), _metric( "lifecycle.onboarding_sla", "lifecycle", MetricKind.COVERAGE, ".20", "percent", "入职 SLA 达成率", "Onboarding SLA attainment", "オンボーディング SLA 達成率", ), _metric( "lifecycle.leaver_failure_rate", "lifecycle", MetricKind.FAILURE_RATE, ".30", "percent", "离职阻塞失败率", "Leaver blocking failure rate", "退職ブロック失敗率", ), _metric( "lifecycle.owner_continuity", "lifecycle", MetricKind.COVERAGE, ".20", "percent", "所有权连续性", "Ownership continuity", "所有権の継続性", ), # v3 asset weights. ``assets.warranty_coverage`` (.15) was retired: # ``asset_items.warranty_until`` has never been written in production # (0 of 1,099 rows, legacy import included) and no intake / edit # control produces it, so the metric could only ever assert an # unmovable 0% and told admins to "complete the uncovered objects" # through a field they cannot fill. It stays a fact # (``warranty_population_count`` / ``warranty_complete_count``); a # scored version needs a real producer and a new version. Its share # left the five surviving asset outcomes at an even .20 — supply # (demand_coverage), truth (inventory_accuracy), circulation # (idle_rate), exit leakage (leaver_hold_rate) and book value # (financial_coverage) are equally load-bearing for the domain. _metric( "assets.demand_coverage", "assets", MetricKind.COVERAGE, ".20", "percent", "近期设备需求覆盖率", "Near-term device demand coverage", "短期デバイス需要カバレッジ", ), _metric( "assets.inventory_accuracy", "assets", MetricKind.COVERAGE, ".20", "percent", "最近盘点准确率", "Latest inventory accuracy", "最新棚卸精度", ), _metric( "assets.idle_rate", "assets", MetricKind.FAILURE_RATE, ".20", "percent", "长期闲置电脑盈余率", "Long-idle surplus computer rate", "長期未使用PC余剰率", ), _metric( "assets.leaver_hold_rate", "assets", MetricKind.FAILURE_RATE, ".20", "percent", "离职人员持有率", "Leaver-held asset rate", "退職者保有資産率", ), _metric( "assets.financial_coverage", "assets", MetricKind.COVERAGE, ".20", "percent", "资产财务完整度", "Asset financial completeness", "資産財務情報の完全性", ), _metric( "software_api.seat_utilization", "software_api", MetricKind.COVERAGE, ".20", "percent", "可观测席位利用率", "Observable seat utilization", "観測可能なシート利用率", ), _metric( "software_api.capacity_headroom", "software_api", MetricKind.COVERAGE, ".15", "percent", "席位容量余量", "Seat capacity headroom", "シート容量余裕", ), _metric( "software_api.owner_coverage", "software_api", MetricKind.COVERAGE, ".20", "percent", "软件与 Key Owner 完整度", "Software and key owner coverage", "ソフトウェアとキーの所有者網羅率", ), _metric( "software_api.reconciliation_coverage", "software_api", MetricKind.COVERAGE, ".20", "percent", "厂商席位对账匹配率", "Vendor seat reconciliation coverage", "ベンダーシート照合率", ), _metric( "software_api.key_expiry_coverage", "software_api", MetricKind.COVERAGE, ".15", "percent", "API Key 有效期治理", "API key expiry governance", "API キー有効期限ガバナンス", ), _metric( "software_api.rotation_failure_rate", "software_api", MetricKind.FAILURE_RATE, ".10", "percent", "Key 轮换退役失败率", "Key rotation retirement failure rate", "キー更新廃止失敗率", ), _metric( "cloud_network.sync_freshness", "cloud_network", MetricKind.FRESHNESS, ".20", "missed_cycles", "云订阅同步新鲜度", "Cloud subscription sync freshness", "クラウドサブスクリプション同期鮮度", ), _metric( "cloud_network.resource_waste_rate", "cloud_network", MetricKind.FAILURE_RATE, ".20", "percent", "可执行资源浪费率", "Actionable resource waste rate", "対応可能なリソース浪費率", ), _metric( "cloud_network.access_governance", "cloud_network", MetricKind.COVERAGE, ".20", "percent", "云授权治理覆盖率", "Cloud access governance coverage", "クラウドアクセス統制カバレッジ", ), _metric( "cloud_network.owner_coverage", "cloud_network", MetricKind.COVERAGE, ".20", "percent", "云与网络 Owner 覆盖率", "Cloud and network owner coverage", "クラウドとネットワーク所有者カバレッジ", ), _metric( "cloud_network.domain_certificate_health", "cloud_network", MetricKind.COVERAGE, ".20", "percent", "域名与证书健康度", "Domain and certificate health", "ドメインと証明書の健全性", ), _metric( "cost_efficiency.business_budget_coverage", "cost_efficiency", MetricKind.COVERAGE, ".20", "percent", "业务口径预算覆盖率", "Business-basis budget coverage", "事業基準予算カバレッジ", ), _metric( "cost_efficiency.anomaly_rate", "cost_efficiency", MetricKind.FAILURE_RATE, ".20", "percent", "成本异常资源率", "Cost-anomalous resource rate", "コスト異常リソース率", ), _metric( "cost_efficiency.optimizable_cost_rate", "cost_efficiency", MetricKind.FAILURE_RATE, ".20", "percent", "可优化成本比例", "Optimizable cost rate", "最適化可能コスト率", ), _metric( "cost_efficiency.allocation_coverage", "cost_efficiency", MetricKind.COVERAGE, ".20", "percent", "成本分摊完整度", "Cost allocation coverage", "コスト配賦カバレッジ", ), _metric( "cost_efficiency.growth_rate", "cost_efficiency", MetricKind.FAILURE_RATE, ".20", "percent", "30 天业务成本增长率", "30-day business cost growth rate", "30 日事業コスト増加率", ), _metric( "security_compliance.mfa_enforcement_coverage", "security_compliance", MetricKind.COVERAGE, ".20", "percent", "MFA 注册与强制覆盖率", "MFA enrollment and enforcement coverage", "MFA 登録・強制適用カバレッジ", ), _metric( "security_compliance.privileged_review_coverage", "security_compliance", MetricKind.COVERAGE, ".20", "percent", "特权访问复核完成率", "Privileged access review coverage", "特権アクセスレビュー完了率", ), _metric( "security_compliance.public_exposure_rate", "security_compliance", MetricKind.FAILURE_RATE, ".20", "percent", "公开暴露未解决率", "Open public exposure rate", "公開露出未解決率", ), _metric( "security_compliance.credential_risk_rate", "security_compliance", MetricKind.FAILURE_RATE, ".20", "percent", "密钥与 PAT 风险率", "Secret and PAT risk rate", "シークレットと PAT のリスク率", ), _metric( "security_compliance.evidence_freshness", "security_compliance", MetricKind.FRESHNESS, ".20", "missed_cycles", "安全证据新鲜度", "Security evidence freshness", "セキュリティ証跡鮮度", ), _metric( "platform_ops.scheduled_job_failure_rate", "platform_ops", MetricKind.FAILURE_RATE, ".20", "percent", "定时任务失败率", "Scheduled job failure rate", "定期ジョブ失敗率", ), _metric( "platform_ops.provisioning_failure_rate", "platform_ops", MetricKind.FAILURE_RATE, ".20", "percent", "自动开通失败率", "Automated provisioning failure rate", "自動プロビジョニング失敗率", ), _metric( "platform_ops.notification_failure_rate", "platform_ops", MetricKind.FAILURE_RATE, ".15", "percent", "通知送达失败率", "Notification delivery failure rate", "通知配信失敗率", ), _metric( "platform_ops.connector_health_coverage", "platform_ops", MetricKind.COVERAGE, ".15", "percent", "连接器健康覆盖率", "Connector health coverage", "コネクター健全性カバレッジ", ), _metric( "platform_ops.client_error_report_rate", "platform_ops", MetricKind.FAILURE_RATE, ".15", "percent", "前端错误报告率", "Client error report rate", "クライアントエラー報告率", ), _metric( "platform_ops.data_quality_finding_rate", "platform_ops", MetricKind.FAILURE_RATE, ".15", "percent", "身份数据质量问题率", "Identity data-quality finding rate", "ID データ品質問題率", ), ) } def metric_definition(key: str, dimension: str) -> MetricDefinition: definition = METRIC_CATALOG.get(key) if definition is None: raise ValueError(f"Metric {key!r} is not registered in {METRIC_CATALOG_VERSION}") if definition.dimension != dimension: raise ValueError(f"Metric {key!r} belongs to {definition.dimension}, not {dimension}") return definition def dimension_metric_keys(dimension: str) -> tuple[str, ...]: """Every metric the catalogue expects for one dimension. Used to state coverage as "observed of expected" rather than as a bare "insufficient": a dimension scoring on 2 of its 6 metrics is a different problem from one whose provider did not run at all. """ return tuple( row.key for row in METRIC_CATALOG.values() if row.dimension == dimension ) def public_metric_catalog(dimension: str) -> list[dict[str, object]]: """Return the immutable, non-sensitive explanation contract for one domain.""" rows = [row for row in METRIC_CATALOG.values() if row.dimension == dimension] if not rows: raise NotFoundError(f"Intelligence metric catalogue {dimension!r} not found") return [ { "key": row.key, "kind": row.kind.value, "weight": str(row.weight), "unit": row.unit, "labels": { "zh-CN": row.label_zh, "en-US": row.label_en, "ja-JP": row.label_ja, }, "rule": { "score_version": SCORE_VERSION, "labels": METRIC_RULE_LABELS[row.kind], }, } for row in rows ] def validate_catalog() -> None: validate_metric_weight_totals( ((row.dimension, row.weight) for row in METRIC_CATALOG.values()), DIMENSION_WEIGHTS, ) validate_metric_key_shape_is_uniform( ((row.key, row.dimension) for row in METRIC_CATALOG.values()), ) validate_catalog()