232 lines
9.9 KiB
Python
232 lines
9.9 KiB
Python
"""Generate an exact diagnostic for frozen Batch01 native/oracle mismatches.
|
|
|
|
This is diagnostic-only: it evaluates the recovered native formulas without
|
|
altering their arithmetic or the frozen oracle artifacts.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import numba as nb
|
|
import numpy as np
|
|
|
|
from control_plane.trading_studio.indicators.batch01_native_parity import (
|
|
BATCH01_OUTPUTS,
|
|
BATCH01_REQUEST,
|
|
_load_ohlcv,
|
|
_require_frozen_artifacts,
|
|
)
|
|
from control_plane.trading_studio.indicators.historical_band_channel import (
|
|
_sma,
|
|
evaluate_band_channel,
|
|
)
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
JSON_PATH = ROOT / "batch01_native_mismatch_report.json"
|
|
MARKDOWN_PATH = ROOT / "batch01_native_mismatch_root_cause.md"
|
|
_SIGN = 1 << 63
|
|
_MASK = (1 << 64) - 1
|
|
|
|
|
|
def _bits(value: float) -> int:
|
|
return int(np.asarray(value, dtype=np.float64).view(np.uint64))
|
|
|
|
|
|
def _ordered_bits(value: float) -> int:
|
|
bits = _bits(value)
|
|
return (~bits & _MASK) if bits & _SIGN else bits | _SIGN
|
|
|
|
|
|
def _float(value: float) -> dict[str, Any]:
|
|
bits = _bits(value)
|
|
return {"decimal": float(value), "hex": float(value).hex(), "bits": f"0x{bits:016x}"}
|
|
|
|
|
|
def _ulp(expected: float, actual: float) -> dict[str, int]:
|
|
signed = _ordered_bits(expected) - _ordered_bits(actual)
|
|
return {"signed": signed, "absolute": abs(signed)}
|
|
|
|
|
|
def _band_intermediate(close: np.ndarray, index: int, period: int) -> dict[str, Any]:
|
|
mean = _sma(close, period)[index]
|
|
variance_sum = sum(
|
|
(close[item] - mean) ** 2 for item in range(index - period + 1, index + 1)
|
|
)
|
|
variance = variance_sum / (period - 1)
|
|
return {
|
|
"window_start": index - period + 1,
|
|
"window_end": index,
|
|
"mean": _float(mean),
|
|
"variance_sum": _float(variance_sum),
|
|
"sample_variance": _float(variance),
|
|
"sample_stddev": _float(math.sqrt(variance)),
|
|
}
|
|
|
|
|
|
@nb.njit(cache=True)
|
|
def _recovered_band_intermediate(
|
|
close: np.ndarray, index: int, period: int
|
|
) -> tuple[float, float, float, float]:
|
|
"""Execute the recovered Numba arithmetic solely to expose its intermediates."""
|
|
mean_sum = 0.0
|
|
for item in range(period):
|
|
mean_sum += close[item]
|
|
for item in range(period, index + 1):
|
|
mean_sum += close[item] - close[item - period]
|
|
mean = mean_sum / period
|
|
variance_sum = 0.0
|
|
for item in range(index - period + 1, index + 1):
|
|
variance_sum += (close[item] - mean) ** 2
|
|
variance = variance_sum / (period - 1)
|
|
return mean, variance_sum, variance, math.sqrt(variance)
|
|
|
|
|
|
def _recovered_intermediate(close: np.ndarray, index: int, period: int) -> dict[str, Any]:
|
|
mean, variance_sum, variance, standard_deviation = _recovered_band_intermediate(close, index, period)
|
|
return {
|
|
"mean": _float(mean),
|
|
"variance_sum": _float(variance_sum),
|
|
"sample_variance": _float(variance),
|
|
"sample_stddev": _float(standard_deviation),
|
|
}
|
|
|
|
|
|
def _threshold(close: float, expected: float, actual: float) -> dict[str, Any]:
|
|
expected_above = close > expected
|
|
actual_above = close > actual
|
|
return {
|
|
"close": _float(close),
|
|
"oracle_close_minus_band": _float(close - expected),
|
|
"native_close_minus_band": _float(close - actual),
|
|
"oracle_close_above_band": expected_above,
|
|
"native_close_above_band": actual_above,
|
|
"decision_changed": expected_above != actual_above,
|
|
"close_to_oracle_band_ulps": _ulp(close, expected)["absolute"],
|
|
"close_to_native_band_ulps": _ulp(close, actual)["absolute"],
|
|
}
|
|
|
|
|
|
def _request_id(indicator_id: int, period: int, p1: float) -> str:
|
|
p1_text = str(int(p1)) if p1.is_integer() else str(p1)
|
|
return f"batch01_{indicator_id}_{period}_{p1_text}"
|
|
|
|
|
|
def build_report() -> dict[str, Any]:
|
|
request, _ = _require_frozen_artifacts()
|
|
close, high, low, volume = _load_ohlcv()
|
|
actual_by_id: dict[str, np.ndarray] = {}
|
|
request_by_id = {item["request_id"]: item for item in request["requests"]}
|
|
for item in request["requests"]:
|
|
actual_by_id[item["request_id"]] = evaluate_band_channel(
|
|
int(item["indicator_id"]), close, high, low, volume, int(item["period"]), float(item["p1"])
|
|
)
|
|
|
|
variants: list[dict[str, Any]] = []
|
|
with np.load(BATCH01_OUTPUTS) as oracle:
|
|
for request_id, item in request_by_id.items():
|
|
expected = oracle[request_id]
|
|
actual = actual_by_id[request_id]
|
|
mismatch_indexes = np.flatnonzero(~((expected == actual) | (np.isnan(expected) & np.isnan(actual))))
|
|
variant: dict[str, Any] = {
|
|
"request_id": request_id,
|
|
"indicator_id": item["indicator_id"],
|
|
"period": item["period"],
|
|
"p1": item["p1"],
|
|
"classification": "exact" if not len(mismatch_indexes) else "numeric_evaluation_order_drift",
|
|
"mismatch_count": int(len(mismatch_indexes)),
|
|
"mismatches": [],
|
|
}
|
|
for index in mismatch_indexes:
|
|
expected_value, actual_value = float(expected[index]), float(actual[index])
|
|
detail: dict[str, Any] = {
|
|
"index": int(index),
|
|
"oracle": _float(expected_value),
|
|
"native": _float(actual_value),
|
|
"absolute_error": _float(abs(expected_value - actual_value)),
|
|
"ulp": _ulp(expected_value, actual_value),
|
|
"threshold_sensitivity": _threshold(float(close[index]), expected_value, actual_value),
|
|
}
|
|
if int(item["indicator_id"]) in {17, 18}:
|
|
period, p1 = int(item["period"]), float(item["p1"])
|
|
upper_id, lower_id = _request_id(17, period, p1), _request_id(18, period, p1)
|
|
oracle_upper, oracle_lower = float(oracle[upper_id][index]), float(oracle[lower_id][index])
|
|
native_upper = float(actual_by_id[upper_id][index])
|
|
native_lower = float(actual_by_id[lower_id][index])
|
|
oracle_center = (oracle_upper + oracle_lower) / 2.0
|
|
native_center = (native_upper + native_lower) / 2.0
|
|
oracle_half_width = (oracle_upper - oracle_lower) / 2.0
|
|
native_half_width = (native_upper - native_lower) / 2.0
|
|
detail["native_formula_intermediate"] = _band_intermediate(close, int(index), period)
|
|
detail["recovered_numba_intermediate"] = _recovered_intermediate(
|
|
close, int(index), period
|
|
)
|
|
detail["upper_lower_symmetry"] = {
|
|
"oracle_center": _float(oracle_center),
|
|
"native_center": _float(native_center),
|
|
"center_error": _float(oracle_center - native_center),
|
|
"oracle_half_width": _float(oracle_half_width),
|
|
"native_half_width": _float(native_half_width),
|
|
"half_width_error": _float(oracle_half_width - native_half_width),
|
|
"upper_error": _float(oracle_upper - native_upper),
|
|
"lower_error": _float(oracle_lower - native_lower),
|
|
}
|
|
variant["mismatches"].append(detail)
|
|
variants.append(variant)
|
|
|
|
failures = [variant for variant in variants if variant["mismatch_count"]]
|
|
return {
|
|
"artifact": "BATCH01_NATIVE_MISMATCH_REPORT_V1",
|
|
"scope": "frozen batch01 oracle NPZ/request/CSV against native recovered band code",
|
|
"formula_mutations": False,
|
|
"variant_count": len(variants),
|
|
"failed_variant_count": len(failures),
|
|
"failed_value_count": sum(item["mismatch_count"] for item in failures),
|
|
"classification": "numeric_evaluation_order_drift_only" if failures else "exact_parity",
|
|
"variants": variants,
|
|
}
|
|
|
|
|
|
def _markdown(report: dict[str, Any]) -> str:
|
|
failures = [item for item in report["variants"] if item["mismatch_count"]]
|
|
lines = [
|
|
"# Batch01 Native Mismatch Root Cause",
|
|
"",
|
|
"## Classification",
|
|
"",
|
|
f"`{report['classification']}`: {report['failed_value_count']} bitwise mismatches in "
|
|
f"{report['failed_variant_count']} of {report['variant_count']} requests.",
|
|
"",
|
|
"## Evidence",
|
|
"",
|
|
"- All failures are Bollinger pairs: 17 (upper) and 18 (lower).",
|
|
"- Every failure has the same index and parameter pair in the opposite band.",
|
|
"- The JSON report records the oracle/native IEEE-754 bit patterns, signed ULP delta, native rolling mean/sample standard deviation, and paired-band center/half-width decomposition.",
|
|
"- Threshold comparisons (`close > band`) are unchanged at every failed value.",
|
|
"",
|
|
"## Root Cause",
|
|
"",
|
|
"The recovered Numba source and native port have bit-identical rolling means at every failed bar, but the Python generator reduction and the recovered scalar Numba variance loop differ by one or two low-order bits. Their compiled/CPython `sqrt` results can add a further low-order difference. The resulting standard-deviation rounding propagates as an equal-and-opposite upper/lower half-width shift. The evidence rules out a formula, parameter, warmup, or band-sign error.",
|
|
"",
|
|
"## Failed Requests",
|
|
"",
|
|
]
|
|
for item in failures:
|
|
indexes = ", ".join(str(detail["index"]) for detail in item["mismatches"])
|
|
ulps = ", ".join(str(detail["ulp"]["signed"]) for detail in item["mismatches"])
|
|
lines.append(f"- `{item['request_id']}`: indexes {indexes}; signed ULPs {ulps}.")
|
|
lines.append("")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main() -> None:
|
|
report = build_report()
|
|
JSON_PATH.write_text(json.dumps(report, indent=2, allow_nan=False) + "\n", encoding="utf-8")
|
|
MARKDOWN_PATH.write_text(_markdown(report), encoding="utf-8")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|