matitos
ac92dd5ac4
- 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.
108 lines
3.1 KiB
Python
108 lines
3.1 KiB
Python
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()
|