데이터 게이트 검증기와 DAG 연결

tools/validate_alpha_feedback_loop_v2.py

@@ -0,0 +1,92 @@
+"""validate_alpha_feedback_loop_v2.py — ALPHA_FEEDBACK_LOOP_VALIDATE_V2
+
+Temp/alpha_feedback_loop_v2.json의 구조와 데이터 게이트 상태를 검증한다.
+"""
+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" / "alpha_feedback_loop_v2.json"
+DEFAULT_OUT = ROOT / "Temp" / "validate_alpha_feedback_loop_v2.json"
+FORMULA_ID = "ALPHA_FEEDBACK_LOOP_VALIDATE_V2"
+
+
+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_dict(value: Any) -> bool:
+    return isinstance(value, dict)
+
+
+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)
+    errors: list[str] = []
+
+    if not _is_dict(payload):
+        errors.append("payload must be object")
+    else:
+        if payload.get("formula_id") != "ALPHA_FEEDBACK_LOOP_V2":
+            errors.append("formula_id mismatch")
+
+        status = str(payload.get("status") or "")
+        if status not in {"DATA_INSUFFICIENT", "ANALYZED"}:
+            errors.append(f"status={status}")
+
+        if "cases_analyzed" not in payload or not isinstance(payload.get("cases_analyzed"), int):
+            errors.append("cases_analyzed must be int")
+        if "recommended_adjustments" not in payload or not isinstance(payload.get("recommended_adjustments"), list):
+            errors.append("recommended_adjustments must be list")
+
+        cases = payload.get("cases_analyzed")
+        recs = payload.get("recommended_adjustments")
+        if isinstance(cases, int):
+            if cases < 10 and status != "DATA_INSUFFICIENT":
+                errors.append("cases_analyzed < 10 requires DATA_INSUFFICIENT")
+            if cases >= 10 and status != "ANALYZED":
+                errors.append("cases_analyzed >= 10 requires ANALYZED")
+        if isinstance(recs, list) and status == "DATA_INSUFFICIENT" and recs:
+            errors.append("DATA_INSUFFICIENT must not carry recommendations")
+        if status == "ANALYZED":
+            for key in [
+                "active_signal_rate_pct", "active_signal_n",
+                "passive_signal_rate_pct", "passive_signal_n",
+                "combined_rate_pct", "sell_signal_rate_pct", "sell_signal_n",
+                "pa1_current_ratio", "pa1_thesis_sum", "pa1_antithesis_sum",
+                "component_analysis", "note",
+            ]:
+                if key not in payload:
+                    errors.append(f"missing field: {key}")
+
+    result = {
+        "formula_id": FORMULA_ID,
+        "gate": "PASS" if not errors else "FAIL",
+        "checked_file": str(Path(args.input).as_posix()),
+        "errors": errors,
+    }
+    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 errors else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())