feat(weight-evolver): 순수 함수 (clamp/perturb/Dirichlet/score/base-rule)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

lotto/app/weight_evolver.py

@@ -0,0 +1,123 @@
+# 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")
+    # Iteratively clamp then normalize until all values satisfy min_w floor.
+    # (Normalizing after clamping can reduce some already-floored values below
+    #  min_w when the denominator is large — iterate to convergence.)
+    vals = [float(w) for w in W]
+    for _ in range(100):  # converges in a few iterations in practice
+        clamped = [max(min_w, v) for v in vals]
+        total = sum(clamped)
+        vals = [v / total for v in clamped]
+        if all(v >= min_w - 1e-12 for v in vals):
+            break
+    return vals
+
+
+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):
+        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"