Artifex/generate_full_historical_oracle_request.py

#!/usr/bin/env python3
"""Build the exact 711-feature code-5056feb oracle request without Django."""

from __future__ import annotations

import argparse
import hashlib
import json
from pathlib import Path

ENGINE_REVISION = "code-5056feb"
INPUT_COLUMNS = ["close", "high", "low", "volume"]


def canonical(value: object) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")


def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def validate_usage_summary(path: Path) -> None:
    payload = json.loads(path.read_text(encoding="utf-8"))
    if payload.get("artifact") != "HISTORICAL_FEATURE_USAGE_SUMMARY_V1":
        raise ValueError("--usage-summary is not HISTORICAL_FEATURE_USAGE_SUMMARY_V1")
    if payload.get("counts", {}).get("role_independent_primitives") != 711:
        raise ValueError("usage summary does not attest to 711 role-independent primitives")


def triples_from_engineering(path: Path) -> set[tuple[int, int, float]]:
    payload = json.loads(path.read_text(encoding="utf-8"))
    if payload.get("artifact") != "ENGINEERING_FAMILY_MAP_V1":
        raise ValueError("--engineering-map is not ENGINEERING_FAMILY_MAP_V1")
    return {
        (int(row["indicator_id"]), int(row["period"]), float(row["p1"]))
        for row in payload["primitives"]
    }


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--usage-summary", type=Path, required=True, help="historical_feature_usage_v1_summary.json")
    parser.add_argument("--engineering-map", type=Path, required=True, help="engineering_family_map_v1.json")
    parser.add_argument("--input-csv", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--deq-targets", type=Path, help="optional JSON metadata copied into the request")
    args = parser.parse_args()

    validate_usage_summary(args.usage_summary)
    engineering = triples_from_engineering(args.engineering_map)
    if len(engineering) != 711:
        raise ValueError(f"engineering map contains {len(engineering)} rather than 711 primitives")
    if not args.input_csv.is_file():
        raise ValueError("--input-csv must be a file")
    semantic_base = {
        "engine_revision": ENGINE_REVISION,
        "input_columns": INPUT_COLUMNS,
        "output_dtype": "float64",
        "window_policy": "continuous_full_history",
    }
    requests = []
    for indicator_id, period, p1 in sorted(engineering):
        semantic = {**semantic_base, "indicator_id": indicator_id, "period": period, "p1": p1}
        requests.append(
            {
                "request_id": f"hs22_{indicator_id}_{period}_{p1:g}",
                "indicator_id": indicator_id,
                "period": period,
                "p1": p1,
                "semantic_fingerprint": hashlib.sha256(canonical(semantic)).hexdigest(),
            }
        )
    payload: dict[str, object] = {
        "schema_version": 1,
        "artifact": "HISTORICAL_FEATURE_ORACLE_V1_REQUEST",
        **semantic_base,
        "input_csv_sha256": sha256(args.input_csv),
        "source_artifacts": {
            "usage_summary": {"path": args.usage_summary.name, "sha256": sha256(args.usage_summary)},
            "engineering_map": {"path": args.engineering_map.name, "sha256": sha256(args.engineering_map)},
        },
        "requests": requests,
    }
    if args.deq_targets:
        payload["deq_targets"] = json.loads(args.deq_targets.read_text(encoding="utf-8"))
        payload["source_artifacts"]["deq_targets"] = {"path": args.deq_targets.name, "sha256": sha256(args.deq_targets)}
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_bytes(canonical(payload) + b"\n")
    print(args.output)


if __name__ == "__main__":
    main()