diff --git a/.env.example b/.env.example index 2cd12b9..b8b4813 100644 --- a/.env.example +++ b/.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 diff --git a/README.md b/README.md index 74bcef7..971bb4f 100644 --- a/README.md +++ b/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` — события. diff --git a/crypto_spot_bot/__init__.py b/crypto_spot_bot/__init__.py index 9bd75b8..f875867 100644 --- a/crypto_spot_bot/__init__.py +++ b/crypto_spot_bot/__init__.py @@ -1,3 +1,3 @@ """Crypto spot trading bot package.""" -__version__ = "1.0.1" +__version__ = "1.0.2" diff --git a/crypto_spot_bot/bybit.py b/crypto_spot_bot/bybit.py index 5462176..3e1a8a1 100644 --- a/crypto_spot_bot/bybit.py +++ b/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, diff --git a/crypto_spot_bot/config.py b/crypto_spot_bot/config.py index da8c0ec..26a66fc 100644 --- a/crypto_spot_bot/config.py +++ b/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)) diff --git a/crypto_spot_bot/dashboard.py b/crypto_spot_bot/dashboard.py index 4d8003b..d423c17 100644 --- a/crypto_spot_bot/dashboard.py +++ b/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, diff --git a/crypto_spot_bot/market_data.py b/crypto_spot_bot/market_data.py index fa23da0..bee5d33 100644 --- a/crypto_spot_bot/market_data.py +++ b/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) - self.orderbook_top[symbol] = (bid, ask) - 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, - ) + bid, bid_size, ask, ask_size = self.client.orderbook_level_one(symbol) + self._update_orderbook(symbol, bid, bid_size, ask, ask_size) 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, []), diff --git a/crypto_spot_bot/storage.py b/crypto_spot_bot/storage.py index 03f45bb..dc5766b 100644 --- a/crypto_spot_bot/storage.py +++ b/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}") diff --git a/tests/test_bybit.py b/tests/test_bybit.py index ecdbdfe..98c5b3e 100644 --- a/tests/test_bybit.py +++ b/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) diff --git a/tests/test_config.py b/tests/test_config.py index f1146de..e7b93be 100644 --- a/tests/test_config.py +++ b/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) diff --git a/tests/test_dashboard.py b/tests/test_dashboard.py index 756f1de..0c0b076 100644 --- a/tests/test_dashboard.py +++ b/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", diff --git a/tests/test_market_data.py b/tests/test_market_data.py index 8ffc785..47a79fd 100644 --- a/tests/test_market_data.py +++ b/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 diff --git a/tests/test_storage.py b/tests/test_storage.py index c323797..5f8ebd9 100644 --- a/tests/test_storage.py +++ b/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) diff --git a/tools/compact_runtime_db.py b/tools/compact_runtime_db.py index d775af5..ae6eac8 100644 --- a/tools/compact_runtime_db.py +++ b/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))