diff --git a/lotto/app/db.py b/lotto/app/db.py index af47d4d..bc58cf2 100644 --- a/lotto/app/db.py +++ b/lotto/app/db.py @@ -125,6 +125,48 @@ def init_db() -> None: "ON simulation_candidates(is_best, score_total DESC);" ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS backtest_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + draw_no INTEGER NOT NULL, + strategy TEXT NOT NULL, + weight_label TEXT NOT NULL DEFAULT '-', + weight_json TEXT, + trial_id INTEGER, + n_tickets INTEGER NOT NULL, + m3 INTEGER NOT NULL DEFAULT 0, + m4 INTEGER NOT NULL DEFAULT 0, + m5 INTEGER NOT NULL DEFAULT 0, + m6 INTEGER NOT NULL DEFAULT 0, + bonus_hits INTEGER NOT NULL DEFAULT 0, + best_match INTEGER NOT NULL DEFAULT 0, + avg_meta_score REAL, + created_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + """ + ) + conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS uq_backtest_run " + "ON backtest_runs(draw_no, strategy, weight_label);") + conn.execute( + """ + CREATE TABLE IF NOT EXISTS winner_calibration ( + draw_no INTEGER PRIMARY KEY, + winning_json TEXT NOT NULL, + score_total REAL NOT NULL, + score_frequency REAL NOT NULL, + score_fingerprint REAL NOT NULL, + score_gap REAL NOT NULL, + score_cooccur REAL NOT NULL, + score_diversity REAL NOT NULL, + percentile REAL, + my_pick_avg REAL, + cache_draws INTEGER NOT NULL, + created_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + """ + ) + conn.execute( """ CREATE TABLE IF NOT EXISTS best_picks ( @@ -1443,3 +1485,80 @@ def get_base_history(limit: int = 12) -> List[Dict[str, Any]]: out.append(d) return out + +# ── backtest_runs / winner_calibration CRUD ─────────────────────────────────── + +def save_backtest_run(draw_no, strategy, weight_label, weight_json, trial_id, + n_tickets, hist, best_match, avg_meta_score) -> None: + with _conn() as conn: + conn.execute( + """ + INSERT INTO backtest_runs + (draw_no, strategy, weight_label, weight_json, trial_id, n_tickets, + m3, m4, m5, m6, bonus_hits, best_match, avg_meta_score) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(draw_no, strategy, weight_label) DO UPDATE SET + weight_json=excluded.weight_json, trial_id=excluded.trial_id, + n_tickets=excluded.n_tickets, m3=excluded.m3, m4=excluded.m4, + m5=excluded.m5, m6=excluded.m6, bonus_hits=excluded.bonus_hits, + best_match=excluded.best_match, avg_meta_score=excluded.avg_meta_score, + created_at=datetime('now') + """, + (draw_no, strategy, weight_label, + json.dumps(weight_json) if weight_json is not None else None, + trial_id, n_tickets, + hist.get("m3",0), hist.get("m4",0), hist.get("m5",0), hist.get("m6",0), + hist.get("bonus_hits",0), best_match, avg_meta_score), + ) + +def get_backtest_runs(draw_no=None, strategy=None) -> List[Dict[str, Any]]: + q = "SELECT * FROM backtest_runs WHERE 1=1" + args = [] + if draw_no is not None: + q += " AND draw_no=?"; args.append(draw_no) + if strategy is not None: + q += " AND strategy=?"; args.append(strategy) + q += " ORDER BY draw_no DESC, strategy, weight_label" + with _conn() as conn: + return [dict(r) for r in conn.execute(q, args).fetchall()] + +def save_winner_calibration(draw_no, winning, scores, percentile, + my_pick_avg, cache_draws) -> None: + with _conn() as conn: + conn.execute( + """ + INSERT INTO winner_calibration + (draw_no, winning_json, score_total, score_frequency, score_fingerprint, + score_gap, score_cooccur, score_diversity, percentile, my_pick_avg, cache_draws) + VALUES (?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(draw_no) DO UPDATE SET + winning_json=excluded.winning_json, score_total=excluded.score_total, + score_frequency=excluded.score_frequency, score_fingerprint=excluded.score_fingerprint, + score_gap=excluded.score_gap, score_cooccur=excluded.score_cooccur, + score_diversity=excluded.score_diversity, percentile=excluded.percentile, + my_pick_avg=excluded.my_pick_avg, cache_draws=excluded.cache_draws, + created_at=datetime('now') + """, + (draw_no, json.dumps(winning), scores["score_total"], scores["score_frequency"], + scores["score_fingerprint"], scores["score_gap"], scores["score_cooccur"], + scores["score_diversity"], percentile, my_pick_avg, cache_draws), + ) + +def get_winner_calibration(draw_no: int) -> Optional[Dict[str, Any]]: + with _conn() as conn: + r = conn.execute("SELECT * FROM winner_calibration WHERE draw_no=?", + (draw_no,)).fetchone() + return dict(r) if r else None + +def get_calibration_history(limit: int = 52) -> List[Dict[str, Any]]: + with _conn() as conn: + rows = conn.execute( + "SELECT * FROM winner_calibration ORDER BY draw_no DESC LIMIT ?", + (limit,)).fetchall() + return [dict(r) for r in rows] + +def get_calibrated_draw_nos() -> set: + with _conn() as conn: + return {r["draw_no"] for r in + conn.execute("SELECT draw_no FROM winner_calibration").fetchall()} + diff --git a/lotto/tests/test_backtest_db.py b/lotto/tests/test_backtest_db.py new file mode 100644 index 0000000..5f47975 --- /dev/null +++ b/lotto/tests/test_backtest_db.py @@ -0,0 +1,31 @@ +import os, tempfile + +def _fresh_db(monkeypatch): + tmp = tempfile.mkdtemp() + path = os.path.join(tmp, "lotto.db") + from app import db + monkeypatch.setattr(db, "DB_PATH", path) + db.init_db() + return db + +def test_backtest_tables_exist(monkeypatch): + db = _fresh_db(monkeypatch) + with db._conn() as conn: + tables = {r["name"] for r in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'").fetchall()} + assert "backtest_runs" in tables + assert "winner_calibration" in tables + +def test_backtest_runs_unique(monkeypatch): + db = _fresh_db(monkeypatch) + db.save_backtest_run(draw_no=100, strategy="random_null", weight_label="-", + weight_json=None, trial_id=None, n_tickets=10, + hist={"m3":1,"m4":0,"m5":0,"m6":0,"bonus_hits":0}, + best_match=3, avg_meta_score=0.5) + db.save_backtest_run(draw_no=100, strategy="random_null", weight_label="-", + weight_json=None, trial_id=None, n_tickets=10, + hist={"m3":2,"m4":0,"m5":0,"m6":0,"bonus_hits":0}, + best_match=3, avg_meta_score=0.6) # 멱등 upsert + rows = db.get_backtest_runs(draw_no=100) + assert len(rows) == 1 + assert rows[0]["m3"] == 2 # 마지막 값으로 갱신