"""Build metadata-only historical coverage and 95% implementation artifacts. This consumes usage aggregates, registry metadata, and the feature catalog. It never imports, evaluates, or ports indicator formulas. """ from __future__ import annotations import argparse import hashlib import json import math from collections import Counter from pathlib import Path DENOMINATOR = 3_514_830 TARGETS = (80, 90, 95, 98, 99, 100) STATE_TO_CLASSIFICATION = { "validated": "native", "source_implemented_unvalidated": "missing", "alias_refused": "unsafe", "unregistered": "ambiguous", } CLASSIFICATIONS = ("native", "missing", "unsafe", "ambiguous") def canonical_json(payload: object) -> bytes: return json.dumps(payload, sort_keys=True, separators=(",", ":"), allow_nan=False).encode( "ascii" ) def sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def frontier(rows: list[dict[str, object]]) -> list[dict[str, object]]: cumulative = 0 cursor = 0 output = [] for target in TARGETS: threshold = math.ceil(DENOMINATOR * target / 100) while cursor < len(rows) and cumulative < threshold: cumulative += int(rows[cursor]["usage_slots"]) cursor += 1 output.append( { "target_percent": target, "minimum_slots": threshold, "canonical_items_required": cursor, "cumulative_slots": cumulative, "cumulative_percent": cumulative * 100 / DENOMINATOR, } ) return output def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--usage-dir", type=Path, required=True) parser.add_argument("--feature-catalog", type=Path, required=True) parser.add_argument("--registry", type=Path, required=True) parser.add_argument("--output-dir", type=Path, required=True) args = parser.parse_args() usage_dir = args.usage_dir summary_path = usage_dir / "historical_feature_usage_v1_summary.json" primitive_path = usage_dir / "historical_primitive_registry_coverage_v1.json" parquet_path = usage_dir / "historical_feature_usage_v1.parquet" summary = json.loads(summary_path.read_text(encoding="utf-8")) primitive_coverage = json.loads(primitive_path.read_text(encoding="utf-8")) catalog = json.loads(args.feature_catalog.read_text(encoding="utf-8")) registry = json.loads(args.registry.read_text(encoding="utf-8")) if summary["counts"]["primitive_slots"] != DENOMINATOR: raise ValueError("Historical usage denominator does not equal 3,514,830 slots.") if primitive_coverage["primitive_slots"] != DENOMINATOR: raise ValueError("Primitive coverage denominator does not equal 3,514,830 slots.") catalog_by_key = { (item["indicator_id"], item["period"], float(item["p1"])): item for item in catalog["features"] } definitions = {item["indicator_id"]: item for item in registry["definitions"]} # The registry coverage artifact retains each primitive's per-role slot # counts, so no formula runtime or Parquet reader is required here. role_triples = [ { "role": role, "primitive": item["primitive"], "usage_count": count, } for item in primitive_coverage["primitives"] for role, count in item["roles"].items() ] triples_by_primitive: dict[str, list[dict[str, object]]] = {} for row in role_triples: triples_by_primitive.setdefault(str(row["primitive"]), []).append(row) primitives: list[dict[str, object]] = [] for source in primitive_coverage["primitives"]: key = source["indicator_id"], source["period"], float(source["p1"]) catalog_item = catalog_by_key.get(key) support_state = catalog_item["support_state"] if catalog_item else "unregistered" classification = STATE_TO_CLASSIFICATION.get(support_state, "ambiguous") definition = definitions.get(source["indicator_id"]) triples = sorted( triples_by_primitive.get(source["primitive"], []), key=lambda item: item["role"] ) primitives.append( { "primitive": source["primitive"], "indicator_id": source["indicator_id"], "period": source["period"], "p1": source["p1"], "usage_slots": source["usage_count"], "usage_percent": source["usage_count"] * 100 / DENOMINATOR, "roles": source["roles"], "role_triples": [ {"role": item["role"], "usage_slots": item["usage_count"]} for item in triples ], "role_triple_count": len(triples), "duplicate_role_assignments": max(0, len(triples) - 1), "cohort_usage_count": source["cohort_usage_count"], "support_state": support_state, "coverage_classification": classification, "causal_classification": ( catalog_item["causal_classification"] if catalog_item else "not_executable" ), "indicator_name": definition["name"] if definition else "UNKNOWN", "registry_status": definition["status"] if definition else "unregistered", "alias_of": definition["alias_of"] if definition else None, } ) primitives.sort(key=lambda item: (-int(item["usage_slots"]), str(item["primitive"]))) primitive_frontier = frontier(primitives) role_triple_rows = [ { "role_triple": f"{row['role']}:{row['primitive']}", "usage_slots": row["usage_count"], } for row in role_triples ] role_triple_rows.sort(key=lambda item: (-int(item["usage_slots"]), str(item["role_triple"]))) role_triple_frontier = frontier(role_triple_rows) class_counts = Counter({classification: 0 for classification in CLASSIFICATIONS}) class_counts.update(item["coverage_classification"] for item in primitives) class_slots = Counter({classification: 0 for classification in CLASSIFICATIONS}) for item in primitives: class_slots[str(item["coverage_classification"])] += int(item["usage_slots"]) aliases = [item for item in primitives if item["registry_status"] == "alias"] registry_aliases = [ { "indicator_id": item["indicator_id"], "indicator_name": item["name"], "alias_of": item["alias_of"], "behavior": item["behavior"], "observed_primitives": sum( primitive["indicator_id"] == item["indicator_id"] for primitive in primitives ), } for item in registry["definitions"] if item["status"] == "alias" ] duplicate_primitives = [ item for item in primitives if int(item["duplicate_role_assignments"]) > 0 ] frontier_80_count = next( item["canonical_items_required"] for item in primitive_frontier if item["target_percent"] == 80 ) frontier_95_count = next( item["canonical_items_required"] for item in primitive_frontier if item["target_percent"] == 95 ) selected = primitives[:frontier_95_count] selected_slots = sum(int(item["usage_slots"]) for item in selected) plan = [] for index, item in enumerate(selected, start=1): if item["coverage_classification"] == "native": continue item = dict(item) item["weighted_rank"] = index item["priority"] = "P0" if index <= frontier_80_count else "P1" item["action"] = { "missing": "implement_and_validate", "unsafe": "retain_refusal_and_recover_alias_semantics", "ambiguous": "recover_metadata_before_implementation", }[str(item["coverage_classification"])] plan.append(item) source = { "historical_usage_summary": {"path": str(summary_path), "sha256": sha256(summary_path)}, "primitive_registry_coverage": { "path": str(primitive_path), "sha256": sha256(primitive_path), }, "usage_parquet": {"path": str(parquet_path), "sha256": sha256(parquet_path)}, "feature_catalog": { "path": str(args.feature_catalog), "sha256": sha256(args.feature_catalog), }, "historical_registry": {"path": str(args.registry), "sha256": sha256(args.registry)}, } coverage = { "schema_version": 1, "artifact": "COVERAGE_FRONTIER_V1", "method": ( "historical usage aggregation plus registry and feature-catalog metadata; " "no formula evaluation, database access, or DEQ" ), "denominator": { "primitive_slots": DENOMINATOR, "lineage_rows": 702_966, "slots_per_row": 5, }, "source_artifacts": source, "counts": { "canonical_primitives": len(primitives), "role_triples": len(role_triples), "native_formula_variants": 80, "classifications": dict(sorted(class_counts.items())), "classification_slots": dict(sorted(class_slots.items())), "aliases_observed": len(aliases), "registry_aliases": len(registry_aliases), "cross_role_duplicate_primitives": len(duplicate_primitives), }, "frontiers": { "canonical_primitives": primitive_frontier, "role_triples": role_triple_frontier, }, "canonical_primitives": primitives, "aliases": aliases, "registry_alias_metadata": registry_aliases, "duplicate_primitives": duplicate_primitives, } implementation = { "schema_version": 1, "artifact": "FEATURE_95PCT_IMPLEMENTATION_PLAN_V1", "method": coverage["method"], "denominator": coverage["denominator"], "source_artifacts": source, "target": { "coverage_percent": 95, "minimum_slots": math.ceil(DENOMINATOR * 0.95), "selected_canonical_primitives": frontier_95_count, "selected_cumulative_slots": selected_slots, "selected_cumulative_percent": selected_slots * 100 / DENOMINATOR, }, "native_coverage": { "formula_variants": 80, "selected_native_primitives": sum( item["coverage_classification"] == "native" for item in selected ), "selected_native_slots": sum( int(item["usage_slots"]) for item in selected if item["coverage_classification"] == "native" ), }, "actual_plan_size": len(plan), "plan_classifications": { classification: sum(item["coverage_classification"] == classification for item in plan) for classification in CLASSIFICATIONS }, "plan": plan, "excluded_after_95_frontier": len(primitives) - frontier_95_count, } args.output_dir.mkdir(parents=True, exist_ok=True) coverage_path = args.output_dir / "coverage_frontier_v1.json" plan_path = args.output_dir / "feature_95pct_implementation_plan_v1.json" coverage_path.write_bytes(canonical_json(coverage) + b"\n") implementation["coverage_frontier_sha256"] = sha256(coverage_path) plan_path.write_bytes(canonical_json(implementation) + b"\n") if __name__ == "__main__": main()