stock-lab 오류 수정, lotto-lab 히트맵 기반 추천 기능 추가

2026-02-05 01:26:20 +09:00
parent 4035432c54
commit c96815c2e3
4 changed files with 219 additions and 128 deletions
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -9,7 +9,7 @@ from .db import (
    save_recommendation_dedup, list_recommendations_ex, delete_recommendation,
    update_recommendation,
 )
-from .recommender import recommend_numbers
+from .recommender import recommend_numbers, recommend_with_heatmap
 from .collector import sync_latest, sync_ensure_all
 from .generator import generate_smart_recommendations
 from .generator import generate_smart_recommendations
@@ -223,6 +223,124 @@ def api_recommend(
 		"tries": tries,
    }

+# ---------- ✅ heatmap-based recommend ----------
+@app.get("/api/lotto/recommend/heatmap")
+def api_recommend_heatmap(
+    heatmap_window: int = 20,
+    heatmap_weight: float = 1.5,
+    recent_window: int = 200,
+    recent_weight: float = 2.0,
+    avoid_recent_k: int = 5,
+    
+    # ---- optional constraints ----
+    sum_min: Optional[int] = None,
+    sum_max: Optional[int] = None,
+    odd_min: Optional[int] = None,
+    odd_max: Optional[int] = None,
+    range_min: Optional[int] = None,
+    range_max: Optional[int] = None,
+    max_overlap_latest: Optional[int] = None,
+    max_try: int = 200,
+):
+    """
+    히트맵 기반 추천: 과거 추천 번호들의 적중률을 분석하여 가중치 부여
+    """
+    draws = get_all_draw_numbers()
+    if not draws:
+        raise HTTPException(status_code=404, detail="No data yet")
+    
+    # 과거 추천 데이터 가져오기 (적중 결과가 있는 것만)
+    past_recs = list_recommendations_ex(limit=100, sort="id_desc")
+    
+    latest = get_latest_draw()
+    
+    params = {
+        "heatmap_window": heatmap_window,
+        "heatmap_weight": float(heatmap_weight),
+        "recent_window": recent_window,
+        "recent_weight": float(recent_weight),
+        "avoid_recent_k": avoid_recent_k,
+        "sum_min": sum_min,
+        "sum_max": sum_max,
+        "odd_min": odd_min,
+        "odd_max": odd_max,
+        "range_min": range_min,
+        "range_max": range_max,
+        "max_overlap_latest": max_overlap_latest,
+        "max_try": int(max_try),
+    }
+    
+    def _accept(nums: List[int]) -> bool:
+        m = calc_metrics(nums)
+        if sum_min is not None and m["sum"] < sum_min:
+            return False
+        if sum_max is not None and m["sum"] > sum_max:
+            return False
+        if odd_min is not None and m["odd"] < odd_min:
+            return False
+        if odd_max is not None and m["odd"] > odd_max:
+            return False
+        if range_min is not None and m["range"] < range_min:
+            return False
+        if range_max is not None and m["range"] > range_max:
+            return False
+        
+        if max_overlap_latest is not None:
+            ov = calc_recent_overlap(nums, draws, last_k=avoid_recent_k)
+            if ov["repeats"] > max_overlap_latest:
+                return False
+        return True
+    
+    chosen = None
+    explain = None
+    
+    tries = 0
+    while tries < max_try:
+        tries += 1
+        result = recommend_with_heatmap(
+            draws,
+            past_recs,
+            heatmap_window=heatmap_window,
+            heatmap_weight=heatmap_weight,
+            recent_window=recent_window,
+            recent_weight=recent_weight,
+            avoid_recent_k=avoid_recent_k,
+        )
+        nums = result["numbers"]
+        if _accept(nums):
+            chosen = nums
+            explain = result["explain"]
+            break
+    
+    if chosen is None:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Constraints too strict. No valid set found in max_try={max_try}.",
+        )
+    
+    # ✅ dedup save
+    saved = save_recommendation_dedup(
+        latest["drw_no"] if latest else None,
+        chosen,
+        params,
+    )
+    
+    metrics = calc_metrics(chosen)
+    overlap = calc_recent_overlap(chosen, draws, last_k=avoid_recent_k)
+    
+    return {
+        "id": saved["id"],
+        "saved": saved["saved"],
+        "deduped": saved["deduped"],
+        "based_on_latest_draw": latest["drw_no"] if latest else None,
+        "numbers": chosen,
+        "explain": explain,
+        "params": params,
+        "metrics": metrics,
+        "recent_overlap": overlap,
+        "tries": tries,
+    }
+
 # ---------- ✅ history list (filter/paging) ----------
@app.get("/api/history")
 def api_history(
--- a/backend/app/recommender.py
+++ b/backend/app/recommender.py
@@ -66,3 +66,98 @@ def recommend_numbers(

    return {"numbers": chosen_sorted, "explain": explain}

+
+def recommend_with_heatmap(
+    draws: List[Tuple[int, List[int]]],
+    past_recommendations: List[Dict[str, Any]],
+    *,
+    heatmap_window: int = 10,
+    heatmap_weight: float = 1.5,
+    recent_window: int = 200,
+    recent_weight: float = 2.0,
+    avoid_recent_k: int = 5,
+    seed: int | None = None,
+) -> Dict[str, Any]:
+    """
+    히트맵 기반 가중치 추천:
+    - 과거 추천 번호들의 적중률을 분석하여 잘 맞춘 번호에 가중치 부여
+    - 기존 통계 기반 추천과 결합
+    
+    Args:
+        draws: 실제 당첨 번호 리스트 [(회차, [번호들]), ...]
+        past_recommendations: 과거 추천 데이터 [{"numbers": [...], "correct_count": N, "based_on_draw": M}, ...]
+        heatmap_window: 히트맵 분석할 최근 추천 개수
+        heatmap_weight: 히트맵 가중치 (높을수록 과거 적중 번호 선호)
+    """
+    if seed is not None:
+        random.seed(seed)
+    
+    # 1. 기존 통계 기반 가중치 계산
+    all_nums = [n for _, nums in draws for n in nums]
+    freq_all = Counter(all_nums)
+    
+    recent = draws[-recent_window:] if len(draws) >= recent_window else draws
+    recent_nums = [n for _, nums in recent for n in nums]
+    freq_recent = Counter(recent_nums)
+    
+    last_k = draws[-avoid_recent_k:] if len(draws) >= avoid_recent_k else draws
+    last_k_nums = set(n for _, nums in last_k for n in nums)
+    
+    # 2. 히트맵 생성: 과거 추천에서 적중한 번호들의 빈도
+    heatmap = Counter()
+    recent_recs = past_recommendations[-heatmap_window:] if len(past_recommendations) >= heatmap_window else past_recommendations
+    
+    for rec in recent_recs:
+        if rec.get("correct_count", 0) > 0:  # 적중한 추천만
+            # 적중 개수에 비례해서 가중치 부여 (많이 맞춘 추천일수록 높은 가중)
+            weight = rec["correct_count"] ** 1.5  # 제곱으로 강조
+            for num in rec["numbers"]:
+                heatmap[num] += weight
+    
+    # 3. 최종 가중치 = 기존 통계 + 히트맵
+    weights = {}
+    for n in range(1, 46):
+        w = freq_all[n] + recent_weight * freq_recent[n]
+        
+        # 히트맵 가중치 추가
+        if n in heatmap:
+            w += heatmap_weight * heatmap[n]
+        
+        # 최근 출현 번호 패널티
+        if n in last_k_nums:
+            w *= 0.6
+        
+        weights[n] = max(w, 0.1)
+    
+    # 4. 가중 샘플링으로 6개 선택
+    chosen = []
+    pool = list(range(1, 46))
+    for _ in range(6):
+        total = sum(weights[n] for n in pool)
+        r = random.random() * total
+        acc = 0.0
+        for n in pool:
+            acc += weights[n]
+            if acc >= r:
+                chosen.append(n)
+                pool.remove(n)
+                break
+    
+    chosen_sorted = sorted(chosen)
+    
+    # 5. 설명 데이터
+    explain = {
+        "recent_window": recent_window,
+        "recent_weight": recent_weight,
+        "avoid_recent_k": avoid_recent_k,
+        "heatmap_window": heatmap_window,
+        "heatmap_weight": heatmap_weight,
+        "top_all": [n for n, _ in freq_all.most_common(10)],
+        "top_recent": [n for n, _ in freq_recent.most_common(10)],
+        "top_heatmap": [n for n, _ in heatmap.most_common(10)],
+        "last_k_draws": [d for d, _ in last_k],
+        "analyzed_recommendations": len(recent_recs),
+    }
+    
+    return {"numbers": chosen_sorted, "explain": explain}
+