197 lines
7.5 KiB
Python
197 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
PLAN_PATH = ROOT / "governance" / "todo" / "v8_9_p3_adoption_plan.yaml"
|
|
DECISION_FLOW_PATH = ROOT / "spec" / "09_decision_flow.yaml"
|
|
MANIFEST_PATH = ROOT / "runtime" / "active_artifact_manifest.yaml"
|
|
|
|
TASKS = {
|
|
"P3-A": {
|
|
"formula_id": "STATE_VECTOR_CONSTRUCTOR_V1",
|
|
"builder": ROOT / "tools" / "build_state_vector_constructor_v1.py",
|
|
"spec_path": ROOT / "spec" / "formulas" / "domains" / "portfolio.yaml",
|
|
"schema_path": ROOT / "schemas" / "generated" / "state_vector_constructor_v1.schema.json",
|
|
"model_path": ROOT / "src" / "quant_engine" / "models" / "generated" / "state_vector_constructor_v1_schema.py",
|
|
"temp_path": ROOT / "Temp" / "state_vector_constructor_v1.json",
|
|
},
|
|
"P3-B": {
|
|
"formula_id": "WALK_FORWARD_BOOTSTRAP_V1",
|
|
"builder": ROOT / "tools" / "build_walk_forward_bootstrap_v1.py",
|
|
"spec_path": ROOT / "spec" / "formulas" / "domains" / "simulation.yaml",
|
|
"schema_path": ROOT / "schemas" / "generated" / "walk_forward_bootstrap_v1.schema.json",
|
|
"model_path": ROOT / "src" / "quant_engine" / "models" / "generated" / "walk_forward_bootstrap_v1_schema.py",
|
|
"temp_path": ROOT / "Temp" / "walk_forward_bootstrap_v1.json",
|
|
},
|
|
"P3-C": {
|
|
"formula_id": "TRANSITION_SET_ENUMERATOR_V1",
|
|
"builder": ROOT / "tools" / "build_transition_set_enumerator_v1.py",
|
|
"spec_path": ROOT / "spec" / "formulas" / "domains" / "portfolio.yaml",
|
|
"schema_path": ROOT / "schemas" / "generated" / "transition_set_enumerator_v1.schema.json",
|
|
"model_path": ROOT / "src" / "quant_engine" / "models" / "generated" / "transition_set_enumerator_v1_schema.py",
|
|
"temp_path": ROOT / "Temp" / "transition_set_enumerator_v1.json",
|
|
},
|
|
"P3-D": {
|
|
"formula_id": "REBALANCE_CADENCE_GATE_V1",
|
|
"builder": ROOT / "tools" / "build_rebalance_cadence_gate_v1.py",
|
|
"spec_path": ROOT / "spec" / "formulas" / "domains" / "portfolio.yaml",
|
|
"schema_path": ROOT / "schemas" / "generated" / "rebalance_cadence_gate_v1.schema.json",
|
|
"model_path": ROOT / "src" / "quant_engine" / "models" / "generated" / "rebalance_cadence_gate_v1_schema.py",
|
|
"temp_path": ROOT / "Temp" / "rebalance_cadence_gate_v1.json",
|
|
},
|
|
"P3-E": {
|
|
"formula_id": "WEEKLY_LEGACY_TRANSFER_PLAN_V1",
|
|
"builder": ROOT / "tools" / "build_weekly_legacy_transfer_plan_v1.py",
|
|
"spec_path": ROOT / "spec" / "formulas" / "domains" / "cash.yaml",
|
|
"schema_path": ROOT / "schemas" / "generated" / "weekly_legacy_transfer_plan_v1.schema.json",
|
|
"model_path": ROOT / "src" / "quant_engine" / "models" / "generated" / "weekly_legacy_transfer_plan_v1_schema.py",
|
|
"temp_path": ROOT / "Temp" / "weekly_legacy_transfer_plan_v1.json",
|
|
},
|
|
}
|
|
|
|
|
|
def _read_text(path: Path) -> str:
|
|
if not path.exists():
|
|
return ""
|
|
return path.read_text(encoding="utf-8", errors="replace")
|
|
|
|
|
|
def _read_json(path: Path) -> dict[str, Any]:
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
return payload if isinstance(payload, dict) else {}
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def _contains_all(text: str, needles: list[str]) -> bool:
|
|
lower = text.lower()
|
|
return all(needle.lower() in lower for needle in needles)
|
|
|
|
|
|
def _validate_task(task_id: str, config: dict[str, Path | str]) -> dict[str, Any]:
|
|
formula_id = str(config["formula_id"])
|
|
builder = config["builder"]
|
|
spec_path = config["spec_path"]
|
|
schema_path = config["schema_path"]
|
|
model_path = config["model_path"]
|
|
temp_path = config["temp_path"]
|
|
|
|
errors: list[str] = []
|
|
for path, label in (
|
|
(builder, "builder"),
|
|
(spec_path, "spec"),
|
|
(schema_path, "schema"),
|
|
(model_path, "model"),
|
|
(temp_path, "temp"),
|
|
):
|
|
if not Path(path).exists():
|
|
errors.append(f"missing_{label}:{Path(path).relative_to(ROOT)}")
|
|
|
|
temp_doc = _read_json(Path(temp_path))
|
|
if temp_doc.get("formula_id") != formula_id:
|
|
errors.append(f"temp_formula_id:{temp_doc.get('formula_id')!r}")
|
|
if task_id == "P3-A" and float(temp_doc.get("state_vector_completeness_pct") or 0.0) < 0.0:
|
|
errors.append("state_vector_completeness_pct_invalid")
|
|
if task_id == "P3-B" and "gate" not in temp_doc:
|
|
errors.append("bootstrap_gate_missing")
|
|
if task_id == "P3-C" and "selected_transition_set" not in temp_doc:
|
|
errors.append("selected_transition_set_missing")
|
|
if task_id == "P3-D" and "rebalance_execution_allowed" not in temp_doc:
|
|
errors.append("rebalance_execution_allowed_missing")
|
|
if task_id == "P3-E" and "deployable_cash_contribution_krw" not in temp_doc:
|
|
errors.append("deployable_cash_contribution_missing")
|
|
|
|
return {
|
|
"formula_id": formula_id,
|
|
"builder_path": str(builder),
|
|
"spec_path": str(spec_path),
|
|
"schema_path": str(schema_path),
|
|
"model_path": str(model_path),
|
|
"temp_path": str(temp_path),
|
|
"gate": "PASS" if not errors else "FAIL",
|
|
"errors": errors,
|
|
"temp_doc": temp_doc,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
plan_text = _read_text(PLAN_PATH)
|
|
decision_flow_text = _read_text(DECISION_FLOW_PATH)
|
|
manifest = _read_text(MANIFEST_PATH)
|
|
plan_doc = yaml.safe_load(plan_text) if plan_text else {}
|
|
|
|
task_results: dict[str, dict[str, Any]] = {}
|
|
errors: list[str] = []
|
|
|
|
for task_id, config in TASKS.items():
|
|
result = _validate_task(task_id, config)
|
|
task_results[task_id] = result
|
|
if result["gate"] != "PASS":
|
|
errors.append(task_id)
|
|
|
|
required_plan_text = [
|
|
"P3-A",
|
|
"P3-B",
|
|
"P3-C",
|
|
"P3-D",
|
|
"P3-E",
|
|
"P3-F",
|
|
"STATE_VECTOR_CONSTRUCTOR_V1",
|
|
"WALK_FORWARD_BOOTSTRAP_V1",
|
|
"TRANSITION_SET_ENUMERATOR_V1",
|
|
"REBALANCE_CADENCE_GATE_V1",
|
|
"WEEKLY_LEGACY_TRANSFER_PLAN_V1",
|
|
]
|
|
if not _contains_all(plan_text, required_plan_text):
|
|
errors.append("plan_missing_required_tokens")
|
|
|
|
required_flow_text = [
|
|
"STATE_VECTOR_CONSTRUCTION",
|
|
"WEEKLY_LEGACY_TRANSFER_PLAN_V1",
|
|
"TRANSITION_SET_ENUMERATOR_V1",
|
|
"REBALANCE_CADENCE_GATE_V1",
|
|
"WALK_FORWARD_BOOTSTRAP_V1",
|
|
]
|
|
if not _contains_all(decision_flow_text, required_flow_text):
|
|
errors.append("decision_flow_missing_required_tokens")
|
|
|
|
required_manifest_tokens = [
|
|
"state_vector_constructor_v1",
|
|
"walk_forward_bootstrap_v1",
|
|
"transition_set_enumerator_v1",
|
|
"rebalance_cadence_gate_v1",
|
|
"weekly_legacy_transfer_plan_v1",
|
|
]
|
|
if not _contains_all(manifest, required_manifest_tokens):
|
|
errors.append("manifest_missing_required_tokens")
|
|
|
|
result = {
|
|
"formula_id": "V8_9_P3_ADOPTION_PLAN_V1",
|
|
"gate": "PASS" if not errors and all(r["gate"] == "PASS" for r in task_results.values()) else "FAIL",
|
|
"plan_path": str(PLAN_PATH),
|
|
"decision_flow_path": str(DECISION_FLOW_PATH),
|
|
"manifest_path": str(MANIFEST_PATH),
|
|
"task_results": task_results,
|
|
"errors": errors,
|
|
}
|
|
|
|
out = ROOT / "Temp" / "v8_9_p3_adoption_plan_v1.json"
|
|
out.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 result["gate"] == "PASS" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|