web-page-backend/docs/superpowers/plans/2026-05-31-lotto-self-learning-backtest.md

로또 자가학습 백테스트 & 캘리브레이션 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 <module>, 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:

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 인덱스 생성 직후에 삽입:

        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 끝부분에:

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

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:

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:

"""로또 자가학습 백테스트 — 순수 연산 (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

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에 추가:

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에 추가:

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

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: 실패 테스트

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에 추가:

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

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: 실패 테스트

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에 추가:

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

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에 추가:

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 패턴 동일):

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

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에 추가:

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에 추가:

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

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에 추가:

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에 추가:

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

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:

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:

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) 다음 줄에:

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

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 실행:

    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

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에 추가:

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에 추가:

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

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에 추가:

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 추가:

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 호출 전에 삽입:

    # 자가학습 강화: 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

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 섹션에 추가:

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

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 디렉토리 생성):

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에 추가:

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"🔍 <b>로또 #{draw_no} 일요 회고</b>", ""]
    if wa:
        lines.append(f"이번 당첨조합 분석치: <b>{wa.get('score_total',0):.2f}</b> "
                     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("📊 <b>이번 회차 가상구매 성적</b>")
    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

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에:

    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 분기 다음):

        if action == "sunday_review":
            return await self.run_sunday_review()

• Step 3: scheduler cron 등록 — agent-office/app/scheduler.py에 래퍼 + 등록:

async def _run_lotto_sunday_review():
    agent = AGENT_REGISTRY.get("lotto")
    if agent:
        await agent.run_sunday_review()

그리고 init_scheduler() 안 lotto cron 그룹에:

    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

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 헬퍼 근처에:

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)

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:

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 (
    <div className="lotto-evolver-card">
      <h3>이번 당첨조합 분석치 (무작위 상위 {pct})</h3>
      <ResponsiveContainer width="100%" height={220}>
        <RadarChart data={data}>
          <PolarGrid stroke="rgba(255,255,255,0.12)" />
          <PolarAngleAxis dataKey="k" tick={{ fill: '#cbd5e1', fontSize: 12 }} />
          <Radar dataKey="v" stroke="#60a5fa" fill="#60a5fa" fillOpacity={0.4} />
        </RadarChart>
      </ResponsiveContainer>
    </div>
  );
}

recharts는 기존 evolver UI(LineChart/Radar)에서 이미 사용 중 — import 경로 동일.

• Step 3: TrackRecordCard 작성 — web-ui/src/pages/lotto/components/TrackRecordCard.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 (
    <div className="lotto-evolver-card">
      <h3>누적 성적표 (전략당 5,000장/회차)</h3>
      <table className="lotto-evolver-table">
        <thead><tr><th>전략</th><th>장수</th><th>3등</th><th>4등</th><th>5등</th></tr></thead>
        <tbody>
          {order.filter((s) => byStrategy[s]).map((s) => {
            const a = byStrategy[s];
            return (
              <tr key={s}>
                <td>{label[s]}</td><td>{a.n_tickets.toLocaleString()}</td>
                <td>{a['3rd']}</td><td>{a['4th']}</td><td>{a['5th']}</td>
              </tr>
            );
          })}
        </tbody>
      </table>
      <p className="lotto-evolver-note">엔진이 무작위를 넘지 못하면 분석에 우위가 없다는 정직한 증거입니다.</p>
    </div>
  );
}

• Step 4: CalibrationChart 작성 — web-ui/src/pages/lotto/components/CalibrationChart.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 (
    <div className="lotto-evolver-card">
      <h3>당첨조합 캘리브레이션 추세</h3>
      <ResponsiveContainer width="100%" height={240}>
        <LineChart data={data}>
          <CartesianGrid stroke="rgba(255,255,255,0.08)" />
          <XAxis dataKey="draw" tick={{ fill: '#94a3b8', fontSize: 11 }} />
          <YAxis domain={[0, 1]} tick={{ fill: '#94a3b8', fontSize: 11 }} />
          <Tooltip contentStyle={{ background: '#0f172a', border: 'none' }} />
          <Line type="monotone" dataKey="score" stroke="#f59e0b" dot={false} name="당첨조합 분석치" />
          <Line type="monotone" dataKey="pct" stroke="#34d399" dot={false} name="무작위 percentile" />
        </LineChart>
      </ResponsiveContainer>
    </div>
  );
}

• Step 5: 자율학습 탭에 통합 — Step 1에서 찾은 탭 컨테이너에 3컴포넌트 추가 + 데이터 로드:

// 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 섹션 하단)
<WinnerAnalysisCard analysis={winner} />
<TrackRecordCard byStrategy={track?.by_strategy} />
<CalibrationChart history={calib} />

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)

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회 트리거:

# 회차가 많으므로 여러 번 호출 (멱등, 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(스키마 일치). 일관.