390 lines
13 KiB
Python
390 lines
13 KiB
Python
# lotto/app/weight_evolver.py
|
||
"""5종 시뮬 점수 가중치 자율 학습 루프.
|
||
|
||
순수 함수 (clamp/perturb/Dirichlet/score/base-rule) + DB 진입점은 별도 섹션.
|
||
"""
|
||
from __future__ import annotations
|
||
import json
|
||
import math
|
||
import random
|
||
from datetime import datetime, timedelta, timezone
|
||
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}
|
||
|
||
LIFT_EPSILON = 10.0 # best-of-engine vs best-of-random margin;
|
||
# selection bias already cancelled by equal group sizes (N_NULL_TRIALS == engine trial count);
|
||
# tune as needed.
|
||
|
||
PRIZE_WEIGHTS = {"m6": 1000.0, "bonus_hits": 50.0, "m5": 30.0, "m4": 4.0, "m3": 1.0}
|
||
|
||
|
||
def select_winner_by_lift(per_w: List[Dict[str, Any]], random_score: float,
|
||
epsilon: float = LIFT_EPSILON) -> Dict[str, Any]:
|
||
"""engine_w 후보들 중 random 대비 lift 최대 선택.
|
||
최대 lift가 epsilon 미만이면 gated=True (노이즈 → base 유지 권고)."""
|
||
scored = [{**w, "lift": w["prize_score"] - random_score} for w in per_w]
|
||
best = max(scored, key=lambda w: w["lift"])
|
||
return {**best, "gated": best["lift"] < epsilon}
|
||
|
||
|
||
def prize_score_from_hist(hist: Dict[str, int]) -> float:
|
||
"""매칭 히스토그램 → 등수 가중 합산 점수.
|
||
1등=m6, 2등=bonus_hits, 3등=m5−bonus_hits, 4등=m4, 5등=m3.
|
||
m3/m4/m5/m6/bonus_hits 키만 읽으며 나머지는 무시하므로
|
||
DB 전체 행(backtest_runs row)을 그대로 넘겨도 안전하다."""
|
||
third = max(0, hist.get("m5", 0) - hist.get("bonus_hits", 0))
|
||
return (hist.get("m6", 0) * PRIZE_WEIGHTS["m6"]
|
||
+ hist.get("bonus_hits", 0) * PRIZE_WEIGHTS["bonus_hits"]
|
||
+ third * PRIZE_WEIGHTS["m5"]
|
||
+ hist.get("m4", 0) * PRIZE_WEIGHTS["m4"]
|
||
+ hist.get("m3", 0) * PRIZE_WEIGHTS["m3"])
|
||
|
||
|
||
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 ----------
|
||
|
||
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["n1"], latest["n2"], latest["n3"],
|
||
latest["n4"], latest["n5"], latest["n6"],
|
||
]
|
||
|
||
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"])
|
||
|
||
# 자가학습 강화: backtest forward 등수점수 lift로 winner 재선정.
|
||
# best-of-engine vs best-of-random 비교 — 동등 그룹 크기로 selection bias 상쇄.
|
||
latest_no = latest["drw_no"]
|
||
runs = db.get_backtest_runs(draw_no=latest_no)
|
||
engine_runs = [r for r in runs if r["strategy"] == "engine_w"]
|
||
null_runs = [r for r in runs if r["strategy"] == "random_null"]
|
||
gated = False # 이후 decide_base_update override에 사용
|
||
if engine_runs and null_runs:
|
||
# base 단독 행이 있고 w* 행도 있으면 base 행 제외 (identity collision 방지)
|
||
has_w_trials = any(r["weight_label"].startswith("w") for r in engine_runs)
|
||
if has_w_trials:
|
||
engine_runs = [r for r in engine_runs if r["weight_label"] != "base"]
|
||
|
||
# best-of-random: 동등 그룹의 최댓값 (selection bias 상쇄)
|
||
random_best = max(prize_score_from_hist(r) for r in null_runs)
|
||
|
||
per_w = []
|
||
for r in engine_runs:
|
||
per_w.append({
|
||
"trial_id": r["trial_id"],
|
||
"weight_label": r["weight_label"],
|
||
"weight": json.loads(r["weight_json"]) if r["weight_json"] else DEFAULT_UNIFORM[:],
|
||
"prize_score": prize_score_from_hist(r),
|
||
"best_match": r["best_match"],
|
||
})
|
||
|
||
lift_winner = select_winner_by_lift(per_w, random_score=random_best)
|
||
if not lift_winner["gated"]:
|
||
# lift winner의 정체성과 채점값을 일관되게 사용
|
||
winner = {
|
||
"trial_id": lift_winner["trial_id"],
|
||
"weight": lift_winner["weight"],
|
||
"max_correct": lift_winner["best_match"], # 이 trial의 실제값
|
||
"avg_score": lift_winner["prize_score"], # lift winner의 prize score
|
||
"lift": lift_winner["lift"],
|
||
}
|
||
else:
|
||
# 노이즈 → gated 플래그 설정; decide_base_update 이후 명시적으로 override
|
||
gated = True
|
||
winner = {**winner, "lift": lift_winner["lift"]}
|
||
|
||
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,
|
||
)
|
||
|
||
# gated path: decide_base_update 결과와 무관하게 base 유지 강제
|
||
if gated:
|
||
new_base = list(current_base) if current_base is not None else DEFAULT_UNIFORM[:]
|
||
reason = "unchanged_gated"
|
||
|
||
next_monday = today + timedelta(days=(7 - today.weekday()) % 7 or 7)
|
||
next_monday_iso = next_monday.isoformat()
|
||
|
||
# Idempotent guard: 같은 effective_from으로 이미 저장된 row가 있으면 skip
|
||
existing = db.get_base_history(limit=1)
|
||
if existing and existing[0]["effective_from"] == next_monday_iso:
|
||
return {
|
||
"ok": True,
|
||
"draw_no": latest["drw_no"],
|
||
"week_start": week_start,
|
||
"previous_base": existing[0].get("weight"),
|
||
"winner": winner,
|
||
"new_base": existing[0]["weight"], # 이미 저장된 값
|
||
"update_reason": existing[0].get("update_reason", "idempotent_skip"),
|
||
"per_day": per_day,
|
||
}
|
||
|
||
db.save_base_history(
|
||
effective_from=next_monday_iso,
|
||
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,
|
||
"previous_base": current_base, # save 이전에 캡처한 값 — diff 계산용
|
||
"winner": winner,
|
||
"new_base": new_base,
|
||
"update_reason": reason,
|
||
"per_day": per_day,
|
||
}
|