feat(lotto): select_winner_by_lift + ε-게이팅

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-31 17:32:37 +09:00
parent add433233a
commit 8e7b4adabd
2 changed files with 40 additions and 0 deletions

View File

@@ -4,6 +4,7 @@
순수 함수 (clamp/perturb/Dirichlet/score/base-rule) + DB 진입점은 별도 섹션. 순수 함수 (clamp/perturb/Dirichlet/score/base-rule) + DB 진입점은 별도 섹션.
""" """
from __future__ import annotations from __future__ import annotations
import json
import math import math
import random import random
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
@@ -18,6 +19,30 @@ DEFAULT_UNIFORM = [0.2] * N_METRICS # cold start
RANK_BY_CORRECT = {6: 1, 5: 3, 4: 4, 3: 5} 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} RANK_BONUS = {1: 1.0, 2: 0.8, 3: 0.6, 4: 0.3, 5: 0.1}
LIFT_EPSILON = 0.5 # 등수점수 노이즈 게이팅 임계 (튜닝 가능)
PRIZE_WEIGHTS = {"m6": 1000.0, "bonus_hits": 50.0, "m5": 30.0, "m4": 4.0, "m3": 1.0}
def select_winner_by_lift(per_w: List[Dict[str, Any]], random_score: float,
epsilon: float = LIFT_EPSILON) -> Dict[str, Any]:
"""engine_w 후보들 중 random 대비 lift 최대 선택.
최대 lift가 epsilon 미만이면 gated=True (노이즈 → base 유지 권고)."""
scored = [{**w, "lift": w["prize_score"] - random_score} for w in per_w]
best = max(scored, key=lambda w: w["lift"])
return {**best, "gated": best["lift"] < epsilon}
def prize_score_from_hist(hist: Dict[str, int]) -> float:
"""매칭 히스토그램 → 등수 가중 합산 점수.
1등=m6, 2등=bonus_hits, 3등=m5bonus_hits, 4등=m4, 5등=m3."""
third = max(0, hist.get("m5", 0) - hist.get("bonus_hits", 0))
return (hist.get("m6", 0) * PRIZE_WEIGHTS["m6"]
+ hist.get("bonus_hits", 0) * PRIZE_WEIGHTS["bonus_hits"]
+ third * PRIZE_WEIGHTS["m5"]
+ hist.get("m4", 0) * PRIZE_WEIGHTS["m4"]
+ hist.get("m3", 0) * PRIZE_WEIGHTS["m3"])
def clamp_and_normalize(W: List[float], min_w: float = MIN_WEIGHT) -> List[float]: def clamp_and_normalize(W: List[float], min_w: float = MIN_WEIGHT) -> List[float]:
"""각 값 ≥ min_w + 합=1.0. 보장 안 되면 raise.""" """각 값 ≥ min_w + 합=1.0. 보장 안 되면 raise."""

View File

@@ -120,3 +120,18 @@ def test_decide_base_update_cold_start_returns_default():
) )
assert new_base == winner_W assert new_base == winner_W
assert reason == "winner_4plus" assert reason == "winner_4plus"
def test_select_winner_by_lift_gating():
# engine_w 3개 + random_null 기준. lift = engine 등수점수 random 등수점수
per_w = [
{"trial_id": 1, "day_of_week": 0, "weight": [0.2]*5, "prize_score": 5.0},
{"trial_id": 2, "day_of_week": 1, "weight": [0.3,0.2,0.2,0.2,0.1], "prize_score": 9.0},
{"trial_id": 3, "day_of_week": 2, "weight": [0.1,0.3,0.2,0.2,0.2], "prize_score": 4.0},
]
# random baseline이 8.0이면 lift는 +1, +1, -4 → 노이즈 ε=2 안에서 게이팅
winner = we.select_winner_by_lift(per_w, random_score=8.0, epsilon=2.0)
assert winner["gated"] is True # 최대 lift(+1) < ε(2) → 게이팅
winner2 = we.select_winner_by_lift(per_w, random_score=3.0, epsilon=2.0)
assert winner2["gated"] is False
assert winner2["trial_id"] == 2 # prize 9 → lift +6