diff --git a/agent-office/app/db.py b/agent-office/app/db.py
index 81f95d7..66ae60f 100644
--- a/agent-office/app/db.py
+++ b/agent-office/app/db.py
@@ -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