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 importlib
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import numpy as np

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_engine(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")
    sys.path.insert(0, str(source))
    engine = importlib.import_module("hyperscalper.fast_engine")
    if Path(engine.__file__).resolve() != engine_file.resolve():
        raise RuntimeError("refused a fast_engine outside --recovered-source")
    return engine


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, engine: 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 = {int(row[0]) for row in engine.FAST_POOL}
    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(source: Path, engine: 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")
    wrapper = importlib.import_module("hyperscalper.paper_replay")
    close, high, low, volume = arrays
    state = wrapper.compute_combo_state(combo, close, high, low, volume)
    slots = ("trend", "signal", "trigger", "confirm", "vol")
    for slot, offset in zip(slots, range(0, 15, 3), strict=True):
        direct = np.asarray(engine.compute(int(combo[offset]), close, high, low, volume, int(combo[offset + 1]), float(combo[offset + 2])), dtype=np.float64)
        if slot not in state or not np.array_equal(direct, np.asarray(state[slot], dtype=np.float64), equal_nan=True):
            raise RuntimeError(f"paper_replay wrapper mismatch for {slot}")
    return {"status": "PASS", "paper_replay_sha256": digest_file(source / "hyperscalper" / "paper_replay.py")}


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; validates paper_replay")
    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:
        engine = load_engine(args.recovered_source)
        arrays = load_csv(args.input_csv)
        requests = load_request(args.request, engine)
        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(engine.compute(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(args.recovered_source, engine, arrays, combo)
        log(event_log, "WRAPPER_VALIDATION", **wrapper)
        records = []
        for number, row in enumerate(requests, start=1):
            values = np.asarray(engine.compute(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(engine.compute(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"), "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()