From 79fa1d5fdec099115dbc4e110eb127ac90c7d113 Mon Sep 17 00:00:00 2001 From: vcmbob Date: Wed, 8 Jul 2026 06:24:39 +0000 Subject: [PATCH 1/4] Add Binance futures derivatives data endpoints --- DATA_LAYER_SERVICE_ACCESS_GUIDE.md | 55 ++++ app/api/routes_binance_derivatives.py | 198 ++++++++++++ app/main.py | 11 +- app/providers/binance/derivatives.py | 353 +++++++++++++++++++++ app/sdk/client.py | 114 +++++++ tests/test_binance_derivatives_contract.py | 140 ++++++++ 6 files changed, 870 insertions(+), 1 deletion(-) create mode 100644 app/api/routes_binance_derivatives.py create mode 100644 app/providers/binance/derivatives.py create mode 100644 tests/test_binance_derivatives_contract.py diff --git a/DATA_LAYER_SERVICE_ACCESS_GUIDE.md b/DATA_LAYER_SERVICE_ACCESS_GUIDE.md index bdb3927..6fcf24a 100644 --- a/DATA_LAYER_SERVICE_ACCESS_GUIDE.md +++ b/DATA_LAYER_SERVICE_ACCESS_GUIDE.md @@ -327,6 +327,61 @@ Important crypto parser note: - This was the root cause of the rsibound warmup issue where `RIFUSDT`, `COMPUSDT`, and similar symbols returned data from data_layer but alpha logged `rows=0`. - The batch endpoint is additive. It does not replace or mutate the existing single-symbol endpoint contract. +### Pattern E: Binance Futures Derivatives Metrics For Basis-Arb + +Basis-arbitrage and futures microstructure alpha must request derivatives metrics through `data_layer`. +Do not call Binance directly from alpha containers. + +These endpoints are direct, non-storage wrappers around official Binance USDⓈ-M Futures market-data APIs: + +- `GET /v1/binance/futures/exchange-info?symbol=BTCUSDT_260925` +- `GET /v1/binance/futures/klines/BTCUSDT_260925?interval=1d&limit=30` +- `GET /v1/binance/futures/depth/BTCUSDT?limit=5` +- `GET /v1/binance/futures/open-interest/BTCUSDT` +- `GET /v1/binance/futures/open-interest-history/BTCUSDT?period=1d&limit=30` +- `GET /v1/binance/futures/long-short/global_account/BTCUSDT?period=1d&limit=30` +- `GET /v1/binance/futures/long-short/top_account/BTCUSDT?period=1d&limit=30` +- `GET /v1/binance/futures/long-short/top_position/BTCUSDT?period=1d&limit=30` +- `GET /v1/binance/futures/taker-long-short/BTCUSDT?period=1d&limit=30` +- `GET /v1/binance/futures/funding-rate/BTCUSDT?limit=100` +- `GET /v1/binance/futures/basis/BTCUSDT?contract_type=CURRENT_QUARTER&period=1d&limit=30` + +Always resolve delivery symbols from `exchange-info` first. Do not guess date suffixes: + +```bash +GET http://data_layer:8100/v1/binance/futures/exchange-info +``` + +At the 2026-07-08 smoke test, Binance listed `BTCUSDT_260925` as `CURRENT_QUARTER` and +`BTCUSDT_261225` as `NEXT_QUARTER`; `BTCUSDT_260926` was rejected by Binance as an invalid symbol. + +Preferred alpha warmup/rebalance request for basis-arb: + +```bash +POST http://data_layer:8100/v1/binance/futures/basis-bundle +Content-Type: application/json + +{ + "perp_symbol": "BTCUSDT", + "delivery_symbol": "BTCUSDT_260925", + "pair": "BTCUSDT", + "interval": "1d", + "period": "1d", + "limit": 30, + "include_depth": true, + "depth_limit": 5, + "contract_type": "CURRENT_QUARTER" +} +``` + +Bundle response contract: + +- `components` contains successful raw provider payloads keyed by component name. +- `errors` contains per-component failures. +- `partial=true` means at least one component failed; alpha must decide whether the missing component is fatal. +- `cached=false` and `stored=false`; data_layer does not persist these derivatives metrics to parquet/Redis history. +- Futures-data history endpoints are latest-30-days style datasets. Consumers must not assume older history is available from these wrappers. + Fallback semantics: - `GET http://data_layer:8100/v1/fallback/crypto/status/BTCUSDT?interval=1m` diff --git a/app/api/routes_binance_derivatives.py b/app/api/routes_binance_derivatives.py new file mode 100644 index 0000000..5f399ee --- /dev/null +++ b/app/api/routes_binance_derivatives.py @@ -0,0 +1,198 @@ +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Body, HTTPException, Query + +from app.providers.binance import derivatives as binance_derivatives +from app.providers.binance.rest import BINANCE_KLINE_INTERVALS, BinanceProviderError + + +router = APIRouter(prefix="/v1/binance/futures", tags=["binance-futures"]) + + +def _bad_request(exc: ValueError, *, supported: Any | None = None): + detail: dict[str, Any] = {"error": str(exc)} + if supported is not None: + detail["supported"] = sorted(supported) + raise HTTPException(status_code=400, detail=detail) + + +def _provider_error(exc: BinanceProviderError): + raise HTTPException(status_code=502, detail={"error": str(exc), "attempts": exc.attempts}) + + +@router.get("/exchange-info") +async def get_exchange_info(symbol: str | None = None): + try: + return binance_derivatives.fetch_exchange_info(symbol) + except BinanceProviderError as exc: + _provider_error(exc) + + +@router.get("/klines/{symbol}") +async def get_derivative_klines( + symbol: str, + interval: str = Query("1d", description="Binance kline interval"), + limit: int = Query(30, ge=1, le=1500), + start_time: int | None = None, + end_time: int | None = None, +): + try: + return binance_derivatives.fetch_klines(symbol, interval, limit, start_time, end_time) + except ValueError as exc: + _bad_request(exc, supported=BINANCE_KLINE_INTERVALS) + except BinanceProviderError as exc: + _provider_error(exc) + + +@router.get("/depth/{symbol}") +async def get_derivative_depth(symbol: str, limit: int = Query(5)): + try: + return binance_derivatives.fetch_depth(symbol, limit) + except ValueError as exc: + _bad_request(exc) + except BinanceProviderError as exc: + _provider_error(exc) + + +@router.get("/open-interest/{symbol}") +async def get_open_interest(symbol: str): + try: + return binance_derivatives.fetch_open_interest(symbol) + except BinanceProviderError as exc: + _provider_error(exc) + + +@router.get("/open-interest-history/{symbol}") +async def get_open_interest_history( + symbol: str, + period: str = Query("1d"), + limit: int = Query(30, ge=1, le=500), + start_time: int | None = None, + end_time: int | None = None, +): + try: + return binance_derivatives.fetch_metric_history( + "open_interest_hist", + symbol, + period, + limit, + start_time, + end_time, + ) + except ValueError as exc: + _bad_request(exc, supported=binance_derivatives.BINANCE_DERIVATIVE_PERIODS) + except BinanceProviderError as exc: + _provider_error(exc) + + +@router.get("/long-short/{kind}/{symbol}") +async def get_long_short_ratio( + kind: str, + symbol: str, + period: str = Query("1d"), + limit: int = Query(30, ge=1, le=500), + start_time: int | None = None, + end_time: int | None = None, +): + try: + return binance_derivatives.fetch_long_short_ratio( + kind, + symbol, + period, + limit, + start_time, + end_time, + ) + except ValueError as exc: + _bad_request(exc, supported=binance_derivatives.LONG_SHORT_KIND_TO_ENDPOINT) + except BinanceProviderError as exc: + _provider_error(exc) + + +@router.get("/taker-long-short/{symbol}") +async def get_taker_long_short_ratio( + symbol: str, + period: str = Query("1d"), + limit: int = Query(30, ge=1, le=500), + start_time: int | None = None, + end_time: int | None = None, +): + try: + return binance_derivatives.fetch_taker_long_short_ratio( + symbol, + period, + limit, + start_time, + end_time, + ) + except ValueError as exc: + _bad_request(exc, supported=binance_derivatives.BINANCE_DERIVATIVE_PERIODS) + except BinanceProviderError as exc: + _provider_error(exc) + + +@router.get("/funding-rate/{symbol}") +async def get_funding_rate( + symbol: str, + limit: int = Query(100, ge=1, le=1000), + start_time: int | None = None, + end_time: int | None = None, +): + try: + return binance_derivatives.fetch_funding_rate(symbol, limit, start_time, end_time) + except ValueError as exc: + _bad_request(exc) + except BinanceProviderError as exc: + _provider_error(exc) + + +@router.get("/basis/{pair}") +async def get_basis( + pair: str, + contract_type: str = Query("CURRENT_QUARTER"), + period: str = Query("1d"), + limit: int = Query(30, ge=1, le=500), + start_time: int | None = None, + end_time: int | None = None, +): + try: + return binance_derivatives.fetch_basis( + pair, + contract_type, + period, + limit, + start_time, + end_time, + ) + except ValueError as exc: + _bad_request(exc) + except BinanceProviderError as exc: + _provider_error(exc) + + +@router.post("/basis-bundle") +async def post_basis_bundle(body: dict[str, Any] = Body(...)): + try: + perp_symbol = str(body["perp_symbol"]) + delivery_symbol = str(body["delivery_symbol"]) + except KeyError as exc: + raise HTTPException(status_code=400, detail=f"missing_required_field:{exc.args[0]}") + + try: + return binance_derivatives.fetch_basis_bundle( + perp_symbol=perp_symbol, + delivery_symbol=delivery_symbol, + pair=body.get("pair"), + interval=str(body.get("interval") or "1d"), + period=str(body.get("period") or "1d"), + limit=int(body.get("limit") or 30), + include_depth=bool(body.get("include_depth", True)), + depth_limit=int(body.get("depth_limit") or 5), + contract_type=str(body.get("contract_type") or "CURRENT_QUARTER"), + start_time=body.get("start_time"), + end_time=body.get("end_time"), + ) + except ValueError as exc: + _bad_request(exc) diff --git a/app/main.py b/app/main.py index 29a5f11..e3bedd2 100644 --- a/app/main.py +++ b/app/main.py @@ -32,7 +32,15 @@ STREAM_STRICT_FEED_HEALTH, ) from app.api.context import DataLayerContext -from app.api import routes_control_plane, routes_fallback, routes_health, routes_history, routes_latest, routes_preload +from app.api import ( + routes_binance_derivatives, + routes_control_plane, + routes_fallback, + routes_health, + routes_history, + routes_latest, + routes_preload, +) from app.cache.redis_cache import RedisCache from app.stream.async_live_feed import start_stream from app.ingestion.supervisor import StreamSupervisor @@ -352,6 +360,7 @@ async def lifespan(app: FastAPI): app.include_router(routes_health.router) app.include_router(routes_latest.router) app.include_router(routes_history.router) +app.include_router(routes_binance_derivatives.router) app.include_router(routes_preload.router) app.include_router(routes_control_plane.router) app.include_router(routes_fallback.router) diff --git a/app/providers/binance/derivatives.py b/app/providers/binance/derivatives.py new file mode 100644 index 0000000..e03faf3 --- /dev/null +++ b/app/providers/binance/derivatives.py @@ -0,0 +1,353 @@ +from __future__ import annotations + +import os +import random +import time +from typing import Any + +import requests + +from app.providers.binance.rest import BINANCE_KLINE_INTERVALS, BinanceProviderError, normalize_interval + + +BINANCE_FUTURES_BASE_URL = "https://fapi.binance.com" + +BINANCE_DERIVATIVE_PERIODS = { + "5m", + "15m", + "30m", + "1h", + "2h", + "4h", + "6h", + "12h", + "1d", +} + +BINANCE_DERIVATIVE_ENDPOINTS = { + "exchange_info": "/fapi/v1/exchangeInfo", + "klines": "/fapi/v1/klines", + "depth": "/fapi/v1/depth", + "funding_rate": "/fapi/v1/fundingRate", + "open_interest": "/fapi/v1/openInterest", + "open_interest_hist": "/futures/data/openInterestHist", + "global_long_short_account_ratio": "/futures/data/globalLongShortAccountRatio", + "top_long_short_account_ratio": "/futures/data/topLongShortAccountRatio", + "top_long_short_position_ratio": "/futures/data/topLongShortPositionRatio", + "taker_long_short_ratio": "/futures/data/takerlongshortRatio", + "basis": "/futures/data/basis", +} + +LONG_SHORT_KIND_TO_ENDPOINT = { + "global_account": "global_long_short_account_ratio", + "top_account": "top_long_short_account_ratio", + "top_position": "top_long_short_position_ratio", +} + + +def normalize_period(period: str) -> str: + value = str(period or "").strip() + if value not in BINANCE_DERIVATIVE_PERIODS: + raise ValueError(f"Unsupported Binance derivatives period: {period}") + return value + + +def normalize_contract_type(contract_type: str) -> str: + value = str(contract_type or "").upper().strip() + supported = {"PERPETUAL", "CURRENT_QUARTER", "NEXT_QUARTER"} + if value not in supported: + raise ValueError(f"Unsupported Binance futures contract_type: {contract_type}") + return value + + +def normalize_depth_limit(limit: int) -> int: + value = int(limit) + supported = {5, 10, 20, 50, 100, 500, 1000} + if value not in supported: + raise ValueError(f"Unsupported Binance depth limit: {limit}; supported={sorted(supported)}") + return value + + +def _clean_params(params: dict[str, Any]) -> dict[str, Any]: + return {key: value for key, value in params.items() if value is not None} + + +def _headers(endpoint_key: str) -> dict[str, str]: + api_key = os.getenv("BINANCE_MARKET_DATA_API_KEY", "").strip() + if not api_key: + return {} + return {"X-MBX-APIKEY": api_key} + + +def _public_get( + endpoint_key: str, + params: dict[str, Any] | None = None, + *, + timeout: float = 10.0, + max_attempts: int = 4, + backoff_seconds: float = 0.25, + http_get=requests.get, +) -> dict[str, Any]: + if endpoint_key not in BINANCE_DERIVATIVE_ENDPOINTS: + raise ValueError(f"Unsupported Binance derivatives endpoint: {endpoint_key}") + url = f"{BINANCE_FUTURES_BASE_URL}{BINANCE_DERIVATIVE_ENDPOINTS[endpoint_key]}" + request_params = _clean_params(params or {}) + attempts: list[dict[str, Any]] = [] + + for attempt in range(1, max_attempts + 1): + try: + resp = http_get(url, params=request_params, headers=_headers(endpoint_key), timeout=timeout) + status_code = int(resp.status_code) + if status_code == 200: + return { + "endpoint": endpoint_key, + "url_path": BINANCE_DERIVATIVE_ENDPOINTS[endpoint_key], + "params": request_params, + "data": resp.json(), + "attempts": attempts + [{"attempt": attempt, "status_code": status_code}], + } + body = getattr(resp, "text", "")[:300] + attempts.append({"attempt": attempt, "status_code": status_code, "body": body}) + if status_code not in {408, 418, 429, 500, 502, 503, 504}: + break + except Exception as exc: + attempts.append({"attempt": attempt, "error": str(exc)}) + + if attempt < max_attempts: + jitter = random.uniform(0, backoff_seconds) + time.sleep(backoff_seconds * (2 ** (attempt - 1)) + jitter) + + raise BinanceProviderError( + f"Failed to fetch Binance derivatives endpoint {endpoint_key}", + attempts=attempts, + ) + + +def _wrap(endpoint_key: str, payload: dict[str, Any], *, symbol: str | None = None, pair: str | None = None) -> dict[str, Any]: + return { + "provider": "binance", + "market": "usdm_futures", + "endpoint": endpoint_key, + "symbol": symbol.upper().strip() if symbol else None, + "pair": pair.upper().strip() if pair else None, + "params": payload["params"], + "data": payload["data"], + "cached": False, + "stored": False, + "attempts": payload.get("attempts", []), + } + + +def fetch_exchange_info(symbol: str | None = None, **kwargs) -> dict[str, Any]: + params = {"symbol": symbol.upper().strip() if symbol else None} + return _wrap("exchange_info", _public_get("exchange_info", params, **kwargs), symbol=symbol) + + +def fetch_klines( + symbol: str, + interval: str = "1d", + limit: int = 30, + start_time: int | None = None, + end_time: int | None = None, + **kwargs, +) -> dict[str, Any]: + provider_interval = normalize_interval(interval) + if int(limit) < 1 or int(limit) > 1500: + raise ValueError("Binance derivatives kline limit must be between 1 and 1500") + params = { + "symbol": symbol.upper().strip(), + "interval": provider_interval, + "limit": int(limit), + "startTime": start_time, + "endTime": end_time, + } + payload = _public_get("klines", params, **kwargs) + wrapped = _wrap("klines", payload, symbol=symbol) + wrapped["requested_interval"] = provider_interval + wrapped["provider_interval"] = provider_interval + return wrapped + + +def fetch_depth(symbol: str, limit: int = 5, **kwargs) -> dict[str, Any]: + params = {"symbol": symbol.upper().strip(), "limit": normalize_depth_limit(limit)} + return _wrap("depth", _public_get("depth", params, **kwargs), symbol=symbol) + + +def fetch_open_interest(symbol: str, **kwargs) -> dict[str, Any]: + params = {"symbol": symbol.upper().strip()} + return _wrap("open_interest", _public_get("open_interest", params, **kwargs), symbol=symbol) + + +def fetch_metric_history( + endpoint_key: str, + symbol: str, + period: str = "1d", + limit: int = 30, + start_time: int | None = None, + end_time: int | None = None, + **kwargs, +) -> dict[str, Any]: + if endpoint_key not in { + "open_interest_hist", + "global_long_short_account_ratio", + "top_long_short_account_ratio", + "top_long_short_position_ratio", + "taker_long_short_ratio", + }: + raise ValueError(f"Unsupported Binance metric history endpoint: {endpoint_key}") + provider_period = normalize_period(period) + if int(limit) < 1 or int(limit) > 500: + raise ValueError("Binance derivatives metric limit must be between 1 and 500") + params = { + "symbol": symbol.upper().strip(), + "period": provider_period, + "limit": int(limit), + "startTime": start_time, + "endTime": end_time, + } + payload = _public_get(endpoint_key, params, **kwargs) + wrapped = _wrap(endpoint_key, payload, symbol=symbol) + wrapped["requested_period"] = provider_period + wrapped["latest_30_days_only"] = True + return wrapped + + +def fetch_long_short_ratio( + kind: str, + symbol: str, + period: str = "1d", + limit: int = 30, + start_time: int | None = None, + end_time: int | None = None, + **kwargs, +) -> dict[str, Any]: + normalized_kind = str(kind or "").lower().strip() + if normalized_kind not in LONG_SHORT_KIND_TO_ENDPOINT: + raise ValueError(f"Unsupported long/short ratio kind: {kind}") + return fetch_metric_history( + LONG_SHORT_KIND_TO_ENDPOINT[normalized_kind], + symbol, + period, + limit, + start_time, + end_time, + **kwargs, + ) + + +def fetch_taker_long_short_ratio( + symbol: str, + period: str = "1d", + limit: int = 30, + start_time: int | None = None, + end_time: int | None = None, + **kwargs, +) -> dict[str, Any]: + return fetch_metric_history("taker_long_short_ratio", symbol, period, limit, start_time, end_time, **kwargs) + + +def fetch_funding_rate( + symbol: str, + limit: int = 100, + start_time: int | None = None, + end_time: int | None = None, + **kwargs, +) -> dict[str, Any]: + if int(limit) < 1 or int(limit) > 1000: + raise ValueError("Binance funding rate limit must be between 1 and 1000") + params = { + "symbol": symbol.upper().strip(), + "limit": int(limit), + "startTime": start_time, + "endTime": end_time, + } + return _wrap("funding_rate", _public_get("funding_rate", params, **kwargs), symbol=symbol) + + +def fetch_basis( + pair: str, + contract_type: str = "CURRENT_QUARTER", + period: str = "1d", + limit: int = 30, + start_time: int | None = None, + end_time: int | None = None, + **kwargs, +) -> dict[str, Any]: + provider_period = normalize_period(period) + normalized_contract_type = normalize_contract_type(contract_type) + if int(limit) < 1 or int(limit) > 500: + raise ValueError("Binance basis limit must be between 1 and 500") + params = { + "pair": pair.upper().strip(), + "contractType": normalized_contract_type, + "period": provider_period, + "limit": int(limit), + "startTime": start_time, + "endTime": end_time, + } + payload = _public_get("basis", params, **kwargs) + wrapped = _wrap("basis", payload, pair=pair) + wrapped["requested_period"] = provider_period + wrapped["contract_type"] = normalized_contract_type + wrapped["latest_30_days_only"] = True + return wrapped + + +def fetch_basis_bundle( + perp_symbol: str, + delivery_symbol: str, + pair: str | None = None, + interval: str = "1d", + period: str = "1d", + limit: int = 30, + include_depth: bool = True, + depth_limit: int = 5, + contract_type: str = "CURRENT_QUARTER", + start_time: int | None = None, + end_time: int | None = None, + **kwargs, +) -> dict[str, Any]: + normalized_perp = perp_symbol.upper().strip() + normalized_delivery = delivery_symbol.upper().strip() + normalized_pair = (pair or normalized_perp).upper().strip() + + bundle: dict[str, Any] = { + "provider": "binance", + "market": "usdm_futures", + "kind": "basis_bundle", + "perp_symbol": normalized_perp, + "delivery_symbol": normalized_delivery, + "pair": normalized_pair, + "requested_interval": normalize_interval(interval), + "requested_period": normalize_period(period), + "limit": int(limit), + "cached": False, + "stored": False, + "components": {}, + "errors": {}, + } + fetches = { + "perp_klines": lambda: fetch_klines(normalized_perp, interval, limit, start_time, end_time, **kwargs), + "delivery_klines": lambda: fetch_klines(normalized_delivery, interval, limit, start_time, end_time, **kwargs), + "funding_rate": lambda: fetch_funding_rate(normalized_perp, min(int(limit), 1000), start_time, end_time, **kwargs), + "open_interest_hist": lambda: fetch_metric_history("open_interest_hist", normalized_perp, period, limit, start_time, end_time, **kwargs), + "global_long_short": lambda: fetch_long_short_ratio("global_account", normalized_perp, period, limit, start_time, end_time, **kwargs), + "top_account_long_short": lambda: fetch_long_short_ratio("top_account", normalized_perp, period, limit, start_time, end_time, **kwargs), + "top_position_long_short": lambda: fetch_long_short_ratio("top_position", normalized_perp, period, limit, start_time, end_time, **kwargs), + "taker_long_short": lambda: fetch_taker_long_short_ratio(normalized_perp, period, limit, start_time, end_time, **kwargs), + "basis": lambda: fetch_basis(normalized_pair, contract_type, period, limit, start_time, end_time, **kwargs), + } + if include_depth: + fetches["perp_depth"] = lambda: fetch_depth(normalized_perp, depth_limit, **kwargs) + fetches["delivery_depth"] = lambda: fetch_depth(normalized_delivery, depth_limit, **kwargs) + + for name, fetcher in fetches.items(): + try: + bundle["components"][name] = fetcher() + except Exception as exc: + bundle["errors"][name] = {"error": str(exc), "attempts": getattr(exc, "attempts", [])} + + bundle["success_count"] = len(bundle["components"]) + bundle["error_count"] = len(bundle["errors"]) + bundle["partial"] = bool(bundle["errors"]) + return bundle diff --git a/app/sdk/client.py b/app/sdk/client.py index bd9ad28..a6061e5 100644 --- a/app/sdk/client.py +++ b/app/sdk/client.py @@ -78,6 +78,10 @@ def _symbols(symbols: str | Iterable[str]) -> list[str]: return [symbols.upper().strip()] return [str(symbol).upper().strip() for symbol in symbols if str(symbol).strip()] + @staticmethod + def _params(params: dict[str, Any]) -> dict[str, Any]: + return {key: value for key, value in params.items() if value is not None} + def health(self) -> dict: return self._get("/v1/health") @@ -213,6 +217,115 @@ def fallback_reference( }, ) + def binance_futures_klines( + self, + symbol: str, + interval: str = "1d", + limit: int = 30, + *, + start_time: int | None = None, + end_time: int | None = None, + ) -> dict: + return self._get( + f"/v1/binance/futures/klines/{symbol.upper().strip()}", + params=self._params({ + "interval": interval, + "limit": limit, + "start_time": start_time, + "end_time": end_time, + }), + ) + + def binance_futures_metric( + self, + metric: str, + symbol: str, + *, + kind: str | None = None, + period: str = "1d", + limit: int = 30, + start_time: int | None = None, + end_time: int | None = None, + ) -> dict: + normalized = metric.lower().strip().replace("_", "-") + symbol = symbol.upper().strip() + params = self._params({ + "period": period, + "limit": limit, + "start_time": start_time, + "end_time": end_time, + }) + if normalized == "open-interest-history": + return self._get(f"/v1/binance/futures/open-interest-history/{symbol}", params=params) + if normalized == "taker-long-short": + return self._get(f"/v1/binance/futures/taker-long-short/{symbol}", params=params) + if normalized == "long-short": + return self._get(f"/v1/binance/futures/long-short/{kind or 'global_account'}/{symbol}", params=params) + if normalized == "funding-rate": + return self._get( + f"/v1/binance/futures/funding-rate/{symbol}", + params=self._params({"limit": limit, "start_time": start_time, "end_time": end_time}), + ) + if normalized == "open-interest": + return self._get(f"/v1/binance/futures/open-interest/{symbol}") + raise ValueError(f"Unsupported Binance futures metric: {metric}") + + def binance_futures_depth(self, symbol: str, limit: int = 5) -> dict: + return self._get(f"/v1/binance/futures/depth/{symbol.upper().strip()}", params={"limit": limit}) + + def binance_futures_basis( + self, + pair: str, + *, + contract_type: str = "CURRENT_QUARTER", + period: str = "1d", + limit: int = 30, + start_time: int | None = None, + end_time: int | None = None, + ) -> dict: + return self._get( + f"/v1/binance/futures/basis/{pair.upper().strip()}", + params=self._params({ + "contract_type": contract_type, + "period": period, + "limit": limit, + "start_time": start_time, + "end_time": end_time, + }), + ) + + def binance_basis_bundle( + self, + perp_symbol: str, + delivery_symbol: str, + *, + pair: str | None = None, + interval: str = "1d", + period: str = "1d", + limit: int = 30, + include_depth: bool = True, + depth_limit: int = 5, + contract_type: str = "CURRENT_QUARTER", + start_time: int | None = None, + end_time: int | None = None, + ) -> dict: + return self._post( + "/v1/binance/futures/basis-bundle", + self._params({ + "perp_symbol": perp_symbol.upper().strip(), + "delivery_symbol": delivery_symbol.upper().strip(), + "pair": pair.upper().strip() if pair else None, + "interval": interval, + "period": period, + "limit": limit, + "include_depth": include_depth, + "depth_limit": depth_limit, + "contract_type": contract_type, + "start_time": start_time, + "end_time": end_time, + }), + ) + def control_contracts(self) -> dict: return { "redis_channels": { @@ -226,6 +339,7 @@ def control_contracts(self) -> dict: "kline": "/v1/binance/kline/{symbol}?interval=1m", "vn_preload": "/v1/preload/{symbol}?interval=1m&limit=1000", "vn_last_quote": "/v1/vn/quote-last/{symbol}", + "basis_bundle": "/v1/binance/futures/basis-bundle", }, "provider_policy": { "binance": "authoritative for Binance crypto live trade/kline data", diff --git a/tests/test_binance_derivatives_contract.py b/tests/test_binance_derivatives_contract.py new file mode 100644 index 0000000..a4979fd --- /dev/null +++ b/tests/test_binance_derivatives_contract.py @@ -0,0 +1,140 @@ +import asyncio +import unittest +from unittest.mock import MagicMock, patch + +from app.api import routes_binance_derivatives +from app.providers.binance import derivatives +from app.providers.binance.rest import BinanceProviderError +from app.sdk.client import DataLayerClient + + +class FakeResponse: + def __init__(self, status_code=200, payload=None, text=""): + self.status_code = status_code + self._payload = payload if payload is not None else [] + self.text = text + + def json(self): + return self._payload + + +class TestBinanceDerivativesContract(unittest.TestCase): + def test_period_validation(self): + self.assertEqual(derivatives.normalize_period("1d"), "1d") + with self.assertRaises(ValueError): + derivatives.normalize_period("7m") + + def test_fetch_metric_history_uses_official_endpoint_and_normalized_payload(self): + def fake_get(url, params=None, headers=None, timeout=None): + self.assertTrue(url.endswith("/futures/data/openInterestHist")) + self.assertEqual(params["symbol"], "BTCUSDT") + self.assertEqual(params["period"], "1d") + self.assertEqual(params["limit"], 30) + return FakeResponse(payload=[{"symbol": "BTCUSDT", "sumOpenInterest": "1", "timestamp": "1"}]) + + payload = derivatives.fetch_metric_history( + "open_interest_hist", + "btcusdt", + "1d", + 30, + http_get=fake_get, + max_attempts=1, + ) + + self.assertEqual(payload["provider"], "binance") + self.assertEqual(payload["endpoint"], "open_interest_hist") + self.assertEqual(payload["symbol"], "BTCUSDT") + self.assertFalse(payload["stored"]) + self.assertTrue(payload["latest_30_days_only"]) + + def test_retry_then_success_on_retryable_status(self): + calls = {"count": 0} + + def fake_get(url, params=None, headers=None, timeout=None): + calls["count"] += 1 + if calls["count"] == 1: + return FakeResponse(status_code=502, text="bad gateway") + return FakeResponse(payload={"symbol": "BTCUSDT", "openInterest": "10", "time": 1}) + + payload = derivatives.fetch_open_interest( + "BTCUSDT", + http_get=fake_get, + max_attempts=2, + backoff_seconds=0, + ) + + self.assertEqual(calls["count"], 2) + self.assertEqual(payload["data"]["openInterest"], "10") + + def test_non_retryable_status_raises_provider_error(self): + def fake_get(url, params=None, headers=None, timeout=None): + return FakeResponse(status_code=400, text="bad request") + + with self.assertRaises(BinanceProviderError) as ctx: + derivatives.fetch_depth("BTCUSDT", http_get=fake_get, max_attempts=3) + self.assertEqual(len(ctx.exception.attempts), 1) + + def test_basis_bundle_is_partial_when_component_fails(self): + def fake_get(url, params=None, headers=None, timeout=None): + if url.endswith("/fapi/v1/depth") and params["symbol"] == "BTCUSDT_260925": + return FakeResponse(status_code=502, text="delivery depth unavailable") + return FakeResponse(payload=[{"ok": True, "params": params}]) + + payload = derivatives.fetch_basis_bundle( + "BTCUSDT", + "BTCUSDT_260925", + pair="BTCUSDT", + include_depth=True, + http_get=fake_get, + max_attempts=1, + ) + + self.assertTrue(payload["partial"]) + self.assertIn("delivery_depth", payload["errors"]) + self.assertIn("perp_klines", payload["components"]) + self.assertEqual(payload["delivery_symbol"], "BTCUSDT_260925") + + def test_route_basis_bundle_requires_symbols(self): + with self.assertRaises(Exception) as ctx: + asyncio.run(routes_binance_derivatives.post_basis_bundle({"perp_symbol": "BTCUSDT"})) + self.assertIn("missing_required_field", str(ctx.exception.detail)) + + def test_sdk_basis_bundle_posts_contract_payload(self): + session = MagicMock() + redis_client = MagicMock() + response = MagicMock(status_code=200, text="") + response.json.return_value = {"kind": "basis_bundle", "components": {}} + response.raise_for_status.return_value = None + session.post.return_value = response + client = DataLayerClient(session=session, redis_client=redis_client) + + payload = client.binance_basis_bundle( + "btcusdt", + "btcusdt_260925", + pair="btcusdt", + include_depth=False, + ) + + self.assertEqual(payload["kind"], "basis_bundle") + session.post.assert_called_once() + _, kwargs = session.post.call_args + self.assertEqual(kwargs["json"]["perp_symbol"], "BTCUSDT") + self.assertEqual(kwargs["json"]["delivery_symbol"], "BTCUSDT_260925") + self.assertFalse(kwargs["json"]["include_depth"]) + + def test_route_klines_delegates_to_provider(self): + with patch("app.api.routes_binance_derivatives.binance_derivatives.fetch_klines") as fetch: + fetch.return_value = {"endpoint": "klines", "symbol": "BTCUSDT_260925"} + payload = asyncio.run( + routes_binance_derivatives.get_derivative_klines( + "BTCUSDT_260925", + interval="1d", + limit=30, + ) + ) + self.assertEqual(payload["symbol"], "BTCUSDT_260925") + fetch.assert_called_once_with("BTCUSDT_260925", "1d", 30, None, None) + + +if __name__ == "__main__": + unittest.main() From 0edd30ffe7416ddba46f47df06c02acd50eb8b8c Mon Sep 17 00:00:00 2001 From: vcmbob Date: Wed, 8 Jul 2026 06:33:35 +0000 Subject: [PATCH 2/4] Add data_layer contribution workflow docs --- .github/pull_request_template.md | 26 ++++++++++++ .github/workflows/ci.yml | 23 +++++++++++ .pre-commit-config.yaml | 19 +++++++++ CODE_OF_CONDUCT.md | 22 +++++++++++ CONTRIBUTING.md | 68 ++++++++++++++++++++++++++++++++ README.md | 65 +++++++++++++++++++++++++++++- SECURITY.md | 27 +++++++++++++ 7 files changed, 249 insertions(+), 1 deletion(-) create mode 100644 .github/pull_request_template.md create mode 100644 .github/workflows/ci.yml create mode 100644 .pre-commit-config.yaml create mode 100644 CODE_OF_CONDUCT.md create mode 100644 CONTRIBUTING.md create mode 100644 SECURITY.md diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md new file mode 100644 index 0000000..9c6f89a --- /dev/null +++ b/.github/pull_request_template.md @@ -0,0 +1,26 @@ +## Summary + +- + +## Type + +- [ ] Data contract / API +- [ ] Provider integration +- [ ] Streaming / Redis +- [ ] VN preload / storage +- [ ] Diagnostics / monitoring +- [ ] Docs / repo hygiene + +## Checks + +- [ ] I did not commit directly to `main`. +- [ ] Docs are updated for public behavior changes. +- [ ] Tests are added or updated. +- [ ] Docker tests pass, or a blocker is documented. +- [ ] No secrets, generated logs, parquet data, or local caches are included. + +## Test Evidence + +```text + +``` diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..0de3844 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,23 @@ +name: CI + +on: + pull_request: + branches: ["dev", "main"] + push: + branches: ["dev"] + +jobs: + unit-tests: + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v3 + + - name: Build data_layer image + run: docker compose build data_layer + + - name: Run unit tests + run: docker compose run --rm test_runner python -m unittest discover -s tests diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..af35456 --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,19 @@ +repos: + - repo: https://github.com/pre-commit/pre-commit-hooks + rev: v4.6.0 + hooks: + - id: trailing-whitespace + - id: end-of-file-fixer + - id: check-added-large-files + args: ["--maxkb=1024"] + - id: check-case-conflict + - id: check-merge-conflict + - id: check-yaml + - id: mixed-line-ending + args: ["--fix=lf"] + + - repo: https://github.com/astral-sh/ruff-pre-commit + rev: v0.5.7 + hooks: + - id: ruff + args: ["--fix", "--exit-non-zero-on-fix"] diff --git a/CODE_OF_CONDUCT.md b/CODE_OF_CONDUCT.md new file mode 100644 index 0000000..c733aeb --- /dev/null +++ b/CODE_OF_CONDUCT.md @@ -0,0 +1,22 @@ +# Code of Conduct + +`data_layer` is maintained as a professional engineering project. Contributors are expected to keep discussions respectful, technical, and focused on improving reliability for downstream users. + +## Expected Behavior + +- Be clear, patient, and constructive. +- Discuss code, design, and operational risk directly without personal attacks. +- Assume good intent, but verify technical claims with tests, logs, docs, or reproducible examples. +- Respect security boundaries and never request or expose credentials. +- Keep issue and pull-request threads on topic. + +## Unacceptable Behavior + +- Harassment, threats, or discriminatory language. +- Publishing private information, credentials, logs with secrets, or production account details. +- Repeated off-topic comments that block maintainers or contributors from resolving work. +- Intentionally submitting malicious code or misleading test results. + +## Enforcement + +Maintainers may edit or remove comments, close issues, block users, or reject contributions that violate this code. Security-sensitive incidents should be reported privately using the process in `SECURITY.md`. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..c4f9a4e --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,68 @@ +# Contributing + +Thanks for helping improve `data_layer`. This service is used as a market-data gateway by execution and alpha services, so changes should keep API contracts stable and observable. + +## Branch Flow + +- Do not commit directly to `main`. +- Work on `dev` or a feature branch from `dev`. +- Open pull requests into `dev` first. +- Merge `dev` into `main` only through a release pull request after tests and smoke checks pass. +- Keep commits focused and use messages that name the subsystem and behavior changed. + +Recommended flow: + +```bash +git checkout dev +git pull origin dev +git checkout -b feature/binance-derivatives-contract +``` + +## Local Checks + +This server is Docker-first. Prefer container tests over installing Python packages on the host: + +```bash +docker compose run --rm test_runner python -m unittest discover -s tests +``` + +For changed Python files, a quick compile check is useful: + +```bash +python3 -m py_compile app/path/to_file.py +``` + +## Pre-Commit + +Install hooks in a development environment: + +```bash +pre-commit install +pre-commit run --all-files +``` + +The hook set checks whitespace, YAML, merge conflicts, large files, and Ruff linting. + +## API Contract Rules + +- Keep existing response shapes stable unless a migration plan is documented. +- Add new endpoints instead of mutating old endpoint contracts when downstream services already depend on them. +- Put provider-specific raw fields under explicit payload sections and include metadata such as provider, market, params, cached, and stored. +- Do not let alpha containers call external providers directly. Add or extend `data_layer` wrappers instead. +- Do not store ephemeral crypto derivatives metrics unless a design note explicitly approves it. + +## Documentation + +Update these files when changing public behavior: + +- `README.md` for project-level capabilities and quickstart. +- `DATA_LAYER_SERVICE_ACCESS_GUIDE.md` for downstream service contracts. +- Tests under `tests/` for endpoint and SDK behavior. + +## Pull Request Checklist + +- [ ] I did not commit directly to `main`. +- [ ] I updated docs for public API or operational changes. +- [ ] I added/updated tests for the changed contract. +- [ ] Docker unit tests pass or the reason is documented. +- [ ] No secrets, credentials, generated logs, parquet data, or local caches are included. diff --git a/README.md b/README.md index 60bc902..f3812a3 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,53 @@ # data_layer -A centralized, high-performance market data gateway service that aggregates and distributes real-time financial market data from multiple sources — **Binance** (crypto) and **Vietnamese stock market** (via DNSE WebSocket + vnstock fallback) — to downstream trading services via **Redis Pub/Sub** and a **REST API**. +`data_layer` is the market-data gateway for Bobby's trading stack. It centralizes provider connectivity, historical warmup, live streaming, fallback recovery, and service-to-service data contracts so alpha and execution services do not connect to exchanges directly. + +It currently serves: + +- Binance crypto spot and USD-M futures market data. +- Binance derivatives metrics for basis-arbitrage research and execution. +- VN stock and derivative data through DNSE/vnstock integrations. +- Redis Pub/Sub streams for live consumers. +- REST endpoints for warmup, recovery, diagnostics, and health checks. + +## Quick Links + +- [Integration guide](./DATA_LAYER_SERVICE_ACCESS_GUIDE.md) +- [Contributing guide](./CONTRIBUTING.md) +- [Security policy](./SECURITY.md) +- [Code of conduct](./CODE_OF_CONDUCT.md) +- [License](./LICENSE) + +## Repository Policy + +`main` is treated as the protected release branch. Do not commit directly to `main`. + +Use this flow: + +```bash +git checkout dev +git pull origin dev +git checkout -b feature/my-change +``` + +Open pull requests into `dev`; merge `dev` into `main` only through a release pull request after tests and smoke checks pass. + +## What This Service Owns + +- Provider connections and retry/backoff policy. +- Normalized service contracts for downstream systems. +- Redis live-stream publication. +- VN preload parquet storage and materialized views. +- Latest-state recovery endpoints. +- Diagnostics for provider health and data freshness. + +## What This Service Does Not Own + +- Trading decisions. +- Portfolio, risk, order routing, or broker account state. +- Alpha-specific signal logic. +- Direct broker execution. +- Long-term storage for ephemeral Binance derivatives metrics unless a design note explicitly approves it. ## Features @@ -8,6 +55,7 @@ A centralized, high-performance market data gateway service that aggregates and - **Automatic failover** — DNSE as primary VN source, vnstock REST poller as secondary fallback - **Redis Pub/Sub distribution** — single upstream connection shared across many downstream consumers - **Historical warmup (VN)** — Parquet-backed preload service for 1-minute OHLCV candle warmup +- **Binance derivatives REST wrappers** — OHLCV, funding, open interest, long/short ratios, taker ratio, depth, and basis bundle endpoints - **Preload watchdog** — auto-refreshes VN candle data during market hours, sleeps until next open otherwise - ⚡ **High-performance serialization** — `orjson` throughout for minimal latency - **Alpha strategy example** — moving-average crossover strategy included as a reference implementation @@ -50,6 +98,21 @@ Downstream services (alpha strategies, paper trading engines, execution services | Runtime | Python 3.10+ | | Container | Docker + Docker Compose | +## Development Checks + +This repo is Docker-first. Prefer container tests over installing Python packages directly on the server: + +```bash +docker compose run --rm test_runner python -m unittest discover -s tests +``` + +Optional pre-commit hooks: + +```bash +pre-commit install +pre-commit run --all-files +``` + ## Project Structure ``` diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 0000000..68ecb8e --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,27 @@ +# Security Policy + +`data_layer` handles market-data connectivity and service-to-service data distribution. It should never contain trading credentials, broker secrets, personal API keys, or production account identifiers in source control. + +## Reporting a Vulnerability + +Please report security issues privately to the maintainers. Do not open a public issue with exploit details, credentials, or production logs. + +Include: + +- affected component or endpoint; +- reproduction steps; +- expected impact; +- relevant logs with secrets removed; +- suggested mitigation, if known. + +## Supported Security Practices + +- Keep `.env` local and out of git. +- Use Docker networks for internal services. +- Do not expose Redis or provider credentials publicly. +- Prefer API wrappers in `data_layer` over direct provider calls from alpha containers. +- Redact API keys, tokens, account IDs, and investor identifiers before sharing logs. + +## Dependency Updates + +Dependency and Docker image updates should be tested in `dev` before merge to `main`. From 1ee1b798c9d01fdf8bd3b3fe65bbcb20a8138231 Mon Sep 17 00:00:00 2001 From: BobbyAxerol Date: Wed, 8 Jul 2026 06:35:13 +0000 Subject: [PATCH 3/4] Map contributors to BobbyAxerol identity --- .mailmap | 2 ++ 1 file changed, 2 insertions(+) create mode 100644 .mailmap diff --git a/.mailmap b/.mailmap new file mode 100644 index 0000000..ac670ad --- /dev/null +++ b/.mailmap @@ -0,0 +1,2 @@ +BobbyAxerol vcmbob +BobbyAxerol root From 196e94fe60f2727b2540bad199460dde9e99125e Mon Sep 17 00:00:00 2001 From: BobbyAxerol Date: Wed, 8 Jul 2026 06:37:54 +0000 Subject: [PATCH 4/4] Keep mailmap local for data_layer contributors --- .gitignore | 1 + .mailmap | 2 -- 2 files changed, 1 insertion(+), 2 deletions(-) delete mode 100644 .mailmap diff --git a/.gitignore b/.gitignore index 4c5aef0..49c0a6e 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ .env +.mailmap .env.local .env.*.local __pycache__/ diff --git a/.mailmap b/.mailmap deleted file mode 100644 index ac670ad..0000000 --- a/.mailmap +++ /dev/null @@ -1,2 +0,0 @@ -BobbyAxerol vcmbob -BobbyAxerol root