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>
1588 lines
50 KiB
Markdown
1588 lines
50 KiB
Markdown
# 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)
|