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

# lotto/app/weight_evolver.py
"""5종 시뮬 점수 가중치 자율 학습 루프.

순수 함수 (clamp/perturb/Dirichlet/score/base-rule) + DB 진입점은 별도 섹션.
"""
from __future__ import annotations
import math
import random
from typing import Any, Dict, List, Optional, Tuple

import numpy as np

MIN_WEIGHT = 0.05
N_METRICS = 5
DEFAULT_UNIFORM = [0.2] * N_METRICS  # cold start

RANK_BY_CORRECT = {6: 1, 5: 3, 4: 4, 3: 5}
RANK_BONUS = {1: 1.0, 2: 0.8, 3: 0.6, 4: 0.3, 5: 0.1}


def clamp_and_normalize(W: List[float], min_w: float = MIN_WEIGHT) -> List[float]:
    """각 값 ≥ min_w + 합=1.0. 보장 안 되면 raise."""
    if len(W) != N_METRICS:
        raise ValueError(f"W must have {N_METRICS} elements")
    # Iteratively clamp then normalize until all values satisfy min_w floor.
    # (Normalizing after clamping can reduce some already-floored values below
    #  min_w when the denominator is large — iterate to convergence.)
    vals = [float(w) for w in W]
    for _ in range(100):  # converges in a few iterations in practice
        clamped = [max(min_w, v) for v in vals]
        total = sum(clamped)
        vals = [v / total for v in clamped]
        if all(v >= min_w - 1e-12 for v in vals):
            break
    return vals


def perturb_weights(
    base: List[float],
    sigma: float = 0.05,
    seed: Optional[int] = None,
) -> List[float]:
    """base에 정규분포 noise(σ) 추가 → clamp+normalize."""
    if seed is not None:
        np.random.seed(seed)
    noise = np.random.normal(0, sigma, size=N_METRICS)
    perturbed = [b + n for b, n in zip(base, noise)]
    return clamp_and_normalize(perturbed)


def dirichlet_weights(
    alpha: float = 2.0,
    seed: Optional[int] = None,
) -> List[float]:
    """Dirichlet(α, α, α, α, α) 샘플 → clamp+normalize."""
    if seed is not None:
        np.random.seed(seed)
    sample = np.random.dirichlet([alpha] * N_METRICS).tolist()
    return clamp_and_normalize(sample)


def generate_weekly_candidates(
    base: Optional[List[float]] = None,
    seed: Optional[int] = None,
) -> List[Dict[str, Any]]:
    """6개 후보 — 4 perturb + 2 dirichlet. day_of_week 0..5 매핑.

    Returns:
        [{"day_of_week": 0, "weight": [...], "source": "perturb"}, ...]
    """
    if base is None:
        base = DEFAULT_UNIFORM[:]
    if seed is not None:
        np.random.seed(seed)

    trials = []
    for i in range(4):
        trials.append({
            "day_of_week": i,
            "weight": perturb_weights(base, sigma=0.05),
            "source": "perturb",
        })
    for i in range(4, 6):
        trials.append({
            "day_of_week": i,
            "weight": dirichlet_weights(alpha=2.0),
            "source": "dirichlet",
        })
    return trials


def count_match(pick: List[int], winning: List[int]) -> int:
    """본번호 6개 일치 개수. 보너스 제외."""
    return len(set(pick) & set(winning[:6]))


def calc_pick_score(pick_numbers: List[int], winning_numbers: List[int]) -> float:
    """correct/6 + RANK_BONUS. 보너스 번호 미고려."""
    correct = count_match(pick_numbers, winning_numbers)
    base = correct / 6.0
    rank = RANK_BY_CORRECT.get(correct)
    bonus = RANK_BONUS.get(rank, 0) if rank else 0
    return base + bonus


def decide_base_update(
    winner_max_correct: int,
    winner_W: List[float],
    current_base: Optional[List[float]],
) -> Tuple[List[float], str]:
    """Hybrid base update rule.

    Returns:
        (new_base, reason) — reason ∈ {'winner_4plus','ema_blend','unchanged','cold_start'}
    """
    if winner_max_correct >= 4:
        return list(winner_W), "winner_4plus"
    if winner_max_correct == 3 and current_base is not None:
        blended = [0.3 * w + 0.7 * c for w, c in zip(winner_W, current_base)]
        return clamp_and_normalize(blended), "ema_blend"
    if current_base is None:
        return DEFAULT_UNIFORM[:], "cold_start"
    return list(current_base), "unchanged"


# ---------- DB-touching entry points ----------

from datetime import datetime, timedelta, timezone


KST = timezone(timedelta(hours=9))


def _db():
    from . import db as _db_mod
    return _db_mod


def _today_kst():
    return datetime.now(KST).date()


def get_week_start(d=None) -> str:
    """주어진 날짜의 월요일 ISO 'YYYY-MM-DD'."""
    if d is None:
        d = _today_kst()
    ws = d - timedelta(days=d.weekday())
    return ws.isoformat()


def get_active_weight() -> Optional[List[float]]:
    """오늘 적용 중인 W. 없으면 None (균등 폴백)."""
    today = _today_kst()
    week_start = get_week_start(today)
    dow = today.weekday()
    if dow == 6:
        dow = 5  # 일요일은 토요일 W 유지
    trial = _db().get_weight_trial(week_start, dow)
    if trial:
        return trial["weight"]
    return None


def generate_weekly_candidates_and_save(seed: Optional[int] = None) -> List[Dict[str, Any]]:
    """월요일 09:00 cron 진입점. 6 trials 생성 후 DB 저장."""
    db = _db()
    base = db.get_current_base()
    if base is None:
        base = DEFAULT_UNIFORM[:]
        db.save_base_history(
            effective_from=get_week_start(),
            weight=base,
            source_trial_id=None,
            update_reason="cold_start",
            winner_score=None,
            winner_max_correct=None,
        )

    candidates = generate_weekly_candidates(base, seed=seed)
    week_start = get_week_start()
    for c in candidates:
        db.save_weight_trial(
            week_start=week_start,
            day_of_week=c["day_of_week"],
            weight=c["weight"],
            source=c["source"],
            base_at_gen=base,
        )
    return candidates


def apply_today_and_pick(n: int = 5) -> Dict[str, Any]:
    """매일 09:00 cron 진입점. 오늘 W로 N=5 세트 추출 후 auto_picks 저장."""
    db = _db()
    from . import analyzer, recommender
    today = _today_kst()
    week_start = get_week_start(today)
    dow = min(today.weekday(), 5)

    trial = db.get_weight_trial(week_start, dow)
    if trial is None:
        return {"ok": False, "reason": "no_trial_for_today"}

    W = trial["weight"]
    draws = db.get_all_draw_numbers()
    cache = analyzer.build_analysis_cache(draws)

    picks_saved = []
    for i in range(1, n + 1):
        try:
            r = recommender.recommend_numbers(draws)
            nums = r["numbers"]
            s = analyzer.score_combination(nums, cache, weights=W)
            pid = db.save_auto_pick(trial["id"], i, nums, meta_score=s["score_total"])
            picks_saved.append({"id": pid, "numbers": nums, "score": s["score_total"]})
        except Exception:
            continue

    return {
        "ok": True,
        "trial_id": trial["id"],
        "weight": W,
        "picks": picks_saved,
    }


def evaluate_weekly() -> Dict[str, Any]:
    """토 22:00 cron 진입점. 6일 trials × N picks 채점 + base 갱신."""
    db = _db()
    today = _today_kst()
    week_start = get_week_start(today)

    trials = db.get_weekly_trials(week_start)
    if not trials:
        return {"ok": False, "reason": "no_trials"}

    latest = db.get_latest_draw()
    if latest is None:
        return {"ok": False, "reason": "no_latest_draw"}
    winning = [
        latest["drw_num1"], latest["drw_num2"], latest["drw_num3"],
        latest["drw_num4"], latest["drw_num5"], latest["drw_num6"],
    ]

    per_day = []
    for trial in trials:
        picks = db.get_auto_picks(trial["id"])
        if not picks:
            continue
        day_scores = []
        max_c = 0
        for p in picks:
            correct = count_match(p["numbers"], winning)
            rank = RANK_BY_CORRECT.get(correct)
            db.update_auto_pick_grade(p["id"], correct, rank)
            day_scores.append(calc_pick_score(p["numbers"], winning))
            if correct > max_c:
                max_c = correct
        avg_score = sum(day_scores) / len(day_scores)
        per_day.append({
            "trial_id": trial["id"],
            "day_of_week": trial["day_of_week"],
            "weight": trial["weight"],
            "avg_score": avg_score,
            "max_correct": max_c,
            "n_picks": len(picks),
        })

    if not per_day:
        return {"ok": False, "reason": "no_picks_graded"}

    winner = max(per_day, key=lambda d: d["avg_score"])

    current_base = db.get_current_base()
    new_base, reason = decide_base_update(
        winner_max_correct=winner["max_correct"],
        winner_W=winner["weight"],
        current_base=current_base,
    )

    next_monday = today + timedelta(days=(7 - today.weekday()) % 7 or 7)
    db.save_base_history(
        effective_from=next_monday.isoformat(),
        weight=new_base,
        source_trial_id=winner["trial_id"],
        update_reason=reason,
        winner_score=winner["avg_score"],
        winner_max_correct=winner["max_correct"],
    )

    return {
        "ok": True,
        "draw_no": latest["drw_no"],
        "week_start": week_start,
        "winner": winner,
        "new_base": new_base,
        "update_reason": reason,
        "per_day": per_day,
    }