Merge pull request #431 from easygap/feat/deployment-health

feat: 집계 배치율 감시 — 설계 대비 미달을 헬스로 표면화

Author: easygap

core/operator_health.py

@@ -126,12 +126,51 @@ def summarize_blockers(blockers: dict[str, Any] | None) -> dict[str, Any]:
     }
 
 
+def summarize_deployment(
+    deployment_ratio: float | None,
+    design_fraction: float | None,
+    *,
+    tolerance: float = 0.05,
+) -> dict[str, Any]:
+    """집계 배치율(총자산 중 실제 주식비중)이 설계 대비 크게 미달인지 판정(순수 함수).
+
+    한 달 운영 리뷰(docs/PAPER_MONTH1_REVIEW_AND_PLAN.md P1-5)의 배경: 종목별 드리프트
+    트리거는 '집계 배치율' 이탈(예: 실효 61% vs 설계 80%)을 영영 못 본다. 여기서 그 이탈을
+    운영자 헬스로 표면화한다. 실제가 설계보다 tolerance(기본 5%p) 초과로 낮으면 ATTENTION.
+    (초과 배치는 리밸런서가 자연 교정하므로 '미달'만 감시한다.)
+
+    반환: {verdict(OK|ATTENTION), note(str|None), deployment_ratio, design_fraction, shortfall}
+    """
+    if deployment_ratio is None or design_fraction is None:
+        return {
+            "verdict": "OK", "note": None,
+            "deployment_ratio": deployment_ratio, "design_fraction": design_fraction,
+            "shortfall": None,
+        }
+    shortfall = float(design_fraction) - float(deployment_ratio)
+    verdict, note = "OK", None
+    if shortfall > tolerance:
+        verdict = "ATTENTION"
+        note = (
+            f"주식 배치율 {deployment_ratio:.0%} < 설계 {design_fraction:.0%} "
+            f"({-shortfall * 100:.1f}%p) — 미체결 슬롯/자본 점검"
+        )
+    return {
+        "verdict": verdict, "note": note,
+        "deployment_ratio": float(deployment_ratio), "design_fraction": float(design_fraction),
+        "shortfall": shortfall,
+    }
+
+
 def summarize_basket_operation(
     enabled_baskets: list[str],
     last_snapshot_date: Any,
     position_count: int,
     today: Any,
     max_stale_calendar_days: int = 4,
+    deployment_ratio: float | None = None,
+    design_fraction: float | None = None,
+    deployment_tolerance: float = 0.05,
 ) -> dict[str, Any]:
     """바스켓 paper 운영(트랙레코드 축적) 상태를 verdict + 요약으로 환원한다.
 
@@ -158,6 +197,8 @@ def _as_date(v: Any):
             "last_snapshot_date": None,
             "position_count": int(position_count or 0),
             "stale_days": None,
+            "deployment_ratio": None,
+            "design_fraction": None,
             "notes": ["enabled 바스켓 없음(운영 안 함)"],
         }
 
@@ -176,12 +217,23 @@ def _as_date(v: Any):
                 "일일 사이클 중단 의심"
             )
 
+    # 집계 배치율 미달 감시 — 종목별 드리프트가 못 보는 설계 대비 이탈을 표면화.
+    dep = summarize_deployment(
+        deployment_ratio, design_fraction, tolerance=deployment_tolerance,
+    )
+    if dep["verdict"] == "ATTENTION":
+        verdict = "ATTENTION"
+        if dep["note"]:
+            notes.append(dep["note"])
+
     return {
         "verdict": verdict,
         "enabled_baskets": list(enabled_baskets),
         "last_snapshot_date": _as_date(last_snapshot_date) if last_snapshot_date is not None else None,
         "position_count": int(position_count or 0),
         "stale_days": stale_days,
+        "deployment_ratio": dep["deployment_ratio"],
+        "design_fraction": dep["design_fraction"],
         "notes": notes,
     }
 

docs/PAPER_MONTH1_REVIEW_AND_PLAN.md

@@ -168,7 +168,13 @@ P0-1(하트비트·재시도)을 먼저 머지하고 재시작하는 순서를 
 >   룩백 14일로 확대(명절 연휴 클러스터 커버), live 동기화 실패 시 CYCLE_ERROR 기록,
 >   run_rebalance 배선 통합 테스트 보강(스킵/critical 분기).
 > - ⬜ P0-2 SMTP 재발급 — 오너 액션
-> - ⬜ P1~P3 — 대기
+> - ✅ **P1-5 집계 배치율 감시** — `core/operator_health.py`에 순수 `summarize_deployment`
+>   추가, `summarize_basket_operation`이 설계 대비 -5%p 초과 미달 시 ATTENTION 승격.
+>   `--mode health`가 최신 스냅샷으로 배치율 계산(네트워크 불필요) → 종목별 드리프트가
+>   못 보던 집계 이탈을 표면화. 실CLI 스모크로 59%<80% ATTENTION 확인(스모크가 `config`
+>   미정의 NameError로 바스켓 헬스가 통째로 죽던 것도 적발·수정). 적대적 리뷰 반영:
+>   배치율 계산을 자체 try로 격리(bad config가 결측 감지를 삼키지 않게), 경계 테스트 보강.
+> - ⬜ P1-4/6/7 (귀속 자동화·스냅샷 시점 정합·주간 리포트), P2~P3 — 대기
 
 ### P0 — 자본 결정과 같은 주에
 

main.py

@@ -1631,11 +1631,30 @@ def run_health_check() -> int:
 
         from core.basket_rebalancer import rebalance_live_strategy_id
 
+        config = Config.get()
         enabled_baskets = BasketRebalancer.get_enabled_baskets()
+        baskets_cfg = BasketRebalancer._load_baskets_config()
+        min_cash_ratio = (
+            config.risk_params.get("diversification", {}).get("min_cash_ratio", 0.20)
+        )
+
+        def _design_fraction(cfg: dict) -> float:
+            """BasketRebalancer._stock_fraction과 동일 규칙(인스턴스 없이 계산)."""
+            max_stock = 1.0 - min_cash_ratio
+            tsw = cfg.get("target_stock_weight")
+            if tsw is None:
+                return max_stock
+            return max(0.0, min(float(tsw), max_stock))
+
         # 바스켓별 전용 계정 키(basket_rebalance:<name>) 기준으로 조회한다.
         # 복수 바스켓이면 '가장 오래된 최신 스냅샷'을 기준으로(가장 뒤처진 사이클 감시).
         last_dates = []
         position_count = 0
+        # 집계 배치율 미달 감시: 최신 스냅샷으로 실제 주식비중을 계산(네트워크 불필요),
+        # 설계 대비 가장 크게 미달인 바스켓을 대표로 넘긴다(가장 뒤처진 배치 감시).
+        worst_shortfall = None
+        worst_dep_ratio = None
+        worst_design = None
         session = get_session()
         try:
             for name in enabled_baskets:
@@ -1648,6 +1667,20 @@ def run_health_check() -> int:
                 )
                 last_dates.append(snap.date if snap else None)
                 position_count += len(get_all_positions(account_key=key) or [])
+                # 배치율은 부가 신호 — 계산 실패(예: baskets.yaml에 float 불가한
+                # target_stock_weight 오타)가 핵심 신호인 결측/staleness 감지를
+                # 통째로 삼키지 않도록 자체 try로 격리한다.
+                try:
+                    if snap and snap.total_value and snap.total_value > 0:
+                        dep_ratio = max(0.0, (snap.total_value - (snap.cash or 0)) / snap.total_value)
+                        design = _design_fraction(baskets_cfg.get(name) or {})
+                        shortfall = design - dep_ratio
+                        if worst_shortfall is None or shortfall > worst_shortfall:
+                            worst_shortfall = shortfall
+                            worst_dep_ratio = dep_ratio
+                            worst_design = design
+                except Exception as dep_exc:
+                    logger.debug("바스켓 '{}' 배치율 계산 생략: {}", name, dep_exc)
         finally:
             session.close()
         oldest_last = (
@@ -1659,6 +1692,8 @@ def run_health_check() -> int:
             "last_snapshot_date": oldest_last,
             "position_count": position_count,
             "today": date.today(),
+            "deployment_ratio": worst_dep_ratio,
+            "design_fraction": worst_design,
         }
     except Exception as exc:
         logger.warning("바스켓 운영 상태 조회 실패: {}", exc)

tests/test_operator_health.py

@@ -197,6 +197,55 @@ def test_datetime_inputs_are_normalized(self):
         )
         assert out["verdict"] == "OK"
 
+    def test_underdeployment_escalates_to_attention(self):
+        # 신선한 스냅샷이라도 배치율 미달(59% vs 설계 80%, -21%p > 5%p)이면 ATTENTION
+        out = self._summ(deployment_ratio=0.59, design_fraction=0.80)
+        assert out["verdict"] == "ATTENTION"
+        assert any("배치율" in n for n in out["notes"])
+        assert out["deployment_ratio"] == 0.59
+
+    def test_deployment_within_tolerance_stays_ok(self):
+        out = self._summ(deployment_ratio=0.77, design_fraction=0.80)  # -3%p ≤ 5%p
+        assert out["verdict"] == "OK"
+
+    def test_deployment_none_is_ok(self):
+        out = self._summ(deployment_ratio=None, design_fraction=None)
+        assert out["verdict"] == "OK"
+
+
+class TestSummarizeDeployment:
+    """집계 배치율 미달 판정(순수 함수)."""
+
+    def _f(self, ratio, design, **kw):
+        from core.operator_health import summarize_deployment
+        return summarize_deployment(ratio, design, **kw)
+
+    def test_shortfall_beyond_tolerance_is_attention(self):
+        out = self._f(0.61, 0.80)
+        assert out["verdict"] == "ATTENTION"
+        assert "61%" in out["note"] and "80%" in out["note"]
+
+    def test_within_tolerance_ok(self):
+        assert self._f(0.76, 0.80)["verdict"] == "OK"
+
+    def test_overdeployment_is_ok(self):
+        # 초과 배치는 리밸런서가 자연 교정 — 감시 대상 아님
+        assert self._f(0.90, 0.80)["verdict"] == "OK"
+
+    def test_none_inputs_ok(self):
+        assert self._f(None, 0.80)["verdict"] == "OK"
+        assert self._f(0.6, None)["verdict"] == "OK"
+
+    def test_custom_tolerance(self):
+        assert self._f(0.75, 0.80, tolerance=0.10)["verdict"] == "OK"   # -5%p ≤ 10%p
+        assert self._f(0.68, 0.80, tolerance=0.10)["verdict"] == "ATTENTION"  # -12%p
+
+    def test_exact_boundary_is_ok_strict_comparison(self):
+        # shortfall == tolerance 이면 OK(엄격 '>' 고정). tolerance=0으로 부동소수 오차 없이
+        # 경계를 못박는다 — '>'를 '>='로 바꾸면 완벽 배치도 ATTENTION이 되어 이 테스트가 잡는다.
+        assert self._f(0.80, 0.80, tolerance=0.0)["verdict"] == "OK"
+        assert self._f(0.79, 0.80, tolerance=0.0)["verdict"] == "ATTENTION"
+
 
 class TestBuildOperatorHealthWithBasket:
     def test_basket_attention_escalates_overall_verdict(self):