feat: collect Bybit orderbook observations for training

.env.example

@@ -30,6 +30,8 @@ FAST_LOOP_INTERVAL_SECONDS=1
 FAST_ENTRY_COOLDOWN_SECONDS=20
 MAX_ENTRIES_PER_MINUTE=12
 WEBSOCKET_ENABLED=true
+MARKET_OBSERVATION_ENABLED=true
+MARKET_OBSERVATION_SAMPLE_SECONDS=30
 MIN_SIGNAL_CONFIDENCE=0.64
 MAX_SPREAD_PERCENT=0.18
 MIN_24H_TURNOVER_USDT=1000000

README.md

@@ -5,6 +5,7 @@ Spot-бот для демо-торговли криптовалютой на р
 ## Что реализовано
 
 - Реальные market data Bybit Spot: REST bootstrap и WebSocket-обновления.
+- Сэмплированный L1-стакан Bybit сохраняется в SQLite: bid/ask size, spread, imbalance и microprice доступны обучающему агенту через защищённый постраничный API. Частота по умолчанию — один sample на пару каждые 30 секунд, а хранилище ограничено 1 200 000 строк.
 - Торговый universe автоматически строится из актуальных Bybit Spot-инструментов: выбираются до 12 ликвидных USDT-пар по `turnover24h`, исключаются stablecoin-to-stablecoin и leveraged-token пары; фиксированный список можно задать только явным `SYMBOLS`.
 - Paper trading с учетом cash, комиссий, проскальзывания, stop-loss, take-profit и trailing stop.
 - Spot-only логика: покупка базовой монеты за USDT и продажа обратно, без short и без плеча.
@@ -144,6 +145,8 @@ FAST_LOOP_INTERVAL_SECONDS=1
 FAST_ENTRY_COOLDOWN_SECONDS=20
 MAX_ENTRIES_PER_MINUTE=12
 WEBSOCKET_ENABLED=true
+MARKET_OBSERVATION_ENABLED=true
+MARKET_OBSERVATION_SAMPLE_SECONDS=30
 MIN_SIGNAL_CONFIDENCE=0.64
 PATTERN_ANALYSIS_ENABLED=true
 PATTERN_SCORE_WEIGHT=0.18
@@ -243,6 +246,7 @@ Live-исполнение ведет журнал order intent до отправ
 - `GET /api/health` — healthcheck.
 - `GET /api/status` — статус бота, account snapshot, позиции.
 - `GET /api/markets` — пары, ticker, свечи, инструменты.
+- `GET /api/training/market-observations?symbol=BTCUSDT&after_id=0&limit=5000` — защищённая training-token выгрузка L1-наблюдений.
 - `GET /api/trades` — последние сделки.
 - `GET /api/signals` — последние сигналы стратегии.
 - `GET /api/events` — события.

crypto_spot_bot/__init__.py

@@ -1,3 +1,3 @@
 """Crypto spot trading bot package."""
 
-__version__ = "1.0.1"
+__version__ = "1.0.2"

crypto_spot_bot/bybit.py

@@ -220,6 +220,10 @@ class BybitClient:
         return candles
 
     def orderbook_top(self, symbol: str) -> tuple[float, float]:
+        bid, _bid_size, ask, _ask_size = self.orderbook_level_one(symbol)
+        return bid, ask
+
+    def orderbook_level_one(self, symbol: str) -> tuple[float, float, float, float]:
         result = self.public_get(
             "/v5/market/orderbook",
             {"category": "spot", "symbol": symbol, "limit": 1},
@@ -227,8 +231,10 @@ class BybitClient:
         bids = result.get("b") or []
         asks = result.get("a") or []
         bid = _float(bids[0][0]) if bids else 0.0
+        bid_size = _float(bids[0][1]) if bids and len(bids[0]) > 1 else 0.0
         ask = _float(asks[0][0]) if asks else 0.0
-        return bid, ask
+        ask_size = _float(asks[0][1]) if asks and len(asks[0]) > 1 else 0.0
+        return bid, bid_size, ask, ask_size
 
     def place_spot_market_order(
         self,

crypto_spot_bot/config.py

@@ -172,6 +172,8 @@ class Settings:
     bybit_rest_base_url_override: str = ""
     bybit_websocket_url_override: str = ""
     time_series_fallback_mode: str = "trend_macd"
+    market_observation_enabled: bool = True
+    market_observation_sample_seconds: float = 30.0
 
     @property
     def rest_base_url(self) -> str:
@@ -358,6 +360,10 @@ def load_settings(env_file: str | Path | None = None) -> Settings:
         time_series_fallback_mode=os.getenv(
             "TIME_SERIES_FALLBACK_MODE", "trend_macd"
         ).strip().lower(),
+        market_observation_enabled=_bool_env("MARKET_OBSERVATION_ENABLED", True),
+        market_observation_sample_seconds=_float_env(
+            "MARKET_OBSERVATION_SAMPLE_SECONDS", 30.0
+        ),
     )
     _validate_settings(settings)
     if settings.trading_mode == "live" and not settings.live_ready:
@@ -396,6 +402,8 @@ def _validate_settings(settings: Settings) -> None:
         errors.append("LIVE_RECONCILIATION_INTERVAL_SECONDS must be positive")
     if settings.time_series_fallback_mode not in {"trend_macd", "legacy"}:
         errors.append("TIME_SERIES_FALLBACK_MODE must be trend_macd or legacy")
+    if settings.market_observation_sample_seconds <= 0:
+        errors.append("MARKET_OBSERVATION_SAMPLE_SECONDS must be positive")
     if errors:
         raise ValueError("; ".join(errors))
 

crypto_spot_bot/dashboard.py

@@ -149,6 +149,27 @@ def create_app(settings: Settings | None = None) -> FastAPI:
     async def training_status(_: None = Depends(authorizer.require)) -> dict[str, Any]:
         return training.status()
 
+    @app.get("/api/training/market-observations")
+    async def training_market_observations(
+        symbol: str,
+        after_id: int = 0,
+        limit: int = 5000,
+        _: None = Depends(authorizer.require_training),
+    ) -> dict[str, Any]:
+        normalized_symbol = symbol.strip().upper()
+        if not normalized_symbol:
+            raise HTTPException(status_code=400, detail="symbol is required")
+        items = storage.market_observations_after(
+            symbol=normalized_symbol,
+            after_id=max(0, after_id),
+            limit=max(1, min(limit, 5000)),
+        )
+        return {
+            "symbol": normalized_symbol,
+            "items": items,
+            "next_after_id": int(items[-1]["id"]) if items else max(0, after_id),
+        }
+
     @app.post("/api/training/retrain")
     async def training_retrain(
         payload: dict[str, Any] | None = None,
@@ -407,6 +428,8 @@ def _safe_config(settings: Settings) -> dict[str, Any]:
         "time_series_require_fresh_model": settings.time_series_require_fresh_model,
         "time_series_model_max_age_hours": settings.time_series_model_max_age_hours,
         "market_ticker_max_age_seconds": settings.market_ticker_max_age_seconds,
+        "market_observation_enabled": settings.market_observation_enabled,
+        "market_observation_sample_seconds": settings.market_observation_sample_seconds,
         "time_series_model_artifact": _time_series_model_artifact(settings),
         "stop_loss_percent": settings.stop_loss_percent,
         "stop_loss_exit_enabled": settings.stop_loss_exit_enabled,

crypto_spot_bot/market_data.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 import json
 import threading
+import time
 from dataclasses import asdict
 from datetime import datetime
 from typing import Any
@@ -51,6 +52,7 @@ class MarketData:
         self.candles: dict[str, list[Candle]] = {}
         self.trend_candles: dict[str, list[Candle]] = {}
         self.orderbook_top: dict[str, tuple[float, float]] = {}
+        self.orderbook_metrics: dict[str, dict[str, Any]] = {}
         self.patterns: dict[str, dict[str, Any]] = {}
         self.forecasts: dict[str, dict[str, Any]] = {}
         self.last_rest_refresh_at: datetime | None = None
@@ -60,6 +62,11 @@ class MarketData:
         self._refresh_lock = threading.Lock()
         self.rest_error_count = 0
         self.last_rest_error = ""
+        self.observation_samples = 0
+        self.last_observation_at: datetime | None = None
+        self.observation_error_count = 0
+        self.last_observation_error = ""
+        self._last_observation_monotonic: dict[str, float] = {}
 
     async def bootstrap(self) -> None:
         self.instruments = await asyncio.to_thread(self.client.instruments)
@@ -112,19 +119,8 @@ class MarketData:
                         trend_candles = _closed_candles(trend_candles, self.settings.trend_interval)
                         add_indicators(trend_candles)
                         self.trend_candles[symbol] = trend_candles
-                    bid, ask = self.client.orderbook_top(symbol)
+                    bid, bid_size, ask, ask_size = self.client.orderbook_level_one(symbol)
-                    self.orderbook_top[symbol] = (bid, ask)
+                    self._update_orderbook(symbol, bid, bid_size, ask, ask_size)
-                    if symbol in self.tickers:
-                        current = self.tickers[symbol]
-                        self.tickers[symbol] = Ticker(
-                            symbol=current.symbol,
-                            last_price=current.last_price,
-                            bid=bid or current.bid,
-                            ask=ask or current.ask,
-                            turnover_24h=current.turnover_24h,
-                            volume_24h=current.volume_24h,
-                            change_24h=current.change_24h,
-                        )
                 except Exception as exc:
                     self.rest_error_count += 1
                     self.last_rest_error = str(exc)
@@ -180,7 +176,11 @@ class MarketData:
         elif topic.startswith("orderbook.") and isinstance(data, dict):
             parts = topic.split(".")
             if len(parts) >= 3:
-                self._handle_orderbook(parts[2], data)
+                self._handle_orderbook(
+                    parts[2],
+                    data,
+                    source_timestamp_ms=int(_float(message.get("ts"))),
+                )
 
     def _handle_ticker(self, symbol: str, data: dict[str, Any]) -> None:
         current = self.tickers.get(symbol)
@@ -222,14 +222,66 @@ class MarketData:
         add_indicators(candles)
         self.candles[symbol] = candles
 
-    def _handle_orderbook(self, symbol: str, data: dict[str, Any]) -> None:
+    def _handle_orderbook(
+        self,
+        symbol: str,
+        data: dict[str, Any],
+        source_timestamp_ms: int = 0,
+    ) -> None:
         bids = data.get("b") or []
         asks = data.get("a") or []
         bid = _float(bids[0][0]) if bids else 0.0
+        bid_size = _float(bids[0][1]) if bids and len(bids[0]) > 1 else 0.0
         ask = _float(asks[0][0]) if asks else 0.0
+        ask_size = _float(asks[0][1]) if asks and len(asks[0]) > 1 else 0.0
+        self._update_orderbook(
+            symbol,
+            bid,
+            bid_size,
+            ask,
+            ask_size,
+            source_timestamp_ms=source_timestamp_ms,
+        )
+
+    def _update_orderbook(
+        self,
+        symbol: str,
+        bid: float,
+        bid_size: float,
+        ask: float,
+        ask_size: float,
+        *,
+        source_timestamp_ms: int = 0,
+    ) -> None:
         if bid > 0 and ask > 0:
             self.orderbook_top[symbol] = (bid, ask)
             current = self.tickers.get(symbol)
+            size_total = max(0.0, bid_size) + max(0.0, ask_size)
+            mid_price = (bid + ask) / 2.0
+            imbalance = (
+                (max(0.0, bid_size) - max(0.0, ask_size)) / size_total
+                if size_total > 0
+                else 0.0
+            )
+            microprice = (
+                (ask * max(0.0, bid_size) + bid * max(0.0, ask_size)) / size_total
+                if size_total > 0
+                else mid_price
+            )
+            observed_at = utc_now()
+            metrics = {
+                "bid_price": bid,
+                "bid_size": max(0.0, bid_size),
+                "ask_price": ask,
+                "ask_size": max(0.0, ask_size),
+                "mid_price": mid_price,
+                "microprice": microprice,
+                "spread_bps": ((ask - bid) / mid_price) * 10_000 if mid_price > 0 else 0.0,
+                "imbalance": imbalance,
+                "source_timestamp_ms": max(0, source_timestamp_ms),
+                "observed_at": observed_at.isoformat(),
+            }
+            self.orderbook_metrics[symbol] = metrics
             if current:
                 self.tickers[symbol] = Ticker(
                     symbol=symbol,
@@ -240,6 +292,44 @@ class MarketData:
                     volume_24h=current.volume_24h,
                     change_24h=current.change_24h,
                 )
+            self._sample_orderbook(symbol, metrics, current.last_price if current else mid_price, observed_at)
+
+    def _sample_orderbook(
+        self,
+        symbol: str,
+        metrics: dict[str, Any],
+        last_price: float,
+        observed_at: datetime,
+    ) -> None:
+        if not self.settings.market_observation_enabled:
+            return
+        now = time.monotonic()
+        previous = self._last_observation_monotonic.get(symbol)
+        if previous is not None and now - previous < self.settings.market_observation_sample_seconds:
+            return
+        try:
+            self.storage.insert_market_observation(
+                symbol=symbol,
+                bid_price=float(metrics["bid_price"]),
+                bid_size=float(metrics["bid_size"]),
+                ask_price=float(metrics["ask_price"]),
+                ask_size=float(metrics["ask_size"]),
+                mid_price=float(metrics["mid_price"]),
+                microprice=float(metrics["microprice"]),
+                spread_bps=float(metrics["spread_bps"]),
+                imbalance=float(metrics["imbalance"]),
+                last_price=last_price,
+                source_timestamp_ms=int(metrics["source_timestamp_ms"]),
+                created_at=observed_at,
+            )
+        except Exception as exc:  # Storage errors must not disconnect market data.
+            self.observation_error_count += 1
+            self.last_observation_error = str(exc)
+            return
+        self._last_observation_monotonic[symbol] = now
+        self.observation_samples += 1
+        self.last_observation_at = observed_at
+        self.last_observation_error = ""
 
     def prices(self) -> dict[str, float]:
         return {symbol: ticker.last_price for symbol, ticker in self.tickers.items()}
@@ -286,6 +376,16 @@ class MarketData:
             "last_ws_message_at": self.last_ws_message_at.isoformat()
             if self.last_ws_message_at
             else None,
+            "observation_collector": {
+                "enabled": self.settings.market_observation_enabled,
+                "sample_seconds": self.settings.market_observation_sample_seconds,
+                "samples_since_start": self.observation_samples,
+                "last_observation_at": self.last_observation_at.isoformat()
+                if self.last_observation_at
+                else None,
+                "error_count": self.observation_error_count,
+                "last_error": self.last_observation_error,
+            },
             "markets": [
                 {
                     "ticker": self.tickers[symbol].as_dict() if symbol in self.tickers else None,
@@ -293,6 +393,7 @@ class MarketData:
                     "trend_candles": [candle.as_dict() for candle in self.trend_candles.get(symbol, [])[-5:]],
                     "pattern": self.patterns.get(symbol),
                     "forecast": self.forecasts.get(symbol),
+                    "orderbook": self.orderbook_metrics.get(symbol),
                     "quality": analyze_symbol_quality(
                         symbol=symbol,
                         candles=self.candles.get(symbol, []),

crypto_spot_bot/storage.py

@@ -4,7 +4,7 @@ import json
 import sqlite3
 import time
 from contextlib import contextmanager
-from datetime import timedelta
+from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Any, Iterator
 
@@ -18,6 +18,7 @@ MAX_RUNTIME_ROWS = {
     "equity": 100_000,
     "events": 20_000,
     "llm_advice": 20_000,
+    "market_observations": 1_200_000,
 }
 _STORED_FORECAST_KEYS = {
     "enabled",
@@ -171,6 +172,21 @@ class Storage:
                     created_at TEXT NOT NULL,
                     updated_at TEXT NOT NULL
                 );
+                CREATE TABLE IF NOT EXISTS market_observations (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    symbol TEXT NOT NULL,
+                    bid_price REAL NOT NULL,
+                    bid_size REAL NOT NULL,
+                    ask_price REAL NOT NULL,
+                    ask_size REAL NOT NULL,
+                    mid_price REAL NOT NULL,
+                    microprice REAL NOT NULL,
+                    spread_bps REAL NOT NULL,
+                    imbalance REAL NOT NULL,
+                    last_price REAL NOT NULL,
+                    source_timestamp_ms INTEGER NOT NULL DEFAULT 0,
+                    created_at TEXT NOT NULL
+                );
                 CREATE INDEX IF NOT EXISTS idx_positions_status_opened
                     ON positions(status, opened_at);
                 CREATE INDEX IF NOT EXISTS idx_trades_closed
@@ -183,6 +199,10 @@ class Storage:
                     ON events(created_at DESC);
                 CREATE INDEX IF NOT EXISTS idx_orders_status_updated
                     ON orders(status, updated_at DESC);
+                CREATE INDEX IF NOT EXISTS idx_market_observations_symbol_id
+                    ON market_observations(symbol, id);
+                CREATE INDEX IF NOT EXISTS idx_market_observations_created
+                    ON market_observations(created_at);
                 """
             )
             columns = {
@@ -458,6 +478,70 @@ class Storage:
             rows = conn.execute("SELECT * FROM signals ORDER BY id DESC LIMIT ?", (limit,)).fetchall()
         return [dict(row) for row in rows]
 
+    def insert_market_observation(
+        self,
+        *,
+        symbol: str,
+        bid_price: float,
+        bid_size: float,
+        ask_price: float,
+        ask_size: float,
+        mid_price: float,
+        microprice: float,
+        spread_bps: float,
+        imbalance: float,
+        last_price: float,
+        source_timestamp_ms: int = 0,
+        created_at: datetime | None = None,
+    ) -> int:
+        timestamp = (created_at or utc_now()).isoformat()
+        with self.connect() as conn:
+            cursor = conn.execute(
+                """
+                INSERT INTO market_observations (
+                    symbol, bid_price, bid_size, ask_price, ask_size,
+                    mid_price, microprice, spread_bps, imbalance, last_price,
+                    source_timestamp_ms, created_at
+                )
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    symbol.upper(),
+                    bid_price,
+                    bid_size,
+                    ask_price,
+                    ask_size,
+                    mid_price,
+                    microprice,
+                    spread_bps,
+                    imbalance,
+                    last_price,
+                    max(0, source_timestamp_ms),
+                    timestamp,
+                ),
+            )
+            return int(cursor.lastrowid)
+
+    def market_observations_after(
+        self,
+        *,
+        symbol: str,
+        after_id: int = 0,
+        limit: int = 5000,
+    ) -> list[dict[str, Any]]:
+        row_limit = max(1, min(limit, 5000))
+        with self.connect() as conn:
+            rows = conn.execute(
+                """
+                SELECT * FROM market_observations
+                WHERE symbol = ? AND id > ?
+                ORDER BY id
+                LIMIT ?
+                """,
+                (symbol.upper(), max(0, after_id), row_limit),
+            ).fetchall()
+        return [dict(row) for row in rows]
+
     def insert_equity(
         self,
         equity: float,
@@ -635,7 +719,7 @@ class Storage:
             return {}
         cutoff = (utc_now() - timedelta(days=retention_days)).isoformat()
         deleted: dict[str, int] = {}
-        for table in ("signals", "equity", "events", "llm_advice"):
+        for table in ("signals", "equity", "events", "llm_advice", "market_observations"):
             with self.connect() as conn:
                 max_id_row = conn.execute(f"SELECT MAX(id) AS value FROM {table}").fetchone()
                 max_id = int(max_id_row["value"] or 0) if max_id_row else 0
@@ -676,7 +760,17 @@ class Storage:
 
     def clear_all(self) -> None:
         with self.connect() as conn:
-            for table in ("positions", "trades", "signals", "equity", "events", "runtime", "llm_advice", "orders"):
+            for table in (
+                "positions",
+                "trades",
+                "signals",
+                "equity",
+                "events",
+                "runtime",
+                "llm_advice",
+                "orders",
+                "market_observations",
+            ):
                 conn.execute(f"DELETE FROM {table}")
 
 

tests/test_bybit.py

@@ -125,3 +125,14 @@ def test_websocket_subscribe_uses_configured_kline_interval() -> None:
 
     assert "kline.60.BTCUSDT" in payload
     assert "kline.1.BTCUSDT" not in payload
+
+
+def test_orderbook_level_one_preserves_sizes(make_settings, tmp_path) -> None:
+    client = BybitClient(make_settings(tmp_path))
+    client.public_get = lambda *_args, **_kwargs: {
+        "b": [["100.5", "2.25"]],
+        "a": [["100.7", "1.75"]],
+    }
+
+    assert client.orderbook_level_one("BTCUSDT") == (100.5, 2.25, 100.7, 1.75)
+    assert client.orderbook_top("BTCUSDT") == (100.5, 100.7)

tests/test_config.py

@@ -180,3 +180,12 @@ def test_load_settings_rejects_unknown_fallback_mode(tmp_path, monkeypatch) -> N
 
     with pytest.raises(ValueError, match="TIME_SERIES_FALLBACK_MODE"):
         load_settings(env_file)
+
+
+def test_load_settings_rejects_non_positive_observation_interval(tmp_path, monkeypatch) -> None:
+    monkeypatch.delenv("MARKET_OBSERVATION_SAMPLE_SECONDS", raising=False)
+    env_file = tmp_path / ".env"
+    env_file.write_text("MARKET_OBSERVATION_SAMPLE_SECONDS=0\n", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="MARKET_OBSERVATION_SAMPLE_SECONDS"):
+        load_settings(env_file)

tests/test_dashboard.py

@@ -46,6 +46,8 @@ def test_safe_config_summarizes_torch_forecast_artifact(make_settings, tmp_path)
     assert config["time_series_probe_size_multiplier"] == 0.40
     assert config["time_series_rebound_fallback_enabled"] is True
     assert config["time_series_fallback_mode"] == "trend_macd"
+    assert config["market_observation_enabled"] is True
+    assert config["market_observation_sample_seconds"] == 30.0
     assert config["time_series_model_artifact"] == {
         "available": True,
         "type": "pytorch_recurrent_forecaster",

tests/test_market_data.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
-from crypto_spot_bot.market_data import _candles_due, _closed_candles, _is_closed_kline_row
+from crypto_spot_bot.market_data import MarketData, _candles_due, _closed_candles, _is_closed_kline_row
 from crypto_spot_bot.models import Candle
+from crypto_spot_bot.storage import Storage
 
 
 def test_closed_candles_excludes_current_open_interval() -> None:
@@ -26,3 +27,37 @@ def test_rest_candles_refresh_only_after_next_bar_closes() -> None:
 
     assert _candles_due([candle], "1", now_ms=11 * 60_000 + 30_000) is False
     assert _candles_due([candle], "1", now_ms=12 * 60_000) is True
+
+
+def test_orderbook_handler_samples_sizes_and_microstructure(make_settings, tmp_path) -> None:
+    settings = make_settings(
+        tmp_path,
+        market_observation_enabled=True,
+        market_observation_sample_seconds=30.0,
+    )
+    storage = Storage(settings.database_path)
+    market = MarketData(settings, object(), storage)
+
+    market._handle_orderbook(
+        "BTCUSDT",
+        {"b": [["100", "3"]], "a": [["101", "1"]]},
+        source_timestamp_ms=1_789_000_000_000,
+    )
+    market._handle_orderbook(
+        "BTCUSDT",
+        {"b": [["100", "4"]], "a": [["101", "1"]]},
+        source_timestamp_ms=1_789_000_001_000,
+    )
+
+    metrics = market.orderbook_metrics["BTCUSDT"]
+    rows = storage.market_observations_after(symbol="BTCUSDT")
+    assert metrics["bid_size"] == 4.0
+    assert metrics["ask_size"] == 1.0
+    assert metrics["imbalance"] == 0.6
+    assert metrics["microprice"] == 100.8
+    assert len(rows) == 1
+    assert rows[0]["bid_size"] == 3.0
+    assert rows[0]["ask_size"] == 1.0
+    assert rows[0]["imbalance"] == 0.5
+    assert rows[0]["microprice"] == 100.75
+    assert rows[0]["source_timestamp_ms"] == 1_789_000_000_000

tests/test_storage.py

@@ -62,6 +62,57 @@ def test_prune_deletes_only_one_bounded_batch_per_table(tmp_path) -> None:
     assert len(storage.recent_signals(PRUNE_BATCH_SIZE + 10)) == 5
 
 
+def test_market_observation_export_is_symbol_scoped_and_paginated(tmp_path) -> None:
+    storage = Storage(tmp_path / "tradebot.sqlite3")
+    first_id = storage.insert_market_observation(
+        symbol="BTCUSDT",
+        bid_price=100.0,
+        bid_size=2.0,
+        ask_price=101.0,
+        ask_size=1.0,
+        mid_price=100.5,
+        microprice=100.6666666667,
+        spread_bps=99.50248756,
+        imbalance=1 / 3,
+        last_price=100.4,
+        source_timestamp_ms=1_789_000_000_000,
+    )
+    second_id = storage.insert_market_observation(
+        symbol="BTCUSDT",
+        bid_price=101.0,
+        bid_size=1.0,
+        ask_price=102.0,
+        ask_size=1.0,
+        mid_price=101.5,
+        microprice=101.5,
+        spread_bps=98.52216749,
+        imbalance=0.0,
+        last_price=101.4,
+        source_timestamp_ms=1_789_000_030_000,
+    )
+    storage.insert_market_observation(
+        symbol="ETHUSDT",
+        bid_price=10.0,
+        bid_size=1.0,
+        ask_price=11.0,
+        ask_size=1.0,
+        mid_price=10.5,
+        microprice=10.5,
+        spread_bps=952.38095238,
+        imbalance=0.0,
+        last_price=10.4,
+    )
+
+    rows = storage.market_observations_after(
+        symbol="BTCUSDT",
+        after_id=first_id,
+        limit=1,
+    )
+
+    assert [row["id"] for row in rows] == [second_id]
+    assert rows[0]["source_timestamp_ms"] == 1_789_000_030_000
+
+
 def test_runtime_compaction_preserves_durable_state_and_bounds_telemetry(tmp_path) -> None:
     database = tmp_path / "tradebot.sqlite3"
     storage = Storage(database)

tools/compact_runtime_db.py

@@ -20,6 +20,7 @@ DEFAULT_RECENT_ROWS = {
     "equity": 5_000,
     "events": 2_000,
     "llm_advice": 1_000,
+    "market_observations": 100_000,
 }
 
 
@@ -114,6 +115,11 @@ def _parse_args() -> argparse.Namespace:
     parser.add_argument("--equity", type=int, default=DEFAULT_RECENT_ROWS["equity"])
     parser.add_argument("--events", type=int, default=DEFAULT_RECENT_ROWS["events"])
     parser.add_argument("--llm-advice", type=int, default=DEFAULT_RECENT_ROWS["llm_advice"])
+    parser.add_argument(
+        "--market-observations",
+        type=int,
+        default=DEFAULT_RECENT_ROWS["market_observations"],
+    )
     return parser.parse_args()
 
 
@@ -127,6 +133,7 @@ def main() -> None:
             "equity": args.equity,
             "events": args.events,
             "llm_advice": args.llm_advice,
+            "market_observations": args.market_observations,
         },
     )
     print(json.dumps(result, ensure_ascii=False, sort_keys=True))