""" 시뮬레이션 엔진 - lotto-lab 고도화 [몬테카를로 시뮬레이션 흐름] 1. 역대 당첨번호 기반 통계 캐시 구성 (build_analysis_cache) 2. 통계 가중치로 N개 후보 조합 생성 (weighted sampling) 3. 5가지 기법으로 각 후보 스코어링 (score_combination) 4. 상위 top_k개 선별하여 DB 저장 (simulation_candidates, best_picks 교체) [시뮬레이션 파라미터] - n_candidates: 1회 시뮬레이션당 생성 후보 수 (기본 20,000) - top_k: 선별 및 저장할 상위 개수 (기본 100) - best_n: best_picks에 올릴 최상위 개수 (기본 20) """ import random from typing import Dict, Any, List, Optional from .db import ( get_latest_draw, get_all_draw_numbers, save_simulation_run, save_simulation_candidates_bulk, replace_best_picks, ) from .analyzer import build_analysis_cache, build_number_weights, score_combination from .utils import weighted_sample_6 def run_simulation( n_candidates: int = 20000, top_k: int = 100, best_n: int = 20, ) -> Dict[str, Any]: """ 몬테카를로 시뮬레이션 실행 메인 함수. Args: n_candidates: 생성할 후보 조합 수 (기본 20,000) top_k: DB에 저장할 상위 후보 수 (기본 100) best_n: best_picks에 올릴 최상위 수 (기본 20) Returns: {run_id, total_generated, top_k_selected, avg_score, best_score, based_on_draw} 또는 {"error": ...} """ draws = get_all_draw_numbers() if not draws: return {"error": "당첨번호 데이터가 없습니다. 먼저 동기화를 실행하세요."} latest = get_latest_draw() based_on_draw = latest["drw_no"] if latest else None # ── 1. 통계 캐시 및 가중치 구성 (시뮬레이션 전체에서 재사용) ──────────── cache = build_analysis_cache(draws) weights = build_number_weights(cache) # ── 2. 후보 생성 및 스코어링 ────────────────────────────────────────────── candidates: List[Dict[str, Any]] = [] seen_keys: set = set() max_attempts = n_candidates * 3 # 중복 제거 여유분 attempts = 0 while len(candidates) < n_candidates and attempts < max_attempts: attempts += 1 nums = weighted_sample_6(weights) key = tuple(sorted(nums)) if key in seen_keys: continue seen_keys.add(key) scores = score_combination(nums, cache) candidates.append({ "numbers": sorted(nums), **scores, }) # ── 3. 점수 내림차순 정렬 및 상위 선별 ────────────────────────────────── candidates.sort(key=lambda x: -x["score_total"]) top_candidates = candidates[:top_k] # is_best 플래그 표시 best_keys = {tuple(c["numbers"]) for c in top_candidates[:best_n]} for c in top_candidates: c["is_best"] = tuple(c["numbers"]) in best_keys avg_score = ( sum(c["score_total"] for c in top_candidates) / len(top_candidates) if top_candidates else 0.0 ) best_score = top_candidates[0]["score_total"] if top_candidates else 0.0 # ── 4. DB 저장 ──────────────────────────────────────────────────────────── run_id = save_simulation_run( strategy="monte_carlo", total_generated=len(candidates), top_k_selected=len(top_candidates), avg_score=avg_score, notes=f"based_on_draw={based_on_draw}, history={len(draws)}회", ) # 상위 top_k개만 DB에 저장 (전체 20,000개는 메모리에서만 처리) save_simulation_candidates_bulk(run_id, top_candidates, based_on_draw) # best_picks 교체 (상위 best_n개) best_picks_data = [ { "numbers": c["numbers"], "score_total": c["score_total"], "rank_in_run": i + 1, } for i, c in enumerate(top_candidates[:best_n]) ] replace_best_picks(best_picks_data, run_id, based_on_draw) return { "run_id": run_id, "total_generated": len(candidates), "top_k_selected": len(top_candidates), "best_n_saved": len(best_picks_data), "avg_score": round(avg_score, 6), "best_score": round(best_score, 6), "based_on_draw": based_on_draw, } def generate_smart_recommendations(count: int = 10) -> int: """ 하위 호환성 유지용 래퍼. 내부적으로 run_simulation을 호출하며, 기존 /api/admin/auto_gen 등에서 계속 사용 가능. """ result = run_simulation(n_candidates=5000, top_k=count, best_n=count) if "error" in result: return 0 return result.get("best_n_saved", 0)