97 lines
3.6 KiB
Python
97 lines
3.6 KiB
Python
"""validate_realized_performance_v1.py — REALIZED_PERFORMANCE_VALIDATE_V1
|
|
|
|
Temp/realized_performance_v1.json의 존재와 기본 스키마를 검증한다.
|
|
데이터가 충분하지 않은 구간은 allow-listed insufficient_data 상태로만 허용한다.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DEFAULT_INPUT = ROOT / "Temp" / "realized_performance_v1.json"
|
|
DEFAULT_OUT = ROOT / "Temp" / "validate_realized_performance_v1.json"
|
|
FORMULA_ID = "REALIZED_PERFORMANCE_VALIDATE_V1"
|
|
|
|
|
|
def _load(path: Path) -> Any:
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def _is_nonempty_dict(v: Any) -> bool:
|
|
return isinstance(v, dict) and len(v) > 0
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--input", default=str(DEFAULT_INPUT))
|
|
ap.add_argument("--out", default=str(DEFAULT_OUT))
|
|
args = ap.parse_args()
|
|
|
|
input_path = Path(args.input)
|
|
input_path = input_path if input_path.is_absolute() else ROOT / input_path
|
|
out_path = Path(args.out)
|
|
out_path = out_path if out_path.is_absolute() else ROOT / out_path
|
|
|
|
payload = _load(input_path)
|
|
missing: list[dict[str, object]] = []
|
|
|
|
if not _is_nonempty_dict(payload):
|
|
missing.append({"field": "file", "reason": "missing_or_invalid_json"})
|
|
else:
|
|
if payload.get("formula_id") != "REALIZED_PERFORMANCE_V1":
|
|
missing.append({
|
|
"field": "formula_id",
|
|
"reason": "unexpected_value",
|
|
"expected": "REALIZED_PERFORMANCE_V1",
|
|
})
|
|
|
|
for field in ["performance_metrics", "current_portfolio_mdd", "insufficient_data_items", "summary"]:
|
|
if field not in payload:
|
|
missing.append({"field": field, "reason": "missing_key"})
|
|
|
|
perf = payload.get("performance_metrics")
|
|
if isinstance(perf, dict):
|
|
for key in ["t1_operational", "t5_operational", "t20_replay_estimated"]:
|
|
block = perf.get(key)
|
|
if not _is_nonempty_dict(block):
|
|
missing.append({"field": f"performance_metrics.{key}", "reason": "missing_block"})
|
|
continue
|
|
if "label" not in block or "n" not in block:
|
|
missing.append({"field": f"performance_metrics.{key}", "reason": "missing_label_or_n"})
|
|
|
|
summary = payload.get("summary")
|
|
if isinstance(summary, dict):
|
|
if "disclaimer" not in summary:
|
|
missing.append({"field": "summary.disclaimer", "reason": "missing_key"})
|
|
if "best_estimated_source" not in summary:
|
|
missing.append({"field": "summary.best_estimated_source", "reason": "missing_key"})
|
|
|
|
mdd = payload.get("current_portfolio_mdd")
|
|
if not _is_nonempty_dict(mdd):
|
|
missing.append({"field": "current_portfolio_mdd", "reason": "missing_block"})
|
|
else:
|
|
if "worst_case_scenario" not in mdd:
|
|
missing.append({"field": "current_portfolio_mdd.worst_case_scenario", "reason": "missing_key"})
|
|
|
|
result = {
|
|
"formula_id": FORMULA_ID,
|
|
"gate": "PASS" if not missing else "FAIL",
|
|
"checked_file": str(Path(args.input).as_posix()),
|
|
"missing": missing,
|
|
}
|
|
out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
return 0 if not missing else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|