309 lines
10 KiB
Python
309 lines
10 KiB
Python
"""Build the conservative Quant60 research-candidate reachability manifest.
|
|
|
|
The hosted-platform claims in this manifest have deliberately narrow
|
|
semantics. For Universe, Alpha, Portfolio, and Risk, ``*_entrypoint_reachable``
|
|
means that the locally computed output and its lineage reach the hosted
|
|
consumer through a verified ``TargetPackageV1``. It does *not* mean that
|
|
JoinQuant or QMT execute those upstream Python implementations. Execution is
|
|
the only layer implemented inside the hosted wrappers.
|
|
|
|
One real JoinQuant TargetPackage smoke observes the execution consumer,
|
|
account/price binding, orders, and fills. It therefore supports only
|
|
``execution.real_platform_observed``. The other four layer observations and
|
|
the derived all-layer observation remain false.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
PROJECT = Path(__file__).resolve().parents[1]
|
|
SOURCE_ROOT = PROJECT / "src"
|
|
if str(SOURCE_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(SOURCE_ROOT))
|
|
|
|
from quant60.ledger import canonical_json # noqa: E402
|
|
from quant60.release_reachability import ( # noqa: E402
|
|
LAYERS,
|
|
build_evidence_artifact,
|
|
build_release_reachability,
|
|
load_verified_release_reachability,
|
|
write_release_reachability,
|
|
)
|
|
|
|
|
|
RELEASE_ID = "quant60-research-candidate-v1"
|
|
RELEASE_DECLARED_AT = "2026-07-26T12:00:00Z"
|
|
DEFAULT_OUTPUT = (
|
|
PROJECT / "releases" / RELEASE_ID / "reachability.json"
|
|
)
|
|
JOINQUANT_TARGET_PACKAGE_RUN_PATH = (
|
|
"evidence/joinquant/TP-20240311-795dcfba/"
|
|
"2457a7c39a276e09e0fabf99e28978e1/run.json"
|
|
)
|
|
JOINQUANT_TARGET_PACKAGE_RUN_ID = "2457a7c39a276e09e0fabf99e28978e1"
|
|
JOINQUANT_TARGET_PACKAGE_OBSERVED_AT = "2026-07-26T08:50:15Z"
|
|
PLATFORM_CLAIM_NOTE = (
|
|
"Upstream layer status means its locally computed output/lineage is "
|
|
"carried through TargetPackageV1 to the hosted consumer; it does not "
|
|
"claim JoinQuant/QMT executes the upstream Python. Execution alone runs "
|
|
"inside the hosted wrapper. One JoinQuant smoke observes only that "
|
|
"execution boundary; it does not observe all five layers."
|
|
)
|
|
|
|
|
|
# Each path names code used by the connected local five-layer decision path.
|
|
# A shared composition module appears in several claims by design; artifact IDs
|
|
# remain unique, as required by the reachability contract.
|
|
IMPLEMENTED_EVIDENCE: dict[str, tuple[tuple[str, str], ...]] = {
|
|
"universe": (
|
|
("baseline-pipeline", "src/quant60/baseline_pipeline.py"),
|
|
),
|
|
"alpha": (
|
|
("frozen-model-bundle", "src/quant60/model_bundle.py"),
|
|
("baseline-pipeline", "src/quant60/baseline_pipeline.py"),
|
|
),
|
|
"portfolio": (
|
|
("optimizer", "src/quant60/optimizer.py"),
|
|
("baseline-pipeline", "src/quant60/baseline_pipeline.py"),
|
|
),
|
|
"risk": (
|
|
("risk-gate", "src/quant60/risk.py"),
|
|
("baseline-pipeline", "src/quant60/baseline_pipeline.py"),
|
|
),
|
|
"execution": (
|
|
("portable-execution-core", "src/quant60/portable_core.py"),
|
|
("baseline-pipeline", "src/quant60/baseline_pipeline.py"),
|
|
),
|
|
}
|
|
|
|
UNIT_TEST_EVIDENCE: dict[str, tuple[tuple[str, str], ...]] = {
|
|
"universe": (
|
|
("five-layer-causality", "tests/test_baseline_pipeline.py"),
|
|
),
|
|
"alpha": (
|
|
("five-layer-causality", "tests/test_baseline_pipeline.py"),
|
|
("frozen-model-replay", "tests/test_model_bundle.py"),
|
|
),
|
|
"portfolio": (
|
|
("five-layer-causality", "tests/test_baseline_pipeline.py"),
|
|
),
|
|
"risk": (
|
|
("five-layer-causality", "tests/test_baseline_pipeline.py"),
|
|
),
|
|
"execution": (
|
|
("five-layer-causality", "tests/test_baseline_pipeline.py"),
|
|
("portable-order-plan", "tests/test_portable_core.py"),
|
|
),
|
|
}
|
|
|
|
LOCAL_ENTRYPOINT_EVIDENCE: tuple[tuple[str, str], ...] = (
|
|
("research-smoke-cli", "src/quant60/cli.py"),
|
|
("research-experiment", "src/quant60/experiment.py"),
|
|
)
|
|
|
|
PLATFORM_ENTRYPOINTS = {
|
|
"jq": {
|
|
"capability": "jq_entrypoint_reachable",
|
|
"platform": "joinquant",
|
|
"path": "platforms/joinquant_strategy.py",
|
|
},
|
|
"qmt": {
|
|
"capability": "qmt_entrypoint_reachable",
|
|
"platform": "qmt",
|
|
"path": "platforms/qmt_builtin_strategy.py",
|
|
},
|
|
}
|
|
|
|
# These labels are part of the signed artifact IDs. They keep the narrow
|
|
# meaning visible even though the v1 schema intentionally has no free-text
|
|
# claim field.
|
|
PLATFORM_OUTPUT_LABELS = {
|
|
"universe": "universe-state-via-targetpackagev1-consumer",
|
|
"alpha": "alpha-lineage-derived-target-via-targetpackagev1-consumer",
|
|
"portfolio": "portfolio-weight-via-targetpackagev1-consumer",
|
|
"risk": "post-risk-weight-constraints-via-targetpackagev1-consumer",
|
|
"execution": "targetpackagev1-hosted-execution-entrypoint",
|
|
}
|
|
|
|
|
|
def _false_claim() -> dict[str, Any]:
|
|
return {"status": False, "artifacts": []}
|
|
|
|
|
|
def _evidence(
|
|
*,
|
|
repository_root: Path,
|
|
artifact_id: str,
|
|
kind: str,
|
|
path: str,
|
|
platform: str,
|
|
observed_at: str | None = None,
|
|
run_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
return build_evidence_artifact(
|
|
artifact_id=artifact_id,
|
|
kind=kind,
|
|
path=path,
|
|
platform=platform,
|
|
repository_root=repository_root,
|
|
observed_at=observed_at,
|
|
run_id=run_id,
|
|
)
|
|
|
|
|
|
def build_manifest(
|
|
*,
|
|
repository_root: str | Path = PROJECT,
|
|
generated_at: str = RELEASE_DECLARED_AT,
|
|
) -> dict[str, Any]:
|
|
"""Build a fully content-addressed manifest from the current repository."""
|
|
|
|
root = Path(repository_root)
|
|
layers: dict[str, dict[str, Any]] = {}
|
|
for layer in LAYERS:
|
|
implemented = [
|
|
_evidence(
|
|
repository_root=root,
|
|
artifact_id=f"{layer}-implemented-{label}",
|
|
kind="source",
|
|
path=path,
|
|
platform="repository",
|
|
)
|
|
for label, path in IMPLEMENTED_EVIDENCE[layer]
|
|
]
|
|
unit_tested = [
|
|
_evidence(
|
|
repository_root=root,
|
|
artifact_id=f"{layer}-unit-tested-{label}",
|
|
kind="test",
|
|
path=path,
|
|
platform="repository",
|
|
)
|
|
for label, path in UNIT_TEST_EVIDENCE[layer]
|
|
]
|
|
local_entrypoint = [
|
|
_evidence(
|
|
repository_root=root,
|
|
artifact_id=f"{layer}-local-{label}",
|
|
kind="entrypoint",
|
|
path=path,
|
|
platform="local",
|
|
)
|
|
for label, path in LOCAL_ENTRYPOINT_EVIDENCE
|
|
]
|
|
claims: dict[str, Any] = {
|
|
"implemented": {"status": True, "artifacts": implemented},
|
|
"unit_tested": {"status": True, "artifacts": unit_tested},
|
|
"local_entrypoint_reachable": {
|
|
"status": True,
|
|
"artifacts": local_entrypoint,
|
|
},
|
|
"real_platform_observed": _false_claim(),
|
|
}
|
|
if layer == "execution":
|
|
claims["real_platform_observed"] = {
|
|
"status": True,
|
|
"artifacts": [
|
|
_evidence(
|
|
repository_root=root,
|
|
artifact_id=(
|
|
"execution-real-platform-targetpackagev1-smoke-jq"
|
|
),
|
|
kind="platform_run",
|
|
path=JOINQUANT_TARGET_PACKAGE_RUN_PATH,
|
|
platform="joinquant",
|
|
observed_at=JOINQUANT_TARGET_PACKAGE_OBSERVED_AT,
|
|
run_id=JOINQUANT_TARGET_PACKAGE_RUN_ID,
|
|
)
|
|
],
|
|
}
|
|
for name, declaration in PLATFORM_ENTRYPOINTS.items():
|
|
capability = declaration["capability"]
|
|
claims[capability] = {
|
|
"status": True,
|
|
"artifacts": [
|
|
_evidence(
|
|
repository_root=root,
|
|
artifact_id=(
|
|
f"{layer}-{PLATFORM_OUTPUT_LABELS[layer]}-{name}"
|
|
),
|
|
kind="entrypoint",
|
|
path=declaration["path"],
|
|
platform=declaration["platform"],
|
|
)
|
|
],
|
|
}
|
|
layers[layer] = claims
|
|
|
|
return build_release_reachability(
|
|
release_id=RELEASE_ID,
|
|
generated_at=generated_at,
|
|
layers=layers,
|
|
repository_root=root,
|
|
)
|
|
|
|
|
|
def _build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--output",
|
|
type=Path,
|
|
default=DEFAULT_OUTPUT,
|
|
help="manifest path (default: releases/<release>/reachability.json)",
|
|
)
|
|
parser.add_argument(
|
|
"--generated-at",
|
|
default=RELEASE_DECLARED_AT,
|
|
help="canonical UTC release declaration time",
|
|
)
|
|
parser.add_argument(
|
|
"--check",
|
|
action="store_true",
|
|
help="verify that the existing manifest exactly matches regeneration",
|
|
)
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = _build_parser().parse_args(argv)
|
|
expected = build_manifest(generated_at=args.generated_at)
|
|
output = args.output.expanduser().resolve()
|
|
if args.check:
|
|
observed = load_verified_release_reachability(
|
|
output,
|
|
repository_root=PROJECT,
|
|
expected_release_id=RELEASE_ID,
|
|
expected_manifest_sha256=expected["manifest_sha256"],
|
|
)
|
|
expected_bytes = (canonical_json(expected) + "\n").encode("utf-8")
|
|
if observed != expected or output.read_bytes() != expected_bytes:
|
|
raise ValueError(
|
|
"reachability manifest does not exactly match regeneration"
|
|
)
|
|
else:
|
|
write_release_reachability(
|
|
expected,
|
|
output,
|
|
repository_root=PROJECT,
|
|
)
|
|
summary = {
|
|
"ok": True,
|
|
"checked": bool(args.check),
|
|
"output": str(output),
|
|
"release_id": RELEASE_ID,
|
|
"manifest_sha256": expected["manifest_sha256"],
|
|
"platform_claim_semantics": PLATFORM_CLAIM_NOTE,
|
|
}
|
|
print(json.dumps(summary, ensure_ascii=False, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|