metals-mcp/test_metals.py

from datetime import datetime, timezone, timedelta

from src.metals_mcp import mcp_tools
from src.metals_mcp import poller as poller_module
from src.metals_mcp.mcp_tools import client, get_capabilities, get_candles, get_last_candle, get_market_snapshot, get_price, get_regime
from src.metals_mcp.storage import init_db, last_candle, prune_candles_older_than, upsert_candle


def test_capabilities():
    caps = get_capabilities()
    assert caps["server"] == "metals-mcp"


def test_scaffold_tools(tmp_path, monkeypatch):
    db_path = tmp_path / "metals.sqlite3"
    init_db(db_path)
    monkeypatch.setattr(mcp_tools, "DB_PATH", db_path)

    assert get_price("XAU/USD")["symbol"] == "XAU/USD"
    assert get_candles("XAU/USD")["candles"] == []
    assert get_last_candle("XAU/USD")["candle"] is None


def test_price_supports_counter_currency(monkeypatch):
    monkeypatch.setattr(client, "pair_is_supported", lambda symbol, counter_currency=None: True)

    class DummyQuote:
        mid = 4049.0
        timestamp = 1234567890

    monkeypatch.setattr(client, "fetch_quote", lambda pair: DummyQuote())
    quote = get_price("XAU", counter_currency="EUR")
    assert quote["pair"] == "XAU/EUR"
    assert quote["counter_currency"] == "EUR"
    assert quote["price"] == 4049.0


def test_retention_prunes_old_candles(tmp_path):
    db_path = tmp_path / "metals.sqlite3"
    init_db(db_path)
    now_ms = int(datetime.now(timezone.utc).timestamp() * 1000)

    upsert_candle(
        db_path,
        {
            "symbol": "XAU/USD",
            "timeframe": "5m",
            "open": 4000.0,
            "high": 4010.0,
            "low": 3990.0,
            "close": 4005.0,
            "start_ts": now_ms - int(timedelta(days=40).total_seconds() * 1000),
            "end_ts": now_ms - int(timedelta(days=40).total_seconds() * 1000) + 300000,
        },
    )
    upsert_candle(
        db_path,
        {
            "symbol": "XAU/USD",
            "timeframe": "5m",
            "open": 4050.0,
            "high": 4060.0,
            "low": 4040.0,
            "close": 4055.0,
            "start_ts": now_ms,
            "end_ts": now_ms + 300000,
        },
    )

    deleted = prune_candles_older_than(db_path, 30)
    assert deleted == 1


def test_poller_persists_current_candle(tmp_path, monkeypatch):
    db_path = tmp_path / "metals.sqlite3"
    init_db(db_path)
    monkeypatch.setattr(poller_module, "DB_PATH", db_path)
    monkeypatch.setattr(poller_module, "METALS_PAIRS", ["XAU/USD"])

    class DummyQuote:
        mid = 4050.0
        timestamp = int(datetime.now(timezone.utc).timestamp() * 1000)

    poller = poller_module.CandlePoller()
    monkeypatch.setattr(poller.client, "fetch_quote", lambda symbol: DummyQuote())
    poller.step()

    candle = last_candle(db_path, "XAU/USD", "5m")
    assert candle is not None
    assert candle["close"] == 4050.0


def test_snapshot_and_regime_use_recent_candles(tmp_path, monkeypatch):
    db_path = tmp_path / "metals.sqlite3"
    init_db(db_path)
    monkeypatch.setattr(mcp_tools, "DB_PATH", db_path)

    base_ts = 1_700_000_000_000
    for idx, close in enumerate((4000.0, 4010.0, 4020.0, 4035.0, 4050.0)):
        upsert_candle(
            db_path,
            {
                "symbol": "XAU/USD",
                "timeframe": "5m",
                "open": close - 5,
                "high": close + 12,
                "low": close - 12,
                "close": close,
                "start_ts": base_ts + idx * 300000,
                "end_ts": base_ts + (idx + 1) * 300000,
            },
        )

    market_snapshot = get_market_snapshot("XAU/USD")
    assert market_snapshot["status"] == "ok"
    assert market_snapshot["components"]["trend_pct"] is not None

    regime = get_regime("XAU/USD")
    assert regime["status"] == "ok"
    assert regime["regime"] in {"bullish", "neutral", "compression"}

    rsi = mcp_tools.get_indicator("XAU/USD", "rsi", params={"period": 3})
    assert rsi["status"] == "ok"