kjh2064/QuantEngineByItz
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

릴리스 DAG와 실현 성과 게이트 연결

Browse Source
This commit is contained in:
김재현
2026-06-18 01:48:17 +09:00
parent fee8131d5d
commit d7f9d3a944
7 changed files with 365 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/build_realized_performance_v1.py
+1 -1
View File
@@ -300,7 +300,7 @@ def main() -> int:
out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
s = result["summary"]
print(
f"[{FORMULA_ID}] T20_replay_CAGR(est)={s.get('best_estimated_cagr_pct')}% "
f"[{FORMULA_ID}] T20_replay_CAGR(est)={s.get('best_estimated_cagr')}% "
f"Sharpe(est)={s.get('best_estimated_sharpe')} "
f"MDD_worst={s.get('worst_case_mdd_pct')}% "
f"T5_win_rate={s.get('t5_win_rate_pct')}% -> {out_path}"
tools/run_release_dag_v3.py
+19 -6
View File
@@ -82,7 +82,7 @@ def main() -> int:
def add_to_closure(nid):
if nid not in closure:
closure.add(nid)
for dep in nodes[nid].get("depends_on") or []:
for dep in sorted(nodes[nid].get("depends_on") or []):
if dep in nodes:
add_to_closure(dep)
@@ -99,14 +99,14 @@ def main() -> int:
raise ValueError(f"Cycle detected involving {nid}")
if nid not in visited:
temp.add(nid)
for dep in nodes[nid].get("depends_on") or []:
for dep in sorted(nodes[nid].get("depends_on") or []):
if dep in closure:
visit(dep)
temp.remove(nid)
visited.add(nid)
order.append(nid)
for nid in closure:
for nid in sorted(closure):
if nid not in visited:
try:
visit(nid)
@@ -116,6 +116,8 @@ def main() -> int:
steps_run = []
success = True
warning_failed = False
warning_failures: list[dict[str, object]] = []
LINEAGE_LOG.parent.mkdir(parents=True, exist_ok=True)
@@ -180,6 +182,11 @@ def main() -> int:
print(proc.stderr)
if node.get("warn_only", False):
print(f"Node {nid} is warn_only - continuing")
warning_failed = True
warning_failures.append({
"node_id": nid,
"returncode": proc.returncode,
})
else:
success = False
if node.get("strict", True) or args.strict:
@@ -191,17 +198,23 @@ def main() -> int:
"formula_id": "RELEASE_DAG_RUN_V4",
"mode": args.mode,
"steps": steps_run,
"gate": "PASS" if success else "FAIL"
"warning_failures": warning_failures,
"gate": "PASS" if success and not warning_failed else "PASS_WITH_WARNINGS" if success else "FAIL"
}, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps({
"formula_id": "RELEASE_DAG_RUN_V4",
"mode": args.mode,
"step_count": len(steps_run),
"gate": "PASS" if success else "FAIL"
"warning_failure_count": len(warning_failures),
"gate": "PASS" if success and not warning_failed else "PASS_WITH_WARNINGS" if success else "FAIL"
}, ensure_ascii=True, indent=2))
return 0 if success else 1
if not success:
return 1
if args.strict and warning_failed:
return 1
return 0
if __name__ == "__main__":
tools/validate_completion_harness_instructions_v1.py
+148
View File
@@ -0,0 +1,148 @@
#!/usr/bin/env python3
from __future__ import annotations
import json
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def _read(rel: str) -> str:
path = ROOT / rel
if not path.exists():
return ""
return path.read_text(encoding="utf-8", errors="replace").lower()
def _require_any(text: str, options: list[str]) -> bool:
return any(option.lower() in text for option in options)
def _require_all(text: str, groups: list[list[str]]) -> bool:
return all(_require_any(text, group) for group in groups)
def main() -> int:
files: dict[str, list[list[str]]] = {
"AGENTS.md": [
["0b. 기본 하네스 완료 조건"],
["yaml"],
["코드"],
["데이터 실체"],
["검증 증빙"],
],
"docs/runbook.md": [
["complete only when", "완료"],
["yaml"],
["code"],
["data artifact", "data artifacts", "data/artifact"],
["validation evidence", "검증 증빙"],
],
"docs/ROADMAP_WBS.md": [
["완료 조건"],
["yaml"],
["코드"],
["데이터 실체"],
["검증 증빙"],
],
"REPORT_GUIDE.md": [
["completion harness"],
["yaml"],
["code"],
["data artifact", "data artifacts", "data/artifact"],
["validation evidence", "검증 증빙"],
],
"docs/doctrine.md": [
["completion harness"],
["yaml"],
["code"],
["data artifact", "data artifacts", "data/artifact"],
["validation evidence", "검증 증빙"],
],
"prompts/analysis_prompt.md": [
["default completion harness"],
["yaml"],
["코드"],
["데이터 실체"],
["검증 증빙"],
],
"prompts/review_prompt.md": [
["default completion harness"],
["yaml"],
["code"],
["data artifact", "data/artifact"],
["validation evidence", "검증 증빙"],
],
"prompts/capture_parse_prompt.md": [
["기본 완료 조건"],
["yaml"],
["코드"],
["데이터 실체"],
["검증 증빙"],
],
"prompts/engine_audit_master_prompt_v2.md": [
["default completion harness"],
["yaml"],
["code"],
["data artifact", "data/artifact"],
["validation evidence", "검증 증빙"],
],
"prompts/engine_audit_master_prompt_v3.md": [
["default completion harness"],
["yaml"],
["code"],
["data artifact", "data/artifact"],
["validation evidence", "검증 증빙"],
],
"prompts/engine_audit_prompt.md": [
["yaml"],
["code"],
["data artifact", "data/artifact"],
["validation evidence", "검증 증빙"],
],
"prompts/low_capability_report_renderer.md": [
["default completion harness"],
["yaml"],
["code"],
["data artifact", "data/artifact"],
["validation evidence", "검증 증빙"],
],
"prompts/report_renderer_prompt.md": [
["yaml"],
["code"],
["data artifact", "data/artifact"],
["validation evidence", "검증 증빙"],
],
"prompts/weekly_operational_report_master_prompt_v1.md": [
["default completion harness"],
["yaml"],
["code"],
["data artifact", "data/artifact"],
["validation evidence", "검증 증빙"],
],
}
missing: list[dict[str, object]] = []
for rel, groups in files.items():
text = _read(rel)
if not text:
missing.append({"file": rel, "reason": "missing_file"})
continue
if not _require_all(text, groups):
missing.append({"file": rel, "reason": "missing_required_text", "required_groups": groups})
result = {
"formula_id": "COMPLETION_HARNESS_INSTRUCTIONS_V1",
"gate": "PASS" if not missing else "FAIL",
"checked_files": len(files),
"missing": missing,
}
out = ROOT / "Temp" / "completion_harness_instructions_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 not missing else 1
if __name__ == "__main__":
raise SystemExit(main())
tools/validate_realized_performance_v1.py
+96
View File
@@ -0,0 +1,96 @@
"""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())