From 6be74737c28d2342750e927ae035b7eee9db23d0 Mon Sep 17 00:00:00 2001 From: gahusb Date: Fri, 22 May 2026 01:38:23 +0900 Subject: [PATCH] =?UTF-8?q?docs(plan):=20Lotto=20Weight=20Evolver=20?= =?UTF-8?q?=EA=B5=AC=ED=98=84=20plan=20(13=20tasks,=20Phase=201-4=20+=20?= =?UTF-8?q?=EB=B0=B0=ED=8F=AC)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Why: spec (2026-05-22-lotto-weight-evolver-design.md)을 13개 atomic task로 분해. TDD red→green→commit 패턴. analyzer.score_combination 기존 fixed 가중치 보존+동적 W 옵션 추가. v1 시그널 자동 cascade. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../plans/2026-05-22-lotto-weight-evolver.md | 1587 +++++++++++++++++ 1 file changed, 1587 insertions(+) create mode 100644 docs/superpowers/plans/2026-05-22-lotto-weight-evolver.md diff --git a/docs/superpowers/plans/2026-05-22-lotto-weight-evolver.md b/docs/superpowers/plans/2026-05-22-lotto-weight-evolver.md new file mode 100644 index 0000000..e0c616e --- /dev/null +++ b/docs/superpowers/plans/2026-05-22-lotto-weight-evolver.md @@ -0,0 +1,1587 @@ +# 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)