Merge pull request #384 from easygap/chore/operability-and-docs

운영 통합 헬스 점검 명령 추가 (운영 편의성)

Author: easygap

README.md

@@ -154,6 +154,11 @@ python main.py --mode dashboard
 # 기본 바인드는 http://127.0.0.1:8080 입니다.
 # 외부 공개가 필요할 때만 인증/reverse proxy 구성 후 --dashboard-host 0.0.0.0 을 명시하세요.
 
+# 운영 통합 헬스 점검 (전 전략 runtime + current_blockers 한눈에)
+python main.py --mode health
+# 종료코드 0=OK / 1=ATTENTION / 2=BLOCKED — 모니터링 스크립트에서 분기 가능.
+# 같은 점검을 tools/paper_runtime_status.py --health (--json) 으로도 실행할 수 있습니다.
+
 # 휴장일 갱신
 python main.py --update-holidays
 ```

core/operator_health.py

@@ -0,0 +1,168 @@
+"""운영자 아침 점검용 통합 헬스 요약.
+
+기존에는 운영자가 시스템 상태를 파악하려면 여러 도구를 따로 돌려야 했다:
+  - paper_runtime_status.py --all       (전략별 runtime state)
+  - evaluate_and_promote.py --check-only (artifact 동기화/freshness)
+  - current_blockers.json 직접 확인       (go_live, hard_blockers)
+
+이 모듈은 그 신호들을 하나로 모아 단일 verdict(OK / ATTENTION / BLOCKED)와
+사람이 읽는 요약을 만든다. 순수 함수라 외부 상태를 직접 읽지 않고 주입받은
+데이터로만 판정하므로 단위 테스트가 쉽다.
+
+verdict 규칙(보수적 — 의심스러우면 주의 이상):
+  - BLOCKED  : frozen 전략 존재, 또는 hard_blocker 존재, 또는 artifact가 stale/손상.
+  - ATTENTION: degraded/blocked_insufficient_evidence 전략 존재, 또는 manual freeze,
+               또는 최근 이상치(anomaly) 존재, 또는 go_live=false인데 live_candidate 표기 불일치.
+  - OK       : 위 어느 것에도 안 걸림.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+# verdict 우선순위 (높을수록 심각)
+_VERDICT_RANK = {"OK": 0, "ATTENTION": 1, "BLOCKED": 2}
+
+# runtime state별 분류
+_BLOCKING_STATES = {"frozen"}
+_ATTENTION_STATES = {"degraded", "blocked_insufficient_evidence"}
+
+
+def _worst(*verdicts: str) -> str:
+    """주어진 verdict 중 가장 심각한 것을 반환."""
+    worst = "OK"
+    for v in verdicts:
+        if _VERDICT_RANK.get(v, 0) > _VERDICT_RANK[worst]:
+            worst = v
+    return worst
+
+
+def summarize_runtime_state(state: Any) -> dict[str, Any]:
+    """단일 RuntimeState(또는 동등 객체)를 verdict + 요약으로 환원한다.
+
+    state는 .state / .strategy / .manual_freeze / .last_anomalies 속성을 가진 객체.
+    """
+    name = getattr(state, "strategy", "?")
+    s = getattr(state, "state", "unknown")
+    manual_freeze = bool(getattr(state, "manual_freeze", False))
+    anomalies = list(getattr(state, "last_anomalies", []) or [])
+
+    if s in _BLOCKING_STATES:
+        verdict = "BLOCKED"
+    elif s in _ATTENTION_STATES or manual_freeze or anomalies:
+        verdict = "ATTENTION"
+    else:
+        verdict = "OK"
+
+    notes = []
+    if s in _BLOCKING_STATES:
+        notes.append(f"state={s}")
+    elif s in _ATTENTION_STATES:
+        notes.append(f"state={s}")
+    if manual_freeze:
+        notes.append("manual_freeze")
+    if anomalies:
+        notes.append(f"anomalies={len(anomalies)}")
+
+    return {
+        "strategy": name,
+        "state": s,
+        "verdict": verdict,
+        "manual_freeze": manual_freeze,
+        "anomaly_count": len(anomalies),
+        "notes": notes,
+    }
+
+
+def summarize_blockers(blockers: dict[str, Any] | None) -> dict[str, Any]:
+    """current_blockers.json 페이로드를 verdict + 요약으로 환원한다."""
+    if not blockers:
+        return {
+            "verdict": "ATTENTION",
+            "go_live": False,
+            "hard_blocker_count": 0,
+            "notes": ["current_blockers 없음/로드 실패"],
+            "freshness_stale": True,
+        }
+
+    hard = list(blockers.get("hard_blockers") or [])
+    go_live = bool(blockers.get("go_live", False))
+    live_candidates = list(blockers.get("live_candidates") or [])
+    freshness = blockers.get("promotion_artifact_freshness") or {}
+    # freshness가 dict면 stale 여부를 본다(없으면 보수적으로 미상=stale 취급하지 않음).
+    stale = False
+    if isinstance(freshness, dict):
+        stale = bool(freshness.get("stale", False)) or freshness.get("status") in ("stale", "expired")
+
+    notes = []
+    verdict = "OK"
+    if hard:
+        verdict = "BLOCKED"
+        notes.append(f"hard_blockers={len(hard)}")
+    if stale:
+        verdict = _worst(verdict, "BLOCKED")
+        notes.append("artifact_stale")
+    # go_live=false인데 live_candidates가 비어있지 않으면 표기 불일치(주의).
+    if not go_live and live_candidates:
+        verdict = _worst(verdict, "ATTENTION")
+        notes.append("go_live=false_but_live_candidates_present")
+
+    return {
+        "verdict": verdict,
+        "go_live": go_live,
+        "live_candidates": live_candidates,
+        "hard_blocker_count": len(hard),
+        "hard_blockers": hard,
+        "freshness_stale": stale,
+        "notes": notes,
+    }
+
+
+def build_operator_health(
+    runtime_states: list[Any],
+    blockers: dict[str, Any] | None,
+) -> dict[str, Any]:
+    """전략별 runtime state + current_blockers를 하나의 헬스 요약으로 합친다.
+
+    반환:
+      {
+        "verdict": "OK" | "ATTENTION" | "BLOCKED",
+        "strategy_count": N,
+        "strategies": [summarize_runtime_state(...), ...],
+        "blockers": summarize_blockers(...),
+        "headline": 사람이 읽는 한 줄 요약,
+        "attention_items": [...],  # 운영자가 봐야 할 항목들
+      }
+    """
+    strat_summaries = [summarize_runtime_state(s) for s in runtime_states]
+    blocker_summary = summarize_blockers(blockers)
+
+    verdict = "OK"
+    for s in strat_summaries:
+        verdict = _worst(verdict, s["verdict"])
+    verdict = _worst(verdict, blocker_summary["verdict"])
+
+    attention_items: list[str] = []
+    for s in strat_summaries:
+        if s["verdict"] != "OK":
+            attention_items.append(f"{s['strategy']}: {', '.join(s['notes']) or s['state']}")
+    if blocker_summary["notes"]:
+        attention_items.append("blockers: " + ", ".join(blocker_summary["notes"]))
+
+    n = len(strat_summaries)
+    n_ok = sum(1 for s in strat_summaries if s["verdict"] == "OK")
+    if verdict == "OK":
+        headline = f"전체 정상 — 전략 {n}개 모두 OK, go_live={blocker_summary['go_live']}"
+    elif verdict == "ATTENTION":
+        headline = f"주의 필요 — 전략 {n}개 중 {n_ok}개 OK, 확인 항목 {len(attention_items)}건"
+    else:
+        headline = f"차단 상태 — 운영 개입 필요, 확인 항목 {len(attention_items)}건"
+
+    return {
+        "verdict": verdict,
+        "strategy_count": n,
+        "strategies": strat_summaries,
+        "blockers": blocker_summary,
+        "headline": headline,
+        "attention_items": attention_items,
+    }

docs/PAPER_TO_LIVE_RUNBOOK.md

@@ -27,8 +27,11 @@
 ```bash
 # 항상 프로젝트 venv 사용 (bare python은 의존성 없음)
 .venv\Scripts\python.exe -m pytest tests/ -q          # 전체 그린 확인
+.venv\Scripts\python.exe tools/paper_runtime_status.py --health   # 통합 헬스 한눈에(아침 점검)
 .venv\Scripts\python.exe tools/evaluate_and_promote.py --check-only   # 운영 artifact 동기화/freshness
 ```
+- `--health`는 전 전략 runtime state + `current_blockers`를 모아 단일 verdict(OK/ATTENTION/BLOCKED)와
+  종료코드(0/1/2)를 준다. 모니터링 스크립트에서 종료코드로 분기하거나 `--json`으로 파싱할 수 있다.
 - `--check-only`가 FAIL(stale)이면 §3의 `--canonical` 재실행 필요.
 - 테스트는 임시 DB로 격리되어 **운영 `data/quant_trader.db`를 건드리지 않는다**
   (이전 "DB 복구(restore)" 반복의 근본 원인이었음 — `tests/conftest.py`로 차단).
@@ -86,7 +89,13 @@ MDD/PF/turnover 게이트 통과 + **정직한(생존자 통제) 백테스트 
 
 1. **시점 유니버스 재검증**(#5): pykrx 동작 환경에서 `--canonical` → `survivorship_controlled=true`로
    target_weight 진짜 엣지 측정. 헤드라인 대비 얼마나 남는지 판정.
-2. **진짜 OOS holdout**(#6): 함수는 구현됨 — `research_candidate_sweep.evaluate_oos_holdout()`
-   (train 구간 rank_score로만 변형 선택 → untouched test 구간 성과/degradation 보고).
-   남은 일: `run_candidate_sweep`에 `--oos-holdout-split` 옵션으로 연결 + 정직한 유니버스 환경에서 실행.
+2. **진짜 OOS holdout**(#6): 구현·연결 완료. sweep 실행 시 `--oos-holdout-split YYYY-MM-DD`를
+   주면 그 이전(train)으로만 변형을 고르고 untouched 그 이후(test) 성과/degradation을
+   artifact의 `oos_holdout`에 보고한다. 예:
+   ```bash
+   .venv\Scripts\python.exe tools/research_candidate_sweep.py \
+     --candidate-family target_weight_drawdown_guard --top-n 200 --oos-holdout-split 2025-01-01
+   ```
+   `oos_holdout.holdout_passes=false` 또는 `sharpe_degradation`이 크면 선택 과적합 신호.
+   남은 일: 정직한(생존자 통제) 유니버스 환경에서 실제 실행해 degradation 수치 확보.
 3. **deflated Sharpe 게이트화**: 현재 report-only인 DSR/생존자 경고를 승격 게이트에 연결 검토.

main.py

@@ -1360,6 +1360,53 @@ def _get_strategy(name: str):
     return create_strategy(name)
 
 
+def run_health_check() -> int:
+    """운영 통합 헬스 점검 — 전 전략 runtime state + current_blockers를 한 번에 요약.
+
+    tools/paper_runtime_status.py --health 와 동일한 로직. main.py 진입점에서도
+    바로 쓸 수 있게 노출한다. 반환 코드: 0=OK, 1=ATTENTION, 2=BLOCKED.
+    """
+    import json
+    from pathlib import Path
+    from database.models import init_database
+    from core.paper_runtime import get_paper_runtime_state
+    from core.strategy_universe import get_paper_strategy_names
+    from core.operator_health import build_operator_health
+
+    init_database()
+    strategies = sorted(get_paper_strategy_names() or [])
+    states = []
+    for strategy in strategies:
+        try:
+            states.append(get_paper_runtime_state(strategy))
+        except Exception as exc:
+            logger.warning("{} 상태 조회 실패: {}", strategy, exc)
+
+    blockers = None
+    blockers_path = Path(__file__).resolve().parent / "reports" / "current_blockers.json"
+    if blockers_path.exists():
+        try:
+            blockers = json.loads(blockers_path.read_text(encoding="utf-8"))
+        except Exception:
+            blockers = None
+
+    health = build_operator_health(states, blockers)
+    icon = {"OK": "✅", "ATTENTION": "⚠️", "BLOCKED": "⛔"}.get(health["verdict"], "❓")
+    logger.info("{} 운영 헬스: {} — {}", icon, health["verdict"], health["headline"])
+    b = health["blockers"]
+    logger.info(
+        "go_live={} | hard_blockers={} | artifact_stale={}",
+        b["go_live"], b["hard_blocker_count"], b["freshness_stale"],
+    )
+    for s in health["strategies"]:
+        extra = f" ({', '.join(s['notes'])})" if s["notes"] else ""
+        logger.info("  - {}: {}{}", s["strategy"], s["state"], extra)
+    for item in health["attention_items"]:
+        logger.info("  확인: {}", item)
+
+    return {"OK": 0, "ATTENTION": 1, "BLOCKED": 2}.get(health["verdict"], 1)
+
+
 def main():
     """메인 진입점"""
     parser = argparse.ArgumentParser(
@@ -1438,8 +1485,9 @@ def main():
             "check_correlation",
             "check_ensemble_correlation",
             "rebalance",
+            "health",
         ],
-        help="실행 모드. backtest_momentum_top: 모멘텀 상위 동일비중 멀티종목. portfolio_backtest: 멀티종목 포트폴리오 백테스트. paper: 워치리스트 1회. schedule: 모의 스케줄 무한 루프(상시 서버). rebalance: 바스켓 리밸런싱.",
+        help="실행 모드. backtest_momentum_top: 모멘텀 상위 동일비중 멀티종목. portfolio_backtest: 멀티종목 포트폴리오 백테스트. paper: 워치리스트 1회. schedule: 모의 스케줄 무한 루프(상시 서버). rebalance: 바스켓 리밸런싱. health: 운영 통합 헬스 점검(전 전략 runtime + blockers).",
     )
     from strategies import get_strategy_names
     parser.add_argument(
@@ -1611,6 +1659,8 @@ def main():
             run_check_ensemble_correlation(args)
         elif args.mode == "rebalance":
             run_rebalance(args)
+        elif args.mode == "health":
+            raise SystemExit(run_health_check())
         else:
             logger.error("알 수 없는 모드: {}", args.mode)
     except KeyboardInterrupt: