fix(lotto): 학습 게이트 정직화 (engine-best vs random-best 6trial·명시적 gated·정체성 일관)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

lotto/app/weight_evolver.py

@@ -19,7 +19,9 @@ 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}
 
-LIFT_EPSILON = 0.5  # 등수점수 노이즈 게이팅 임계 (튜닝 가능)
+LIFT_EPSILON = 10.0  # best-of-engine vs best-of-random margin;
+# selection bias already cancelled by equal group sizes (N_NULL_TRIALS == engine trial count);
+# tune as needed.
 
 PRIZE_WEIGHTS = {"m6": 1000.0, "bonus_hits": 50.0, "m5": 30.0, "m4": 4.0, "m3": 1.0}
 
@@ -35,7 +37,9 @@ def select_winner_by_lift(per_w: List[Dict[str, Any]], random_score: float,
 
 def prize_score_from_hist(hist: Dict[str, int]) -> float:
     """매칭 히스토그램 → 등수 가중 합산 점수.
-    1등=m6, 2등=bonus_hits, 3등=m5−bonus_hits, 4등=m4, 5등=m3."""
+    1등=m6, 2등=bonus_hits, 3등=m5−bonus_hits, 4등=m4, 5등=m3.
+    m3/m4/m5/m6/bonus_hits 키만 읽으며 나머지는 무시하므로
+    DB 전체 행(backtest_runs row)을 그대로 넘겨도 안전하다."""
     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"]
@@ -294,34 +298,46 @@ def evaluate_weekly() -> Dict[str, Any]:
 
     winner = max(per_day, key=lambda d: d["avg_score"])
 
-    # 자가학습 강화: backtest forward 등수점수 lift로 winner 재선정
+    # 자가학습 강화: backtest forward 등수점수 lift로 winner 재선정.
+    # best-of-engine vs best-of-random 비교 — 동등 그룹 크기로 selection bias 상쇄.
     latest_no = latest["drw_no"]
     runs = db.get_backtest_runs(draw_no=latest_no)
     engine_runs = [r for r in runs if r["strategy"] == "engine_w"]
     null_runs = [r for r in runs if r["strategy"] == "random_null"]
+    gated = False  # 이후 decide_base_update override에 사용
     if engine_runs and null_runs:
-        random_score = prize_score_from_hist(null_runs[0])
+        # base 단독 행이 있고 w* 행도 있으면 base 행 제외 (identity collision 방지)
+        has_w_trials = any(r["weight_label"].startswith("w") for r in engine_runs)
+        if has_w_trials:
+            engine_runs = [r for r in engine_runs if r["weight_label"] != "base"]
+
+        # best-of-random: 동등 그룹의 최댓값 (selection bias 상쇄)
+        random_best = max(prize_score_from_hist(r) for r in null_runs)
+
         per_w = []
         for r in engine_runs:
             per_w.append({
                 "trial_id": r["trial_id"],
-                "day_of_week": int(r["weight_label"][1:]) if r["weight_label"].startswith("w") else 0,
+                "weight_label": r["weight_label"],
                 "weight": json.loads(r["weight_json"]) if r["weight_json"] else DEFAULT_UNIFORM[:],
                 "prize_score": prize_score_from_hist(r),
+                "best_match": r["best_match"],
             })
-        lift_winner = select_winner_by_lift(per_w, random_score=random_score)
+
+        lift_winner = select_winner_by_lift(per_w, random_score=random_best)
         if not lift_winner["gated"]:
+            # lift winner의 정체성과 채점값을 일관되게 사용
             winner = {
                 "trial_id": lift_winner["trial_id"],
-                "day_of_week": lift_winner["day_of_week"],
                 "weight": lift_winner["weight"],
-                "avg_score": winner["avg_score"],
-                "max_correct": winner["max_correct"],
+                "max_correct": lift_winner["best_match"],   # 이 trial의 실제값
+                "avg_score": lift_winner["prize_score"],    # lift winner의 prize score
                 "lift": lift_winner["lift"],
             }
         else:
-            # 노이즈 → base 유지 강제 (max_correct를 0으로 낮춰 unchanged 유도)
-            winner = {**winner, "max_correct": min(winner["max_correct"], 2), "lift": lift_winner["lift"]}
+            # 노이즈 → gated 플래그 설정; decide_base_update 이후 명시적으로 override
+            gated = True
+            winner = {**winner, "lift": lift_winner["lift"]}
 
     current_base = db.get_current_base()
     new_base, reason = decide_base_update(
@@ -330,6 +346,11 @@ def evaluate_weekly() -> Dict[str, Any]:
         current_base=current_base,
     )
 
+    # gated path: decide_base_update 결과와 무관하게 base 유지 강제
+    if gated:
+        new_base = list(current_base) if current_base is not None else DEFAULT_UNIFORM[:]
+        reason = "unchanged_gated"
+
     next_monday = today + timedelta(days=(7 - today.weekday()) % 7 or 7)
     next_monday_iso = next_monday.isoformat()