Observability: Prometheus metrics, log panel, runbook
- GET /api/metrics renders Prometheus text (equity, P&L, per-strategy equity/pnl/allocation/halted, info) with no client library. - In-memory ring log handler (app/logs.py) exposed at GET /api/logs and a dashboard "Server logs" panel; Runtime mirrors bot events into the logger. - docs/runbook.md covers deploys, the kill switch, resetting paper state and going live. Adds tests for metrics rendering and the log ring.
This commit is contained in:
1 parent
c605a0f316
commit
ac92dd5ac4
12 files changed
+462
-1
No files matched your search
@@ -4,10 +4,20 @@ import asyncio
|
||||
from typing import Annotated
|
||||
|
||||
import httpx
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request, WebSocket, WebSocketDisconnect
|
||||
from fastapi import (
|
||||
APIRouter,
|
||||
Depends,
|
||||
HTTPException,
|
||||
Request,
|
||||
Response,
|
||||
WebSocket,
|
||||
WebSocketDisconnect,
|
||||
)
|
||||
|
||||
from app.config import Settings
|
||||
from app.engine.strategies import get_strategy, list_strategies, strategy_spec
|
||||
from app.logs import get_log_handler
|
||||
from app.metrics import render_metrics
|
||||
from app.models import OrderSide, Position, PositionStatus
|
||||
from app.runtime import Runtime
|
||||
from app.schemas import (
|
||||
@@ -65,6 +75,21 @@ def health() -> dict[str, str]:
|
||||
return {"status": "ok"}
|
||||
|
||||
|
||||
@router.get("/metrics")
|
||||
def metrics(runtime: GuardedRuntimeDep) -> Response:
|
||||
"""Prometheus text exposition, unauthenticated so a scraper can reach it."""
|
||||
status = runtime.multi_status()
|
||||
body = render_metrics(status, runtime.settings.mode, runtime.settings.exchange)
|
||||
return Response(content=body, media_type="text/plain; version=0.0.4")
|
||||
|
||||
|
||||
@router.get("/logs")
|
||||
def logs(runtime: GuardedRuntimeDep, limit: int = 100) -> dict:
|
||||
handler = get_log_handler()
|
||||
entries = handler.snapshot(limit) if handler is not None else []
|
||||
return {"entries": entries}
|
||||
|
||||
|
||||
@router.get("/version")
|
||||
def version(request: Request) -> dict[str, str]:
|
||||
"""Build version of the running backend, used by the client to self-heal
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections import deque
|
||||
from datetime import UTC, datetime
|
||||
|
||||
|
||||
class RingLogHandler(logging.Handler):
|
||||
"""Keeps the most recent log records in memory for the dashboard.
|
||||
|
||||
Deliberately small and dependency-free: the goal is a debugging aid, not a
|
||||
log store. Attach once at startup.
|
||||
"""
|
||||
|
||||
def __init__(self, capacity: int = 200) -> None:
|
||||
super().__init__()
|
||||
self.records: deque[dict] = deque(maxlen=capacity)
|
||||
|
||||
def emit(self, record: logging.LogRecord) -> None:
|
||||
try:
|
||||
self.records.append(
|
||||
{
|
||||
"time": datetime.fromtimestamp(record.created, UTC).isoformat(),
|
||||
"level": record.levelname.lower(),
|
||||
"logger": record.name,
|
||||
"message": record.getMessage(),
|
||||
}
|
||||
)
|
||||
except Exception: # never let logging break the app
|
||||
self.handleError(record)
|
||||
|
||||
def snapshot(self, limit: int = 100) -> list[dict]:
|
||||
items = list(self.records)
|
||||
return items[-limit:]
|
||||
|
||||
|
||||
_handler: RingLogHandler | None = None
|
||||
|
||||
|
||||
def install_log_capture(capacity: int = 200) -> RingLogHandler:
|
||||
"""Attach the ring handler to the root logger (idempotent)."""
|
||||
global _handler
|
||||
if _handler is not None:
|
||||
return _handler
|
||||
_handler = RingLogHandler(capacity)
|
||||
_handler.setLevel(logging.INFO)
|
||||
root = logging.getLogger()
|
||||
root.addHandler(_handler)
|
||||
if root.level > logging.INFO or root.level == logging.NOTSET:
|
||||
root.setLevel(logging.INFO)
|
||||
return _handler
|
||||
|
||||
|
||||
def get_log_handler() -> RingLogHandler | None:
|
||||
return _handler
|
||||
@@ -10,6 +10,7 @@ from fastapi.middleware.cors import CORSMiddleware
|
||||
from app.api.routes import router
|
||||
from app.config import Settings, get_settings
|
||||
from app.db import init_db
|
||||
from app.logs import install_log_capture
|
||||
from app.runtime import Runtime
|
||||
|
||||
|
||||
@@ -22,6 +23,7 @@ def create_app(
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
||||
install_log_capture()
|
||||
init_db()
|
||||
app.state.runtime = runtime
|
||||
yield
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from app.schemas import MultiStatusOut
|
||||
|
||||
# Rendered as Prometheus text exposition format (no client library needed).
|
||||
# The route serves this at GET /api/metrics for the existing Prometheus scrape.
|
||||
|
||||
|
||||
def _escape(value: str) -> str:
|
||||
return value.replace("\\", "\\\\").replace('"', '\\"')
|
||||
|
||||
|
||||
def _metric(
|
||||
name: str, help_text: str, value: float | int, labels: dict[str, str] | None = None
|
||||
) -> str:
|
||||
lines = [f"# HELP {name} {help_text}", f"# TYPE {name} gauge"]
|
||||
if labels:
|
||||
rendered = ",".join(f'{k}="{_escape(v)}"' for k, v in labels.items())
|
||||
lines.append(f"{name}{{{rendered}}} {value}")
|
||||
else:
|
||||
lines.append(f"{name} {value}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def render_metrics(status: MultiStatusOut, mode: str, exchange: str) -> str:
|
||||
"""Render the current portfolio state as Prometheus metrics."""
|
||||
parts = [
|
||||
_metric("crypto_running", "1 when the bot is running", int(status.running)),
|
||||
_metric("crypto_equity", "Total account equity", status.equity),
|
||||
_metric("crypto_starting_balance", "Starting balance", status.starting_balance),
|
||||
_metric("crypto_total_pnl", "Total profit and loss", status.total_pnl),
|
||||
_metric(
|
||||
"crypto_open_positions",
|
||||
"Number of open positions across strategies",
|
||||
sum(s.open_positions for s in status.strategies),
|
||||
),
|
||||
_metric(
|
||||
"crypto_strategies",
|
||||
"Number of configured strategies",
|
||||
len(status.strategies),
|
||||
),
|
||||
]
|
||||
|
||||
if status.price is not None:
|
||||
parts.append(_metric("crypto_price", "Latest mark price", status.price))
|
||||
|
||||
# Common labels shared by the per-strategy families.
|
||||
lines = [
|
||||
"# HELP crypto_strategy_equity Equity per strategy",
|
||||
"# TYPE crypto_strategy_equity gauge",
|
||||
]
|
||||
for s in status.strategies:
|
||||
labels = {"strategy": s.strategy_key, "symbol": s.symbol}
|
||||
rendered = ",".join(f'{k}="{_escape(v)}"' for k, v in labels.items())
|
||||
lines.append(f"crypto_strategy_equity{{{rendered}}} {s.equity}")
|
||||
parts.append("\n".join(lines))
|
||||
|
||||
lines = [
|
||||
"# HELP crypto_strategy_pnl Total P&L per strategy",
|
||||
"# TYPE crypto_strategy_pnl gauge",
|
||||
]
|
||||
for s in status.strategies:
|
||||
labels = {"strategy": s.strategy_key, "symbol": s.symbol}
|
||||
rendered = ",".join(f'{k}="{_escape(v)}"' for k, v in labels.items())
|
||||
lines.append(f"crypto_strategy_pnl{{{rendered}}} {s.total_pnl}")
|
||||
parts.append("\n".join(lines))
|
||||
|
||||
lines = [
|
||||
"# HELP crypto_strategy_allocation Fraction of the balance per strategy",
|
||||
"# TYPE crypto_strategy_allocation gauge",
|
||||
]
|
||||
for s in status.strategies:
|
||||
labels = {"strategy": s.strategy_key}
|
||||
rendered = ",".join(f'{k}="{_escape(v)}"' for k, v in labels.items())
|
||||
lines.append(f"crypto_strategy_allocation{{{rendered}}} {s.allocation}")
|
||||
parts.append("\n".join(lines))
|
||||
|
||||
lines = [
|
||||
"# HELP crypto_strategy_halted 1 when a strategy is halted",
|
||||
"# TYPE crypto_strategy_halted gauge",
|
||||
]
|
||||
for s in status.strategies:
|
||||
labels = {"strategy": s.strategy_key}
|
||||
rendered = ",".join(f'{k}="{_escape(v)}"' for k, v in labels.items())
|
||||
lines.append(f"crypto_strategy_halted{{{rendered}}} {int(s.halted)}")
|
||||
parts.append("\n".join(lines))
|
||||
|
||||
info_labels = {"mode": mode, "exchange": exchange}
|
||||
rendered = ",".join(f'{k}="{_escape(v)}"' for k, v in info_labels.items())
|
||||
parts.append(
|
||||
"# HELP crypto_info Build information\n"
|
||||
"# TYPE crypto_info gauge\n"
|
||||
f"crypto_info{{{rendered}}} 1"
|
||||
)
|
||||
|
||||
return "\n".join(parts) + "\n"
|
||||
@@ -32,6 +32,15 @@ from app.store import Store
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_LOG_LEVELS = {
|
||||
"debug": logging.DEBUG,
|
||||
"info": logging.INFO,
|
||||
"trade": logging.INFO,
|
||||
"warning": logging.WARNING,
|
||||
"error": logging.ERROR,
|
||||
"critical": logging.CRITICAL,
|
||||
}
|
||||
|
||||
FeedFactory = Callable[[Settings], MarketData]
|
||||
BrokerFactory = Callable[[Settings], Broker]
|
||||
NotifierFactory = Callable[[Settings], Notifier | None]
|
||||
@@ -91,8 +100,15 @@ class Runtime:
|
||||
with suppress(asyncio.QueueFull):
|
||||
queue.put_nowait(payload)
|
||||
if payload.get("type") == "event":
|
||||
self._log_event(payload["data"])
|
||||
self._maybe_notify(payload["data"])
|
||||
|
||||
def _log_event(self, event: dict[str, Any]) -> None:
|
||||
"""Mirror bot events into the logger so the dashboard log view is useful."""
|
||||
level = str(event.get("level", "info")).lower()
|
||||
message = event.get("message", "")
|
||||
logger.log(_LOG_LEVELS.get(level, logging.INFO), "%s", message)
|
||||
|
||||
def _maybe_notify(self, event: dict[str, Any]) -> None:
|
||||
if event.get("level") not in self.notify_levels:
|
||||
return
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from app.logs import RingLogHandler
|
||||
from app.metrics import render_metrics
|
||||
from app.schemas import MultiStatusOut, StrategyStatusOut
|
||||
|
||||
|
||||
def make_status(**overrides) -> MultiStatusOut:
|
||||
strategy = StrategyStatusOut(
|
||||
strategy_key="sma_cross",
|
||||
strategy_name="SMA Crossover",
|
||||
params={},
|
||||
symbol="BTC/USDT",
|
||||
timeframe="5m",
|
||||
allocation=0.5,
|
||||
cash=5000,
|
||||
equity=5100,
|
||||
realized_pnl=100,
|
||||
unrealized_pnl=0,
|
||||
total_pnl=100,
|
||||
open_positions=1,
|
||||
halted=False,
|
||||
trades_today=2,
|
||||
)
|
||||
base = {
|
||||
"running": True,
|
||||
"mode": "paper",
|
||||
"exchange": "binance",
|
||||
"price": 100.0,
|
||||
"starting_balance": 10000,
|
||||
"equity": 10100,
|
||||
"total_pnl": 100,
|
||||
"strategies": [strategy],
|
||||
}
|
||||
base.update(overrides)
|
||||
return MultiStatusOut(**base)
|
||||
|
||||
|
||||
def test_render_metrics_contains_core_series() -> None:
|
||||
body = render_metrics(make_status(), "paper", "binance")
|
||||
assert "crypto_running 1" in body
|
||||
assert "crypto_equity 10100" in body
|
||||
assert "crypto_total_pnl 100" in body
|
||||
assert "crypto_open_positions 1" in body
|
||||
assert 'crypto_strategy_equity{strategy="sma_cross",symbol="BTC/USDT"} 5100' in body
|
||||
assert 'crypto_strategy_allocation{strategy="sma_cross"} 0.5' in body
|
||||
assert 'crypto_info{mode="paper",exchange="binance"} 1' in body
|
||||
|
||||
|
||||
def test_render_metrics_omits_price_when_none() -> None:
|
||||
body = render_metrics(make_status(price=None), "paper", "binance")
|
||||
assert "crypto_price" not in body
|
||||
|
||||
|
||||
def test_ring_log_handler_captures() -> None:
|
||||
handler = RingLogHandler(capacity=5)
|
||||
logger = logging.getLogger("test.ring")
|
||||
logger.addHandler(handler)
|
||||
logger.setLevel(logging.INFO)
|
||||
try:
|
||||
logger.info("hello %s", "world")
|
||||
entries = handler.snapshot()
|
||||
assert entries[-1]["message"] == "hello world"
|
||||
assert entries[-1]["level"] == "info"
|
||||
finally:
|
||||
logger.removeHandler(handler)
|
||||
|
||||
|
||||
def test_ring_log_handler_respects_capacity() -> None:
|
||||
handler = RingLogHandler(capacity=3)
|
||||
logger = logging.getLogger("test.ring.cap")
|
||||
logger.addHandler(handler)
|
||||
logger.setLevel(logging.INFO)
|
||||
try:
|
||||
for i in range(10):
|
||||
logger.info("m%d", i)
|
||||
entries = handler.snapshot()
|
||||
assert len(entries) == 3
|
||||
assert entries[-1]["message"] == "m9"
|
||||
finally:
|
||||
logger.removeHandler(handler)
|
||||
|
||||
|
||||
def test_metrics_endpoint(runtime) -> None:
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.main import create_app
|
||||
|
||||
app = create_app(runtime.settings, runtime)
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/api/metrics")
|
||||
assert response.status_code == 200
|
||||
assert "crypto_running" in response.text
|
||||
|
||||
|
||||
def test_logs_endpoint(runtime) -> None:
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.main import create_app
|
||||
|
||||
app = create_app(runtime.settings, runtime)
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/api/logs")
|
||||
assert response.status_code == 200
|
||||
assert "entries" in response.json()
|
||||
Reference in new issue
Block a user