로또 프리미엄 Phase 2 — 구매 이력 + 개인 패턴 분석 + 주간 리포트 캐싱

- purchase_history 테이블 추가 (draw_no, amount, sets, prize, note) - weekly_reports 캐시 테이블 추가 (drw_no UNIQUE, report JSON) - GET /api/lotto/purchase 구매 이력 조회 (draw_no, days 필터) - POST /api/lotto/purchase 구매 이력 추가 - PUT /api/lotto/purchase/:id 구매 이력 수정 (당첨금 업데이트) - DELETE /api/lotto/purchase/:id 구매 이력 삭제 - GET /api/lotto/purchase/stats 투자 수익률 통계 - GET /api/lotto/analysis/personal 개인 패턴 분석 (top/least picks, 홀짝/구간/연속번호) - GET /api/lotto/report/history 저장된 주간 리포트 목록 - GET /api/lotto/report/:drw_no 캐시 우선 조회 + cached 플래그 - 스케줄러: 토요일 09:00 주간 리포트 자동 생성 및 DB 캐싱 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-19 23:59:07 +09:00
parent 2ce118baba
commit 732d78becc
3 changed files with 336 additions and 5 deletions
--- a/backend/app/db.py
+++ b/backend/app/db.py
@@ -251,6 +251,34 @@ def init_db() -> None:
            "CREATE INDEX IF NOT EXISTS idx_sub_items_created ON subscription_items(created_at DESC);"
        )

+        # ── purchase_history 테이블 ────────────────────────────────────────────
+        conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS purchase_history (
+                id         INTEGER PRIMARY KEY AUTOINCREMENT,
+                draw_no    INTEGER NOT NULL,
+                amount     INTEGER NOT NULL,
+                sets       INTEGER NOT NULL DEFAULT 1,
+                prize      INTEGER NOT NULL DEFAULT 0,
+                note       TEXT    NOT NULL DEFAULT '',
+                created_at TEXT    NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
+            );
+            """
+        )
+        conn.execute("CREATE INDEX IF NOT EXISTS idx_purchase_draw ON purchase_history(draw_no DESC);")
+
+        # ── weekly_reports 캐시 테이블 ──────────────────────────────────────────
+        conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS weekly_reports (
+                id           INTEGER PRIMARY KEY AUTOINCREMENT,
+                drw_no       INTEGER UNIQUE NOT NULL,
+                report       TEXT    NOT NULL,
+                generated_at TEXT    NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
+            );
+            """
+        )
+
        # ── subscription_profile 테이블 (싱글톤 id=1) ──────────────────────────
        conn.execute(
            """
@@ -1109,6 +1137,144 @@ def get_subscription_profile() -> Optional[Dict[str, Any]]:
    return _profile_row_to_dict(r) if r else None


+# ── purchase_history CRUD ─────────────────────────────────────────────────────
+
+def _purchase_row_to_dict(r) -> Dict[str, Any]:
+    return {
+        "id": r["id"],
+        "draw_no": r["draw_no"],
+        "amount": r["amount"],
+        "sets": r["sets"],
+        "prize": r["prize"],
+        "note": r["note"],
+        "created_at": r["created_at"],
+    }
+
+
+def add_purchase(draw_no: int, amount: int, sets: int, prize: int = 0, note: str = "") -> Dict[str, Any]:
+    with _conn() as conn:
+        conn.execute(
+            "INSERT INTO purchase_history (draw_no, amount, sets, prize, note) VALUES (?, ?, ?, ?, ?)",
+            (draw_no, amount, sets, prize, note),
+        )
+        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]]:
+    conditions, params = [], []
+    if draw_no is not None:
+        conditions.append("draw_no = ?")
+        params.append(draw_no)
+    if days:
+        conditions.append("created_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 purchase_history {where} ORDER BY draw_no DESC, id DESC",
+            params,
+        ).fetchall()
+    return [_purchase_row_to_dict(r) for r in rows]
+
+
+def update_purchase(purchase_id: int, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    allowed = {"draw_no", "amount", "sets", "prize", "note"}
+    updates = {k: v for k, v in data.items() if k in allowed}
+    if not updates:
+        with _conn() as conn:
+            row = conn.execute("SELECT * FROM purchase_history WHERE id = ?", (purchase_id,)).fetchone()
+        return _purchase_row_to_dict(row) if row else None
+    set_clause = ", ".join(f"{k} = ?" for k in updates)
+    with _conn() as conn:
+        cur = conn.execute(
+            f"UPDATE purchase_history SET {set_clause} WHERE id = ?",
+            list(updates.values()) + [purchase_id],
+        )
+        if cur.rowcount == 0:
+            return None
+        row = conn.execute("SELECT * FROM purchase_history WHERE id = ?", (purchase_id,)).fetchone()
+    return _purchase_row_to_dict(row)
+
+
+def delete_purchase(purchase_id: int) -> bool:
+    with _conn() as conn:
+        cur = conn.execute("DELETE FROM purchase_history WHERE id = ?", (purchase_id,))
+    return cur.rowcount > 0
+
+
+def get_purchase_stats() -> Dict[str, Any]:
+    with _conn() as conn:
+        rows = conn.execute("SELECT amount, prize FROM purchase_history").fetchall()
+    if not rows:
+        return {
+            "total_records": 0,
+            "total_invested": 0,
+            "total_prize": 0,
+            "net": 0,
+            "return_rate": 0.0,
+            "prize_count": 0,
+            "max_prize": 0,
+        }
+    amounts = [r["amount"] for r in rows]
+    prizes = [r["prize"] for r in rows]
+    total_invested = sum(amounts)
+    total_prize = sum(prizes)
+    return {
+        "total_records": len(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),
+    }
+
+
+# ── weekly_reports CRUD ───────────────────────────────────────────────────────
+
+def save_weekly_report(drw_no: int, report_json: str) -> None:
+    with _conn() as conn:
+        conn.execute(
+            """
+            INSERT INTO weekly_reports (drw_no, report)
+            VALUES (?, ?)
+            ON CONFLICT(drw_no) DO UPDATE SET
+                report = excluded.report,
+                generated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')
+            """,
+            (drw_no, report_json),
+        )
+
+
+def get_weekly_report_list(limit: int = 10) -> List[Dict[str, Any]]:
+    with _conn() as conn:
+        rows = conn.execute(
+            "SELECT drw_no, generated_at FROM weekly_reports ORDER BY drw_no DESC LIMIT ?",
+            (limit,),
+        ).fetchall()
+    return [dict(r) for r in rows]
+
+
+def get_weekly_report(drw_no: int) -> Optional[Dict[str, Any]]:
+    with _conn() as conn:
+        row = conn.execute(
+            "SELECT drw_no, report, generated_at FROM weekly_reports WHERE drw_no = ?",
+            (drw_no,),
+        ).fetchone()
+    if not row:
+        return None
+    import json as _json
+    return {"drw_no": row["drw_no"], "generated_at": row["generated_at"], **_json.loads(row["report"])}
+
+
+def get_all_recommendation_numbers() -> List[List[int]]:
+    """개인 패턴 분석용: 저장된 모든 추천 번호 반환"""
+    with _conn() as conn:
+        rows = conn.execute("SELECT numbers FROM recommendations ORDER BY id DESC").fetchall()
+    return [json.loads(r["numbers"]) for r in rows]
+
+
 def upsert_subscription_profile(data: Dict[str, Any]) -> Dict[str, Any]:
    field_map = {
        "isHouseholdHead": "is_household_head",