lotto-lab: 구매 CRUD 확장 + strategy_performance/weights CRUD 추가
- _purchase_row_to_dict: numbers/is_real/source_detail/results/total_prize 신규 컬럼 포함 - add_purchase: numbers, is_real, source_strategy, source_detail 파라미터 추가 - get_purchases: is_real, strategy, checked 필터 추가 - get_purchase_stats: total/real/virtual/by_strategy 분리 통계 + 하위호환 필드 유지 - update_purchase: allowed 셋에 numbers/is_real/source_strategy 추가 - 신규: upsert_strategy_performance, get_strategy_performance - 신규: get_strategy_weights, update_strategy_weight - 신규: update_purchase_results (체커 연동용) - 테스트 5건 추가 (TDD) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -810,6 +810,11 @@ def get_simulation_candidates(run_id: int, limit: int = 100) -> List[Dict[str, A
|
||||
# ── purchase_history CRUD ─────────────────────────────────────────────────────
|
||||
|
||||
def _purchase_row_to_dict(r) -> Dict[str, Any]:
|
||||
import json as _json
|
||||
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"],
|
||||
@@ -818,20 +823,37 @@ def _purchase_row_to_dict(r) -> Dict[str, Any]:
|
||||
"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 = "") -> Dict[str, Any]:
|
||||
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]:
|
||||
import json as _json
|
||||
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) VALUES (?, ?, ?, ?, ?)",
|
||||
(draw_no, amount, sets, prize, note),
|
||||
"""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) -> List[Dict[str, Any]]:
|
||||
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 = ?")
|
||||
@@ -839,6 +861,15 @@ def get_purchases(draw_no: int = None, days: int = None) -> List[Dict[str, Any]]
|
||||
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(
|
||||
@@ -849,7 +880,7 @@ def get_purchases(draw_no: int = None, days: int = None) -> List[Dict[str, Any]]
|
||||
|
||||
|
||||
def update_purchase(purchase_id: int, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||||
allowed = {"draw_no", "amount", "sets", "prize", "note"}
|
||||
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:
|
||||
@@ -874,30 +905,74 @@ def delete_purchase(purchase_id: int) -> bool:
|
||||
|
||||
|
||||
def get_purchase_stats() -> Dict[str, Any]:
|
||||
with _conn() as conn:
|
||||
rows = conn.execute("SELECT amount, prize FROM purchase_history").fetchall()
|
||||
if not rows:
|
||||
import json as _json
|
||||
|
||||
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 {
|
||||
"total_records": 0,
|
||||
"total_invested": 0,
|
||||
"total_prize": 0,
|
||||
"net": 0,
|
||||
"return_rate": 0.0,
|
||||
"prize_count": 0,
|
||||
"max_prize": 0,
|
||||
"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,
|
||||
}
|
||||
amounts = [r["amount"] for r in rows]
|
||||
prizes = [r["prize"] for r in rows]
|
||||
total_invested = sum(amounts)
|
||||
total_prize = sum(prizes)
|
||||
|
||||
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_records": len(rows),
|
||||
"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,
|
||||
"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),
|
||||
"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),
|
||||
}
|
||||
|
||||
|
||||
@@ -945,4 +1020,70 @@ def get_all_recommendation_numbers() -> List[List[int]]:
|
||||
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:
|
||||
"""구매 건의 결과를 갱신 (체커 호출 후)"""
|
||||
import json as _json
|
||||
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),
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user