From c4406b9ecd8c7c319aa97640a822f24b733ea9a0 Mon Sep 17 00:00:00 2001 From: gahusb Date: Sun, 5 Apr 2026 22:10:16 +0900 Subject: [PATCH] =?UTF-8?q?lotto-lab:=20=EA=B5=AC=EB=A7=A4=20=EC=97=B0?= =?UTF-8?q?=EB=8F=99=20+=20=EC=A0=84=EB=9E=B5=20=EC=A7=84=ED=99=94=20?= =?UTF-8?q?=EC=8B=9C=EC=8A=A4=ED=85=9C=20=EA=B5=AC=ED=98=84=20=EA=B3=84?= =?UTF-8?q?=ED=9A=8D=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 --- ...04-05-lotto-purchase-strategy-evolution.md | 1498 +++++++++++++++++ 1 file changed, 1498 insertions(+) create mode 100644 docs/superpowers/plans/2026-04-05-lotto-purchase-strategy-evolution.md diff --git a/docs/superpowers/plans/2026-04-05-lotto-purchase-strategy-evolution.md b/docs/superpowers/plans/2026-04-05-lotto-purchase-strategy-evolution.md new file mode 100644 index 0000000..045105e --- /dev/null +++ b/docs/superpowers/plans/2026-04-05-lotto-purchase-strategy-evolution.md @@ -0,0 +1,1498 @@ +# Lotto 구매 연동 + 전략 진화 시스템 구현 계획 + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 로또 추천 번호의 실제/가상 구매, 자동 결과 체크, EMA 기반 전략 진화 메타 추천 시스템 구현 + +**Architecture:** 기존 lotto-backend(backend/) 서비스에 `purchase_manager.py`, `strategy_evolver.py` 2개 모듈 추가. 기존 `purchase_history` 테이블을 ALTER TABLE로 확장하고, `strategy_performance`, `strategy_weights` 2개 테이블 신규 생성. checker.py에서 purchase 체크 연동하여 자동 순환 파이프라인 완성. + +**Tech Stack:** Python 3.12, FastAPI, SQLite, APScheduler (기존 스택 그대로) + +**Spec:** `docs/superpowers/specs/2026-04-05-lotto-purchase-strategy-evolution-design.md` + +--- + +## 파일 구조 + +### 신규 파일 +| 파일 | 역할 | +|------|------| +| `backend/app/purchase_manager.py` | 구매 이력 관리 + 결과 체크 로직 | +| `backend/app/strategy_evolver.py` | EMA 계산 + Softmax 가중치 + 스마트 추천 | +| `backend/tests/test_purchase_manager.py` | purchase_manager 단위 테스트 | +| `backend/tests/test_strategy_evolver.py` | strategy_evolver 단위 테스트 | +| `backend/tests/test_integration.py` | 체커 연동 통합 테스트 | + +### 수정 파일 +| 파일 | 변경 내용 | +|------|----------| +| `backend/app/db.py` | purchase_history ALTER + 신규 테이�� 2개 + CRUD 함수 | +| `backend/app/checker.py` | check_results_for_draw() 끝에 purchase 체크 호출 | +| `backend/app/main.py` | 신규 API 엔드포인트 + Pydantic 모델 + import | + +--- + +## Task 1: DB 스키마 확장 — purchase_history ALTER + 신규 테이블 + +**Files:** +- Modify: `backend/app/db.py:254-268` (purchase_history 생성 부분) +- Modify: `backend/app/db.py:21` (init_db 함수 내부에 신규 ��이블 추가) +- Test: `backend/tests/test_purchase_manager.py` + +- [ ] **Step 1: 테스트 파일 생성 — DB 스키마 검증 테스트** + +```python +# 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 + +# 테스트용 임시 DB 경로 +TEST_DB = ":memory:" + +def _get_test_conn(): + conn = sqlite3.connect(TEST_DB) + conn.row_factory = sqlite3.Row + return conn + +def test_purchase_history_has_new_columns(): + """purchase_history 테이블에 신규 컬럼이 존재하는지 검증""" + with patch("db.DB_PATH", TEST_DB): + import db + db.DB_PATH = TEST_DB + db.init_db() + + conn = db._conn() + cols = {r["name"] for r in conn.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 + conn.close() + + +def test_strategy_performance_table_exists(): + """strategy_performance 테이블이 생성되는지 검증""" + with patch("db.DB_PATH", TEST_DB): + import db + db.DB_PATH = TEST_DB + db.init_db() + + conn = db._conn() + cols = {r["name"] for r in conn.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 + conn.close() + + +def test_strategy_weights_table_exists(): + """strategy_weights 테이블이 생성되고 초기값이 있는지 검증""" + with patch("db.DB_PATH", TEST_DB): + import db + db.DB_PATH = TEST_DB + db.init_db() + + conn = db._conn() + rows = conn.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 + conn.close() +``` + +- [ ] **Step 2: 테스트 실행 — 실패 확인** + +Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_purchase_manager.py -v` +Expected: FAIL — 신규 컬럼/테이블 미존재 + +- [ ] **Step 3: db.py init_db()에 purchase_history ALTER 추가** + +`backend/app/db.py`의 `init_db()` 함수에서, 기존 `purchase_history` CREATE TABLE 이후(268행 근처)에 추가: + +```python + # ── 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") +``` + +- [ ] **Step 4: db.py init_db()에 strategy_performance 테이블 추가** + +기존 `weekly_reports` 테이블 생성 부분(270행 근처) 이후에 추가: + +```python + # ── 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), + ) +``` + +- [ ] **Step 5: 테스트 실행 — 통과 확인** + +Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_purchase_manager.py -v` +Expected: 3 tests PASS + +- [ ] **Step 6: 커밋** + +```bash +cd C:/Users/jaeoh/Desktop/workspace/web-backend +git add backend/app/db.py backend/tests/test_purchase_manager.py +git commit -m "lotto-lab: DB 스키마 확장 — purchase_history ALTER + strategy 테이블 추가" +``` + +--- + +## Task 2: DB CRUD 함수 — 구매 이력 확장 + 전략 성과/가중치 + +**Files:** +- Modify: `backend/app/db.py:1140-1230` (기존 purchase CRUD 확장 + 신규 함수) +- Test: `backend/tests/test_purchase_manager.py` (추가) + +- [ ] **Step 1: 테스트 추가 — 확장된 purchase CRUD** + +`backend/tests/test_purchase_manager.py`에 추가: + +```python +def test_add_purchase_with_numbers(): + """번호 포함 구매 등록""" + with patch("db.DB_PATH", TEST_DB): + import db + db.DB_PATH = TEST_DB + db.init_db() + + result = db.add_purchase( + draw_no=1125, + amount=2000, + sets=2, + prize=0, + note="테스트", + numbers=[[3, 12, 23, 34, 38, 45], [7, 14, 21, 29, 36, 42]], + is_real=True, + source_strategy="combined", + source_detail={"recommendation_ids": [1, 2]}, + ) + assert result["draw_no"] == 1125 + assert result["is_real"] == 1 + assert result["source_strategy"] == "combined" + assert result["numbers"] == [[3, 12, 23, 34, 38, 45], [7, 14, 21, 29, 36, 42]] + assert result["checked"] == 0 + + +def test_get_purchases_filter_is_real(): + """is_real 필터 동작""" + with patch("db.DB_PATH", TEST_DB): + import db + db.DB_PATH = TEST_DB + db.init_db() + + db.add_purchase(draw_no=1125, amount=1000, sets=1, is_real=True, source_strategy="combined") + db.add_purchase(draw_no=1125, amount=1000, sets=1, is_real=False, source_strategy="simulation") + + real = db.get_purchases(is_real=True) + virtual = db.get_purchases(is_real=False) + assert all(r["is_real"] == 1 for r in real) + assert all(r["is_real"] == 0 for r in virtual) + + +def test_get_purchase_stats_by_type(): + """실제/가상 분리 통계""" + with patch("db.DB_PATH", TEST_DB): + import db + db.DB_PATH = TEST_DB + db.init_db() + + db.add_purchase(draw_no=1125, amount=2000, sets=2, prize=5000, is_real=True, source_strategy="combined") + db.add_purchase(draw_no=1125, amount=1000, sets=1, prize=0, is_real=False, source_strategy="simulation") + + stats = db.get_purchase_stats() + assert stats["total"]["invested"] == 3000 + assert stats["real"]["invested"] == 2000 + assert stats["virtual"]["invested"] == 1000 + + +def test_upsert_strategy_performance(): + """전략 성과 upsert""" + with patch("db.DB_PATH", TEST_DB): + import db + db.DB_PATH = TEST_DB + db.init_db() + + db.upsert_strategy_performance("combined", 1125, sets_count=2, total_correct=5, max_correct=3, avg_score=0.42) + rows = db.get_strategy_performance("combined") + assert len(rows) == 1 + assert rows[0]["sets_count"] == 2 + + # upsert: 같은 전략+회차 → 업데이트 + db.upsert_strategy_performance("combined", 1125, sets_count=3, total_correct=7, max_correct=4, avg_score=0.50) + rows = db.get_strategy_performance("combined") + assert len(rows) == 1 + assert rows[0]["sets_count"] == 3 + + +def test_update_strategy_weight(): + """전략 가중치 업데이트""" + with patch("db.DB_PATH", TEST_DB): + import db + db.DB_PATH = TEST_DB + db.init_db() + + db.update_strategy_weight("combined", weight=0.35, ema_score=0.28, total_sets=15, total_hits_3plus=3) + weights = db.get_strategy_weights() + combined = next(w for w in weights if w["strategy"] == "combined") + assert combined["weight"] == 0.35 + assert combined["ema_score"] == 0.28 +``` + +- [ ] **Step 2: 테스트 실행 — 실패 확인** + +Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_purchase_manager.py -v` +Expected: 새 테스트들 FAIL — 함수 미존재 + +- [ ] **Step 3: db.py — add_purchase 함수 시그니처 확장** + +기존 `add_purchase` (db.py:1154) 를 확장: + +```python +def add_purchase(draw_no: int, amount: int, sets: int, prize: int = 0, note: str = "", + numbers: list = None, is_real: bool = True, + source_strategy: str = "manual", source_detail: dict = None) -> Dict[str, Any]: + import json as _json + numbers_json = _json.dumps(numbers or [], ensure_ascii=False) + detail_json = _json.dumps(source_detail or {}, ensure_ascii=False) + is_real_int = 1 if is_real else 0 + with _conn() as conn: + conn.execute( + """INSERT INTO purchase_history + (draw_no, amount, sets, prize, note, numbers, is_real, source_strategy, source_detail) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (draw_no, amount, sets, prize, note, numbers_json, is_real_int, source_strategy, detail_json), + ) + row = conn.execute("SELECT * FROM purchase_history WHERE rowid = last_insert_rowid()").fetchone() + return _purchase_row_to_dict(row) +``` + +- [ ] **Step 4: db.py — _purchase_row_to_dict 확장** + +기존 `_purchase_row_to_dict` (db.py:1142) 를 확장: + +```python +def _purchase_row_to_dict(r) -> Dict[str, Any]: + import json as _json + numbers_raw = r["numbers"] if "numbers" in r.keys() else "[]" + detail_raw = r["source_detail"] if "source_detail" in r.keys() else "{}" + results_raw = r["results"] if "results" in r.keys() else "[]" + return { + "id": r["id"], + "draw_no": r["draw_no"], + "amount": r["amount"], + "sets": r["sets"], + "prize": r["prize"], + "note": r["note"], + "created_at": r["created_at"], + "numbers": _json.loads(numbers_raw) if numbers_raw else [], + "is_real": r["is_real"] if "is_real" in r.keys() else 1, + "source_strategy": r["source_strategy"] if "source_strategy" in r.keys() else "manual", + "source_detail": _json.loads(detail_raw) if detail_raw else {}, + "checked": r["checked"] if "checked" in r.keys() else 0, + "results": _json.loads(results_raw) if results_raw else [], + "total_prize": r["total_prize"] if "total_prize" in r.keys() else 0, + } +``` + +- [ ] **Step 5: db.py — get_purchases에 is_real, strategy 필터 추가** + +기존 `get_purchases` (db.py:1164) 확장: + +```python +def get_purchases(draw_no: int = None, days: int = None, + is_real: bool = None, strategy: str = None, + checked: bool = None) -> List[Dict[str, Any]]: + conditions, params = [], [] + if draw_no is not None: + conditions.append("draw_no = ?") + params.append(draw_no) + if days: + conditions.append("created_at >= datetime('now', ? || ' days')") + params.append(f"-{days}") + if is_real is not None: + conditions.append("is_real = ?") + params.append(1 if is_real else 0) + if strategy is not None: + conditions.append("source_strategy = ?") + params.append(strategy) + if checked is not None: + conditions.append("checked = ?") + params.append(1 if checked else 0) + where = f"WHERE {' AND '.join(conditions)}" if conditions else "" + with _conn() as conn: + rows = conn.execute( + f"SELECT * FROM purchase_history {where} ORDER BY draw_no DESC, id DESC", + params, + ).fetchall() + return [_purchase_row_to_dict(r) for r in rows] +``` + +- [ ] **Step 6: db.py — get_purchase_stats 확장 (전체/실제/가상 + 전략별)** + +기존 `get_purchase_stats` (db.py:1206) 를 교체: + +```python +def get_purchase_stats() -> Dict[str, Any]: + import json as _json + + def _calc_group(rows): + if not rows: + return {"sets": 0, "invested": 0, "prize": 0, "roi": 0.0, "win_rate": 0.0} + invested = sum(r["amount"] for r in rows) + prize = sum(r["total_prize"] or r["prize"] for r in rows) + wins = sum(1 for r in rows if (r["total_prize"] or r["prize"]) > 0) + return { + "sets": sum(r["sets"] for r in rows), + "invested": invested, + "prize": prize, + "roi": round((prize / invested * 100 - 100) if invested else 0.0, 2), + "win_rate": round(wins / len(rows) * 100, 2) if rows else 0.0, + } + + with _conn() as conn: + rows = conn.execute("SELECT * FROM purchase_history").fetchall() + + all_rows = [dict(r) for r in rows] + real_rows = [r for r in all_rows if r.get("is_real", 1) == 1] + virtual_rows = [r for r in all_rows if r.get("is_real", 1) == 0] + + # 전략별 집계 + by_strategy = {} + for r in all_rows: + strat = r.get("source_strategy", "manual") + if strat not in by_strategy: + by_strategy[strat] = [] + by_strategy[strat].append(r) + + strategy_stats = {} + for strat, srows in by_strategy.items(): + s = _calc_group(srows) + # results에서 correct 수 추출 + total_correct = 0 + count_sets = 0 + hits_3plus = 0 + for r in srows: + results_raw = r.get("results", "[]") + try: + results = _json.loads(results_raw) if isinstance(results_raw, str) else results_raw + except Exception: + results = [] + for res in results: + count_sets += 1 + c = res.get("correct", 0) + total_correct += c + if c >= 3: + hits_3plus += 1 + s["avg_correct"] = round(total_correct / count_sets, 2) if count_sets else 0.0 + s["hits_3plus"] = hits_3plus + strategy_stats[strat] = s + + return { + "total": _calc_group(all_rows), + "real": _calc_group(real_rows), + "virtual": _calc_group(virtual_rows), + "by_strategy": strategy_stats, + # 하위호환: 기존 필드도 유지 + "total_records": len(all_rows), + "total_invested": sum(r["amount"] for r in all_rows), + "total_prize": sum(r.get("total_prize", 0) or r["prize"] for r in all_rows), + "net": sum(r.get("total_prize", 0) or r["prize"] for r in all_rows) - sum(r["amount"] for r in all_rows), + "return_rate": round((sum(r.get("total_prize", 0) or r["prize"] for r in all_rows) / sum(r["amount"] for r in all_rows) * 100) if sum(r["amount"] for r in all_rows) else 0.0, 2), + "prize_count": sum(1 for r in all_rows if (r.get("total_prize", 0) or r["prize"]) > 0), + "max_prize": max((r.get("total_prize", 0) or r["prize"] for r in all_rows), default=0), + } +``` + +- [ ] **Step 7: db.py — strategy_performance CRUD 함수 추가** + +db.py 파일 끝에 추가: + +```python +# ── strategy_performance CRUD ──────────────────────────────────────────────── + +def upsert_strategy_performance(strategy: str, draw_no: int, sets_count: int = 0, + total_correct: int = 0, max_correct: int = 0, + prize_total: int = 0, avg_score: float = 0.0) -> None: + with _conn() as conn: + conn.execute( + """INSERT INTO strategy_performance (strategy, draw_no, sets_count, total_correct, max_correct, prize_total, avg_score) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(strategy, draw_no) DO UPDATE SET + sets_count=excluded.sets_count, total_correct=excluded.total_correct, + max_correct=excluded.max_correct, prize_total=excluded.prize_total, + avg_score=excluded.avg_score, + updated_at=strftime('%Y-%m-%dT%H:%M:%fZ','now')""", + (strategy, draw_no, sets_count, total_correct, max_correct, prize_total, avg_score), + ) + + +def get_strategy_performance(strategy: str = None, days: int = None) -> List[Dict[str, Any]]: + conditions, params = [], [] + if strategy: + conditions.append("strategy = ?") + params.append(strategy) + if days: + conditions.append("updated_at >= datetime('now', ? || ' days')") + params.append(f"-{days}") + where = f"WHERE {' AND '.join(conditions)}" if conditions else "" + with _conn() as conn: + rows = conn.execute( + f"SELECT * FROM strategy_performance {where} ORDER BY draw_no ASC", + params, + ).fetchall() + return [dict(r) for r in rows] + + +# ── strategy_weights CRUD ───────────────────────────────────────────────────── + +def get_strategy_weights() -> List[Dict[str, Any]]: + with _conn() as conn: + rows = conn.execute("SELECT * FROM strategy_weights ORDER BY weight DESC").fetchall() + return [dict(r) for r in rows] + + +def update_strategy_weight(strategy: str, weight: float, ema_score: float, + total_sets: int = None, total_hits_3plus: int = None) -> None: + with _conn() as conn: + fields = "weight=?, ema_score=?, updated_at=strftime('%Y-%m-%dT%H:%M:%fZ','now')" + params = [weight, ema_score] + if total_sets is not None: + fields += ", total_sets=?" + params.append(total_sets) + if total_hits_3plus is not None: + fields += ", total_hits_3plus=?" + params.append(total_hits_3plus) + params.append(strategy) + conn.execute(f"UPDATE strategy_weights SET {fields} WHERE strategy=?", params) + + +def update_purchase_results(purchase_id: int, results: list, total_prize: int) -> None: + """구매 건의 결과를 갱신 (체커 호출 후)""" + import json as _json + with _conn() as conn: + conn.execute( + "UPDATE purchase_history SET results=?, total_prize=?, checked=1 WHERE id=?", + (_json.dumps(results, ensure_ascii=False), total_prize, purchase_id), + ) +``` + +- [ ] **Step 8: db.py — exports 업데이트 (main.py import용)** + +db.py 파일 상단이나 기존 함수들이 이미 직접 import되므로, main.py의 import 문에 새 함수를 추가할 준비만 해둠. (Task 5에서 실행) + +- [ ] **Step 9: 테스트 실행 — 통과 확인** + +Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_purchase_manager.py -v` +Expected: ALL PASS + +- [ ] **Step 10: 커밋** + +```bash +cd C:/Users/jaeoh/Desktop/workspace/web-backend +git add backend/app/db.py backend/tests/test_purchase_manager.py +git commit -m "lotto-lab: 구매 CRUD 확장 + strategy_performance/weights CRUD 추가" +``` + +--- + +## Task 3: purchase_manager.py — 구매 결과 체크 로직 + +**Files:** +- Create: `backend/app/purchase_manager.py` +- Test: `backend/tests/test_purchase_manager.py` (추가) + +- [ ] **Step 1: 테스트 추가 — 구매 결과 체크** + +`backend/tests/test_purchase_manager.py`에 추가: + +```python +def test_check_purchases_for_draw(): + """특정 회차 구매 건들의 결과 체크""" + with patch("db.DB_PATH", TEST_DB): + import db + db.DB_PATH = TEST_DB + db.init_db() + + # draws 테이블에 테스트 데이터 삽입 + conn = db._conn() + conn.execute( + "INSERT OR IGNORE INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + (1125, "2026-04-04", 3, 12, 23, 34, 38, 45, 7) + ) + conn.commit() + conn.close() + + # 1125회차 대상 구매 등록 (draw_no=1125) + db.add_purchase( + draw_no=1125, amount=2000, sets=2, numbers=[[3, 12, 23, 34, 38, 45], [1, 2, 3, 4, 5, 6]], + is_real=True, source_strategy="combined", + ) + + from purchase_manager import check_purchases_for_draw + checked = check_purchases_for_draw(1125) + assert checked == 1 # 1건 체크됨 + + purchases = db.get_purchases(draw_no=1125) + p = purchases[0] + assert p["checked"] == 1 + assert len(p["results"]) == 2 + # 첫 번째 세트: 6개 전부 맞음 = 1등 + assert p["results"][0]["rank"] == 1 + assert p["results"][0]["correct"] == 6 + # 두 번째 세트: 3개 맞음 (3, 34 -> 아니 3만 맞음... 확인) + # [1,2,3,4,5,6] vs [3,12,23,34,38,45] → 3 하나만 맞음 + assert p["results"][1]["correct"] == 1 + assert p["results"][1]["rank"] == 0 + + +def test_check_purchases_updates_strategy_performance(): + """결과 체크 후 strategy_performance가 갱신되는지 검증""" + with patch("db.DB_PATH", TEST_DB): + import db + db.DB_PATH = TEST_DB + db.init_db() + + conn = db._conn() + conn.execute( + "INSERT OR IGNORE INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + (1126, "2026-04-11", 5, 15, 25, 35, 40, 44, 10) + ) + conn.commit() + conn.close() + + db.add_purchase( + draw_no=1126, amount=1000, sets=1, numbers=[[5, 15, 25, 10, 11, 12]], + is_real=False, source_strategy="simulation", + ) + + from purchase_manager import check_purchases_for_draw + check_purchases_for_draw(1126) + + perfs = db.get_strategy_performance("simulation") + assert len(perfs) >= 1 + p = perfs[-1] + assert p["draw_no"] == 1126 + assert p["sets_count"] == 1 +``` + +- [ ] **Step 2: 테스트 실행 — 실패 확인** + +Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_purchase_manager.py::test_check_purchases_for_draw -v` +Expected: FAIL — `purchase_manager` 모듈 미존재 + +- [ ] **Step 3: purchase_manager.py 작성** + +```python +# backend/app/purchase_manager.py +""" +구매 이력 관리 + 결과 체크 모듈. + +- check_purchases_for_draw(): 특정 회차 구매 건들의 결과를 자동 체크 +- 체커의 _calc_rank 재사용 +- 결과 체크 후 strategy_performance 자동 갱신 +""" +import json +import logging +from .db import ( + _conn, get_draw, get_purchases, update_purchase_results, + upsert_strategy_performance, +) +from .checker import _calc_rank + +logger = logging.getLogger("lotto-backend") + +RANK_PRIZE = {1: 0, 2: 0, 3: 1_500_000, 4: 50_000, 5: 5_000} + + +def check_purchases_for_draw(drw_no: int) -> int: + """ + 특정 회차 결과로 해당 회차 구매 건들을 채점한다. + + Returns: 채점한 구매 건 수 + """ + win_row = get_draw(drw_no) + if not win_row: + return 0 + + win_nums = [win_row["n1"], win_row["n2"], win_row["n3"], + win_row["n4"], win_row["n5"], win_row["n6"]] + bonus = win_row["bonus"] + + # draw_no가 해당 회차이고 아직 체크 안 된 구매 건 + unchecked = get_purchases(draw_no=drw_no, checked=False) + + # 전략별 집계 + strategy_agg = {} # strategy -> {sets_count, total_correct, max_correct, prize_total, scores[]} + + count = 0 + for purchase in unchecked: + numbers_list = purchase["numbers"] + if not numbers_list: + continue + + results = [] + for nums in numbers_list: + rank, correct, has_bonus = _calc_rank(nums, win_nums, bonus) + prize = RANK_PRIZE.get(rank, 0) + results.append({ + "numbers": nums, + "rank": rank, + "correct": correct, + "has_bonus": has_bonus, + "prize": prize, + }) + + total_prize = sum(r["prize"] for r in results) + update_purchase_results(purchase["id"], results, total_prize) + + # 전략별 집계 + strat = purchase["source_strategy"] + if strat not in strategy_agg: + strategy_agg[strat] = {"sets_count": 0, "total_correct": 0, "max_correct": 0, "prize_total": 0, "scores": []} + agg = strategy_agg[strat] + for r in results: + agg["sets_count"] += 1 + agg["total_correct"] += r["correct"] + agg["max_correct"] = max(agg["max_correct"], r["correct"]) + agg["prize_total"] += r["prize"] + agg["scores"].append(r["correct"] / 6.0) + + count += 1 + + # strategy_performance 갱신 + for strat, agg in strategy_agg.items(): + avg_score = sum(agg["scores"]) / len(agg["scores"]) if agg["scores"] else 0.0 + upsert_strategy_performance( + strategy=strat, + draw_no=drw_no, + sets_count=agg["sets_count"], + total_correct=agg["total_correct"], + max_correct=agg["max_correct"], + prize_total=agg["prize_total"], + avg_score=round(avg_score, 4), + ) + + logger.info(f"[purchase_manager] {drw_no}회차 구매 {count}건 체크 완료") + return count +``` + +- [ ] **Step 4: 테스트 실행 — 통과 확인** + +Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_purchase_manager.py -v` +Expected: ALL PASS + +- [ ] **Step 5: 커밋** + +```bash +cd C:/Users/jaeoh/Desktop/workspace/web-backend +git add backend/app/purchase_manager.py backend/tests/test_purchase_manager.py +git commit -m "lotto-lab: purchase_manager — 구매 결과 자동 체크 + 전략 성과 집계" +``` + +--- + +## Task 4: strategy_evolver.py — EMA + Softmax 가중치 진화 + 스마트 추천 + +**Files:** +- Create: `backend/app/strategy_evolver.py` +- Test: `backend/tests/test_strategy_evolver.py` + +- [ ] **Step 1: 테스트 파일 생성** + +```python +# backend/tests/test_strategy_evolver.py +import sys, os +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app")) + +import math +import pytest +from unittest.mock import patch + +TEST_DB = ":memory:" + + +def test_calc_draw_score_basic(): + """세트별 결과 → draw_score 계산""" + from strategy_evolver import calc_draw_score + + results = [ + {"correct": 3, "rank": 5}, # 3/6 + 0.1 = 0.6 + {"correct": 1, "rank": 0}, # 1/6 + 0 = 0.167 + ] + score = calc_draw_score(results) + expected = ((3/6 + 0.1) + (1/6)) / 2 + assert abs(score - expected) < 0.01 + + +def test_calc_draw_score_empty(): + """빈 결과 → 0""" + from strategy_evolver import calc_draw_score + assert calc_draw_score([]) == 0.0 + + +def test_recalculate_weights_softmax(): + """EMA → Softmax 가중치 변환""" + from strategy_evolver import _softmax_weights + + ema_scores = { + "combined": 0.30, + "simulation": 0.25, + "heatmap": 0.15, + "manual": 0.10, + "custom": 0.05, + } + weights = _softmax_weights(ema_scores) + + # 합이 1.0 + assert abs(sum(weights.values()) - 1.0) < 0.001 + # combined가 가장 높아야 함 + assert weights["combined"] > weights["simulation"] + assert weights["simulation"] > weights["heatmap"] + # 최소 가중치 5% 보장 + assert all(w >= 0.049 for w in weights.values()) + + +def test_recalculate_weights_min_weight(): + """한 전략의 EMA가 매우 낮아도 최소 5% 보장""" + from strategy_evolver import _softmax_weights + + ema_scores = { + "combined": 0.50, + "simulation": 0.01, + "heatmap": 0.01, + "manual": 0.01, + "custom": 0.01, + } + weights = _softmax_weights(ema_scores) + + assert weights["simulation"] >= 0.049 + assert weights["custom"] >= 0.049 + assert abs(sum(weights.values()) - 1.0) < 0.001 + + +def test_update_ema(): + """EMA 갱신 공식 검증""" + from strategy_evolver import ALPHA + + old_ema = 0.15 + draw_score = 0.40 + new_ema = ALPHA * draw_score + (1 - ALPHA) * old_ema + expected = 0.3 * 0.40 + 0.7 * 0.15 # = 0.12 + 0.105 = 0.225 + assert abs(new_ema - expected) < 0.001 +``` + +- [ ] **Step 2: 테스트 실행 — 실패 확인** + +Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_strategy_evolver.py -v` +Expected: FAIL — `strategy_evolver` 모듈 미존재 + +- [ ] **Step 3: strategy_evolver.py 작성** + +```python +# backend/app/strategy_evolver.py +""" +전략 진화 엔진 — EMA + Softmax 기반 적응형 가중치 관리. + +- calc_draw_score(): 구매 결과 → 성과 점수 +- update_ema_for_strategy(): 특정 전략의 EMA 갱신 +- recalculate_weights(): 전 전략 EMA → Softmax → 가중치 저장 +- generate_smart_recommendation(): 가중치 기반 메타 전략 추천 +""" +import math +import json +import logging +from typing import Dict, List, Any, Tuple + +from .db import ( + get_strategy_weights, update_strategy_weight, + get_strategy_performance, get_best_picks, get_all_draw_numbers, get_latest_draw, +) +from .recommender import recommend_numbers, recommend_with_heatmap +from .analyzer import generate_combined_recommendation, score_combination, build_analysis_cache +from .db import list_recommendations_ex + +logger = logging.getLogger("lotto-backend") + +ALPHA = 0.3 # EMA 감쇠율 +TEMPERATURE = 2.0 # Softmax 온도 +MIN_WEIGHT = 0.05 # 최소 가중치 +INITIAL_EMA = 0.15 # 콜드스타트 초기값 +MIN_DATA_DRAWS = 10 # 학습 최소 회차 + +STRATEGIES = ["combined", "simulation", "heatmap", "manual", "custom"] + +RANK_BONUS = {5: 0.1, 4: 0.3, 3: 0.6, 2: 0.8, 1: 1.0} + + +def calc_draw_score(results: List[Dict]) -> float: + """구매 결과 리스트 → 평균 성과 점수""" + if not results: + return 0.0 + scores = [] + for r in results: + s = r.get("correct", 0) / 6.0 + s += RANK_BONUS.get(r.get("rank", 0), 0) + scores.append(s) + return sum(scores) / len(scores) + + +def _softmax_weights(ema_scores: Dict[str, float]) -> Dict[str, float]: + """EMA 점수 → Softmax → 최소 가중치 보장 → 정규화""" + raw = {s: math.exp(ema / TEMPERATURE) for s, ema in ema_scores.items()} + total = sum(raw.values()) + weights = {s: v / total for s, v in raw.items()} + + # 최소 가중치 보장 + clamped = {} + surplus = 0.0 + unclamped = [] + for s, w in weights.items(): + if w < MIN_WEIGHT: + clamped[s] = MIN_WEIGHT + surplus += MIN_WEIGHT - w + else: + unclamped.append(s) + clamped[s] = w + + # surplus를 unclamped에서 비례 차감 + if surplus > 0 and unclamped: + unclamped_total = sum(clamped[s] for s in unclamped) + for s in unclamped: + clamped[s] -= surplus * (clamped[s] / unclamped_total) + + # 최종 정규화 + final_total = sum(clamped.values()) + return {s: round(v / final_total, 4) for s, v in clamped.items()} + + +def update_ema_for_strategy(strategy: str, draw_score: float) -> float: + """특정 전략의 EMA 갱신 + DB 저장. 새 EMA 반환.""" + weights = get_strategy_weights() + current = next((w for w in weights if w["strategy"] == strategy), None) + old_ema = current["ema_score"] if current else INITIAL_EMA + new_ema = ALPHA * draw_score + (1 - ALPHA) * old_ema + return new_ema + + +def recalculate_weights() -> Dict[str, float]: + """전 전략 EMA → Softmax → 가중치 재계산 + DB 저장""" + weights_rows = get_strategy_weights() + ema_scores = {w["strategy"]: w["ema_score"] for w in weights_rows} + + # 누락된 전략 보충 + for s in STRATEGIES: + if s not in ema_scores: + ema_scores[s] = INITIAL_EMA + + new_weights = _softmax_weights(ema_scores) + + for s, w in new_weights.items(): + row = next((r for r in weights_rows if r["strategy"] == s), None) + update_strategy_weight( + strategy=s, + weight=w, + ema_score=ema_scores[s], + total_sets=row["total_sets"] if row else 0, + total_hits_3plus=row["total_hits_3plus"] if row else 0, + ) + + logger.info(f"[strategy_evolver] 가중치 재계산: {new_weights}") + return new_weights + + +def evolve_after_check(strategy: str, draw_no: int, results: List[Dict]) -> None: + """결과 체크 후 EMA 갱신 + 가중치 재계산 (purchase_manager에서 호출)""" + draw_score = calc_draw_score(results) + new_ema = update_ema_for_strategy(strategy, draw_score) + + weights_rows = get_strategy_weights() + current = next((w for w in weights_rows if w["strategy"] == strategy), None) + hits_3plus = sum(1 for r in results if r.get("correct", 0) >= 3) + + update_strategy_weight( + strategy=strategy, + weight=current["weight"] if current else 0.2, + ema_score=new_ema, + total_sets=(current["total_sets"] if current else 0) + len(results), + total_hits_3plus=(current["total_hits_3plus"] if current else 0) + hits_3plus, + ) + + recalculate_weights() + + +def get_weights_with_trend() -> Dict[str, Any]: + """현재 가중치 + trend 정보 반환""" + weights = get_strategy_weights() + perfs = get_strategy_performance() + + # 전략별 최근 EMA 변화 추적 (최근 5회차) + strat_perfs = {} + for p in perfs: + s = p["strategy"] + if s not in strat_perfs: + strat_perfs[s] = [] + strat_perfs[s].append(p) + + result = [] + for w in weights: + # trend 계산: 최근 5회차 avg_score 변화 + sp = strat_perfs.get(w["strategy"], []) + if len(sp) >= 5: + recent_avg = sum(p["avg_score"] for p in sp[-3:]) / 3 + older_avg = sum(p["avg_score"] for p in sp[-5:-2]) / 3 + delta = recent_avg - older_avg + trend = "up" if delta > 0.02 else ("down" if delta < -0.02 else "stable") + else: + trend = "stable" + + result.append({ + "strategy": w["strategy"], + "weight": w["weight"], + "ema_score": w["ema_score"], + "total_sets": w["total_sets"], + "hits_3plus": w["total_hits_3plus"], + "trend": trend, + }) + + # 학습 상태 + all_draws = set() + for p in perfs: + all_draws.add(p["draw_no"]) + + return { + "weights": result, + "last_evolved": weights[0]["updated_at"] if weights else None, + "min_data_draws": MIN_DATA_DRAWS, + "current_data_draws": len(all_draws), + "status": "active" if len(all_draws) >= MIN_DATA_DRAWS else "learning", + } + + +def generate_smart_recommendation(sets: int = 5) -> Dict[str, Any]: + """ + 전략 가중치 기반 메타 전략 추천. + + 1. 가중치 로드 + 2. 각 전략에서 후보 10세트 생성 + 3. meta_score = original_score × strategy_weight + 4. 상위 N세트 선출 (중복 제거) + """ + weights_data = get_strategy_weights() + weight_map = {w["strategy"]: w["weight"] for w in weights_data} + draws = get_all_draw_numbers() + if not draws: + return {"error": "No draw data"} + + latest = get_latest_draw() + cache = build_analysis_cache(draws) + past_recs = list_recommendations_ex(limit=100, sort="id_desc") + + candidates = [] # [{numbers, score, strategy, meta_score}] + seen_keys = set() + + def _add_candidate(nums: list, strategy: str, raw_score: float = None): + key = tuple(sorted(nums)) + if key in seen_keys: + return + seen_keys.add(key) + if raw_score is None: + sc = score_combination(nums, cache) + raw_score = sc["score_total"] + meta = raw_score * weight_map.get(strategy, 0.1) + candidates.append({ + "numbers": sorted(nums), + "raw_score": round(raw_score, 4), + "strategy": strategy, + "meta_score": round(meta, 4), + }) + + # combined: 10세트 + for _ in range(10): + try: + r = generate_combined_recommendation(draws) + if "final_numbers" in r: + _add_candidate(r["final_numbers"], "combined") + except Exception: + pass + + # simulation: best_picks 상위 10개 + best = get_best_picks(limit=10) + for b in best: + nums = json.loads(b["numbers"]) if isinstance(b["numbers"], str) else b["numbers"] + _add_candidate(nums, "simulation", b.get("score_total")) + + # heatmap: 10세트 + for _ in range(10): + try: + r = recommend_with_heatmap(draws, past_recs) + _add_candidate(r["numbers"], "heatmap") + except Exception: + pass + + # manual: 10세트 + for _ in range(10): + try: + r = recommend_numbers(draws) + _add_candidate(r["numbers"], "manual") + except Exception: + pass + + # meta_score 기준 정렬, 상위 N개 + candidates.sort(key=lambda c: -c["meta_score"]) + top = candidates[:sets] + + # contribution 계산 + result_sets = [] + for c in top: + # 이 번호에 기여한 전략들의 비율 + sc = score_combination(c["numbers"], cache) + contributions = {} + for strat in STRATEGIES: + contributions[strat] = round(weight_map.get(strat, 0) * sc["score_total"], 4) + contrib_total = sum(contributions.values()) or 1 + contributions = {s: round(v / contrib_total, 3) for s, v in contributions.items()} + + result_sets.append({ + "numbers": c["numbers"], + "meta_score": c["meta_score"], + "source_strategy": c["strategy"], + "contribution": contributions, + "individual_scores": {k: round(v, 4) for k, v in sc.items()}, + }) + + # 학습 상태 + perfs = get_strategy_performance() + data_draws = len(set(p["draw_no"] for p in perfs)) + status = "active" if data_draws >= MIN_DATA_DRAWS else "learning" + + return { + "sets": result_sets, + "strategy_weights_used": weight_map, + "learning_status": { + "draws_learned": data_draws, + "status": status, + "message": "" if status == "active" else f"{MIN_DATA_DRAWS}회차 이상 데이터 필요 (현재 {data_draws}회차)", + }, + "based_on_latest_draw": latest["drw_no"] if latest else None, + } +``` + +- [ ] **Step 4: 테스트 실행 — 통과 확인** + +Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_strategy_evolver.py -v` +Expected: ALL PASS + +- [ ] **Step 5: 커밋** + +```bash +cd C:/Users/jaeoh/Desktop/workspace/web-backend +git add backend/app/strategy_evolver.py backend/tests/test_strategy_evolver.py +git commit -m "lotto-lab: strategy_evolver — EMA/Softmax 가중치 진화 + 스마트 추천" +``` + +--- + +## Task 5: checker.py 연동 — 자동 파이프라인 + +**Files:** +- Modify: `backend/app/checker.py:28-66` +- Test: `backend/tests/test_integration.py` + +- [ ] **Step 1: 통합 테스트 작성** + +```python +# backend/tests/test_integration.py +import sys, os +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app")) + +import pytest +from unittest.mock import patch + +TEST_DB = ":memory:" + + +def test_check_results_triggers_purchase_check(): + """check_results_for_draw가 purchase 체크도 트리거하는지 검증""" + with patch("db.DB_PATH", TEST_DB): + import db + db.DB_PATH = TEST_DB + db.init_db() + + # 당첨번호 삽입: 1124회차 (base), 1125회차 (결과) + conn = db._conn() + conn.execute( + "INSERT INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + (1124, "2026-03-28", 1, 2, 3, 4, 5, 6, 7) + ) + conn.execute( + "INSERT INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + (1125, "2026-04-04", 10, 20, 30, 35, 40, 44, 15) + ) + conn.commit() + conn.close() + + # 1125회차 대상 구매 (draw_no=1125) + db.add_purchase( + draw_no=1125, amount=1000, sets=1, + numbers=[[10, 20, 30, 1, 2, 3]], + is_real=True, source_strategy="combined", + ) + + from checker import check_results_for_draw + check_results_for_draw(1125) + + # purchase도 체크되었는지 확인 + purchases = db.get_purchases(draw_no=1125) + assert purchases[0]["checked"] == 1 + assert purchases[0]["results"][0]["correct"] == 3 # 10, 20, 30 맞음 +``` + +- [ ] **Step 2: 테스트 실행 — 실패 확인** + +Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_integration.py -v` +Expected: FAIL — checker.py에 purchase 연동 없음 + +- [ ] **Step 3: checker.py 수정 — purchase 체크 연동** + +`backend/app/checker.py` 파일 끝(66행 이후)의 `check_results_for_draw` 함수에 추가: + +기존 함수의 마지막 `return count` 바로 위에 추가: + +```python + # ── 구매 이력 체크 연동 ────────────────────────────────────── + try: + from .purchase_manager import check_purchases_for_draw + purchase_count = check_purchases_for_draw(drw_no) + if purchase_count > 0: + # 전략 가중치 재계산 + from .strategy_evolver import recalculate_weights + recalculate_weights() + except ImportError: + pass # purchase_manager 미설치 시 무시 (하위호환) +``` + +- [ ] **Step 4: 테스트 실행 — 통과 확인** + +Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_integration.py -v` +Expected: PASS + +- [ ] **Step 5: 커밋** + +```bash +cd C:/Users/jaeoh/Desktop/workspace/web-backend +git add backend/app/checker.py backend/tests/test_integration.py +git commit -m "lotto-lab: checker 연동 — 추첨 결과 시 purchase 자동 체크 + 가중치 재계산" +``` + +--- + +## Task 6: main.py — 신규 API 엔드포인트 + +**Files:** +- Modify: `backend/app/main.py:12-36` (import 확장) +- Modify: `backend/app/main.py:258-308` (purchase API 확장) +- Modify: `backend/app/main.py` (신규 전략/스마트 추천 API 추가) + +- [ ] **Step 1: main.py import 확장** + +`backend/app/main.py:12-36`의 import 블록에 추가: + +```python +from .db import ( + # ... 기존 import 유지 ... + # 신규: 전략 관련 + get_strategy_weights as db_get_strategy_weights, + get_strategy_performance as db_get_strategy_performance, + upsert_strategy_performance, + update_strategy_weight, + update_purchase_results, +) +from .purchase_manager import check_purchases_for_draw +from .strategy_evolver import ( + get_weights_with_trend, recalculate_weights, + generate_smart_recommendation, +) +``` + +- [ ] **Step 2: PurchaseCreate Pydantic 모델 확장** + +기존 `PurchaseCreate` (main.py:260-265) 교체: + +```python +class PurchaseCreate(BaseModel): + draw_no: int + amount: int + sets: int = 1 + prize: int = 0 + note: str = "" + # 신규 필드 + numbers: List[List[int]] = [] + is_real: bool = True + source_strategy: str = "manual" + source_detail: dict = {} +``` + +- [ ] **Step 3: PurchaseUpdate 확장** + +기존 `PurchaseUpdate` (main.py:268-273) 확장: + +```python +class PurchaseUpdate(BaseModel): + draw_no: Optional[int] = None + amount: Optional[int] = None + sets: Optional[int] = None + prize: Optional[int] = None + note: Optional[str] = None + # 신규 필드 + numbers: Optional[List[List[int]]] = None + is_real: Optional[bool] = None + source_strategy: Optional[str] = None +``` + +- [ ] **Step 4: 기존 purchase API 수정** + +`api_purchase_create` (main.py:288-291) 수정: + +```python +@app.post("/api/lotto/purchase", status_code=201) +def api_purchase_create(body: PurchaseCreate): + """구매 이력 추가 (실제/가상)""" + sets = body.sets if body.sets > 0 else max(len(body.numbers), 1) + amount = body.amount if body.amount > 0 else sets * 1000 + return add_purchase( + draw_no=body.draw_no, + amount=amount, + sets=sets, + prize=body.prize, + note=body.note, + numbers=body.numbers, + is_real=body.is_real, + source_strategy=body.source_strategy, + source_detail=body.source_detail, + ) +``` + +`api_purchase_list` (main.py:282-285) 수정: + +```python +@app.get("/api/lotto/purchase") +def api_purchase_list(draw_no: Optional[int] = None, days: Optional[int] = None, + is_real: Optional[bool] = None, strategy: Optional[str] = None): + """구매 이력 조회 (필터: draw_no, days, is_real, strategy)""" + return {"records": get_purchases(draw_no=draw_no, days=days, is_real=is_real, strategy=strategy)} +``` + +- [ ] **Step 5: 전략 API 추가** + +`main.py`의 purchase API 블록 이후에 추가: + +```python +# ── 전략 진화 API ───────────────────────────────────────��──────────────────── + +@app.get("/api/lotto/strategy/weights") +def api_strategy_weights(): + """현재 전략별 가중치 + 성과 요약 + trend""" + return get_weights_with_trend() + + +@app.get("/api/lotto/strategy/performance") +def api_strategy_performance(strategy: Optional[str] = None, days: Optional[int] = None): + """전략별 회차 성과 이력 (차트용)""" + rows = db_get_strategy_performance(strategy=strategy, days=days) + return {"records": rows} + + +@app.post("/api/lotto/strategy/evolve") +def api_strategy_evolve(): + """수동 가중치 재계산 트리거""" + new_weights = recalculate_weights() + return {"ok": True, "weights": new_weights} + + +# ── 스마트 추천 API ────────────────────────────────────────────────────────── + +@app.get("/api/lotto/recommend/smart") +def api_recommend_smart(sets: int = 5): + """전략 가중치 기반 메타 전략 추천""" + sets = max(1, min(sets, 10)) + result = generate_smart_recommendation(sets=sets) + if "error" in result: + raise HTTPException(status_code=500, detail=result["error"]) + return result +``` + +- [ ] **Step 6: 테스트 — API 스모크 테스트** + +Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -c "from backend.app.main import app; print('import OK')"` 또는 docker compose up 후 curl 테스트. + +Expected: import 성공, 문법 오류 없음 + +- [ ] **Step 7: 커밋** + +```bash +cd C:/Users/jaeoh/Desktop/workspace/web-backend +git add backend/app/main.py +git commit -m "lotto-lab: 구매/전략/스마트추천 API 엔드포인트 추가" +``` + +--- + +## Task 7: 통합 검증 — Docker 빌드 + 전체 테스트 + +**Files:** (수정 없음, 검증만) + +- [ ] **Step 1: 전체 테스트 실행** + +Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/ -v` +Expected: ALL PASS + +- [ ] **Step 2: Docker 빌드 테스트** + +Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && docker compose build lotto-backend` +Expected: 빌드 성공 + +- [ ] **Step 3: 로컬 실행 테스트** + +Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && docker compose up -d lotto-backend` + +API 엔드포인트 확인: +```bash +# 전략 가중치 조회 +curl http://localhost:18000/api/lotto/strategy/weights + +# 스마트 추천 +curl "http://localhost:18000/api/lotto/recommend/smart?sets=3" + +# 가상 구매 등록 +curl -X POST http://localhost:18000/api/lotto/purchase \ + -H "Content-Type: application/json" \ + -d '{"draw_no":1125,"numbers":[[3,12,23,34,38,45]],"is_real":false,"amount":1000,"sets":1,"source_strategy":"smart"}' + +# 구매 이력 조회 (가상만) +curl "http://localhost:18000/api/lotto/purchase?is_real=false" + +# 구매 통계 +curl http://localhost:18000/api/lotto/purchase/stats +``` + +Expected: 모든 API가 200 응답 + +- [ ] **Step 4: CLAUDE.md 업데이트** + +`backend/` CLAUDE.md의 lotto-lab API 목록에 신규 API 추가: + +```markdown +| GET | `/api/lotto/purchase` | 구매 이력 조회 (is_real, strategy, draw_no, days 필터) | +| POST | `/api/lotto/purchase` | 구매 등록 (실제/가상, 번호, 전략 출처 포함) | +| GET | `/api/lotto/purchase/stats` | 구매 통계 (전체/실제/가상 + 전략별) | +| GET | `/api/lotto/strategy/weights` | 전략별 가중치 + 성과 + trend | +| GET | `/api/lotto/strategy/performance` | 전략별 회차 성과 이력 (차트용) | +| POST | `/api/lotto/strategy/evolve` | 수동 가중치 재계산 | +| GET | `/api/lotto/recommend/smart` | 전략 진화 기반 메타 추천 | +``` + +lotto.db 테이블에 추가: + +```markdown +| `strategy_performance` | 전략별 회차 성과 (EMA 입력 데이터) | +| `strategy_weights` | 메타 전략 가중치 (EMA + Softmax) | +``` + +- [ ] **Step 5: 커밋** + +```bash +cd C:/Users/jaeoh/Desktop/workspace/web-backend +git add CLAUDE.md +git commit -m "lotto-lab: CLAUDE.md 신규 API + 테이블 문서 업데이트" +``` + +--- + +## 요약 + +| Task | 내용 | 파일 | +|------|------|------| +| 1 | DB 스키마 확장 | db.py (ALTER + 테이블 생성) | +| 2 | DB CRUD 함수 | db.py (purchase 확장 + strategy CRUD) | +| 3 | purchase_manager.py | 구매 결과 체크 + 전략 성과 집계 | +| 4 | strategy_evolver.py | EMA + Softmax + 스마트 추천 | +| 5 | checker.py 연동 | 자동 파이프라인 완성 | +| 6 | main.py API | 9개 엔드포인트 추가 | +| 7 | 통합 검증 | Docker 빌드 + API 테스트 + 문서 | + +**프론트엔드 작업은 별도 플랜으로 분리합니다** (web-ui 레포가 별도 Git 저장소이므로).