feat(lotto-lab): purchase_manager — 구매 결과 자동 체크 + 전략 성과 집계
- backend/app/purchase_manager.py 신규 생성 - check_purchases_for_draw(): 회차별 미채점 구매 건 자동 채점 - checker._calc_rank 재사용, RANK_PRIZE 상수 정의 - 채점 후 strategy_performance 자동 upsert (전략별 집계) - backend/tests/test_purchase_manager.py에 통합 테스트 2건 추가 - test_check_purchases_for_draw: 1등/낙첨 결과 검증 - test_check_purchases_updates_strategy_performance: 성과 테이블 갱신 검증 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
90
backend/app/purchase_manager.py
Normal file
90
backend/app/purchase_manager.py
Normal file
@@ -0,0 +1,90 @@
|
||||
"""
|
||||
구매 이력 관리 + 결과 체크 모듈.
|
||||
|
||||
- check_purchases_for_draw(): 특정 회차 구매 건들의 결과를 자동 체크
|
||||
- 체커의 _calc_rank 재사용
|
||||
- 결과 체크 후 strategy_performance 자동 갱신
|
||||
"""
|
||||
import logging
|
||||
from .db import (
|
||||
get_draw, get_purchases, update_purchase_results,
|
||||
upsert_strategy_performance,
|
||||
)
|
||||
from .checker import _calc_rank
|
||||
|
||||
logger = logging.getLogger("lotto-backend")
|
||||
|
||||
RANK_PRIZE = {1: 0, 2: 0, 3: 1_500_000, 4: 50_000, 5: 5_000}
|
||||
|
||||
|
||||
def check_purchases_for_draw(drw_no: int) -> int:
|
||||
"""
|
||||
특정 회차 결과로 해당 회차 구매 건들을 채점한다.
|
||||
Returns: 채점한 구매 건 수
|
||||
"""
|
||||
win_row = get_draw(drw_no)
|
||||
if not win_row:
|
||||
return 0
|
||||
|
||||
win_nums = [win_row["n1"], win_row["n2"], win_row["n3"],
|
||||
win_row["n4"], win_row["n5"], win_row["n6"]]
|
||||
bonus = win_row["bonus"]
|
||||
|
||||
unchecked = get_purchases(draw_no=drw_no, checked=False)
|
||||
|
||||
strategy_agg = {}
|
||||
|
||||
count = 0
|
||||
for purchase in unchecked:
|
||||
numbers_list = purchase["numbers"]
|
||||
if not numbers_list:
|
||||
continue
|
||||
|
||||
results = []
|
||||
for nums in numbers_list:
|
||||
rank, correct, has_bonus = _calc_rank(nums, win_nums, bonus)
|
||||
prize = RANK_PRIZE.get(rank, 0)
|
||||
results.append({
|
||||
"numbers": nums,
|
||||
"rank": rank,
|
||||
"correct": correct,
|
||||
"has_bonus": has_bonus,
|
||||
"prize": prize,
|
||||
})
|
||||
|
||||
total_prize = sum(r["prize"] for r in results)
|
||||
update_purchase_results(purchase["id"], results, total_prize)
|
||||
|
||||
strat = purchase["source_strategy"]
|
||||
if strat not in strategy_agg:
|
||||
strategy_agg[strat] = {
|
||||
"sets_count": 0,
|
||||
"total_correct": 0,
|
||||
"max_correct": 0,
|
||||
"prize_total": 0,
|
||||
"scores": [],
|
||||
}
|
||||
agg = strategy_agg[strat]
|
||||
for r in results:
|
||||
agg["sets_count"] += 1
|
||||
agg["total_correct"] += r["correct"]
|
||||
agg["max_correct"] = max(agg["max_correct"], r["correct"])
|
||||
agg["prize_total"] += r["prize"]
|
||||
agg["scores"].append(r["correct"] / 6.0)
|
||||
|
||||
count += 1
|
||||
|
||||
for strat, agg in strategy_agg.items():
|
||||
avg_score = sum(agg["scores"]) / len(agg["scores"]) if agg["scores"] else 0.0
|
||||
upsert_strategy_performance(
|
||||
strategy=strat,
|
||||
draw_no=drw_no,
|
||||
sets_count=agg["sets_count"],
|
||||
total_correct=agg["total_correct"],
|
||||
max_correct=agg["max_correct"],
|
||||
prize_total=agg["prize_total"],
|
||||
avg_score=round(avg_score, 4),
|
||||
)
|
||||
|
||||
logger.info(f"[purchase_manager] {drw_no}회차 구매 {count}건 체크 완료")
|
||||
return count
|
||||
Reference in New Issue
Block a user