diff --git a/lotto/app/db.py b/lotto/app/db.py index 8862309..af47d4d 100644 --- a/lotto/app/db.py +++ b/lotto/app/db.py @@ -300,7 +300,51 @@ def init_db() -> None: _ensure_column(conn, "lotto_briefings", "tier_rationale", "ALTER TABLE lotto_briefings ADD COLUMN tier_rationale TEXT NOT NULL DEFAULT '{}'") - + # ── weight_trials / auto_picks / weight_base_history 테이블 ────────── + conn.execute(""" + CREATE TABLE IF NOT EXISTS weight_trials ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + week_start TEXT NOT NULL, + day_of_week INTEGER NOT NULL, + weight_json TEXT NOT NULL, + source TEXT NOT NULL, + base_at_gen TEXT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), + UNIQUE(week_start, day_of_week) + ) + """) + conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_wt_week + ON weight_trials(week_start, day_of_week) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS auto_picks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + trial_id INTEGER NOT NULL REFERENCES weight_trials(id) ON DELETE CASCADE, + pick_no INTEGER NOT NULL, + numbers TEXT NOT NULL, + meta_score REAL, + correct INTEGER, + rank INTEGER, + graded_at TEXT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), + UNIQUE(trial_id, pick_no) + ) + """) + conn.execute("CREATE INDEX IF NOT EXISTS idx_ap_trial ON auto_picks(trial_id)") + conn.execute("CREATE INDEX IF NOT EXISTS idx_ap_graded ON auto_picks(graded_at)") + conn.execute(""" + CREATE TABLE IF NOT EXISTS weight_base_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + effective_from TEXT NOT NULL, + weight_json TEXT NOT NULL, + source_trial_id INTEGER REFERENCES weight_trials(id), + update_reason TEXT, + winner_score REAL, + winner_max_correct INTEGER, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) + ) + """) def upsert_draw(row: Dict[str, Any]) -> None: @@ -1247,3 +1291,155 @@ def list_reviews(limit: int = 10) -> List[Dict[str, Any]]: ).fetchall() return [_review_row(r) for r in rows] + +# --- weight_trials / auto_picks / weight_base_history CRUD --- + +def save_weight_trial( + week_start: str, + day_of_week: int, + weight: List[float], + source: str, + base_at_gen: Optional[List[float]] = None, +) -> int: + with _conn() as conn: + cur = conn.execute( + """ + INSERT INTO weight_trials (week_start, day_of_week, weight_json, source, base_at_gen) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(week_start, day_of_week) DO UPDATE SET + weight_json = excluded.weight_json, + source = excluded.source, + base_at_gen = excluded.base_at_gen + """, + (week_start, day_of_week, json.dumps(weight), + source, json.dumps(base_at_gen) if base_at_gen else None), + ) + if cur.lastrowid: + return cur.lastrowid + row = conn.execute( + "SELECT id FROM weight_trials WHERE week_start=? AND day_of_week=?", + (week_start, day_of_week), + ).fetchone() + return int(row["id"]) + + +def get_weight_trial(week_start: str, day_of_week: int) -> Optional[Dict[str, Any]]: + with _conn() as conn: + row = conn.execute( + "SELECT * FROM weight_trials WHERE week_start=? AND day_of_week=?", + (week_start, day_of_week), + ).fetchone() + if not row: + return None + d = dict(row) + d["weight"] = json.loads(d.pop("weight_json")) + if d.get("base_at_gen"): + d["base_at_gen"] = json.loads(d["base_at_gen"]) + return d + + +def get_weekly_trials(week_start: str) -> List[Dict[str, Any]]: + with _conn() as conn: + rows = conn.execute( + "SELECT * FROM weight_trials WHERE week_start=? ORDER BY day_of_week", + (week_start,), + ).fetchall() + out = [] + for r in rows: + d = dict(r) + d["weight"] = json.loads(d.pop("weight_json")) + if d.get("base_at_gen"): + d["base_at_gen"] = json.loads(d["base_at_gen"]) + out.append(d) + return out + + +def save_auto_pick( + trial_id: int, + pick_no: int, + numbers: List[int], + meta_score: Optional[float] = None, +) -> int: + with _conn() as conn: + cur = conn.execute( + """ + INSERT OR REPLACE INTO auto_picks (trial_id, pick_no, numbers, meta_score) + VALUES (?, ?, ?, ?) + """, + (trial_id, pick_no, json.dumps(sorted(numbers)), meta_score), + ) + return cur.lastrowid + + +def get_auto_picks(trial_id: int) -> List[Dict[str, Any]]: + with _conn() as conn: + rows = conn.execute( + "SELECT * FROM auto_picks WHERE trial_id=? ORDER BY pick_no", + (trial_id,), + ).fetchall() + out = [] + for r in rows: + d = dict(r) + d["numbers"] = json.loads(d["numbers"]) + out.append(d) + return out + + +def update_auto_pick_grade(pick_id: int, correct: int, rank: Optional[int]) -> None: + with _conn() as conn: + conn.execute( + """ + UPDATE auto_picks + SET correct=?, rank=?, graded_at=strftime('%Y-%m-%dT%H:%M:%fZ','now') + WHERE id=? + """, + (correct, rank, pick_id), + ) + + +def get_current_base() -> Optional[List[float]]: + """weight_base_history 최신 row의 weight. 없으면 None (cold start).""" + with _conn() as conn: + row = conn.execute( + "SELECT weight_json FROM weight_base_history ORDER BY id DESC LIMIT 1", + ).fetchone() + if not row: + return None + return json.loads(row["weight_json"]) + + +def save_base_history( + effective_from: str, + weight: List[float], + source_trial_id: Optional[int], + update_reason: str, + winner_score: Optional[float], + winner_max_correct: Optional[int], +) -> int: + with _conn() as conn: + cur = conn.execute( + """ + INSERT INTO weight_base_history + (effective_from, weight_json, source_trial_id, update_reason, + winner_score, winner_max_correct) + VALUES (?, ?, ?, ?, ?, ?) + """, + (effective_from, json.dumps(weight), source_trial_id, + update_reason, winner_score, winner_max_correct), + ) + return cur.lastrowid + + +def get_base_history(limit: int = 12) -> List[Dict[str, Any]]: + with _conn() as conn: + rows = conn.execute( + "SELECT * FROM weight_base_history ORDER BY id DESC LIMIT ?", + (limit,), + ).fetchall() + out = [] + for r in rows: + d = dict(r) + d["weight"] = json.loads(d.pop("weight_json")) + out.append(d) + return out +