diff --git a/agent-office/app/agents/lotto.py b/agent-office/app/agents/lotto.py index ae791e8..66d17ca 100644 --- a/agent-office/app/agents/lotto.py +++ b/agent-office/app/agents/lotto.py @@ -22,6 +22,8 @@ class LottoAgent(BaseAgent): return await self.run_signal_check(source=source) if action == "daily_digest": return await self.run_daily_digest() + if action == "sunday_review": + return await self.run_sunday_review() return {"ok": False, "message": f"unknown action: {action}"} async def on_approval(self, task_id: str, approved: bool, feedback: str = "") -> None: @@ -155,6 +157,29 @@ class LottoAgent(BaseAgent): add_log("lotto", f"daily_digest 예외: {e}", level="error", task_id=task_id) return {"ok": False, "message": f"{type(e).__name__}: {e}"} + async def run_sunday_review(self) -> dict: + """일 09:00 — 최신 회차 forward+calibration 보장 후 회고 텔레그램.""" + from ..service_proxy import lotto_latest_draw, lotto_backtest_review + from ..notifiers.telegram_lotto import send_sunday_review + from ..db import create_task, update_task_status, add_log + + task_id = create_task("lotto", "sunday_review", {}) + try: + draw_no = await lotto_latest_draw() + if not draw_no: + update_task_status(task_id, "failed", result_data={"reason": "no_draw"}) + return {"ok": False, "message": "no latest draw"} + # forward는 lotto cron이 이미 돌렸을 수 있으나 멱등이라 안전 — review만 호출 + payload = await lotto_backtest_review(draw_no) + await send_sunday_review(payload) + update_task_status(task_id, "succeeded", result_data={"draw_no": draw_no}) + add_log("lotto", f"sunday_review 발송: #{draw_no}", task_id=task_id) + return {"ok": True, "draw_no": draw_no} + except Exception as e: + update_task_status(task_id, "failed", result_data={"error": str(e)}) + add_log("lotto", f"sunday_review 예외: {e}", level="error", task_id=task_id) + return {"ok": False, "message": f"{type(e).__name__}: {e}"} + async def run_weekly_evolution_report(self) -> dict: """토 22:15 — lotto-lab evaluate-now 트리거 후 텔레그램 리포트. task_id wrap.""" from ..service_proxy import lotto_evolver_evaluate, lotto_evolver_status diff --git a/agent-office/app/notifiers/telegram_lotto.py b/agent-office/app/notifiers/telegram_lotto.py index 7835629..b1ff050 100644 --- a/agent-office/app/notifiers/telegram_lotto.py +++ b/agent-office/app/notifiers/telegram_lotto.py @@ -225,3 +225,42 @@ async def send_evolution_report(eval_result: Dict[str, Any], current_base: List[ await send_raw(text) except Exception as e: logger.warning(f"[telegram_lotto] evolution report send failed: {e}") + + +# ---------- 일요 회고 브리핑 ---------- + +def format_sunday_review(payload: Dict[str, Any]) -> str: + """일요 회고 브리핑 텍스트 (HTML parse_mode).""" + wa = payload.get("winner_analysis") or {} + draw_no = payload.get("draw_no") or "?" + pct = wa.get("percentile") + pct_txt = f"{pct*100:.0f}%" if pct is not None else "—" + lines = [f"🔍 로또 #{draw_no} 일요 회고", ""] + if wa: + lines.append(f"이번 당첨조합 분석치: {wa.get('score_total',0):.2f} " + f"(무작위 분포 상위 {pct_txt})") + lines.append(f" 빈도 {wa.get('score_frequency',0):.2f} · 지문 {wa.get('score_fingerprint',0):.2f} " + f"· 갭 {wa.get('score_gap',0):.2f} · 공동출현 {wa.get('score_cooccur',0):.2f} " + f"· 다양성 {wa.get('score_diversity',0):.2f}") + lines.append("") + if payload.get("forward"): + lines.append("📊 이번 회차 가상구매 성적") + for f in payload.get("forward", []): + p = f.get("prizes") or {} + name = {"engine_w": f"엔진({f.get('label','')})", "random_null": "무작위", "coverage": "커버리지"}.get( + f.get("strategy", ""), f.get("strategy", "?")) + lines.append(f" {name}: 최고 {f.get('best_match','?')}일치 / " + f"4등 {p.get('4th', 0)} · 5등 {p.get('5th', 0)}") + else: + lines.append("📊 이번 회차 가상구매 성적: 데이터 없음 (아직 집계 전)") + lines.append("") + lines.append("ℹ️ 무작위 대비 우위가 통계적으로 의미있을 때만 가중치가 진화합니다.") + return "\n".join(lines) + + +async def send_sunday_review(payload: Dict[str, Any]) -> None: + text = format_sunday_review(payload) + try: + await send_raw(text) + except Exception as e: + logger.warning(f"[telegram_lotto] sunday review send failed: {e}") diff --git a/agent-office/app/scheduler.py b/agent-office/app/scheduler.py index 02a0ddf..abbc00d 100644 --- a/agent-office/app/scheduler.py +++ b/agent-office/app/scheduler.py @@ -68,6 +68,11 @@ async def _run_lotto_sync_evolver_activity(): if agent: await agent.sync_evolver_activity() +async def _run_lotto_sunday_review(): + agent = AGENT_REGISTRY.get("lotto") + if agent: + await agent.run_sunday_review() + async def _run_youtube_research(): agent = AGENT_REGISTRY.get("youtube") if agent: @@ -116,6 +121,7 @@ def init_scheduler(): scheduler.add_job(_run_lotto_deep_check, "cron", day_of_week="sun,wed", hour=21, minute=15, id="lotto_deep_check") scheduler.add_job(_run_lotto_daily_digest, "cron", hour=9, minute=25, id="lotto_digest") scheduler.add_job(_run_lotto_weekly_evolution_report, "cron", day_of_week="sat", hour=22, minute=15, id="lotto_evolution_weekly") + scheduler.add_job(_run_lotto_sunday_review, "cron", day_of_week="sun", hour=9, minute=0, id="lotto_sunday_review") scheduler.add_job( _run_lotto_sync_evolver_activity, "cron", hour=9, minute=30, diff --git a/agent-office/app/service_proxy.py b/agent-office/app/service_proxy.py index 4db5dd1..30408f5 100644 --- a/agent-office/app/service_proxy.py +++ b/agent-office/app/service_proxy.py @@ -399,6 +399,14 @@ async def lotto_evolver_evaluate() -> Dict[str, Any]: return resp.json() +async def lotto_backtest_review(draw_no: int) -> Dict[str, Any]: + from .config import LOTTO_BACKEND_URL + resp = await _client.get(f"{LOTTO_BACKEND_URL}/api/lotto/backtest/review/{draw_no}") + resp.raise_for_status() + return resp.json() + + + from .config import AGENT_CONTAINER_MAP diff --git a/agent-office/tests/test_sunday_review.py b/agent-office/tests/test_sunday_review.py new file mode 100644 index 0000000..e971911 --- /dev/null +++ b/agent-office/tests/test_sunday_review.py @@ -0,0 +1,38 @@ +import sys, os +sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) + +from app.notifiers import telegram_lotto as tl + + +def test_format_sunday_review_text(): + payload = { + "draw_no": 1170, + "winner_analysis": {"score_total": 0.41, "percentile": 0.33, + "score_frequency": 0.4, "score_fingerprint": 0.5, "score_gap": 0.3, + "score_cooccur": 0.45, "score_diversity": 0.6}, + "forward": [ + {"strategy": "engine_w", "label": "w1", "prizes": {"1st":0,"2nd":0,"3rd":0,"4th":1,"5th":12}, "best_match": 4, "avg_meta_score": 0.55}, + {"strategy": "random_null", "label": "-", "prizes": {"1st":0,"2nd":0,"3rd":0,"4th":0,"5th":10}, "best_match": 3, "avg_meta_score": 0.33}, + ], + "track_record": {}, + "calibration_trend": [{"draw_no":1170,"score_total":0.41,"percentile":0.33}], + } + txt = tl.format_sunday_review(payload) + assert "1170" in txt + assert "%" in txt # percentile 표기 + assert "engine" in txt.lower() or "엔진" in txt + + +def test_format_sunday_review_no_calibration(): + payload = {"draw_no": 1171, "winner_analysis": None, "forward": []} + txt = tl.format_sunday_review(payload) + assert "1171" in txt + assert "%" not in txt # no percentile section when calibration absent + assert "데이터 없음" in txt + + +def test_format_sunday_review_missing_prizes_no_crash(): + payload = {"draw_no": 1171, "winner_analysis": None, + "forward": [{"strategy": "engine_w", "label": "w1", "best_match": 3}]} # no 'prizes' + txt = tl.format_sunday_review(payload) # must NOT raise + assert "1171" in txt diff --git a/docs/superpowers/plans/2026-05-31-lotto-self-learning-backtest.md b/docs/superpowers/plans/2026-05-31-lotto-self-learning-backtest.md new file mode 100644 index 0000000..0aeba6e --- /dev/null +++ b/docs/superpowers/plans/2026-05-31-lotto-self-learning-backtest.md @@ -0,0 +1,1328 @@ +# 로또 자가학습 백테스트 & 캘리브레이션 Implementation Plan + +> **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:** 로또 분석 엔진을 forward 가상구매(전략당 5,000장/회차) + 당첨조합 캘리브레이션 + null-model 대비 학습 + 일요 회고 브리핑으로 고도화한다. + +**Architecture:** lotto-lab에 순수 연산 모듈 `backtest.py`를 추가하고 집계 전용 2테이블(`backtest_runs`/`winner_calibration`)에 저장. `weight_evolver`의 학습 신호를 W-무관 N=5에서 W가 선택을 바꾸는 forward + null-model lift로 승격. agent-office가 일 09:00에 회고를 텔레그램 발송하고, web-ui /lotto 자율학습 탭에 성적표·캘리브레이션을 시각화. 신규 ML 없이 기존 `score_combination`/`run_simulation`/`weight_evolver` 100% 재활용. + +**Tech Stack:** Python 3.12, FastAPI, SQLite(WAL), APScheduler, pytest / React+Vite(web-ui, 별도 repo). + +**Spec:** `docs/superpowers/specs/2026-05-31-lotto-self-learning-backtest-design.md` + +--- + +## 기존 자산 (재사용 — 시그니처 확인됨) +- `app/analyzer.py`: `build_analysis_cache(draws)`, `score_combination(numbers, cache, weights=None)` → `{score_total, score_frequency, ...}`, `build_number_weights(cache)` +- `app/utils.py`: `weighted_sample_6(weights: dict[int,float]) -> list[int]` +- `app/weight_evolver.py`: `count_match(pick, winning)`, `calc_pick_score`, `RANK_BY_CORRECT`, `decide_base_update`, `get_week_start`, `get_weekly_trials`(via db), `evaluate_weekly` +- `app/db.py`: `_conn()`, `_ensure_column()`, `get_all_draw_numbers() -> [(drw_no,[n1..n6]),...] 오름차순`, `get_latest_draw() -> {drw_no,n1..n6,bonus,...}`, `get_draw(drw_no)`, `get_weekly_trials(week_start)`, `get_weight_trial(week_start,dow)`, `get_current_base()`, `save_base_history(...)`, `get_base_history(limit)` +- 테스트 컨벤션: `from app import `, pytest, seed 고정. 기존 `lotto/tests/test_weight_evolver.py` 참조. + +--- + +# Phase 1 — 데이터 모델 + 구매/채점 순수 로직 + +## Task 1.1: backtest_runs / winner_calibration 테이블 + +**Files:** +- Modify: `lotto/app/db.py` (init_db 내부, simulation 테이블 블록 뒤) +- Test: `lotto/tests/test_backtest_db.py` + +- [ ] **Step 1: 실패하는 테스트 작성** + +`lotto/tests/test_backtest_db.py`: +```python +import os, tempfile, importlib + +def _fresh_db(monkeypatch): + tmp = tempfile.mkdtemp() + path = os.path.join(tmp, "lotto.db") + from app import db + monkeypatch.setattr(db, "DB_PATH", path) + db.init_db() + return db + +def test_backtest_tables_exist(monkeypatch): + db = _fresh_db(monkeypatch) + with db._conn() as conn: + tables = {r["name"] for r in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'").fetchall()} + assert "backtest_runs" in tables + assert "winner_calibration" in tables + +def test_backtest_runs_unique(monkeypatch): + db = _fresh_db(monkeypatch) + db.save_backtest_run(draw_no=100, strategy="random_null", weight_label="-", + weight_json=None, trial_id=None, n_tickets=10, + hist={"m3":1,"m4":0,"m5":0,"m6":0,"bonus_hits":0}, + best_match=3, avg_meta_score=0.5) + db.save_backtest_run(draw_no=100, strategy="random_null", weight_label="-", + weight_json=None, trial_id=None, n_tickets=10, + hist={"m3":2,"m4":0,"m5":0,"m6":0,"bonus_hits":0}, + best_match=3, avg_meta_score=0.6) # 멱등 upsert + rows = db.get_backtest_runs(draw_no=100) + assert len(rows) == 1 + assert rows[0]["m3"] == 2 # 마지막 값으로 갱신 +``` + +- [ ] **Step 2: 실패 확인** — Run: `cd lotto && python -m pytest tests/test_backtest_db.py -v` Expected: FAIL (`save_backtest_run` 없음) + +- [ ] **Step 3: 테이블 DDL 추가** — `lotto/app/db.py` `init_db()` 안 simulation_candidates 인덱스 생성 직후에 삽입: +```python + conn.execute( + """ + CREATE TABLE IF NOT EXISTS backtest_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + draw_no INTEGER NOT NULL, + strategy TEXT NOT NULL, + weight_label TEXT NOT NULL DEFAULT '-', + weight_json TEXT, + trial_id INTEGER, + n_tickets INTEGER NOT NULL, + m3 INTEGER NOT NULL DEFAULT 0, + m4 INTEGER NOT NULL DEFAULT 0, + m5 INTEGER NOT NULL DEFAULT 0, + m6 INTEGER NOT NULL DEFAULT 0, + bonus_hits INTEGER NOT NULL DEFAULT 0, + best_match INTEGER NOT NULL DEFAULT 0, + avg_meta_score REAL, + created_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + """ + ) + conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS uq_backtest_run " + "ON backtest_runs(draw_no, strategy, weight_label);") + conn.execute( + """ + CREATE TABLE IF NOT EXISTS winner_calibration ( + draw_no INTEGER PRIMARY KEY, + winning_json TEXT NOT NULL, + score_total REAL NOT NULL, + score_frequency REAL NOT NULL, + score_fingerprint REAL NOT NULL, + score_gap REAL NOT NULL, + score_cooccur REAL NOT NULL, + score_diversity REAL NOT NULL, + percentile REAL, + my_pick_avg REAL, + cache_draws INTEGER NOT NULL, + created_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + """ + ) +``` + +- [ ] **Step 4: CRUD 함수 추가** — `lotto/app/db.py` 끝부분에: +```python +def save_backtest_run(draw_no, strategy, weight_label, weight_json, trial_id, + n_tickets, hist, best_match, avg_meta_score) -> None: + with _conn() as conn: + conn.execute( + """ + INSERT INTO backtest_runs + (draw_no, strategy, weight_label, weight_json, trial_id, n_tickets, + m3, m4, m5, m6, bonus_hits, best_match, avg_meta_score) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(draw_no, strategy, weight_label) DO UPDATE SET + weight_json=excluded.weight_json, trial_id=excluded.trial_id, + n_tickets=excluded.n_tickets, m3=excluded.m3, m4=excluded.m4, + m5=excluded.m5, m6=excluded.m6, bonus_hits=excluded.bonus_hits, + best_match=excluded.best_match, avg_meta_score=excluded.avg_meta_score, + created_at=datetime('now') + """, + (draw_no, strategy, weight_label, + json.dumps(weight_json) if weight_json is not None else None, + trial_id, n_tickets, + hist.get("m3",0), hist.get("m4",0), hist.get("m5",0), hist.get("m6",0), + hist.get("bonus_hits",0), best_match, avg_meta_score), + ) + +def get_backtest_runs(draw_no=None, strategy=None) -> List[Dict[str, Any]]: + q = "SELECT * FROM backtest_runs WHERE 1=1" + args = [] + if draw_no is not None: + q += " AND draw_no=?"; args.append(draw_no) + if strategy is not None: + q += " AND strategy=?"; args.append(strategy) + q += " ORDER BY draw_no DESC, strategy, weight_label" + with _conn() as conn: + return [dict(r) for r in conn.execute(q, args).fetchall()] + +def save_winner_calibration(draw_no, winning, scores, percentile, + my_pick_avg, cache_draws) -> None: + with _conn() as conn: + conn.execute( + """ + INSERT INTO winner_calibration + (draw_no, winning_json, score_total, score_frequency, score_fingerprint, + score_gap, score_cooccur, score_diversity, percentile, my_pick_avg, cache_draws) + VALUES (?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(draw_no) DO UPDATE SET + winning_json=excluded.winning_json, score_total=excluded.score_total, + score_frequency=excluded.score_frequency, score_fingerprint=excluded.score_fingerprint, + score_gap=excluded.score_gap, score_cooccur=excluded.score_cooccur, + score_diversity=excluded.score_diversity, percentile=excluded.percentile, + my_pick_avg=excluded.my_pick_avg, cache_draws=excluded.cache_draws, + created_at=datetime('now') + """, + (draw_no, json.dumps(winning), scores["score_total"], scores["score_frequency"], + scores["score_fingerprint"], scores["score_gap"], scores["score_cooccur"], + scores["score_diversity"], percentile, my_pick_avg, cache_draws), + ) + +def get_winner_calibration(draw_no: int) -> Optional[Dict[str, Any]]: + with _conn() as conn: + r = conn.execute("SELECT * FROM winner_calibration WHERE draw_no=?", + (draw_no,)).fetchone() + return dict(r) if r else None + +def get_calibration_history(limit: int = 52) -> List[Dict[str, Any]]: + with _conn() as conn: + rows = conn.execute( + "SELECT * FROM winner_calibration ORDER BY draw_no DESC LIMIT ?", + (limit,)).fetchall() + return [dict(r) for r in rows] + +def get_calibrated_draw_nos() -> set: + with _conn() as conn: + return {r["draw_no"] for r in + conn.execute("SELECT draw_no FROM winner_calibration").fetchall()} +``` + +- [ ] **Step 5: 통과 확인** — Run: `cd lotto && python -m pytest tests/test_backtest_db.py -v` Expected: PASS + +- [ ] **Step 6: Commit** +```bash +git add lotto/app/db.py lotto/tests/test_backtest_db.py +git commit -m "feat(lotto): backtest_runs/winner_calibration 테이블 + CRUD" +``` + +## Task 1.2: grade_tickets — 매칭 채점 + 등수 매핑 + +**Files:** +- Create: `lotto/app/backtest.py` +- Test: `lotto/tests/test_backtest.py` + +- [ ] **Step 1: 실패 테스트** + +`lotto/tests/test_backtest.py`: +```python +from app import backtest as bt + +def test_grade_tickets_histogram_and_prizes(): + winning6 = [1, 2, 3, 4, 5, 6] + bonus = 7 + tickets = [ + [1, 2, 3, 4, 5, 6], # 6일치 = 1등 + [1, 2, 3, 4, 5, 7], # 5일치 + 보너스 = 2등 + [1, 2, 3, 4, 5, 8], # 5일치 = 3등 + [1, 2, 3, 4, 9, 10], # 4일치 = 4등 + [1, 2, 3, 11, 12, 13], # 3일치 = 5등 + [40, 41, 42, 43, 44, 45], # 0일치 + ] + r = bt.grade_tickets(tickets, winning6, bonus) + assert r["m6"] == 1 + assert r["m5"] == 2 # 5일치 총 2장(보너스 포함) + assert r["bonus_hits"] == 1 # 그 중 보너스 1장 + assert r["m4"] == 1 + assert r["m3"] == 1 + assert r["best_match"] == 6 + # 등수 매핑 헬퍼 + prizes = bt.prize_counts(r) + assert prizes == {"1st": 1, "2nd": 1, "3rd": 1, "4th": 1, "5th": 1} +``` + +- [ ] **Step 2: 실패 확인** — Run: `cd lotto && python -m pytest tests/test_backtest.py::test_grade_tickets_histogram_and_prizes -v` Expected: FAIL (모듈 없음) + +- [ ] **Step 3: 구현** — `lotto/app/backtest.py`: +```python +"""로또 자가학습 백테스트 — 순수 연산 (FastAPI 의존성 0, Windows 이전 대비).""" +import random +from typing import Any, Dict, List, Optional, Tuple + +from .analyzer import build_analysis_cache, build_number_weights, score_combination +from .utils import weighted_sample_6 +from .weight_evolver import count_match + + +def grade_tickets(tickets: List[List[int]], winning6: List[int], bonus: int) -> Dict[str, Any]: + """티켓 묶음을 당첨번호로 채점 → 매칭 히스토그램 + 보너스 + best_match. + 2등 판정: 5일치 AND 보너스 번호를 티켓이 포함.""" + win = set(winning6) + hist = {"m3": 0, "m4": 0, "m5": 0, "m6": 0, "bonus_hits": 0} + best = 0 + for t in tickets: + c = len(set(t) & win) + if c > best: + best = c + if c == 6: + hist["m6"] += 1 + elif c == 5: + hist["m5"] += 1 + if bonus in t: + hist["bonus_hits"] += 1 + elif c == 4: + hist["m4"] += 1 + elif c == 3: + hist["m3"] += 1 + return {**hist, "best_match": best} + + +def prize_counts(hist: Dict[str, Any]) -> Dict[str, int]: + """매칭 히스토그램 → 등수 카운트. + 1등=m6, 2등=bonus_hits, 3등=m5−bonus_hits, 4등=m4, 5등=m3.""" + return { + "1st": hist.get("m6", 0), + "2nd": hist.get("bonus_hits", 0), + "3rd": hist.get("m5", 0) - hist.get("bonus_hits", 0), + "4th": hist.get("m4", 0), + "5th": hist.get("m3", 0), + } +``` + +- [ ] **Step 4: 통과 확인** — Run: `cd lotto && python -m pytest tests/test_backtest.py -v` Expected: PASS + +- [ ] **Step 5: Commit** +```bash +git add lotto/app/backtest.py lotto/tests/test_backtest.py +git commit -m "feat(lotto): grade_tickets 매칭 채점 + 등수 매핑" +``` + +## Task 1.3: 티켓 생성 전략 — engine_w / random_null / coverage + +**Files:** +- Modify: `lotto/app/backtest.py` +- Test: `lotto/tests/test_backtest.py` + +- [ ] **Step 1: 실패 테스트** — `test_backtest.py`에 추가: +```python +def _toy_draws(n=120): + # 결정적 가짜 회차: 분석 캐시 구성용 (오름차순 (drw_no, [6 nums])) + import random as _r + _r.seed(1) + out = [] + for i in range(1, n + 1): + nums = sorted(_r.sample(range(1, 46), 6)) + out.append((i, nums)) + return out + +def test_purchase_tickets_distinct_and_count(): + draws = _toy_draws() + cache = bt.build_analysis_cache(draws) + nw = bt.build_number_weights(cache) + pool = bt.generate_pool(cache, nw, n=2000, seed=7) + W = [0.25, 0.30, 0.20, 0.15, 0.10] + bought = bt.purchase_tickets(pool, cache, W, k=50) + assert len(bought) == 50 + assert len({tuple(t) for t in bought}) == 50 # distinct + # W로 랭킹된 상위 → 평균 분석치가 풀 평균보다 높아야 + avg_bought = sum(score_combination(t, cache, W)["score_total"] for t in bought) / 50 + assert avg_bought > 0 + +def test_random_null_and_coverage_distinct(): + rnd = bt.random_null_tickets(k=50, seed=3) + assert len(rnd) == 50 and len({tuple(t) for t in rnd}) == 50 + cov = bt.coverage_tickets(k=9, seed=3) # 9장 = 54슬롯 ≥ 45번호 전수 커버 가능 + flat = {n for t in cov for n in t} + assert len(cov) == 9 and len({tuple(t) for t in cov}) == 9 + assert len(flat) >= 40 # 커버리지 전략은 번호를 넓게 퍼뜨림 +``` + +- [ ] **Step 2: 실패 확인** — Run: `cd lotto && python -m pytest tests/test_backtest.py -v` Expected: FAIL + +- [ ] **Step 3: 구현** — `lotto/app/backtest.py`에 추가: +```python +def generate_pool(cache, number_weights, n: int = 20000, + seed: Optional[int] = None) -> List[List[int]]: + """가중 샘플링으로 distinct 후보 풀 생성.""" + if seed is not None: + random.seed(seed) + seen, pool = set(), [] + attempts, cap = 0, n * 4 + while len(pool) < n and attempts < cap: + attempts += 1 + nums = tuple(sorted(weighted_sample_6(number_weights))) + if nums in seen: + continue + seen.add(nums) + pool.append(list(nums)) + return pool + + +def purchase_tickets(pool, cache, W: List[float], k: int) -> List[List[int]]: + """풀을 score_combination(·, W)로 랭킹 → 상위 k장 distinct.""" + ranked = sorted(pool, key=lambda t: -score_combination(t, cache, W)["score_total"]) + return ranked[:k] + + +def random_null_tickets(k: int, seed: Optional[int] = None) -> List[List[int]]: + """무작위 distinct 티켓 k장 (null-model 대조군).""" + if seed is not None: + random.seed(seed) + seen, out = set(), [] + while len(out) < k: + nums = tuple(sorted(random.sample(range(1, 46), 6))) + if nums in seen: + continue + seen.add(nums) + out.append(list(nums)) + return out + + +def coverage_tickets(k: int, seed: Optional[int] = None) -> List[List[int]]: + """greedy 커버리지 — 아직 덜 쓰인 번호를 우선 배치해 번호를 넓게 분산. + (휠링/보장설계는 향후. 현재는 distinct + 번호 사용 균등화)""" + if seed is not None: + random.seed(seed) + usage = {n: 0 for n in range(1, 46)} + seen, out = set(), [] + guard = 0 + while len(out) < k and guard < k * 50: + guard += 1 + ranked = sorted(range(1, 46), key=lambda n: (usage[n], random.random())) + nums = tuple(sorted(ranked[:6])) + if nums in seen: + # 동점 흔들기: 약간 더 깊은 풀에서 샘플 + nums = tuple(sorted(random.sample(ranked[:12], 6))) + if nums in seen: + continue + seen.add(nums) + out.append(list(nums)) + for n in nums: + usage[n] += 1 + return out +``` + +- [ ] **Step 4: 통과 확인** — Run: `cd lotto && python -m pytest tests/test_backtest.py -v` Expected: PASS + +- [ ] **Step 5: Commit** +```bash +git add lotto/app/backtest.py lotto/tests/test_backtest.py +git commit -m "feat(lotto): 티켓 생성 3전략 (engine_w/random_null/coverage)" +``` + +--- + +# Phase 2 — 캘리브레이션 + forward 실행 + 백필 + +## Task 2.1: point-in-time 캐시 헬퍼 (대상 회차 제외 검증) + +**Files:** +- Modify: `lotto/app/backtest.py` +- Test: `lotto/tests/test_backtest.py` + +- [ ] **Step 1: 실패 테스트** +```python +def test_point_in_time_excludes_target_draw(): + draws = _toy_draws(50) # drw_no 1..50 + pit = bt.point_in_time_draws(draws, target_draw_no=30) + assert all(d < 30 for d, _ in pit) # 30 이상 제외 + assert max(d for d, _ in pit) == 29 + assert len(pit) == 29 +``` + +- [ ] **Step 2: 실패 확인** — Run: `cd lotto && python -m pytest tests/test_backtest.py::test_point_in_time_excludes_target_draw -v` Expected: FAIL + +- [ ] **Step 3: 구현** — `lotto/app/backtest.py`에 추가: +```python +def point_in_time_draws(draws: List[Tuple[int, List[int]]], + target_draw_no: int) -> List[Tuple[int, List[int]]]: + """target 회차 추첨 '직전' 시점의 데이터 — target_draw_no 미만만.""" + return [(d, nums) for d, nums in draws if d < target_draw_no] +``` + +- [ ] **Step 4: 통과 확인** — Run: `cd lotto && python -m pytest tests/test_backtest.py::test_point_in_time_excludes_target_draw -v` Expected: PASS + +- [ ] **Step 5: Commit** +```bash +git add lotto/app/backtest.py lotto/tests/test_backtest.py +git commit -m "feat(lotto): point_in_time_draws 헬퍼" +``` + +## Task 2.2: calibrate_winner — 당첨조합 역분석 + percentile + +**Files:** +- Modify: `lotto/app/backtest.py` +- Test: `lotto/tests/test_backtest.py` + +- [ ] **Step 1: 실패 테스트** +```python +def test_calibrate_winner_scores_and_percentile(): + draws = _toy_draws(60) + winning6 = [3, 11, 19, 27, 35, 44] + res = bt.calibrate_winner_compute(draws, target_draw_no=60, + winning6=winning6, sample_m=500, seed=9) + assert set(res["scores"].keys()) >= {"score_total", "score_frequency", + "score_fingerprint", "score_gap", "score_cooccur", "score_diversity"} + assert 0.0 <= res["percentile"] <= 1.0 + assert res["cache_draws"] == 59 # 1..59 +``` + +- [ ] **Step 2: 실패 확인** — Run: `cd lotto && python -m pytest tests/test_backtest.py::test_calibrate_winner_scores_and_percentile -v` Expected: FAIL + +- [ ] **Step 3: 구현** — `lotto/app/backtest.py`에 추가: +```python +def calibrate_winner_compute(draws, target_draw_no, winning6, + sample_m: int = 2000, seed: Optional[int] = None) -> Dict[str, Any]: + """순수 연산: point-in-time 캐시로 당첨조합 채점 + 무작위 M표본 percentile.""" + pit = point_in_time_draws(draws, target_draw_no) + cache = build_analysis_cache(pit) + scores = score_combination(sorted(winning6), cache) + win_total = scores["score_total"] + samples = random_null_tickets(sample_m, seed=seed) + le = sum(1 for t in samples + if score_combination(t, cache)["score_total"] <= win_total) + percentile = le / max(len(samples), 1) + return {"scores": scores, "percentile": percentile, "cache_draws": len(pit)} +``` + +- [ ] **Step 4: 통과 확인** — Run: `cd lotto && python -m pytest tests/test_backtest.py::test_calibrate_winner_scores_and_percentile -v` Expected: PASS + +- [ ] **Step 5: Commit** +```bash +git add lotto/app/backtest.py lotto/tests/test_backtest.py +git commit -m "feat(lotto): calibrate_winner_compute 당첨조합 역분석+percentile" +``` + +## Task 2.3: DB 진입점 — calibrate_winner + backfill (멱등) + +**Files:** +- Modify: `lotto/app/backtest.py` +- Test: `lotto/tests/test_backtest_db.py` + +- [ ] **Step 1: 실패 테스트** — `test_backtest_db.py`에 추가: +```python +def _seed_draws(db, n=40): + rows = [] + import random as _r; _r.seed(2) + for i in range(1, n + 1): + s = sorted(_r.sample(range(1, 46), 6)) + rows.append({"drw_no": i, "drw_date": f"2020-01-{(i%28)+1:02d}", + "n1": s[0], "n2": s[1], "n3": s[2], "n4": s[3], + "n5": s[4], "n6": s[5], "bonus": ((s[5] % 45) + 1)}) + db.upsert_many_draws(rows) + +def test_backfill_calibration_idempotent(monkeypatch): + db = _fresh_db(monkeypatch) + _seed_draws(db, 40) + from app import backtest as bt + r1 = bt.backfill_calibration(batch=15, sample_m=200) + # 첫 회차는 point-in-time 데이터가 빈약 → min_history 이후만 처리 + done1 = len(db.get_calibrated_draw_nos()) + assert done1 > 0 + r2 = bt.backfill_calibration(batch=100, sample_m=200) # 나머지 + done2 = len(db.get_calibrated_draw_nos()) + assert done2 >= done1 + r3 = bt.backfill_calibration(batch=100, sample_m=200) # 재실행 → 추가 0 + assert r3["calibrated"] == 0 +``` + +- [ ] **Step 2: 실패 확인** — Run: `cd lotto && python -m pytest tests/test_backtest_db.py -v` Expected: FAIL + +- [ ] **Step 3: 구현** — `lotto/app/backtest.py`에 추가 (db는 함수 내 지연 import — weight_evolver 패턴 동일): +```python +MIN_HISTORY = 30 # point-in-time 캐시 최소 회차 (이 미만은 캘리브레이션 skip) + + +def _db(): + from . import db as _db_mod + return _db_mod + + +def calibrate_winner(draw_no: int, sample_m: int = 2000) -> Dict[str, Any]: + """DB 진입점: 회차 1개 캘리브레이션 후 저장 (멱등).""" + db = _db() + draws = db.get_all_draw_numbers() + row = db.get_draw(draw_no) + if row is None: + return {"ok": False, "reason": "no_draw"} + pit = point_in_time_draws(draws, draw_no) + if len(pit) < MIN_HISTORY: + return {"ok": False, "reason": "insufficient_history"} + winning6 = [row["n1"], row["n2"], row["n3"], row["n4"], row["n5"], row["n6"]] + res = calibrate_winner_compute(draws, draw_no, winning6, sample_m=sample_m) + db.save_winner_calibration( + draw_no=draw_no, winning=winning6, scores=res["scores"], + percentile=res["percentile"], my_pick_avg=None, + cache_draws=res["cache_draws"], + ) + return {"ok": True, "draw_no": draw_no, **res} + + +def backfill_calibration(batch: int = 50, sample_m: int = 2000) -> Dict[str, Any]: + """미처리 회차만 batch개 캘리브레이션 (멱등·재개 가능).""" + db = _db() + draws = db.get_all_draw_numbers() + done = db.get_calibrated_draw_nos() + todo = [d for d, _ in draws if d not in done and d > MIN_HISTORY] + todo.sort() + n = 0 + for draw_no in todo[:batch]: + r = calibrate_winner(draw_no, sample_m=sample_m) + if r.get("ok"): + n += 1 + return {"calibrated": n, "remaining": max(0, len(todo) - batch)} +``` + +- [ ] **Step 4: 통과 확인** — Run: `cd lotto && python -m pytest tests/test_backtest_db.py -v` Expected: PASS + +- [ ] **Step 5: Commit** +```bash +git add lotto/app/backtest.py lotto/tests/test_backtest_db.py +git commit -m "feat(lotto): calibrate_winner + backfill (멱등·청크)" +``` + +## Task 2.4: run_forward_purchase — 3전략 구매·채점·저장 + +**Files:** +- Modify: `lotto/app/backtest.py` +- Test: `lotto/tests/test_backtest_db.py` + +- [ ] **Step 1: 실패 테스트** — `test_backtest_db.py`에 추가: +```python +def test_run_forward_purchase_persists_all_strategies(monkeypatch): + db = _fresh_db(monkeypatch) + _seed_draws(db, 40) + from app import backtest as bt + # 작은 규모로 빠르게 + res = bt.run_forward_purchase(draw_no=40, k=20, pool_n=500, sample_seed=5) + assert res["ok"] is True + rows = db.get_backtest_runs(draw_no=40) + strategies = {r["strategy"] for r in rows} + assert "random_null" in strategies + assert "coverage" in strategies + assert "engine_w" in strategies # base 가중치로 최소 1건 + for r in rows: + assert r["n_tickets"] == 20 +``` + +- [ ] **Step 2: 실패 확인** — Run: `cd lotto && python -m pytest tests/test_backtest_db.py::test_run_forward_purchase_persists_all_strategies -v` Expected: FAIL + +- [ ] **Step 3: 구현** — `lotto/app/backtest.py`에 추가: +```python +def run_forward_purchase(draw_no: int, k: int = 5000, pool_n: int = 20000, + sample_seed: Optional[int] = None) -> Dict[str, Any]: + """회차 추첨 '직전' 시점 데이터로 3전략 구매 → 당첨번호로 채점 → 저장(멱등). + engine_w: 그 주 weight_trials 6개(없으면 current_base 1개)로 각각 구매.""" + import json as _json + db = _db() + draws = db.get_all_draw_numbers() + row = db.get_draw(draw_no) + if row is None: + return {"ok": False, "reason": "no_draw"} + pit = point_in_time_draws(draws, draw_no) + if len(pit) < MIN_HISTORY: + return {"ok": False, "reason": "insufficient_history"} + winning6 = [row["n1"], row["n2"], row["n3"], row["n4"], row["n5"], row["n6"]] + bonus = row["bonus"] + + cache = build_analysis_cache(pit) + nw = build_number_weights(cache) + pool = generate_pool(cache, nw, n=pool_n, seed=sample_seed) + + def _store(strategy, label, weight_json, trial_id, tickets): + graded = grade_tickets(tickets, winning6, bonus) + avg_meta = (sum(score_combination(t, cache)["score_total"] for t in tickets) + / max(len(tickets), 1)) + db.save_backtest_run( + draw_no=draw_no, strategy=strategy, weight_label=label, + weight_json=weight_json, trial_id=trial_id, n_tickets=len(tickets), + hist=graded, best_match=graded["best_match"], avg_meta_score=avg_meta, + ) + + # 1) engine_w — 그 주 trials(있으면) 아니면 current_base + from . import weight_evolver as we + week_start = we.get_week_start() + trials = db.get_weekly_trials(week_start) if hasattr(db, "get_weekly_trials") else [] + if trials: + for t in trials: + bought = purchase_tickets(pool, cache, t["weight"], k) + _store("engine_w", f"w{t['day_of_week']}", t["weight"], t["id"], bought) + else: + base = db.get_current_base() or [0.2] * 5 + bought = purchase_tickets(pool, cache, base, k) + _store("engine_w", "base", base, None, bought) + + # 2) random_null + _store("random_null", "-", None, None, random_null_tickets(k, seed=sample_seed)) + # 3) coverage + _store("coverage", "-", None, None, coverage_tickets(k, seed=sample_seed)) + + return {"ok": True, "draw_no": draw_no} +``` + +- [ ] **Step 4: 통과 확인** — Run: `cd lotto && python -m pytest tests/test_backtest_db.py -v` Expected: PASS + +- [ ] **Step 5: Commit** +```bash +git add lotto/app/backtest.py lotto/tests/test_backtest_db.py +git commit -m "feat(lotto): run_forward_purchase 3전략 구매·채점·저장" +``` + +--- + +# Phase 3 — API 라우터 + main 통합 + +## Task 3.1: track-record / calibration 집계 + build_review_payload + +**Files:** +- Modify: `lotto/app/backtest.py` +- Test: `lotto/tests/test_backtest_db.py` + +- [ ] **Step 1: 실패 테스트** — `test_backtest_db.py`에 추가: +```python +def test_track_record_and_review_payload(monkeypatch): + db = _fresh_db(monkeypatch) + _seed_draws(db, 40) + from app import backtest as bt + bt.run_forward_purchase(draw_no=40, k=20, pool_n=500, sample_seed=5) + bt.calibrate_winner(40, sample_m=200) + + tr = bt.track_record() + assert "random_null" in tr["by_strategy"] + assert tr["by_strategy"]["random_null"]["n_tickets"] >= 20 + + payload = bt.build_review_payload(40) + assert payload["draw_no"] == 40 + assert "winner_analysis" in payload # 당첨조합 5분석치 + assert "forward" in payload # 이번 회차 전략별 성적 + assert "calibration_trend" in payload +``` + +- [ ] **Step 2: 실패 확인** — Run: `cd lotto && python -m pytest tests/test_backtest_db.py::test_track_record_and_review_payload -v` Expected: FAIL + +- [ ] **Step 3: 구현** — `lotto/app/backtest.py`에 추가: +```python +def track_record() -> Dict[str, Any]: + """전략별 누적 등수 집계 (engine_w는 라벨 합산).""" + db = _db() + rows = db.get_backtest_runs() + agg: Dict[str, Dict[str, int]] = {} + for r in rows: + a = agg.setdefault(r["strategy"], { + "n_tickets": 0, "1st": 0, "2nd": 0, "3rd": 0, "4th": 0, "5th": 0, "draws": 0}) + p = prize_counts(r) + a["n_tickets"] += r["n_tickets"] + for tier in ("1st", "2nd", "3rd", "4th", "5th"): + a[tier] += p[tier] + a["draws"] += 1 + return {"by_strategy": agg} + + +def build_review_payload(draw_no: int) -> Dict[str, Any]: + """일요 회고 브리핑용 조립.""" + db = _db() + cal = db.get_winner_calibration(draw_no) + runs = db.get_backtest_runs(draw_no=draw_no) + hist = db.get_calibration_history(limit=12) + forward = [] + for r in runs: + forward.append({"strategy": r["strategy"], "label": r["weight_label"], + "prizes": prize_counts(r), "best_match": r["best_match"], + "avg_meta_score": r["avg_meta_score"]}) + return { + "draw_no": draw_no, + "winner_analysis": cal, # score_* + percentile + "forward": forward, + "track_record": track_record()["by_strategy"], + "calibration_trend": [ + {"draw_no": h["draw_no"], "score_total": h["score_total"], + "percentile": h["percentile"]} for h in hist + ], + } +``` + +- [ ] **Step 4: 통과 확인** — Run: `cd lotto && python -m pytest tests/test_backtest_db.py -v` Expected: PASS + +- [ ] **Step 5: Commit** +```bash +git add lotto/app/backtest.py lotto/tests/test_backtest_db.py +git commit -m "feat(lotto): track_record + build_review_payload 집계" +``` + +## Task 3.2: backtest 라우터 + +**Files:** +- Create: `lotto/app/routers/backtest.py` +- Modify: `lotto/app/main.py` (router include) +- Test: `lotto/tests/test_backtest_api.py` + +- [ ] **Step 1: 실패 테스트** + +`lotto/tests/test_backtest_api.py`: +```python +import os, tempfile +from fastapi.testclient import TestClient + +def _client(monkeypatch): + tmp = tempfile.mkdtemp() + from app import db + monkeypatch.setattr(db, "DB_PATH", os.path.join(tmp, "lotto.db")) + db.init_db() + from app.main import app + return TestClient(app), db + +def test_backtest_endpoints(monkeypatch): + client, db = _client(monkeypatch) + r = client.get("/api/lotto/backtest/track-record") + assert r.status_code == 200 + assert "by_strategy" in r.json() + r2 = client.get("/api/lotto/backtest/calibration?weeks=4") + assert r2.status_code == 200 + assert isinstance(r2.json().get("history"), list) +``` + +- [ ] **Step 2: 실패 확인** — Run: `cd lotto && python -m pytest tests/test_backtest_api.py -v` Expected: FAIL (404/import error) + +- [ ] **Step 3: 라우터 구현** — `lotto/app/routers/backtest.py`: +```python +from fastapi import APIRouter, BackgroundTasks, Query +from .. import backtest, db + +router = APIRouter(prefix="/api/lotto/backtest", tags=["backtest"]) + + +@router.get("/track-record") +def track_record(): + return backtest.track_record() + + +@router.get("/calibration") +def calibration(weeks: int = Query(52, ge=1, le=520)): + return {"history": db.get_calibration_history(limit=weeks)} + + +@router.get("/review/{draw_no}") +def review(draw_no: int): + return backtest.build_review_payload(draw_no) + + +@router.post("/run-forward") +def run_forward(draw_no: int = Query(...), k: int = 5000, pool_n: int = 20000): + return backtest.run_forward_purchase(draw_no=draw_no, k=k, pool_n=pool_n) + + +@router.post("/backfill") +def backfill(background_tasks: BackgroundTasks, batch: int = 50, sample_m: int = 2000): + background_tasks.add_task(backtest.backfill_calibration, batch, sample_m) + return {"ok": True, "message": f"backfill 시작 (batch={batch})"} +``` + +- [ ] **Step 4: main.py 등록** — `lotto/app/main.py` 상단 라우터 import 구역(`from .routers import review as review_router` 아래)에 `from .routers import backtest as backtest_router` 추가(기존과 동일한 **상대 import** 스타일), `app.include_router(review_router.router)` 다음 줄에: +```python +app.include_router(backtest_router.router) +``` + +- [ ] **Step 5: 통과 확인** — Run: `cd lotto && python -m pytest tests/test_backtest_api.py -v` Expected: PASS + +- [ ] **Step 6: Commit** +```bash +git add lotto/app/routers/backtest.py lotto/app/main.py lotto/tests/test_backtest_api.py +git commit -m "feat(lotto): backtest API 라우터 + main 등록" +``` + +## Task 3.3: 주간 forward + 캘리브레이션 스케줄러 잡 + +**Files:** +- Modify: `lotto/app/main.py` (`_sync_and_check` 확장) + +- [ ] **Step 1: `_sync_and_check` 확장** — `lotto/app/main.py`의 `_sync_and_check` 함수를 수정해 새 회차 채점 직후 forward+calibration 실행: +```python + def _sync_and_check(): + res = sync_latest(LATEST_URL) + if res["was_new"]: + check_results_for_draw(res["drawNo"]) + _refresh_perf_cache() + # 자가학습 백테스트 — 새 회차 forward 구매 + 당첨조합 캘리브레이션 + try: + from app import backtest + backtest.run_forward_purchase(draw_no=res["drawNo"]) + backtest.calibrate_winner(res["drawNo"]) + except Exception as e: + logger.warning(f"backtest 갱신 실패: {e}") +``` + +- [ ] **Step 2: 회귀 확인** — Run: `cd lotto && python -m pytest tests/ -q` Expected: 전체 PASS (기존 + 신규) + +- [ ] **Step 3: Commit** +```bash +git add lotto/app/main.py +git commit -m "feat(lotto): 새 회차 동기화 시 forward+calibration 자동 실행" +``` + +--- + +# Phase 4 — weight_evolver 학습 신호 업그레이드 + +## Task 4.1: lift-over-random 승자 선택 + ε-게이팅 (순수 함수) + +**Files:** +- Modify: `lotto/app/weight_evolver.py` +- Test: `lotto/tests/test_weight_evolver.py` + +- [ ] **Step 1: 실패 테스트** — `test_weight_evolver.py`에 추가: +```python +def test_select_winner_by_lift_gating(): + # engine_w 3개 + random_null 기준. lift = engine 등수점수 − random 등수점수 + per_w = [ + {"trial_id": 1, "day_of_week": 0, "weight": [0.2]*5, "prize_score": 5.0}, + {"trial_id": 2, "day_of_week": 1, "weight": [0.3,0.2,0.2,0.2,0.1], "prize_score": 9.0}, + {"trial_id": 3, "day_of_week": 2, "weight": [0.1,0.3,0.2,0.2,0.2], "prize_score": 4.0}, + ] + # random baseline이 8.0이면 lift는 +1, +1, -4 → 노이즈 ε=2 안에서 게이팅 + winner = we.select_winner_by_lift(per_w, random_score=8.0, epsilon=2.0) + assert winner["gated"] is True # 최대 lift(+1) < ε(2) → 게이팅 + winner2 = we.select_winner_by_lift(per_w, random_score=3.0, epsilon=2.0) + assert winner2["gated"] is False + assert winner2["trial_id"] == 2 # prize 9 → lift +6 +``` + +- [ ] **Step 2: 실패 확인** — Run: `cd lotto && python -m pytest tests/test_weight_evolver.py::test_select_winner_by_lift_gating -v` Expected: FAIL + +- [ ] **Step 3: 구현** — `lotto/app/weight_evolver.py`에 추가: +```python +LIFT_EPSILON = 0.5 # 등수점수 노이즈 게이팅 임계 (튜닝 가능) + + +def select_winner_by_lift(per_w: List[Dict[str, Any]], random_score: float, + epsilon: float = LIFT_EPSILON) -> Dict[str, Any]: + """engine_w 후보들 중 random 대비 lift 최대 선택. + 최대 lift가 epsilon 미만이면 gated=True (노이즈 → base 유지 권고).""" + scored = [{**w, "lift": w["prize_score"] - random_score} for w in per_w] + best = max(scored, key=lambda w: w["lift"]) + return {**best, "gated": best["lift"] < epsilon} +``` + +- [ ] **Step 4: 통과 확인** — Run: `cd lotto && python -m pytest tests/test_weight_evolver.py::test_select_winner_by_lift_gating -v` Expected: PASS + +- [ ] **Step 5: Commit** +```bash +git add lotto/app/weight_evolver.py lotto/tests/test_weight_evolver.py +git commit -m "feat(lotto): select_winner_by_lift + ε-게이팅" +``` + +## Task 4.2: evaluate_weekly가 forward 등수점수를 학습 신호로 사용 + +**Files:** +- Modify: `lotto/app/weight_evolver.py` (`evaluate_weekly`) +- Test: `lotto/tests/test_weight_evolver.py` + +- [ ] **Step 1: 등수점수 헬퍼 + 실패 테스트** — `test_weight_evolver.py`에 추가: +```python +def test_prize_score_from_hist(): + # 등수 가중치: 1등 매우 큼, 하위는 작게 + s = we.prize_score_from_hist({"m3": 10, "m4": 2, "m5": 0, "m6": 0, "bonus_hits": 0}) + s_big = we.prize_score_from_hist({"m3": 0, "m4": 0, "m5": 0, "m6": 1, "bonus_hits": 0}) + assert s_big > s # 1등 1장이 5등 다수보다 큼 +``` + +- [ ] **Step 2: 실패 확인** — Run: `cd lotto && python -m pytest tests/test_weight_evolver.py::test_prize_score_from_hist -v` Expected: FAIL + +- [ ] **Step 3: 구현** — `lotto/app/weight_evolver.py`에 `prize_score_from_hist` 추가: +```python +PRIZE_WEIGHTS = {"m6": 1000.0, "bonus_hits": 50.0, "m5": 30.0, "m4": 4.0, "m3": 1.0} + + +def prize_score_from_hist(hist: Dict[str, int]) -> float: + """매칭 히스토그램 → 등수 가중 합산 점수. + 1등=m6, 2등=bonus_hits, 3등=m5−bonus_hits, 4등=m4, 5등=m3.""" + third = max(0, hist.get("m5", 0) - hist.get("bonus_hits", 0)) + return (hist.get("m6", 0) * PRIZE_WEIGHTS["m6"] + + hist.get("bonus_hits", 0) * PRIZE_WEIGHTS["bonus_hits"] + + third * PRIZE_WEIGHTS["m5"] + + hist.get("m4", 0) * PRIZE_WEIGHTS["m4"] + + hist.get("m3", 0) * PRIZE_WEIGHTS["m3"]) +``` + +- [ ] **Step 4: `evaluate_weekly` 학습 신호 교체** — `lotto/app/weight_evolver.py`의 `evaluate_weekly` 내 winner 선택 로직을 backtest 기반으로 교체. 기존 per_day(auto_picks 평균) 계산은 유지하되, **base 갱신 결정은 backtest 등수점수 lift로** 수행. `winner = max(per_day, key=avg_score)` 블록 뒤·`decide_base_update` 호출 전에 삽입: +```python + # 자가학습 강화: backtest forward 등수점수 lift로 winner 재선정 + from . import backtest as bt + latest_no = latest["drw_no"] + runs = db.get_backtest_runs(draw_no=latest_no) + engine_runs = [r for r in runs if r["strategy"] == "engine_w"] + null_runs = [r for r in runs if r["strategy"] == "random_null"] + if engine_runs and null_runs: + random_score = bt.prize_score_from_hist(null_runs[0]) if False else \ + prize_score_from_hist(null_runs[0]) + per_w = [] + for r in engine_runs: + per_w.append({ + "trial_id": r["trial_id"], + "day_of_week": int(r["weight_label"][1:]) if r["weight_label"].startswith("w") else 0, + "weight": json.loads(r["weight_json"]) if r["weight_json"] else DEFAULT_UNIFORM[:], + "prize_score": prize_score_from_hist(r), + }) + lift_winner = select_winner_by_lift(per_w, random_score=random_score) + if not lift_winner["gated"]: + winner = { + "trial_id": lift_winner["trial_id"], + "day_of_week": lift_winner["day_of_week"], + "weight": lift_winner["weight"], + "avg_score": winner["avg_score"], + "max_correct": winner["max_correct"], + "lift": lift_winner["lift"], + } + else: + # 노이즈 → base 유지 강제 (max_correct를 0으로 낮춰 unchanged 유도) + winner = {**winner, "max_correct": min(winner["max_correct"], 2), "lift": lift_winner["lift"]} +``` +> 주의: `import json`은 weight_evolver.py 상단에 이미 없으면 추가. `prize_score_from_hist`/`select_winner_by_lift`는 동일 모듈 함수. + +- [ ] **Step 5: 회귀 확인** — Run: `cd lotto && python -m pytest tests/test_weight_evolver.py -q` Expected: PASS (기존 evolve 테스트 포함) + +- [ ] **Step 6: Commit** +```bash +git add lotto/app/weight_evolver.py lotto/tests/test_weight_evolver.py +git commit -m "feat(lotto): evaluate_weekly 학습 신호를 forward lift로 승격" +``` + +--- + +# Phase 5 — agent-office 일요 회고 + +## Task 5.1: service_proxy 백테스트 호출 + +**Files:** +- Modify: `agent-office/app/service_proxy.py` (lotto 섹션, `lotto_latest_draw` 부근) + +- [ ] **Step 1: 함수 추가** — `agent-office/app/service_proxy.py`의 lotto 섹션에 추가: +```python +async def lotto_backtest_review(draw_no: int) -> Dict[str, Any]: + from .config import LOTTO_BACKEND_URL + resp = await _client.get(f"{LOTTO_BACKEND_URL}/api/lotto/backtest/review/{draw_no}") + resp.raise_for_status() + return resp.json() + + +async def lotto_backtest_run_forward(draw_no: int) -> Dict[str, Any]: + from .config import LOTTO_BACKEND_URL + resp = await _client.post(f"{LOTTO_BACKEND_URL}/api/lotto/backtest/run-forward", + params={"draw_no": draw_no}) + resp.raise_for_status() + return resp.json() +``` + +- [ ] **Step 2: import 확인** — Run: `cd agent-office && python -c "from app import service_proxy"` Expected: 에러 없음 + +- [ ] **Step 3: Commit** +```bash +git add agent-office/app/service_proxy.py +git commit -m "feat(agent-office): lotto backtest review/run-forward 프록시" +``` + +## Task 5.2: 일요 회고 텔레그램 포매터 + +**Files:** +- Modify: `agent-office/app/notifiers/telegram_lotto.py` +- Test: `agent-office/app/test_db.py` 또는 신규 `agent-office/tests/test_sunday_review.py` + +- [ ] **Step 1: 실패 테스트** — 신규 `agent-office/tests/test_sunday_review.py` (없으면 tests 디렉토리 생성): +```python +from app.notifiers import telegram_lotto as tl + +def test_format_sunday_review_text(): + payload = { + "draw_no": 1170, + "winner_analysis": {"score_total": 0.41, "percentile": 0.33, + "score_frequency": 0.4, "score_fingerprint": 0.5, "score_gap": 0.3, + "score_cooccur": 0.45, "score_diversity": 0.6}, + "forward": [ + {"strategy": "engine_w", "label": "w1", "prizes": {"1st":0,"2nd":0,"3rd":0,"4th":1,"5th":12}, "best_match": 4, "avg_meta_score": 0.55}, + {"strategy": "random_null", "label": "-", "prizes": {"1st":0,"2nd":0,"3rd":0,"4th":0,"5th":10}, "best_match": 3, "avg_meta_score": 0.33}, + ], + "track_record": {}, + "calibration_trend": [{"draw_no":1170,"score_total":0.41,"percentile":0.33}], + } + txt = tl.format_sunday_review(payload) + assert "1170" in txt + assert "%" in txt # percentile 표기 + assert "engine" in txt.lower() or "엔진" in txt +``` + +- [ ] **Step 2: 실패 확인** — Run: `cd agent-office && python -m pytest tests/test_sunday_review.py -v` Expected: FAIL + +- [ ] **Step 3: 구현** — `agent-office/app/notifiers/telegram_lotto.py`에 추가: +```python +def format_sunday_review(payload: Dict[str, Any]) -> str: + """일요 회고 브리핑 텍스트 (HTML parse_mode).""" + wa = payload.get("winner_analysis") or {} + draw_no = payload.get("draw_no") + pct = wa.get("percentile") + pct_txt = f"{pct*100:.0f}%" if pct is not None else "—" + lines = [f"🔍 로또 #{draw_no} 일요 회고", ""] + if wa: + lines.append(f"이번 당첨조합 분석치: {wa.get('score_total',0):.2f} " + f"(무작위 분포 상위 {pct_txt})") + lines.append(f" 빈도 {wa.get('score_frequency',0):.2f} · 지문 {wa.get('score_fingerprint',0):.2f} " + f"· 갭 {wa.get('score_gap',0):.2f} · 공동출현 {wa.get('score_cooccur',0):.2f} " + f"· 다양성 {wa.get('score_diversity',0):.2f}") + lines.append("") + lines.append("📊 이번 회차 가상구매 성적") + for f in payload.get("forward", []): + p = f["prizes"] + name = {"engine_w": f"엔진({f['label']})", "random_null": "무작위", "coverage": "커버리지"}.get( + f["strategy"], f["strategy"]) + lines.append(f" {name}: 최고 {f['best_match']}일치 / " + f"4등 {p['4th']} · 5등 {p['5th']}") + lines.append("") + lines.append("ℹ️ 무작위 대비 우위가 통계적으로 의미있을 때만 가중치가 진화합니다.") + return "\n".join(lines) + + +async def send_sunday_review(payload: Dict[str, Any]) -> None: + from ..telegram.messaging import send_raw + await send_raw(format_sunday_review(payload)) +``` + +- [ ] **Step 4: 통과 확인** — Run: `cd agent-office && python -m pytest tests/test_sunday_review.py -v` Expected: PASS + +- [ ] **Step 5: Commit** +```bash +git add agent-office/app/notifiers/telegram_lotto.py agent-office/tests/test_sunday_review.py +git commit -m "feat(agent-office): 일요 회고 텔레그램 포매터" +``` + +## Task 5.3: LottoAgent.run_sunday_review + cron + +**Files:** +- Modify: `agent-office/app/agents/lotto.py` (메서드 + on_command) +- Modify: `agent-office/app/scheduler.py` (cron 등록) + +- [ ] **Step 1: run_sunday_review 메서드 추가** — `agent-office/app/agents/lotto.py` `LottoAgent`에: +```python + async def run_sunday_review(self) -> dict: + """일 09:00 — 최신 회차 forward+calibration 보장 후 회고 텔레그램.""" + from ..service_proxy import lotto_latest_draw, lotto_backtest_review, lotto_backtest_run_forward + from ..notifiers.telegram_lotto import send_sunday_review + from ..db import create_task, update_task_status, add_log + + task_id = create_task("lotto", "sunday_review", {}) + try: + draw_no = await lotto_latest_draw() + if not draw_no: + update_task_status(task_id, "failed", result_data={"reason": "no_draw"}) + return {"ok": False, "message": "no latest draw"} + # forward는 lotto cron이 이미 돌렸을 수 있으나 멱등이라 안전 — review만 호출 + payload = await lotto_backtest_review(draw_no) + await send_sunday_review(payload) + update_task_status(task_id, "succeeded", result_data={"draw_no": draw_no}) + add_log("lotto", f"sunday_review 발송: #{draw_no}", task_id=task_id) + return {"ok": True, "draw_no": draw_no} + except Exception as e: + update_task_status(task_id, "failed", result_data={"error": str(e)}) + add_log("lotto", f"sunday_review 예외: {e}", level="error", task_id=task_id) + return {"ok": False, "message": f"{type(e).__name__}: {e}"} +``` + +- [ ] **Step 2: on_command 분기 추가** — `lotto.py`의 `on_command`에 (`daily_digest` 분기 다음): +```python + if action == "sunday_review": + return await self.run_sunday_review() +``` + +- [ ] **Step 3: scheduler cron 등록** — `agent-office/app/scheduler.py`에 래퍼 + 등록: +```python +async def _run_lotto_sunday_review(): + agent = AGENT_REGISTRY.get("lotto") + if agent: + await agent.run_sunday_review() +``` +그리고 `init_scheduler()` 안 lotto cron 그룹에: +```python + scheduler.add_job(_run_lotto_sunday_review, "cron", day_of_week="sun", hour=9, minute=0, id="lotto_sunday_review") +``` + +- [ ] **Step 4: import 확인** — Run: `cd agent-office && python -c "from app import scheduler; from app.agents.lotto import LottoAgent"` Expected: 에러 없음 + +- [ ] **Step 5: Commit** +```bash +git add agent-office/app/agents/lotto.py agent-office/app/scheduler.py +git commit -m "feat(agent-office): LottoAgent 일 09:00 sunday_review cron" +``` + +--- + +# Phase 6 — web-ui 자율학습 탭 확장 (별도 repo: web-ui) + +> **주의:** web-ui는 별도 Git 저장소. 커밋은 `web-ui/`에서 수행([[feedback-commit-repo]]). 배포는 자동 안 됨 — `npm run release:nas` 수동. + +## Task 6.1: api.js 헬퍼 + +**Files:** +- Modify: `web-ui/src/api.js` + +- [ ] **Step 1: 헬퍼 추가** — `web-ui/src/api.js` 기존 lotto 헬퍼 근처에: +```javascript +export const lottoBacktestTrackRecord = () => get('/api/lotto/backtest/track-record'); +export const lottoBacktestCalibration = (weeks = 52) => + get(`/api/lotto/backtest/calibration?weeks=${weeks}`); +export const lottoBacktestReview = (drawNo) => + get(`/api/lotto/backtest/review/${drawNo}`); +``` +> `get` 헬퍼 이름은 기존 api.js 컨벤션 확인 후 맞출 것 (다를 경우 동일 패턴 사용). + +- [ ] **Step 2: Commit** (web-ui repo) +```bash +cd ../web-ui && git add src/api.js && git commit -m "feat: 로또 백테스트 API 헬퍼" +``` + +## Task 6.2: 성적표·캘리브레이션 컴포넌트 + 자율학습 탭 통합 + +**Files:** +- Create: `web-ui/src/pages/lotto/components/TrackRecordCard.jsx` +- Create: `web-ui/src/pages/lotto/components/CalibrationChart.jsx` +- Create: `web-ui/src/pages/lotto/components/WinnerAnalysisCard.jsx` +- Modify: 기존 `/lotto` "자율 학습" 탭 컨테이너 (Lotto 페이지의 evolver 탭 컴포넌트) + +- [ ] **Step 1: 탭 컴포넌트 위치 확인** — Run: `cd ../web-ui && grep -rn "자율 학습\|evolver\|lotto-evolver" src/pages/lotto/ | head` 로 탭 컨테이너 파일 식별. + +- [ ] **Step 2: WinnerAnalysisCard 작성** — `web-ui/src/pages/lotto/components/WinnerAnalysisCard.jsx`: +```jsx +import { Radar, RadarChart, PolarGrid, PolarAngleAxis, ResponsiveContainer } from 'recharts'; + +export default function WinnerAnalysisCard({ analysis }) { + if (!analysis) return null; + const data = [ + { k: '빈도', v: analysis.score_frequency }, + { k: '지문', v: analysis.score_fingerprint }, + { k: '갭', v: analysis.score_gap }, + { k: '공동출현', v: analysis.score_cooccur }, + { k: '다양성', v: analysis.score_diversity }, + ]; + const pct = analysis.percentile != null ? `${(analysis.percentile * 100).toFixed(0)}%` : '—'; + return ( +
+

이번 당첨조합 분석치 (무작위 상위 {pct})

+ + + + + + + +
+ ); +} +``` +> recharts는 기존 evolver UI(LineChart/Radar)에서 이미 사용 중 — import 경로 동일. + +- [ ] **Step 3: TrackRecordCard 작성** — `web-ui/src/pages/lotto/components/TrackRecordCard.jsx`: +```jsx +export default function TrackRecordCard({ byStrategy }) { + if (!byStrategy) return null; + const order = ['engine_w', 'random_null', 'coverage']; + const label = { engine_w: '엔진', random_null: '무작위', coverage: '커버리지' }; + return ( +
+

누적 성적표 (전략당 5,000장/회차)

+ + + + {order.filter((s) => byStrategy[s]).map((s) => { + const a = byStrategy[s]; + return ( + + + + + ); + })} + +
전략장수3등4등5등
{label[s]}{a.n_tickets.toLocaleString()}{a['3rd']}{a['4th']}{a['5th']}
+

엔진이 무작위를 넘지 못하면 분석에 우위가 없다는 정직한 증거입니다.

+
+ ); +} +``` + +- [ ] **Step 4: CalibrationChart 작성** — `web-ui/src/pages/lotto/components/CalibrationChart.jsx`: +```jsx +import { LineChart, Line, XAxis, YAxis, CartesianGrid, Tooltip, ResponsiveContainer } from 'recharts'; + +export default function CalibrationChart({ history }) { + if (!history?.length) return null; + const data = [...history].reverse().map((h) => ({ + draw: h.draw_no, score: h.score_total, pct: h.percentile != null ? h.percentile : null, + })); + return ( +
+

당첨조합 캘리브레이션 추세

+ + + + + + + + + + +
+ ); +} +``` + +- [ ] **Step 5: 자율학습 탭에 통합** — Step 1에서 찾은 탭 컨테이너에 3컴포넌트 추가 + 데이터 로드: +```jsx +// imports +import { lottoBacktestTrackRecord, lottoBacktestCalibration, lottoBacktestReview, lottoLatest } from '../../api'; +import TrackRecordCard from './components/TrackRecordCard'; +import CalibrationChart from './components/CalibrationChart'; +import WinnerAnalysisCard from './components/WinnerAnalysisCard'; + +// state + effect (기존 useState/useEffect 패턴 따름) +const [track, setTrack] = useState(null); +const [calib, setCalib] = useState([]); +const [winner, setWinner] = useState(null); +useEffect(() => { + (async () => { + setTrack(await lottoBacktestTrackRecord()); + const c = await lottoBacktestCalibration(52); setCalib(c.history || []); + const latest = await lottoLatest(); + if (latest?.drwNo || latest?.drw_no) { + const review = await lottoBacktestReview(latest.drwNo || latest.drw_no); + setWinner(review.winner_analysis); + } + })().catch(() => {}); +}, []); + +// render (기존 evolver 섹션 하단) + + + +``` +> `lottoLatest` 헬퍼 이름·응답 필드(`drwNo` vs `drw_no`)는 기존 api.js 확인 후 맞출 것. + +- [ ] **Step 6: 빌드 확인** — Run: `cd ../web-ui && npm run build` Expected: exit 0 + +- [ ] **Step 7: Commit** (web-ui repo) +```bash +git add src/pages/lotto/ && git commit -m "feat: 로또 자율학습 탭 — 성적표·캘리브레이션·당첨조합 분석" +``` + +--- + +# Phase 7 — 통합 검증 & 운영 트리거 + +## Task 7.1: 전체 회귀 + 백필 트리거 안내 + +- [ ] **Step 1: lotto 전체 테스트** — Run: `cd lotto && python -m pytest tests/ -q` Expected: 전체 PASS +- [ ] **Step 2: agent-office 전체 테스트** — Run: `cd agent-office && python -m pytest -q` Expected: 전체 PASS +- [ ] **Step 3: 배포 후 1회 캘리브레이션 백필** — NAS 배포(git push → webhook) 후 운영자가 1회 트리거: +```bash +# 회차가 많으므로 여러 번 호출 (멱등, batch=50씩) +curl -X POST "http://localhost:8080/api/lotto/backtest/backfill?batch=50&sample_m=1000" +# remaining=0 될 때까지 반복 (track-record/calibration로 진행 확인) +``` +> NAS 부하가 크면 sample_m을 낮추거나(예: 500), Windows WSL 이전(spec §5)을 검토. + +- [ ] **Step 4: 최종 커밋 없음 (검증만)** — 이상 없으면 finishing-a-development-branch로 머지 절차 진행. + +--- + +## Self-Review 체크리스트 결과 +- **Spec 커버리지**: 축A(forward)=Task 1.3/2.4/3.3, 축B(calibration)=Task 2.2/2.3, 축C(회고)=Task 5.2/5.3 + UI Phase 6. 데이터모델=1.1, evolver 결함수정=Phase 4, null-model=1.3/2.2, 멱등=1.1/2.3. 모두 매핑됨. +- **placeholder**: 모든 코드 step에 실제 코드 포함. UI Task 6.5의 api.js 헬퍼명/응답필드는 "기존 확인 후 맞출 것" 명시(코드베이스 의존, 합리적). +- **타입 일관성**: `grade_tickets`→`{m3..m6,bonus_hits,best_match}`, `prize_counts`/`prize_score_from_hist`가 동일 키 사용. `save_backtest_run(hist=...)`가 동일 dict 수용. `build_review_payload`의 `winner_analysis`=`get_winner_calibration` row(스키마 일치). 일관. diff --git a/docs/superpowers/specs/2026-05-31-lotto-self-learning-backtest-design.md b/docs/superpowers/specs/2026-05-31-lotto-self-learning-backtest-design.md new file mode 100644 index 0000000..0459b79 --- /dev/null +++ b/docs/superpowers/specs/2026-05-31-lotto-self-learning-backtest-design.md @@ -0,0 +1,191 @@ +# 로또 자가학습 백테스트 & 캘리브레이션 — 설계 Spec + +- **작성일**: 2026-05-31 +- **상태**: 설계 승인 (구현 plan 대기) +- **대상 서비스**: `lotto` (lotto-lab) + `agent-office` (LottoAgent) + `web-ui` (/lotto 자율학습 탭) +- **사이클**: 스마트 에이전트 고도화 3종(로또/주식/인스타) 중 **1번 로또**. 주식·인스타는 후속 사이클. + +--- + +## 1. 배경 & 목표 + +사용자(CEO)는 로또 에이전트를 "분석 번호를 계속 가상구매해 시도횟수를 늘리고, 실제 당첨조합을 역분석해 스스로 학습·디벨롭하며 일요일에 회고 브리핑하는 스마트 에이전트"로 고도화하길 원한다. 명시 목표는 "로또 1등". + +### ⚠️ 정직성 전제 (설계의 토대) +로또는 매 회차 균등·독립 추첨이다. C(45,6)=8,145,060 조합이 전부 동일 확률이며 회차 간 독립이다. 따라서: +- **과거 데이터(빈도·갭·공동출현)의 미래 예측력은 수학적으로 0.** 통계 분석으로 1등 확률을 올릴 수 없다. +- 고정 예산 N장으로 1등 확률을 최대화하는 유일한 방법은 **서로 다른(distinct) 조합 N개**를 사는 것이다. + +이 사실을 부정하지 않고 **시스템에 내장**한다. 본 프로젝트의 가치는 "예측"이 아니라: +1. **정직한 측정** — "내 분석 엔진이 무작위를 이기는가?"를 null-model 대조군으로 매번 엄밀히 검정. +2. **자가학습 엔진 인프라** — 측정→학습→회고 루프 자체의 엔지니어링. +3. **커버리지 최적화** — 1등이 목표라면 distinct 조합 커버리지 최대화가 수학적 최적. + +→ 사용자 결정(2026-05-31): **"정직한 측정 + 커버리지 최적"** 프레이밍 채택. 패턴 학습은 계속하되 모든 백테스트에 null-model 베이스라인을 내장한다. + +### 기존 자산 (100% 재활용, 신규 ML 없음) +- `analyzer.build_analysis_cache(draws)` / `score_combination(numbers, cache, weights)` — 임의 조합의 5개 sub-score + 종합점수(0~1) = **"분석치"**. +- `analyzer.build_number_weights` + `utils.weighted_sample_6` — 가중 후보 생성. +- `generator.run_simulation` — 20k 후보를 `score_combination(·, active_weights)`로 랭킹→best_picks. **W가 선택을 바꾸는 경로가 이미 존재.** +- `weight_evolver` — 토 22:00 주간 6 가중치 후보 채점→base 갱신. + +### 발견된 잠재 결함 (본 작업으로 수정) +`weight_evolver.apply_today_and_pick`은 `recommend_numbers(draws)`(W 미사용)로 픽을 뽑은 뒤 W로 점수만 매긴다. 즉 **현재 daily 픽은 W와 무관**하고, evolver가 평가하는 매칭 결과도 W-독립이라 가중치 진화가 픽 품질에 연결돼 있지 않다. → forward 가상구매를 **시뮬레이션 선택 경로(풀 생성→W 랭킹→상위 K 구매)**로 구현하면 W가 결과를 실제로 바꿔 가중치 학습이 비로소 의미를 갖고 이 결함도 해소된다. + +--- + +## 2. 핵심 개념 — Self-Learning Backtest Loop + +세 축으로 구성: + +### 축 A — Forward 가상구매 (매주, 회차당 수천 장) +매 회차 추첨 후, 각 전략별로 대량 후보를 생성·랭킹해 상위 K장을 "구매"로 간주 → 실제 당첨번호로 채점 → **회차별 집계 1행만 영구 저장**. 개별 티켓 미저장. +- 전략: `engine_w`(6개 trial 가중치 각각) / `random_null`(무작위 대조군) / `coverage`(distinct 최대화). +- 이 매칭 결과가 evolver의 학습 신호가 된다. + +### 축 B — Winner 캘리브레이션 (역대 전체 백필 + 매주 증분) +각 회차의 **실제 당첨조합을 그 시점 이전 데이터로 만든 캐시(point-in-time)에 넣어** 5개 분석치 + 종합점수 + percentile을 기록. +- percentile = 당첨조합 score_total이 그 시점 무작위 M개 표본 분포에서 차지하는 위치. +- "내 엔진이 실제 당첨번호에 높은 점수를 주는가?"의 가장 정직한 신호. 당첨조합이 일관되게 낮은 percentile이면 엔진은 헛다리. + +### 축 C — 일요일 회고 브리핑 +토 추첨(20:45)→동기화(21:10)→기존 evolver 리포트(토 22:15) 이후, **일 09:00**에 차분히 회고. 이번 회차 forward 성적 + 당첨조합 역분석 + 내 추천과 비교 + 캘리브레이션 추세 + 가중치 진화를 텔레그램 1통 + UI. + +--- + +## 3. 데이터 모델 (lotto.db 신규) + +집계 전용 — row 수 ≈ 회차 × 전략 (수천 규모, 무시 가능). + +### `backtest_runs` — forward 가상구매 집계 +``` +id INTEGER PK +draw_no INTEGER NOT NULL -- 채점 대상(당첨 확정된) 회차 +strategy TEXT NOT NULL -- 'engine_w' | 'random_null' | 'coverage' +weight_label TEXT NOT NULL -- engine_w는 trial day_of_week('w0'..'w5'), 그 외 '-' +weight_json TEXT -- 사용한 W (random/coverage는 NULL) +trial_id INTEGER -- FK weight_trials (engine_w만, nullable) +n_tickets INTEGER NOT NULL -- 구매(채점) 장수 +m3 INTEGER NOT NULL DEFAULT 0 -- 3개 일치 장수 +m4 INTEGER NOT NULL DEFAULT 0 +m5 INTEGER NOT NULL DEFAULT 0 +m6 INTEGER NOT NULL DEFAULT 0 +bonus_hits INTEGER NOT NULL DEFAULT 0 -- 5+보너스(2등) 장수 +best_match INTEGER NOT NULL DEFAULT 0 +avg_meta_score REAL -- 구매 티켓 평균 분석치 +created_at TEXT NOT NULL +UNIQUE(draw_no, strategy, weight_label) -- 멱등 +``` +- 등수 매핑: 1등=m6, 2등=bonus_hits, 3등=m5−bonus_hits, 4등=m4, 5등=m3. + +### `winner_calibration` — 회차별 당첨조합 역분석 +``` +draw_no INTEGER PK -- 멱등 +winning_json TEXT NOT NULL -- [n1..n6] (보너스 별도 보관 안 함) +score_total REAL NOT NULL +score_frequency REAL NOT NULL +score_fingerprint REAL NOT NULL +score_gap REAL NOT NULL +score_cooccur REAL NOT NULL +score_diversity REAL NOT NULL +percentile REAL -- 0~1, 무작위 M표본 대비 당첨조합 점수 위치 +my_pick_avg REAL -- 그 회차 engine 추천 평균 분석치(있으면) +cache_draws INTEGER NOT NULL -- point-in-time 캐시에 쓰인 회차 수 +created_at TEXT NOT NULL +``` + +> 누적 성적표(track record)는 `backtest_runs` SUM 집계로 on-the-fly 계산 — 별도 테이블 불필요. + +--- + +## 4. 컴포넌트 + +### 4.1 lotto-lab `app/backtest.py` (순수 연산 — FastAPI 의존성 0, Windows 이전 대비) +- `generate_pool(cache, number_weights, n) -> list[tuple]` — `weighted_sample_6` 반복으로 distinct 후보 풀. +- `purchase_tickets(pool, cache, W, k) -> list[dict]` — 풀을 `score_combination(·, W)`로 랭킹→상위 k장 distinct. +- `coverage_select(pool, k) -> list` — distinct 보장 상위 커버리지(초기엔 단순 distinct, 휠링은 향후). +- `grade_tickets(tickets, winning6, bonus) -> dict` — 매칭 히스토그램 + 등수 카운트 + best_match + avg_meta. `bonus`는 draws 레코드에서 가져옴(2등=5일치+보너스 판정용). +- `run_forward_purchase(draw_no, k=5000, pool_n=20000) -> dict` — engine(6 W)+random_null+coverage 각각 **전략당 k=5000장(수천 장)** 구매·채점·`backtest_runs` 저장(멱등). 풀 pool_n=20000에서 랭킹. +- `calibrate_winner(draw_no, sample_m=2000) -> dict` — `draws[:idx]`(대상 회차 제외) 캐시로 당첨조합 채점 + 무작위 sample_m 표본 percentile → `winner_calibration` 저장(멱등). +- `backfill_calibration(batch=50) -> dict` — 미처리 회차만 청크 처리, 재개 가능. +- `build_review_payload(draw_no) -> dict` — 회고 브리핑용 조립(당첨조합 분해 + 내 추천 비교 + forward 성적 + 캘리브레이션 추세 + 진화 결과). + +### 4.2 lotto-lab `app/routers/backtest.py` +| 메서드 | 경로 | 설명 | +|--------|------|------| +| GET | `/api/lotto/backtest/track-record` | 누적 성적표(전략별 등수 카운트, engine vs random) | +| GET | `/api/lotto/backtest/calibration?weeks=N` | 캘리브레이션 이력 + 추세 | +| GET | `/api/lotto/backtest/review/{draw_no}` | 회고 payload | +| POST | `/api/lotto/backtest/run-forward?draw_no=` | forward 수동 트리거 | +| POST | `/api/lotto/backtest/backfill` | 캘리브레이션 백필(백그라운드) | + +### 4.3 weight_evolver 업그레이드 +- `evaluate_weekly`: 학습 신호를 N=5(W-무관)에서 **forward 가상구매(engine_w 6전략) + null-model 대비 lift**로 승격. + - lift = engine_w 등수 점수 − random_null 등수 점수(동일 회차). + - 승자 = lift 최대 trial. **모든 W의 lift가 노이즈 범위(±ε) 내면 base `unchanged`** → 노이즈 과적합 방지. +- `decide_base_update` 규칙은 유지하되 입력(winner)을 backtest 기반으로 교체. +- 기존 `auto_picks` 경로는 하위호환·일일 활동표시용으로 유지(evolver 결정에는 미사용). + +--- + +## 5. 플로우 + +1. **캘리브레이션 백필 (1회)**: `POST /backtest/backfill` → 백그라운드 청크(50회차/배치, 멱등 재개). 이후 회차마다 증분. +2. **주간 forward**: 당첨번호 동기화 직후 `run_forward_purchase(latest)`. 참고: 6 W × 20k 풀은 기존 시뮬이 **하루 6회** 돌리는 부하보다 가벼움 → NAS 부담 작음. +3. **일 09:00 회고 (agent-office 신규 cron)**: `LottoAgent.run_sunday_review()` → forward+calibration 보장 → `GET /backtest/review/{latest}` → 텔레그램 1통. +4. **evolver (토 22:00, 기존 cron)**: backtest 집계를 학습 신호로 소비. + +### Windows 이전 경로 (NAS 부하 측정 후 필요시) +`backtest.py`가 순수 함수라, lotto-lab은 system-of-record 유지 + 무거운 연산만 Windows WSL docker 워커에 위임(`/api/internal/lotto/*` webhook, 기존 music/video/image 워커 패턴 재활용) + agent 폴링. 코드 경계가 깨끗해 마이그레이션 비용 최소. **초기 구현은 NAS-first**, 측정 후 결정. + +--- + +## 6. 출력 + +### 6.1 텔레그램 (일 09:00, `notifiers/telegram_lotto.py` 신규 섹션) +이번 당첨조합 5분석치 분해 + 내 추천 평균과 비교 + 이번주 forward 성적(등수 카운트, **무작위 대비 lift**) + 캘리브레이션 percentile 추세 + 가중치 진화 결과. + +### 6.2 web-ui `/lotto` "자율 학습" 탭 확장 (`.lotto-evolver-*` 다크 네임스페이스 재활용) +- **TrackRecordCard**: 누적 "매주 전략당 5,000장 샀다면" 등수 — engine vs random_null 나란히 + 총지출 대비 당첨금(정직하게 적자 표시). +- **CalibrationChart**: 당첨조합 score_total 추세 + 내 추천 평균 오버레이 + percentile 밴드 → "우위 없음"을 시각화. +- **WinnerAnalysisCard**: 이번 회차 당첨조합 5분석치 레이더 + 내 추천 비교. + +--- + +## 7. 에러·성능·멱등 +- **멱등성**: `winner_calibration` UNIQUE(draw_no), `backtest_runs` UNIQUE(draw_no,strategy,weight_label) → 재실행 skip. +- **NAS 성능**: 주간 forward는 기존 시뮬보다 가벼움. 백필만 1회 무거움(≈1100 point-in-time 캐시 재구성) → 청크+백그라운드+멱등 재개. 야간/유휴 트리거 권장. +- **텔레그램 실패**: 로그만 남기고 job은 성공 처리(기존 패턴). 회고 데이터는 이미 DB에 있어 UI는 영향 없음. + +## 8. 테스트 전략 +- 등수 매핑(m3~m6/bonus → 1~5등) 단위 테스트. +- null-model 기대값 + lift 계산. +- percentile 계산 정확성. +- **point-in-time 캐시가 대상 회차를 제외하는지** (calibrate_winner 정직성 핵심). +- 멱등 백필(재실행 시 중복 row 없음, 중단 후 재개). +- evolver의 lift-over-random 승자 선택 + ε-게이팅(노이즈 시 unchanged). +- 기존 `count_match`/`calc_pick_score` 테스트 유지. + +## 9. 리스크 & 완화 +| 리스크 | 완화 | +|--------|------| +| 무작위성 → 실제 우위 없음 | null-model 정직 프레이밍, 우위 없음을 데이터로 보고하는 게 목표 | +| Celeron 백필 부하 | 청크+1회성+멱등 재개, 필요시 Windows 이전 | +| evolver 노이즈 추종 | lift-over-random + ε-게이팅으로 unchanged 처리 | +| DB 증가 | 집계 전용, row 수 무시 가능 | +| forward 풀 중복으로 커버리지 손실 | distinct 강제 + coverage 전략 별도 측정 | + +## 10. 결정 로그 (2026-05-31 brainstorming) +1. 3종 중 **로또 먼저**, 주식·인스타는 후속 사이클. +2. 회고 브리핑 = **토 추첨 직후 일 09:00**. +3. 시도 규모 = **수천 장/회차 + 집계만 저장**. +4. 자율성 = **가중치 자동튜닝 강화**(산식 구조 고정). +5. 백테스트 범위 = **캘리브레이션 전체 백필 + 가상구매 forward**. +6. 출력 = **텔레그램 + 기존 자율학습 탭 확장**. +7. 프레이밍 = **정직한 측정(null-model) + 커버리지 최적**. +8. 연산 위치 = **NAS-first, 필요시 Windows WSL 이전**. + +## 11. 스코프 밖 / 향후 +- 주식 에이전트(보유종목 집중 분석+차트 매수/매도 시그널), 인스타 에이전트(자율 카드 발급) — 별도 사이클. +- 휠링/커버링 디자인(하위 등수 최소 보장) — coverage 전략 고도화로 향후. +- Windows WSL 워커 분리 — NAS 부하 측정 후. diff --git a/lotto/app/backtest.py b/lotto/app/backtest.py new file mode 100644 index 0000000..a9ef617 --- /dev/null +++ b/lotto/app/backtest.py @@ -0,0 +1,272 @@ +"""로또 자가학습 백테스트 — 순수 연산 (FastAPI 의존성 0, Windows 이전 대비).""" +import logging +import random +from typing import Any, Dict, List, Optional, Tuple + +from .analyzer import build_analysis_cache, build_number_weights, score_combination +from .utils import weighted_sample_6 + +# engine_w trials 수와 동일하게 맞춰 selection bias를 상쇄한다. +N_NULL_TRIALS = 6 + + +def grade_tickets(tickets: List[List[int]], winning6: List[int], bonus: int) -> Dict[str, Any]: + """티켓 묶음을 당첨번호로 채점 → 매칭 히스토그램 + 보너스 + best_match. + 2등 판정: 5일치 AND 보너스 번호를 티켓이 포함.""" + win = set(winning6) + hist = {"m3": 0, "m4": 0, "m5": 0, "m6": 0, "bonus_hits": 0} + best = 0 + for t in tickets: + c = len(set(t) & win) + if c > best: + best = c + if c == 6: + hist["m6"] += 1 + elif c == 5: + hist["m5"] += 1 + if bonus in t: + hist["bonus_hits"] += 1 + elif c == 4: + hist["m4"] += 1 + elif c == 3: + hist["m3"] += 1 + return {**hist, "best_match": best} + + +def prize_counts(hist: Dict[str, Any]) -> Dict[str, int]: + """매칭 히스토그램 → 등수 카운트. + 1등=m6, 2등=bonus_hits, 3등=m5−bonus_hits, 4등=m4, 5등=m3.""" + return { + "1st": hist.get("m6", 0), + "2nd": hist.get("bonus_hits", 0), + "3rd": hist.get("m5", 0) - hist.get("bonus_hits", 0), + "4th": hist.get("m4", 0), + "5th": hist.get("m3", 0), + } + + +def generate_pool(cache, number_weights, n: int = 20000, + seed: Optional[int] = None) -> List[List[int]]: + """가중 샘플링으로 distinct 후보 풀 생성.""" + rng = random.Random(seed) + seen, pool = set(), [] + attempts, cap = 0, n * 4 + while len(pool) < n and attempts < cap: + attempts += 1 + nums = tuple(sorted(weighted_sample_6(number_weights))) + if nums in seen: + continue + seen.add(nums) + pool.append(list(nums)) + if len(pool) < n: + logging.getLogger(__name__).warning( + "generate_pool: requested %d, got %d", n, len(pool) + ) + return pool + + +def purchase_tickets(pool, cache, W: List[float], k: int) -> List[List[int]]: + """풀을 score_combination(·, W)로 랭킹 → 상위 k장 distinct.""" + if k > len(pool): + raise ValueError(f"k={k} exceeds pool size {len(pool)}") + ranked = sorted(pool, key=lambda t: -score_combination(t, cache, W)["score_total"]) + return ranked[:k] + + +def random_null_tickets(k: int, seed: Optional[int] = None) -> List[List[int]]: + """무작위 distinct 티켓 k장 (null-model 대조군).""" + rng = random.Random(seed) + seen, out = set(), [] + guard = 0 + while len(out) < k and guard < k * 200: + guard += 1 + nums = tuple(sorted(rng.sample(range(1, 46), 6))) + if nums in seen: + continue + seen.add(nums) + out.append(list(nums)) + return out + + +def point_in_time_draws(draws: List[Tuple[int, List[int]]], + target_draw_no: int) -> List[Tuple[int, List[int]]]: + """target 회차 추첨 '직전' 시점의 데이터 — target_draw_no 미만만.""" + return [(d, nums) for d, nums in draws if d < target_draw_no] + + +def calibrate_winner_compute(draws, target_draw_no, winning6, + sample_m: int = 2000, seed: Optional[int] = None) -> Dict[str, Any]: + """순수 연산: point-in-time 캐시로 당첨조합 채점 + 무작위 M표본 percentile.""" + pit = point_in_time_draws(draws, target_draw_no) + cache = build_analysis_cache(pit) + scores = score_combination(sorted(winning6), cache) + win_total = scores["score_total"] + samples = random_null_tickets(sample_m, seed=seed) + le = sum(1 for t in samples + if score_combination(t, cache)["score_total"] <= win_total) + percentile = le / max(len(samples), 1) + return {"scores": scores, "percentile": percentile, "cache_draws": len(pit)} + + +MIN_HISTORY = 30 # point-in-time 캐시 최소 회차 (이 미만은 캘리브레이션 skip) + + +def _db(): + from . import db as _db_mod + return _db_mod + + +def calibrate_winner(draw_no: int, sample_m: int = 2000, draws=None) -> Dict[str, Any]: + """DB 진입점: 회차 1개 캘리브레이션 후 저장 (멱등). + draws를 외부에서 전달하면 N+1 조회를 방지한다.""" + db = _db() + if draws is None: + draws = db.get_all_draw_numbers() + row = db.get_draw(draw_no) + if row is None: + return {"ok": False, "reason": "no_draw"} + pit = point_in_time_draws(draws, draw_no) + if len(pit) < MIN_HISTORY: + return {"ok": False, "reason": "insufficient_history"} + winning6 = [row["n1"], row["n2"], row["n3"], row["n4"], row["n5"], row["n6"]] + res = calibrate_winner_compute(draws, draw_no, winning6, sample_m=sample_m) + db.save_winner_calibration( + draw_no=draw_no, winning=winning6, scores=res["scores"], + percentile=res["percentile"], my_pick_avg=None, + cache_draws=res["cache_draws"], + ) + return {"ok": True, "draw_no": draw_no, **res} + + +def backfill_calibration(batch: int = 50, sample_m: int = 2000) -> Dict[str, Any]: + """미처리 회차만 batch개 캘리브레이션 (멱등·재개 가능).""" + db = _db() + draws = db.get_all_draw_numbers() + done = db.get_calibrated_draw_nos() + todo = [d for d, _ in draws if d not in done and d > MIN_HISTORY] + todo.sort() + n = 0 + for draw_no in todo[:batch]: + r = calibrate_winner(draw_no, sample_m=sample_m, draws=draws) + if r.get("ok"): + n += 1 + return {"calibrated": n, "remaining": max(0, len(todo) - batch)} + + +def run_forward_purchase(draw_no: int, k: int = 5000, pool_n: int = 20000, + sample_seed: Optional[int] = None) -> Dict[str, Any]: + """회차 추첨 '직전' 시점 데이터로 3전략 구매 → 당첨번호로 채점 → 저장(멱등). + engine_w: 그 주 weight_trials 6개(없으면 current_base 1개)로 각각 구매.""" + db = _db() + draws = db.get_all_draw_numbers() + row = db.get_draw(draw_no) + if row is None: + return {"ok": False, "reason": "no_draw"} + pit = point_in_time_draws(draws, draw_no) + if len(pit) < MIN_HISTORY: + return {"ok": False, "reason": "insufficient_history"} + winning6 = [row["n1"], row["n2"], row["n3"], row["n4"], row["n5"], row["n6"]] + bonus = row["bonus"] + + cache = build_analysis_cache(pit) + nw = build_number_weights(cache) + pool = generate_pool(cache, nw, n=pool_n, seed=sample_seed) + + def _store(strategy, label, weight_json, trial_id, tickets): + graded = grade_tickets(tickets, winning6, bonus) + avg_meta = (sum(score_combination(t, cache)["score_total"] for t in tickets) + / max(len(tickets), 1)) + db.save_backtest_run( + draw_no=draw_no, strategy=strategy, weight_label=label, + weight_json=weight_json, trial_id=trial_id, n_tickets=len(tickets), + hist=graded, best_match=graded["best_match"], avg_meta_score=avg_meta, + ) + + # 1) engine_w — 그 주 trials(있으면) 아니면 uniform fallback (leak-free) + from datetime import date as _date + from . import weight_evolver as we + draw_date = _date.fromisoformat(row["drw_date"]) + week_start = we.get_week_start(draw_date) + trials = db.get_weekly_trials(week_start) + if trials: + for t in trials: + bought = purchase_tickets(pool, cache, t["weight"], k) + _store("engine_w", f"w{t['day_of_week']}", t["weight"], t["id"], bought) + else: + base = [0.2] * 5 + bought = purchase_tickets(pool, cache, base, k) + _store("engine_w", "base", base, None, bought) + + # 2) random_null — N_NULL_TRIALS 개 (engine_w 수와 동일해 selection bias 상쇄) + for _i in range(N_NULL_TRIALS): + seed_i = None if sample_seed is None else sample_seed + 100 + _i + _store("random_null", f"r{_i}", None, None, random_null_tickets(k, seed=seed_i)) + # 3) coverage + _store("coverage", "-", None, None, coverage_tickets(k, seed=sample_seed)) + + return {"ok": True, "draw_no": draw_no} + + +def track_record() -> Dict[str, Any]: + """전략별 누적 등수 집계 (engine_w는 라벨 합산).""" + db = _db() + rows = db.get_backtest_runs() + agg: Dict[str, Dict[str, int]] = {} + draw_sets: Dict[str, set] = {} + for r in rows: + a = agg.setdefault(r["strategy"], { + "n_tickets": 0, "1st": 0, "2nd": 0, "3rd": 0, "4th": 0, "5th": 0, "draws": 0}) + p = prize_counts(r) + a["n_tickets"] += r["n_tickets"] + for tier in ("1st", "2nd", "3rd", "4th", "5th"): + a[tier] += p[tier] + draw_sets.setdefault(r["strategy"], set()).add(r["draw_no"]) + for strat, s in draw_sets.items(): + agg[strat]["draws"] = len(s) + return {"by_strategy": agg} + + +def build_review_payload(draw_no: int) -> Dict[str, Any]: + """일요 회고 브리핑용 조립.""" + db = _db() + cal = db.get_winner_calibration(draw_no) + runs = db.get_backtest_runs(draw_no=draw_no) + hist = db.get_calibration_history(limit=12) + forward = [] + for r in runs: + forward.append({"strategy": r["strategy"], "label": r["weight_label"], + "prizes": prize_counts(r), "best_match": r["best_match"], + "avg_meta_score": r["avg_meta_score"]}) + return { + "draw_no": draw_no, + "winner_analysis": cal, # score_* + percentile + "forward": forward, + "track_record": track_record()["by_strategy"], + "calibration_trend": [ + {"draw_no": h["draw_no"], "score_total": h["score_total"], + "percentile": h["percentile"]} for h in hist + ], + } + + +def coverage_tickets(k: int, seed: Optional[int] = None) -> List[List[int]]: + """greedy 커버리지 — 아직 덜 쓰인 번호를 우선 배치해 번호를 넓게 분산. + (휠링/보장설계는 향후. 현재는 distinct + 번호 사용 균등화)""" + rng = random.Random(seed) + usage = {n: 0 for n in range(1, 46)} + seen, out = set(), [] + guard = 0 + while len(out) < k and guard < k * 50: + guard += 1 + ranked = sorted(range(1, 46), key=lambda n: (usage[n], rng.random())) + nums = tuple(sorted(ranked[:6])) + if nums in seen: + # 동점 흔들기: top-6과 disjoint한 영역에서 샘플 + nums = tuple(sorted(rng.sample(ranked[6:12], 6))) + if nums in seen: + continue + seen.add(nums) + out.append(list(nums)) + for n in nums: + usage[n] += 1 + return out diff --git a/lotto/app/db.py b/lotto/app/db.py index af47d4d..a85a53e 100644 --- a/lotto/app/db.py +++ b/lotto/app/db.py @@ -125,6 +125,48 @@ def init_db() -> None: "ON simulation_candidates(is_best, score_total DESC);" ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS backtest_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + draw_no INTEGER NOT NULL, + strategy TEXT NOT NULL, + weight_label TEXT NOT NULL DEFAULT '-', + weight_json TEXT, + trial_id INTEGER, + n_tickets INTEGER NOT NULL, + m3 INTEGER NOT NULL DEFAULT 0, + m4 INTEGER NOT NULL DEFAULT 0, + m5 INTEGER NOT NULL DEFAULT 0, + m6 INTEGER NOT NULL DEFAULT 0, + bonus_hits INTEGER NOT NULL DEFAULT 0, + best_match INTEGER NOT NULL DEFAULT 0, + avg_meta_score REAL, + created_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + """ + ) + conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS uq_backtest_run " + "ON backtest_runs(draw_no, strategy, weight_label);") + conn.execute( + """ + CREATE TABLE IF NOT EXISTS winner_calibration ( + draw_no INTEGER PRIMARY KEY, + winning_json TEXT NOT NULL, + score_total REAL NOT NULL, + score_frequency REAL NOT NULL, + score_fingerprint REAL NOT NULL, + score_gap REAL NOT NULL, + score_cooccur REAL NOT NULL, + score_diversity REAL NOT NULL, + percentile REAL, + my_pick_avg REAL, + cache_draws INTEGER NOT NULL, + created_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + """ + ) + conn.execute( """ CREATE TABLE IF NOT EXISTS best_picks ( @@ -1443,3 +1485,79 @@ def get_base_history(limit: int = 12) -> List[Dict[str, Any]]: out.append(d) return out + +# ── backtest_runs / winner_calibration CRUD ─────────────────────────────────── + +def save_backtest_run(draw_no, strategy, weight_label, weight_json, trial_id, + n_tickets, hist, best_match, avg_meta_score) -> None: + with _conn() as conn: + conn.execute( + """ + INSERT INTO backtest_runs + (draw_no, strategy, weight_label, weight_json, trial_id, n_tickets, + m3, m4, m5, m6, bonus_hits, best_match, avg_meta_score) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(draw_no, strategy, weight_label) DO UPDATE SET + weight_json=excluded.weight_json, trial_id=excluded.trial_id, + n_tickets=excluded.n_tickets, m3=excluded.m3, m4=excluded.m4, + m5=excluded.m5, m6=excluded.m6, bonus_hits=excluded.bonus_hits, + best_match=excluded.best_match, avg_meta_score=excluded.avg_meta_score + """, + (draw_no, strategy, weight_label, + # weight_json must be a dict/list (not a pre-serialized string) to avoid double-encoding + json.dumps(weight_json) if weight_json is not None else None, + trial_id, n_tickets, + hist.get("m3",0), hist.get("m4",0), hist.get("m5",0), hist.get("m6",0), + hist.get("bonus_hits",0), best_match, avg_meta_score), + ) + +def get_backtest_runs(draw_no=None, strategy=None) -> List[Dict[str, Any]]: + q = "SELECT * FROM backtest_runs WHERE 1=1" + args = [] + if draw_no is not None: + q += " AND draw_no=?"; args.append(draw_no) + if strategy is not None: + q += " AND strategy=?"; args.append(strategy) + q += " ORDER BY draw_no DESC, strategy, weight_label" + with _conn() as conn: + return [dict(r) for r in conn.execute(q, args).fetchall()] + +def save_winner_calibration(draw_no, winning, scores, percentile, + my_pick_avg, cache_draws) -> None: + with _conn() as conn: + conn.execute( + """ + INSERT INTO winner_calibration + (draw_no, winning_json, score_total, score_frequency, score_fingerprint, + score_gap, score_cooccur, score_diversity, percentile, my_pick_avg, cache_draws) + VALUES (?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(draw_no) DO UPDATE SET + winning_json=excluded.winning_json, score_total=excluded.score_total, + score_frequency=excluded.score_frequency, score_fingerprint=excluded.score_fingerprint, + score_gap=excluded.score_gap, score_cooccur=excluded.score_cooccur, + score_diversity=excluded.score_diversity, percentile=excluded.percentile, + my_pick_avg=excluded.my_pick_avg, cache_draws=excluded.cache_draws + """, + (draw_no, json.dumps(winning), scores["score_total"], scores["score_frequency"], + scores["score_fingerprint"], scores["score_gap"], scores["score_cooccur"], + scores["score_diversity"], percentile, my_pick_avg, cache_draws), + ) + +def get_winner_calibration(draw_no: int) -> Optional[Dict[str, Any]]: + with _conn() as conn: + r = conn.execute("SELECT * FROM winner_calibration WHERE draw_no=?", + (draw_no,)).fetchone() + return dict(r) if r else None + +def get_calibration_history(limit: int = 52) -> List[Dict[str, Any]]: + with _conn() as conn: + rows = conn.execute( + "SELECT * FROM winner_calibration ORDER BY draw_no DESC LIMIT ?", + (limit,)).fetchall() + return [dict(r) for r in rows] + +def get_calibrated_draw_nos() -> set[int]: + with _conn() as conn: + return {r["draw_no"] for r in + conn.execute("SELECT draw_no FROM winner_calibration").fetchall()} + diff --git a/lotto/app/main.py b/lotto/app/main.py index f6e9497..9e6f86b 100644 --- a/lotto/app/main.py +++ b/lotto/app/main.py @@ -47,13 +47,16 @@ from .weight_evolver import ( from .routers import curator as curator_router from .routers import briefing as briefing_router from .routers import review as review_router +from .routers import backtest as backtest_router from .jobs.grade_weekly_review import run_for_latest as grade_run_for_latest +from . import backtest app = FastAPI() install_access_log(app) app.include_router(curator_router.router) app.include_router(briefing_router.router) app.include_router(review_router.router) +app.include_router(backtest_router.router) scheduler = BackgroundScheduler(timezone=os.getenv("TZ", "Asia/Seoul")) ALL_URL = os.getenv("LOTTO_ALL_URL", "https://smok95.github.io/lotto/results/all.json") @@ -82,6 +85,12 @@ def on_startup(): if res["was_new"]: check_results_for_draw(res["drawNo"]) _refresh_perf_cache() # 새 채점 결과 반영 → 즉시 갱신 + # 자가학습 백테스트 — 새 회차 forward 구매 + 당첨조합 캘리브레이션 + try: + backtest.run_forward_purchase(draw_no=res["drawNo"]) + backtest.calibrate_winner(res["drawNo"]) + except Exception as e: + logger.warning(f"backtest 갱신 실패: {e}") scheduler.add_job(_sync_and_check, "cron", hour="9,21", minute=10) diff --git a/lotto/app/routers/backtest.py b/lotto/app/routers/backtest.py new file mode 100644 index 0000000..b445419 --- /dev/null +++ b/lotto/app/routers/backtest.py @@ -0,0 +1,39 @@ +from fastapi import APIRouter, BackgroundTasks, Query +from .. import backtest, db + +router = APIRouter(prefix="/api/lotto/backtest", tags=["backtest"]) + + +@router.get("/track-record") +def track_record(): + return backtest.track_record() + + +@router.get("/calibration") +def calibration(weeks: int = Query(52, ge=1, le=520)): + return {"history": db.get_calibration_history(limit=weeks)} + + +@router.get("/review/{draw_no}") +def review(draw_no: int): + if db.get_draw(draw_no) is None: + from fastapi import HTTPException + raise HTTPException(404, f"no draw {draw_no}") + return backtest.build_review_payload(draw_no) + + +@router.post("/run-forward") +def run_forward( + background_tasks: BackgroundTasks, + draw_no: int = Query(...), + k: int = Query(5000, ge=1, le=5000), + pool_n: int = Query(20000, ge=1000, le=20000), +): + background_tasks.add_task(backtest.run_forward_purchase, draw_no, k, pool_n) + return {"ok": True, "queued": True, "draw_no": draw_no} + + +@router.post("/backfill") +def backfill(background_tasks: BackgroundTasks, batch: int = 50, sample_m: int = 2000): + background_tasks.add_task(backtest.backfill_calibration, batch, sample_m) + return {"ok": True, "message": f"backfill 시작 (batch={batch})"} diff --git a/lotto/app/weight_evolver.py b/lotto/app/weight_evolver.py index 6969748..c323e58 100644 --- a/lotto/app/weight_evolver.py +++ b/lotto/app/weight_evolver.py @@ -4,6 +4,7 @@ 순수 함수 (clamp/perturb/Dirichlet/score/base-rule) + DB 진입점은 별도 섹션. """ from __future__ import annotations +import json import math import random from datetime import datetime, timedelta, timezone @@ -18,6 +19,34 @@ DEFAULT_UNIFORM = [0.2] * N_METRICS # cold start RANK_BY_CORRECT = {6: 1, 5: 3, 4: 4, 3: 5} RANK_BONUS = {1: 1.0, 2: 0.8, 3: 0.6, 4: 0.3, 5: 0.1} +LIFT_EPSILON = 10.0 # best-of-engine vs best-of-random margin; +# selection bias already cancelled by equal group sizes (N_NULL_TRIALS == engine trial count); +# tune as needed. + +PRIZE_WEIGHTS = {"m6": 1000.0, "bonus_hits": 50.0, "m5": 30.0, "m4": 4.0, "m3": 1.0} + + +def select_winner_by_lift(per_w: List[Dict[str, Any]], random_score: float, + epsilon: float = LIFT_EPSILON) -> Dict[str, Any]: + """engine_w 후보들 중 random 대비 lift 최대 선택. + 최대 lift가 epsilon 미만이면 gated=True (노이즈 → base 유지 권고).""" + scored = [{**w, "lift": w["prize_score"] - random_score} for w in per_w] + best = max(scored, key=lambda w: w["lift"]) + return {**best, "gated": best["lift"] < epsilon} + + +def prize_score_from_hist(hist: Dict[str, int]) -> float: + """매칭 히스토그램 → 등수 가중 합산 점수. + 1등=m6, 2등=bonus_hits, 3등=m5−bonus_hits, 4등=m4, 5등=m3. + m3/m4/m5/m6/bonus_hits 키만 읽으며 나머지는 무시하므로 + DB 전체 행(backtest_runs row)을 그대로 넘겨도 안전하다.""" + third = max(0, hist.get("m5", 0) - hist.get("bonus_hits", 0)) + return (hist.get("m6", 0) * PRIZE_WEIGHTS["m6"] + + hist.get("bonus_hits", 0) * PRIZE_WEIGHTS["bonus_hits"] + + third * PRIZE_WEIGHTS["m5"] + + hist.get("m4", 0) * PRIZE_WEIGHTS["m4"] + + hist.get("m3", 0) * PRIZE_WEIGHTS["m3"]) + def clamp_and_normalize(W: List[float], min_w: float = MIN_WEIGHT) -> List[float]: """각 값 ≥ min_w + 합=1.0. 보장 안 되면 raise.""" @@ -269,6 +298,47 @@ def evaluate_weekly() -> Dict[str, Any]: winner = max(per_day, key=lambda d: d["avg_score"]) + # 자가학습 강화: backtest forward 등수점수 lift로 winner 재선정. + # best-of-engine vs best-of-random 비교 — 동등 그룹 크기로 selection bias 상쇄. + latest_no = latest["drw_no"] + runs = db.get_backtest_runs(draw_no=latest_no) + engine_runs = [r for r in runs if r["strategy"] == "engine_w"] + null_runs = [r for r in runs if r["strategy"] == "random_null"] + gated = False # 이후 decide_base_update override에 사용 + if engine_runs and null_runs: + # base 단독 행이 있고 w* 행도 있으면 base 행 제외 (identity collision 방지) + has_w_trials = any(r["weight_label"].startswith("w") for r in engine_runs) + if has_w_trials: + engine_runs = [r for r in engine_runs if r["weight_label"] != "base"] + + # best-of-random: 동등 그룹의 최댓값 (selection bias 상쇄) + random_best = max(prize_score_from_hist(r) for r in null_runs) + + per_w = [] + for r in engine_runs: + per_w.append({ + "trial_id": r["trial_id"], + "weight_label": r["weight_label"], + "weight": json.loads(r["weight_json"]) if r["weight_json"] else DEFAULT_UNIFORM[:], + "prize_score": prize_score_from_hist(r), + "best_match": r["best_match"], + }) + + lift_winner = select_winner_by_lift(per_w, random_score=random_best) + if not lift_winner["gated"]: + # lift winner의 정체성과 채점값을 일관되게 사용 + winner = { + "trial_id": lift_winner["trial_id"], + "weight": lift_winner["weight"], + "max_correct": lift_winner["best_match"], # 이 trial의 실제값 + "avg_score": lift_winner["prize_score"], # lift winner의 prize score + "lift": lift_winner["lift"], + } + else: + # 노이즈 → gated 플래그 설정; decide_base_update 이후 명시적으로 override + gated = True + winner = {**winner, "lift": lift_winner["lift"]} + current_base = db.get_current_base() new_base, reason = decide_base_update( winner_max_correct=winner["max_correct"], @@ -276,6 +346,11 @@ def evaluate_weekly() -> Dict[str, Any]: current_base=current_base, ) + # gated path: decide_base_update 결과와 무관하게 base 유지 강제 + if gated: + new_base = list(current_base) if current_base is not None else DEFAULT_UNIFORM[:] + reason = "unchanged_gated" + next_monday = today + timedelta(days=(7 - today.weekday()) % 7 or 7) next_monday_iso = next_monday.isoformat() diff --git a/lotto/tests/test_backtest.py b/lotto/tests/test_backtest.py new file mode 100644 index 0000000..517aa3f --- /dev/null +++ b/lotto/tests/test_backtest.py @@ -0,0 +1,100 @@ +from app import backtest as bt +from app.analyzer import build_analysis_cache, build_number_weights, score_combination + + +def _toy_draws(n=120): + # 결정적 가짜 회차: 분석 캐시 구성용 (오름차순 (drw_no, [6 nums])) + import random as _r + _r.seed(1) + out = [] + for i in range(1, n + 1): + nums = sorted(_r.sample(range(1, 46), 6)) + out.append((i, nums)) + return out + + +def test_grade_tickets_histogram_and_prizes(): + winning6 = [1, 2, 3, 4, 5, 6] + bonus = 7 + tickets = [ + [1, 2, 3, 4, 5, 6], # 6일치 = 1등 + [1, 2, 3, 4, 5, 7], # 5일치 + 보너스 = 2등 + [1, 2, 3, 4, 5, 8], # 5일치 = 3등 + [1, 2, 3, 4, 9, 10], # 4일치 = 4등 + [1, 2, 3, 11, 12, 13], # 3일치 = 5등 + [40, 41, 42, 43, 44, 45], # 0일치 + ] + r = bt.grade_tickets(tickets, winning6, bonus) + assert r["m6"] == 1 + assert r["m5"] == 2 # 5일치 총 2장(보너스 포함) + assert r["bonus_hits"] == 1 # 그 중 보너스 1장 + assert r["m4"] == 1 + assert r["m3"] == 1 + assert r["best_match"] == 6 + # 등수 매핑 헬퍼 + prizes = bt.prize_counts(r) + assert prizes == {"1st": 1, "2nd": 1, "3rd": 1, "4th": 1, "5th": 1} + + +def test_purchase_tickets_distinct_and_count(): + draws = _toy_draws() + cache = build_analysis_cache(draws) + nw = build_number_weights(cache) + pool = bt.generate_pool(cache, nw, n=2000, seed=7) + W = [0.25, 0.30, 0.20, 0.15, 0.10] + bought = bt.purchase_tickets(pool, cache, W, k=50) + assert len(bought) == 50 + assert len({tuple(t) for t in bought}) == 50 # distinct + # W로 랭킹된 상위 k → 평균 점수가 풀 전체 평균 이상이어야 + avg_bought = sum(score_combination(t, cache, W)["score_total"] for t in bought) / 50 + avg_pool = sum(score_combination(t, cache, W)["score_total"] for t in pool) / len(pool) + assert avg_bought >= avg_pool + + +def test_random_null_and_coverage_distinct(): + rnd = bt.random_null_tickets(k=50, seed=3) + assert len(rnd) == 50 and len({tuple(t) for t in rnd}) == 50 + cov = bt.coverage_tickets(k=9, seed=3) # 9장 = 54슬롯 ≥ 45번호 전수 커버 가능 + flat = {n for t in cov for n in t} + assert len(cov) == 9 and len({tuple(t) for t in cov}) == 9 + assert len(flat) >= 40 # 커버리지 전략은 번호를 넓게 퍼뜨림 + + +def test_point_in_time_excludes_target_draw(): + draws = _toy_draws(50) # drw_no 1..50 + pit = bt.point_in_time_draws(draws, target_draw_no=30) + assert all(d < 30 for d, _ in pit) # 30 이상 제외 + assert max(d for d, _ in pit) == 29 + assert len(pit) == 29 + + +def test_calibrate_winner_scores_and_percentile(): + draws = _toy_draws(60) + winning6 = [3, 11, 19, 27, 35, 44] + res = bt.calibrate_winner_compute(draws, target_draw_no=60, + winning6=winning6, sample_m=500, seed=9) + assert set(res["scores"].keys()) >= {"score_total", "score_frequency", + "score_fingerprint", "score_gap", "score_cooccur", "score_diversity"} + assert 0.0 <= res["percentile"] <= 1.0 + assert res["cache_draws"] == 59 # 1..59 + + +def test_generate_pool_partial_fill(monkeypatch): + """weighted_sample_6이 항상 같은 조합만 반환하도록 패치 → cap에 먼저 걸려 len < n — 예외 없이 반환.""" + import random as _r + _r.seed(42) + tiny_draws = [(i, sorted(_r.sample(range(1, 46), 6))) for i in range(1, 10)] + cache = build_analysis_cache(tiny_draws) + nw = build_number_weights(cache) + + # weighted_sample_6을 항상 동일한 하나의 조합만 반환하도록 패치 + # → 두 번째 시도부터 seen에 막혀 n개를 채울 수 없고 cap=n*4 이후 종료 + import app.backtest as _bt_mod + monkeypatch.setattr(_bt_mod, "weighted_sample_6", lambda _w: [1, 2, 3, 4, 5, 6]) + + n = 50 + pool = bt.generate_pool(cache, nw, n=n, seed=0) + # 예외 없이 반환해야 하고, 결과는 n 미만이어야 하며 모두 distinct + assert isinstance(pool, list) + assert len(pool) < n + assert len({tuple(t) for t in pool}) == len(pool) diff --git a/lotto/tests/test_backtest_api.py b/lotto/tests/test_backtest_api.py new file mode 100644 index 0000000..459d54e --- /dev/null +++ b/lotto/tests/test_backtest_api.py @@ -0,0 +1,75 @@ +import os, sys, tempfile, random as _r + +# _shared lives in web-backend/_shared; add the parent dir so it can be found +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) + +from fastapi.testclient import TestClient + +def _client(monkeypatch): + tmp = tempfile.mkdtemp() + from app import db + monkeypatch.setattr(db, "DB_PATH", os.path.join(tmp, "lotto.db")) + db.init_db() + from app.main import app + return TestClient(app), db + + +def _seed_draws(db, n=40): + rows = [] + _r.seed(2) + for i in range(1, n + 1): + s = sorted(_r.sample(range(1, 46), 6)) + rows.append({"drw_no": i, "drw_date": f"2020-01-{(i % 28) + 1:02d}", + "n1": s[0], "n2": s[1], "n3": s[2], "n4": s[3], + "n5": s[4], "n6": s[5], "bonus": ((s[5] % 45) + 1)}) + db.upsert_many_draws(rows) + + +def test_backtest_endpoints(monkeypatch): + client, db = _client(monkeypatch) + r = client.get("/api/lotto/backtest/track-record") + assert r.status_code == 200 + assert "by_strategy" in r.json() + r2 = client.get("/api/lotto/backtest/calibration?weeks=4") + assert r2.status_code == 200 + assert isinstance(r2.json().get("history"), list) + + +def test_track_record_with_data(monkeypatch): + """seed 40 draws + forward run → track-record contains random_null.""" + client, db_mod = _client(monkeypatch) + _seed_draws(db_mod, 40) + from app import backtest as bt + bt.run_forward_purchase(40, k=20, pool_n=500, sample_seed=5) + r = client.get("/api/lotto/backtest/track-record") + assert r.status_code == 200 + body = r.json() + assert "by_strategy" in body + assert "random_null" in body["by_strategy"] + + +def test_review_known_and_unknown(monkeypatch): + """Known draw with calibration → 200 + non-null winner_analysis; unknown → 404.""" + client, db_mod = _client(monkeypatch) + _seed_draws(db_mod, 40) + from app import backtest as bt + bt.run_forward_purchase(40, k=20, pool_n=500, sample_seed=5) + bt.calibrate_winner(40, sample_m=200) + + r = client.get("/api/lotto/backtest/review/40") + assert r.status_code == 200 + body = r.json() + assert body["winner_analysis"] is not None + assert "score_total" in body["winner_analysis"] + + r2 = client.get("/api/lotto/backtest/review/99999") + assert r2.status_code == 404 + + +def test_calibration_weeks_bounds(monkeypatch): + """weeks=0 and weeks=521 should be rejected with 422.""" + client, _ = _client(monkeypatch) + r0 = client.get("/api/lotto/backtest/calibration?weeks=0") + assert r0.status_code == 422 + r521 = client.get("/api/lotto/backtest/calibration?weeks=521") + assert r521.status_code == 422 diff --git a/lotto/tests/test_backtest_db.py b/lotto/tests/test_backtest_db.py new file mode 100644 index 0000000..f8876bc --- /dev/null +++ b/lotto/tests/test_backtest_db.py @@ -0,0 +1,320 @@ +import os, tempfile + +def _fresh_db(monkeypatch): + tmp = tempfile.mkdtemp() + path = os.path.join(tmp, "lotto.db") + from app import db + monkeypatch.setattr(db, "DB_PATH", path) + db.init_db() + return db + +def test_backtest_tables_exist(monkeypatch): + db = _fresh_db(monkeypatch) + with db._conn() as conn: + tables = {r["name"] for r in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'").fetchall()} + assert "backtest_runs" in tables + assert "winner_calibration" in tables + +def test_backtest_runs_unique(monkeypatch): + db = _fresh_db(monkeypatch) + db.save_backtest_run(draw_no=100, strategy="random_null", weight_label="-", + weight_json=None, trial_id=None, n_tickets=10, + hist={"m3":1,"m4":0,"m5":0,"m6":0,"bonus_hits":0}, + best_match=3, avg_meta_score=0.5) + db.save_backtest_run(draw_no=100, strategy="random_null", weight_label="-", + weight_json=None, trial_id=None, n_tickets=10, + hist={"m3":2,"m4":0,"m5":0,"m6":0,"bonus_hits":0}, + best_match=3, avg_meta_score=0.6) # 멱등 upsert + rows = db.get_backtest_runs(draw_no=100) + assert len(rows) == 1 + assert rows[0]["m3"] == 2 # 마지막 값으로 갱신 + + +_SCORES = { + "score_total": 1.23, + "score_frequency": 0.30, + "score_fingerprint": 0.25, + "score_gap": 0.20, + "score_cooccur": 0.28, + "score_diversity": 0.20, +} + + +def test_winner_calibration_upsert(monkeypatch): + """save_winner_calibration 두 번 호출 시 upsert — 행 1개, 값은 마지막 것.""" + db = _fresh_db(monkeypatch) + winning = [3, 7, 15, 22, 33, 41] + db.save_winner_calibration(draw_no=200, winning=winning, + scores=_SCORES, percentile=75.0, + my_pick_avg=0.9, cache_draws=100) + # 두 번째 저장 — percentile, my_pick_avg 업데이트 + scores2 = {**_SCORES, "score_total": 2.00} + db.save_winner_calibration(draw_no=200, winning=winning, + scores=scores2, percentile=80.0, + my_pick_avg=1.1, cache_draws=110) + row = db.get_winner_calibration(200) + assert row is not None + # 행이 1개만 존재하는지 확인 + with db._conn() as conn: + cnt = conn.execute( + "SELECT COUNT(*) AS c FROM winner_calibration WHERE draw_no=200" + ).fetchone()["c"] + assert cnt == 1 + assert row["percentile"] == 80.0 + assert row["score_total"] == 2.00 + + +def _seed_draws(db, n=40): + rows = [] + import random as _r; _r.seed(2) + for i in range(1, n + 1): + s = sorted(_r.sample(range(1, 46), 6)) + rows.append({"drw_no": i, "drw_date": f"2020-01-{(i%28)+1:02d}", + "n1": s[0], "n2": s[1], "n3": s[2], "n4": s[3], + "n5": s[4], "n6": s[5], "bonus": ((s[5] % 45) + 1)}) + db.upsert_many_draws(rows) + +def test_backfill_calibration_idempotent(monkeypatch): + db = _fresh_db(monkeypatch) + _seed_draws(db, 40) + from app import backtest as bt + r1 = bt.backfill_calibration(batch=15, sample_m=200) + # 첫 회차는 point-in-time 데이터가 빈약 → min_history 이후만 처리 + done1 = len(db.get_calibrated_draw_nos()) + assert done1 > 0 + r2 = bt.backfill_calibration(batch=100, sample_m=200) # 나머지 + done2 = len(db.get_calibrated_draw_nos()) + assert done2 >= done1 + r3 = bt.backfill_calibration(batch=100, sample_m=200) # 재실행 → 추가 0 + assert r3["calibrated"] == 0 + + +def test_run_forward_purchase_persists_all_strategies(monkeypatch): + db = _fresh_db(monkeypatch) + _seed_draws(db, 40) + from app import backtest as bt + # 작은 규모로 빠르게 + res = bt.run_forward_purchase(draw_no=40, k=20, pool_n=500, sample_seed=5) + assert res["ok"] is True + rows = db.get_backtest_runs(draw_no=40) + strategies = {r["strategy"] for r in rows} + assert "random_null" in strategies + assert "coverage" in strategies + assert "engine_w" in strategies # base 가중치로 최소 1건 + for r in rows: + assert r["n_tickets"] == 20 + + +def test_calibrate_winner_no_draw(monkeypatch): + """DB에 없는 회차 번호 → ok=False, reason='no_draw'.""" + db = _fresh_db(monkeypatch) + _seed_draws(db, 40) + from app import backtest as bt + r = bt.calibrate_winner(99999) + assert r["ok"] is False + assert r["reason"] == "no_draw" + + +def test_calibrate_winner_insufficient_history(monkeypatch): + """point-in-time 이력이 MIN_HISTORY(30) 미만인 회차 → reason='insufficient_history'. + draw_no=20이면 PIT 이력이 19개(draws 1~19)로 30 미만.""" + db = _fresh_db(monkeypatch) + _seed_draws(db, 40) + from app import backtest as bt + r = bt.calibrate_winner(20) + assert r["ok"] is False + assert r["reason"] == "insufficient_history" + + +def test_run_forward_purchase_with_trials(monkeypatch): + """그 주 weight_trials가 존재하면 engine_w 행의 weight_label이 'w0'..'w5' 형식이어야 한다.""" + db = _fresh_db(monkeypatch) + _seed_draws(db, 40) + # draw 40: drw_date='2020-01-13' → week_start='2020-01-13' + from datetime import date, timedelta + draw_date = date.fromisoformat("2020-01-13") + ws = (draw_date - timedelta(days=draw_date.weekday())).isoformat() + # 해당 주에 trial 2개 심기 (day_of_week 0, 1) + db.save_weight_trial(ws, 0, [0.1, 0.3, 0.2, 0.2, 0.2], "perturb") + db.save_weight_trial(ws, 1, [0.25, 0.25, 0.25, 0.15, 0.1], "perturb") + from app import backtest as bt + res = bt.run_forward_purchase(draw_no=40, k=20, pool_n=500, sample_seed=5) + assert res["ok"] is True + rows = db.get_backtest_runs(draw_no=40) + engine_w_labels = {r["weight_label"] for r in rows if r["strategy"] == "engine_w"} + # trials가 있으므로 'base'가 아닌 'w0', 'w1' 형식이어야 한다 + assert "base" not in engine_w_labels + assert any(lbl.startswith("w") for lbl in engine_w_labels) + + +def test_run_forward_purchase_idempotent(monkeypatch): + """run_forward_purchase 두 번 호출 시 upsert — 행 수 변화 없음.""" + db = _fresh_db(monkeypatch) + _seed_draws(db, 40) + from app import backtest as bt + bt.run_forward_purchase(draw_no=40, k=20, pool_n=500, sample_seed=5) + count_after_first = len(db.get_backtest_runs(draw_no=40)) + bt.run_forward_purchase(draw_no=40, k=20, pool_n=500, sample_seed=5) + count_after_second = len(db.get_backtest_runs(draw_no=40)) + assert count_after_second == count_after_first + + +def test_get_calibrated_draw_nos(monkeypatch): + """저장된 draw_no 집합이 get_calibrated_draw_nos에 포함되어야 한다.""" + db = _fresh_db(monkeypatch) + winning = [1, 2, 3, 4, 5, 6] + for draw_no in (301, 302, 303): + db.save_winner_calibration(draw_no=draw_no, winning=winning, + scores=_SCORES, percentile=50.0, + my_pick_avg=0.5, cache_draws=50) + nos = db.get_calibrated_draw_nos() + assert isinstance(nos, set) + assert {301, 302, 303}.issubset(nos) + + +def test_track_record_and_review_payload(monkeypatch): + db = _fresh_db(monkeypatch) + _seed_draws(db, 40) + from app import backtest as bt + bt.run_forward_purchase(draw_no=40, k=20, pool_n=500, sample_seed=5) + bt.calibrate_winner(40, sample_m=200) + + tr = bt.track_record() + assert "random_null" in tr["by_strategy"] + # 이제 random_null은 N_NULL_TRIALS=6 행이므로 6*20=120장 + assert tr["by_strategy"]["random_null"]["n_tickets"] >= 20 + + payload = bt.build_review_payload(40) + assert payload["draw_no"] == 40 + assert "winner_analysis" in payload # 당첨조합 5분석치 + assert "forward" in payload # 이번 회차 전략별 성적 + assert "calibration_trend" in payload + assert payload["winner_analysis"] is not None + assert "score_total" in payload["winner_analysis"] + + +def test_run_forward_purchase_random_null_count(monkeypatch): + """run_forward_purchase는 random_null을 N_NULL_TRIALS=6개 저장해야 한다.""" + db = _fresh_db(monkeypatch) + _seed_draws(db, 40) + from app import backtest as bt + res = bt.run_forward_purchase(draw_no=40, k=20, pool_n=500, sample_seed=7) + assert res["ok"] is True + rows = db.get_backtest_runs(draw_no=40) + null_rows = [r for r in rows if r["strategy"] == "random_null"] + assert len(null_rows) == bt.N_NULL_TRIALS # 6개 + null_labels = {r["weight_label"] for r in null_rows} + assert null_labels == {f"r{i}" for i in range(bt.N_NULL_TRIALS)} + for r in null_rows: + assert r["n_tickets"] == 20 + + +def test_evaluate_weekly_gated_keeps_base_unchanged(monkeypatch): + """Fix 5 통합 테스트 (end-to-end gated path). + + 접근: DB에 draws, weight_trials, auto_picks, backtest_runs, base_history를 직접 심어 + evaluate_weekly()의 gated 분기가 base를 바꾸지 않음을 검증한다. + + gated 조건: engine_w 최고 prize_score − random_best < LIFT_EPSILON(10.0). + engine_best=5, random_best=20 → lift=-15 → gated. + + evaluate_weekly 내부 흐름: + - get_weekly_trials(week_start) : _today_kst() 기준 week_start 사용 + - get_latest_draw() : draws 테이블에서 max(drw_no) 반환 + 두 참조가 같은 날짜 기준이어야 하므로 _today_kst를 monkeypatch로 고정하고 + draws의 최신 회차 날짜(drw_date)를 해당 주의 날짜로 맞춘다. + """ + import json as _json + from datetime import date, timedelta, datetime as _dt, timezone as _tz, timedelta as _td + + db = _fresh_db(monkeypatch) + + # KST 오늘 날짜 — evaluate_weekly가 이 날짜를 기준으로 week_start 계산 + KST = _tz(_td(hours=9)) + today_kst = _dt.now(KST).date() + from app import weight_evolver as we + week_start = we.get_week_start(today_kst) + + # 1) draws 심기 — 최신 회차의 drw_date를 week_start 주 안의 날짜로 맞춤 + import random as _r; _r.seed(99) + rows = [] + for i in range(1, 41): + s = sorted(_r.sample(range(1, 46), 6)) + # 마지막 회차(40)는 오늘 날짜 사용 (week_start 주 내) + if i == 40: + drw_date = today_kst.isoformat() + else: + drw_date = f"2020-01-{(i % 28) + 1:02d}" + rows.append({ + "drw_no": i, "drw_date": drw_date, + "n1": s[0], "n2": s[1], "n3": s[2], + "n4": s[3], "n5": s[4], "n6": s[5], + "bonus": (s[5] % 45) + 1, + }) + db.upsert_many_draws(rows) + latest = db.get_latest_draw() + assert latest is not None + assert latest["drw_date"] == today_kst.isoformat() + + # 2) weight trial 1개 심기 (day_of_week=0, week_start=오늘 주) + trial_w = [0.2, 0.2, 0.2, 0.2, 0.2] + db.save_weight_trial(week_start, 0, trial_w, "perturb") + trial_rows = db.get_weekly_trials(week_start) + assert len(trial_rows) == 1 + trial_id = trial_rows[0]["id"] + + # 3) auto_picks 1개 심기 (winning 번호와 2개 일치 → max_correct=2) + winning6 = [latest["n1"], latest["n2"], latest["n3"], + latest["n4"], latest["n5"], latest["n6"]] + pick = winning6[:2] + [40, 41, 42, 43] + db.save_auto_pick(trial_id, 1, pick, meta_score=0.5) + + # 4) backtest_runs: engine_w prize_score=5, random_null 6개 prize_score=20 (gated 확실) + LOW_HIST = {"m3": 5, "m4": 0, "m5": 0, "m6": 0, "bonus_hits": 0} # prize=5 + HIGH_HIST = {"m3": 20, "m4": 0, "m5": 0, "m6": 0, "bonus_hits": 0} # prize=20 + draw_no = latest["drw_no"] + db.save_backtest_run( + draw_no=draw_no, strategy="engine_w", weight_label="w0", + weight_json=_json.dumps(trial_w), trial_id=trial_id, n_tickets=20, + hist=LOW_HIST, best_match=2, avg_meta_score=0.5, + ) + from app import backtest as bt + for i in range(bt.N_NULL_TRIALS): + db.save_backtest_run( + draw_no=draw_no, strategy="random_null", weight_label=f"r{i}", + weight_json=None, trial_id=None, n_tickets=20, + hist=HIGH_HIST, best_match=3, avg_meta_score=0.5, + ) + + # 5) current base 저장 (이전 주 월요일 effective_from) + base_w = [0.2, 0.2, 0.2, 0.2, 0.2] + prev_monday = (today_kst - timedelta(weeks=1, days=today_kst.weekday())).isoformat() + db.save_base_history( + effective_from=prev_monday, + weight=base_w, + source_trial_id=None, + update_reason="cold_start", + winner_score=None, + winner_max_correct=None, + ) + assert db.get_current_base() == base_w + + # 6) evaluate_weekly 호출 — _today_kst()를 monkeypatch로 오늘 날짜 고정 + monkeypatch.setattr(we, "_today_kst", lambda: today_kst) + + result = we.evaluate_weekly() + + assert result.get("ok") is True, f"evaluate_weekly 실패: {result}" + + # gated path 검증 + update_reason = result.get("update_reason", "") + assert update_reason in ("unchanged_gated", "idempotent_skip"), ( + f"gated여야 하는데 reason='{update_reason}' — 게이팅 로직 깨짐" + ) + + # base가 바뀌지 않았는지 검증 + new_base = result.get("new_base") + assert new_base == base_w, ( + f"gated인데 base가 변경됨: {new_base} != {base_w}" + ) diff --git a/lotto/tests/test_weight_evolver.py b/lotto/tests/test_weight_evolver.py index c7d9d0d..45ba6f8 100644 --- a/lotto/tests/test_weight_evolver.py +++ b/lotto/tests/test_weight_evolver.py @@ -120,3 +120,79 @@ def test_decide_base_update_cold_start_returns_default(): ) assert new_base == winner_W assert reason == "winner_4plus" + + +def test_select_winner_by_lift_gating(): + # engine_w 3개 + random_null 기준. lift = engine 등수점수 − random 등수점수 + per_w = [ + {"trial_id": 1, "day_of_week": 0, "weight": [0.2]*5, "prize_score": 5.0}, + {"trial_id": 2, "day_of_week": 1, "weight": [0.3,0.2,0.2,0.2,0.1], "prize_score": 9.0}, + {"trial_id": 3, "day_of_week": 2, "weight": [0.1,0.3,0.2,0.2,0.2], "prize_score": 4.0}, + ] + # random baseline이 8.0이면 lift는 -3, +1, -4 → 최대 lift(+1) < ε(2) → 게이팅 + winner = we.select_winner_by_lift(per_w, random_score=8.0, epsilon=2.0) + assert winner["gated"] is True # 최대 lift(+1) < ε(2) → 게이팅 + winner2 = we.select_winner_by_lift(per_w, random_score=3.0, epsilon=2.0) + assert winner2["gated"] is False + assert winner2["trial_id"] == 2 # prize 9 → lift +6 + + +def test_prize_score_from_hist(): + # 등수 가중치: 1등 매우 큼, 하위는 작게 + s = we.prize_score_from_hist({"m3": 10, "m4": 2, "m5": 0, "m6": 0, "bonus_hits": 0}) + s_big = we.prize_score_from_hist({"m3": 0, "m4": 0, "m5": 0, "m6": 1, "bonus_hits": 0}) + assert s_big > s # 1등 1장이 5등 다수보다 큼 + + +def test_select_winner_by_lift_preserves_all_keys(): + """select_winner_by_lift는 per_w 항목의 모든 키를 보존해야 한다. + best_match, weight_label 등 identity 필드가 누락되면 evaluate_weekly가 깨진다.""" + per_w = [ + { + "trial_id": 10, + "weight_label": "w0", + "weight": [0.2] * 5, + "prize_score": 3.0, + "best_match": 3, + }, + { + "trial_id": 11, + "weight_label": "w1", + "weight": [0.3, 0.2, 0.2, 0.2, 0.1], + "prize_score": 20.0, + "best_match": 4, + }, + ] + result = we.select_winner_by_lift(per_w, random_score=5.0, epsilon=2.0) + assert result["gated"] is False + assert result["trial_id"] == 11 + assert result["weight_label"] == "w1" # identity 키 보존 + assert result["best_match"] == 4 # best_match 키 보존 + assert "lift" in result # lift 추가됨 + assert result["lift"] == pytest.approx(15.0) + + +def test_gated_path_keeps_base_via_select_winner(): + """gated=True일 때 select_winner_by_lift의 반환값 검증. + evaluate_weekly 내의 gated 분기가 올바른 값에 의존함을 확인한다.""" + per_w = [ + {"trial_id": 1, "weight_label": "w0", "weight": [0.2]*5, + "prize_score": 5.0, "best_match": 2}, + {"trial_id": 2, "weight_label": "w1", "weight": [0.3,0.2,0.2,0.2,0.1], + "prize_score": 7.0, "best_match": 3}, + ] + # random_best=8.0 → 최대 engine lift=7-8=-1 → gated + result = we.select_winner_by_lift(per_w, random_score=8.0, epsilon=we.LIFT_EPSILON) + assert result["gated"] is True + assert result["lift"] < 0 + + # decide_base_update를 통해 gated가 unchanged를 유도하는지 확인 + # (gated override가 없더라도, 현재 LIFT_EPSILON=10.0 하에서 lift<0이면 항상 gated) + current = [0.2, 0.2, 0.2, 0.2, 0.2] + # gated이면 evaluate_weekly가 current_base를 그대로 유지해야 함 + # 여기서는 override 로직을 직접 재현해 검증한다 + gated = result["gated"] + new_base_override = list(current) if gated else None + reason_override = "unchanged_gated" if gated else "should_not_reach" + assert new_base_override == current + assert reason_override == "unchanged_gated"