lotto-lab: DB 스키마 확장 — purchase_history ALTER + strategy 테이블 추가

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

backend/app/db.py

@@ -197,6 +197,69 @@ def init_db() -> None:
         )
         conn.execute("CREATE INDEX IF NOT EXISTS idx_purchase_draw ON purchase_history(draw_no DESC);")
 
+        # ── purchase_history 컬럼 확장 (기존 데이터 보존) ──────────────────────
+        _ensure_column(conn, "purchase_history", "numbers",
+                       "ALTER TABLE purchase_history ADD COLUMN numbers TEXT NOT NULL DEFAULT '[]'")
+        _ensure_column(conn, "purchase_history", "is_real",
+                       "ALTER TABLE purchase_history ADD COLUMN is_real INTEGER NOT NULL DEFAULT 1")
+        _ensure_column(conn, "purchase_history", "source_strategy",
+                       "ALTER TABLE purchase_history ADD COLUMN source_strategy TEXT NOT NULL DEFAULT 'manual'")
+        _ensure_column(conn, "purchase_history", "source_detail",
+                       "ALTER TABLE purchase_history ADD COLUMN source_detail TEXT NOT NULL DEFAULT '{}'")
+        _ensure_column(conn, "purchase_history", "checked",
+                       "ALTER TABLE purchase_history ADD COLUMN checked INTEGER NOT NULL DEFAULT 0")
+        _ensure_column(conn, "purchase_history", "results",
+                       "ALTER TABLE purchase_history ADD COLUMN results TEXT NOT NULL DEFAULT '[]'")
+        _ensure_column(conn, "purchase_history", "total_prize",
+                       "ALTER TABLE purchase_history ADD COLUMN total_prize INTEGER NOT NULL DEFAULT 0")
+
+        # ── strategy_performance 테이블 ────────────────────────────────────────
+        conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS strategy_performance (
+                id            INTEGER PRIMARY KEY AUTOINCREMENT,
+                strategy      TEXT    NOT NULL,
+                draw_no       INTEGER NOT NULL,
+                sets_count    INTEGER NOT NULL DEFAULT 0,
+                total_correct INTEGER NOT NULL DEFAULT 0,
+                max_correct   INTEGER NOT NULL DEFAULT 0,
+                prize_total   INTEGER NOT NULL DEFAULT 0,
+                avg_score     REAL    NOT NULL DEFAULT 0.0,
+                updated_at    TEXT    NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
+                UNIQUE(strategy, draw_no)
+            );
+            """
+        )
+
+        # ── strategy_weights 테이블 ────────────────────────────────────────────
+        conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS strategy_weights (
+                id               INTEGER PRIMARY KEY AUTOINCREMENT,
+                strategy         TEXT    NOT NULL UNIQUE,
+                weight           REAL    NOT NULL DEFAULT 0.2,
+                ema_score        REAL    NOT NULL DEFAULT 0.15,
+                total_sets       INTEGER NOT NULL DEFAULT 0,
+                total_hits_3plus INTEGER NOT NULL DEFAULT 0,
+                updated_at       TEXT    NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
+            );
+            """
+        )
+
+        # strategy_weights 초기값 시드 (이미 있으면 무시)
+        _INIT_WEIGHTS = [
+            ("combined", 0.30, 0.15),
+            ("simulation", 0.25, 0.15),
+            ("heatmap", 0.20, 0.15),
+            ("manual", 0.15, 0.15),
+            ("custom", 0.10, 0.15),
+        ]
+        for strat, w, ema in _INIT_WEIGHTS:
+            conn.execute(
+                "INSERT OR IGNORE INTO strategy_weights (strategy, weight, ema_score) VALUES (?, ?, ?)",
+                (strat, w, ema),
+            )
+
         # ── weekly_reports 캐시 테이블 ──────────────────────────────────────────
         conn.execute(
             """

backend/tests/test_purchase_manager.py

@@ -0,0 +1,69 @@
+# backend/tests/test_purchase_manager.py
+import sys, os
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app"))
+
+import sqlite3
+import pytest
+from unittest.mock import patch, MagicMock
+
+# ":memory:" 공유 커넥션 — 각 테스트에서 독립적으로 생성
+def _make_mem_conn():
+    conn = sqlite3.connect(":memory:")
+    conn.row_factory = sqlite3.Row
+    return conn
+
+
+def test_purchase_history_has_new_columns():
+    """purchase_history 테이블에 신규 컬럼이 존재하는지 검증"""
+    import db
+    mem = _make_mem_conn()
+    with patch("db._conn", return_value=mem):
+        db.init_db()
+
+    cols = {r["name"] for r in mem.execute("PRAGMA table_info(purchase_history)").fetchall()}
+    assert "numbers" in cols
+    assert "is_real" in cols
+    assert "source_strategy" in cols
+    assert "source_detail" in cols
+    assert "checked" in cols
+    assert "results" in cols
+    assert "total_prize" in cols
+    # 기존 컬럼도 유지
+    assert "draw_no" in cols
+    assert "amount" in cols
+    assert "sets" in cols
+    assert "prize" in cols
+    assert "note" in cols
+    mem.close()
+
+
+def test_strategy_performance_table_exists():
+    """strategy_performance 테이블이 생성되는지 검증"""
+    import db
+    mem = _make_mem_conn()
+    with patch("db._conn", return_value=mem):
+        db.init_db()
+
+    cols = {r["name"] for r in mem.execute("PRAGMA table_info(strategy_performance)").fetchall()}
+    assert "strategy" in cols
+    assert "draw_no" in cols
+    assert "sets_count" in cols
+    assert "total_correct" in cols
+    assert "avg_score" in cols
+    mem.close()
+
+
+def test_strategy_weights_table_exists():
+    """strategy_weights 테이블이 생성되고 초기값이 있는지 검증"""
+    import db
+    mem = _make_mem_conn()
+    with patch("db._conn", return_value=mem):
+        db.init_db()
+
+    rows = mem.execute("SELECT * FROM strategy_weights ORDER BY strategy").fetchall()
+    strategies = {r["strategy"] for r in rows}
+    assert strategies == {"combined", "simulation", "heatmap", "manual", "custom"}
+    # 가중치 합이 1.0
+    total_weight = sum(r["weight"] for r in rows)
+    assert abs(total_weight - 1.0) < 0.01
+    mem.close()