From c9f10aca4afc680904bf311889c0c9ca059975ea Mon Sep 17 00:00:00 2001 From: gahusb Date: Mon, 6 Apr 2026 21:15:42 +0900 Subject: [PATCH] =?UTF-8?q?lotto-lab:=20strategy=5Fevolver=20=E2=80=94=20E?= =?UTF-8?q?MA/Softmax=20=EA=B0=80=EC=A4=91=EC=B9=98=20=EC=A7=84=ED=99=94?= =?UTF-8?q?=20+=20=EC=8A=A4=EB=A7=88=ED=8A=B8=20=EC=B6=94=EC=B2=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- backend/app/strategy_evolver.py | 277 +++++++++++++++++++++++++ backend/tests/test_strategy_evolver.py | 72 +++++++ 2 files changed, 349 insertions(+) create mode 100644 backend/app/strategy_evolver.py create mode 100644 backend/tests/test_strategy_evolver.py diff --git a/backend/app/strategy_evolver.py b/backend/app/strategy_evolver.py new file mode 100644 index 0000000..d69f046 --- /dev/null +++ b/backend/app/strategy_evolver.py @@ -0,0 +1,277 @@ +""" +전략 진화 엔진 — EMA + Softmax 기반 적응형 가중치 관리. +""" +import math +import json +import logging +from typing import Dict, List, Any + +logger = logging.getLogger("lotto-backend") + +# ── Constants (importable without DB) ───────────────────────────────────────── +ALPHA = 0.3 # EMA 감쇠율 +TEMPERATURE = 2.0 # Softmax 온도 +MIN_WEIGHT = 0.05 # 최소 가중치 +INITIAL_EMA = 0.15 # 콜드스타트 초기값 +MIN_DATA_DRAWS = 10 # 학습 최소 회차 + +STRATEGIES = ["combined", "simulation", "heatmap", "manual", "custom"] +RANK_BONUS = {5: 0.1, 4: 0.3, 3: 0.6, 2: 0.8, 1: 1.0} + + +# ── Pure functions (no DB dependency) ───────────────────────────────────────── + +def calc_draw_score(results: List[Dict]) -> float: + """구매 결과 리스트 → 평균 성과 점수""" + if not results: + return 0.0 + scores = [] + for r in results: + s = r.get("correct", 0) / 6.0 + s += RANK_BONUS.get(r.get("rank", 0), 0) + scores.append(s) + return sum(scores) / len(scores) + + +def _softmax_weights(ema_scores: Dict[str, float]) -> Dict[str, float]: + """EMA 점수 → Softmax → 최소 가중치 보장 → 정규화""" + raw = {s: math.exp(ema / TEMPERATURE) for s, ema in ema_scores.items()} + total = sum(raw.values()) + weights = {s: v / total for s, v in raw.items()} + + clamped = {} + surplus = 0.0 + unclamped = [] + for s, w in weights.items(): + if w < MIN_WEIGHT: + clamped[s] = MIN_WEIGHT + surplus += MIN_WEIGHT - w + else: + unclamped.append(s) + clamped[s] = w + + if surplus > 0 and unclamped: + unclamped_total = sum(clamped[s] for s in unclamped) + for s in unclamped: + clamped[s] -= surplus * (clamped[s] / unclamped_total) + + final_total = sum(clamped.values()) + return {s: round(v / final_total, 4) for s, v in clamped.items()} + + +# ── DB-dependent functions (use lazy imports) ───────────────────────────────── + +def _db(): + """Lazy import to avoid circular/relative import issues in tests""" + from . import db as _db_mod + return _db_mod + + +def _recommender(): + from . import recommender as _rec_mod + return _rec_mod + + +def _analyzer(): + from . import analyzer as _ana_mod + return _ana_mod + + +def update_ema_for_strategy(strategy: str, draw_score: float) -> float: + db = _db() + weights = db.get_strategy_weights() + current = next((w for w in weights if w["strategy"] == strategy), None) + old_ema = current["ema_score"] if current else INITIAL_EMA + new_ema = ALPHA * draw_score + (1 - ALPHA) * old_ema + return new_ema + + +def recalculate_weights() -> Dict[str, float]: + db = _db() + weights_rows = db.get_strategy_weights() + ema_scores = {w["strategy"]: w["ema_score"] for w in weights_rows} + + for s in STRATEGIES: + if s not in ema_scores: + ema_scores[s] = INITIAL_EMA + + new_weights = _softmax_weights(ema_scores) + + for s, w in new_weights.items(): + row = next((r for r in weights_rows if r["strategy"] == s), None) + db.update_strategy_weight( + strategy=s, + weight=w, + ema_score=ema_scores[s], + total_sets=row["total_sets"] if row else 0, + total_hits_3plus=row["total_hits_3plus"] if row else 0, + ) + + logger.info(f"[strategy_evolver] 가중치 재계산: {new_weights}") + return new_weights + + +def evolve_after_check(strategy: str, draw_no: int, results: List[Dict]) -> None: + db = _db() + draw_score = calc_draw_score(results) + new_ema = update_ema_for_strategy(strategy, draw_score) + + weights_rows = db.get_strategy_weights() + current = next((w for w in weights_rows if w["strategy"] == strategy), None) + hits_3plus = sum(1 for r in results if r.get("correct", 0) >= 3) + + db.update_strategy_weight( + strategy=strategy, + weight=current["weight"] if current else 0.2, + ema_score=new_ema, + total_sets=(current["total_sets"] if current else 0) + len(results), + total_hits_3plus=(current["total_hits_3plus"] if current else 0) + hits_3plus, + ) + + recalculate_weights() + + +def get_weights_with_trend() -> Dict[str, Any]: + db = _db() + weights = db.get_strategy_weights() + perfs = db.get_strategy_performance() + + strat_perfs = {} + for p in perfs: + s = p["strategy"] + if s not in strat_perfs: + strat_perfs[s] = [] + strat_perfs[s].append(p) + + result = [] + for w in weights: + sp = strat_perfs.get(w["strategy"], []) + if len(sp) >= 5: + recent_avg = sum(p["avg_score"] for p in sp[-3:]) / 3 + older_avg = sum(p["avg_score"] for p in sp[-5:-2]) / 3 + delta = recent_avg - older_avg + trend = "up" if delta > 0.02 else ("down" if delta < -0.02 else "stable") + else: + trend = "stable" + + result.append({ + "strategy": w["strategy"], + "weight": w["weight"], + "ema_score": w["ema_score"], + "total_sets": w["total_sets"], + "hits_3plus": w["total_hits_3plus"], + "trend": trend, + }) + + all_draws = set() + for p in perfs: + all_draws.add(p["draw_no"]) + + return { + "weights": result, + "last_evolved": weights[0]["updated_at"] if weights else None, + "min_data_draws": MIN_DATA_DRAWS, + "current_data_draws": len(all_draws), + "status": "active" if len(all_draws) >= MIN_DATA_DRAWS else "learning", + } + + +def generate_smart_recommendation(sets: int = 5) -> Dict[str, Any]: + db = _db() + rec = _recommender() + ana = _analyzer() + + weights_data = db.get_strategy_weights() + weight_map = {w["strategy"]: w["weight"] for w in weights_data} + draws = db.get_all_draw_numbers() + if not draws: + return {"error": "No draw data"} + + latest = db.get_latest_draw() + cache = ana.build_analysis_cache(draws) + past_recs = db.list_recommendations_ex(limit=100, sort="id_desc") + + candidates = [] + seen_keys = set() + + def _add_candidate(nums: list, strategy: str, raw_score: float = None): + key = tuple(sorted(nums)) + if key in seen_keys: + return + seen_keys.add(key) + if raw_score is None: + sc = ana.score_combination(nums, cache) + raw_score = sc["score_total"] + meta = raw_score * weight_map.get(strategy, 0.1) + candidates.append({ + "numbers": sorted(nums), + "raw_score": round(raw_score, 4), + "strategy": strategy, + "meta_score": round(meta, 4), + }) + + # combined: 10세트 + for _ in range(10): + try: + r = ana.generate_combined_recommendation(draws) + if "final_numbers" in r: + _add_candidate(r["final_numbers"], "combined") + except Exception: + pass + + # simulation: best_picks 상위 10개 + best = db.get_best_picks(limit=10) + for b in best: + nums = json.loads(b["numbers"]) if isinstance(b["numbers"], str) else b["numbers"] + _add_candidate(nums, "simulation", b.get("score_total")) + + # heatmap: 10세트 + for _ in range(10): + try: + r = rec.recommend_with_heatmap(draws, past_recs) + _add_candidate(r["numbers"], "heatmap") + except Exception: + pass + + # manual: 10세트 + for _ in range(10): + try: + r = rec.recommend_numbers(draws) + _add_candidate(r["numbers"], "manual") + except Exception: + pass + + candidates.sort(key=lambda c: -c["meta_score"]) + top = candidates[:sets] + + result_sets = [] + for c in top: + sc = ana.score_combination(c["numbers"], cache) + contributions = {} + for strat in STRATEGIES: + contributions[strat] = round(weight_map.get(strat, 0) * sc["score_total"], 4) + contrib_total = sum(contributions.values()) or 1 + contributions = {s: round(v / contrib_total, 3) for s, v in contributions.items()} + + result_sets.append({ + "numbers": c["numbers"], + "meta_score": c["meta_score"], + "source_strategy": c["strategy"], + "contribution": contributions, + "individual_scores": {k: round(v, 4) for k, v in sc.items()}, + }) + + perfs = db.get_strategy_performance() + data_draws = len(set(p["draw_no"] for p in perfs)) + status = "active" if data_draws >= MIN_DATA_DRAWS else "learning" + + return { + "sets": result_sets, + "strategy_weights_used": weight_map, + "learning_status": { + "draws_learned": data_draws, + "status": status, + "message": "" if status == "active" else f"{MIN_DATA_DRAWS}회차 이상 데이터 필요 (현재 {data_draws}회차)", + }, + "based_on_latest_draw": latest["drw_no"] if latest else None, + } diff --git a/backend/tests/test_strategy_evolver.py b/backend/tests/test_strategy_evolver.py new file mode 100644 index 0000000..e204642 --- /dev/null +++ b/backend/tests/test_strategy_evolver.py @@ -0,0 +1,72 @@ +import sys, os +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app")) + +import math +import pytest + + +def test_calc_draw_score_basic(): + """세트별 결과 → draw_score 계산""" + from strategy_evolver import calc_draw_score + + results = [ + {"correct": 3, "rank": 5}, # 3/6 + 0.1 = 0.6 + {"correct": 1, "rank": 0}, # 1/6 + 0 = 0.167 + ] + score = calc_draw_score(results) + expected = ((3/6 + 0.1) + (1/6)) / 2 + assert abs(score - expected) < 0.01 + + +def test_calc_draw_score_empty(): + """빈 결과 → 0""" + from strategy_evolver import calc_draw_score + assert calc_draw_score([]) == 0.0 + + +def test_recalculate_weights_softmax(): + """EMA → Softmax 가중치 변환""" + from strategy_evolver import _softmax_weights + + ema_scores = { + "combined": 0.30, + "simulation": 0.25, + "heatmap": 0.15, + "manual": 0.10, + "custom": 0.05, + } + weights = _softmax_weights(ema_scores) + + assert abs(sum(weights.values()) - 1.0) < 0.001 + assert weights["combined"] > weights["simulation"] + assert weights["simulation"] > weights["heatmap"] + assert all(w >= 0.049 for w in weights.values()) + + +def test_recalculate_weights_min_weight(): + """한 전략의 EMA가 매우 낮아도 최소 5% 보장""" + from strategy_evolver import _softmax_weights + + ema_scores = { + "combined": 0.50, + "simulation": 0.01, + "heatmap": 0.01, + "manual": 0.01, + "custom": 0.01, + } + weights = _softmax_weights(ema_scores) + + assert weights["simulation"] >= 0.049 + assert weights["custom"] >= 0.049 + assert abs(sum(weights.values()) - 1.0) < 0.001 + + +def test_update_ema(): + """EMA 갱신 공식 검증""" + from strategy_evolver import ALPHA + + old_ema = 0.15 + draw_score = 0.40 + new_ema = ALPHA * draw_score + (1 - ALPHA) * old_ema + expected = 0.3 * 0.40 + 0.7 * 0.15 # = 0.225 + assert abs(new_ema - expected) < 0.001