# 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** ```python # 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** ```bash 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** ```python # 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** ```bash 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 전) 추가: ```python 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** ```python # --- 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** ```bash 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** ```bash 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** ```python # 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) 수정: ```python # 시그니처 변경 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): ```python # 기존 score_total = ( score_frequency * 0.25 + score_fingerprint * 0.30 + score_gap * 0.20 + score_cooccur * 0.15 + score_diversity * 0.10 ) ``` 다음으로 교체: ```python 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** ```bash 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** ```bash 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** ```python # ---------- 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** ```bash 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: ```bash cd lotto && pytest tests/ -v 2>&1 | tail -10 ``` Expected: no regression - [ ] **Step 3: Commit** ```bash 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 추가: ```python from .weight_evolver import get_active_weight ``` 그리고 `run_simulation` 함수 시작 부분에서 active W 조회 + `score_combination` 호출에 전달: 기존: ```python scores = score_combination(nums, cache) ``` 변경: ```python scores = score_combination(nums, cache, weights=active_weights) ``` `active_weights = get_active_weight()`를 `run_simulation` 함수 진입 직후 한 번 호출하고 변수로 보관 (루프 안에서 매번 DB 호출 방지). 함수 시작부 예시: ```python 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** ```bash 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: ```bash cd lotto && pytest tests/ -v 2>&1 | tail -10 ``` Expected: no regression - [ ] **Step 3: Commit** ```bash 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` 등 있는 영역)에 다음 추가: ```python # (기존 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줄 추가: ```python 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** ```bash 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** ```bash 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 근처에 추가: ```python @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** ```bash 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** ```bash 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** ```python 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** ```bash cd agent-office && python -c "from app.service_proxy import lotto_evolver_status, lotto_evolver_evaluate; print('OK')" ``` - [ ] **Step 3: Commit** ```bash 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** ```python # 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` 파일 끝에 추가: ```python # ---------- 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** ```bash 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** ```bash 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` 클래스 끝에 추가: ```python 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 근처에 함수 + 등록 추가: ```python 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, ...)` 근처에: ```python scheduler.add_job(_run_lotto_weekly_evolution_report, "cron", day_of_week="sat", hour=22, minute=15, id="lotto_evolution_weekly") ``` - [ ] **Step 3: Verify** ```bash 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** ```bash 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 섹션 끝): ```markdown | 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** 항목에 추가: ```markdown - 월요일 09:00 — weight_evolver_weekly (6개 후보 생성 + 그날 N=5 추출) - 매일 09:00 — weight_evolver_daily (월요일 제외, 오늘 W로 N=5 추출) - 토요일 22:00 — weight_evolver_eval (회고 + 다음주 base 갱신) ``` lotto-lab "테이블" 항목에 추가: ```markdown | `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"에 추가: ```markdown - 토요일 22:15 — 로또 weight_evolver 주간 텔레그램 리포트 ``` - [ ] **Step 2: Commit** ```bash 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로 자동 배포** ```bash git push ``` - [ ] **Step 2: 컨테이너 로그 확인** ```bash 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: 수동 후보 생성** ```bash curl -X POST "https://gahusb.synology.me/api/lotto/evolver/generate-now" ``` 기대: `{"ok": true, "candidates_count": 6, "candidates": [...]}` - [ ] **Step 4: status 확인** ```bash 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을 기다리거나 수동 호출 가능하면) ```bash # 또는 직접 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: ```bash curl -X POST "https://gahusb.synology.me/api/lotto/evolver/evaluate-now" | python -m json.tool ``` 이번 회차 winning numbers와 매칭 + 다음주 base 결정 - [ ] **Step 7: 텔레그램 폼 강제 확인 (선택)** ```bash 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)