Stabilize metals candle store

PROJECT.md

@@ -38,12 +38,12 @@ Tool calls are performed via FastMCP’s message transport under `/mcp/messages/
 - Live Swissquote `get_price` for metals like `XAU`
 - Background poller that keeps the server alive and updating
 - Clock-aligned 5m candle storage in SQLite
-- Crypto-like tool naming where it helps client compatibility
+- Crypto-like tool naming on the shared surface, with `get_last_candle` as a small convenience helper
+- Gold, silver, platinum, and palladium candles are tracked by default, with context surfaced through the shared `get_market_snapshot` and `get_regime` names
 
 ## Planned
 
 - Real regime calculations for metals
-- Retention policy for historical candles
 - Additional pair coverage only if it proves useful
 
 ## Notes
@@ -51,6 +51,7 @@ Tool calls are performed via FastMCP’s message transport under `/mcp/messages/
 - Keep the transport small and predictable.
 - Mirror the crypto server’s MCP/HTTP shape where it helps client compatibility.
 - Keep Swissquote polling, storage, and candle aggregation behind the tools.
+- Candle retention defaults to 30 days via `METALS_CANDLE_RETENTION_DAYS`, with periodic pruning in the poller.
 
 ## Verification
 

README.md

@@ -19,6 +19,14 @@ pip install -r requirements.txt
 
 Default URL base: `http://127.0.0.1:8515`
 
+### Docker / Compose
+
+```bash
+docker compose up -d --build
+```
+
+The compose file persists `data/` and `logs/` into the project directory and keeps the candle DB bounded.
+
 ## HTTP
 
 - `GET /` → health + tool list
@@ -27,20 +35,26 @@ Default URL base: `http://127.0.0.1:8515`
 ## Tools
 
 - `get_price`
-- `get_candles`
+- `get_ohlcv`
 - `get_last_candle`
+- `get_market_snapshot`
+- `get_indicator`
+- `get_top_movers`
 - `get_capabilities`
+- `get_regime`
 
 ## Notes
 
 ### Done
-- MCP surface mirrors `crypto-mcp` tool names where it makes sense.
+- Shared MCP surface mirrors `crypto-mcp` tool names, with `get_last_candle` kept as a small metals-specific convenience.
 - `get_price` fetches live Swissquote quotes for metals like `XAU`.
 - An internal background poller keeps the server self-sufficient.
-- 5m candles are clock-aligned and persisted in SQLite.
+- 5m candles are clock-aligned and persisted in SQLite for `XAU/USD`, `XAG/USD`, `XPT/USD`, and `XPD/USD` by default.
+- `get_market_snapshot` and `get_regime` now provide the useful metals context in crypto-style naming.
+- Candle retention is bounded by `METALS_CANDLE_RETENTION_DAYS` (default 30), with periodic pruning to keep the DB bounded.
 
 ### Planned
-- Fill out `get_indicator`, `get_market_snapshot`, `get_top_movers`, and `get_regime` with real regime math.
+- Keep refining `get_indicator`, `get_market_snapshot`, `get_top_movers`, and `get_regime` as more metals behavior becomes useful to Argus.
 - Add retention cleanup for older candles.
 - Expand the watched pair set only if it proves useful for market orientation.
 

docker-compose.yml

@@ -1,7 +1,15 @@
 services:
   metals-mcp:
     build: .
+    restart: unless-stopped
     ports:
       - "8515:8515"
     environment:
       - METALS_PORT=8515
+      - METALS_DATA_DIR=/app/data
+      - METALS_LOG_DIR=/app/logs
+      - METALS_DB_PATH=/app/data/metals.sqlite3
+      - METALS_CANDLE_RETENTION_DAYS=30
+    volumes:
+      - ./data:/app/data
+      - ./logs:/app/logs

src/metals_mcp/config.py

@@ -10,4 +10,5 @@ DATA_DIR = Path(os.getenv("METALS_DATA_DIR", BASE_DIR / "data"))
 LOG_DIR = Path(os.getenv("METALS_LOG_DIR", BASE_DIR / "logs"))
 DB_PATH = Path(os.getenv("METALS_DB_PATH", DATA_DIR / "metals.sqlite3"))
 POLL_INTERVAL_SECONDS = float(os.getenv("SWISSQUOTE_POLL_INTERVAL_SECONDS", "0.2"))
-METALS_PAIRS = [p.strip().upper() for p in os.getenv("METALS_PAIRS", "XAU/USD,XAG/USD").split(",") if p.strip()]
+METALS_PAIRS = [p.strip().upper() for p in os.getenv("METALS_PAIRS", "XAU/USD,XAG/USD,XPT/USD,XPD/USD").split(",") if p.strip()]
+METALS_CANDLE_RETENTION_DAYS = int(os.getenv("METALS_CANDLE_RETENTION_DAYS", "30"))

src/metals_mcp/mcp_tools.py

@@ -1,13 +1,174 @@
 from __future__ import annotations
 
+from datetime import datetime, timezone
+from statistics import pstdev
 from typing import Any
 
-from .config import DB_PATH
+from .config import DB_PATH, METALS_PAIRS
 from .storage import last_candle, latest_candles, stats
 from .swissquote import SwissquoteClient
 
 client = SwissquoteClient()
 
+WATCHLIST = tuple(METALS_PAIRS)
+
+
+def _canonical_symbol(symbol: str) -> str:
+    return client.normalize_symbol(symbol).upper()
+
+
+def _to_float(value: Any) -> float | None:
+    if value is None:
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _closes(candles: list[dict[str, Any]]) -> list[float]:
+    closes: list[float] = []
+    for candle in candles:
+        close = _to_float(candle.get("close"))
+        if close is not None:
+            closes.append(close)
+    return closes
+
+
+def _sma(values: list[float], period: int) -> float | None:
+    if period <= 0 or len(values) < period:
+        return None
+    window = values[-period:]
+    return sum(window) / len(window)
+
+
+def _ema(values: list[float], period: int) -> float | None:
+    if period <= 0 or len(values) < period:
+        return None
+    k = 2.0 / (period + 1.0)
+    ema = sum(values[:period]) / period
+    for value in values[period:]:
+        ema = value * k + ema * (1.0 - k)
+    return ema
+
+
+def _rsi(values: list[float], period: int = 14) -> float | None:
+    if period <= 0 or len(values) < period + 1:
+        return None
+    gains = 0.0
+    losses = 0.0
+    for previous, current in zip(values[-(period + 1) : -1], values[-period:]):
+        delta = current - previous
+        if delta >= 0:
+            gains += delta
+        else:
+            losses -= delta
+    if gains == 0 and losses == 0:
+        return 50.0
+    if losses == 0:
+        return 100.0
+    rs = gains / losses
+    return 100.0 - (100.0 / (1.0 + rs))
+
+
+def _atr_pct(candles: list[dict[str, Any]], period: int = 14) -> float | None:
+    if period <= 0 or len(candles) < period + 1:
+        return None
+    true_ranges: list[float] = []
+    for idx in range(1, len(candles)):
+        high = _to_float(candles[idx].get("high"))
+        low = _to_float(candles[idx].get("low"))
+        prev_close = _to_float(candles[idx - 1].get("close"))
+        if high is None or low is None or prev_close is None:
+            continue
+        true_ranges.append(max(high - low, abs(high - prev_close), abs(low - prev_close)))
+    if len(true_ranges) < period:
+        return None
+    atr = sum(true_ranges[-period:]) / period
+    last_close = _to_float(candles[-1].get("close"))
+    if last_close in (None, 0):
+        return None
+    return (atr / last_close) * 100.0
+
+
+def _trend_pct(closes: list[float]) -> float | None:
+    if len(closes) < 2:
+        return None
+    first = closes[0]
+    last = closes[-1]
+    if first == 0:
+        return None
+    return ((last - first) / first) * 100.0
+
+
+def _range_position(candles: list[dict[str, Any]]) -> float | None:
+    highs = [_to_float(c.get("high")) for c in candles]
+    lows = [_to_float(c.get("low")) for c in candles]
+    closes = [_to_float(c.get("close")) for c in candles]
+    highs = [v for v in highs if v is not None]
+    lows = [v for v in lows if v is not None]
+    closes = [v for v in closes if v is not None]
+    if not highs or not lows or not closes:
+        return None
+    low = min(lows)
+    high = max(highs)
+    if high <= low:
+        return 0.5
+    return (closes[-1] - low) / (high - low)
+
+
+def _derive_regime(trend_pct: float | None, rsi: float | None, atr_pct: float | None) -> tuple[str, float, str]:
+    if trend_pct is None:
+        return "no_data", 0.0, "Not enough candle history to classify the market."
+
+    compression = atr_pct is not None and atr_pct < 0.25
+    bullish = trend_pct >= 0.35 and (rsi is None or rsi >= 55)
+    bearish = trend_pct <= -0.35 and (rsi is None or rsi <= 45)
+
+    if compression:
+        confidence = min(0.65, max(0.2, 0.35 + (0.25 - (atr_pct or 0.0))))
+        return "compression", confidence, "Price action is compressed and range-like."
+    if bullish:
+        confidence = min(0.9, 0.5 + min(abs(trend_pct) / 4.0, 0.4))
+        return "bullish", confidence, "Momentum is supportive and the trend is holding up."
+    if bearish:
+        confidence = min(0.9, 0.5 + min(abs(trend_pct) / 4.0, 0.4))
+        return "bearish", confidence, "Momentum is weak and sellers are in control."
+
+    confidence = 0.35
+    if rsi is not None:
+        confidence += min(abs(rsi - 50.0) / 100.0, 0.25)
+    return "neutral", confidence, "Signals are mixed and the market is not strongly directional."
+
+
+def _symbol_snapshot(symbol: str, timeframe: str = "5m", limit: int = 20) -> dict[str, Any]:
+    canonical = _canonical_symbol(symbol)
+    candles = latest_candles(DB_PATH, canonical, timeframe, limit)
+    last = last_candle(DB_PATH, canonical, timeframe)
+    closes = _closes(candles)
+    trend_pct = _trend_pct(closes)
+    rsi = _rsi(closes)
+    atr_pct = _atr_pct(candles)
+    range_position = _range_position(candles)
+    regime, confidence, summary = _derive_regime(trend_pct, rsi, atr_pct)
+
+    return {
+        "symbol": symbol.upper(),
+        "pair": canonical,
+        "timeframe": timeframe,
+        "candle_count": len(candles),
+        "price": last["close"] if last else None,
+        "change_pct": trend_pct,
+        "rsi": rsi,
+        "atr_pct": atr_pct,
+        "range_position": range_position,
+        "regime": regime,
+        "confidence": confidence,
+        "summary": summary,
+        "last_candle": last,
+        "candles": candles,
+    }
+
 
 def get_capabilities() -> dict[str, Any]:
     return {
@@ -19,6 +180,7 @@ def get_capabilities() -> dict[str, Any]:
             "get_indicator",
             "get_market_snapshot",
             "get_top_movers",
+            "get_last_candle",
             "get_capabilities",
             "get_regime",
         ],
@@ -63,8 +225,9 @@ def get_price(symbol: str, counter_currency: str | None = None) -> dict[str, Any
 
 
 def get_ohlcv(symbol: str, timeframe: str = "5m", limit: int = 100) -> dict[str, Any]:
-    candles = latest_candles(DB_PATH, symbol.upper(), timeframe, limit)
-    return {"symbol": symbol.upper(), "timeframe": timeframe, "limit": limit, "candles": candles}
+    canonical = _canonical_symbol(symbol)
+    candles = latest_candles(DB_PATH, canonical, timeframe, limit)
+    return {"symbol": canonical, "timeframe": timeframe, "limit": limit, "candles": candles}
 
 
 def get_candles(symbol: str, timeframe: str = "5m", limit: int = 100) -> dict[str, Any]:
@@ -72,26 +235,150 @@ def get_candles(symbol: str, timeframe: str = "5m", limit: int = 100) -> dict[st
 
 
 def get_last_candle(symbol: str, timeframe: str = "5m") -> dict[str, Any]:
-    return {"symbol": symbol.upper(), "timeframe": timeframe, "candle": last_candle(DB_PATH, symbol.upper(), timeframe)}
+    canonical = _canonical_symbol(symbol)
+    return {"symbol": canonical, "timeframe": timeframe, "candle": last_candle(DB_PATH, canonical, timeframe)}
 
 
 def get_indicator(symbol: str, indicator: str, timeframe: str = "5m", params: dict[str, Any] | None = None) -> dict[str, Any]:
-    return {"symbol": symbol, "indicator": indicator, "timeframe": timeframe, "params": params or {}, "value": None, "status": "scaffolded"}
+    params = params or {}
+    limit = int(params.get("limit", 50))
+    canonical = _canonical_symbol(symbol)
+    candles = latest_candles(DB_PATH, canonical, timeframe, max(limit, 2))
+    closes = _closes(candles)
+    name = indicator.lower().strip()
+
+    value: float | None
+    if name == "sma":
+        value = _sma(closes, int(params.get("period", 14)))
+    elif name == "ema":
+        value = _ema(closes, int(params.get("period", 14)))
+    elif name == "rsi":
+        value = _rsi(closes, int(params.get("period", 14)))
+    elif name == "atr":
+        value = _atr_pct(candles, int(params.get("period", 14)))
+    elif name == "return_pct":
+        value = _trend_pct(closes)
+    elif name == "volatility":
+        returns = [((curr - prev) / prev) * 100.0 for prev, curr in zip(closes, closes[1:]) if prev]
+        value = pstdev(returns) if len(returns) >= 2 else None
+    else:
+        value = None
+
+    status = "ok" if value is not None else "unavailable"
+    return {
+        "symbol": canonical,
+        "indicator": name,
+        "timeframe": timeframe,
+        "params": params,
+        "value": value,
+        "status": status,
+    }
 
 
 def get_market_snapshot(symbol: str) -> dict[str, Any]:
-    candle = last_candle(DB_PATH, symbol.upper(), "5m")
-    price = candle["close"] if candle else None
-    return {"symbol": symbol.upper(), "price": price, "trend_bias": "range", "status": "scaffolded"}
+    snapshot = _symbol_snapshot(symbol, "5m", 20)
+    return {
+        "symbol": snapshot["symbol"],
+        "timeframe": snapshot["timeframe"],
+        "price": snapshot["price"],
+        "change_pct": snapshot["change_pct"],
+        "trend_bias": snapshot["regime"],
+        "confidence": snapshot["confidence"],
+        "summary": snapshot["summary"],
+        "components": {
+            "trend_pct": snapshot["change_pct"],
+            "rsi": snapshot["rsi"],
+            "atr_pct": snapshot["atr_pct"],
+            "range_position": snapshot["range_position"],
+        },
+        "last_candle": snapshot["last_candle"],
+        "status": "ok" if snapshot["candle_count"] else "no_data",
+    }
+
+
+def get_combined_snapshot(timeframe: str = "5m", limit: int = 20) -> dict[str, Any]:
+    symbols = {symbol: _symbol_snapshot(symbol, timeframe, limit) for symbol in WATCHLIST}
+    gold = symbols.get("XAU/USD")
+    silver = symbols.get("XAG/USD")
+
+    ratio = None
+    ratio_change = None
+    if gold and silver and gold["price"] and silver["price"]:
+        ratio = gold["price"] / silver["price"]
+        if gold["change_pct"] is not None and silver["change_pct"] is not None:
+            ratio_change = gold["change_pct"] - silver["change_pct"]
+
+    regimes = [item["regime"] for item in symbols.values() if item["regime"] != "no_data"]
+    if not regimes:
+        regime = "no_data"
+        confidence = 0.0
+        summary = "No metals data available yet."
+    elif all(r == "bullish" for r in regimes):
+        regime = "hard_asset_bid"
+        confidence = min(0.95, sum(item["confidence"] for item in symbols.values()) / len(symbols))
+        summary = "Gold and silver are both trending higher."
+    elif all(r == "bearish" for r in regimes):
+        regime = "hard_asset_pressure"
+        confidence = min(0.95, sum(item["confidence"] for item in symbols.values()) / len(symbols))
+        summary = "Gold and silver are both trending lower."
+    elif any(r == "compression" for r in regimes):
+        regime = "compression"
+        confidence = 0.45
+        summary = "Metals are compressed and waiting for a break."
+    else:
+        regime = "neutral"
+        confidence = 0.4
+        summary = "Metals are mixed and not giving a clean directional read."
+
+    return {
+        "server": "metals-mcp",
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "timeframe": timeframe,
+        "symbols": symbols,
+        "cross_asset": {
+            "gold_silver_ratio": ratio,
+            "gold_silver_ratio_change_pct": ratio_change,
+        },
+        "regime": regime,
+        "confidence": confidence,
+        "summary": summary,
+        "status": "ok" if symbols else "no_data",
+    }
 
 
 def get_top_movers(limit: int = 10) -> dict[str, Any]:
-    return {"limit": limit, "movers": [], "status": "scaffolded"}
+    movers = []
+    for symbol in WATCHLIST:
+        snapshot = _symbol_snapshot(symbol, "5m", 20)
+        movers.append(
+            {
+                "symbol": symbol,
+                "change_pct": snapshot["change_pct"],
+                "regime": snapshot["regime"],
+                "confidence": snapshot["confidence"],
+            }
+        )
+    movers = sorted(movers, key=lambda item: abs(item["change_pct"] or 0.0), reverse=True)[:limit]
+    return {"limit": limit, "movers": movers, "status": "ok"}
 
 
 def get_regime(symbol: str, timeframe: str = "5m") -> dict[str, Any]:
-    candles = latest_candles(DB_PATH, symbol.upper(), timeframe, 20)
-    return {"symbol": symbol.upper(), "timeframe": timeframe, "candles": candles, "regime": None, "status": "scaffolded"}
+    snapshot = _symbol_snapshot(symbol, timeframe, 20)
+    return {
+        "symbol": snapshot["symbol"],
+        "timeframe": snapshot["timeframe"],
+        "regime": snapshot["regime"],
+        "confidence": snapshot["confidence"],
+        "summary": snapshot["summary"],
+        "components": {
+            "trend_pct": snapshot["change_pct"],
+            "rsi": snapshot["rsi"],
+            "atr_pct": snapshot["atr_pct"],
+            "range_position": snapshot["range_position"],
+        },
+        "candles": snapshot["candles"],
+        "status": "ok" if snapshot["candle_count"] else "no_data",
+    }
 
 
 def get_health() -> dict[str, Any]:

src/metals_mcp/poller.py

@@ -1,13 +1,12 @@
 from __future__ import annotations
 
-import math
 import time
 from dataclasses import dataclass
 from typing import Any
 import logging
 
-from .config import DB_PATH, METALS_PAIRS, POLL_INTERVAL_SECONDS
-from .storage import init_db, upsert_candle
+from .config import DB_PATH, METALS_CANDLE_RETENTION_DAYS, METALS_PAIRS, POLL_INTERVAL_SECONDS
+from .storage import init_db, prune_candles_older_than, upsert_candle
 from .swissquote import SwissquoteClient
 
 TIMEFRAME_SECONDS = 300
@@ -46,6 +45,7 @@ class CandlePoller:
     def __init__(self) -> None:
         self.client = SwissquoteClient()
         self.states: dict[str, CandleState] = {}
+        self._last_prune_ts = 0.0
         init_db(DB_PATH)
 
     def bucket_start(self, ts_ms: int) -> int:
@@ -73,6 +73,14 @@ class CandlePoller:
             else:
                 state.update(quote.mid)
 
+        for state in self.states.values():
+            upsert_candle(DB_PATH, state.to_row())
+
+        now = time.monotonic()
+        if now - self._last_prune_ts >= 3600:
+            prune_candles_older_than(DB_PATH, METALS_CANDLE_RETENTION_DAYS)
+            self._last_prune_ts = now
+
     def flush(self) -> None:
         for state in self.states.values():
             upsert_candle(DB_PATH, state.to_row())

src/metals_mcp/server_fastmcp.py

@@ -12,6 +12,7 @@ from .mcp_tools import (
     get_indicator as _get_indicator,
     get_market_snapshot as _get_market_snapshot,
     get_ohlcv as _get_ohlcv,
+    get_last_candle as _get_last_candle,
     get_price as _get_price,
     get_regime as _get_regime,
     get_top_movers as _get_top_movers,
@@ -38,6 +39,11 @@ def get_ohlcv(symbol: str, timeframe: str = "5m", limit: int = 100):
     return _get_ohlcv(symbol, timeframe=timeframe, limit=limit)
 
 
+@mcp.tool()
+def get_last_candle(symbol: str, timeframe: str = "5m"):
+    return _get_last_candle(symbol, timeframe=timeframe)
+
+
 @mcp.tool()
 def get_indicator(symbol: str, indicator: str, timeframe: str = "5m", params: dict | None = None):
     return _get_indicator(symbol, indicator, timeframe=timeframe, params=params)
@@ -65,7 +71,20 @@ def get_regime(symbol: str, timeframe: str = "5m"):
 
 @app.get("/")
 def root():
-    return {"ok": True, "server": "metals-mcp", "tools": ["get_price", "get_candles", "get_last_candle", "get_capabilities"]}
+    return {
+        "ok": True,
+        "server": "metals-mcp",
+        "tools": [
+            "get_price",
+            "get_ohlcv",
+            "get_last_candle",
+            "get_indicator",
+            "get_market_snapshot",
+            "get_top_movers",
+            "get_regime",
+            "get_capabilities",
+        ],
+    }
 
 
 @app.get("/health")

src/metals_mcp/storage.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import sqlite3
+from datetime import datetime, timezone, timedelta
 from pathlib import Path
 from typing import Any
 
@@ -31,6 +32,9 @@ def init_db(db_path: str | Path) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     with connect(path) as conn:
         conn.executescript(SCHEMA)
+        conn.execute(
+            "CREATE UNIQUE INDEX IF NOT EXISTS idx_candles_symbol_timeframe_start_ts ON candles(symbol, timeframe, start_ts)"
+        )
         conn.commit()
 
 
@@ -95,3 +99,13 @@ def stats(db_path: str | Path) -> dict[str, Any]:
     with connect(db_path) as conn:
         candles = conn.execute("SELECT COUNT(*) AS n FROM candles").fetchone()["n"]
     return {"candles": candles}
+
+
+def prune_candles_older_than(db_path: str | Path, days: int) -> int:
+    if days <= 0:
+        return 0
+    cutoff = int((datetime.now(timezone.utc) - timedelta(days=days)).timestamp() * 1000)
+    with connect(db_path) as conn:
+        cursor = conn.execute("DELETE FROM candles WHERE end_ts < ?", (cutoff,))
+        conn.commit()
+        return int(cursor.rowcount or 0)

test_metals.py

@@ -1,4 +1,9 @@
-from src.metals_mcp.mcp_tools import client, get_capabilities, get_candles, get_last_candle, get_price
+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():
@@ -6,7 +11,11 @@ def test_capabilities():
     assert caps["server"] == "metals-mcp"
 
 
-def test_scaffold_tools():
+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
@@ -24,3 +33,91 @@ def test_price_supports_counter_currency(monkeypatch):
     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"