Artifex/run_historical_oracle_overnight.py

#!/usr/bin/env python3
"""Resumable, Django-free full-history oracle runner for recovered code-5056feb."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import numpy as np

from historical_feature_oracle_direct_registry_v1 import load_direct_registry

ENGINE_REVISION = "code-5056feb"
REQUIRED_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 digest_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def digest_file(path: Path) -> str:
    return digest_bytes(path.read_bytes())


def array_digest(values: np.ndarray) -> str:
    values = np.ascontiguousarray(values)
    return digest_bytes(canonical({"dtype": values.dtype.str, "shape": values.shape}) + b"\0" + values.tobytes())


def atomic_bytes(path: Path, data: bytes) -> None:
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_bytes(data)
    os.replace(temporary, path)


def now() -> str:
    return datetime.now(timezone.utc).isoformat()


def log(path: Path, event: str, **fields: object) -> None:
    atomic_append = canonical({"at": now(), "event": event, **fields}) + b"\n"
    with path.open("ab") as handle:
        handle.write(atomic_append)
        handle.flush()
        os.fsync(handle.fileno())


def load_registry(source: Path):
    engine_file = source / "hyperscalper" / "fast_engine.py"
    if not engine_file.is_file() or not (source / "hyperscalper" / "__init__.py").is_file():
        raise ValueError("--recovered-source must contain hyperscalper/fast_engine.py")
    return load_direct_registry(source)


def load_csv(path: Path) -> tuple[np.ndarray, ...]:
    table = np.atleast_1d(np.genfromtxt(path, delimiter=",", names=True, dtype=None, encoding="utf-8"))
    names = set(table.dtype.names or ())
    if set(REQUIRED_COLUMNS) - names:
        raise ValueError("CSV must contain close, high, low, volume")
    arrays = tuple(np.asarray(table[name], dtype=np.float64) for name in REQUIRED_COLUMNS)
    if not arrays[0].size or any(values.shape != arrays[0].shape for values in arrays):
        raise ValueError("CSV OHLCV columns must be non-empty, equal-length vectors")
    return arrays


def load_request(path: Path, registry: dict[int, Any]) -> list[dict[str, Any]]:
    payload = json.loads(path.read_text(encoding="utf-8"))
    if payload.get("artifact") != "HISTORICAL_FEATURE_ORACLE_V1_REQUEST" or payload.get("engine_revision") != ENGINE_REVISION:
        raise ValueError("unsupported request artifact or engine revision")
    if payload.get("input_columns") != list(REQUIRED_COLUMNS) or payload.get("output_dtype") != "float64":
        raise ValueError("unsupported input/output semantics")
    rows = payload.get("requests")
    if not isinstance(rows, list) or len(rows) != 711:
        raise ValueError("full historical request must contain exactly 711 features")
    known = set(registry)
    base = {key: payload[key] for key in ("engine_revision", "input_columns", "output_dtype", "window_policy")}
    result = []
    for row in rows:
        item = {"request_id": str(row["request_id"]), "indicator_id": int(row["indicator_id"]), "period": int(row["period"]), "p1": float(row["p1"])}
        if item["indicator_id"] not in known or item["period"] <= 0 or not np.isfinite(item["p1"]):
            raise ValueError(f"invalid historical request: {item['request_id']}")
        expected = digest_bytes(canonical({**base, **{key: item[key] for key in ("indicator_id", "period", "p1")}}))
        if row.get("semantic_fingerprint") != expected:
            raise ValueError(f"semantic fingerprint mismatch: {item['request_id']}")
        result.append(item)
    if len({row["request_id"] for row in result}) != len(result):
        raise ValueError("request IDs must be unique")
    return result


def checkpoint(path: Path, row: dict[str, Any], values: np.ndarray) -> dict[str, object]:
    npy = path / f"{row['request_id']}.npy"
    meta = path / f"{row['request_id']}.json"
    if npy.is_file() and meta.is_file():
        saved = json.loads(meta.read_text(encoding="utf-8"))
        loaded = np.load(npy, allow_pickle=False)
        if saved["sha256"] == array_digest(loaded) and tuple(saved["shape"]) == loaded.shape:
            return saved
    temporary = npy.with_suffix(".tmp.npy")
    np.save(temporary, values, allow_pickle=False)
    os.replace(temporary, npy)
    saved = {**row, "path": npy.name, "dtype": values.dtype.str, "shape": list(values.shape), "sha256": array_digest(values)}
    atomic_bytes(meta, canonical(saved) + b"\n")
    return saved


def wrapper_validate(registry: dict[int, Any], arrays: tuple[np.ndarray, ...], combo: list[float]) -> dict[str, object]:
    if len(combo) < 15:
        raise ValueError("--wrapper-combo requires at least 15 numeric combo values")
    close, high, low, volume = arrays
    slots = ("trend", "signal", "trigger", "confirm", "vol")
    for slot, offset in zip(slots, range(0, 15, 3), strict=True):
        values = np.asarray(registry[int(combo[offset])](close, high, low, volume, int(combo[offset + 1]), float(combo[offset + 2])), dtype=np.float64)
        if values.shape != close.shape:
            raise RuntimeError(f"direct registry shape mismatch for {slot}")
    return {"status": "PASS", "method": "five representative direct recovered helper calls"}


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--request", type=Path, required=True)
    parser.add_argument("--input-csv", type=Path, required=True)
    parser.add_argument("--recovered-source", type=Path, required=True)
    parser.add_argument("--output-dir", type=Path, required=True)
    parser.add_argument("--pilot-count", type=int, default=3)
    parser.add_argument("--wrapper-combo", type=Path, required=True, help="combo JSON or required_variants.json; exercises five direct registry entries")
    args = parser.parse_args()
    if args.pilot_count < 1:
        raise ValueError("--pilot-count must be positive")
    args.output_dir.mkdir(parents=True, exist_ok=True)
    checkpoints = args.output_dir / "checkpoints"
    checkpoints.mkdir(exist_ok=True)
    event_log = args.output_dir / "events.jsonl"
    status = args.output_dir / "final_status.json"
    try:
        registry, provenance = load_registry(args.recovered_source)
        arrays = load_csv(args.input_csv)
        requests = load_request(args.request, registry)
        if json.loads(args.request.read_text(encoding="utf-8")).get("input_csv_sha256") not in (None, digest_file(args.input_csv)):
            raise ValueError("input CSV hash does not match request")
        log(event_log, "START", request_count=len(requests), row_count=len(arrays[0]))
        pilot = requests[: args.pilot_count]
        started = time.perf_counter()
        for row in pilot:
            values = np.asarray(registry[row["indicator_id"]](*arrays, row["period"], row["p1"]), dtype=np.float64)
            if values.shape != arrays[0].shape:
                raise RuntimeError(f"pilot shape mismatch: {row['request_id']}")
        pilot_seconds = time.perf_counter() - started
        log(event_log, "PILOT_PASS", count=len(pilot), seconds=pilot_seconds, seconds_per_feature=pilot_seconds / len(pilot))
        raw = json.loads(args.wrapper_combo.read_text(encoding="utf-8"))
        if isinstance(raw, dict) and "combos" in raw:
            raw = raw["combos"][0]
        combo = raw.get("combo", raw) if isinstance(raw, dict) else raw
        wrapper = wrapper_validate(registry, arrays, combo)
        log(event_log, "WRAPPER_VALIDATION", **wrapper)
        records = []
        for number, row in enumerate(requests, start=1):
            values = np.asarray(registry[row["indicator_id"]](*arrays, row["period"], row["p1"]), dtype=np.float64)
            if values.shape != arrays[0].shape:
                raise RuntimeError(f"shape mismatch: {row['request_id']}")
            record = checkpoint(checkpoints, row, values)
            records.append(record)
            log(event_log, "CHECKPOINT", number=number, request_id=row["request_id"], sha256=record["sha256"])
        representative = requests[len(requests) // 2]
        first = np.load(checkpoints / f"{representative['request_id']}.npy", allow_pickle=False)
        second = np.asarray(registry[representative["indicator_id"]](*arrays, representative["period"], representative["p1"]), dtype=np.float64)
        if not np.array_equal(first, second, equal_nan=True):
            raise RuntimeError("determinism rerun failed")
        final = {"schema_version": 1, "artifact": "HISTORICAL_FEATURE_ORACLE_FULL_STATUS_V1", "status": "PASS", "finished_at": now(), "engine_revision": ENGINE_REVISION, "request_sha256": digest_file(args.request), "input_csv_sha256": digest_file(args.input_csv), "engine_sha256": digest_file(args.recovered_source / "hyperscalper" / "fast_engine.py"), "direct_registry_provenance": provenance, "row_count": len(arrays[0]), "completed_features": len(records), "pilot": {"count": len(pilot), "seconds": pilot_seconds}, "wrapper_validation": wrapper, "determinism": {"request_id": representative["request_id"], "sha256": array_digest(second)}, "deq_targets": json.loads(args.request.read_text(encoding="utf-8")).get("deq_targets"), "checkpoints": records}
        atomic_bytes(status, canonical(final) + b"\n")
        log(event_log, "PASS", completed_features=len(records))
    except Exception as error:
        failure = {"schema_version": 1, "artifact": "HISTORICAL_FEATURE_ORACLE_FULL_STATUS_V1", "status": "FAIL", "finished_at": now(), "error": f"{type(error).__name__}: {error}"}
        atomic_bytes(status, canonical(failure) + b"\n")
        log(event_log, "FAIL", error=failure["error"])
        raise


if __name__ == "__main__":
    main()