From f017a61c79bb204f40ea7ff5181620d7c62d1d62 Mon Sep 17 00:00:00 2001 From: gahusb Date: Fri, 22 May 2026 03:08:56 +0900 Subject: [PATCH] =?UTF-8?q?feat(weight-evolver):=20DB=20=ED=86=B5=ED=95=A9?= =?UTF-8?q?=20=EC=A7=84=EC=9E=85=EC=A0=90=20(generate=5Fweekly/apply=5Ftod?= =?UTF-8?q?ay/evaluate=5Fweekly)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- lotto/app/weight_evolver.py | 176 ++++++++++++++++++++++++++++++++++++ 1 file changed, 176 insertions(+) diff --git a/lotto/app/weight_evolver.py b/lotto/app/weight_evolver.py index 591cec7..f10b97a 100644 --- a/lotto/app/weight_evolver.py +++ b/lotto/app/weight_evolver.py @@ -121,3 +121,179 @@ def decide_base_update( if current_base is None: return DEFAULT_UNIFORM[:], "cold_start" return list(current_base), "unchanged" + + +# ---------- 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): + 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, + ) + + 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, + }