Files
web-page-backend/docs/superpowers/plans/2026-05-22-lotto-weight-evolver.md
gahusb 6be74737c2 docs(plan): Lotto Weight Evolver 구현 plan (13 tasks, Phase 1-4 + 배포)
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) <noreply@anthropic.com>
2026-05-22 01:38:23 +09:00

1588 lines
50 KiB
Markdown
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 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)