473 lines
14 KiB
Python
473 lines
14 KiB
Python
import os
|
|
from typing import Optional, List, Dict, Any, Tuple
|
|
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
|
|
from .db import (
|
|
init_db, get_draw, get_latest_draw, get_all_draw_numbers,
|
|
save_recommendation_dedup, list_recommendations_ex, delete_recommendation,
|
|
update_recommendation,
|
|
)
|
|
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
|
|
from .checker import check_results_for_draw
|
|
from .utils import calc_metrics, calc_recent_overlap
|
|
|
|
app = FastAPI()
|
|
scheduler = BackgroundScheduler(timezone=os.getenv("TZ", "Asia/Seoul"))
|
|
|
|
ALL_URL = os.getenv("LOTTO_ALL_URL", "https://smok95.github.io/lotto/results/all.json")
|
|
LATEST_URL = os.getenv("LOTTO_LATEST_URL", "https://smok95.github.io/lotto/results/latest.json")
|
|
|
|
@app.on_event("startup")
|
|
def on_startup():
|
|
init_db()
|
|
|
|
# 1. 로또 당첨번호 동기화 (매일 9시, 21시 10분)
|
|
# 동기화 후 새로운 회차가 있으면 채점(check)까지 수행
|
|
def _sync_and_check():
|
|
res = sync_latest(LATEST_URL)
|
|
if res["was_new"]:
|
|
# 새로운 회차(예: 1000회)가 나오면, 999회차 기반 추천들을 채점
|
|
check_results_for_draw(res["drawNo"])
|
|
|
|
scheduler.add_job(_sync_and_check, "cron", hour="9,21", minute=10)
|
|
|
|
# 2. 매일 아침 8시: 지능형 자동 추천 (10개씩)
|
|
scheduler.add_job(lambda: generate_smart_recommendations(10), "cron", hour="8", minute=0)
|
|
|
|
scheduler.start()
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {"ok": True}
|
|
|
|
@app.get("/api/lotto/latest")
|
|
def api_latest():
|
|
row = get_latest_draw()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="No data yet")
|
|
return {
|
|
"drawNo": row["drw_no"],
|
|
"date": row["drw_date"],
|
|
"numbers": [row["n1"], row["n2"], row["n3"], row["n4"], row["n5"], row["n6"]],
|
|
"bonus": row["bonus"],
|
|
"metrics": calc_metrics([row["n1"], row["n2"], row["n3"], row["n4"], row["n5"], row["n6"]]),
|
|
}
|
|
|
|
@app.get("/api/lotto/{drw_no:int}")
|
|
def api_draw(drw_no: int):
|
|
row = get_draw(drw_no)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
return {
|
|
"drwNo": row["drw_no"],
|
|
"date": row["drw_date"],
|
|
"numbers": [row["n1"], row["n2"], row["n3"], row["n4"], row["n5"], row["n6"]],
|
|
"bonus": row["bonus"],
|
|
"metrics": calc_metrics([row["n1"], row["n2"], row["n3"], row["n4"], row["n5"], row["n6"]]),
|
|
}
|
|
|
|
@app.post("/api/admin/sync_latest")
|
|
def admin_sync_latest():
|
|
res = sync_latest(LATEST_URL)
|
|
# 수동 동기화 시에도 신규 회차면 채점
|
|
if res["was_new"]:
|
|
check_results_for_draw(res["drawNo"])
|
|
return res
|
|
|
|
@app.post("/api/admin/auto_gen")
|
|
def admin_auto_gen(count: int = 10):
|
|
"""지능형 자동 생성 수동 트리거"""
|
|
n = generate_smart_recommendations(count)
|
|
return {"generated": n}
|
|
|
|
@app.get("/api/lotto/stats")
|
|
def api_stats():
|
|
# 1. 데이터 완전성 보장 (없으면 가져옴)
|
|
sync_ensure_all(LATEST_URL, ALL_URL)
|
|
|
|
# 2. 전체 데이터 조회
|
|
draws = get_all_draw_numbers()
|
|
if not draws:
|
|
raise HTTPException(status_code=404, detail="No data yet")
|
|
|
|
# 1~45번 빈도 초기화
|
|
frequency = {n: 0 for n in range(1, 46)}
|
|
|
|
total_draws = len(draws)
|
|
|
|
for _, nums in draws:
|
|
for n in nums:
|
|
frequency[n] += 1
|
|
|
|
# 리스트 형태로 변환 (프론트엔드 차트용)
|
|
# x: 번호, y: 횟수
|
|
stats = [
|
|
{"number": n, "count": frequency[n]}
|
|
for n in range(1, 46)
|
|
]
|
|
|
|
return {
|
|
"total_draws": total_draws,
|
|
"frequency": stats
|
|
}
|
|
|
|
# ---------- ✅ recommend (dedup save) ----------
|
|
@app.get("/api/lotto/recommend")
|
|
def api_recommend(
|
|
recent_window: int = 200,
|
|
recent_weight: float = 2.0,
|
|
avoid_recent_k: int = 5,
|
|
|
|
# ---- optional constraints (Lotto Lab) ----
|
|
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, # 최근 avoid_recent_k 회차와 중복 허용 개수
|
|
max_try: int = 200, # 조건 맞는 조합 찾기 재시도
|
|
):
|
|
draws = get_all_draw_numbers()
|
|
if not draws:
|
|
raise HTTPException(status_code=404, detail="No data yet")
|
|
|
|
latest = get_latest_draw()
|
|
|
|
params = {
|
|
"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_numbers(
|
|
draws,
|
|
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}. "
|
|
f"Try relaxing sum/odd/range/overlap constraints.",
|
|
)
|
|
|
|
# ✅ 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,
|
|
}
|
|
|
|
# ---------- ✅ 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(
|
|
limit: int = 30,
|
|
offset: int = 0,
|
|
favorite: Optional[bool] = None,
|
|
tag: Optional[str] = None,
|
|
q: Optional[str] = None,
|
|
sort: str = "id_desc",
|
|
):
|
|
items = list_recommendations_ex(
|
|
limit=limit,
|
|
offset=offset,
|
|
favorite=favorite,
|
|
tag=tag,
|
|
q=q,
|
|
sort=sort,
|
|
)
|
|
|
|
draws = get_all_draw_numbers()
|
|
|
|
out = []
|
|
for it in items:
|
|
nums = it["numbers"]
|
|
out.append({
|
|
**it,
|
|
"metrics": calc_metrics(nums),
|
|
"recent_overlap": calc_recent_overlap(
|
|
nums, draws, last_k=int(it["params"].get("avoid_recent_k", 0) or 0)
|
|
),
|
|
})
|
|
|
|
return {
|
|
"items": out,
|
|
"limit": limit,
|
|
"offset": offset,
|
|
"filters": {"favorite": favorite, "tag": tag, "q": q, "sort": sort},
|
|
}
|
|
|
|
@app.delete("/api/history/{rec_id:int}")
|
|
def api_history_delete(rec_id: int):
|
|
ok = delete_recommendation(rec_id)
|
|
if not ok:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
return {"deleted": True, "id": rec_id}
|
|
|
|
# ---------- ✅ history update (favorite/note/tags) ----------
|
|
class HistoryUpdate(BaseModel):
|
|
favorite: Optional[bool] = None
|
|
note: Optional[str] = None
|
|
tags: Optional[List[str]] = None
|
|
|
|
@app.patch("/api/history/{rec_id:int}")
|
|
def api_history_patch(rec_id: int, body: HistoryUpdate):
|
|
ok = update_recommendation(rec_id, favorite=body.favorite, note=body.note, tags=body.tags)
|
|
if not ok:
|
|
raise HTTPException(status_code=404, detail="Not found or no changes")
|
|
return {"updated": True, "id": rec_id}
|
|
|
|
# ---------- ✅ batch recommend ----------
|
|
def _batch_unique(draws, count: int, recent_window: int, recent_weight: float, avoid_recent_k: int, max_try: int = 200):
|
|
items = []
|
|
seen = set()
|
|
|
|
tries = 0
|
|
while len(items) < count and tries < max_try:
|
|
tries += 1
|
|
r = recommend_numbers(draws, recent_window=recent_window, recent_weight=recent_weight, avoid_recent_k=avoid_recent_k)
|
|
key = tuple(sorted(r["numbers"]))
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
items.append(r)
|
|
|
|
return items
|
|
|
|
@app.get("/api/lotto/recommend/batch")
|
|
def api_recommend_batch(
|
|
count: int = 5,
|
|
recent_window: int = 200,
|
|
recent_weight: float = 2.0,
|
|
avoid_recent_k: int = 5,
|
|
):
|
|
count = max(1, min(count, 20))
|
|
draws = get_all_draw_numbers()
|
|
if not draws:
|
|
raise HTTPException(status_code=404, detail="No data yet")
|
|
|
|
latest = get_latest_draw()
|
|
params = {
|
|
"recent_window": recent_window,
|
|
"recent_weight": float(recent_weight),
|
|
"avoid_recent_k": avoid_recent_k,
|
|
"count": count,
|
|
}
|
|
|
|
items = _batch_unique(draws, count, recent_window, float(recent_weight), avoid_recent_k)
|
|
|
|
return {
|
|
"based_on_latest_draw": latest["drw_no"] if latest else None,
|
|
"count": count,
|
|
"items": [{
|
|
"numbers": it["numbers"],
|
|
"explain": it["explain"],
|
|
"metrics": calc_metrics(it["numbers"]),
|
|
} for it in items],
|
|
"params": params,
|
|
}
|
|
|
|
class BatchSave(BaseModel):
|
|
items: List[List[int]]
|
|
params: dict
|
|
|
|
@app.post("/api/lotto/recommend/batch")
|
|
def api_recommend_batch_save(body: BatchSave):
|
|
latest = get_latest_draw()
|
|
based = latest["drw_no"] if latest else None
|
|
|
|
created, deduped = [], []
|
|
for nums in body.items:
|
|
saved = save_recommendation_dedup(based, nums, body.params)
|
|
(created if saved["saved"] else deduped).append(saved["id"])
|
|
|
|
return {"saved": True, "created_ids": created, "deduped_ids": deduped}
|
|
|
|
@app.get("/api/version")
|
|
def version():
|
|
import os
|
|
return {"version": os.getenv("APP_VERSION", "dev")}
|