Adriano Dal Pastro
b6539802e0
- mv src/multi_swarm → src/multi_swarm_core/multi_swarm_core (member layout) - sed-replace globale degli import: from/import multi_swarm.* → multi_swarm_core.* - 115 occorrenze aggiornate in src/ scripts/ tests/ - multi_swarm_coevolutive (nome repo) preservato dal word boundary Pre-condizione per il setup uv workspace della Fase 3. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
141 lines
4.2 KiB
Python
141 lines
4.2 KiB
Python
"""Tests for CerberoOHLCVLoader (mocked CerberoClient)."""
|
|
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from multi_swarm_core.data.cerbero_ohlcv import CerberoOHLCVLoader, OHLCVRequest
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_records_object() -> list[dict[str, float | int]]:
|
|
base = int(datetime(2024, 1, 1, tzinfo=UTC).timestamp() * 1000)
|
|
return [
|
|
{
|
|
"ts": base + i * 3600 * 1000,
|
|
"open": 40000 + i,
|
|
"high": 40100 + i,
|
|
"low": 39900 + i,
|
|
"close": 40050 + i,
|
|
"volume": 100.0 + i,
|
|
}
|
|
for i in range(48)
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_records_array() -> list[list[float | int]]:
|
|
base = int(datetime(2024, 1, 1, tzinfo=UTC).timestamp() * 1000)
|
|
return [
|
|
[base + i * 3600 * 1000, 40000 + i, 40100 + i, 39900 + i, 40050 + i, 100.0 + i]
|
|
for i in range(48)
|
|
]
|
|
|
|
|
|
def test_loader_parses_object_records(
|
|
tmp_path: Path, mocker, sample_records_object
|
|
) -> None:
|
|
fake_client = mocker.MagicMock()
|
|
fake_client.call_tool.return_value = {"candles": sample_records_object}
|
|
|
|
loader = CerberoOHLCVLoader(client=fake_client, cache_dir=tmp_path)
|
|
df = loader.load(
|
|
OHLCVRequest(
|
|
symbol="BTC-PERPETUAL",
|
|
timeframe="1h",
|
|
start=datetime(2024, 1, 1, tzinfo=UTC),
|
|
end=datetime(2024, 1, 3, tzinfo=UTC),
|
|
exchange="deribit",
|
|
)
|
|
)
|
|
|
|
assert list(df.columns) == ["open", "high", "low", "close", "volume"]
|
|
assert len(df) == 48
|
|
assert df.index.tz is not None
|
|
fake_client.call_tool.assert_called_once_with(
|
|
"deribit",
|
|
"get_historical",
|
|
{
|
|
"instrument": "BTC-PERPETUAL",
|
|
"start_date": "2024-01-01T00:00:00+00:00",
|
|
"end_date": "2024-01-03T00:00:00+00:00",
|
|
"resolution": "1h",
|
|
},
|
|
)
|
|
|
|
|
|
def test_loader_parses_array_records(
|
|
tmp_path: Path, mocker, sample_records_array
|
|
) -> None:
|
|
fake_client = mocker.MagicMock()
|
|
fake_client.call_tool.return_value = {"candles": sample_records_array}
|
|
|
|
loader = CerberoOHLCVLoader(client=fake_client, cache_dir=tmp_path)
|
|
df = loader.load(
|
|
OHLCVRequest(
|
|
symbol="BTC-PERPETUAL",
|
|
timeframe="1h",
|
|
start=datetime(2024, 1, 1, tzinfo=UTC),
|
|
end=datetime(2024, 1, 3, tzinfo=UTC),
|
|
)
|
|
)
|
|
assert len(df) == 48
|
|
|
|
|
|
def test_loader_uses_cache_on_second_call(
|
|
tmp_path: Path, mocker, sample_records_object
|
|
) -> None:
|
|
fake_client = mocker.MagicMock()
|
|
fake_client.call_tool.return_value = {"candles": sample_records_object}
|
|
|
|
loader = CerberoOHLCVLoader(client=fake_client, cache_dir=tmp_path)
|
|
req = OHLCVRequest(
|
|
symbol="BTC-PERPETUAL",
|
|
timeframe="1h",
|
|
start=datetime(2024, 1, 1, tzinfo=UTC),
|
|
end=datetime(2024, 1, 3, tzinfo=UTC),
|
|
)
|
|
df1 = loader.load(req)
|
|
fake_client.call_tool.reset_mock()
|
|
df2 = loader.load(req)
|
|
assert fake_client.call_tool.call_count == 0
|
|
pd.testing.assert_frame_equal(df1, df2)
|
|
|
|
|
|
def test_loader_unsupported_exchange_raises(tmp_path: Path, mocker) -> None:
|
|
fake_client = mocker.MagicMock()
|
|
loader = CerberoOHLCVLoader(client=fake_client, cache_dir=tmp_path)
|
|
req = OHLCVRequest(
|
|
symbol="X",
|
|
timeframe="1h",
|
|
start=datetime(2024, 1, 1, tzinfo=UTC),
|
|
end=datetime(2024, 1, 2, tzinfo=UTC),
|
|
exchange="kraken",
|
|
)
|
|
with pytest.raises(ValueError, match="unsupported exchange"):
|
|
loader.load(req)
|
|
|
|
|
|
def test_loader_bybit_args(tmp_path: Path, mocker, sample_records_object) -> None:
|
|
fake_client = mocker.MagicMock()
|
|
fake_client.call_tool.return_value = {"candles": sample_records_object}
|
|
|
|
loader = CerberoOHLCVLoader(client=fake_client, cache_dir=tmp_path)
|
|
loader.load(
|
|
OHLCVRequest(
|
|
symbol="BTCUSDT",
|
|
timeframe="1h",
|
|
start=datetime(2024, 1, 1, tzinfo=UTC),
|
|
end=datetime(2024, 1, 3, tzinfo=UTC),
|
|
exchange="bybit",
|
|
)
|
|
)
|
|
args = fake_client.call_tool.call_args.args
|
|
assert args[0] == "bybit"
|
|
assert args[1] == "get_historical"
|
|
payload = args[2]
|
|
assert payload["symbol"] == "BTCUSDT"
|
|
assert payload["interval"] == 60
|