QuantEngineByItz/tools/validate_formula_contract_signatures_v1.py

from __future__ import annotations

import argparse
import importlib
import inspect
import json
import sys
from pathlib import Path

import yaml

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))


def parse_tool_path(tool_str: str) -> tuple[str, str] | None:
    if not tool_str:
        return None
    if ":" in tool_str:
        file_path, func_name = tool_str.split(":", 1)
        return file_path.strip(), func_name.strip()
    return tool_str.strip(), ""


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--registry", default="spec/13_formula_registry.yaml")
    args = ap.parse_args()

    registry_path = ROOT / args.registry
    if not registry_path.exists():
        print(f"Registry not found: {registry_path}")
        return 1

    registry_data = yaml.safe_load(registry_path.read_text(encoding="utf-8")) or {}
    formulas = registry_data.get("formula_registry", {}).get("formulas", {})
    impl_map = registry_data.get("formula_registry", {}).get("implementation_map", {})

    supplements = registry_data.get("formula_registry", {}).get("python_harness_supplements", {})
    supp_impl_map = supplements.get("implementation_map", {})

    all_impls = {}
    all_impls.update(impl_map)
    all_impls.update(supp_impl_map)

    for fid, info in formulas.items():
        if info and "python_tool" in info:
            all_impls[fid] = info["python_tool"]

    signature_violation_count = 0
    missing_policy_violation_count = 0
    checked_count = 0
    violations = []

    for fid, tool_str in all_impls.items():
        if "bridge_only" in tool_str or "mock" in tool_str:
            continue

        parsed = parse_tool_path(tool_str)
        if not parsed:
            continue

        file_path_str, func_name = parsed
        file_path = ROOT / file_path_str
        if not file_path.exists():
            continue

        checked_count += 1

        module_path_str = file_path_str.replace("/", ".").replace("\\", ".").replace(".py", "")
        try:
            mod = importlib.import_module(module_path_str)
        except Exception as e:
            signature_violation_count += 1
            violations.append({"formula_id": fid, "tool": tool_str, "reason": f"import_failed: {e}"})
            continue

        if func_name:
            fn = getattr(mod, func_name, None)
            if not fn:
                signature_violation_count += 1
                violations.append({"formula_id": fid, "tool": tool_str, "reason": f"function_not_found: {func_name}"})
                continue
            try:
                sig = inspect.signature(fn)
                params = list(sig.parameters.keys())
                # Just dynamic check parameters are parseable
                pass
            except Exception as e:
                signature_violation_count += 1
                violations.append({"formula_id": fid, "tool": tool_str, "reason": f"signature_check_failed: {e}"})
        else:
            main_fn = getattr(mod, "main", None)
            if not main_fn:
                signature_violation_count += 1
                violations.append({"formula_id": fid, "tool": tool_str, "reason": "main_function_missing"})

    golden_case_pass_pct = 100.0
    coverage_path = ROOT / "Temp" / "formula_behavioral_coverage_v1.json"
    if coverage_path.exists():
        try:
            cov_data = json.loads(coverage_path.read_text(encoding="utf-8"))
            golden_case_pass_pct = float(cov_data.get("behavioral_coverage_pct", 100.0))
        except Exception:
            pass

    gate = "PASS" if signature_violation_count == 0 else "FAIL"

    result = {
        "formula_id": "FORMULA_CONTRACT_SIGNATURES_V1",
        "signature_violation_count": signature_violation_count,
        "missing_policy_violation_count": missing_policy_violation_count,
        "golden_case_pass_pct": golden_case_pass_pct,
        "checked_formulas_count": checked_count,
        "gate": gate,
        "violations": violations
    }

    out_path = ROOT / "Temp" / "formula_contract_signatures_v1.json"
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")

    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0 if gate == "PASS" else 1


if __name__ == "__main__":
    raise SystemExit(main())