852 lines
28 KiB
Python
852 lines
28 KiB
Python
import os
|
|
import time
|
|
import logging
|
|
from typing import Optional, List, Dict, Any, Tuple
|
|
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(levelname)s %(message)s")
|
|
logger = logging.getLogger("lotto-backend")
|
|
|
|
from .db import (
|
|
init_db, get_draw, get_latest_draw, get_all_draw_numbers,
|
|
save_recommendation_dedup, list_recommendations_ex, delete_recommendation,
|
|
update_recommendation,
|
|
# 시뮬레이션 관련
|
|
get_best_picks, get_simulation_runs, get_simulation_candidates,
|
|
# 성과 통계
|
|
get_recommendation_performance,
|
|
# Phase 2: 구매 이력
|
|
add_purchase, get_purchases, update_purchase, delete_purchase, get_purchase_stats,
|
|
# Phase 2: 주간 리포트 캐시
|
|
save_weekly_report, get_weekly_report_list, get_weekly_report,
|
|
# Phase 2: 개인 패턴 분석
|
|
get_all_recommendation_numbers,
|
|
# Phase 3: 전략 관련
|
|
get_strategy_performance as db_get_strategy_performance,
|
|
)
|
|
from .recommender import recommend_numbers, recommend_with_heatmap
|
|
from .collector import sync_latest, sync_ensure_all
|
|
from .generator import run_simulation, generate_smart_recommendations
|
|
from .checker import check_results_for_draw
|
|
from .utils import calc_metrics, calc_recent_overlap
|
|
from .analyzer import get_statistical_report, generate_weekly_report, analyze_personal_patterns, generate_combined_recommendation
|
|
from .purchase_manager import check_purchases_for_draw
|
|
from .strategy_evolver import (
|
|
get_weights_with_trend, recalculate_weights,
|
|
generate_smart_recommendation,
|
|
)
|
|
from .routers import curator as curator_router
|
|
from .routers import briefing as briefing_router
|
|
from .routers import review as review_router
|
|
from .jobs.grade_weekly_review import run_for_latest as grade_run_for_latest
|
|
|
|
app = FastAPI()
|
|
app.include_router(curator_router.router)
|
|
app.include_router(briefing_router.router)
|
|
app.include_router(review_router.router)
|
|
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")
|
|
|
|
# ── 성과 통계 인메모리 캐시 ───────────────────────────────────────────────────
|
|
# 채점 데이터는 하루 2번 스케줄러 실행 시에만 갱신되므로 인메모리 캐시로 충분
|
|
_PERF_CACHE: Dict[str, Any] = {"data": None, "at": 0.0}
|
|
_PERF_CACHE_TTL = 3600 # 1시간 (스케줄러 미실행 상황 대비 폴백)
|
|
|
|
|
|
def _refresh_perf_cache() -> None:
|
|
_PERF_CACHE["data"] = get_recommendation_performance()
|
|
_PERF_CACHE["at"] = time.time()
|
|
logger.info("성과 통계 캐시 갱신")
|
|
|
|
|
|
@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"]:
|
|
check_results_for_draw(res["drawNo"])
|
|
_refresh_perf_cache() # 새 채점 결과 반영 → 즉시 갱신
|
|
|
|
scheduler.add_job(_sync_and_check, "cron", hour="9,21", minute=10)
|
|
|
|
# 2. 몬테카를로 시뮬레이션 (하루 6회: 0, 4, 8, 12, 16, 20시)
|
|
# 20,000개 후보 생성 → 스코어링 → 상위 100개 저장 → best_picks 교체
|
|
def _run_simulation_job():
|
|
run_simulation(n_candidates=20000, top_k=100, best_n=20)
|
|
|
|
scheduler.add_job(_run_simulation_job, "cron", hour="0,4,8,12,16,20", minute=5)
|
|
|
|
# 3. 토요일 오전 9시 — 다음 회차 공략 리포트 자동 캐싱
|
|
def _save_weekly_report_job():
|
|
import json as _json
|
|
draws = get_all_draw_numbers()
|
|
latest = get_latest_draw()
|
|
if not draws or not latest:
|
|
return
|
|
target = latest["drw_no"] + 1
|
|
report = generate_weekly_report(draws, target)
|
|
save_weekly_report(target, _json.dumps(report, ensure_ascii=False))
|
|
logger.info(f"{target}회차 리포트 저장 완료")
|
|
|
|
scheduler.add_job(_save_weekly_report_job, "cron", day_of_week="sat", hour=9, minute=0)
|
|
|
|
# 4. 주간 채점 (매주 일요일 03:00 KST — 토요일 추첨 다음날 새벽)
|
|
# 당첨번호 sync 이후 추천 vs 실제 결과 비교 → reviews 테이블 저장
|
|
scheduler.add_job(
|
|
grade_run_for_latest,
|
|
"cron",
|
|
day_of_week="sun",
|
|
hour=3,
|
|
minute=0,
|
|
id="grade_weekly_review",
|
|
)
|
|
|
|
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.post("/api/admin/simulate")
|
|
def admin_simulate(n_candidates: int = 20000, top_k: int = 100, best_n: int = 20):
|
|
"""
|
|
몬테카를로 시뮬레이션 수동 트리거.
|
|
백그라운드 스케줄과 동일한 동작을 즉시 실행.
|
|
"""
|
|
result = run_simulation(
|
|
n_candidates=max(1000, min(n_candidates, 50000)),
|
|
top_k=max(10, min(top_k, 500)),
|
|
best_n=max(10, min(best_n, 50)),
|
|
)
|
|
if "error" in result:
|
|
raise HTTPException(status_code=500, detail=result["error"])
|
|
return result
|
|
|
|
|
|
@app.get("/api/lotto/stats")
|
|
def api_stats():
|
|
sync_ensure_all(LATEST_URL, ALL_URL)
|
|
|
|
draws = get_all_draw_numbers()
|
|
if not draws:
|
|
raise HTTPException(status_code=404, detail="No data yet")
|
|
|
|
frequency = {n: 0 for n in range(1, 46)}
|
|
total_draws = len(draws)
|
|
|
|
for _, nums in draws:
|
|
for n in nums:
|
|
frequency[n] += 1
|
|
|
|
stats = [
|
|
{"number": n, "count": frequency[n]}
|
|
for n in range(1, 46)
|
|
]
|
|
|
|
return {
|
|
"total_draws": total_draws,
|
|
"frequency": stats,
|
|
}
|
|
|
|
|
|
# ── 추천 성과 통계 (Phase 1, 인메모리 캐시) ──────────────────────────────────
|
|
@app.get("/api/lotto/stats/performance")
|
|
def api_performance_stats():
|
|
"""
|
|
채점된 추천 이력 기반 성과 통계 (캐시 반환).
|
|
캐시 갱신 시점: 새 회차 채점 직후 | TTL 1시간 만료 시
|
|
"""
|
|
if _PERF_CACHE["data"] is None or time.time() - _PERF_CACHE["at"] > _PERF_CACHE_TTL:
|
|
_refresh_perf_cache()
|
|
return _PERF_CACHE["data"]
|
|
|
|
|
|
# ── 회차 공략 리포트 (Phase 1) ────────────────────────────────────────────────
|
|
@app.get("/api/lotto/report/latest")
|
|
def api_report_latest():
|
|
"""
|
|
다음 회차 공략 리포트 (최신 회차 기준으로 자동 계산).
|
|
- 과출현/냉각/오버듀 번호 분석
|
|
- 최근 3회 패턴
|
|
- 3가지 전략별 추천 번호
|
|
- AI 신뢰도 점수
|
|
"""
|
|
draws = get_all_draw_numbers()
|
|
if not draws:
|
|
raise HTTPException(status_code=404, detail="No data yet")
|
|
latest = get_latest_draw()
|
|
target = latest["drw_no"] + 1
|
|
return generate_weekly_report(draws, target)
|
|
|
|
|
|
@app.get("/api/lotto/report/history")
|
|
def api_report_history(limit: int = 10):
|
|
"""저장된 주간 리포트 목록 (자동 저장된 것만)"""
|
|
return {"reports": get_weekly_report_list(limit=min(limit, 52))}
|
|
|
|
|
|
@app.get("/api/lotto/report/{drw_no}")
|
|
def api_report_by_draw(drw_no: int):
|
|
"""
|
|
특정 회차 공략 리포트 (캐시 우선, 없으면 실시간 생성).
|
|
"""
|
|
cached = get_weekly_report(drw_no)
|
|
if cached:
|
|
return {**cached, "cached": True}
|
|
|
|
draws = get_all_draw_numbers()
|
|
if not draws:
|
|
raise HTTPException(status_code=404, detail="No data yet")
|
|
base_draws = [(no, nums) for no, nums in draws if no < drw_no]
|
|
if not base_draws:
|
|
raise HTTPException(status_code=400, detail=f"{drw_no}회차 이전 데이터가 없습니다")
|
|
return {**generate_weekly_report(base_draws, drw_no), "cached": False}
|
|
|
|
|
|
# ── 개인 패턴 분석 (Phase 2) ─────────────────────────────────────────────────
|
|
@app.get("/api/lotto/analysis/personal")
|
|
def api_personal_analysis():
|
|
"""
|
|
저장된 추천 이력 기반 개인 패턴 분석.
|
|
- 자주 선택한 번호 TOP 10 / 한 번도 선택 안 한 번호
|
|
- 홀짝 비율, 합계, 범위, 연속번호 포함률
|
|
- 구간별 분포, 역대 당첨번호 평균과 비교
|
|
"""
|
|
all_numbers = get_all_recommendation_numbers()
|
|
draws = get_all_draw_numbers()
|
|
return analyze_personal_patterns(all_numbers, draws)
|
|
|
|
|
|
# ── 구매 이력 API (Phase 2) ───────────────────────────────────────────────────
|
|
|
|
class PurchaseCreate(BaseModel):
|
|
draw_no: int
|
|
amount: int
|
|
sets: int = 1
|
|
prize: int = 0
|
|
note: str = ""
|
|
numbers: List[List[int]] = []
|
|
is_real: bool = True
|
|
source_strategy: str = "manual"
|
|
source_detail: dict = {}
|
|
|
|
|
|
class PurchaseUpdate(BaseModel):
|
|
draw_no: Optional[int] = None
|
|
amount: Optional[int] = None
|
|
sets: Optional[int] = None
|
|
prize: Optional[int] = None
|
|
note: Optional[str] = None
|
|
numbers: Optional[List[List[int]]] = None
|
|
is_real: Optional[bool] = None
|
|
source_strategy: Optional[str] = None
|
|
|
|
|
|
@app.get("/api/lotto/purchase/stats")
|
|
def api_purchase_stats():
|
|
"""투자 수익률 통계 (총 투자금, 총 당첨금, 수익률 등)"""
|
|
return get_purchase_stats()
|
|
|
|
|
|
@app.get("/api/lotto/purchase")
|
|
def api_purchase_list(draw_no: Optional[int] = None, days: Optional[int] = None,
|
|
is_real: Optional[bool] = None, strategy: Optional[str] = None):
|
|
"""구매 이력 조회 (필터: draw_no, days, is_real, strategy)"""
|
|
return {"records": get_purchases(draw_no=draw_no, days=days, is_real=is_real, strategy=strategy)}
|
|
|
|
|
|
@app.post("/api/lotto/purchase", status_code=201)
|
|
def api_purchase_create(body: PurchaseCreate):
|
|
"""구매 이력 추가 (실제/가상)"""
|
|
sets = body.sets if body.sets > 0 else max(len(body.numbers), 1)
|
|
amount = body.amount if body.amount > 0 else sets * 1000
|
|
return add_purchase(
|
|
draw_no=body.draw_no,
|
|
amount=amount,
|
|
sets=sets,
|
|
prize=body.prize,
|
|
note=body.note,
|
|
numbers=body.numbers,
|
|
is_real=body.is_real,
|
|
source_strategy=body.source_strategy,
|
|
source_detail=body.source_detail,
|
|
)
|
|
|
|
|
|
@app.put("/api/lotto/purchase/{purchase_id}")
|
|
def api_purchase_update(purchase_id: int, body: PurchaseUpdate):
|
|
"""구매 이력 수정 (당첨금 업데이트 등)"""
|
|
updated = update_purchase(purchase_id, body.model_dump(exclude_none=True))
|
|
if updated is None:
|
|
raise HTTPException(status_code=404, detail="Purchase not found")
|
|
return updated
|
|
|
|
|
|
@app.delete("/api/lotto/purchase/{purchase_id}")
|
|
def api_purchase_delete(purchase_id: int):
|
|
"""구매 이력 삭제"""
|
|
if not delete_purchase(purchase_id):
|
|
raise HTTPException(status_code=404, detail="Purchase not found")
|
|
return {"ok": True}
|
|
|
|
|
|
# ── 전략 진화 API ──────────────────────────────────────────────────────────
|
|
|
|
@app.get("/api/lotto/strategy/weights")
|
|
def api_strategy_weights():
|
|
"""현재 전략별 가중치 + 성과 요약 + trend"""
|
|
return get_weights_with_trend()
|
|
|
|
|
|
@app.get("/api/lotto/strategy/performance")
|
|
def api_strategy_performance(strategy: Optional[str] = None, days: Optional[int] = None):
|
|
"""전략별 회차 성과 이력 (차트용)"""
|
|
rows = db_get_strategy_performance(strategy=strategy, days=days)
|
|
return {"records": rows}
|
|
|
|
|
|
@app.post("/api/lotto/strategy/evolve")
|
|
def api_strategy_evolve():
|
|
"""수동 가중치 재계산 트리거"""
|
|
new_weights = recalculate_weights()
|
|
return {"ok": True, "weights": new_weights}
|
|
|
|
|
|
# ── 스마트 추천 API ────────────────────────────────────────────────────────
|
|
|
|
@app.get("/api/lotto/recommend/smart")
|
|
def api_recommend_smart(sets: int = 5):
|
|
"""전략 가중치 기반 메타 전략 추천"""
|
|
sets = max(1, min(sets, 10))
|
|
result = generate_smart_recommendation(sets=sets)
|
|
if "error" in result:
|
|
raise HTTPException(status_code=500, detail=result["error"])
|
|
return result
|
|
|
|
|
|
# ── 통계 분석 리포트 ────────────────────────────────────────────────────────
|
|
@app.get("/api/lotto/analysis")
|
|
def api_analysis():
|
|
"""
|
|
5가지 통계 기법 기반 분석 리포트.
|
|
- 번호별 빈도, Z-score, 갭
|
|
- 핫/콜드/오버듀 번호
|
|
- 역대 합계 분포, 홀짝 분포
|
|
"""
|
|
draws = get_all_draw_numbers()
|
|
if not draws:
|
|
raise HTTPException(status_code=404, detail="No data yet")
|
|
return get_statistical_report(draws)
|
|
|
|
|
|
# ── 시뮬레이션 best_picks (메인 추천 엔드포인트) ────────────────────────────
|
|
@app.get("/api/lotto/best")
|
|
def api_best_picks(limit: int = 20):
|
|
"""
|
|
시뮬레이션을 통해 선별된 최적 번호 조합 반환 (기본 20쌍).
|
|
하루 6회 시뮬레이션 후 자동 갱신됨.
|
|
각 조합에 점수 및 메트릭 포함.
|
|
"""
|
|
limit = max(1, min(limit, 50))
|
|
picks = get_best_picks(limit=limit)
|
|
if not picks:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="시뮬레이션 결과가 없습니다. /api/admin/simulate로 먼저 실행하세요.",
|
|
)
|
|
|
|
draws = get_all_draw_numbers()
|
|
|
|
result = []
|
|
for p in picks:
|
|
nums = p["numbers"]
|
|
result.append({
|
|
"rank": p["rank_in_run"],
|
|
"numbers": nums,
|
|
"score_total": p["score_total"],
|
|
"based_on_draw": p["based_on_draw"],
|
|
"simulation_run_id": p["source_run_id"],
|
|
"created_at": p["created_at"],
|
|
"metrics": calc_metrics(nums),
|
|
})
|
|
|
|
latest = get_latest_draw()
|
|
return {
|
|
"based_on_draw": latest["drw_no"] if latest else None,
|
|
"count": len(result),
|
|
"items": result,
|
|
}
|
|
|
|
|
|
# ── 시뮬레이션 전체 결과 조회 (상세 API) ────────────────────────────────────
|
|
@app.get("/api/lotto/simulation")
|
|
def api_simulation(run_id: Optional[int] = None, runs_limit: int = 5):
|
|
"""
|
|
시뮬레이션 실행 기록 및 상위 후보 상세 조회.
|
|
run_id 미지정 시: 최근 runs_limit개 실행 기록 + 가장 최근 run의 후보 반환.
|
|
run_id 지정 시: 해당 run의 후보만 반환.
|
|
"""
|
|
runs = get_simulation_runs(limit=runs_limit)
|
|
if not runs:
|
|
raise HTTPException(status_code=404, detail="시뮬레이션 기록이 없습니다.")
|
|
|
|
target_run_id = run_id if run_id is not None else runs[0]["id"]
|
|
candidates = get_simulation_candidates(target_run_id, limit=100)
|
|
|
|
# 후보에 메트릭 추가
|
|
enriched = []
|
|
for c in candidates:
|
|
enriched.append({
|
|
**c,
|
|
"metrics": calc_metrics(c["numbers"]),
|
|
})
|
|
|
|
return {
|
|
"runs": runs,
|
|
"selected_run_id": target_run_id,
|
|
"candidates_count": len(enriched),
|
|
"candidates": enriched,
|
|
}
|
|
|
|
|
|
# ── 종합 추론 추천 ───────────────────────────────────────────────────────────
|
|
|
|
@app.get("/api/lotto/recommend/combined")
|
|
def api_recommend_combined():
|
|
"""5가지 통계 기법 종합 추론 추천 — 결과를 이력에 저장한다."""
|
|
draws = get_all_draw_numbers()
|
|
if not draws:
|
|
raise HTTPException(status_code=404, detail="No data")
|
|
|
|
latest = get_latest_draw()
|
|
result = generate_combined_recommendation(draws)
|
|
if "error" in result:
|
|
raise HTTPException(status_code=500, detail=result["error"])
|
|
|
|
# 추천 이력 저장 (태그: 종합추론)
|
|
params = {"method": "combined"}
|
|
saved = save_recommendation_dedup(
|
|
latest["drw_no"] if latest else None,
|
|
result["final_numbers"],
|
|
params,
|
|
)
|
|
if saved["saved"]:
|
|
update_recommendation(saved["id"], tags=["종합추론"])
|
|
|
|
return {
|
|
**result,
|
|
"id": saved["id"],
|
|
"saved": saved["saved"],
|
|
"deduped": saved["deduped"],
|
|
"based_on_latest_draw": latest["drw_no"] if latest else None,
|
|
}
|
|
|
|
|
|
@app.get("/api/lotto/recommend/combined/history")
|
|
def api_combined_history(limit: int = 30):
|
|
"""종합추론 추천 이력 조회."""
|
|
items = list_recommendations_ex(limit=limit, tag="종합추론", sort="id_desc")
|
|
return {"items": items, "total": len(items)}
|
|
|
|
|
|
# ── 기존 수동 추천 API (하위 호환 유지) ─────────────────────────────────────
|
|
@app.get("/api/lotto/recommend")
|
|
def api_recommend(
|
|
recent_window: int = 200,
|
|
recent_weight: float = 2.0,
|
|
avoid_recent_k: int = 5,
|
|
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")
|
|
|
|
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}.",
|
|
)
|
|
|
|
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,
|
|
}
|
|
|
|
|
|
# ── 히트맵 기반 추천 (하위 호환 유지) ────────────────────────────────────────
|
|
@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,
|
|
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}.",
|
|
)
|
|
|
|
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,
|
|
}
|
|
|
|
|
|
# ── 추천 이력 ────────────────────────────────────────────────────────────────
|
|
@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}
|
|
|
|
|
|
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}
|
|
|
|
|
|
# ── 배치 추천 (하위 호환 유지) ───────────────────────────────────────────────
|
|
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():
|
|
return {"version": os.getenv("APP_VERSION", "dev")}
|
|
|
|
|