fix(lotto): curator_helpers 시그니처 정합 (recommender/analyzer/strategy_evolver 실제 시그니처에 맞춤)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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]]:
+def collect_candidates(n: int, hot: Set[int], cold: Set[int]) -> List[Dict[str, Any]]:
-    """여러 엔진에서 후보를 모으고 중복을 제거. 최대 n세트 반환.
+    """우선순위: simulation best_picks → meta → heatmap → statistics. 중복 제거 후 최대 n세트."""
+    seen: Dict[str, Dict[str, Any]] = {}
+    order: List[str] = []
 
-    우선순위: simulation best_picks → meta → heatmap → statistics
+    def _add(numbers: List[int], source: str) -> None:
-    """
+        if not numbers:
-    seen = {}
+            return
-    sources_order = []
+        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):
+    try:
-        numbers = row.get("numbers") or []
+        for row in db.get_best_picks(limit=n):
-        if not numbers:
+            _add(row.get("numbers") or [], "simulation")
-            continue
+    except Exception:
-        k = _key(numbers)
+        pass
-        if k not in seen:
+
-            seen[k] = {"numbers": sorted(numbers), "source": "simulation"}
+    # draws는 한 번만 로드
-            sources_order.append(k)
+    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 []
+            _add(s.get("numbers") or [], "meta")
-            k = _key(numbers)
-            if k not in seen and numbers:
-                seen[k] = {"numbers": sorted(numbers), "source": "meta"}
-                sources_order.append(k)
     except Exception:
         pass
 
-    # 3. heatmap
+    # 3. heatmap (n번 호출, 중복 회피)
-    try:
+    if draws:
-        hm = recommend_with_heatmap(count=n)
+        try:
-        for numbers in hm:
+            for _ in range(n * 2):
-            k = _key(numbers)
+                if len(order) >= n * 2:
-            if k not in seen and numbers:
+                    break
-                seen[k] = {"numbers": sorted(numbers), "source": "heatmap"}
+                r = recommend_with_heatmap(draws, [])
-                sources_order.append(k)
+                _add(r.get("numbers") or [], "heatmap")
-    except Exception:
+        except Exception:
-        pass
+            pass
 
     # 4. statistics
-    try:
+    if draws:
-        st = recommend_numbers(count=n)
+        try:
-        for numbers in st:
+            for _ in range(n * 2):
-            k = _key(numbers)
+                if len(order) >= n * 2:
-            if k not in seen and numbers:
+                    break
-                seen[k] = {"numbers": sorted(numbers), "source": "statistics"}
+                r = recommend_numbers(draws)
-                sources_order.append(k)
+                _add(r.get("numbers") or [], "statistics")
-    except Exception:
+        except Exception:
-        pass
+            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]:
+def build_context(hot_limit: int = 10, cold_limit: int = 10) -> Dict[str, Any]:
-    """주간 맥락 패키지."""
+    """주간 맥락 패키지 — get_statistical_report가 이미 hot/cold를 제공."""
-    report = get_statistical_report()
+    hot: List[int] = []
-    latest = db.get_latest_draw()
+    cold: List[int] = []
-    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]]
-
     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)

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})
+        results = p.get("results") or []
-        by_draw[d]["purchased_sets"] += int(p.get("sets") or 1)
+        max_correct = max((int(r.get("correct") or 0) for r in results), default=0)
-        by_draw[d]["best_match"] = max(by_draw[d]["best_match"], int(p.get("correct_count") or 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]