web-page-backend/docs/superpowers/plans/2026-05-22-lotto-weight-evolver.md

Lotto Weight Evolver 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: lotto-lab에 5종 시뮬 점수 가중치 자율 학습 루프 추가 — 주간 6 trials → 토요일 회고 → winner 기반 base 갱신 → 무한 반복

Architecture: lotto-lab 내부에 weight_evolver.py 신설 + analyzer.score_combination 시그니처 확장 + lotto.db 3 신규 테이블. agent-office는 토요일 22:15 텔레그램 리포트만 추가. v1 시그널은 W 변화로 자동 cascade.

Tech Stack: Python 3.12, FastAPI, APScheduler, SQLite, numpy, httpx, pytest

Spec: docs/superpowers/specs/2026-05-22-lotto-weight-evolver-design.md

---

Important Spec Adjustment Noted in Plan

The spec describes the existing analyzer.score_combination as "균등 합산". Actual code at lotto/app/analyzer.py:285-291 uses fixed weights [0.25, 0.30, 0.20, 0.15, 0.10] for [frequency, fingerprint, gap, cooccur, diversity]. The plan preserves this as default when weights=None, and uses dynamic W when weights=[...]. Cold start W_base = [0.2]*5 remains as spec, applied only when weight_base_history is empty.

---

File Structure

경로	작업	책임
lotto/app/weight_evolver.py	Create	순수 함수 (clamp/normalize/perturb/Dirichlet/score) + 진입점
lotto/app/db.py	Modify	3개 신규 테이블 + CRUD
lotto/app/analyzer.py	Modify	score_combination(numbers, cache, weights=None) 시그니처 확장
lotto/app/generator.py	Modify	시뮬 cron이 active W를 읽어 전달
lotto/app/main.py	Modify	cron 3종 + API 5종 등록
lotto/tests/test_weight_evolver.py	Create	순수 함수 + base update rule + score 테스트
lotto/tests/test_analyzer_weighted.py	Create	가중 합산 정확성 테스트
agent-office/app/service_proxy.py	Modify	lotto_evolver_status() helper
agent-office/app/notifiers/telegram_lotto.py	Modify	send_evolution_report + _format_evolution_report
agent-office/app/agents/lotto.py	Modify	run_weekly_evolution_report
agent-office/app/scheduler.py	Modify	cron 1종 추가
agent-office/tests/test_lotto_evolution_format.py	Create	텔레그램 폼 테스트
web-backend/CLAUDE.md	Modify	lotto-lab 섹션 갱신

---

Phase 1 — DB + 순수 함수 + 테스트

Task 1: weight_evolver.py 순수 함수 테스트 (TDD red)

Files:

• Create: lotto/tests/test_weight_evolver.py

• Step 1: Write failing tests

# lotto/tests/test_weight_evolver.py
import json
import math
import pytest

from app import weight_evolver as we


def test_clamp_and_normalize_min_floor():
    """모든 값이 0.05 이상이 되도록 보장 + 합=1.0."""
    W = we.clamp_and_normalize([0.01, 0.6, 0.2, 0.1, 0.09])
    assert all(w >= 0.05 - 1e-9 for w in W)
    assert abs(sum(W) - 1.0) < 1e-9


def test_clamp_and_normalize_negative_becomes_floor():
    W = we.clamp_and_normalize([-0.1, 0.5, 0.3, 0.2, 0.1])
    assert W[0] >= 0.05 - 1e-9
    assert abs(sum(W) - 1.0) < 1e-9


def test_perturbation_changes_around_base():
    """σ=0.05 정규분포 perturbation 후 정규화 — 각 값이 합리적 범위 안."""
    base = [0.2, 0.2, 0.2, 0.2, 0.2]
    W = we.perturb_weights(base, sigma=0.05, seed=42)
    assert abs(sum(W) - 1.0) < 1e-9
    assert all(w >= 0.05 - 1e-9 for w in W)


def test_dirichlet_random_distribution():
    """Dirichlet α=2 — 5종 비음수 합=1."""
    W = we.dirichlet_weights(alpha=2.0, seed=42)
    assert abs(sum(W) - 1.0) < 1e-9
    assert all(0.05 - 1e-9 <= w <= 1.0 for w in W)


def test_generate_weekly_candidates_count():
    """6개 후보 생성 — 4 perturb + 2 dirichlet."""
    base = [0.2, 0.2, 0.2, 0.2, 0.2]
    trials = we.generate_weekly_candidates(base, seed=42)
    assert len(trials) == 6
    sources = [t["source"] for t in trials]
    assert sources.count("perturb") == 4
    assert sources.count("dirichlet") == 2
    # day_of_week 0..5 모두 존재
    days = sorted(t["day_of_week"] for t in trials)
    assert days == [0, 1, 2, 3, 4, 5]


def test_calc_pick_score_six_match():
    """6개 모두 일치 → 1등 → base=1.0 + bonus 1.0 = 2.0."""
    score = we.calc_pick_score([1, 2, 3, 4, 5, 6], [1, 2, 3, 4, 5, 6])
    assert score == pytest.approx(2.0)


def test_calc_pick_score_four_match():
    """4개 일치 → 4등 → base=4/6 + bonus 0.3."""
    score = we.calc_pick_score([1, 2, 3, 4, 7, 8], [1, 2, 3, 4, 5, 6])
    assert score == pytest.approx(4/6 + 0.3)


def test_calc_pick_score_three_match():
    """3개 일치 → 5등 → base=3/6 + bonus 0.1."""
    score = we.calc_pick_score([1, 2, 3, 7, 8, 9], [1, 2, 3, 4, 5, 6])
    assert score == pytest.approx(3/6 + 0.1)


def test_calc_pick_score_two_match_no_bonus():
    """2개 일치 → 미당첨 → base=2/6 + bonus 0."""
    score = we.calc_pick_score([1, 2, 7, 8, 9, 10], [1, 2, 3, 4, 5, 6])
    assert score == pytest.approx(2/6)


def test_decide_base_update_winner_4plus_replaces():
    """winner_max_correct ≥ 4 → 교체."""
    current = [0.2, 0.2, 0.2, 0.2, 0.2]
    winner_W = [0.1, 0.3, 0.2, 0.3, 0.1]
    new_base, reason = we.decide_base_update(
        winner_max_correct=4,
        winner_W=winner_W,
        current_base=current,
    )
    assert new_base == winner_W
    assert reason == "winner_4plus"


def test_decide_base_update_winner_3_ema_blend():
    """winner_max_correct = 3 → 0.3*winner + 0.7*current."""
    current = [0.2, 0.2, 0.2, 0.2, 0.2]
    winner_W = [0.1, 0.3, 0.2, 0.3, 0.1]
    new_base, reason = we.decide_base_update(
        winner_max_correct=3,
        winner_W=winner_W,
        current_base=current,
    )
    expected = [0.3 * w + 0.7 * c for w, c in zip(winner_W, current)]
    assert all(abs(a - b) < 1e-9 for a, b in zip(new_base, expected))
    assert reason == "ema_blend"


def test_decide_base_update_winner_lt3_unchanged():
    """winner_max_correct ≤ 2 → 직전 base 유지."""
    current = [0.2, 0.2, 0.2, 0.2, 0.2]
    winner_W = [0.1, 0.3, 0.2, 0.3, 0.1]
    new_base, reason = we.decide_base_update(
        winner_max_correct=2,
        winner_W=winner_W,
        current_base=current,
    )
    assert new_base == current
    assert reason == "unchanged"


def test_decide_base_update_cold_start_returns_default():
    """current_base=None (첫 회) → 균등 default 반환."""
    winner_W = [0.1, 0.3, 0.2, 0.3, 0.1]
    new_base, reason = we.decide_base_update(
        winner_max_correct=4,
        winner_W=winner_W,
        current_base=None,
    )
    # current_base가 None이어도 winner_max_correct>=4면 winner 교체 가능
    assert new_base == winner_W
    assert reason == "winner_4plus"

• Step 2: Run test to verify it fails

Run: cd lotto && pytest tests/test_weight_evolver.py -v Expected: FAIL with ModuleNotFoundError: No module named 'app.weight_evolver'

• Step 3: Commit

git add lotto/tests/test_weight_evolver.py
git commit -m "test(weight-evolver): 순수 함수 + base update rule 단위 테스트"

---

Task 2: weight_evolver.py 순수 함수 구현

Files:

• Create: lotto/app/weight_evolver.py

• Step 1: Implement pure functions

# lotto/app/weight_evolver.py
"""5종 시뮬 점수 가중치 자율 학습 루프.

순수 함수 (clamp/perturb/Dirichlet/score/base-rule) + DB 진입점은 별도 섹션.
"""
from __future__ import annotations
import math
import random
from typing import Any, Dict, List, Optional, Tuple

import numpy as np

MIN_WEIGHT = 0.05
N_METRICS = 5
DEFAULT_UNIFORM = [0.2] * N_METRICS  # cold start

RANK_BY_CORRECT = {6: 1, 5: 3, 4: 4, 3: 5}
RANK_BONUS = {1: 1.0, 2: 0.8, 3: 0.6, 4: 0.3, 5: 0.1}


def clamp_and_normalize(W: List[float], min_w: float = MIN_WEIGHT) -> List[float]:
    """각 값 ≥ min_w + 합=1.0. 보장 안 되면 raise."""
    if len(W) != N_METRICS:
        raise ValueError(f"W must have {N_METRICS} elements")
    clamped = [max(min_w, float(w)) for w in W]
    total = sum(clamped)
    return [w / total for w in clamped]


def perturb_weights(
    base: List[float],
    sigma: float = 0.05,
    seed: Optional[int] = None,
) -> List[float]:
    """base에 정규분포 noise(σ) 추가 → clamp+normalize."""
    if seed is not None:
        np.random.seed(seed)
    noise = np.random.normal(0, sigma, size=N_METRICS)
    perturbed = [b + n for b, n in zip(base, noise)]
    return clamp_and_normalize(perturbed)


def dirichlet_weights(
    alpha: float = 2.0,
    seed: Optional[int] = None,
) -> List[float]:
    """Dirichlet(α, α, α, α, α) 샘플 → clamp+normalize."""
    if seed is not None:
        np.random.seed(seed)
    sample = np.random.dirichlet([alpha] * N_METRICS).tolist()
    return clamp_and_normalize(sample)


def generate_weekly_candidates(
    base: Optional[List[float]] = None,
    seed: Optional[int] = None,
) -> List[Dict[str, Any]]:
    """6개 후보 — 4 perturb + 2 dirichlet. day_of_week 0..5 매핑.

    Returns:
        [{"day_of_week": 0, "weight": [...], "source": "perturb"}, ...]
    """
    if base is None:
        base = DEFAULT_UNIFORM[:]
    if seed is not None:
        np.random.seed(seed)

    trials = []
    for i in range(4):
        # 각 trial에 다른 seed 변동 — np.random.normal 호출 시점에 자동
        trials.append({
            "day_of_week": i,
            "weight": perturb_weights(base, sigma=0.05),
            "source": "perturb",
        })
    for i in range(4, 6):
        trials.append({
            "day_of_week": i,
            "weight": dirichlet_weights(alpha=2.0),
            "source": "dirichlet",
        })
    return trials


def count_match(pick: List[int], winning: List[int]) -> int:
    """본번호 6개 일치 개수. 보너스 제외."""
    return len(set(pick) & set(winning[:6]))


def calc_pick_score(pick_numbers: List[int], winning_numbers: List[int]) -> float:
    """correct/6 + RANK_BONUS. 보너스 번호 미고려."""
    correct = count_match(pick_numbers, winning_numbers)
    base = correct / 6.0
    rank = RANK_BY_CORRECT.get(correct)
    bonus = RANK_BONUS.get(rank, 0) if rank else 0
    return base + bonus


def decide_base_update(
    winner_max_correct: int,
    winner_W: List[float],
    current_base: Optional[List[float]],
) -> Tuple[List[float], str]:
    """Hybrid base update rule.

    Returns:
        (new_base, reason) — reason ∈ {'winner_4plus','ema_blend','unchanged','cold_start'}
    """
    if winner_max_correct >= 4:
        return list(winner_W), "winner_4plus"
    if winner_max_correct == 3 and current_base is not None:
        blended = [0.3 * w + 0.7 * c for w, c in zip(winner_W, current_base)]
        return clamp_and_normalize(blended), "ema_blend"
    if current_base is None:
        return DEFAULT_UNIFORM[:], "cold_start"
    return list(current_base), "unchanged"

• Step 2: Run tests to verify they pass

Run: cd lotto && pytest tests/test_weight_evolver.py -v Expected: All 12 tests PASS

• Step 3: Commit

git add lotto/app/weight_evolver.py
git commit -m "feat(weight-evolver): 순수 함수 (clamp/perturb/Dirichlet/score/base-rule)"

---

Task 3: lotto.db 마이그레이션 — 3 신규 테이블 + CRUD

Files:

• Modify: lotto/app/db.py

• Step 1: Add 3 CREATE TABLE blocks

lotto/app/db.py의 init_db() 함수 마지막 (다른 CREATE TABLE 뒤, seed insert 전) 추가:

        conn.execute("""
            CREATE TABLE IF NOT EXISTS weight_trials (
                id            INTEGER PRIMARY KEY AUTOINCREMENT,
                week_start    TEXT NOT NULL,
                day_of_week   INTEGER NOT NULL,
                weight_json   TEXT NOT NULL,
                source        TEXT NOT NULL,
                base_at_gen   TEXT,
                created_at    TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
                UNIQUE(week_start, day_of_week)
            )
        """)
        conn.execute("""
            CREATE INDEX IF NOT EXISTS idx_wt_week
            ON weight_trials(week_start, day_of_week)
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS auto_picks (
                id            INTEGER PRIMARY KEY AUTOINCREMENT,
                trial_id      INTEGER NOT NULL REFERENCES weight_trials(id) ON DELETE CASCADE,
                pick_no       INTEGER NOT NULL,
                numbers       TEXT NOT NULL,
                meta_score    REAL,
                correct       INTEGER,
                rank          INTEGER,
                graded_at     TEXT,
                created_at    TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
                UNIQUE(trial_id, pick_no)
            )
        """)
        conn.execute("CREATE INDEX IF NOT EXISTS idx_ap_trial ON auto_picks(trial_id)")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_ap_graded ON auto_picks(graded_at)")
        conn.execute("""
            CREATE TABLE IF NOT EXISTS weight_base_history (
                id              INTEGER PRIMARY KEY AUTOINCREMENT,
                effective_from  TEXT NOT NULL,
                weight_json     TEXT NOT NULL,
                source_trial_id INTEGER REFERENCES weight_trials(id),
                update_reason   TEXT,
                winner_score    REAL,
                winner_max_correct INTEGER,
                created_at      TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
            )
        """)

• Step 2: Add CRUD helpers at end of db.py

# --- weight_trials / auto_picks / weight_base_history CRUD ---

def save_weight_trial(
    week_start: str,
    day_of_week: int,
    weight: List[float],
    source: str,
    base_at_gen: Optional[List[float]] = None,
) -> int:
    with _conn() as conn:
        cur = conn.execute(
            """
            INSERT INTO weight_trials (week_start, day_of_week, weight_json, source, base_at_gen)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(week_start, day_of_week) DO UPDATE SET
              weight_json = excluded.weight_json,
              source = excluded.source,
              base_at_gen = excluded.base_at_gen
            """,
            (week_start, day_of_week, json.dumps(weight),
             source, json.dumps(base_at_gen) if base_at_gen else None),
        )
        # ON CONFLICT 경로에서는 lastrowid가 0 → 별도 조회
        if cur.lastrowid:
            return cur.lastrowid
        row = conn.execute(
            "SELECT id FROM weight_trials WHERE week_start=? AND day_of_week=?",
            (week_start, day_of_week),
        ).fetchone()
        return int(row["id"])


def get_weight_trial(week_start: str, day_of_week: int) -> Optional[Dict[str, Any]]:
    with _conn() as conn:
        row = conn.execute(
            "SELECT * FROM weight_trials WHERE week_start=? AND day_of_week=?",
            (week_start, day_of_week),
        ).fetchone()
    if not row:
        return None
    d = dict(row)
    d["weight"] = json.loads(d.pop("weight_json"))
    if d.get("base_at_gen"):
        d["base_at_gen"] = json.loads(d["base_at_gen"])
    return d


def get_weekly_trials(week_start: str) -> List[Dict[str, Any]]:
    with _conn() as conn:
        rows = conn.execute(
            "SELECT * FROM weight_trials WHERE week_start=? ORDER BY day_of_week",
            (week_start,),
        ).fetchall()
    out = []
    for r in rows:
        d = dict(r)
        d["weight"] = json.loads(d.pop("weight_json"))
        if d.get("base_at_gen"):
            d["base_at_gen"] = json.loads(d["base_at_gen"])
        out.append(d)
    return out


def save_auto_pick(
    trial_id: int,
    pick_no: int,
    numbers: List[int],
    meta_score: Optional[float] = None,
) -> int:
    with _conn() as conn:
        cur = conn.execute(
            """
            INSERT OR REPLACE INTO auto_picks (trial_id, pick_no, numbers, meta_score)
            VALUES (?, ?, ?, ?)
            """,
            (trial_id, pick_no, json.dumps(sorted(numbers)), meta_score),
        )
        return cur.lastrowid


def get_auto_picks(trial_id: int) -> List[Dict[str, Any]]:
    with _conn() as conn:
        rows = conn.execute(
            "SELECT * FROM auto_picks WHERE trial_id=? ORDER BY pick_no",
            (trial_id,),
        ).fetchall()
    out = []
    for r in rows:
        d = dict(r)
        d["numbers"] = json.loads(d["numbers"])
        out.append(d)
    return out


def update_auto_pick_grade(pick_id: int, correct: int, rank: Optional[int]) -> None:
    with _conn() as conn:
        conn.execute(
            """
            UPDATE auto_picks
            SET correct=?, rank=?, graded_at=strftime('%Y-%m-%dT%H:%M:%fZ','now')
            WHERE id=?
            """,
            (correct, rank, pick_id),
        )


def get_current_base() -> Optional[List[float]]:
    """weight_base_history 최신 row의 weight. 없으면 None (cold start)."""
    with _conn() as conn:
        row = conn.execute(
            "SELECT weight_json FROM weight_base_history ORDER BY id DESC LIMIT 1",
        ).fetchone()
    if not row:
        return None
    return json.loads(row["weight_json"])


def save_base_history(
    effective_from: str,
    weight: List[float],
    source_trial_id: Optional[int],
    update_reason: str,
    winner_score: Optional[float],
    winner_max_correct: Optional[int],
) -> int:
    with _conn() as conn:
        cur = conn.execute(
            """
            INSERT INTO weight_base_history
              (effective_from, weight_json, source_trial_id, update_reason,
               winner_score, winner_max_correct)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (effective_from, json.dumps(weight), source_trial_id,
             update_reason, winner_score, winner_max_correct),
        )
        return cur.lastrowid


def get_base_history(limit: int = 12) -> List[Dict[str, Any]]:
    with _conn() as conn:
        rows = conn.execute(
            "SELECT * FROM weight_base_history ORDER BY id DESC LIMIT ?",
            (limit,),
        ).fetchall()
    out = []
    for r in rows:
        d = dict(r)
        d["weight"] = json.loads(d.pop("weight_json"))
        out.append(d)
    return out

Note: db.py already imports os, json, sqlite3, List, Dict, Optional, Any from typing. Verify if Optional is present at top; if not, add.

• Step 3: Smoke test

cd lotto && python -c "
import os
os.environ['LOTTO_DB_PATH'] = '/tmp/test_evolver.db' if os.name != 'nt' else 'test_evolver.db'
from app.db import init_db, save_weight_trial, get_weight_trial, save_base_history, get_current_base
init_db()
tid = save_weight_trial('2026-05-25', 0, [0.2,0.2,0.2,0.2,0.2], 'perturb', [0.2]*5)
print('trial id:', tid)
print('get:', get_weight_trial('2026-05-25', 0))
save_base_history('2026-05-25', [0.2]*5, None, 'cold_start', None, None)
print('current base:', get_current_base())
"

Expected: trial id 출력 + dict 출력 + current base [0.2, 0.2, 0.2, 0.2, 0.2]

• Step 4: Commit

git add lotto/app/db.py
git commit -m "feat(weight-evolver): lotto.db에 weight_trials/auto_picks/weight_base_history + CRUD"

---

Phase 2 — analyzer 확장 + active weight 적용

Task 4: analyzer.score_combination 시그니처 확장 + 테스트

Files:

• Create: lotto/tests/test_analyzer_weighted.py

• Modify: lotto/app/analyzer.py:173-... (score_combination 함수)

• Step 1: Write failing test

# lotto/tests/test_analyzer_weighted.py
import pytest
from app.analyzer import score_combination, build_analysis_cache


@pytest.fixture
def cache():
    # 최소 더미 cache — 실제 회차 데이터가 없어도 score_combination이 동작하도록
    # build_analysis_cache는 회차 list가 필요하므로 fake draws로
    fake_draws = [
        {"drw_no": 1, "drw_num1": 1, "drw_num2": 2, "drw_num3": 3,
         "drw_num4": 4, "drw_num5": 5, "drw_num6": 6, "bnus_no": 7,
         "drw_date": "2024-01-01"},
        {"drw_no": 2, "drw_num1": 7, "drw_num2": 8, "drw_num3": 9,
         "drw_num4": 10, "drw_num5": 11, "drw_num6": 12, "bnus_no": 13,
         "drw_date": "2024-01-08"},
    ]
    return build_analysis_cache(fake_draws)


def test_score_default_uses_fixed_weights(cache):
    """weights=None은 기존 fixed [0.25, 0.30, 0.20, 0.15, 0.10]과 동등."""
    s = score_combination([1, 2, 3, 4, 5, 6], cache)
    # 기존 score_total은 0~1.5 범위. 정확값은 입력 의존, 단지 키 있고 0이상.
    assert "score_total" in s
    assert 0.0 <= s["score_total"] <= 2.0
    # 5종 점수 모두 키 존재
    for k in ("score_frequency", "score_fingerprint", "score_gap",
              "score_cooccur", "score_diversity"):
        assert k in s


def test_score_with_custom_weights_sums_correctly(cache):
    """weights=[1,0,0,0,0]은 score_total == score_frequency."""
    s = score_combination([1, 2, 3, 4, 5, 6], cache, weights=[1.0, 0.0, 0.0, 0.0, 0.0])
    assert s["score_total"] == pytest.approx(s["score_frequency"], rel=1e-3)


def test_score_with_uniform_weights(cache):
    """weights=[0.2]*5는 단순 평균."""
    s = score_combination([1, 2, 3, 4, 5, 6], cache, weights=[0.2]*5)
    expected = 0.2 * (s["score_frequency"] + s["score_fingerprint"]
                      + s["score_gap"] + s["score_cooccur"] + s["score_diversity"])
    assert s["score_total"] == pytest.approx(expected, rel=1e-3)


def test_score_weights_wrong_length_raises(cache):
    with pytest.raises((ValueError, AssertionError)):
        score_combination([1, 2, 3, 4, 5, 6], cache, weights=[0.5, 0.5])

Run: cd lotto && pytest tests/test_analyzer_weighted.py -v Expected: FAIL — score_combination() got unexpected keyword 'weights'

• Step 2: Modify score_combination

lotto/app/analyzer.py 파일에서 함수 시그니처 + 합산 부분 (line 173, line 285-291) 수정:

# 시그니처 변경
def score_combination(
    numbers: List[int],
    cache: Dict[str, Any],
    weights: Optional[List[float]] = None,
) -> Dict[str, float]:
    """5종 점수 + score_total 계산.

    weights=None: 기존 fixed [0.25, 0.30, 0.20, 0.15, 0.10] 사용 (호환성)
    weights=[w_freq, w_finger, w_gap, w_cooccur, w_diversity]: 동적 가중치 사용
    """
    ...

Optional이 import 안 되어 있으면 추가: from typing import Any, Dict, List, Optional

함수 본문 끝의 score_total 합산 부분 (현재 line 285-291):

# 기존
score_total = (
    score_frequency   * 0.25
    + score_fingerprint * 0.30
    + score_gap         * 0.20
    + score_cooccur     * 0.15
    + score_diversity   * 0.10
)

다음으로 교체:

if weights is None:
    weights = [0.25, 0.30, 0.20, 0.15, 0.10]
if len(weights) != 5:
    raise ValueError("weights must have 5 elements")
score_total = (
    score_frequency   * weights[0]
    + score_fingerprint * weights[1]
    + score_gap         * weights[2]
    + score_cooccur     * weights[3]
    + score_diversity   * weights[4]
)

• Step 3: Run tests to verify pass

cd lotto && pytest tests/test_analyzer_weighted.py -v
cd lotto && pytest tests/ -v 2>&1 | tail -10

Expected: 4 new tests PASS + 기존 tests still pass (no regression)

• Step 4: Commit

git add lotto/app/analyzer.py lotto/tests/test_analyzer_weighted.py
git commit -m "feat(analyzer): score_combination에 weights 파라미터 추가 (None=기존 fixed)"

---

Task 5: weight_evolver.py에 get_active_weight() + DB 통합 진입점

Files:

• Modify: lotto/app/weight_evolver.py

• Step 1: Add DB-touching entry points at end of file

# ---------- DB-touching entry points ----------

from datetime import datetime, timedelta, timezone


KST = timezone(timedelta(hours=9))


def _db():
    from . import db as _db_mod
    return _db_mod


def _today_kst():
    return datetime.now(KST).date()


def get_week_start(d=None) -> str:
    """주어진 날짜의 월요일 ISO 'YYYY-MM-DD'."""
    if d is None:
        d = _today_kst()
    ws = d - timedelta(days=d.weekday())
    return ws.isoformat()


def get_active_weight() -> Optional[List[float]]:
    """오늘 적용 중인 W. 없으면 None (균등 폴백)."""
    today = _today_kst()
    week_start = get_week_start(today)
    dow = today.weekday()
    if dow == 6:
        dow = 5  # 일요일은 토요일 W 유지
    trial = _db().get_weight_trial(week_start, dow)
    if trial:
        return trial["weight"]
    return None


def generate_weekly_candidates_and_save(seed: Optional[int] = None) -> List[Dict[str, Any]]:
    """월요일 09:00 cron 진입점. 6 trials 생성 후 DB 저장."""
    db = _db()
    base = db.get_current_base()
    if base is None:
        base = DEFAULT_UNIFORM[:]
        db.save_base_history(
            effective_from=get_week_start(),
            weight=base,
            source_trial_id=None,
            update_reason="cold_start",
            winner_score=None,
            winner_max_correct=None,
        )

    candidates = generate_weekly_candidates(base, seed=seed)
    week_start = get_week_start()
    for c in candidates:
        db.save_weight_trial(
            week_start=week_start,
            day_of_week=c["day_of_week"],
            weight=c["weight"],
            source=c["source"],
            base_at_gen=base,
        )
    return candidates


def apply_today_and_pick(n: int = 5) -> Dict[str, Any]:
    """매일 09:00 cron 진입점. 오늘 W로 N=5 세트 추출 후 auto_picks 저장."""
    db = _db()
    from . import analyzer, recommender
    today = _today_kst()
    week_start = get_week_start(today)
    dow = min(today.weekday(), 5)  # 일요일은 토요일과 같이

    trial = db.get_weight_trial(week_start, dow)
    if trial is None:
        return {"ok": False, "reason": "no_trial_for_today"}

    W = trial["weight"]
    draws = db.get_all_draw_numbers()
    cache = analyzer.build_analysis_cache(draws)

    picks_saved = []
    for i in range(1, n + 1):
        # recommender.recommend_numbers() — 단순 추천. 5번 호출로 다양성 확보.
        try:
            r = recommender.recommend_numbers(draws)
            nums = r["numbers"]
            s = analyzer.score_combination(nums, cache, weights=W)
            pid = db.save_auto_pick(trial["id"], i, nums, meta_score=s["score_total"])
            picks_saved.append({"id": pid, "numbers": nums, "score": s["score_total"]})
        except Exception:
            continue

    return {
        "ok": True,
        "trial_id": trial["id"],
        "weight": W,
        "picks": picks_saved,
    }


def evaluate_weekly() -> Dict[str, Any]:
    """토 22:00 cron 진입점. 6일 trials × N picks 채점 + base 갱신."""
    db = _db()
    today = _today_kst()
    week_start = get_week_start(today)

    trials = db.get_weekly_trials(week_start)
    if not trials:
        return {"ok": False, "reason": "no_trials"}

    latest = db.get_latest_draw()
    if latest is None:
        return {"ok": False, "reason": "no_latest_draw"}
    winning = [
        latest["drw_num1"], latest["drw_num2"], latest["drw_num3"],
        latest["drw_num4"], latest["drw_num5"], latest["drw_num6"],
    ]

    per_day = []
    for trial in trials:
        picks = db.get_auto_picks(trial["id"])
        if not picks:
            continue
        day_scores = []
        max_c = 0
        for p in picks:
            correct = count_match(p["numbers"], winning)
            rank = RANK_BY_CORRECT.get(correct)
            db.update_auto_pick_grade(p["id"], correct, rank)
            day_scores.append(calc_pick_score(p["numbers"], winning))
            if correct > max_c:
                max_c = correct
        avg_score = sum(day_scores) / len(day_scores)
        per_day.append({
            "trial_id": trial["id"],
            "day_of_week": trial["day_of_week"],
            "weight": trial["weight"],
            "avg_score": avg_score,
            "max_correct": max_c,
            "n_picks": len(picks),
        })

    if not per_day:
        return {"ok": False, "reason": "no_picks_graded"}

    winner = max(per_day, key=lambda d: d["avg_score"])

    current_base = db.get_current_base()
    new_base, reason = decide_base_update(
        winner_max_correct=winner["max_correct"],
        winner_W=winner["weight"],
        current_base=current_base,
    )

    # 다음 주 월요일 base로 저장
    next_monday = today + timedelta(days=(7 - today.weekday()) % 7 or 7)
    db.save_base_history(
        effective_from=next_monday.isoformat(),
        weight=new_base,
        source_trial_id=winner["trial_id"],
        update_reason=reason,
        winner_score=winner["avg_score"],
        winner_max_correct=winner["max_correct"],
    )

    return {
        "ok": True,
        "draw_no": latest["drw_no"],
        "week_start": week_start,
        "winner": winner,
        "new_base": new_base,
        "update_reason": reason,
        "per_day": per_day,
    }

• Step 2: Smoke test all entry points compile

cd lotto && python -c "
from app.weight_evolver import (
    get_active_weight, generate_weekly_candidates_and_save,
    apply_today_and_pick, evaluate_weekly, get_week_start
)
print('all imports OK')
print('week_start:', get_week_start())
"

Expected: all imports OK + 오늘 주의 월요일 날짜

Also run all tests:

cd lotto && pytest tests/ -v 2>&1 | tail -10

Expected: no regression

• Step 3: Commit

git add lotto/app/weight_evolver.py
git commit -m "feat(weight-evolver): DB 통합 진입점 (generate_weekly/apply_today/evaluate_weekly)"

---

Task 6: 기존 시뮬레이션 cron에 active W 적용

Files:

• Modify: lotto/app/generator.py

• Step 1: Inject active W into run_simulation

lotto/app/generator.py의 run_simulation (line 30) — analyzer.score_combination 호출(line 72) 부분 수정:

기존 from .analyzer import build_analysis_cache, build_number_weights, score_combination 옆에 다음 import 추가:

from .weight_evolver import get_active_weight

그리고 run_simulation 함수 시작 부분에서 active W 조회 + score_combination 호출에 전달:

기존:

scores = score_combination(nums, cache)

변경:

scores = score_combination(nums, cache, weights=active_weights)

active_weights = get_active_weight()를 run_simulation 함수 진입 직후 한 번 호출하고 변수로 보관 (루프 안에서 매번 DB 호출 방지).

함수 시작부 예시:

def run_simulation(...):
    ...
    active_weights = get_active_weight()  # None이면 기존 fixed 사용
    cache = build_analysis_cache(draws)
    ...
    for ... in candidates:
        scores = score_combination(nums, cache, weights=active_weights)
        ...

• Step 2: Smoke test simulation runs

cd lotto && python -c "
import os
os.environ['LOTTO_DB_PATH'] = 'test_sim.db' if os.name == 'nt' else '/tmp/test_sim.db'
from app.db import init_db
init_db()
from app.generator import run_simulation
# fake draws — run_simulation은 draws 필요. 빈 DB에선 fail 가능 — 그냥 import만 확인.
print('import OK')
"

Expected: import OK

Run all tests:

cd lotto && pytest tests/ -v 2>&1 | tail -10

Expected: no regression

• Step 3: Commit

git add lotto/app/generator.py
git commit -m "feat(weight-evolver): run_simulation이 active W를 score_combination에 전달"

---

Phase 3 — cron + API

Task 7: lotto-lab cron 3종 + 진입점 wiring

Files:

• Modify: lotto/app/main.py

• Step 1: Add async wrapper functions + cron registrations

lotto/app/main.py의 scheduler 등록 부분(현재 _sync_and_check, _run_simulation_job 등 있는 영역)에 다음 추가:

# (기존 import 줄에 추가)
from .weight_evolver import (
    generate_weekly_candidates_and_save,
    apply_today_and_pick,
    evaluate_weekly,
)


async def _run_weight_evolver_weekly():
    """월 09:00 — 6개 후보 생성 후 inline으로 apply_today도 호출."""
    try:
        generate_weekly_candidates_and_save()
        # 같은 시각 race 방지 — generate 후 inline으로 apply 호출
        apply_today_and_pick(n=5)
    except Exception as e:
        logger.error(f"[weight_evolver_weekly] {e}")


async def _run_weight_evolver_daily():
    """매일 09:00 (월요일 제외 — 월은 weekly cron이 inline으로 처리)."""
    try:
        from datetime import datetime, timezone, timedelta
        KST = timezone(timedelta(hours=9))
        if datetime.now(KST).weekday() == 0:
            return  # 월요일은 weekly cron에서 처리됨
        apply_today_and_pick(n=5)
    except Exception as e:
        logger.error(f"[weight_evolver_daily] {e}")


async def _run_weight_evolver_eval():
    """토 22:00 — 회고 + 다음주 base 갱신."""
    try:
        evaluate_weekly()
    except Exception as e:
        logger.error(f"[weight_evolver_eval] {e}")

기존 scheduler 등록 (예: scheduler.add_job(_sync_and_check, ...) 근처)에 다음 3줄 추가:

scheduler.add_job(_run_weight_evolver_weekly, "cron", day_of_week="mon", hour=9, minute=0, id="weight_evolver_weekly")
scheduler.add_job(_run_weight_evolver_daily,  "cron", hour=9, minute=0,                  id="weight_evolver_daily")
scheduler.add_job(_run_weight_evolver_eval,   "cron", day_of_week="sat", hour=22, minute=0, id="weight_evolver_eval")

• Step 2: Verify scheduler loads

cd lotto && python -c "
from app.main import _run_weight_evolver_weekly, _run_weight_evolver_daily, _run_weight_evolver_eval
print('cron wrappers OK')
"

Expected: cron wrappers OK

• Step 3: Commit

git add lotto/app/main.py
git commit -m "feat(weight-evolver): cron 3종 등록 (월 generate+apply / 일 apply / 토 evaluate)"

---

Task 8: lotto-lab API 5종 endpoint

Files:

• Modify: lotto/app/main.py

• Step 1: Add 5 endpoints

lotto/app/main.py에서 다른 endpoint 근처에 추가:

@app.get("/api/lotto/evolver/status")
async def evolver_status():
    """현재 base + 이번주 trials 진행."""
    from .weight_evolver import get_week_start
    from .db import get_current_base, get_weekly_trials, get_auto_picks, get_latest_draw
    ws = get_week_start()
    trials = get_weekly_trials(ws)
    trials_with_picks = []
    for t in trials:
        picks = get_auto_picks(t["id"])
        trials_with_picks.append({**t, "picks": picks})
    latest = get_latest_draw()
    return {
        "week_start": ws,
        "current_base": get_current_base(),
        "trials": trials_with_picks,
        "latest_draw": latest["drw_no"] if latest else None,
    }


@app.get("/api/lotto/evolver/history")
async def evolver_history(weeks: int = 12):
    from .db import get_base_history
    return {"items": get_base_history(limit=weeks)}


@app.get("/api/lotto/evolver/trials/{week_start}")
async def evolver_trials(week_start: str):
    from .db import get_weekly_trials, get_auto_picks
    trials = get_weekly_trials(week_start)
    out = []
    for t in trials:
        picks = get_auto_picks(t["id"])
        out.append({**t, "picks": picks})
    return {"week_start": week_start, "trials": out}


@app.post("/api/lotto/evolver/generate-now")
async def evolver_generate_now():
    from .weight_evolver import generate_weekly_candidates_and_save
    candidates = generate_weekly_candidates_and_save()
    return {"ok": True, "candidates_count": len(candidates), "candidates": candidates}


@app.post("/api/lotto/evolver/evaluate-now")
async def evolver_evaluate_now():
    from .weight_evolver import evaluate_weekly
    return evaluate_weekly()

• Step 2: Verify routes

cd lotto && python -c "
from app.main import app
routes = sorted({r.path for r in app.routes if 'evolver' in r.path})
print('routes:', routes)
"

Expected: 5 routes listed including /api/lotto/evolver/status, /history, /trials/{week_start}, /generate-now, /evaluate-now

• Step 3: Commit

git add lotto/app/main.py
git commit -m "feat(weight-evolver): evolver API 5종 (status/history/trials/generate-now/evaluate-now)"

---

Phase 4 — agent-office 텔레그램 통합

Task 9: service_proxy에 lotto_evolver_status 추가

Files:

• Modify: agent-office/app/service_proxy.py

• Step 1: Append helper at end of file

async def lotto_evolver_status() -> Dict[str, Any]:
    """GET /api/lotto/evolver/status — 이번주 trials + 다음주 base 정보."""
    from .config import LOTTO_BACKEND_URL
    resp = await _client.get(f"{LOTTO_BACKEND_URL}/api/lotto/evolver/status")
    resp.raise_for_status()
    return resp.json()


async def lotto_evolver_evaluate() -> Dict[str, Any]:
    """POST /api/lotto/evolver/evaluate-now — 회고 트리거 (텔레그램 리포트용)."""
    from .config import LOTTO_BACKEND_URL
    async with httpx.AsyncClient(timeout=60.0) as client:
        resp = await client.post(f"{LOTTO_BACKEND_URL}/api/lotto/evolver/evaluate-now")
        resp.raise_for_status()
        return resp.json()

• Step 2: Verify import

cd agent-office && python -c "from app.service_proxy import lotto_evolver_status, lotto_evolver_evaluate; print('OK')"

• Step 3: Commit

git add agent-office/app/service_proxy.py
git commit -m "feat(lotto-evolver): service_proxy.lotto_evolver_status/evaluate helpers"

---

Task 10: 텔레그램 evolution_report 포맷 + 테스트 (TDD)

Files:

• Create: agent-office/tests/test_lotto_evolution_format.py

• Modify: agent-office/app/notifiers/telegram_lotto.py

• Step 1: Write failing tests

# agent-office/tests/test_lotto_evolution_format.py
from app.notifiers.telegram_lotto import _format_evolution_report


def test_evolution_report_winner_4plus():
    eval_result = {
        "draw_no": 1225,
        "week_start": "2026-05-18",
        "winner": {
            "day_of_week": 3,  # 목요일
            "weight": [0.18, 0.32, 0.20, 0.22, 0.08],
            "avg_score": 0.42,
            "max_correct": 4,
            "n_picks": 5,
        },
        "new_base": [0.18, 0.32, 0.20, 0.22, 0.08],
        "update_reason": "winner_4plus",
        "per_day": [
            {"day_of_week": 0, "avg_score": 0.20, "max_correct": 2},
            {"day_of_week": 3, "avg_score": 0.42, "max_correct": 4},
        ],
    }
    current_base = [0.20, 0.20, 0.20, 0.20, 0.20]
    text = _format_evolution_report(eval_result, current_base)
    assert "🧬" in text
    assert "1225" in text
    assert "목요일" in text or "Winner" in text
    assert "4개 일치" in text
    assert "winner_4plus" in text


def test_evolution_report_unchanged():
    eval_result = {
        "draw_no": 1226,
        "week_start": "2026-05-25",
        "winner": {
            "day_of_week": 1,
            "weight": [0.21, 0.19, 0.20, 0.20, 0.20],
            "avg_score": 0.10,
            "max_correct": 2,
            "n_picks": 5,
        },
        "new_base": [0.20, 0.20, 0.20, 0.20, 0.20],
        "update_reason": "unchanged",
        "per_day": [],
    }
    current_base = [0.20, 0.20, 0.20, 0.20, 0.20]
    text = _format_evolution_report(eval_result, current_base)
    assert "unchanged" in text or "유지" in text
    assert "2개 일치" in text or "max=2" in text


def test_evolution_report_empty_returns_empty():
    """evaluate가 ok=False면 빈 문자열 (발송 skip)."""
    text = _format_evolution_report({"ok": False, "reason": "no_trials"}, [0.2]*5)
    assert text == ""

Run: cd agent-office && pytest tests/test_lotto_evolution_format.py -v Expected: FAIL ImportError: cannot import name '_format_evolution_report'

• Step 2: Implement

agent-office/app/notifiers/telegram_lotto.py 파일 끝에 추가:

# ---------- Weight Evolver 주간 리포트 ----------

_DAY_NAMES = ["월", "화", "수", "목", "금", "토"]
_METRIC_NAMES = ["freq", "finger", "gap", "cooccur", "divers"]
_REASON_LABEL = {
    "winner_4plus": "4개 이상 일치 → base 교체",
    "ema_blend": "3개 일치 → EMA blend (0.3)",
    "unchanged": "유효 성과 없음 → base 유지",
    "cold_start": "초기 균등 적용",
}


def _format_evolution_report(eval_result: Dict[str, Any], current_base: List[float]) -> str:
    """주간 weight evolution 텔레그램 메시지."""
    if not eval_result or not eval_result.get("ok", True) is True and "winner" not in eval_result:
        return ""
    if "winner" not in eval_result:
        return ""

    draw_no = eval_result.get("draw_no", "?")
    winner = eval_result["winner"]
    new_base = eval_result["new_base"]
    reason = eval_result.get("update_reason", "")
    dow = winner.get("day_of_week", 0)
    day_name = _DAY_NAMES[dow] if 0 <= dow < len(_DAY_NAMES) else "?"

    lines = [
        f"🧬 로또 학습 주간 리포트 ({draw_no}회차)",
        "",
        f"이번주 시도: 6일 × {winner.get('n_picks', 5)}세트",
        "",
        f"🏆 Winner: {day_name}요일",
        f"  W = [" + ", ".join(
            f"{name} {w:.2f}" for name, w in zip(_METRIC_NAMES, winner["weight"])
        ) + "]",
        f"  최고 적중: {winner.get('max_correct', 0)}개 일치 (max={winner.get('max_correct', 0)})",
        f"  평균 점수: {winner.get('avg_score', 0):.2f}",
        "",
        f"📊 다음주 base 변경 ({reason}):",
    ]
    for i, (cur, new) in enumerate(zip(current_base or [0]*5, new_base or [0]*5)):
        diff = new - cur
        if abs(diff) < 0.005:
            marker = "="
        elif diff > 0:
            marker = "+" if diff < 0.05 else "++"
        else:
            marker = "-" if diff > -0.05 else "--"
        lines.append(f"  {_METRIC_NAMES[i]:8s} {cur:.2f} → {new:.2f}  ({marker})")
    lines.append("")
    lines.append(f"  → {_REASON_LABEL.get(reason, reason)}")
    lines.append("")
    lines.append(f"[웹에서 차트 보기] ({LOTTO_URL}/evolver)")
    return "\n".join(lines)


async def send_evolution_report(eval_result: Dict[str, Any], current_base: List[float]) -> None:
    text = _format_evolution_report(eval_result, current_base)
    if not text:
        return
    try:
        await send_raw(text)
    except Exception as e:
        logger.warning(f"[telegram_lotto] evolution report send failed: {e}")

List 가 위쪽 typing import에 있는지 확인. 없으면 추가.

• Step 3: Run tests pass

cd agent-office && pytest tests/test_lotto_evolution_format.py -v
cd agent-office && pytest tests/ -v 2>&1 | tail -5

Expected: 3 new + existing all pass (66 total)

• Step 4: Commit

git add agent-office/app/notifiers/telegram_lotto.py agent-office/tests/test_lotto_evolution_format.py
git commit -m "feat(lotto-evolver): 텔레그램 주간 evolution report 포맷 + 발송"

---

Task 11: LottoAgent.run_weekly_evolution_report + cron

Files:

• Modify: agent-office/app/agents/lotto.py

• Modify: agent-office/app/scheduler.py

• Step 1: Add method to LottoAgent

agent-office/app/agents/lotto.py 클래스 끝에 추가:

    async def run_weekly_evolution_report(self) -> dict:
        """토요일 22:15 — lotto-lab evaluate-now 트리거 후 텔레그램 리포트."""
        from ..service_proxy import lotto_evolver_evaluate, lotto_evolver_status
        from ..notifiers.telegram_lotto import send_evolution_report
        from ..db import add_log

        try:
            eval_result = await lotto_evolver_evaluate()
            status = await lotto_evolver_status()
            current_base = status.get("current_base") or [0.2] * 5
            await send_evolution_report(eval_result, current_base)
            add_log(
                self.agent_id,
                f"weekly_evolution_report 발송: draw={eval_result.get('draw_no')} reason={eval_result.get('update_reason')}",
            )
            return {"ok": True, **eval_result}
        except Exception as e:
            add_log(self.agent_id, f"weekly_evolution_report 예외: {e}", level="error")
            return {"ok": False, "message": f"{type(e).__name__}: {e}"}

• Step 2: Add cron in scheduler.py

agent-office/app/scheduler.py의 기존 lotto cron 근처에 함수 + 등록 추가:

async def _run_lotto_weekly_evolution_report():
    agent = AGENT_REGISTRY.get("lotto")
    if agent:
        await agent.run_weekly_evolution_report()

기존 scheduler.add_job(_run_lotto_daily_digest, ...) 근처에:

scheduler.add_job(_run_lotto_weekly_evolution_report, "cron", day_of_week="sat", hour=22, minute=15, id="lotto_evolution_weekly")

• Step 3: Verify

cd agent-office && python -c "
from app.scheduler import _run_lotto_weekly_evolution_report
print('cron OK')
"
cd agent-office && pytest tests/ -v 2>&1 | tail -5

Expected: import OK, all tests still pass

• Step 4: Commit

git add agent-office/app/agents/lotto.py agent-office/app/scheduler.py
git commit -m "feat(lotto-evolver): LottoAgent.run_weekly_evolution_report + 토 22:15 cron"

---

Task 12: CLAUDE.md 업데이트

Files:

• Modify: web-backend/CLAUDE.md

• Step 1: Update lotto-lab section

web-backend/CLAUDE.md의 lotto-lab API 목록 표에 추가 (lotto API 섹션 끝):

| GET | `/api/lotto/evolver/status` | weight_evolver 이번주 trials + current_base + 진행 상황 |
| GET | `/api/lotto/evolver/history?weeks=12` | base 변경 이력 |
| GET | `/api/lotto/evolver/trials/{week_start}` | 특정 주 6 trials + 채점 결과 |
| POST | `/api/lotto/evolver/generate-now` | 수동 트리거 — 이번주 후보 생성 |
| POST | `/api/lotto/evolver/evaluate-now` | 수동 회고 + 다음주 base 갱신 |

스케줄러 job 항목에 추가:

- 월요일 09:00 — weight_evolver_weekly (6개 후보 생성 + 그날 N=5 추출)
- 매일 09:00 — weight_evolver_daily (월요일 제외, 오늘 W로 N=5 추출)
- 토요일 22:00 — weight_evolver_eval (회고 + 다음주 base 갱신)

lotto-lab "테이블" 항목에 추가:

| `weight_trials` | 주별 6일치 후보 가중치 (4 perturb + 2 dirichlet) |
| `auto_picks` | 매일 N=5 시도 번호 + 채점 결과 |
| `weight_base_history` | base 갱신 이력 (winner_4plus / ema_blend / unchanged / cold_start) |

agent-office API 표에는 추가 API 없음. agent-office "스케줄러 job"에 추가:

- 토요일 22:15 — 로또 weight_evolver 주간 텔레그램 리포트

• Step 2: Commit

git -C C:/Users/jaeoh/Desktop/workspace/web-backend add CLAUDE.md
git -C C:/Users/jaeoh/Desktop/workspace/web-backend commit -m "docs(CLAUDE): lotto-lab weight_evolver API/스케줄러/테이블 추가"

---

Task 13: NAS 배포 + 수동 트리거 검증 (사용자 수동)

• Step 1: push로 자동 배포

git push

• Step 2: 컨테이너 로그 확인

ssh nas "docker logs lotto --tail 80"
ssh nas "docker logs agent-office --tail 60"

체크:

• lotto: weight_evolver_weekly / weight_evolver_daily / weight_evolver_eval 3 cron 등록 메시지

• agent-office: lotto_evolution_weekly cron 등록

• Step 3: 수동 후보 생성

curl -X POST "https://gahusb.synology.me/api/lotto/evolver/generate-now"

기대: {"ok": true, "candidates_count": 6, "candidates": [...]}

• Step 4: status 확인

curl "https://gahusb.synology.me/api/lotto/evolver/status" | python -m json.tool

체크:

• week_start 이번주 월요일

• current_base: cold_start로 [0.2, 0.2, 0.2, 0.2, 0.2]

• trials 6개, each with weight 5종

• Step 5: 오늘 W로 picks 추출

(weight_evolver_daily 09:00을 기다리거나 수동 호출 가능하면)

# 또는 직접 docker exec
ssh nas "docker exec lotto python -c 'from app.weight_evolver import apply_today_and_pick; print(apply_today_and_pick(n=5))'"

기대: 5세트 출력 + auto_picks 테이블에 저장

• Step 6: 토요일 22:00 자동 evaluate 대기 + 22:15 텔레그램 리포트 도착 확인

또는 강제 evaluate-now:

curl -X POST "https://gahusb.synology.me/api/lotto/evolver/evaluate-now" | python -m json.tool

이번 회차 winning numbers와 매칭 + 다음주 base 결정

• Step 7: 텔레그램 폼 강제 확인 (선택)

ssh nas "docker exec agent-office python -c \"
import asyncio
from app.service_proxy import lotto_evolver_evaluate, lotto_evolver_status
from app.notifiers.telegram_lotto import send_evolution_report

async def main():
    e = await lotto_evolver_evaluate()
    s = await lotto_evolver_status()
    await send_evolution_report(e, s.get('current_base') or [0.2]*5)
    print('sent')

asyncio.run(main())
\""

기대: 텔레그램에 🧬 메시지 도착

---

Self-Review (수행 완료)

1. Spec coverage check

Spec 섹션	구현 task
4.1 Weight Vector	Task 2 (MIN_WEIGHT, N_METRICS)
4.2 6개 후보 생성	Task 2 (perturb + dirichlet), Task 5 (generate_weekly_candidates_and_save)
4.3 일일 W 적용	Task 5 (apply_today_and_pick), Task 6 (analyzer 통합)
4.4 토요일 회고	Task 5 (evaluate_weekly), Task 2 (calc_pick_score)
4.5 Base 갱신 Hybrid	Task 2 (decide_base_update)
4.6 Cold start	Task 2 (DEFAULT_UNIFORM), Task 5 (cold_start path in generate_weekly_candidates_and_save)
5.1-5.3 DB 스키마	Task 3
6 analyzer 시그니처	Task 4
6.1 get_active_weight	Task 5
7 API 5종	Task 8
8 cron 3종	Task 7
9 텔레그램 리포트	Task 10, 11
10 v1 cascade	자동 (Task 6 분석기 변경 효과)
11 Phase 1-4	Task 1-12
12 비기능 요구	Task 1 (단위테스트), Task 4 (analyzer 보강 테스트), Task 10 (텔레그램 폼 테스트)

모두 매핑됨.

2. Placeholder scan: 없음. 모든 step에 실제 코드 또는 명확한 명령.

3. Type consistency:

• generate_weekly_candidates(base, seed) → returns List[Dict] with keys day_of_week, weight, source — consistent across Task 2 (정의), Task 5 (호출), Task 8 (API 반환)
• decide_base_update(winner_max_correct, winner_W, current_base) → returns Tuple[List[float], str] — Task 2 정의, Task 5 호출 일치
• DB CRUD 함수명 (save_weight_trial, get_weekly_trials, save_base_history, etc.) — Task 3 정의, Task 5/8 호출 일치
• _format_evolution_report(eval_result, current_base) — Task 10 정의, Task 11 호출 일치
• lotto_evolver_status 응답 키 (week_start, current_base, trials, latest_draw) — Task 8 정의, Task 11 사용 일치

이슈 없음.

---

비목표 (Out of scope, v3 후속)

• 메타 전략(strategy_evolver) 가중치 동시 학습 — v3에서 검토
• Multi-objective 점수 (적중 + 분포 균등 등)
• 자동 구매
• 프론트 /lotto/evolver UI — web-ui repo 별도 PR
• 강화학습 (UCB1/policy gradient)