From 4a8b0092d763fd31df88dd6ced522917e8c0dc4a Mon Sep 17 00:00:00 2001 From: gahusb Date: Wed, 15 Apr 2026 08:20:51 +0900 Subject: [PATCH] =?UTF-8?q?feat(lotto):=20curator=5Fhelpers=20=E2=80=94=20?= =?UTF-8?q?=ED=9B=84=EB=B3=B4=20=EB=B3=91=ED=95=A9=C2=B7=ED=94=BC=EC=B2=98?= =?UTF-8?q?=C2=B7=EB=A7=A5=EB=9D=BD?= 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 | 131 ++++++++++++++++++++++++++++++++ backend/app/purchase_manager.py | 15 ++++ 2 files changed, 146 insertions(+) create mode 100644 backend/app/curator_helpers.py diff --git a/backend/app/curator_helpers.py b/backend/app/curator_helpers.py new file mode 100644 index 0000000..f3c5ec9 --- /dev/null +++ b/backend/app/curator_helpers.py @@ -0,0 +1,131 @@ +"""큐레이터용 후보 가공 — 여러 엔진 결과를 하나로 병합, 중복 제거, 피처 계산.""" +from typing import Dict, List, Any +from . import db +from .recommender import recommend_numbers, recommend_with_heatmap +from .analyzer import get_statistical_report + + +LOW_HIGH_CUT = 22 # 1~22 저구간, 23~45 고구간 + + +def compute_features(numbers: List[int], hot: set, cold: set) -> 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 + for n in nums: + if n <= 10: buckets[0] += 1 + elif n <= 20: buckets[1] += 1 + elif n <= 30: buckets[2] += 1 + elif n <= 40: buckets[3] += 1 + else: buckets[4] += 1 + consecutive = any(nums[i+1] - nums[i] == 1 for i in range(len(nums) - 1)) + return { + "odd_count": odd, + "even_count": 6 - odd, + "low_count": low, + "high_count": 6 - low, + "range_distribution": buckets, + "has_consecutive": consecutive, + "hot_number_count": len(set(nums) & hot), + "cold_number_count": len(set(nums) & cold), + "sum": sum(nums), + } + + +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세트 반환. + + 우선순위: simulation best_picks → meta → heatmap → statistics + """ + seen = {} + sources_order = [] + + # 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) + + # 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) + 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 + + # 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 + + out = [] + for k in sources_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]] + + 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: + from .purchase_manager import get_recent_performance + my_perf = get_recent_performance(limit=3) + except Exception: + my_perf = [] + + return { + "hot_numbers": hot, + "cold_numbers": cold, + "last_draw_summary": last_summary, + "my_recent_performance": my_perf, + } diff --git a/backend/app/purchase_manager.py b/backend/app/purchase_manager.py index feec529..3260fab 100644 --- a/backend/app/purchase_manager.py +++ b/backend/app/purchase_manager.py @@ -97,3 +97,18 @@ def check_purchases_for_draw(drw_no: int) -> int: logger.info(f"[purchase_manager] {drw_no}회차 구매 {count}건 체크 완료") return count + + +def get_recent_performance(limit: int = 3) -> list: + """최근 N회차 내 구매 성과 요약. 없으면 빈 리스트.""" + from . import db + purchases = db.get_purchases(days=None) 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)) + return sorted(by_draw.values(), key=lambda x: -x["draw_no"])[:limit]