눈금 투자 기록 화면 개편과 위험 판단 자료 검증 보완

README.md

@@ -6,14 +6,15 @@
 
 정해 둔 비중에 맞춰 주식과 ETF를 사고팔고, 원금·수익률·보유 종목을 한 화면에서 확인하는 개인용 투자 도구입니다. 과거 데이터로 운용 규칙을 비교하고, 모의투자로 실제 동작을 점검할 수 있습니다.
 
-![눈금의 계좌 현황 화면 — 총자산, 투자원금, 수익률과 자산 추이](docs/images/dashboard-account-20260917.png)
+![날짜 눈금으로 살펴보는 눈금의 실제 투자 기록](docs/images/dashboard-account-20260922.png)
 
-*2026년 9월 17일 모의투자 기록을 불러온 실제 화면입니다. 아래 백테스트와는 별도의 계좌 기록입니다.*
+*2026년 9월 22일 모의투자 기록을 불러온 실제 화면입니다. 아래 백테스트와는 별도의 계좌 기록입니다.*
 
 ## 할 수 있는 일
 
 - **계좌 확인** — 총자산, 투자원금, 현금, 보유 종목과 목표 비중을 확인합니다.
-- **성과 확인** — 적립금을 수익에서 제외한 수익률, 자산 추이, 고점 대비 하락률을 봅니다.
+- **날짜별 성과 탐색** — 날짜 눈금을 움직이면 그날의 수익률·원금·평가금액이 함께 바뀝니다. 적립금은 수익에서 제외합니다.
+- **입체·평면 차트와 CSV** — 같은 투자 기록을 두 방식으로 살펴보고, 조회한 기간의 원본 수치를 내려받습니다.
 - **자동 리밸런싱** — 실제 비중이 설정 범위를 벗어나면 조정합니다. 최소 주문금액과 1주 단위를 반영합니다.
 - **모의투자 검증** — 운영 기록, 누락된 자산 기록, 거래비용과 미해결 주문을 확인합니다.
 - **운영 점검** — 자동매매 실행 여부, 거래 중지 상태, 증권사 연결과 데이터 갱신 상태를 봅니다.
@@ -22,11 +23,17 @@ Python·aiohttp·SQLite·한국투자증권 KIS API를 사용합니다. 화면
 
 ## 화면
 
+### 날짜를 고르면 그날의 계좌가 보입니다
+
+파란 차트의 띠는 실제 누적 수익률입니다. 날짜 눈금은 마우스·터치·방향키로 움직일 수 있습니다. 입체감은 선에 폭을 준 표현이며, 별도의 지표를 뜻하지 않습니다. 정확한 값은 평면 차트와 일별 기록 표에서도 확인할 수 있습니다.
+
+한글 글꼴은 로컬에서 읽고, 입체 표현은 가벼운 Canvas로 그립니다. 흰색·검정·코발트·시안의 역할과 참고한 실제 디자인 사례는 [화면 설계와 검증 기록](docs/DESIGN_20260922.md)에 정리했습니다.
+
 ### 보유 종목과 비중
 
 매입금액과 목표 비중을 나란히 보여 줍니다. 계좌를 바꾸면 요약, 종목표, 차트, 적립금 입력 대상이 함께 바뀝니다.
 
-![대형주 분산 투자 계좌의 보유 종목](docs/images/dashboard-holdings-20260917.png)
+![대형주 분산 투자 계좌의 보유 종목과 투자 비중의 기준](docs/images/dashboard-holdings-20260922.png)
 
 ### 모바일과 운영 상태
 
@@ -35,14 +42,14 @@ Python·aiohttp·SQLite·한국투자증권 KIS API를 사용합니다. 화면
 <details>
 <summary>모바일 화면 보기</summary>
 
-<img src="docs/images/dashboard-mobile-20260917.png" alt="모바일 계좌 현황 — 잔액, 현금, 원금, 수익률" width="390">
+<img src="docs/images/dashboard-mobile-20260922.png" alt="모바일 투자 기록 — 날짜 탐색, 원금, 수익률, 보유 종목과 운영 상태" width="390">
 
 </details>
 
 <details>
 <summary>자동매매 상태 화면 보기</summary>
 
-![거래 안전, 장 상태, 자동매매와 데이터 갱신 확인](docs/images/dashboard-operations-20260917.png)
+![거래 안전, 장 상태, 자동매매와 데이터 갱신 확인](docs/images/dashboard-operations-20260922.png)
 
 </details>
 
@@ -61,21 +68,23 @@ ETF 적립의 기본 목표는 주식 ETF 47.5%·CD금리 ETF 47.5%·현금 5%
 
 적립금이 들어오면 낙폭이 작아 보이던 백테스트 계산을 고쳤습니다. 이어서 추세·낙폭 조건이 겹칠 때 비중을 중복해서 줄이지 않고, 주식 축소분을 기존 CD금리 ETF로 옮기도록 바꿨습니다.
 
-![동일 기간·비용으로 비교한 세 가지 운용 방식의 수익과 낙폭](docs/images/risk-review-20260917.png)
+![동일 기간·비용으로 비교한 세 가지 운용 방식의 수익과 낙폭](docs/images/risk-review-20260922.png)
+
+**2020-07-07~2026-09-17, 실제 ETF 종가·1주 단위·월 10만원 적립 비교**
 
-**2020-07-07~2026-09-16, 실제 ETF 종가·1주 단위·월 10만원 적립 비교**
+9월 22일 수집 자료에서 지수 종가가 17일까지 제공되어, 모든 자료가 있는 17일까지 비교했습니다. ETF에만 있는 18일·21일 기록을 섞어 기간을 늘리지 않았습니다.
 
 | 방식 | 연환산 수익률 | 최대 낙폭 | 샤프² |
 |---|---:|---:|---:|
-| 고정 비중 | 13.54% | -21.72% | 0.80 |
-| 기존 위험 관리¹ | 10.74% | -16.02% | 0.74 |
-| 변경한 위험 관리 | 12.97% | -15.74% | 0.90 |
+| 고정 비중 | 13.53% | -21.72% | 0.79 |
+| 기존 위험 관리¹ | 10.73% | -16.02% | 0.74 |
+| 변경한 위험 관리 | 12.96% | -15.74% | 0.90 |
 
 ¹ 기존 방식도 적립금 계산 오류를 수정한 뒤 같은 조건으로 다시 계산했습니다. ² 샤프 계산의 기준금리는 연 3%로 고정했습니다.
 
 변경한 방식은 이 전체 기간에서 기존 방식보다 수익률이 높고 최대 낙폭이 작았습니다. **고정 비중보다 수익률은 낮았고, 2023~2025년에는 기존 방식보다 낙폭이 컸습니다.** 수수료·슬리피지와 CD ETF의 보수적 세금 근사를 반영했지만, ETF 분배금·실시간 호가·미체결은 재현하지 못했습니다. 이미 살펴본 과거 자료로 비교한 결과이며 향후 수익을 보장하지 않습니다.
 
-기간별 결과, 비용 3배 조건, 2014년부터의 보조 실험, 2026년 9월까지 확인한 자료와 코드 변경 근거는 [위험 관리 검증 보고서](docs/RISK_REVIEW_20260917.md)에 정리했습니다.
+9월 22일에는 오래된 시세로 주식 비중을 다시 늘리던 가능성을 막고, 모의·실전 계좌의 위험 판단 기록을 분리했습니다. 백테스트 계산도 같은 일별 결과를 유지하면서 빨라졌습니다. [최신 검증과 자료 기준일](docs/RISK_REVIEW_20260922.md), [운용 규칙을 변경한 근거](docs/RISK_REVIEW_20260917.md)에서 결과와 한계를 확인할 수 있습니다.
 
 ## 시작하기
 
@@ -101,14 +110,14 @@ Copy-Item .env.example .env
 .\.venv\Scripts\python.exe main.py --mode rebalance --basket kr_pocket
 
 # 같은 조건으로 백테스트 재현
-.\.venv\Scripts\python.exe tools/risk_review.py --as-of 2026-09-17
+.\.venv\Scripts\python.exe tools/risk_review.py --as-of 2026-09-22
 ```
 
 매일 자동 실행하려면 `main.py --mode schedule`을 사용합니다. KIS API 키 등 개인 설정은 `.env`에 넣고 Git에는 올리지 않습니다. 대시보드는 계좌 정보를 표시하므로 기본 설정대로 이 PC에서만 접속해 사용합니다.
 
 ## 자세히 보기
 
-- [위험 관리 검증과 한계](docs/RISK_REVIEW_20260917.md)
-- [화면 설계·한글 글꼴·성능 측정](docs/DASHBOARD_REVIEW_20260917.md)
+- [위험 관리 검증과 한계](docs/RISK_REVIEW_20260922.md)
+- [화면 설계·참고한 디자인·성능 측정](docs/DESIGN_20260922.md)
 - [모의투자 평가 기준](docs/BASKET_PAPER_EVALUATION.md) · [실전 전환 절차](docs/PAPER_TO_LIVE_RUNBOOK.md)
 - [거래 안전장치](docs/SAFETY_MODEL.md) · [프로젝트 구조](docs/PROJECT_GUIDE.md)

core/basket_rebalancer.py

@@ -195,10 +195,14 @@ def overlay_decision(self):
             self._overlay_decision = None
         if not self._overlay_cfg.any_enabled:
             return None
-        if getattr(self, "_overlay_decision", None) is not None:
+        evaluation_date = datetime.now(_KST).date()
+        if (getattr(self, "_overlay_decision", None) is not None
+                and getattr(self, "_overlay_evaluation_date", None) == evaluation_date):
             return self._overlay_decision
 
-        prev = load_overlay_state(self.basket_name)
+        self._overlay_input_issues = []
+        self._overlay_source_dates = {}
+        prev = load_overlay_state(self.basket_name, mode=self._ledger_mode())
         index_closes = None
         if self._overlay_cfg.trend.enabled:
             index_closes = self._fetch_index_closes(
@@ -215,8 +219,10 @@ def overlay_decision(self):
             daily_returns=daily,
             prev_state=prev,
         )
+        decision.data_issues.extend(self._overlay_input_issues)
+        decision.source_dates = dict(self._overlay_source_dates)
         try:
-            save_overlay_state(self.basket_name, decision)
+            save_overlay_state(self.basket_name, decision, mode=self._ledger_mode())
         except OSError as exc:
             logger.warning("바스켓 '{}' 오버레이 상태 저장 실패: {}", self.basket_name, exc)
         if decision.data_issues:
@@ -226,8 +232,61 @@ def overlay_decision(self):
         else:
             logger.info("바스켓 '{}' 리스크 오버레이 계산 완료 (배수 {})", self.basket_name, decision.scale)
         self._overlay_decision = decision
+        self._overlay_evaluation_date = evaluation_date
         return decision
 
+    def _overlay_previous_session(self):
+        """한국 거래일 기준 직전 완료 일봉. 주말·설정된 휴장일은 건너뛴다."""
+        from core.trading_hours import TradingHours
+
+        calendar = TradingHours(self.config)
+        day = datetime.now(_KST) - timedelta(days=1)
+        for _ in range(31):
+            if calendar.is_trading_day(day):
+                return day.date()
+            day -= timedelta(days=1)
+        raise ValueError("직전 거래일을 확인할 수 없습니다")
+
+    def _overlay_data_issue(self, message: str):
+        if not hasattr(self, "_overlay_input_issues"):
+            self._overlay_input_issues = []
+        self._overlay_input_issues.append(message)
+
+    def _overlay_dated_frame(self, frame, source: str):
+        """날짜 오류·중복·오래된 자료는 버림 없이 전체 입력을 보류한다."""
+        import pandas as pd
+
+        try:
+            values = frame["date"] if "date" in frame.columns else frame.index
+            if "date" not in frame.columns and not isinstance(frame.index, pd.DatetimeIndex):
+                raise ValueError("날짜 열 없음")
+            dates = pd.to_datetime(values, errors="coerce")
+            parsed = [
+                (value.tz_convert(_KST) if value.tzinfo is not None else value).date()
+                if not pd.isna(value) else None
+                for value in dates
+            ]
+            if any(value is None for value in parsed):
+                raise ValueError("날짜를 읽지 못한 행 있음")
+            today = datetime.now(_KST).date()
+            keep = [value < today for value in parsed]
+            result = frame.loc[keep].copy()
+            result["_overlay_date"] = [value for value in parsed if value < today]
+            result = result.sort_values("_overlay_date")
+            if result["_overlay_date"].duplicated().any():
+                raise ValueError("같은 날짜의 기록이 둘 이상 있음")
+            expected = self._overlay_previous_session()
+            latest = result["_overlay_date"].iloc[-1] if not result.empty else None
+            if not hasattr(self, "_overlay_source_dates"):
+                self._overlay_source_dates = {}
+            self._overlay_source_dates[source] = str(latest) if latest else None
+            if latest != expected:
+                raise ValueError(f"최근 기록 {latest or '없음'}, 필요한 기준일 {expected}")
+            return result
+        except (ValueError, TypeError, OverflowError) as exc:
+            self._overlay_data_issue(f"{source}: {exc}. 자료 확인 전 비중 확대 보류")
+            return None
+
     def _fetch_index_closes(self, symbol: str, ma_days: int) -> list[float] | None:
         """추세 판단용 지수 종가(오래된 순). 오늘 날짜 봉은 제외 — 전일까지의 정보만."""
         start, end = self._recent_range(int(ma_days * 1.7) + 30)
@@ -238,26 +297,8 @@ def _fetch_index_closes(self, symbol: str, ma_days: int) -> list[float] | None:
             return None
         if df is None or df.empty or "close" not in df.columns:
             return None
-        frame = df.copy()
-        today = datetime.now(_KST).date()
-        dates = None
-        if "date" in frame.columns:
-            dates = frame["date"]
-        elif hasattr(frame.index, "date"):
-            dates = frame.index.to_series()
-        if dates is not None:
-            try:
-                import pandas as pd
-                parsed = pd.to_datetime(dates).dt.date if hasattr(pd.to_datetime(dates), "dt") else pd.to_datetime(dates).date
-                mask = [d < today for d in parsed]
-                frame = frame[mask].copy()
-                frame["_overlay_date"] = [d for d, keep in zip(parsed, mask) if keep]
-                frame = frame.sort_values("_overlay_date")
-                if frame["_overlay_date"].duplicated().any():
-                    return None
-            except Exception:
-                return None
-        else:
+        frame = self._overlay_dated_frame(df, "지수 종가")
+        if frame is None:
             return None
         # 누락 봉을 삭제하면 이동평균 창이 과거로 밀려 잘못 복귀할 수 있다.
         try:
@@ -278,11 +319,9 @@ def _nav_series_for_overlay(self) -> tuple[list[float] | None, list[float] | Non
             return None, None
         if snaps is None or getattr(snaps, "empty", True) or "cumulative_return" not in snaps.columns:
             return None, None
-        if "date" in snaps.columns:
-            import pandas as pd
-            snaps = snaps.copy()
-            snaps["date"] = pd.to_datetime(snaps["date"], errors="coerce")
-            snaps = snaps[snaps["date"].dt.date < datetime.now(_KST).date()].sort_values("date")
+        snaps = self._overlay_dated_frame(snaps, "계좌 기록")
+        if snaps is None:
+            return None, None
         try:
             cumulative = [float(v) if v is not None else float("nan") for v in snaps["cumulative_return"].tolist()]
         except (TypeError, ValueError):

core/risk_overlays.py

@@ -71,6 +71,7 @@ class OverlayDecision:
     vol_scale: float | None = None
     realized_vol: float | None = None
     evaluated_at: str = ""
+    source_dates: dict[str, str | None] = field(default_factory=dict)
 
     def to_dict(self) -> dict[str, Any]:
         return asdict(self)
@@ -229,7 +230,7 @@ def compute_decision(
     prev_state: dict[str, Any] | None = None,
     now: datetime | None = None,
 ) -> OverlayDecision:
-    """설정된 오버레이를 모두 평가해 배수 하나로 합친다(곱)."""
+    """설정한 결합 방식에 따라 하나의 위험 배수를 계산한다."""
     prev = prev_state or {}
     decision = OverlayDecision(evaluated_at=(now or datetime.now()).isoformat(timespec="seconds"))
     scale = 1.0
@@ -358,21 +359,49 @@ def overlay_state_path(basket_name: str, state_dir: str | os.PathLike | None = N
     return overlay_state_dir(state_dir) / f"{safe}.json"
 
 
-def load_overlay_state(basket_name: str, state_dir: str | os.PathLike | None = None) -> dict[str, Any] | None:
-    path = overlay_state_path(basket_name, state_dir)
+def load_overlay_state(
+    basket_name: str, state_dir: str | os.PathLike | None = None, *, mode: str | None = None,
+) -> dict[str, Any] | None:
+    directory = overlay_state_dir(state_dir)
+    if mode is not None and mode not in {"paper", "live"}:
+        raise ValueError("위험 관리 기록의 모드는 paper 또는 live여야 합니다")
+    path = overlay_state_path(basket_name, directory / mode if mode else directory)
+    # 이전 버전의 공용 기록은 모의투자만 이어받는다. 실전 계좌로 전파하지 않는다.
+    if mode == "paper" and not path.exists():
+        path = overlay_state_path(basket_name, directory)
     if not path.exists():
         return None
     try:
         data = json.loads(path.read_text(encoding="utf-8"))
     except (OSError, ValueError):
         return None
-    return data if isinstance(data, dict) else None
+    if not isinstance(data, dict):
+        return None
+    try:
+        scale = float(data["scale"])
+        if not math.isfinite(scale) or not 0 <= scale <= 1:
+            return None
+    except (KeyError, TypeError, ValueError):
+        return None
+    for key in ("trend_below", "drawdown_active"):
+        if data.get(key) is not None and not isinstance(data[key], bool):
+            return None
+    for key in ("reasons", "data_issues"):
+        if key in data and (not isinstance(data[key], list)
+                            or any(not isinstance(item, str) for item in data[key])):
+            return None
+    data["scale"] = scale
+    return data
 
 
 def save_overlay_state(
     basket_name: str, decision: OverlayDecision, state_dir: str | os.PathLike | None = None,
+    *, mode: str | None = None,
 ) -> Path:
-    path = overlay_state_path(basket_name, state_dir)
+    directory = overlay_state_dir(state_dir)
+    if mode is not None and mode not in {"paper", "live"}:
+        raise ValueError("위험 관리 기록의 모드는 paper 또는 live여야 합니다")
+    path = overlay_state_path(basket_name, directory / mode if mode else directory)
     path.parent.mkdir(parents=True, exist_ok=True)
     payload = decision.to_dict()
     payload["basket"] = str(basket_name)