feat(lotto): backtest_runs/winner_calibration 테이블 + CRUD

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-31 16:49:03 +09:00
parent 160fc27279
commit bb0e771a4a
2 changed files with 150 additions and 0 deletions

View File

@@ -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()}