"""Build source-only feature-usage and registry-coverage archaeology artifacts.""" from __future__ import annotations import argparse import hashlib import json from collections import Counter, defaultdict from pathlib import Path import pyarrow as pa import pyarrow.parquet as pq ROLES = ("trend", "signal", "trigger", "confirm", "vol") FRONTIER_TARGETS = (80, 90, 95, 98, 99, 100) def canonical_key(indicator_id: int, period: int, p1: float) -> tuple[int, int, float]: return indicator_id, period, float(p1) def key_text(key: tuple[int, int, float]) -> str: return f"{key[0]}:{key[1]}:{key[2]:g}" def digest(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def frontier(counts: Counter[object]) -> list[dict[str, object]]: total = sum(counts.values()) ordered = sorted(counts.items(), key=lambda item: (-item[1], str(item[0]))) result = [] cumulative = 0 cursor = 0 for target in FRONTIER_TARGETS: threshold = total * target / 100 while cursor < len(ordered) and cumulative < threshold: cumulative += ordered[cursor][1] cursor += 1 result.append( { "target_percent": target, "items_required": cursor, "cumulative_usage": cumulative, "cumulative_percent": cumulative * 100 / total if total else 0.0, } ) return result def load_catalog(path: Path) -> dict[tuple[int, int, float], str]: payload = json.loads(path.read_text(encoding="utf-8")) return { canonical_key(item["indicator_id"], item["period"], item["p1"]): item["support_state"] for item in payload["features"] } def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--lineage", type=Path, required=True) parser.add_argument("--cohort-manifest", type=Path, required=True) parser.add_argument("--feature-catalog", type=Path, required=True) parser.add_argument("--output-dir", type=Path, required=True) args = parser.parse_args() output_dir = args.output_dir output_dir.mkdir(parents=True, exist_ok=True) catalog = load_catalog(args.feature_catalog) cohort = json.loads(args.cohort_manifest.read_text(encoding="utf-8")) cohort_members = cohort["members"] cohort_by_family: dict[str, list[dict[str, object]]] = defaultdict(list) cohort_by_strategy_id: dict[int, str] = {} for member in cohort_members: combo = member["combo"] triples = tuple( canonical_key(int(combo[index]), int(combo[index + 1]), float(combo[index + 2])) for index in range(0, 15, 3) ) family = json.dumps(triples, separators=(",", ":")) cohort_by_family[family].append(member) cohort_by_strategy_id[int(member["source_strategy_id"])] = family id_counts: Counter[int] = Counter() primitive_counts: Counter[tuple[int, int, float]] = Counter() role_counts: Counter[str] = Counter() role_triple_counts: Counter[tuple[str, int, int, float]] = Counter() family_counts: Counter[str] = Counter() primitive_roles: dict[tuple[int, int, float], Counter[str]] = defaultdict(Counter) found_cohort_strategy_ids: set[int] = set() source = pq.ParquetFile(args.lineage) rows = 0 for batch in source.iter_batches(columns=["strategy_id", "combo_json"], batch_size=65_536): for record in batch.to_pylist(): combo = json.loads(record["combo_json"]) if len(combo) < 15: raise ValueError(f"strategy {record['strategy_id']} has an incomplete primitive combo") triples = tuple( canonical_key(int(combo[index]), int(combo[index + 1]), float(combo[index + 2])) for index in range(0, 15, 3) ) family = json.dumps(triples, separators=(",", ":")) strategy_id = int(record["strategy_id"]) expected_family = cohort_by_strategy_id.get(strategy_id) if expected_family is not None: if family != expected_family: raise ValueError(f"cohort strategy {strategy_id} does not match its manifest combo") found_cohort_strategy_ids.add(strategy_id) family_counts[family] += 1 rows += 1 for role, primitive in zip(ROLES, triples, strict=True): indicator_id, period, p1 = primitive id_counts[indicator_id] += 1 primitive_counts[primitive] += 1 role_counts[role] += 1 role_triple_counts[(role, indicator_id, period, p1)] += 1 primitive_roles[primitive][role] += 1 if rows != source.metadata.num_rows: raise ValueError("Parquet batch iteration did not cover every lineage row") if found_cohort_strategy_ids != set(cohort_by_strategy_id): raise ValueError("not every cohort source_strategy_id is present in the lineage") cohort_family_counts = {family: len(members) for family, members in cohort_by_family.items()} cohort_primitive_counts: Counter[tuple[int, int, float]] = Counter() cohort_role_triple_counts: Counter[tuple[str, int, int, float]] = Counter() for family, members in cohort_by_family.items(): triples = json.loads(family) for _member in members: for role, raw in zip(ROLES, triples, strict=True): primitive = canonical_key(*raw) cohort_primitive_counts[primitive] += 1 cohort_role_triple_counts[(role, *primitive)] += 1 usage_rows: list[dict[str, object]] = [] for indicator_id, count in id_counts.items(): usage_rows.append({"entity_type": "indicator_id", "indicator_id": indicator_id, "usage_count": count}) for role, count in role_counts.items(): usage_rows.append({"entity_type": "role", "role": role, "usage_count": count}) for primitive, count in primitive_counts.items(): usage_rows.append( { "entity_type": "primitive", "primitive": key_text(primitive), "indicator_id": primitive[0], "period": primitive[1], "p1": primitive[2], "usage_count": count, "cohort_usage_count": cohort_primitive_counts[primitive], "classification": catalog.get(primitive, "unregistered"), } ) for (role, indicator_id, period, p1), count in role_triple_counts.items(): primitive = canonical_key(indicator_id, period, p1) usage_rows.append( { "entity_type": "role_triple", "role": role, "primitive": key_text(primitive), "indicator_id": indicator_id, "period": period, "p1": p1, "usage_count": count, "cohort_usage_count": cohort_role_triple_counts[(role, indicator_id, period, p1)], "classification": catalog.get(primitive, "unregistered"), } ) for family, count in family_counts.items(): members = cohort_by_family.get(family, []) usage_rows.append( { "entity_type": "base_family", "base_family_signature": family, "usage_count": count, "cohort_usage_count": len(members), "cohort_base_families": ",".join(str(member["base_family"]) for member in members) or None, } ) usage_path = output_dir / "historical_feature_usage_v1.parquet" pq.write_table(pa.Table.from_pylist(usage_rows), usage_path, compression="zstd") primitive_registry = [] for primitive, count in sorted(primitive_counts.items(), key=lambda item: (-item[1], item[0])): classification = catalog.get(primitive, "unregistered") primitive_registry.append( { "primitive": key_text(primitive), "indicator_id": primitive[0], "period": primitive[1], "p1": primitive[2], "usage_count": count, "usage_percent": count * 100 / (rows * len(ROLES)), "roles": dict(sorted(primitive_roles[primitive].items())), "cohort_usage_count": cohort_primitive_counts[primitive], "classification": classification, "currently_validated": classification == "validated", } ) registry_payload = { "schema_version": 1, "artifact": "HISTORICAL_PRIMITIVE_REGISTRY_COVERAGE_V1", "lineage_rows": rows, "primitive_slots": rows * len(ROLES), "observed_primitives": len(primitive_counts), "currently_validated_formula_variants": sum( classification == "validated" for classification in catalog.values() ), "observed_currently_validated_primitives": sum( item["currently_validated"] for item in primitive_registry ), "weighted_primitive_frontier": frontier(primitive_counts), "primitives": primitive_registry, } registry_path = output_dir / "historical_primitive_registry_coverage_v1.json" registry_path.write_text(json.dumps(registry_payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") summary = { "schema_version": 1, "artifact": "HISTORICAL_FEATURE_USAGE_SUMMARY_V1", "source": { "lineage": str(args.lineage), "lineage_sha256": digest(args.lineage), "lineage_rows": rows, "cohort_manifest": str(args.cohort_manifest), "cohort_manifest_sha256": digest(args.cohort_manifest), }, "counts": { "unique_strategy_ids": rows, "primitive_slots": rows * len(ROLES), "indicator_ids": len(id_counts), "exact_role_triples": len(role_triple_counts), "role_independent_primitives": len(primitive_counts), "roles": len(role_counts), "base_families": len(family_counts), "cohort_members": len(cohort_members), "cohort_source_strategy_ids_found": len(found_cohort_strategy_ids), "cohort_base_families_found": sum(family in family_counts for family in cohort_by_family), "cohort_role_triples": len(cohort_role_triple_counts), "cohort_primitives": len(cohort_primitive_counts), "currently_validated_formula_variants": sum( classification == "validated" for classification in catalog.values() ), "observed_currently_validated_primitives": sum( catalog.get(primitive) == "validated" for primitive in primitive_counts ), }, "role_usage": dict(sorted(role_counts.items())), "weighted_cumulative_frontiers": { "role_triples": frontier(role_triple_counts), "primitives": frontier(primitive_counts), }, "artifacts": { "usage_parquet": str(usage_path), "primitive_registry_coverage": str(registry_path), }, } summary_path = output_dir / "historical_feature_usage_v1_summary.json" summary_path.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8") summary["artifacts"]["usage_parquet_sha256"] = digest(usage_path) summary["artifacts"]["primitive_registry_coverage_sha256"] = digest(registry_path) summary_path.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8") if __name__ == "__main__": main()