gahusb
535ffea45a
P0 버그 수정: - stock-lab: trade 엔드포인트 NameError 수정 (resp 미정의) - deployer: 동시 배포 시 HTTP 200 → 503 반환 P1 데드코드 제거: - stock-lab: fetch_overseas_news(), get_broker_cash() 제거 - blog-lab: 미사용 urlparse import 제거 - lotto-lab: 중복 inline import json 7곳 제거 P2 성능/효율 개선: - lotto-lab: 가중 샘플링 3중 복사 → utils.weighted_sample_6() 통합 - lotto-lab: DB 인덱스 3개 추가 (recommendations, purchase_history) - stock-lab: Pydantic .dict() → .model_dump() 호환 - blog-lab: 페이지네이션 상한(le=100) 추가 P3 보안/인프라: - nginx: X-Frame-Options, X-Content-Type-Options, Referrer-Policy 헤더 추가 - docker-compose: travel-proxy CORS 와일드카드 → localhost 전용 - Dockerfile: music-lab, blog-lab, realestate-lab에 PYTHONUNBUFFERED 추가 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1099 lines
43 KiB
Python
1099 lines
43 KiB
Python
# backend/app/db.py
|
|
import os
|
|
import sqlite3
|
|
import json
|
|
import hashlib
|
|
from typing import Any, Dict, Optional, List
|
|
|
|
DB_PATH = "/app/data/lotto.db"
|
|
|
|
def _conn() -> sqlite3.Connection:
|
|
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
def _ensure_column(conn: sqlite3.Connection, table: str, col: str, ddl: str) -> None:
|
|
cols = {r["name"] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()}
|
|
if col not in cols:
|
|
conn.execute(ddl)
|
|
|
|
def init_db() -> None:
|
|
with _conn() as conn:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS draws (
|
|
drw_no INTEGER PRIMARY KEY,
|
|
drw_date TEXT NOT NULL,
|
|
n1 INTEGER NOT NULL,
|
|
n2 INTEGER NOT NULL,
|
|
n3 INTEGER NOT NULL,
|
|
n4 INTEGER NOT NULL,
|
|
n5 INTEGER NOT NULL,
|
|
n6 INTEGER NOT NULL,
|
|
bonus INTEGER NOT NULL,
|
|
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
|
|
);
|
|
"""
|
|
)
|
|
conn.execute("CREATE INDEX IF NOT EXISTS idx_draws_date ON draws(drw_date);")
|
|
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS recommendations (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
|
based_on_draw INTEGER,
|
|
numbers TEXT NOT NULL,
|
|
params TEXT NOT NULL
|
|
);
|
|
"""
|
|
)
|
|
conn.execute("CREATE INDEX IF NOT EXISTS idx_reco_created ON recommendations(created_at DESC);")
|
|
|
|
# ✅ 확장 컬럼들(기존 DB에도 자동 추가)
|
|
_ensure_column(conn, "recommendations", "numbers_sorted",
|
|
"ALTER TABLE recommendations ADD COLUMN numbers_sorted TEXT;")
|
|
_ensure_column(conn, "recommendations", "dedup_hash",
|
|
"ALTER TABLE recommendations ADD COLUMN dedup_hash TEXT;")
|
|
_ensure_column(conn, "recommendations", "favorite",
|
|
"ALTER TABLE recommendations ADD COLUMN favorite INTEGER NOT NULL DEFAULT 0;")
|
|
_ensure_column(conn, "recommendations", "note",
|
|
"ALTER TABLE recommendations ADD COLUMN note TEXT NOT NULL DEFAULT '';")
|
|
_ensure_column(conn, "recommendations", "tags",
|
|
"ALTER TABLE recommendations ADD COLUMN tags TEXT NOT NULL DEFAULT '[]';")
|
|
|
|
# ✅ 결과 채점용 컬럼 추가
|
|
_ensure_column(conn, "recommendations", "rank",
|
|
"ALTER TABLE recommendations ADD COLUMN rank INTEGER;")
|
|
_ensure_column(conn, "recommendations", "correct_count",
|
|
"ALTER TABLE recommendations ADD COLUMN correct_count INTEGER DEFAULT 0;")
|
|
_ensure_column(conn, "recommendations", "has_bonus",
|
|
"ALTER TABLE recommendations ADD COLUMN has_bonus INTEGER DEFAULT 0;")
|
|
_ensure_column(conn, "recommendations", "checked",
|
|
"ALTER TABLE recommendations ADD COLUMN checked INTEGER DEFAULT 0;")
|
|
|
|
|
|
# ✅ UNIQUE 인덱스(중복 저장 방지)
|
|
conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS uq_reco_dedup ON recommendations(dedup_hash);")
|
|
|
|
# ── 시뮬레이션 테이블 ─────────────────────────────────────────────────
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS simulation_runs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
run_at TEXT NOT NULL DEFAULT (datetime('now')),
|
|
strategy TEXT NOT NULL DEFAULT 'monte_carlo',
|
|
total_generated INTEGER NOT NULL DEFAULT 0,
|
|
top_k_selected INTEGER NOT NULL DEFAULT 0,
|
|
avg_score REAL,
|
|
notes TEXT DEFAULT ''
|
|
);
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_simrun_at ON simulation_runs(run_at DESC);"
|
|
)
|
|
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS simulation_candidates (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
run_id INTEGER NOT NULL,
|
|
numbers TEXT NOT NULL,
|
|
score_total REAL NOT NULL,
|
|
score_frequency REAL,
|
|
score_fingerprint REAL,
|
|
score_gap REAL,
|
|
score_cooccur REAL,
|
|
score_diversity REAL,
|
|
is_best INTEGER DEFAULT 0,
|
|
based_on_draw INTEGER,
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
|
FOREIGN KEY(run_id) REFERENCES simulation_runs(id)
|
|
);
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_simcand_run "
|
|
"ON simulation_candidates(run_id, score_total DESC);"
|
|
)
|
|
conn.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_simcand_best "
|
|
"ON simulation_candidates(is_best, score_total DESC);"
|
|
)
|
|
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS best_picks (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
numbers TEXT NOT NULL,
|
|
score_total REAL NOT NULL,
|
|
rank_in_run INTEGER,
|
|
source_run_id INTEGER,
|
|
based_on_draw INTEGER,
|
|
is_active INTEGER DEFAULT 1,
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
|
FOREIGN KEY(source_run_id) REFERENCES simulation_runs(id)
|
|
);
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_bestpicks_active "
|
|
"ON best_picks(is_active, score_total DESC);"
|
|
)
|
|
|
|
# ── todos 테이블 ───────────────────────────────────────────────────────
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS todos (
|
|
id TEXT PRIMARY KEY
|
|
DEFAULT (lower(hex(randomblob(4))) || '-' || lower(hex(randomblob(2)))),
|
|
title TEXT NOT NULL,
|
|
description TEXT,
|
|
status TEXT NOT NULL DEFAULT 'todo'
|
|
CHECK(status IN ('todo','in_progress','done')),
|
|
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
|
|
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
|
|
);
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_todos_created ON todos(created_at DESC);"
|
|
)
|
|
|
|
# ── blog_posts 테이블 ──────────────────────────────────────────────────
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS blog_posts (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
title TEXT NOT NULL,
|
|
body TEXT NOT NULL DEFAULT '',
|
|
excerpt TEXT NOT NULL DEFAULT '',
|
|
tags TEXT NOT NULL DEFAULT '[]',
|
|
date TEXT NOT NULL DEFAULT (date('now','localtime')),
|
|
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
|
|
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
|
|
);
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_blog_date ON blog_posts(date 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);")
|
|
|
|
# ── purchase_history 컬럼 확장 (기존 데이터 보존) ──────────────────────
|
|
_ensure_column(conn, "purchase_history", "numbers",
|
|
"ALTER TABLE purchase_history ADD COLUMN numbers TEXT NOT NULL DEFAULT '[]'")
|
|
_ensure_column(conn, "purchase_history", "is_real",
|
|
"ALTER TABLE purchase_history ADD COLUMN is_real INTEGER NOT NULL DEFAULT 1")
|
|
_ensure_column(conn, "purchase_history", "source_strategy",
|
|
"ALTER TABLE purchase_history ADD COLUMN source_strategy TEXT NOT NULL DEFAULT 'manual'")
|
|
_ensure_column(conn, "purchase_history", "source_detail",
|
|
"ALTER TABLE purchase_history ADD COLUMN source_detail TEXT NOT NULL DEFAULT '{}'")
|
|
_ensure_column(conn, "purchase_history", "checked",
|
|
"ALTER TABLE purchase_history ADD COLUMN checked INTEGER NOT NULL DEFAULT 0")
|
|
_ensure_column(conn, "purchase_history", "results",
|
|
"ALTER TABLE purchase_history ADD COLUMN results TEXT NOT NULL DEFAULT '[]'")
|
|
_ensure_column(conn, "purchase_history", "total_prize",
|
|
"ALTER TABLE purchase_history ADD COLUMN total_prize INTEGER NOT NULL DEFAULT 0")
|
|
|
|
# ── strategy_performance 테이블 ────────────────────────────────────────
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS strategy_performance (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
strategy TEXT NOT NULL,
|
|
draw_no INTEGER NOT NULL,
|
|
sets_count INTEGER NOT NULL DEFAULT 0,
|
|
total_correct INTEGER NOT NULL DEFAULT 0,
|
|
max_correct INTEGER NOT NULL DEFAULT 0,
|
|
prize_total INTEGER NOT NULL DEFAULT 0,
|
|
avg_score REAL NOT NULL DEFAULT 0.0,
|
|
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
|
|
UNIQUE(strategy, draw_no)
|
|
);
|
|
"""
|
|
)
|
|
|
|
# ── strategy_weights 테이블 ────────────────────────────────────────────
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS strategy_weights (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
strategy TEXT NOT NULL UNIQUE,
|
|
weight REAL NOT NULL DEFAULT 0.2,
|
|
ema_score REAL NOT NULL DEFAULT 0.15,
|
|
total_sets INTEGER NOT NULL DEFAULT 0,
|
|
total_hits_3plus INTEGER NOT NULL DEFAULT 0,
|
|
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
|
|
);
|
|
"""
|
|
)
|
|
|
|
# strategy_weights 초기값 시드 (이미 있으면 무시)
|
|
_INIT_WEIGHTS = [
|
|
("combined", 0.30, 0.15),
|
|
("simulation", 0.25, 0.15),
|
|
("heatmap", 0.20, 0.15),
|
|
("manual", 0.15, 0.15),
|
|
("custom", 0.10, 0.15),
|
|
]
|
|
for strat, w, ema in _INIT_WEIGHTS:
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO strategy_weights (strategy, weight, ema_score) VALUES (?, ?, ?)",
|
|
(strat, w, ema),
|
|
)
|
|
|
|
# ── 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'))
|
|
);
|
|
"""
|
|
)
|
|
|
|
# ── 추가 인덱스 ───────────────────────────────────────────────────────
|
|
conn.execute("CREATE INDEX IF NOT EXISTS idx_reco_based_checked ON recommendations(based_on_draw, checked)")
|
|
conn.execute("CREATE INDEX IF NOT EXISTS idx_purchase_strategy ON purchase_history(source_strategy)")
|
|
conn.execute("CREATE INDEX IF NOT EXISTS idx_purchase_checked ON purchase_history(draw_no, checked)")
|
|
|
|
|
|
# ── todos CRUD ───────────────────────────────────────────────────────────────
|
|
|
|
def _todo_row_to_dict(r) -> Dict[str, Any]:
|
|
return {
|
|
"id": r["id"],
|
|
"title": r["title"],
|
|
"description": r["description"],
|
|
"status": r["status"],
|
|
"created_at": r["created_at"],
|
|
"updated_at": r["updated_at"],
|
|
}
|
|
|
|
|
|
def get_all_todos() -> List[Dict[str, Any]]:
|
|
with _conn() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM todos ORDER BY created_at DESC"
|
|
).fetchall()
|
|
return [_todo_row_to_dict(r) for r in rows]
|
|
|
|
|
|
def create_todo(title: str, description: Optional[str], status: str) -> Dict[str, Any]:
|
|
with _conn() as conn:
|
|
conn.execute(
|
|
"INSERT INTO todos (title, description, status) VALUES (?, ?, ?)",
|
|
(title, description, status),
|
|
)
|
|
row = conn.execute(
|
|
"SELECT * FROM todos WHERE rowid = last_insert_rowid()"
|
|
).fetchone()
|
|
return _todo_row_to_dict(row)
|
|
|
|
|
|
def update_todo(todo_id: str, fields: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""fields에 있는 항목만 업데이트 (PATCH 방식), updated_at 자동 갱신"""
|
|
allowed = {"title", "description", "status"}
|
|
updates = {k: v for k, v in fields.items() if k in allowed}
|
|
if not updates:
|
|
with _conn() as conn:
|
|
row = conn.execute("SELECT * FROM todos WHERE id = ?", (todo_id,)).fetchone()
|
|
return _todo_row_to_dict(row) if row else None
|
|
|
|
set_clauses = ", ".join(f"{k} = ?" for k in updates)
|
|
set_clauses += ", updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')"
|
|
args = list(updates.values()) + [todo_id]
|
|
|
|
with _conn() as conn:
|
|
conn.execute(
|
|
f"UPDATE todos SET {set_clauses} WHERE id = ?",
|
|
args,
|
|
)
|
|
row = conn.execute("SELECT * FROM todos WHERE id = ?", (todo_id,)).fetchone()
|
|
return _todo_row_to_dict(row) if row else None
|
|
|
|
|
|
def delete_todo(todo_id: str) -> bool:
|
|
with _conn() as conn:
|
|
cur = conn.execute("DELETE FROM todos WHERE id = ?", (todo_id,))
|
|
return cur.rowcount > 0
|
|
|
|
|
|
def delete_done_todos() -> int:
|
|
with _conn() as conn:
|
|
cur = conn.execute("DELETE FROM todos WHERE status = 'done'")
|
|
return cur.rowcount
|
|
|
|
|
|
# ── blog_posts CRUD ──────────────────────────────────────────────────────────
|
|
|
|
def _post_row_to_dict(r) -> Dict[str, Any]:
|
|
return {
|
|
"id": r["id"],
|
|
"title": r["title"],
|
|
"body": r["body"],
|
|
"excerpt": r["excerpt"],
|
|
"tags": json.loads(r["tags"]) if r["tags"] else [],
|
|
"date": r["date"],
|
|
"created_at": r["created_at"],
|
|
"updated_at": r["updated_at"],
|
|
}
|
|
|
|
|
|
def get_all_posts() -> List[Dict[str, Any]]:
|
|
with _conn() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM blog_posts ORDER BY date DESC, id DESC"
|
|
).fetchall()
|
|
return [_post_row_to_dict(r) for r in rows]
|
|
|
|
|
|
def create_post(title: str, body: str, excerpt: str, tags: List[str], date: str) -> Dict[str, Any]:
|
|
with _conn() as conn:
|
|
conn.execute(
|
|
"INSERT INTO blog_posts (title, body, excerpt, tags, date) VALUES (?, ?, ?, ?, ?)",
|
|
(title, body, excerpt, json.dumps(tags), date),
|
|
)
|
|
row = conn.execute(
|
|
"SELECT * FROM blog_posts WHERE rowid = last_insert_rowid()"
|
|
).fetchone()
|
|
return _post_row_to_dict(row)
|
|
|
|
|
|
def update_post(post_id: int, fields: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
allowed = {"title", "body", "excerpt", "tags", "date"}
|
|
updates = {k: v for k, v in fields.items() if k in allowed}
|
|
if not updates:
|
|
with _conn() as conn:
|
|
row = conn.execute("SELECT * FROM blog_posts WHERE id = ?", (post_id,)).fetchone()
|
|
return _post_row_to_dict(row) if row else None
|
|
|
|
if "tags" in updates:
|
|
updates["tags"] = json.dumps(updates["tags"])
|
|
|
|
set_clauses = ", ".join(f"{k} = ?" for k in updates)
|
|
set_clauses += ", updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')"
|
|
args = list(updates.values()) + [post_id]
|
|
|
|
with _conn() as conn:
|
|
conn.execute(f"UPDATE blog_posts SET {set_clauses} WHERE id = ?", args)
|
|
row = conn.execute("SELECT * FROM blog_posts WHERE id = ?", (post_id,)).fetchone()
|
|
return _post_row_to_dict(row) if row else None
|
|
|
|
|
|
def delete_post(post_id: int) -> bool:
|
|
with _conn() as conn:
|
|
cur = conn.execute("DELETE FROM blog_posts WHERE id = ?", (post_id,))
|
|
return cur.rowcount > 0
|
|
|
|
|
|
def upsert_draw(row: Dict[str, Any]) -> None:
|
|
with _conn() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(drw_no) DO UPDATE SET
|
|
drw_date=excluded.drw_date,
|
|
n1=excluded.n1, n2=excluded.n2, n3=excluded.n3,
|
|
n4=excluded.n4, n5=excluded.n5, n6=excluded.n6,
|
|
bonus=excluded.bonus,
|
|
updated_at=datetime('now')
|
|
""",
|
|
(
|
|
int(row["drw_no"]),
|
|
str(row["drw_date"]),
|
|
int(row["n1"]), int(row["n2"]), int(row["n3"]),
|
|
int(row["n4"]), int(row["n5"]), int(row["n6"]),
|
|
int(row["bonus"]),
|
|
),
|
|
)
|
|
|
|
def upsert_many_draws(rows: List[Dict[str, Any]]) -> None:
|
|
data = [
|
|
(
|
|
int(r["drw_no"]), str(r["drw_date"]),
|
|
int(r["n1"]), int(r["n2"]), int(r["n3"]),
|
|
int(r["n4"]), int(r["n5"]), int(r["n6"]),
|
|
int(r["bonus"])
|
|
) for r in rows
|
|
]
|
|
with _conn() as conn:
|
|
conn.executemany(
|
|
"""
|
|
INSERT INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(drw_no) DO UPDATE SET
|
|
drw_date=excluded.drw_date,
|
|
n1=excluded.n1, n2=excluded.n2, n3=excluded.n3,
|
|
n4=excluded.n4, n5=excluded.n5, n6=excluded.n6,
|
|
bonus=excluded.bonus,
|
|
updated_at=datetime('now')
|
|
""",
|
|
data
|
|
)
|
|
|
|
def get_latest_draw() -> Optional[Dict[str, Any]]:
|
|
with _conn() as conn:
|
|
r = conn.execute("SELECT * FROM draws ORDER BY drw_no DESC LIMIT 1").fetchone()
|
|
return dict(r) if r else None
|
|
|
|
def get_draw(drw_no: int) -> Optional[Dict[str, Any]]:
|
|
with _conn() as conn:
|
|
r = conn.execute("SELECT * FROM draws WHERE drw_no = ?", (drw_no,)).fetchone()
|
|
return dict(r) if r else None
|
|
|
|
def count_draws() -> int:
|
|
with _conn() as conn:
|
|
r = conn.execute("SELECT COUNT(*) AS c FROM draws").fetchone()
|
|
return int(r["c"])
|
|
|
|
def get_all_draw_numbers():
|
|
with _conn() as conn:
|
|
rows = conn.execute(
|
|
"SELECT drw_no, n1, n2, n3, n4, n5, n6 FROM draws ORDER BY drw_no ASC"
|
|
).fetchall()
|
|
return [(int(r["drw_no"]), [int(r["n1"]), int(r["n2"]), int(r["n3"]), int(r["n4"]), int(r["n5"]), int(r["n6"])]) for r in rows]
|
|
|
|
# ---------- ✅ recommendation helpers ----------
|
|
|
|
def _canonical_params(params: dict) -> str:
|
|
return json.dumps(params, sort_keys=True, separators=(",", ":"))
|
|
|
|
def _numbers_sorted_str(numbers: List[int]) -> str:
|
|
return ",".join(str(x) for x in sorted(numbers))
|
|
|
|
def _dedup_hash(based_on_draw: Optional[int], numbers: List[int], params: dict) -> str:
|
|
s = f"{based_on_draw or ''}|{_numbers_sorted_str(numbers)}|{_canonical_params(params)}"
|
|
return hashlib.sha1(s.encode("utf-8")).hexdigest()
|
|
|
|
def save_recommendation_dedup(based_on_draw: Optional[int], numbers: List[int], params: dict) -> Dict[str, Any]:
|
|
"""
|
|
✅ 동일 추천(번호+params+based_on_draw)이면 중복 저장 없이 기존 id 반환
|
|
"""
|
|
ns = _numbers_sorted_str(numbers)
|
|
h = _dedup_hash(based_on_draw, numbers, params)
|
|
|
|
with _conn() as conn:
|
|
# 이미 있으면 반환
|
|
r = conn.execute("SELECT id FROM recommendations WHERE dedup_hash = ?", (h,)).fetchone()
|
|
if r:
|
|
return {"id": int(r["id"]), "saved": False, "deduped": True}
|
|
|
|
cur = conn.execute(
|
|
"""
|
|
INSERT INTO recommendations (based_on_draw, numbers, params, numbers_sorted, dedup_hash)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""",
|
|
(based_on_draw, json.dumps(numbers), json.dumps(params), ns, h),
|
|
)
|
|
return {"id": int(cur.lastrowid), "saved": True, "deduped": False}
|
|
|
|
def list_recommendations_ex(
|
|
limit: int = 30,
|
|
offset: int = 0,
|
|
favorite: Optional[bool] = None,
|
|
tag: Optional[str] = None,
|
|
q: Optional[str] = None,
|
|
sort: str = "id_desc", # id_desc|created_desc|favorite_desc
|
|
) -> List[Dict[str, Any]]:
|
|
where = []
|
|
args: list[Any] = []
|
|
|
|
if favorite is not None:
|
|
where.append("favorite = ?")
|
|
args.append(1 if favorite else 0)
|
|
|
|
if q:
|
|
where.append("note LIKE ?")
|
|
args.append(f"%{q}%")
|
|
|
|
# tags는 JSON 문자열이므로 단순 LIKE로 처리(가볍게 시작)
|
|
if tag:
|
|
where.append("tags LIKE ?")
|
|
args.append(f"%{tag}%")
|
|
|
|
where_sql = ("WHERE " + " AND ".join(where)) if where else ""
|
|
|
|
if sort == "created_desc":
|
|
order = "created_at DESC"
|
|
elif sort == "favorite_desc":
|
|
# favorite(1)이 먼저, 그 다음 최신
|
|
order = "favorite DESC, id DESC"
|
|
else:
|
|
order = "id DESC"
|
|
|
|
sql = f"""
|
|
SELECT id, created_at, based_on_draw, numbers, params, favorite, note, tags
|
|
FROM recommendations
|
|
{where_sql}
|
|
ORDER BY {order}
|
|
LIMIT ? OFFSET ?
|
|
"""
|
|
args.extend([int(limit), int(offset)])
|
|
|
|
with _conn() as conn:
|
|
rows = conn.execute(sql, args).fetchall()
|
|
|
|
out = []
|
|
for r in rows:
|
|
out.append({
|
|
"id": int(r["id"]),
|
|
"created_at": r["created_at"],
|
|
"based_on_draw": r["based_on_draw"],
|
|
"numbers": json.loads(r["numbers"]),
|
|
"params": json.loads(r["params"]),
|
|
"favorite": bool(r["favorite"]) if r["favorite"] is not None else False,
|
|
"note": r["note"],
|
|
"tags": json.loads(r["tags"]) if r["tags"] else [],
|
|
})
|
|
return out
|
|
|
|
def update_recommendation(rec_id: int, favorite: Optional[bool] = None, note: Optional[str] = None, tags: Optional[List[str]] = None) -> bool:
|
|
fields = []
|
|
args: list[Any] = []
|
|
|
|
if favorite is not None:
|
|
fields.append("favorite = ?")
|
|
args.append(1 if favorite else 0)
|
|
if note is not None:
|
|
fields.append("note = ?")
|
|
args.append(note)
|
|
if tags is not None:
|
|
fields.append("tags = ?")
|
|
args.append(json.dumps(tags))
|
|
|
|
if not fields:
|
|
return False
|
|
|
|
args.append(rec_id)
|
|
|
|
with _conn() as conn:
|
|
cur = conn.execute(
|
|
f"UPDATE recommendations SET {', '.join(fields)} WHERE id = ?",
|
|
args,
|
|
)
|
|
return cur.rowcount > 0
|
|
|
|
def delete_recommendation(rec_id: int) -> bool:
|
|
with _conn() as conn:
|
|
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(
|
|
"""
|
|
UPDATE recommendations
|
|
SET rank = ?, correct_count = ?, has_bonus = ?, checked = 1
|
|
WHERE id = ?
|
|
""",
|
|
(rank, correct_count, 1 if has_bonus else 0, rec_id)
|
|
)
|
|
return cur.rowcount > 0
|
|
|
|
|
|
# ── 시뮬레이션 CRUD ─────────────────────────────────────────────────────────
|
|
|
|
def save_simulation_run(
|
|
strategy: str,
|
|
total_generated: int,
|
|
top_k_selected: int,
|
|
avg_score: float,
|
|
notes: str = "",
|
|
) -> int:
|
|
"""시뮬레이션 실행 기록 저장, 생성된 ID 반환"""
|
|
with _conn() as conn:
|
|
cur = conn.execute(
|
|
"""
|
|
INSERT INTO simulation_runs (strategy, total_generated, top_k_selected, avg_score, notes)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""",
|
|
(strategy, total_generated, top_k_selected, round(avg_score, 6), notes),
|
|
)
|
|
return int(cur.lastrowid)
|
|
|
|
|
|
def save_simulation_candidates_bulk(
|
|
run_id: int,
|
|
candidates: List[Dict[str, Any]],
|
|
based_on_draw: Optional[int],
|
|
) -> None:
|
|
"""
|
|
상위 후보들을 simulation_candidates 테이블에 일괄 저장.
|
|
candidates 각 항목: {"numbers": [...], "score_total": ..., "score_*": ..., "is_best": bool}
|
|
"""
|
|
data = [
|
|
(
|
|
run_id,
|
|
json.dumps(sorted(c["numbers"])),
|
|
c["score_total"],
|
|
c.get("score_frequency"),
|
|
c.get("score_fingerprint"),
|
|
c.get("score_gap"),
|
|
c.get("score_cooccur"),
|
|
c.get("score_diversity"),
|
|
1 if c.get("is_best") else 0,
|
|
based_on_draw,
|
|
)
|
|
for c in candidates
|
|
]
|
|
with _conn() as conn:
|
|
conn.executemany(
|
|
"""
|
|
INSERT INTO simulation_candidates
|
|
(run_id, numbers, score_total, score_frequency, score_fingerprint,
|
|
score_gap, score_cooccur, score_diversity, is_best, based_on_draw)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
data,
|
|
)
|
|
|
|
|
|
def replace_best_picks(
|
|
picks: List[Dict[str, Any]],
|
|
run_id: int,
|
|
based_on_draw: Optional[int],
|
|
) -> None:
|
|
"""
|
|
기존 활성 best_picks를 비활성화하고 새 picks로 교체.
|
|
picks 각 항목: {"numbers": [...], "score_total": ..., "rank_in_run": int}
|
|
"""
|
|
with _conn() as conn:
|
|
conn.execute("UPDATE best_picks SET is_active = 0 WHERE is_active = 1")
|
|
data = [
|
|
(
|
|
json.dumps(sorted(p["numbers"])),
|
|
p["score_total"],
|
|
p.get("rank_in_run"),
|
|
run_id,
|
|
based_on_draw,
|
|
)
|
|
for p in picks
|
|
]
|
|
conn.executemany(
|
|
"""
|
|
INSERT INTO best_picks (numbers, score_total, rank_in_run, source_run_id, based_on_draw, is_active)
|
|
VALUES (?, ?, ?, ?, ?, 1)
|
|
""",
|
|
data,
|
|
)
|
|
|
|
|
|
def get_best_picks(limit: int = 20) -> List[Dict[str, Any]]:
|
|
"""현재 활성화된 best_picks 조회 (점수 내림차순)"""
|
|
with _conn() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, numbers, score_total, rank_in_run, source_run_id, based_on_draw, created_at
|
|
FROM best_picks
|
|
WHERE is_active = 1
|
|
ORDER BY score_total DESC
|
|
LIMIT ?
|
|
""",
|
|
(limit,),
|
|
).fetchall()
|
|
return [
|
|
{
|
|
"id": int(r["id"]),
|
|
"numbers": json.loads(r["numbers"]),
|
|
"score_total": r["score_total"],
|
|
"rank_in_run": r["rank_in_run"],
|
|
"source_run_id": r["source_run_id"],
|
|
"based_on_draw": r["based_on_draw"],
|
|
"created_at": r["created_at"],
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
|
|
def get_simulation_runs(limit: int = 10) -> List[Dict[str, Any]]:
|
|
"""최근 시뮬레이션 실행 기록 조회"""
|
|
with _conn() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, run_at, strategy, total_generated, top_k_selected, avg_score, notes
|
|
FROM simulation_runs
|
|
ORDER BY id DESC
|
|
LIMIT ?
|
|
""",
|
|
(limit,),
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
def get_simulation_candidates(run_id: int, limit: int = 100) -> List[Dict[str, Any]]:
|
|
"""특정 시뮬레이션 실행의 후보 목록 조회 (점수 내림차순)"""
|
|
with _conn() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, numbers, score_total, score_frequency, score_fingerprint,
|
|
score_gap, score_cooccur, score_diversity, is_best, based_on_draw, created_at
|
|
FROM simulation_candidates
|
|
WHERE run_id = ?
|
|
ORDER BY score_total DESC
|
|
LIMIT ?
|
|
""",
|
|
(run_id, limit),
|
|
).fetchall()
|
|
return [
|
|
{**dict(r), "numbers": json.loads(r["numbers"])}
|
|
for r in rows
|
|
]
|
|
|
|
|
|
# ── purchase_history CRUD ─────────────────────────────────────────────────────
|
|
|
|
def _purchase_row_to_dict(r) -> Dict[str, Any]:
|
|
|
|
keys = r.keys()
|
|
numbers_raw = r["numbers"] if "numbers" in keys else "[]"
|
|
detail_raw = r["source_detail"] if "source_detail" in keys else "{}"
|
|
results_raw = r["results"] if "results" in keys else "[]"
|
|
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"],
|
|
"numbers": json.loads(numbers_raw) if numbers_raw else [],
|
|
"is_real": r["is_real"] if "is_real" in keys else 1,
|
|
"source_strategy": r["source_strategy"] if "source_strategy" in keys else "manual",
|
|
"source_detail": json.loads(detail_raw) if detail_raw else {},
|
|
"checked": r["checked"] if "checked" in keys else 0,
|
|
"results": json.loads(results_raw) if results_raw else [],
|
|
"total_prize": r["total_prize"] if "total_prize" in keys else 0,
|
|
}
|
|
|
|
|
|
def add_purchase(draw_no: int, amount: int, sets: int, prize: int = 0, note: str = "",
|
|
numbers: list = None, is_real: bool = True,
|
|
source_strategy: str = "manual", source_detail: dict = None) -> Dict[str, Any]:
|
|
|
|
numbers_json = json.dumps(numbers or [], ensure_ascii=False)
|
|
detail_json = json.dumps(source_detail or {}, ensure_ascii=False)
|
|
is_real_int = 1 if is_real else 0
|
|
with _conn() as conn:
|
|
conn.execute(
|
|
"""INSERT INTO purchase_history
|
|
(draw_no, amount, sets, prize, note, numbers, is_real, source_strategy, source_detail)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(draw_no, amount, sets, prize, note, numbers_json, is_real_int, source_strategy, detail_json),
|
|
)
|
|
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,
|
|
is_real: bool = None, strategy: str = None,
|
|
checked: bool = 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}")
|
|
if is_real is not None:
|
|
conditions.append("is_real = ?")
|
|
params.append(1 if is_real else 0)
|
|
if strategy is not None:
|
|
conditions.append("source_strategy = ?")
|
|
params.append(strategy)
|
|
if checked is not None:
|
|
conditions.append("checked = ?")
|
|
params.append(1 if checked else 0)
|
|
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", "numbers", "is_real", "source_strategy"}
|
|
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
|
|
# SQLite에 전달 전 타입 변환
|
|
if "numbers" in updates:
|
|
updates["numbers"] = json.dumps(updates["numbers"], ensure_ascii=False)
|
|
if "is_real" in updates:
|
|
updates["is_real"] = 1 if updates["is_real"] else 0
|
|
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]:
|
|
|
|
|
|
def _calc_group(rows):
|
|
if not rows:
|
|
return {"sets": 0, "invested": 0, "prize": 0, "roi": 0.0, "win_rate": 0.0}
|
|
invested = sum(r["amount"] for r in rows)
|
|
prize = sum(r.get("total_prize") or r["prize"] for r in rows)
|
|
wins = sum(1 for r in rows if (r.get("total_prize") or r["prize"]) > 0)
|
|
return {
|
|
"sets": sum(r["sets"] for r in rows),
|
|
"invested": invested,
|
|
"prize": prize,
|
|
"roi": round((prize / invested * 100 - 100) if invested else 0.0, 2),
|
|
"win_rate": round(wins / len(rows) * 100, 2) if rows else 0.0,
|
|
}
|
|
|
|
with _conn() as conn:
|
|
rows = conn.execute("SELECT * FROM purchase_history").fetchall()
|
|
|
|
all_rows = [dict(r) for r in rows]
|
|
real_rows = [r for r in all_rows if r.get("is_real", 1) == 1]
|
|
virtual_rows = [r for r in all_rows if r.get("is_real", 1) == 0]
|
|
|
|
# 전략별 집계
|
|
by_strategy: Dict[str, list] = {}
|
|
for r in all_rows:
|
|
strat = r.get("source_strategy", "manual")
|
|
if strat not in by_strategy:
|
|
by_strategy[strat] = []
|
|
by_strategy[strat].append(r)
|
|
|
|
strategy_stats: Dict[str, Any] = {}
|
|
for strat, srows in by_strategy.items():
|
|
s = _calc_group(srows)
|
|
total_correct = 0
|
|
count_sets = 0
|
|
hits_3plus = 0
|
|
for r in srows:
|
|
results_raw = r.get("results", "[]")
|
|
try:
|
|
results = json.loads(results_raw) if isinstance(results_raw, str) else (results_raw or [])
|
|
except Exception:
|
|
results = []
|
|
for res in results:
|
|
count_sets += 1
|
|
c = res.get("correct", 0)
|
|
total_correct += c
|
|
if c >= 3:
|
|
hits_3plus += 1
|
|
s["avg_correct"] = round(total_correct / count_sets, 2) if count_sets else 0.0
|
|
s["hits_3plus"] = hits_3plus
|
|
strategy_stats[strat] = s
|
|
|
|
total_invested = sum(r["amount"] for r in all_rows)
|
|
total_prize_sum = sum(r.get("total_prize") or r["prize"] for r in all_rows)
|
|
return {
|
|
"total": _calc_group(all_rows),
|
|
"real": _calc_group(real_rows),
|
|
"virtual": _calc_group(virtual_rows),
|
|
"by_strategy": strategy_stats,
|
|
# 하위호환
|
|
"total_records": len(all_rows),
|
|
"total_invested": total_invested,
|
|
"total_prize": total_prize_sum,
|
|
"net": total_prize_sum - total_invested,
|
|
"return_rate": round((total_prize_sum / total_invested * 100) if total_invested else 0.0, 2),
|
|
"prize_count": sum(1 for r in all_rows if (r.get("total_prize") or r["prize"]) > 0),
|
|
"max_prize": max((r.get("total_prize") or r["prize"] for r in all_rows), default=0),
|
|
}
|
|
|
|
|
|
# ── 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
|
|
|
|
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]
|
|
|
|
|
|
# ── strategy_performance CRUD ─────────────────────────────────────────────────
|
|
|
|
def upsert_strategy_performance(strategy: str, draw_no: int, sets_count: int = 0,
|
|
total_correct: int = 0, max_correct: int = 0,
|
|
prize_total: int = 0, avg_score: float = 0.0) -> None:
|
|
with _conn() as conn:
|
|
conn.execute(
|
|
"""INSERT INTO strategy_performance (strategy, draw_no, sets_count, total_correct, max_correct, prize_total, avg_score)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(strategy, draw_no) DO UPDATE SET
|
|
sets_count=excluded.sets_count, total_correct=excluded.total_correct,
|
|
max_correct=excluded.max_correct, prize_total=excluded.prize_total,
|
|
avg_score=excluded.avg_score,
|
|
updated_at=strftime('%Y-%m-%dT%H:%M:%fZ','now')""",
|
|
(strategy, draw_no, sets_count, total_correct, max_correct, prize_total, avg_score),
|
|
)
|
|
|
|
|
|
def get_strategy_performance(strategy: str = None, days: int = None) -> List[Dict[str, Any]]:
|
|
conditions, params = [], []
|
|
if strategy:
|
|
conditions.append("strategy = ?")
|
|
params.append(strategy)
|
|
if days:
|
|
conditions.append("updated_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 strategy_performance {where} ORDER BY draw_no ASC",
|
|
params,
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
# ── strategy_weights CRUD ─────────────────────────────────────────────────────
|
|
|
|
def get_strategy_weights() -> List[Dict[str, Any]]:
|
|
with _conn() as conn:
|
|
rows = conn.execute("SELECT * FROM strategy_weights ORDER BY weight DESC").fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
def update_strategy_weight(strategy: str, weight: float, ema_score: float,
|
|
total_sets: int = None, total_hits_3plus: int = None) -> None:
|
|
with _conn() as conn:
|
|
fields = "weight=?, ema_score=?, updated_at=strftime('%Y-%m-%dT%H:%M:%fZ','now')"
|
|
params = [weight, ema_score]
|
|
if total_sets is not None:
|
|
fields += ", total_sets=?"
|
|
params.append(total_sets)
|
|
if total_hits_3plus is not None:
|
|
fields += ", total_hits_3plus=?"
|
|
params.append(total_hits_3plus)
|
|
params.append(strategy)
|
|
conn.execute(f"UPDATE strategy_weights SET {fields} WHERE strategy=?", params)
|
|
|
|
|
|
def update_purchase_results(purchase_id: int, results: list, total_prize: int) -> None:
|
|
"""구매 건의 결과를 갱신 (체커 호출 후)"""
|
|
|
|
with _conn() as conn:
|
|
conn.execute(
|
|
"UPDATE purchase_history SET results=?, total_prize=?, checked=1 WHERE id=?",
|
|
(json.dumps(results, ensure_ascii=False), total_prize, purchase_id),
|
|
)
|
|
|