From 4c6e96d59cd3a29677dcc08123954f558f171740 Mon Sep 17 00:00:00 2001 From: gahusb Date: Mon, 6 Apr 2026 21:09:59 +0900 Subject: [PATCH] =?UTF-8?q?lotto-lab:=20=EA=B5=AC=EB=A7=A4=20CRUD=20?= =?UTF-8?q?=ED=99=95=EC=9E=A5=20+=20strategy=5Fperformance/weights=20CRUD?= =?UTF-8?q?=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _purchase_row_to_dict: numbers/is_real/source_detail/results/total_prize 신규 컬럼 포함 - add_purchase: numbers, is_real, source_strategy, source_detail 파라미터 추가 - get_purchases: is_real, strategy, checked 필터 추가 - get_purchase_stats: total/real/virtual/by_strategy 분리 통계 + 하위호환 필드 유지 - update_purchase: allowed 셋에 numbers/is_real/source_strategy 추가 - 신규: upsert_strategy_performance, get_strategy_performance - 신규: get_strategy_weights, update_strategy_weight - 신규: update_purchase_results (체커 연동용) - 테스트 5건 추가 (TDD) Co-Authored-By: Claude Sonnet 4.6 --- backend/app/db.py | 191 +++++++++++++++++++++---- backend/tests/test_purchase_manager.py | 129 +++++++++++++++++ 2 files changed, 295 insertions(+), 25 deletions(-) diff --git a/backend/app/db.py b/backend/app/db.py index 4285201..30d3ddb 100644 --- a/backend/app/db.py +++ b/backend/app/db.py @@ -810,6 +810,11 @@ def get_simulation_candidates(run_id: int, limit: int = 100) -> List[Dict[str, A # ── purchase_history CRUD ───────────────────────────────────────────────────── def _purchase_row_to_dict(r) -> Dict[str, Any]: + import json as _json + keys = r.keys() + numbers_raw = r["numbers"] if "numbers" in keys else "[]" + detail_raw = r["source_detail"] if "source_detail" in keys else "{}" + results_raw = r["results"] if "results" in keys else "[]" return { "id": r["id"], "draw_no": r["draw_no"], @@ -818,20 +823,37 @@ def _purchase_row_to_dict(r) -> Dict[str, Any]: "prize": r["prize"], "note": r["note"], "created_at": r["created_at"], + "numbers": _json.loads(numbers_raw) if numbers_raw else [], + "is_real": r["is_real"] if "is_real" in keys else 1, + "source_strategy": r["source_strategy"] if "source_strategy" in keys else "manual", + "source_detail": _json.loads(detail_raw) if detail_raw else {}, + "checked": r["checked"] if "checked" in keys else 0, + "results": _json.loads(results_raw) if results_raw else [], + "total_prize": r["total_prize"] if "total_prize" in keys else 0, } -def add_purchase(draw_no: int, amount: int, sets: int, prize: int = 0, note: str = "") -> Dict[str, Any]: +def add_purchase(draw_no: int, amount: int, sets: int, prize: int = 0, note: str = "", + numbers: list = None, is_real: bool = True, + source_strategy: str = "manual", source_detail: dict = None) -> Dict[str, Any]: + import json as _json + numbers_json = _json.dumps(numbers or [], ensure_ascii=False) + detail_json = _json.dumps(source_detail or {}, ensure_ascii=False) + is_real_int = 1 if is_real else 0 with _conn() as conn: conn.execute( - "INSERT INTO purchase_history (draw_no, amount, sets, prize, note) VALUES (?, ?, ?, ?, ?)", - (draw_no, amount, sets, prize, note), + """INSERT INTO purchase_history + (draw_no, amount, sets, prize, note, numbers, is_real, source_strategy, source_detail) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (draw_no, amount, sets, prize, note, numbers_json, is_real_int, source_strategy, detail_json), ) row = conn.execute("SELECT * FROM purchase_history WHERE rowid = last_insert_rowid()").fetchone() return _purchase_row_to_dict(row) -def get_purchases(draw_no: int = None, days: int = None) -> List[Dict[str, Any]]: +def get_purchases(draw_no: int = None, days: int = None, + is_real: bool = None, strategy: str = None, + checked: bool = None) -> List[Dict[str, Any]]: conditions, params = [], [] if draw_no is not None: conditions.append("draw_no = ?") @@ -839,6 +861,15 @@ def get_purchases(draw_no: int = None, days: int = None) -> List[Dict[str, Any]] if days: conditions.append("created_at >= datetime('now', ? || ' days')") params.append(f"-{days}") + if is_real is not None: + conditions.append("is_real = ?") + params.append(1 if is_real else 0) + if strategy is not None: + conditions.append("source_strategy = ?") + params.append(strategy) + if checked is not None: + conditions.append("checked = ?") + params.append(1 if checked else 0) where = f"WHERE {' AND '.join(conditions)}" if conditions else "" with _conn() as conn: rows = conn.execute( @@ -849,7 +880,7 @@ def get_purchases(draw_no: int = None, days: int = None) -> List[Dict[str, Any]] def update_purchase(purchase_id: int, data: Dict[str, Any]) -> Optional[Dict[str, Any]]: - allowed = {"draw_no", "amount", "sets", "prize", "note"} + allowed = {"draw_no", "amount", "sets", "prize", "note", "numbers", "is_real", "source_strategy"} updates = {k: v for k, v in data.items() if k in allowed} if not updates: with _conn() as conn: @@ -874,30 +905,74 @@ def delete_purchase(purchase_id: int) -> bool: def get_purchase_stats() -> Dict[str, Any]: - with _conn() as conn: - rows = conn.execute("SELECT amount, prize FROM purchase_history").fetchall() - if not rows: + import json as _json + + def _calc_group(rows): + if not rows: + return {"sets": 0, "invested": 0, "prize": 0, "roi": 0.0, "win_rate": 0.0} + invested = sum(r["amount"] for r in rows) + prize = sum(r.get("total_prize") or r["prize"] for r in rows) + wins = sum(1 for r in rows if (r.get("total_prize") or r["prize"]) > 0) return { - "total_records": 0, - "total_invested": 0, - "total_prize": 0, - "net": 0, - "return_rate": 0.0, - "prize_count": 0, - "max_prize": 0, + "sets": sum(r["sets"] for r in rows), + "invested": invested, + "prize": prize, + "roi": round((prize / invested * 100 - 100) if invested else 0.0, 2), + "win_rate": round(wins / len(rows) * 100, 2) if rows else 0.0, } - amounts = [r["amount"] for r in rows] - prizes = [r["prize"] for r in rows] - total_invested = sum(amounts) - total_prize = sum(prizes) + + with _conn() as conn: + rows = conn.execute("SELECT * FROM purchase_history").fetchall() + + all_rows = [dict(r) for r in rows] + real_rows = [r for r in all_rows if r.get("is_real", 1) == 1] + virtual_rows = [r for r in all_rows if r.get("is_real", 1) == 0] + + # 전략별 집계 + by_strategy: Dict[str, list] = {} + for r in all_rows: + strat = r.get("source_strategy", "manual") + if strat not in by_strategy: + by_strategy[strat] = [] + by_strategy[strat].append(r) + + strategy_stats: Dict[str, Any] = {} + for strat, srows in by_strategy.items(): + s = _calc_group(srows) + total_correct = 0 + count_sets = 0 + hits_3plus = 0 + for r in srows: + results_raw = r.get("results", "[]") + try: + results = _json.loads(results_raw) if isinstance(results_raw, str) else (results_raw or []) + except Exception: + results = [] + for res in results: + count_sets += 1 + c = res.get("correct", 0) + total_correct += c + if c >= 3: + hits_3plus += 1 + s["avg_correct"] = round(total_correct / count_sets, 2) if count_sets else 0.0 + s["hits_3plus"] = hits_3plus + strategy_stats[strat] = s + + total_invested = sum(r["amount"] for r in all_rows) + total_prize_sum = sum(r.get("total_prize") or r["prize"] for r in all_rows) return { - "total_records": len(rows), + "total": _calc_group(all_rows), + "real": _calc_group(real_rows), + "virtual": _calc_group(virtual_rows), + "by_strategy": strategy_stats, + # 하위호환 + "total_records": len(all_rows), "total_invested": total_invested, - "total_prize": total_prize, - "net": total_prize - total_invested, - "return_rate": round((total_prize / total_invested * 100) if total_invested else 0.0, 2), - "prize_count": sum(1 for p in prizes if p > 0), - "max_prize": max(prizes), + "total_prize": total_prize_sum, + "net": total_prize_sum - total_invested, + "return_rate": round((total_prize_sum / total_invested * 100) if total_invested else 0.0, 2), + "prize_count": sum(1 for r in all_rows if (r.get("total_prize") or r["prize"]) > 0), + "max_prize": max((r.get("total_prize") or r["prize"] for r in all_rows), default=0), } @@ -945,4 +1020,70 @@ def get_all_recommendation_numbers() -> List[List[int]]: return [json.loads(r["numbers"]) for r in rows] +# ── strategy_performance CRUD ───────────────────────────────────────────────── + +def upsert_strategy_performance(strategy: str, draw_no: int, sets_count: int = 0, + total_correct: int = 0, max_correct: int = 0, + prize_total: int = 0, avg_score: float = 0.0) -> None: + with _conn() as conn: + conn.execute( + """INSERT INTO strategy_performance (strategy, draw_no, sets_count, total_correct, max_correct, prize_total, avg_score) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(strategy, draw_no) DO UPDATE SET + sets_count=excluded.sets_count, total_correct=excluded.total_correct, + max_correct=excluded.max_correct, prize_total=excluded.prize_total, + avg_score=excluded.avg_score, + updated_at=strftime('%Y-%m-%dT%H:%M:%fZ','now')""", + (strategy, draw_no, sets_count, total_correct, max_correct, prize_total, avg_score), + ) + + +def get_strategy_performance(strategy: str = None, days: int = None) -> List[Dict[str, Any]]: + conditions, params = [], [] + if strategy: + conditions.append("strategy = ?") + params.append(strategy) + if days: + conditions.append("updated_at >= datetime('now', ? || ' days')") + params.append(f"-{days}") + where = f"WHERE {' AND '.join(conditions)}" if conditions else "" + with _conn() as conn: + rows = conn.execute( + f"SELECT * FROM strategy_performance {where} ORDER BY draw_no ASC", + params, + ).fetchall() + return [dict(r) for r in rows] + + +# ── strategy_weights CRUD ───────────────────────────────────────────────────── + +def get_strategy_weights() -> List[Dict[str, Any]]: + with _conn() as conn: + rows = conn.execute("SELECT * FROM strategy_weights ORDER BY weight DESC").fetchall() + return [dict(r) for r in rows] + + +def update_strategy_weight(strategy: str, weight: float, ema_score: float, + total_sets: int = None, total_hits_3plus: int = None) -> None: + with _conn() as conn: + fields = "weight=?, ema_score=?, updated_at=strftime('%Y-%m-%dT%H:%M:%fZ','now')" + params = [weight, ema_score] + if total_sets is not None: + fields += ", total_sets=?" + params.append(total_sets) + if total_hits_3plus is not None: + fields += ", total_hits_3plus=?" + params.append(total_hits_3plus) + params.append(strategy) + conn.execute(f"UPDATE strategy_weights SET {fields} WHERE strategy=?", params) + + +def update_purchase_results(purchase_id: int, results: list, total_prize: int) -> None: + """구매 건의 결과를 갱신 (체커 호출 후)""" + import json as _json + with _conn() as conn: + conn.execute( + "UPDATE purchase_history SET results=?, total_prize=?, checked=1 WHERE id=?", + (_json.dumps(results, ensure_ascii=False), total_prize, purchase_id), + ) diff --git a/backend/tests/test_purchase_manager.py b/backend/tests/test_purchase_manager.py index 4f45db3..a044d87 100644 --- a/backend/tests/test_purchase_manager.py +++ b/backend/tests/test_purchase_manager.py @@ -67,3 +67,132 @@ def test_strategy_weights_table_exists(): total_weight = sum(r["weight"] for r in rows) assert abs(total_weight - 1.0) < 0.01 mem.close() + + +def test_add_purchase_with_numbers(): + """번호 포함 구매 등록""" + import db + mem = _make_mem_conn() + with patch("db._conn", return_value=mem): + db.init_db() + result = db.add_purchase( + draw_no=1150, + amount=5000, + sets=5, + numbers=[[1, 2, 3, 4, 5, 6], [7, 8, 9, 10, 11, 12]], + is_real=False, + source_strategy="simulation", + source_detail={"run_id": 42}, + ) + assert result["draw_no"] == 1150 + assert result["amount"] == 5000 + assert result["is_real"] == 0 + assert result["source_strategy"] == "simulation" + assert result["numbers"] == [[1, 2, 3, 4, 5, 6], [7, 8, 9, 10, 11, 12]] + assert result["source_detail"] == {"run_id": 42} + mem.close() + + +def test_get_purchases_filter_is_real(): + """is_real 필터 동작""" + import db + mem = _make_mem_conn() + with patch("db._conn", return_value=mem): + db.init_db() + db.add_purchase(draw_no=1150, amount=5000, sets=5, is_real=True) + db.add_purchase(draw_no=1150, amount=1000, sets=1, is_real=False) + real_only = db.get_purchases(is_real=True) + virtual_only = db.get_purchases(is_real=False) + assert len(real_only) == 1 + assert real_only[0]["is_real"] == 1 + assert len(virtual_only) == 1 + assert virtual_only[0]["is_real"] == 0 + mem.close() + + +def test_get_purchase_stats_by_type(): + """실제/가상 분리 통계""" + import db + mem = _make_mem_conn() + with patch("db._conn", return_value=mem): + db.init_db() + db.add_purchase(draw_no=1150, amount=5000, sets=5, is_real=True, source_strategy="manual") + db.add_purchase(draw_no=1150, amount=1000, sets=1, is_real=False, source_strategy="simulation") + stats = db.get_purchase_stats() + assert "total" in stats + assert "real" in stats + assert "virtual" in stats + assert "by_strategy" in stats + assert stats["total"]["sets"] == 6 + assert stats["real"]["sets"] == 5 + assert stats["virtual"]["sets"] == 1 + assert "manual" in stats["by_strategy"] + assert "simulation" in stats["by_strategy"] + # 하위호환 필드 + assert "total_records" in stats + assert stats["total_records"] == 2 + mem.close() + + +def test_upsert_strategy_performance(): + """전략 성과 upsert""" + import db + mem = _make_mem_conn() + with patch("db._conn", return_value=mem): + db.init_db() + # 최초 insert + db.upsert_strategy_performance( + strategy="simulation", + draw_no=1150, + sets_count=10, + total_correct=30, + max_correct=5, + prize_total=5000, + avg_score=3.0, + ) + rows = db.get_strategy_performance(strategy="simulation") + assert len(rows) == 1 + assert rows[0]["sets_count"] == 10 + assert rows[0]["avg_score"] == 3.0 + # upsert (동일 strategy+draw_no) + db.upsert_strategy_performance( + strategy="simulation", + draw_no=1150, + sets_count=20, + total_correct=60, + max_correct=6, + prize_total=10000, + avg_score=4.5, + ) + rows = db.get_strategy_performance(strategy="simulation") + assert len(rows) == 1 # 중복 없이 1개 + assert rows[0]["sets_count"] == 20 + assert rows[0]["avg_score"] == 4.5 + mem.close() + + +def test_update_strategy_weight(): + """전략 가중치 업데이트""" + import db + mem = _make_mem_conn() + with patch("db._conn", return_value=mem): + db.init_db() + # 초기값 확인 + weights_before = db.get_strategy_weights() + combined_before = next(w for w in weights_before if w["strategy"] == "combined") + original_weight = combined_before["weight"] + # 업데이트 + db.update_strategy_weight( + strategy="combined", + weight=0.5, + ema_score=0.75, + total_sets=100, + total_hits_3plus=20, + ) + weights_after = db.get_strategy_weights() + combined_after = next(w for w in weights_after if w["strategy"] == "combined") + assert combined_after["weight"] == 0.5 + assert combined_after["ema_score"] == 0.75 + assert combined_after["total_sets"] == 100 + assert combined_after["total_hits_3plus"] == 20 + mem.close()