성과 통계 인메모리 캐시 추가 (GET /api/lotto/stats/performance)

매 요청마다 전체 recommendations 조회하던 구조를 캐시로 개선.
갱신 시점: 새 회차 채점 직후(_sync_and_check) + TTL 1시간 만료 폴백

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

backend/app/analyzer.py

@@ -18,6 +18,7 @@
 
 import math
 from collections import Counter, defaultdict
+from datetime import datetime, timezone
 from typing import List, Tuple, Dict, Any, Optional
 
 # 구간 정의: (시작, 끝) 포함
@@ -352,3 +353,177 @@ def get_statistical_report(draws: List[Tuple[int, List[int]]]) -> Dict[str, Any]
         "overdue_numbers": [x["number"] for x in sorted_by_gap[:10]],
         "sum_distribution": sum_buckets,
     }
+
+
+def analyze_personal_patterns(
+    all_numbers: List[List[int]],
+    draws: List[Tuple[int, List[int]]],
+) -> Dict[str, Any]:
+    """
+    사용자 추천 이력 기반 개인 패턴 분석.
+    all_numbers: 저장된 모든 추천 번호 리스트 (각 원소는 6개짜리 리스트)
+    draws: 역대 당첨번호 (홀짝/합계 평균 비교용)
+    """
+    if not all_numbers:
+        return {"total_analyzed": 0, "message": "추천 이력이 없습니다"}
+
+    total = len(all_numbers)
+    flat = [n for nums in all_numbers for n in nums]
+    freq = Counter(flat)
+
+    # 번호별 선택 빈도
+    number_frequency = {n: freq.get(n, 0) for n in range(1, 46)}
+    top_picks = sorted(range(1, 46), key=lambda n: -freq.get(n, 0))[:10]
+    least_picks = [n for n in sorted(range(1, 46), key=lambda n: freq.get(n, 0)) if freq.get(n, 0) == 0][:10]
+
+    # 패턴 지표
+    odd_counts = [sum(1 for n in nums if n % 2 == 1) for nums in all_numbers]
+    sums = [sum(nums) for nums in all_numbers]
+    ranges = [max(nums) - min(nums) for nums in all_numbers]
+    consecutive_count = sum(
+        1 for nums in all_numbers
+        if any(sorted(nums)[i + 1] - sorted(nums)[i] == 1 for i in range(5))
+    )
+
+    zone_totals = {k: 0 for k in ["1-9", "10-19", "20-29", "30-39", "40-45"]}
+    zone_ranges = [("1-9", 1, 9), ("10-19", 10, 19), ("20-29", 20, 29), ("30-39", 30, 39), ("40-45", 40, 45)]
+    for nums in all_numbers:
+        for label, lo, hi in zone_ranges:
+            zone_totals[label] += sum(1 for n in nums if lo <= n <= hi)
+    zone_avg = {k: round(v / total, 2) for k, v in zone_totals.items()}
+
+    avg_odd = sum(odd_counts) / total
+    avg_sum = sum(sums) / total
+    avg_range = sum(ranges) / total
+
+    # 역대 당첨번호 평균과 비교
+    if draws:
+        draw_odd_avg = sum(sum(1 for n in nums if n % 2 == 1) for _, nums in draws) / len(draws)
+        draw_sum_avg = sum(sum(nums) for _, nums in draws) / len(draws)
+    else:
+        draw_odd_avg = 3.0
+        draw_sum_avg = 138.0
+
+    return {
+        "total_analyzed": total,
+        "number_frequency": number_frequency,
+        "top_picks": top_picks,
+        "least_picks": least_picks,
+        "pattern": {
+            "avg_odd_count": round(avg_odd, 2),
+            "avg_sum": round(avg_sum, 1),
+            "avg_range": round(avg_range, 1),
+            "consecutive_rate": round(consecutive_count / total, 3),
+            "zone_avg": zone_avg,
+        },
+        "vs_draw_avg": {
+            "odd_diff": round(avg_odd - draw_odd_avg, 2),
+            "sum_diff": round(avg_sum - draw_sum_avg, 1),
+            "odd_tendency": "홀수 선호" if avg_odd > draw_odd_avg + 0.2 else ("짝수 선호" if avg_odd < draw_odd_avg - 0.2 else "균형"),
+            "sum_tendency": "고합계 선호" if avg_sum > draw_sum_avg + 5 else ("저합계 선호" if avg_sum < draw_sum_avg - 5 else "균형"),
+        },
+    }
+
+
+def generate_weekly_report(draws: List[Tuple[int, List[int]]], target_drw_no: int) -> Dict[str, Any]:
+    """
+    특정 회차 공략 리포트 생성.
+    target_drw_no: 공략 대상 회차 (아직 추첨 안 된 회차)
+    draws: target_drw_no 이전까지의 당첨번호 (오름차순)
+    """
+    if not draws:
+        return {"error": "데이터 없음"}
+
+    cache = build_analysis_cache(draws)
+    total_draws = cache["total_draws"]
+    freq_all = cache["freq_all"]
+    last_seen_gap = cache["last_seen_gap"]
+
+    recent_10 = draws[-10:] if len(draws) >= 10 else draws
+    recent_3 = draws[-3:] if len(draws) >= 3 else draws
+
+    # 과출현: 최근 10회에 2회 이상 출현 번호 (출현 많은 순)
+    r10_nums = [n for _, nums in recent_10 for n in nums]
+    r10_freq = Counter(r10_nums)
+    hot_numbers = [n for n, _ in sorted(r10_freq.items(), key=lambda x: -x[1]) if r10_freq[n] >= 2]
+
+    # 냉각: 역대 출현 빈도 낮은 번호
+    cold_numbers = sorted(range(1, 46), key=lambda n: freq_all.get(n, 0))[:10]
+
+    # 오버듀: 가장 오래 미출현 번호
+    overdue_numbers = sorted(range(1, 46), key=lambda n: -last_seen_gap.get(n, 0))[:10]
+
+    # 최근 3회 연속 출현 (2회 이상)
+    r3_nums = [n for _, nums in recent_3 for n in nums]
+    r3_freq = Counter(r3_nums)
+    triple_appear = sorted(n for n, cnt in r3_freq.items() if cnt >= 2)
+
+    recent_sums = [sum(nums) for _, nums in recent_10]
+    recent_odd = [sum(1 for n in nums if n % 2 == 1) for _, nums in recent_10]
+
+    # 갭 기반 가중치 (오래된 번호일수록 높음)
+    gap_w = {n: last_seen_gap.get(n, 0) for n in range(1, 46)}
+
+    def _pick(exclude=None, prefer=None, n=6):
+        ex = set(exclude or [])
+        chosen = []
+        # prefer에서 최대 3개 우선 선택
+        for p in (prefer or []):
+            if p not in ex and len(chosen) < 3:
+                chosen.append(p)
+        # 구간별 1개씩 (갭 우선)
+        for lo, hi in [(1, 9), (10, 19), (20, 29), (30, 39), (40, 45)]:
+            if len(chosen) >= n:
+                break
+            cands = [x for x in range(lo, hi + 1) if x not in ex and x not in chosen]
+            if cands:
+                chosen.append(max(cands, key=lambda x: gap_w.get(x, 0)))
+        # 부족하면 나머지에서 갭 순
+        rest = sorted(
+            [x for x in range(1, 46) if x not in ex and x not in chosen],
+            key=lambda x: -gap_w.get(x, 0),
+        )
+        while len(chosen) < n and rest:
+            chosen.append(rest.pop(0))
+        return sorted(chosen[:n])
+
+    set1 = _pick(exclude=hot_numbers[:5], prefer=overdue_numbers[:5])
+    set2 = _pick()
+    set3 = _pick(exclude=hot_numbers)
+
+    # 신뢰도 점수
+    data_vol = min(total_draws / 500, 1.0)
+    if len(recent_sums) > 1:
+        avg_s = sum(recent_sums) / len(recent_sums)
+        std_s = (sum((s - avg_s) ** 2 for s in recent_sums) / len(recent_sums)) ** 0.5
+        pattern = max(0.0, 1.0 - std_s / 60.0)
+    else:
+        pattern = 0.5
+    trend = max(0.0, 1.0 - len(hot_numbers) / max(len(r10_nums), 1))
+    confidence = round((data_vol * 0.4 + pattern * 0.35 + trend * 0.25) * 100)
+
+    return {
+        "target_drw_no": target_drw_no,
+        "based_on_draw": draws[-1][0],
+        "generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
+        "hot_numbers": hot_numbers[:8],
+        "cold_numbers": cold_numbers,
+        "overdue_numbers": overdue_numbers,
+        "recent_pattern": {
+            "last3_numbers": sorted(set(r3_nums)),
+            "triple_appear": triple_appear,
+            "recent_sum_avg": round(sum(recent_sums) / len(recent_sums), 1) if recent_sums else 0,
+            "recent_odd_avg": round(sum(recent_odd) / len(recent_odd), 1) if recent_odd else 0,
+        },
+        "recommended_sets": [
+            {"strategy": "냉각번호 중심", "numbers": set1, "description": "오랫동안 미출현 번호 위주 + 과출현 제외"},
+            {"strategy": "균형형", "numbers": set2, "description": "구간 균형 + 갭 최적화"},
+            {"strategy": "과출현 피하기", "numbers": set3, "description": "최근 자주 나온 번호 완전 제외"},
+        ],
+        "confidence_score": confidence,
+        "confidence_factors": {
+            "data_volume": round(data_vol * 100),
+            "pattern_consistency": round(pattern * 100),
+            "recent_trend": round(trend * 100),
+        },
+    }

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(
             """
@@ -596,6 +624,54 @@ def delete_recommendation(rec_id: int) -> bool:
         cur = conn.execute("DELETE FROM recommendations WHERE id = ?", (rec_id,))
         return cur.rowcount > 0
 
+def get_recommendation_performance() -> Dict[str, Any]:
+    """채점된 추천 이력 기반 성과 통계"""
+    with _conn() as conn:
+        rows = conn.execute(
+            "SELECT correct_count, rank FROM recommendations WHERE checked = 1"
+        ).fetchall()
+
+    if not rows:
+        return {
+            "total_checked": 0,
+            "avg_correct": 0.0,
+            "distribution": {str(i): 0 for i in range(7)},
+            "rate_3plus": 0.0,
+            "rate_4plus": 0.0,
+            "by_rank": {"rank_1": 0, "rank_2": 0, "rank_3": 0, "rank_4": 0, "rank_5": 0, "no_prize": 0},
+            "vs_random": {"our_avg": 0.0, "random_avg": 0.8, "improvement_pct": 0.0},
+        }
+
+    total = len(rows)
+    corrects = [r["correct_count"] or 0 for r in rows]
+    ranks = [r["rank"] or 0 for r in rows]
+    avg_correct = sum(corrects) / total
+
+    RANDOM_AVG = 0.8  # 이론 기댓값: 6 * (6/45)
+    improvement = (avg_correct - RANDOM_AVG) / RANDOM_AVG * 100
+
+    return {
+        "total_checked": total,
+        "avg_correct": round(avg_correct, 3),
+        "distribution": {str(i): corrects.count(i) for i in range(7)},
+        "rate_3plus": round(sum(1 for c in corrects if c >= 3) / total, 4),
+        "rate_4plus": round(sum(1 for c in corrects if c >= 4) / total, 4),
+        "by_rank": {
+            "rank_1": ranks.count(1),
+            "rank_2": ranks.count(2),
+            "rank_3": ranks.count(3),
+            "rank_4": ranks.count(4),
+            "rank_5": ranks.count(5),
+            "no_prize": ranks.count(0),
+        },
+        "vs_random": {
+            "our_avg": round(avg_correct, 3),
+            "random_avg": RANDOM_AVG,
+            "improvement_pct": round(improvement, 1),
+        },
+    }
+
+
 def update_recommendation_result(rec_id: int, rank: int, correct_count: int, has_bonus: bool) -> bool:
     with _conn() as conn:
         cur = conn.execute(
@@ -1061,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",

backend/app/main.py

@@ -1,4 +1,5 @@
 import os
+import time
 from typing import Optional, List, Dict, Any, Tuple
 from fastapi import FastAPI, HTTPException
 from pydantic import BaseModel
@@ -20,13 +21,21 @@ from .db import (
     get_all_subscription_items, create_subscription_item,
     update_subscription_item, delete_subscription_item,
     get_subscription_profile, upsert_subscription_profile,
+    # 성과 통계
+    get_recommendation_performance,
+    # Phase 2: 구매 이력
+    add_purchase, get_purchases, update_purchase, delete_purchase, get_purchase_stats,
+    # Phase 2: 주간 리포트 캐시
+    save_weekly_report, get_weekly_report_list, get_weekly_report,
+    # Phase 2: 개인 패턴 분석
+    get_all_recommendation_numbers,
 )
 from .recommender import recommend_numbers, recommend_with_heatmap
 from .collector import sync_latest, sync_ensure_all
 from .generator import run_simulation, generate_smart_recommendations
 from .checker import check_results_for_draw
 from .utils import calc_metrics, calc_recent_overlap
-from .analyzer import get_statistical_report
+from .analyzer import get_statistical_report, generate_weekly_report, analyze_personal_patterns
 
 app = FastAPI()
 scheduler = BackgroundScheduler(timezone=os.getenv("TZ", "Asia/Seoul"))
@@ -34,6 +43,17 @@ scheduler = BackgroundScheduler(timezone=os.getenv("TZ", "Asia/Seoul"))
 ALL_URL = os.getenv("LOTTO_ALL_URL", "https://smok95.github.io/lotto/results/all.json")
 LATEST_URL = os.getenv("LOTTO_LATEST_URL", "https://smok95.github.io/lotto/results/latest.json")
 
+# ── 성과 통계 인메모리 캐시 ───────────────────────────────────────────────────
+# 채점 데이터는 하루 2번 스케줄러 실행 시에만 갱신되므로 인메모리 캐시로 충분
+_PERF_CACHE: Dict[str, Any] = {"data": None, "at": 0.0}
+_PERF_CACHE_TTL = 3600  # 1시간 (스케줄러 미실행 상황 대비 폴백)
+
+
+def _refresh_perf_cache() -> None:
+    _PERF_CACHE["data"] = get_recommendation_performance()
+    _PERF_CACHE["at"] = time.time()
+    print("[PerfCache] 성과 통계 캐시 갱신")
+
 
 @app.on_event("startup")
 def on_startup():
@@ -45,6 +65,7 @@ def on_startup():
         res = sync_latest(LATEST_URL)
         if res["was_new"]:
             check_results_for_draw(res["drawNo"])
+            _refresh_perf_cache()  # 새 채점 결과 반영 → 즉시 갱신
 
     scheduler.add_job(_sync_and_check, "cron", hour="9,21", minute=10)
 
@@ -55,6 +76,20 @@ def on_startup():
 
     scheduler.add_job(_run_simulation_job, "cron", hour="0,4,8,12,16,20", minute=5)
 
+    # 3. 토요일 오전 9시 — 다음 회차 공략 리포트 자동 캐싱
+    def _save_weekly_report_job():
+        import json as _json
+        draws = get_all_draw_numbers()
+        latest = get_latest_draw()
+        if not draws or not latest:
+            return
+        target = latest["drw_no"] + 1
+        report = generate_weekly_report(draws, target)
+        save_weekly_report(target, _json.dumps(report, ensure_ascii=False))
+        print(f"[WeeklyReport] {target}회차 리포트 저장 완료")
+
+    scheduler.add_job(_save_weekly_report_job, "cron", day_of_week="sat", hour=9, minute=0)
+
     scheduler.start()
 
 
@@ -148,6 +183,127 @@ def api_stats():
     }
 
 
+# ── 추천 성과 통계 (Phase 1, 인메모리 캐시) ──────────────────────────────────
+@app.get("/api/lotto/stats/performance")
+def api_performance_stats():
+    """
+    채점된 추천 이력 기반 성과 통계 (캐시 반환).
+    캐시 갱신 시점: 새 회차 채점 직후 | TTL 1시간 만료 시
+    """
+    if _PERF_CACHE["data"] is None or time.time() - _PERF_CACHE["at"] > _PERF_CACHE_TTL:
+        _refresh_perf_cache()
+    return _PERF_CACHE["data"]
+
+
+# ── 회차 공략 리포트 (Phase 1) ────────────────────────────────────────────────
+@app.get("/api/lotto/report/latest")
+def api_report_latest():
+    """
+    다음 회차 공략 리포트 (최신 회차 기준으로 자동 계산).
+    - 과출현/냉각/오버듀 번호 분석
+    - 최근 3회 패턴
+    - 3가지 전략별 추천 번호
+    - AI 신뢰도 점수
+    """
+    draws = get_all_draw_numbers()
+    if not draws:
+        raise HTTPException(status_code=404, detail="No data yet")
+    latest = get_latest_draw()
+    target = latest["drw_no"] + 1
+    return generate_weekly_report(draws, target)
+
+
+@app.get("/api/lotto/report/history")
+def api_report_history(limit: int = 10):
+    """저장된 주간 리포트 목록 (자동 저장된 것만)"""
+    return {"reports": get_weekly_report_list(limit=min(limit, 52))}
+
+
+@app.get("/api/lotto/report/{drw_no}")
+def api_report_by_draw(drw_no: int):
+    """
+    특정 회차 공략 리포트 (캐시 우선, 없으면 실시간 생성).
+    """
+    cached = get_weekly_report(drw_no)
+    if cached:
+        return {**cached, "cached": True}
+
+    draws = get_all_draw_numbers()
+    if not draws:
+        raise HTTPException(status_code=404, detail="No data yet")
+    base_draws = [(no, nums) for no, nums in draws if no < drw_no]
+    if not base_draws:
+        raise HTTPException(status_code=400, detail=f"{drw_no}회차 이전 데이터가 없습니다")
+    return {**generate_weekly_report(base_draws, drw_no), "cached": False}
+
+
+# ── 개인 패턴 분석 (Phase 2) ─────────────────────────────────────────────────
+@app.get("/api/lotto/analysis/personal")
+def api_personal_analysis():
+    """
+    저장된 추천 이력 기반 개인 패턴 분석.
+    - 자주 선택한 번호 TOP 10 / 한 번도 선택 안 한 번호
+    - 홀짝 비율, 합계, 범위, 연속번호 포함률
+    - 구간별 분포, 역대 당첨번호 평균과 비교
+    """
+    all_numbers = get_all_recommendation_numbers()
+    draws = get_all_draw_numbers()
+    return analyze_personal_patterns(all_numbers, draws)
+
+
+# ── 구매 이력 API (Phase 2) ───────────────────────────────────────────────────
+
+class PurchaseCreate(BaseModel):
+    draw_no: int
+    amount: int
+    sets: int = 1
+    prize: int = 0
+    note: str = ""
+
+
+class PurchaseUpdate(BaseModel):
+    draw_no: Optional[int] = None
+    amount: Optional[int] = None
+    sets: Optional[int] = None
+    prize: Optional[int] = None
+    note: Optional[str] = None
+
+
+@app.get("/api/lotto/purchase/stats")
+def api_purchase_stats():
+    """투자 수익률 통계 (총 투자금, 총 당첨금, 수익률 등)"""
+    return get_purchase_stats()
+
+
+@app.get("/api/lotto/purchase")
+def api_purchase_list(draw_no: Optional[int] = None, days: Optional[int] = None):
+    """구매 이력 조회 (draw_no, days 필터 선택)"""
+    return {"records": get_purchases(draw_no=draw_no, days=days)}
+
+
+@app.post("/api/lotto/purchase", status_code=201)
+def api_purchase_create(body: PurchaseCreate):
+    """구매 이력 추가"""
+    return add_purchase(body.draw_no, body.amount, body.sets, body.prize, body.note)
+
+
+@app.put("/api/lotto/purchase/{purchase_id}")
+def api_purchase_update(purchase_id: int, body: PurchaseUpdate):
+    """구매 이력 수정 (당첨금 업데이트 등)"""
+    updated = update_purchase(purchase_id, body.model_dump(exclude_none=True))
+    if updated is None:
+        raise HTTPException(status_code=404, detail="Purchase not found")
+    return updated
+
+
+@app.delete("/api/lotto/purchase/{purchase_id}")
+def api_purchase_delete(purchase_id: int):
+    """구매 이력 삭제"""
+    if not delete_purchase(purchase_id):
+        raise HTTPException(status_code=404, detail="Purchase not found")
+    return {"ok": True}
+
+
 # ── 통계 분석 리포트 ────────────────────────────────────────────────────────
 @app.get("/api/lotto/analysis")
 def api_analysis():