gahusb
706ca410ca
- 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>
91 lines
2.8 KiB
Python
91 lines
2.8 KiB
Python
"""
|
|
구매 이력 관리 + 결과 체크 모듈.
|
|
|
|
- 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
|