web-page-backend/lotto/app/curator_helpers.py

"""큐레이터용 후보 가공 — 여러 엔진 결과를 하나로 병합, 중복 제거, 피처 계산."""
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


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]
    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[int], cold: Set[int]) -> List[Dict[str, Any]]:
    """우선순위: simulation best_picks → meta → heatmap → statistics. 중복 제거 후 최대 n세트."""
    seen: Dict[str, Dict[str, Any]] = {}
    order: List[str] = []

    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
    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:
        meta = generate_smart_recommendation(sets=n)
        for s in meta.get("sets", []):
            _add(s.get("numbers") or [], "meta")
    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
    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 order[:n]:
        item = seen[k]
        item["features"] = compute_features(item["numbers"], hot, cold)
        out.append(item)
    return out


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 = ""

    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)
    except Exception:
        my_perf = []

    return {
        "hot_numbers": hot,
        "cold_numbers": cold,
        "last_draw_summary": last_summary,
        "my_recent_performance": my_perf,
    }