quant-os/tools/export_mock_parity.py

"""Export and verify the deterministic JoinQuant/QMT wrapper parity fixture.

This evidence proves only the local wrapper contract. It never represents a
real hosted-platform, licensed QMT, broker, or investment-performance pass.
"""

from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import json
import os
import tempfile
from pathlib import Path
from typing import Any

from platforms import joinquant_strategy, qmt_builtin_strategy
from platforms.fake_joinquant_harness import FakeJoinQuantHarness
from platforms.fake_qmt_harness import FakeQmtContext, FakeQmtHarness


PROJECT = Path(__file__).resolve().parents[1]


def _canonical_json(payload: Any) -> str:
    return json.dumps(
        payload,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        allow_nan=False,
    )


def _sha256_bytes(payload: bytes) -> str:
    return hashlib.sha256(payload).hexdigest()


def _sha256_payload(payload: Any) -> str:
    return _sha256_bytes(_canonical_json(payload).encode("utf-8"))


def _price_histories() -> dict[str, list[float]]:
    return {
        "600000.XSHG": [15.0 - index * 0.02 for index in range(21)],
        "000001.XSHE": [10.0 + index * 0.10 for index in range(21)],
        "300750.XSHE": [100.0 + index * 1.5 for index in range(21)],
        "000333.XSHE": [20.0 + index * 0.05 for index in range(21)],
    }


def _source_hashes() -> dict[str, str]:
    paths = (
        PROJECT / "platforms/joinquant_strategy.py",
        PROJECT / "platforms/qmt_builtin_strategy.py",
        PROJECT / "src/quant60/portable_core.py",
        PROJECT / "configs/baseline.json",
    )
    return {
        str(path.relative_to(PROJECT)): _sha256_bytes(path.read_bytes())
        for path in paths
    }


def build_report() -> dict[str, Any]:
    histories = _price_histories()
    jq = FakeJoinQuantHarness(histories)
    jq.initialize(joinquant_strategy)
    jq.context.previous_date = dt.date(2026, 7, 24)
    jq.context.current_dt = dt.datetime(2026, 7, 27, 9, 30)
    jq_plan = joinquant_strategy.compute_plan(jq.context)

    qmt_histories = {
        symbol.replace(".XSHE", ".SZ").replace(".XSHG", ".SH"): values
        for symbol, values in histories.items()
    }
    qmt_context = FakeQmtContext(
        qmt_histories,
        bar_time=dt.datetime(2026, 7, 24, 15, 0),
    )
    FakeQmtHarness(qmt_context).install(qmt_builtin_strategy)
    qmt_builtin_strategy.init(qmt_context)
    qmt_plan = qmt_builtin_strategy.compute_plan(
        qmt_context,
        dt.datetime(2026, 7, 24, 15, 0),
    )

    sources = _source_hashes()
    body = {
        "schema_version": "1.0",
        "evidence_class": "mock_contract_only",
        "real_platform_pass": False,
        "investment_value_claim": False,
        "fixture": {
            "as_of": "2026-07-24T15:00:00+08:00",
            "history_length": 21,
            "symbols": sorted(histories),
            "price_history_sha256": _sha256_payload(histories),
        },
        "tolerances": {
            "score_absolute": 1e-12,
            "target_weight_absolute": 1e-10,
            "target_quantity": "exact_integer",
            "order_delta": "exact_integer",
        },
        "joinquant_plan": jq_plan,
        "qmt_plan": qmt_plan,
        "comparisons": {
            "whole_plan_exact": jq_plan == qmt_plan,
            "score_exact": jq_plan["scores"] == qmt_plan["scores"],
            "target_weight_exact": (
                jq_plan["weights"] == qmt_plan["weights"]
            ),
            "target_quantity_exact": (
                jq_plan["targets"] == qmt_plan["targets"]
            ),
            "order_delta_exact": jq_plan["orders"] == qmt_plan["orders"],
        },
        "source_sha256": sources,
        "source_aggregate_sha256": _sha256_payload(sources),
        "gate_credit": [],
        "required_next_evidence": [
            "JoinQuant hosted export",
            "licensed QMT backtest export",
            "canonical clock/data/fee/fill difference report",
        ],
    }
    if not all(body["comparisons"].values()):
        raise RuntimeError("local JoinQuant/QMT wrapper parity failed")
    body["report_sha256"] = _sha256_payload(body)
    return body


def write_report(path: Path) -> dict[str, Any]:
    report = build_report()
    output = path.expanduser().resolve()
    output.parent.mkdir(parents=True, exist_ok=True)
    encoded = (
        json.dumps(
            report,
            ensure_ascii=False,
            indent=2,
            sort_keys=True,
            allow_nan=False,
        )
        + "\n"
    ).encode("utf-8")
    descriptor, temporary = tempfile.mkstemp(
        prefix=f".{output.name}.",
        suffix=".tmp",
        dir=str(output.parent),
    )
    try:
        with os.fdopen(descriptor, "wb") as handle:
            handle.write(encoded)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, output)
    except BaseException:
        try:
            os.unlink(temporary)
        except FileNotFoundError:
            pass
        raise
    return report


def verify_report(path: Path) -> dict[str, Any]:
    report = json.loads(path.read_text(encoding="utf-8"))
    saved_hash = report.pop("report_sha256", None)
    if not isinstance(saved_hash, str) or _sha256_payload(report) != saved_hash:
        raise ValueError("mock parity report hash mismatch")
    expected = build_report()
    if saved_hash != expected["report_sha256"]:
        raise ValueError("mock parity report does not match current sources")
    return {
        "ok": True,
        "evidence_class": report["evidence_class"],
        "report_sha256": saved_hash,
        "whole_plan_exact": report["comparisons"]["whole_plan_exact"],
    }


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    subparsers = parser.add_subparsers(dest="command", required=True)
    export_parser = subparsers.add_parser("export")
    export_parser.add_argument("--output", required=True, type=Path)
    verify_parser = subparsers.add_parser("verify")
    verify_parser.add_argument("report", type=Path)
    args = parser.parse_args(argv)
    if args.command == "export":
        result = write_report(args.output)
        summary = {
            "ok": True,
            "output": str(args.output.expanduser().resolve()),
            "evidence_class": result["evidence_class"],
            "report_sha256": result["report_sha256"],
        }
    else:
        summary = verify_report(args.report)
    print(json.dumps(summary, ensure_ascii=False, sort_keys=True))
    return 0


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