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(스키마 일치). 일관.