From 4063f29cd32162bda87ac3fa0c021625c1454056 Mon Sep 17 00:00:00 2001 From: gahusb Date: Sun, 31 May 2026 17:47:52 +0900 Subject: [PATCH] =?UTF-8?q?fix(lotto):=20=ED=95=99=EC=8A=B5=20=EA=B2=8C?= =?UTF-8?q?=EC=9D=B4=ED=8A=B8=20=EC=A0=95=EC=A7=81=ED=99=94=20(engine-best?= =?UTF-8?q?=20vs=20random-best=206trial=C2=B7=EB=AA=85=EC=8B=9C=EC=A0=81?= =?UTF-8?q?=20gated=C2=B7=EC=A0=95=EC=B2=B4=EC=84=B1=20=EC=9D=BC=EA=B4=80)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.8 (1M context) --- lotto/app/backtest.py | 9 +- lotto/app/weight_evolver.py | 43 +++++++--- lotto/tests/test_backtest_db.py | 127 +++++++++++++++++++++++++++++ lotto/tests/test_weight_evolver.py | 56 ++++++++++++- 4 files changed, 221 insertions(+), 14 deletions(-) diff --git a/lotto/app/backtest.py b/lotto/app/backtest.py index c86a809..a9ef617 100644 --- a/lotto/app/backtest.py +++ b/lotto/app/backtest.py @@ -6,6 +6,9 @@ from typing import Any, Dict, List, Optional, Tuple from .analyzer import build_analysis_cache, build_number_weights, score_combination from .utils import weighted_sample_6 +# engine_w trials 수와 동일하게 맞춰 selection bias를 상쇄한다. +N_NULL_TRIALS = 6 + def grade_tickets(tickets: List[List[int]], winning6: List[int], bonus: int) -> Dict[str, Any]: """티켓 묶음을 당첨번호로 채점 → 매칭 히스토그램 + 보너스 + best_match. @@ -194,8 +197,10 @@ def run_forward_purchase(draw_no: int, k: int = 5000, pool_n: int = 20000, bought = purchase_tickets(pool, cache, base, k) _store("engine_w", "base", base, None, bought) - # 2) random_null - _store("random_null", "-", None, None, random_null_tickets(k, seed=sample_seed)) + # 2) random_null — N_NULL_TRIALS 개 (engine_w 수와 동일해 selection bias 상쇄) + for _i in range(N_NULL_TRIALS): + seed_i = None if sample_seed is None else sample_seed + 100 + _i + _store("random_null", f"r{_i}", None, None, random_null_tickets(k, seed=seed_i)) # 3) coverage _store("coverage", "-", None, None, coverage_tickets(k, seed=sample_seed)) diff --git a/lotto/app/weight_evolver.py b/lotto/app/weight_evolver.py index 2ff91fb..c323e58 100644 --- a/lotto/app/weight_evolver.py +++ b/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() diff --git a/lotto/tests/test_backtest_db.py b/lotto/tests/test_backtest_db.py index 1a7dad2..f8876bc 100644 --- a/lotto/tests/test_backtest_db.py +++ b/lotto/tests/test_backtest_db.py @@ -182,6 +182,7 @@ def test_track_record_and_review_payload(monkeypatch): tr = bt.track_record() assert "random_null" in tr["by_strategy"] + # 이제 random_null은 N_NULL_TRIALS=6 행이므로 6*20=120장 assert tr["by_strategy"]["random_null"]["n_tickets"] >= 20 payload = bt.build_review_payload(40) @@ -191,3 +192,129 @@ def test_track_record_and_review_payload(monkeypatch): assert "calibration_trend" in payload assert payload["winner_analysis"] is not None assert "score_total" in payload["winner_analysis"] + + +def test_run_forward_purchase_random_null_count(monkeypatch): + """run_forward_purchase는 random_null을 N_NULL_TRIALS=6개 저장해야 한다.""" + db = _fresh_db(monkeypatch) + _seed_draws(db, 40) + from app import backtest as bt + res = bt.run_forward_purchase(draw_no=40, k=20, pool_n=500, sample_seed=7) + assert res["ok"] is True + rows = db.get_backtest_runs(draw_no=40) + null_rows = [r for r in rows if r["strategy"] == "random_null"] + assert len(null_rows) == bt.N_NULL_TRIALS # 6개 + null_labels = {r["weight_label"] for r in null_rows} + assert null_labels == {f"r{i}" for i in range(bt.N_NULL_TRIALS)} + for r in null_rows: + assert r["n_tickets"] == 20 + + +def test_evaluate_weekly_gated_keeps_base_unchanged(monkeypatch): + """Fix 5 통합 테스트 (end-to-end gated path). + + 접근: DB에 draws, weight_trials, auto_picks, backtest_runs, base_history를 직접 심어 + evaluate_weekly()의 gated 분기가 base를 바꾸지 않음을 검증한다. + + gated 조건: engine_w 최고 prize_score − random_best < LIFT_EPSILON(10.0). + engine_best=5, random_best=20 → lift=-15 → gated. + + evaluate_weekly 내부 흐름: + - get_weekly_trials(week_start) : _today_kst() 기준 week_start 사용 + - get_latest_draw() : draws 테이블에서 max(drw_no) 반환 + 두 참조가 같은 날짜 기준이어야 하므로 _today_kst를 monkeypatch로 고정하고 + draws의 최신 회차 날짜(drw_date)를 해당 주의 날짜로 맞춘다. + """ + import json as _json + from datetime import date, timedelta, datetime as _dt, timezone as _tz, timedelta as _td + + db = _fresh_db(monkeypatch) + + # KST 오늘 날짜 — evaluate_weekly가 이 날짜를 기준으로 week_start 계산 + KST = _tz(_td(hours=9)) + today_kst = _dt.now(KST).date() + from app import weight_evolver as we + week_start = we.get_week_start(today_kst) + + # 1) draws 심기 — 최신 회차의 drw_date를 week_start 주 안의 날짜로 맞춤 + import random as _r; _r.seed(99) + rows = [] + for i in range(1, 41): + s = sorted(_r.sample(range(1, 46), 6)) + # 마지막 회차(40)는 오늘 날짜 사용 (week_start 주 내) + if i == 40: + drw_date = today_kst.isoformat() + else: + drw_date = f"2020-01-{(i % 28) + 1:02d}" + rows.append({ + "drw_no": i, "drw_date": drw_date, + "n1": s[0], "n2": s[1], "n3": s[2], + "n4": s[3], "n5": s[4], "n6": s[5], + "bonus": (s[5] % 45) + 1, + }) + db.upsert_many_draws(rows) + latest = db.get_latest_draw() + assert latest is not None + assert latest["drw_date"] == today_kst.isoformat() + + # 2) weight trial 1개 심기 (day_of_week=0, week_start=오늘 주) + trial_w = [0.2, 0.2, 0.2, 0.2, 0.2] + db.save_weight_trial(week_start, 0, trial_w, "perturb") + trial_rows = db.get_weekly_trials(week_start) + assert len(trial_rows) == 1 + trial_id = trial_rows[0]["id"] + + # 3) auto_picks 1개 심기 (winning 번호와 2개 일치 → max_correct=2) + winning6 = [latest["n1"], latest["n2"], latest["n3"], + latest["n4"], latest["n5"], latest["n6"]] + pick = winning6[:2] + [40, 41, 42, 43] + db.save_auto_pick(trial_id, 1, pick, meta_score=0.5) + + # 4) backtest_runs: engine_w prize_score=5, random_null 6개 prize_score=20 (gated 확실) + LOW_HIST = {"m3": 5, "m4": 0, "m5": 0, "m6": 0, "bonus_hits": 0} # prize=5 + HIGH_HIST = {"m3": 20, "m4": 0, "m5": 0, "m6": 0, "bonus_hits": 0} # prize=20 + draw_no = latest["drw_no"] + db.save_backtest_run( + draw_no=draw_no, strategy="engine_w", weight_label="w0", + weight_json=_json.dumps(trial_w), trial_id=trial_id, n_tickets=20, + hist=LOW_HIST, best_match=2, avg_meta_score=0.5, + ) + from app import backtest as bt + for i in range(bt.N_NULL_TRIALS): + db.save_backtest_run( + draw_no=draw_no, strategy="random_null", weight_label=f"r{i}", + weight_json=None, trial_id=None, n_tickets=20, + hist=HIGH_HIST, best_match=3, avg_meta_score=0.5, + ) + + # 5) current base 저장 (이전 주 월요일 effective_from) + base_w = [0.2, 0.2, 0.2, 0.2, 0.2] + prev_monday = (today_kst - timedelta(weeks=1, days=today_kst.weekday())).isoformat() + db.save_base_history( + effective_from=prev_monday, + weight=base_w, + source_trial_id=None, + update_reason="cold_start", + winner_score=None, + winner_max_correct=None, + ) + assert db.get_current_base() == base_w + + # 6) evaluate_weekly 호출 — _today_kst()를 monkeypatch로 오늘 날짜 고정 + monkeypatch.setattr(we, "_today_kst", lambda: today_kst) + + result = we.evaluate_weekly() + + assert result.get("ok") is True, f"evaluate_weekly 실패: {result}" + + # gated path 검증 + update_reason = result.get("update_reason", "") + assert update_reason in ("unchanged_gated", "idempotent_skip"), ( + f"gated여야 하는데 reason='{update_reason}' — 게이팅 로직 깨짐" + ) + + # base가 바뀌지 않았는지 검증 + new_base = result.get("new_base") + assert new_base == base_w, ( + f"gated인데 base가 변경됨: {new_base} != {base_w}" + ) diff --git a/lotto/tests/test_weight_evolver.py b/lotto/tests/test_weight_evolver.py index 2d642f5..45ba6f8 100644 --- a/lotto/tests/test_weight_evolver.py +++ b/lotto/tests/test_weight_evolver.py @@ -129,7 +129,7 @@ def test_select_winner_by_lift_gating(): {"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 안에서 게이팅 + # random baseline이 8.0이면 lift는 -3, +1, -4 → 최대 lift(+1) < ε(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) @@ -142,3 +142,57 @@ def test_prize_score_from_hist(): s = we.prize_score_from_hist({"m3": 10, "m4": 2, "m5": 0, "m6": 0, "bonus_hits": 0}) s_big = we.prize_score_from_hist({"m3": 0, "m4": 0, "m5": 0, "m6": 1, "bonus_hits": 0}) assert s_big > s # 1등 1장이 5등 다수보다 큼 + + +def test_select_winner_by_lift_preserves_all_keys(): + """select_winner_by_lift는 per_w 항목의 모든 키를 보존해야 한다. + best_match, weight_label 등 identity 필드가 누락되면 evaluate_weekly가 깨진다.""" + per_w = [ + { + "trial_id": 10, + "weight_label": "w0", + "weight": [0.2] * 5, + "prize_score": 3.0, + "best_match": 3, + }, + { + "trial_id": 11, + "weight_label": "w1", + "weight": [0.3, 0.2, 0.2, 0.2, 0.1], + "prize_score": 20.0, + "best_match": 4, + }, + ] + result = we.select_winner_by_lift(per_w, random_score=5.0, epsilon=2.0) + assert result["gated"] is False + assert result["trial_id"] == 11 + assert result["weight_label"] == "w1" # identity 키 보존 + assert result["best_match"] == 4 # best_match 키 보존 + assert "lift" in result # lift 추가됨 + assert result["lift"] == pytest.approx(15.0) + + +def test_gated_path_keeps_base_via_select_winner(): + """gated=True일 때 select_winner_by_lift의 반환값 검증. + evaluate_weekly 내의 gated 분기가 올바른 값에 의존함을 확인한다.""" + per_w = [ + {"trial_id": 1, "weight_label": "w0", "weight": [0.2]*5, + "prize_score": 5.0, "best_match": 2}, + {"trial_id": 2, "weight_label": "w1", "weight": [0.3,0.2,0.2,0.2,0.1], + "prize_score": 7.0, "best_match": 3}, + ] + # random_best=8.0 → 최대 engine lift=7-8=-1 → gated + result = we.select_winner_by_lift(per_w, random_score=8.0, epsilon=we.LIFT_EPSILON) + assert result["gated"] is True + assert result["lift"] < 0 + + # decide_base_update를 통해 gated가 unchanged를 유도하는지 확인 + # (gated override가 없더라도, 현재 LIFT_EPSILON=10.0 하에서 lift<0이면 항상 gated) + current = [0.2, 0.2, 0.2, 0.2, 0.2] + # gated이면 evaluate_weekly가 current_base를 그대로 유지해야 함 + # 여기서는 override 로직을 직접 재현해 검증한다 + gated = result["gated"] + new_base_override = list(current) if gated else None + reason_override = "unchanged_gated" if gated else "should_not_reach" + assert new_base_override == current + assert reason_override == "unchanged_gated"