From 706ca410ca04b94a12f5bfdc6f08408295f31e0c Mon Sep 17 00:00:00 2001 From: gahusb Date: Mon, 6 Apr 2026 21:12:43 +0900 Subject: [PATCH] =?UTF-8?q?feat(lotto-lab):=20purchase=5Fmanager=20?= =?UTF-8?q?=E2=80=94=20=EA=B5=AC=EB=A7=A4=20=EA=B2=B0=EA=B3=BC=20=EC=9E=90?= =?UTF-8?q?=EB=8F=99=20=EC=B2=B4=ED=81=AC=20+=20=EC=A0=84=EB=9E=B5=20?= =?UTF-8?q?=EC=84=B1=EA=B3=BC=20=EC=A7=91=EA=B3=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- backend/app/purchase_manager.py | 90 ++++++++++++++++++++ backend/tests/test_purchase_manager.py | 111 +++++++++++++++++++++++++ 2 files changed, 201 insertions(+) create mode 100644 backend/app/purchase_manager.py diff --git a/backend/app/purchase_manager.py b/backend/app/purchase_manager.py new file mode 100644 index 0000000..5db6dce --- /dev/null +++ b/backend/app/purchase_manager.py @@ -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 diff --git a/backend/tests/test_purchase_manager.py b/backend/tests/test_purchase_manager.py index a044d87..fa29e7f 100644 --- a/backend/tests/test_purchase_manager.py +++ b/backend/tests/test_purchase_manager.py @@ -1,6 +1,8 @@ # backend/tests/test_purchase_manager.py import sys, os sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app")) +# Also insert the backend root so that "backend.app" package is importable +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) import sqlite3 import pytest @@ -196,3 +198,112 @@ def test_update_strategy_weight(): assert combined_after["total_sets"] == 100 assert combined_after["total_hits_3plus"] == 20 mem.close() + + +# ── purchase_manager 테스트 ─────────────────────────────────────────────────── + +def _import_purchase_manager_with_mem(mem_conn): + """purchase_manager를 메모리 DB에 연결된 상태로 임포트.""" + import db + import importlib + # backend.app 패키지로 로드해 상대 임포트가 동작하게 함 + import backend.app.purchase_manager as pm + return pm + + +def test_check_purchases_for_draw(): + """특정 회차 구매 건들의 결과 체크""" + import db + import backend.app.purchase_manager as pm + + mem = _make_mem_conn() + with patch("db._conn", return_value=mem): + db.init_db() + # 당첨번호 삽입: 1125회 [3,12,23,34,38,45] bonus=7 + mem.execute( + """INSERT INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (1125, "2024-12-01", 3, 12, 23, 34, 38, 45, 7), + ) + mem.commit() + + # 구매 등록: 1등 번호 세트 + 낙첨 세트 + purchase = db.add_purchase( + draw_no=1125, + amount=2000, + sets=2, + numbers=[[3, 12, 23, 34, 38, 45], [1, 2, 3, 4, 5, 6]], + is_real=False, + source_strategy="simulation", + ) + + with patch("db._conn", return_value=mem), \ + patch("backend.app.purchase_manager.get_draw", side_effect=lambda drw: db.get_draw(drw)), \ + patch("backend.app.purchase_manager.get_purchases", side_effect=lambda **kw: db.get_purchases(**kw)), \ + patch("backend.app.purchase_manager.update_purchase_results", side_effect=lambda *a, **kw: db.update_purchase_results(*a, **kw)), \ + patch("backend.app.purchase_manager.upsert_strategy_performance", side_effect=lambda **kw: db.upsert_strategy_performance(**kw)): + count = pm.check_purchases_for_draw(1125) + + assert count == 1 + + # 결과 확인 + with patch("db._conn", return_value=mem): + checked = db.get_purchases(draw_no=1125, checked=True) + assert len(checked) == 1 + results = checked[0]["results"] + assert results is not None + assert len(results) == 2 + # 첫 번째 세트: 6개 일치 → 1등 + assert results[0]["rank"] == 1 + assert results[0]["correct"] == 6 + # 두 번째 세트: 3 하나만 일치 → 낙첨(correct=1) + assert results[1]["rank"] == 0 + assert results[1]["correct"] == 1 + + mem.close() + + +def test_check_purchases_updates_strategy_performance(): + """결과 체크 후 strategy_performance가 갱신되는지 검증""" + import db + import backend.app.purchase_manager as pm + + mem = _make_mem_conn() + with patch("db._conn", return_value=mem): + db.init_db() + # 당첨번호 삽입: 1126회 + mem.execute( + """INSERT INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (1126, "2024-12-08", 1, 2, 3, 4, 5, 6, 7), + ) + mem.commit() + + db.add_purchase( + draw_no=1126, + amount=5000, + sets=5, + numbers=[[1, 2, 3, 4, 5, 6], [10, 20, 30, 40, 41, 42]], + is_real=False, + source_strategy="simulation", + ) + + with patch("db._conn", return_value=mem), \ + patch("backend.app.purchase_manager.get_draw", side_effect=lambda drw: db.get_draw(drw)), \ + patch("backend.app.purchase_manager.get_purchases", side_effect=lambda **kw: db.get_purchases(**kw)), \ + patch("backend.app.purchase_manager.update_purchase_results", side_effect=lambda *a, **kw: db.update_purchase_results(*a, **kw)), \ + patch("backend.app.purchase_manager.upsert_strategy_performance", side_effect=lambda **kw: db.upsert_strategy_performance(**kw)): + count = pm.check_purchases_for_draw(1126) + + assert count == 1 + + with patch("db._conn", return_value=mem): + perf = db.get_strategy_performance(strategy="simulation") + + assert len(perf) >= 1 + entry = next((p for p in perf if p["draw_no"] == 1126), None) + assert entry is not None, "draw_no=1126 에 대한 strategy_performance 없음" + assert entry["strategy"] == "simulation" + assert entry["sets_count"] == 2 # 2개 세트 + + mem.close()