feat(cerbero): tools wrapper for Phase 1 indicator subset

Espone sma/rsi/atr/macd/realized_vol/funding_rate sopra CerberoClient
delegando a call_tool(exchange, tool, args).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-09 19:27:34 +02:00
parent 14b1b84a47
commit 7290900dc7
2 changed files with 92 additions and 0 deletions
+60
View File
@@ -0,0 +1,60 @@
from __future__ import annotations
from typing import Any
from .client import CerberoClient
class CerberoTools:
"""Sottoinsieme di tool MCP esposti agli agenti in Phase 1."""
def __init__(self, client: CerberoClient):
self._client = client
def sma(self, exchange: str, symbol: str, timeframe: str, length: int) -> Any:
return self._client.call_tool(
exchange, "sma", {"symbol": symbol, "timeframe": timeframe, "length": length}
)
def rsi(self, exchange: str, symbol: str, timeframe: str, length: int = 14) -> Any:
return self._client.call_tool(
exchange, "rsi", {"symbol": symbol, "timeframe": timeframe, "length": length}
)
def atr(self, exchange: str, symbol: str, timeframe: str, length: int = 14) -> Any:
return self._client.call_tool(
exchange, "atr", {"symbol": symbol, "timeframe": timeframe, "length": length}
)
def macd(
self,
exchange: str,
symbol: str,
timeframe: str,
fast: int = 12,
slow: int = 26,
signal: int = 9,
) -> Any:
return self._client.call_tool(
exchange,
"macd",
{
"symbol": symbol,
"timeframe": timeframe,
"fast": fast,
"slow": slow,
"signal": signal,
},
)
def realized_vol(
self, exchange: str, symbol: str, timeframe: str, window: int = 24
) -> Any:
return self._client.call_tool(
exchange,
"realized_vol",
{"symbol": symbol, "timeframe": timeframe, "window": window},
)
def funding_rate(self, exchange: str, symbol: str) -> Any:
return self._client.call_tool(exchange, "funding_rate", {"symbol": symbol})
+32
View File
@@ -0,0 +1,32 @@
import pytest
from multi_swarm.cerbero.tools import CerberoTools
def test_tools_dispatch_sma(mocker):
fake_client = mocker.MagicMock()
fake_client.call_tool.return_value = {"value": 100.0}
t = CerberoTools(fake_client)
out = t.sma(exchange="bybit", symbol="BTCUSDT", timeframe="1h", length=20)
fake_client.call_tool.assert_called_once_with(
"bybit", "sma", {"symbol": "BTCUSDT", "timeframe": "1h", "length": 20}
)
assert out == {"value": 100.0}
def test_tools_dispatch_rsi(mocker):
fake_client = mocker.MagicMock()
fake_client.call_tool.return_value = {"value": 55.0}
t = CerberoTools(fake_client)
out = t.rsi(exchange="bybit", symbol="BTCUSDT", timeframe="1h", length=14)
fake_client.call_tool.assert_called_once_with(
"bybit", "rsi", {"symbol": "BTCUSDT", "timeframe": "1h", "length": 14}
)
assert out == {"value": 55.0}
def test_tools_unknown_raises(mocker):
fake_client = mocker.MagicMock()
t = CerberoTools(fake_client)
with pytest.raises(AttributeError):
t.nonexistent_tool() # type: ignore[attr-defined]