53 KiB
53 KiB
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 스키마 검증 테스트
# 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행 근처)에 추가:
# ── 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행 근처) 이후에 추가:
# ── 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: 커밋
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에 추가:
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) 를 확장:
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) 를 확장:
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) 확장:
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) 를 교체:
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 파일 끝에 추가:
# ── 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: 커밋
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에 추가:
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 작성
# 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: 커밋
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: 테스트 파일 생성
# 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 작성
# 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: 커밋
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: 통합 테스트 작성
# 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 바로 위에 추가:
# ── 구매 이력 체크 연동 ──────────────────────────────────────
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: 커밋
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 블록에 추가:
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) 교체:
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) 확장:
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) 수정:
@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) 수정:
@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 블록 이후에 추가:
# ── 전략 진화 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: 커밋
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 엔드포인트 확인:
# 전략 가중치 조회
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 추가:
| 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 테이블에 추가:
| `strategy_performance` | 전략별 회차 성과 (EMA 입력 데이터) |
| `strategy_weights` | 메타 전략 가중치 (EMA + Softmax) |
- Step 5: 커밋
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 저장소이므로).