Fix DS4 candle-limit freshness and completeness handling

hermes_overlay/tests/test_candle_limit_completeness.py

@@ -0,0 +1,239 @@
+"""Regression tests for the DS4 candle-limit clamp / completeness fix.
+
+Datasource 4's Short Hunter snapshot endpoint caps OHLCV history at
+DS4_MAX_OHLCV_CANDLES (200) candles per request. Requesting a wider
+window than DS4 can fill used to make it return the OLDEST candles in
+that window, which made the freshest returned candle look stale even
+though fresher data existed. The fix clamps the *request window* to
+the provider cap while still reporting the caller's original requested
+limit, so a 240-candle request against a 200-candle provider cap comes
+back temporally fresh and only quantity-partial -- never silently
+padded/fabricated and never marked stale for that reason alone.
+
+Mocked HTTP only -- no live network calls.
+"""
+import asyncio
+import os
+import sys
+import time
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))  # hermes_overlay/
+
+import trading.dual_datasource_client as ddc
+import tools.futures_dashboard_api as fda
+
+
+@pytest.fixture(autouse=True)
+def _disable_live_binance(monkeypatch):
+    async def _unavailable(symbol, needed):
+        return {}, []
+    monkeypatch.setattr(ddc.binance_public, "get_binance_public_snapshot", _unavailable)
+
+
+class _FakeResponse:
+    def __init__(self, payload, status_code=200):
+        self._payload = payload
+        self.status_code = status_code
+
+    def raise_for_status(self):
+        pass
+
+    def json(self):
+        return self._payload
+
+
+def _make_fake_client(ds4_payload, captured_params=None):
+    class _FakeClient:
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *a):
+            return False
+
+        async def get(self, url, timeout=None, params=None, **_kwargs):
+            if "short-hunter/snapshot" in url:
+                if captured_params is not None:
+                    captured_params.update(params or {})
+                return _FakeResponse(ds4_payload)
+            return _FakeResponse({})
+
+    return _FakeClient()
+
+
+def _make_candles(n, interval_seconds=300, end_ms=None):
+    """n candles at `interval_seconds` spacing, the last one dated `end_ms` (default: now)."""
+    end_ms = end_ms if end_ms is not None else int(time.time() * 1000)
+    return [
+        {
+            "timestamp": end_ms - (n - 1 - i) * interval_seconds * 1000,
+            "open": 100.0, "high": 101.0, "low": 99.0, "close": 100.5, "volume": 10.0,
+        }
+        for i in range(n)
+    ]
+
+
+def _ds4_payload(ohlcv):
+    return {
+        "success": True,
+        "symbol": "BTCUSDT",
+        "dataState": "live",
+        "noTradeGuard": False,
+        "timestamp": None,
+        "warnings": [],
+        "errors": [],
+        "data": {
+            "contract": {"symbol": "XBTUSDTM", "contractType": "PERPETUAL", "status": "TRADING"},
+            "ticker": {"lastPrice": 64364.4, "change24h": 1.2},
+            "ohlcv": ohlcv,
+            "orderbook": {"bids": [[64364.4, 1]], "asks": [[64364.5, 1]]},
+            "funding": {"currentFundingRate": -8.2e-05},
+            "openInterest": {"openInterest": 27000928.0, "history": []},
+            "indicators": {"rsi14": 49.3},
+            "sentiment": None,
+        },
+    }
+
+
+# ---------------------------------------------------------------------------
+# 1 & 2: within-provider-cap limits (60, 120) -- fresh and complete
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize("requested", [60, 120])
+def test_within_cap_limit_returns_fresh_complete(monkeypatch, requested):
+    payload = _ds4_payload(_make_candles(requested))
+    monkeypatch.setattr(ddc.httpx, "AsyncClient", lambda *a, **kw: _make_fake_client(payload))
+    result = asyncio.run(ddc.get_market_context("BTCUSDT", limit=requested))
+
+    assert result["requestedLimit"] == requested
+    assert result["providerLimit"] == requested
+    assert result["returnedCandleCount"] == requested
+    assert result["candleCompleteness"] == "complete"
+    assert result["freshness"]["datasource4"] == "fresh"
+
+
+# ---------------------------------------------------------------------------
+# 3: requested 240, provider cap 200 -> request window built for newest 200
+# ---------------------------------------------------------------------------
+
+def test_over_cap_request_window_uses_provider_cap(monkeypatch):
+    payload = _ds4_payload(_make_candles(200))
+    captured_params = {}
+    monkeypatch.setattr(
+        ddc.httpx, "AsyncClient",
+        lambda *a, **kw: _make_fake_client(payload, captured_params),
+    )
+    result = asyncio.run(ddc.get_market_context("BTCUSDT", limit=240))
+
+    assert captured_params.get("limit") == 200
+    assert result["providerLimit"] == 200
+
+
+# ---------------------------------------------------------------------------
+# 4: requested 240, returned 200 recent candles -> fresh and partial
+# ---------------------------------------------------------------------------
+
+def test_over_cap_returns_fresh_partial_not_stale(monkeypatch):
+    payload = _ds4_payload(_make_candles(200))
+    monkeypatch.setattr(ddc.httpx, "AsyncClient", lambda *a, **kw: _make_fake_client(payload))
+    result = asyncio.run(ddc.get_market_context("BTCUSDT", limit=240))
+
+    assert result["requestedLimit"] == 240
+    assert result["providerLimit"] == 200
+    assert result["returnedCandleCount"] == 200
+    assert result["candleCompleteness"] == "partial"
+    assert result["freshness"]["datasource4"] == "fresh"
+
+
+# ---------------------------------------------------------------------------
+# 5: genuinely old final candle -> stale
+# ---------------------------------------------------------------------------
+
+def test_genuinely_old_final_candle_is_stale(monkeypatch):
+    old_end = int(time.time() * 1000) - 3600 * 1000  # 1 hour old, well past the 15-minute window
+    payload = _ds4_payload(_make_candles(60, end_ms=old_end))
+    monkeypatch.setattr(ddc.httpx, "AsyncClient", lambda *a, **kw: _make_fake_client(payload))
+    result = asyncio.run(ddc.get_market_context("BTCUSDT", limit=60))
+
+    assert result["freshness"]["datasource4"] == "stale"
+
+
+# ---------------------------------------------------------------------------
+# 6: malformed candles -> invalid/unavailable
+# ---------------------------------------------------------------------------
+
+def test_malformed_candles_are_invalid_unavailable(monkeypatch):
+    malformed = [{"timestamp": None, "open": 1, "high": 1, "low": 1, "close": 1, "volume": 1}] * 5
+    payload = _ds4_payload(malformed)
+    monkeypatch.setattr(ddc.httpx, "AsyncClient", lambda *a, **kw: _make_fake_client(payload))
+    result = asyncio.run(ddc.get_market_context("BTCUSDT", limit=60))
+
+    assert result["merged"]["ohlcv"] is None
+    assert result["sources"]["ohlcv"] == "unavailable"
+    assert result["returnedCandleCount"] == 0
+
+
+# ---------------------------------------------------------------------------
+# 7: candle quantity mismatch does not invalidate other fresh fields
+# ---------------------------------------------------------------------------
+
+def test_candle_shortfall_does_not_invalidate_other_fresh_fields(monkeypatch):
+    payload = _ds4_payload(_make_candles(200))
+    monkeypatch.setattr(ddc.httpx, "AsyncClient", lambda *a, **kw: _make_fake_client(payload))
+    result = asyncio.run(ddc.get_market_context("BTCUSDT", limit=240))
+
+    assert result["candleCompleteness"] == "partial"
+    assert result["merged"]["ticker"]["lastPrice"] == 64364.4
+    assert result["merged"]["ticker"]["change24h"] == 1.2
+    assert result["merged"]["funding"] == {"currentFundingRate": -8.2e-05}
+    assert result["merged"]["orderbook"] == {"bids": [[64364.4, 1]], "asks": [[64364.5, 1]]}
+    assert result["merged"]["openInterest"] == {"openInterest": 27000928.0, "history": []}
+
+
+# ---------------------------------------------------------------------------
+# 8: API response exposes requestedLimit/providerLimit/returnedCount/completeness
+# ---------------------------------------------------------------------------
+
+def test_futures_market_endpoint_exposes_completeness_fields(monkeypatch):
+    payload = _ds4_payload(_make_candles(200))
+    monkeypatch.setattr(ddc.httpx, "AsyncClient", lambda *a, **kw: _make_fake_client(payload))
+    monkeypatch.setattr(fda, "_dash_state", fda._dash_state)  # keep real state module (no-op sanity)
+    monkeypatch.setattr(fda._dash_state, "record_market_context", lambda *a, **kw: None)
+
+    response = asyncio.run(fda.futures_market(symbol="BTCUSDT", interval="5m", limit=240))
+    body = response.body
+    import json as _json
+    data = _json.loads(body)
+
+    assert data["requestedLimit"] == 240
+    assert data["providerLimit"] == 200
+    assert data["returnedCount"] == 200
+    assert data["candleCompleteness"] == "partial"
+    assert data["freshness"] == "fresh"
+    assert data["state"] in {"available", "partial"}
+    assert len(data["candles"]) == 200
+
+
+# ---------------------------------------------------------------------------
+# 9: no candle padding or fabrication
+# ---------------------------------------------------------------------------
+
+def test_normalize_candles_does_not_pad_or_fabricate():
+    raw = _make_candles(200)
+    normalized = fda._normalize_candles(raw, 240)
+    assert len(normalized) == 200
+    # timestamps preserved verbatim, newest candle kept, nothing synthesized
+    assert normalized[-1]["timestamp"] == raw[-1]["timestamp"]
+
+
+# ---------------------------------------------------------------------------
+# 10: warning is emitted once
+# ---------------------------------------------------------------------------
+
+def test_over_cap_warning_emitted_once(monkeypatch):
+    payload = _ds4_payload(_make_candles(200))
+    monkeypatch.setattr(ddc.httpx, "AsyncClient", lambda *a, **kw: _make_fake_client(payload))
+    result = asyncio.run(ddc.get_market_context("BTCUSDT", limit=240))
+
+    matches = [w for w in result["warnings"] if "Datasource 4's maximum is 200" in w]
+    assert len(matches) == 1

hermes_overlay/tools/futures_dashboard_api.py

@@ -743,6 +743,10 @@ async def futures_market(
             "symbol": norm.ds4,
             "interval": interval,
             "limit": limit,
+            "requestedLimit": limit,
+            "providerLimit": None,
+            "returnedCount": 0,
+            "candleCompleteness": "unavailable",
             "candles": [],
             "currentPrice": None,
             "markPrice": None,
@@ -865,6 +869,10 @@ async def futures_market(
         "symbol": norm.ds4,
         "interval": interval,
         "limit": limit,
+        "requestedLimit": context.get("requestedLimit", limit),
+        "providerLimit": context.get("providerLimit"),
+        "returnedCount": context.get("returnedCandleCount", len(candles)),
+        "candleCompleteness": context.get("candleCompleteness"),
         "candles": candles,
         "currentPrice": price,
         "markPrice": mark_price,

hermes_overlay/trading/dual_datasource_client.py

@@ -41,6 +41,14 @@ FUTURES_CRITICAL_FIELDS = ("contract", "ticker", "orderbook", "funding", "openIn
 DS4_FRESH_STATES = {"live", "fresh", "ok", "real"}
 SUPPORTED_INTERVALS = {"1m": 60, "5m": 300, "15m": 900, "1h": 3600}
 MAX_MARKET_LIMIT = 500
+# Datasource 4's Short Hunter snapshot endpoint caps OHLCV history at 200
+# candles per request regardless of the `limit` requested. Requesting a wider
+# time window than DS4 can actually fill causes it to return the OLDEST
+# candles within that window rather than the most recent ones -- the request
+# window must therefore be built from this provider cap, never from the raw
+# caller-requested limit, or the "freshest" candle in the response can be
+# hours old. See dual-datasource candle-count completeness fix.
+DS4_MAX_OHLCV_CANDLES = 200
 
 
 def normalize_epoch_milliseconds(value: Any) -> int:
@@ -118,19 +126,26 @@ async def _get_json(
 
 async def _fetch_ds4_snapshot(
     client: httpx.AsyncClient, ds4_symbol: str, timeframe: str = "5m", limit: int = 120
-) -> tuple[Optional[dict], list[str], dict]:
+) -> tuple[Optional[dict], list[str], dict, int]:
     warnings: list[str] = []
     bounded_limit = max(1, min(int(limit), MAX_MARKET_LIMIT))
-    start_ms, end_ms = build_kucoin_time_range(timeframe, bounded_limit)
+    # Build the request window from what DS4 can actually deliver, not from
+    # the raw requested limit. Asking for a window wider than DS4's own cap
+    # makes it return the oldest candles in that window (stale), not the
+    # most recent ones -- see DS4_MAX_OHLCV_CANDLES above.
+    provider_limit = min(bounded_limit, DS4_MAX_OHLCV_CANDLES)
+    start_ms, end_ms = build_kucoin_time_range(timeframe, provider_limit)
     endpoint = f"{DS4_BASE}/api/short-hunter/snapshot/{ds4_symbol}"
     data, meta = await _get_json_with_meta(
         client, endpoint, DS4_TIMEOUT_S,
-        params={"interval": timeframe, "limit": bounded_limit,
+        params={"interval": timeframe, "limit": provider_limit,
                 "from": start_ms, "to": end_ms},
     )
     if data is None:
         warnings.append("Datasource 4 (Short Hunter) unreachable or errored")
-    return data, warnings, meta
+    # A more precise candle-count-mismatch warning (aware of how many candles
+    # were actually returned) is added by the caller, get_market_context.
+    return data, warnings, meta, provider_limit
 
 
 async def _fetch_ds2_context(
@@ -708,12 +723,13 @@ async def get_market_context(symbol: str, timeframe: str = "5m", limit: int = 12
     no_trade_reasons: list[str] = []
 
     async with httpx.AsyncClient() as client:
-        (ds4, ds4_warnings, ds4_transport), (ds2_context, ds2_warnings, ds2_transport) = await asyncio.gather(
+        (ds4, ds4_warnings, ds4_transport, ds4_provider_limit), (ds2_context, ds2_warnings, ds2_transport) = await asyncio.gather(
             _fetch_ds4_snapshot(client, norm.ds4, timeframe, limit),
             _fetch_ds2_context(client, norm.ds2),
         )
     warnings.extend(ds4_warnings)
     warnings.extend(ds2_warnings)
+    requested_limit = max(1, min(int(limit), MAX_MARKET_LIMIT))
 
     ds4_warnings_payload = _find_section(ds4, "warnings") if isinstance(ds4, dict) else None
     ds4_errors_payload = _find_section(ds4, "errors") if isinstance(ds4, dict) else None
@@ -1027,8 +1043,27 @@ async def get_market_context(symbol: str, timeframe: str = "5m", limit: int = 12
     trading_readiness = "ready" if not no_trade_guard else "blocked"
     merge_status = "complete" if not missing_critical and not stale_critical else "partial"
 
+    ohlcv_value = merged.get("ohlcv")
+    returned_candle_count = len(ohlcv_value) if isinstance(ohlcv_value, list) else 0
+    # Candle-count completeness is independent from temporal freshness: a
+    # response can hold fewer candles than requested (because the provider
+    # caps history length) while every candle in it is fully current. This
+    # compares against what the caller actually asked for, not the provider
+    # cap, so a 240-request answered with 200 fresh candles is reported as
+    # quantity-partial rather than silently rounded up to "complete".
+    candle_completeness = "complete" if returned_candle_count >= requested_limit else "partial"
+    if candle_completeness == "partial" and ds4_provider_limit < requested_limit and returned_candle_count >= ds4_provider_limit:
+        warnings.append(
+            f"Requested {requested_limit} candles; Datasource 4's maximum is "
+            f"{ds4_provider_limit} and {returned_candle_count} fresh candles were returned"
+        )
+
     return {
         "symbol": norm.ds4,
+        "requestedLimit": requested_limit,
+        "providerLimit": ds4_provider_limit,
+        "returnedCandleCount": returned_candle_count,
+        "candleCompleteness": candle_completeness,
         "verifiedFutures": bool(futures_verification.get("verified")),
         "futuresVerification": futures_verification,
         "primary": ds4 or {},