1499 lines
53 KiB
Markdown
1499 lines
53 KiB
Markdown
# 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 + 신규 테이<ED858C><EC9DB4> 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 함수 내부에 신규 <20><>이블 추가)
|
||
- 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 ───────────────────────────────────────<E29480><E29480>────────────────────
|
||
|
||
@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 저장소이므로).
|