feat: smart recommendation generator with feedback loop and result checker
This commit is contained in:
100
backend/app/generator.py
Normal file
100
backend/app/generator.py
Normal file
@@ -0,0 +1,100 @@
|
||||
import random
|
||||
import json
|
||||
from typing import Dict, Any, List, Optional
|
||||
|
||||
from .db import _conn, save_recommendation_dedup, get_latest_draw, get_all_draw_numbers
|
||||
from .recommender import recommend_numbers
|
||||
from .main import calc_metrics, calc_recent_overlap # main에 있는 헬퍼 재사용(순환참조 주의 필요 -> 사실 헬퍼는 utils로 빼는게 좋으나 일단 진행)
|
||||
|
||||
# 순환 참조 방지를 위해 main.py의 calc_metrics 등을 utils.py가 아닌 여기서 재정의하거나
|
||||
# main.py에서 generator를 import할 때 함수 내부에서 하도록 처리.
|
||||
# 여기서는 코드가 중복되더라도 안전하게 독립적으로 구현하거나, db/collector만 import.
|
||||
|
||||
def _get_top_performing_params(limit: int = 20) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
최근 1~5등에 당첨된 추천들의 파라미터 조회
|
||||
"""
|
||||
sql = """
|
||||
SELECT params
|
||||
FROM recommendations
|
||||
WHERE rank > 0 AND rank <= 5
|
||||
ORDER BY id DESC
|
||||
LIMIT ?
|
||||
"""
|
||||
with _conn() as conn:
|
||||
rows = conn.execute(sql, (limit,)).fetchall()
|
||||
|
||||
return [json.loads(r["params"]) for r in rows]
|
||||
|
||||
def _perturb_param(val: float, delta: float, min_val: float, max_val: float, is_int: bool = False) -> float:
|
||||
change = random.uniform(-delta, delta)
|
||||
new_val = val + change
|
||||
new_val = max(min_val, min(new_val, max_val))
|
||||
return int(round(new_val)) if is_int else round(new_val, 2)
|
||||
|
||||
def generate_smart_recommendations(count: int = 10) -> int:
|
||||
"""
|
||||
지능형 자동 생성: 과거 성적 우수 파라미터 기반으로 생성
|
||||
"""
|
||||
draws = get_all_draw_numbers()
|
||||
if not draws:
|
||||
return 0
|
||||
|
||||
latest = get_latest_draw()
|
||||
based_on = latest["drw_no"] if latest else None
|
||||
|
||||
# 1. 성공 사례 조회 (Feedback)
|
||||
top_params = _get_top_performing_params()
|
||||
|
||||
generated_count = 0
|
||||
|
||||
for _ in range(count):
|
||||
# 전략 선택: 이력이 있으면 70% 확률로 모방(Exploitation), 30%는 랜덤(Exploration)
|
||||
use_history = (len(top_params) > 0) and (random.random() < 0.7)
|
||||
|
||||
if use_history:
|
||||
# 과거 우수 파라미터 중 하나 선택하여 변형
|
||||
base = random.choice(top_params)
|
||||
|
||||
# 파라미터 변형 (유전 알고리즘과 유사)
|
||||
p_window = _perturb_param(base.get("recent_window", 200), 50, 10, 500, True)
|
||||
p_weight = _perturb_param(base.get("recent_weight", 2.0), 1.0, 0.1, 10.0, False)
|
||||
p_avoid = _perturb_param(base.get("avoid_recent_k", 5), 2, 0, 20, True)
|
||||
|
||||
# Constraints 로직은 복잡하니 일단 랜덤성 부여하거나 유지
|
||||
# (여기서는 기본 파라미터 위주로 튜닝)
|
||||
|
||||
params = {
|
||||
"recent_window": p_window,
|
||||
"recent_weight": p_weight,
|
||||
"avoid_recent_k": p_avoid,
|
||||
"strategy": "smart_feedback"
|
||||
}
|
||||
else:
|
||||
# 완전 랜덤 탐색
|
||||
params = {
|
||||
"recent_window": random.randint(50, 400),
|
||||
"recent_weight": round(random.uniform(0.5, 5.0), 2),
|
||||
"avoid_recent_k": random.randint(0, 10),
|
||||
"strategy": "random_exploration"
|
||||
}
|
||||
|
||||
# 생성 시도
|
||||
try:
|
||||
# recommend_numbers는 db.py/main.py 로직과 독립적이므로 여기서 사용 가능
|
||||
# 단, recommend_numbers 함수가 어디 있는지 확인 (recommender.py)
|
||||
res = recommend_numbers(
|
||||
draws,
|
||||
recent_window=params["recent_window"],
|
||||
recent_weight=params["recent_weight"],
|
||||
avoid_recent_k=params["avoid_recent_k"]
|
||||
)
|
||||
|
||||
save_recommendation_dedup(based_on, res["numbers"], params)
|
||||
generated_count += 1
|
||||
|
||||
except Exception as e:
|
||||
print(f"Gen Error: {e}")
|
||||
continue
|
||||
|
||||
return generated_count
|
||||
Reference in New Issue
Block a user