From d1fec71bdc24c8506cdcbcedbfc0242d34fb3721 Mon Sep 17 00:00:00 2001 From: gahusb Date: Wed, 15 Apr 2026 08:22:56 +0900 Subject: [PATCH] =?UTF-8?q?fix(lotto):=20curator=5Fhelpers=20=EC=8B=9C?= =?UTF-8?q?=EA=B7=B8=EB=8B=88=EC=B2=98=20=EC=A0=95=ED=95=A9=20(recommender?= =?UTF-8?q?/analyzer/strategy=5Fevolver=20=EC=8B=A4=EC=A0=9C=20=EC=8B=9C?= =?UTF-8?q?=EA=B7=B8=EB=8B=88=EC=B2=98=EC=97=90=20=EB=A7=9E=EC=B6=A4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- backend/app/curator_helpers.py | 142 ++++++++++++++++++-------------- backend/app/purchase_manager.py | 10 ++- 2 files changed, 87 insertions(+), 65 deletions(-) diff --git a/backend/app/curator_helpers.py b/backend/app/curator_helpers.py index f3c5ec9..f5f8198 100644 --- a/backend/app/curator_helpers.py +++ b/backend/app/curator_helpers.py @@ -1,18 +1,19 @@ """큐레이터용 후보 가공 — 여러 엔진 결과를 하나로 병합, 중복 제거, 피처 계산.""" -from typing import Dict, List, Any +from typing import Dict, List, Any, Set from . import db from .recommender import recommend_numbers, recommend_with_heatmap from .analyzer import get_statistical_report +from .strategy_evolver import generate_smart_recommendation -LOW_HIGH_CUT = 22 # 1~22 저구간, 23~45 고구간 +LOW_HIGH_CUT = 22 -def compute_features(numbers: List[int], hot: set, cold: set) -> Dict[str, Any]: +def compute_features(numbers: List[int], hot: Set[int], cold: Set[int]) -> Dict[str, Any]: nums = sorted(numbers) odd = sum(1 for n in nums if n % 2 == 1) low = sum(1 for n in nums if n <= LOW_HIGH_CUT) - buckets = [0, 0, 0, 0, 0] # 1-10, 11-20, 21-30, 31-40, 41-45 + buckets = [0, 0, 0, 0, 0] for n in nums: if n <= 10: buckets[0] += 1 elif n <= 20: buckets[1] += 1 @@ -37,86 +38,105 @@ def _key(numbers: List[int]) -> str: return ",".join(str(n) for n in sorted(numbers)) -def collect_candidates(n: int, hot: set, cold: set) -> List[Dict[str, Any]]: - """여러 엔진에서 후보를 모으고 중복을 제거. 최대 n세트 반환. +def collect_candidates(n: int, hot: Set[int], cold: Set[int]) -> List[Dict[str, Any]]: + """우선순위: simulation best_picks → meta → heatmap → statistics. 중복 제거 후 최대 n세트.""" + seen: Dict[str, Dict[str, Any]] = {} + order: List[str] = [] - 우선순위: simulation best_picks → meta → heatmap → statistics - """ - seen = {} - sources_order = [] + def _add(numbers: List[int], source: str) -> None: + if not numbers: + return + k = _key(numbers) + if k in seen: + return + seen[k] = {"numbers": sorted(numbers), "source": source} + order.append(k) # 1. simulation best_picks - for row in db.get_best_picks(limit=n): - numbers = row.get("numbers") or [] - if not numbers: - continue - k = _key(numbers) - if k not in seen: - seen[k] = {"numbers": sorted(numbers), "source": "simulation"} - sources_order.append(k) + try: + for row in db.get_best_picks(limit=n): + _add(row.get("numbers") or [], "simulation") + except Exception: + pass + + # draws는 한 번만 로드 + draws = [] + try: + draws = db.get_all_draw_numbers() + except Exception: + pass # 2. meta-strategy (smart) try: - from .generator import generate_smart_recommendation meta = generate_smart_recommendation(sets=n) for s in meta.get("sets", []): - numbers = s.get("numbers") or [] - k = _key(numbers) - if k not in seen and numbers: - seen[k] = {"numbers": sorted(numbers), "source": "meta"} - sources_order.append(k) + _add(s.get("numbers") or [], "meta") except Exception: pass - # 3. heatmap - try: - hm = recommend_with_heatmap(count=n) - for numbers in hm: - k = _key(numbers) - if k not in seen and numbers: - seen[k] = {"numbers": sorted(numbers), "source": "heatmap"} - sources_order.append(k) - except Exception: - pass + # 3. heatmap (n번 호출, 중복 회피) + if draws: + try: + for _ in range(n * 2): + if len(order) >= n * 2: + break + r = recommend_with_heatmap(draws, []) + _add(r.get("numbers") or [], "heatmap") + except Exception: + pass # 4. statistics - try: - st = recommend_numbers(count=n) - for numbers in st: - k = _key(numbers) - if k not in seen and numbers: - seen[k] = {"numbers": sorted(numbers), "source": "statistics"} - sources_order.append(k) - except Exception: - pass + if draws: + try: + for _ in range(n * 2): + if len(order) >= n * 2: + break + r = recommend_numbers(draws) + _add(r.get("numbers") or [], "statistics") + except Exception: + pass out = [] - for k in sources_order[:n]: + for k in order[:n]: item = seen[k] item["features"] = compute_features(item["numbers"], hot, cold) out.append(item) return out -def build_context(hot_limit: int = 3, cold_limit: int = 3) -> Dict[str, Any]: - """주간 맥락 패키지.""" - report = get_statistical_report() - latest = db.get_latest_draw() - freq = report.get("frequency", {}) - sorted_freq = sorted(freq.items(), key=lambda x: -x[1]) - hot = [int(k) for k, _ in sorted_freq[:hot_limit]] - - sorted_cold = sorted(freq.items(), key=lambda x: x[1]) - cold = [int(k) for k, _ in sorted_cold[:cold_limit]] - +def build_context(hot_limit: int = 10, cold_limit: int = 10) -> Dict[str, Any]: + """주간 맥락 패키지 — get_statistical_report가 이미 hot/cold를 제공.""" + hot: List[int] = [] + cold: List[int] = [] last_summary = "" - if latest: - nums = [latest.get(f"drwtNo{i}") for i in range(1, 7)] - odd = sum(1 for n in nums if n and n % 2 == 1) - low = sum(1 for n in nums if n and n <= LOW_HIGH_CUT) - last_summary = f"{latest['drwNo']}회: {', '.join(str(n) for n in nums)} (홀{odd}짝{6-odd}, 저{low}고{6-low})" - my_perf = [] + try: + draws = db.get_all_draw_numbers() + except Exception: + draws = [] + + if draws: + try: + report = get_statistical_report(draws) + hot = list(report.get("hot_numbers", []))[:hot_limit] + cold = list(report.get("cold_numbers", []))[:cold_limit] + except Exception: + pass + + try: + latest = db.get_latest_draw() + except Exception: + latest = None + + if latest: + nums = [latest.get(f"n{i}") for i in range(1, 7)] + nums = [n for n in nums if n is not None] + if nums: + odd = sum(1 for n in nums if n % 2 == 1) + low = sum(1 for n in nums if n <= LOW_HIGH_CUT) + last_summary = f"{latest.get('drw_no')}회: {', '.join(str(n) for n in nums)} (홀{odd}짝{6-odd}, 저{low}고{6-low})" + + my_perf: List[Dict[str, Any]] = [] try: from .purchase_manager import get_recent_performance my_perf = get_recent_performance(limit=3) diff --git a/backend/app/purchase_manager.py b/backend/app/purchase_manager.py index 3260fab..7f5554d 100644 --- a/backend/app/purchase_manager.py +++ b/backend/app/purchase_manager.py @@ -102,13 +102,15 @@ def check_purchases_for_draw(drw_no: int) -> int: def get_recent_performance(limit: int = 3) -> list: """최근 N회차 내 구매 성과 요약. 없으면 빈 리스트.""" from . import db - purchases = db.get_purchases(days=None) or [] + purchases = db.get_purchases() or [] by_draw: dict = {} for p in purchases: d = p.get("draw_no") if not d: continue - by_draw.setdefault(d, {"draw_no": d, "purchased_sets": 0, "best_match": 0}) - by_draw[d]["purchased_sets"] += int(p.get("sets") or 1) - by_draw[d]["best_match"] = max(by_draw[d]["best_match"], int(p.get("correct_count") or 0)) + results = p.get("results") or [] + max_correct = max((int(r.get("correct") or 0) for r in results), default=0) + slot = by_draw.setdefault(d, {"draw_no": d, "purchased_sets": 0, "best_match": 0}) + slot["purchased_sets"] += int(p.get("sets") or 1) + slot["best_match"] = max(slot["best_match"], max_correct) return sorted(by_draw.values(), key=lambda x: -x["draw_no"])[:limit]