Files
crypto_trading/backend/app/logs.py
T
matitos ac92dd5ac4 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.
2026-09-22 17:10:38 +00:00

56 lines
1.6 KiB
Python

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