# 로또 자가학습 백테스트 & 캘리브레이션 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(스키마 일치). 일관.