feat(lotto-signals): lotto_signals/lotto_baselines 테이블 + CRUD

agent-office DB에 lotto_signals, lotto_baselines 테이블 추가 및
insert/mark/query/upsert CRUD 헬퍼 함수 구현 (throttle, z-score, baseline 관리)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-05-20 02:43:27 +09:00
parent e5465ad136
commit 9e1001b935

View File

@@ -98,6 +98,39 @@ def init_db() -> None:
completed_at TEXT
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS lotto_signals (
id INTEGER PRIMARY KEY AUTOINCREMENT,
triggered_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
source TEXT NOT NULL,
metric TEXT NOT NULL,
value REAL NOT NULL,
baseline_mu REAL,
baseline_sigma REAL,
z_score REAL,
fire_level TEXT NOT NULL,
notified_at TEXT,
payload TEXT
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_ls_triggered
ON lotto_signals(triggered_at DESC)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_ls_fire
ON lotto_signals(fire_level, notified_at)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS lotto_baselines (
metric TEXT PRIMARY KEY,
window_values TEXT NOT NULL DEFAULT '[]',
mu REAL NOT NULL DEFAULT 0.0,
sigma REAL NOT NULL DEFAULT 0.0,
last_pushed_draw_no INTEGER,
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
)
""")
# Seed default agent configs
for agent_id, name in [
("stock", "주식 트레이더"),
@@ -556,3 +589,153 @@ def get_latest_youtube_research_job() -> Optional[Dict[str, Any]]:
"started_at": row["started_at"],
"completed_at": row["completed_at"],
}
# --- lotto_signals / lotto_baselines CRUD ---
def insert_lotto_signal(
source: str,
metric: str,
value: float,
baseline_mu: Optional[float],
baseline_sigma: Optional[float],
z_score: Optional[float],
fire_level: str,
payload: Optional[Dict[str, Any]] = None,
) -> int:
with _conn() as conn:
cur = conn.execute(
"""
INSERT INTO lotto_signals
(source, metric, value, baseline_mu, baseline_sigma, z_score, fire_level, payload)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
source, metric, value,
baseline_mu, baseline_sigma, z_score, fire_level,
json.dumps(payload or {}, ensure_ascii=False),
),
)
return cur.lastrowid
def mark_signal_notified(signal_id: int) -> None:
with _conn() as conn:
conn.execute(
"UPDATE lotto_signals SET notified_at = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id = ?",
(signal_id,),
)
def get_recent_lotto_signals(hours: int = 24, min_fire: str = "normal") -> List[Dict[str, Any]]:
"""지난 N시간 발화 시그널. min_fire='normal'이면 normal+urgent."""
levels = ("urgent",) if min_fire == "urgent" else ("normal", "urgent")
placeholders = ",".join("?" * len(levels))
with _conn() as conn:
rows = conn.execute(
f"""
SELECT * FROM lotto_signals
WHERE triggered_at >= datetime('now', ?)
AND fire_level IN ({placeholders})
ORDER BY triggered_at DESC
""",
(f"-{int(hours)} hours", *levels),
).fetchall()
return [dict(r) for r in rows]
def get_signals_history(days: int = 7) -> List[Dict[str, Any]]:
"""차트/이력 페이지용 — 모든 fire_level 포함."""
with _conn() as conn:
rows = conn.execute(
"""
SELECT * FROM lotto_signals
WHERE triggered_at >= datetime('now', ?)
ORDER BY triggered_at DESC
""",
(f"-{int(days)} days",),
).fetchall()
return [dict(r) for r in rows]
def get_recent_urgent_count(hours: int = 24) -> int:
with _conn() as conn:
row = conn.execute(
"""
SELECT COUNT(*) AS c FROM lotto_signals
WHERE triggered_at >= datetime('now', ?)
AND fire_level = 'urgent'
AND notified_at IS NOT NULL
""",
(f"-{int(hours)} hours",),
).fetchone()
return int(row["c"]) if row else 0
def get_last_signal_notification(metric: str, fire_level: str, hours: int) -> Optional[str]:
"""같은 metric+fire_level이 hours 내에 알림 발송된 마지막 시각. throttle용."""
with _conn() as conn:
row = conn.execute(
"""
SELECT notified_at FROM lotto_signals
WHERE metric = ?
AND fire_level = ?
AND notified_at IS NOT NULL
AND notified_at >= datetime('now', ?)
ORDER BY notified_at DESC LIMIT 1
""",
(metric, fire_level, f"-{int(hours)} hours"),
).fetchone()
return row["notified_at"] if row else None
def get_baseline(metric: str) -> Optional[Dict[str, Any]]:
with _conn() as conn:
row = conn.execute(
"SELECT * FROM lotto_baselines WHERE metric = ?",
(metric,),
).fetchone()
if not row:
return None
d = dict(row)
d["window_values"] = json.loads(d["window_values"])
return d
def upsert_baseline(
metric: str,
window_values: List[float],
mu: float,
sigma: float,
last_pushed_draw_no: Optional[int],
) -> None:
with _conn() as conn:
conn.execute(
"""
INSERT INTO lotto_baselines
(metric, window_values, mu, sigma, last_pushed_draw_no, updated_at)
VALUES (?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
ON CONFLICT(metric) DO UPDATE SET
window_values = excluded.window_values,
mu = excluded.mu,
sigma = excluded.sigma,
last_pushed_draw_no = excluded.last_pushed_draw_no,
updated_at = excluded.updated_at
""",
(
metric,
json.dumps(window_values),
mu, sigma, last_pushed_draw_no,
),
)
def get_all_baselines() -> List[Dict[str, Any]]:
with _conn() as conn:
rows = conn.execute("SELECT * FROM lotto_baselines ORDER BY metric").fetchall()
out = []
for r in rows:
d = dict(r)
d["window_values"] = json.loads(d["window_values"])
out.append(d)
return out