diff --git a/README.md b/README.md index 685d669aa..b961c148b 100644 --- a/README.md +++ b/README.md @@ -271,6 +271,13 @@ PostgreSQL migration guidance, see [Installation troubleshooting](docs/deployment/INSTALL_TROUBLESHOOTING.md) and the [cloud deployment guide](docs/deployment/CLOUD_DEPLOYMENT_EN.md). +### Optional Qveris data source + +QuantDinger can use Qveris as an opt-in unified market-data layer while keeping +the existing market providers as automatic fallbacks. See the +[Qveris data-source guide](docs/integrations/QVERIS_DATA_SOURCE.md) for setup, +supported markets, safety behavior, and verification. + ## Production deployment Validate secrets before starting a production stack: diff --git a/backend_api_python/app/data_sources/factory.py b/backend_api_python/app/data_sources/factory.py index 38903b613..206a85695 100644 --- a/backend_api_python/app/data_sources/factory.py +++ b/backend_api_python/app/data_sources/factory.py @@ -187,27 +187,38 @@ def _create_source(cls, market: str) -> BaseDataSource: """创建数据源实例""" if market == 'Crypto': from app.data_sources.crypto import CryptoDataSource - return CryptoDataSource() + source = CryptoDataSource() elif market == 'CNStock': from app.data_sources.cn_stock import CNStockDataSource - return CNStockDataSource() + source = CNStockDataSource() elif market == 'HKStock': from app.data_sources.hk_stock import HKStockDataSource - return HKStockDataSource() + source = HKStockDataSource() elif market == 'USStock': from app.data_sources.us_stock import USStockDataSource - return USStockDataSource() + source = USStockDataSource() elif market == 'Forex': from app.data_sources.forex import ForexDataSource - return ForexDataSource() + source = ForexDataSource() elif market == 'Futures': from app.data_sources.futures import FuturesDataSource - return FuturesDataSource() + source = FuturesDataSource() elif market == 'MOEX': from app.data_sources.moex import MOEXDataSource - return MOEXDataSource() + source = MOEXDataSource() else: raise UnsupportedMarketError(market) + return cls._wrap_optional_source(market, source) + + @staticmethod + def _wrap_optional_source(market: str, source: BaseDataSource) -> BaseDataSource: + """Wrap an existing source with Qveris only when explicitly enabled.""" + from app.data_sources.qveris import QverisDataSource + + if QverisDataSource.is_enabled_for(market): + logger.info("Qveris data source enabled for %s with %s fallback", market, source.name) + return QverisDataSource(market, source) + return source @classmethod def get_kline( diff --git a/backend_api_python/app/data_sources/qveris.py b/backend_api_python/app/data_sources/qveris.py new file mode 100644 index 000000000..f9270bb76 --- /dev/null +++ b/backend_api_python/app/data_sources/qveris.py @@ -0,0 +1,576 @@ +"""Optional Qveris market-data adapter with existing-source fallback. + +Qveris discovers a suitable upstream tool, executes it, and returns the +provider response. This adapter keeps provider-specific response shapes inside +the integration boundary and normalizes common OHLCV/quote layouts for +QuantDinger. +""" + +from __future__ import annotations + +import json +import os +import re +import time +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +import requests + +from app.data_sources.base import BaseDataSource +from app.utils.logger import get_logger + +logger = get_logger(__name__) + + +class QverisError(RuntimeError): + """Raised when Qveris discovery, execution, or normalization fails.""" + + +_PARAM_ALIASES = { + "symbol": { + "symbol", + "symbols", + "symbolexchange", + "ticker", + "tickers", + "pair", + "instrument", + "instid", + "code", + "asset", + }, + "market": {"market", "assetclass", "assettype"}, + "timeframe": {"timeframe", "interval", "resolution", "period", "granularity", "frequency"}, + "limit": {"limit", "count", "rows", "size", "outputsize", "numresults", "numberofresults"}, + "start": {"start", "startdate", "starttime", "from", "fromdate", "fromtime", "since"}, + "end": {"end", "enddate", "endtime", "to", "todate", "totime", "until", "before"}, +} + +_ROW_ALIASES = { + "time": {"time", "times", "timestamp", "timestamps", "datetime", "datetimes", "date", "dates", "opentime", "t"}, + "open": {"open", "o", "1open"}, + "high": {"high", "h", "2high"}, + "low": {"low", "l", "3low"}, + "close": {"close", "c", "price", "4close", "adjustedclose", "adjclose"}, + "volume": {"volume", "v", "5volume", "6volume"}, +} + +_QUOTE_ALIASES = { + "last": {"last", "price", "close", "currentprice", "latestprice", "regularmarketprice", "c"}, + "change": {"change", "netchange", "d"}, + "changePercent": {"changepercent", "percentchange", "percentagechange", "changepct", "dp"}, + "high": {"high", "dayhigh", "regularmarketdayhigh", "h"}, + "low": {"low", "daylow", "regularmarketdaylow", "l"}, + "open": {"open", "regularmarketopen", "o"}, + "previousClose": {"previousclose", "prevclose", "regularmarketpreviousclose", "pc"}, +} + +_TIMEFRAME_OPTIONS = { + "1m": ("1m", "1min", "1minute"), + "3m": ("3m", "3min", "3minute"), + "5m": ("5m", "5min", "5minute"), + "15m": ("15m", "15min", "15minute"), + "30m": ("30m", "30min", "30minute"), + "1H": ("1H", "1h", "60m", "hour", "hourly"), + "4H": ("4H", "4h", "240m", "4hour"), + "1D": ("1D", "1d", "d", "day", "daily"), + "1W": ("1W", "1w", "w", "week", "weekly"), +} + +_MARKET_RANK_TERMS = { + "USStock": ({"stock", "equity", "nasdaq", "nyse"}, {"crypto", "cryptocurrency", "forex"}), + "CNStock": ({"stock", "equity", "china", "a-share"}, {"crypto", "cryptocurrency", "forex"}), + "HKStock": ({"stock", "equity", "hong kong", "hk"}, {"crypto", "cryptocurrency", "forex"}), + "Crypto": ({"crypto", "cryptocurrency", "digital asset"}, {"stock", "equity", "forex"}), + "Forex": ({"forex", "foreign exchange", "currency"}, {"stock", "equity", "crypto"}), + "Futures": ({"futures", "commodity", "derivative"}, {"stock", "equity", "crypto"}), + "MOEX": ({"moex", "russia", "russian"}, {"crypto", "cryptocurrency", "forex"}), +} + + +def _key(value: Any) -> str: + return re.sub(r"[^a-z0-9]", "", str(value or "").lower()) + + +def _float(value: Any) -> Optional[float]: + if value is None or isinstance(value, bool): + return None + try: + if isinstance(value, str): + value = value.strip().replace(",", "").replace("$", "").replace("%", "") + if not value or value.lower() in {"none", "null", "n/a", "nan", "-"}: + return None + return float(value) + except (TypeError, ValueError): + return None + + +def _timestamp(value: Any) -> Optional[int]: + if value is None or isinstance(value, bool): + return None + if isinstance(value, (int, float)): + number = float(value) + if number > 10_000_000_000: + number /= 1000.0 + return int(number) if number > 0 else None + text = str(value).strip() + if not text: + return None + if re.fullmatch(r"\d+(?:\.\d+)?", text): + return _timestamp(float(text)) + try: + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return int(parsed.timestamp()) + except ValueError: + return None + + +def _field(row: Dict[str, Any], aliases: set[str]) -> Any: + normalized = {_key(name): value for name, value in row.items()} + for alias in aliases: + if alias in normalized: + return normalized[alias] + return None + + +class QverisDataSource(BaseDataSource): + """Use Qveris first and preserve the configured QuantDinger fallback.""" + + name = "Qveris" + DEFAULT_BASE_URL = "https://qveris.ai/api/v1" + + def __init__( + self, + market: str, + fallback: BaseDataSource, + *, + api_key: Optional[str] = None, + base_url: Optional[str] = None, + timeout: Optional[float] = None, + session: Optional[requests.Session] = None, + ): + self.market = str(market or "").strip() + self.fallback = fallback + self.api_key = str(api_key if api_key is not None else os.getenv("QVERIS_API_KEY", "")).strip() + self.base_url = str(base_url or os.getenv("QVERIS_BASE_URL") or self.DEFAULT_BASE_URL).rstrip("/") + self.timeout = float(timeout or os.getenv("QVERIS_TIMEOUT", "30")) + self.discovery_ttl = max(0, int(os.getenv("QVERIS_DISCOVERY_TTL_SECONDS", "3600"))) + self.session = session or requests.Session() + self.name = f"Qveris/{self.market}+{fallback.name}" + self._discovery_cache: Dict[str, tuple[float, Dict[str, Any], str]] = {} + + @classmethod + def is_enabled_for(cls, market: str) -> bool: + if not (os.getenv("QVERIS_API_KEY") or "").strip(): + return False + configured = { + value.strip().lower() + for value in (os.getenv("QVERIS_DATA_SOURCE_MARKETS") or "").split(",") + if value.strip() + } + return "*" in configured or str(market or "").strip().lower() in configured + + def get_kline( + self, + symbol: str, + timeframe: str, + limit: int, + before_time: Optional[int] = None, + after_time: Optional[int] = None, + ) -> List[Dict[str, Any]]: + try: + result = self._execute( + "kline", + symbol=symbol, + timeframe=timeframe, + limit=max(1, int(limit or 1)), + before_time=before_time, + after_time=after_time, + ) + rows = self._normalize_klines(result) + rows = self.filter_and_limit( + rows, + limit=max(1, int(limit or 1)), + before_time=before_time, + after_time=after_time, + truncate=(after_time is None), + ) + if rows: + self.log_result(symbol, rows, timeframe) + return rows + raise QverisError("Qveris returned no normalizable OHLCV rows") + except Exception as exc: + logger.warning( + "Qveris K-line request failed for %s:%s (%s); using %s", + self.market, + symbol, + str(exc)[:240], + self.fallback.name, + ) + return self.fallback.get_kline(symbol, timeframe, limit, before_time, after_time) + + def get_ticker(self, symbol: str) -> Dict[str, Any]: + try: + result = self._execute("ticker", symbol=symbol) + quote = self._normalize_ticker(result) + if quote and float(quote.get("last") or 0) > 0: + quote.setdefault("symbol", symbol) + return quote + raise QverisError("Qveris returned no normalizable quote") + except Exception as exc: + logger.warning( + "Qveris ticker request failed for %s:%s (%s); using %s", + self.market, + symbol, + str(exc)[:240], + self.fallback.name, + ) + return self.fallback.get_ticker(symbol) + + def _execute(self, operation: str, **context: Any) -> Any: + if not self.api_key: + raise QverisError("QVERIS_API_KEY is not configured") + tool, discovery_id = self._discover(operation, str(context.get("timeframe") or "")) + parameters = self._build_parameters(self._tool_parameters(tool), context) + payload = self._request_json( + "/tools/execute", + params={"tool_id": tool.get("tool_id")}, + json={ + "search_id": discovery_id, + "parameters": parameters, + "max_response_size": 262144, + }, + ) + if payload.get("success") is False: + raise QverisError(str(payload.get("error_message") or "Qveris tool execution failed")) + result = payload.get("result", payload) + if isinstance(result, str): + try: + return json.loads(result) + except ValueError: + return result + return result + + def _discover(self, operation: str, timeframe: str) -> tuple[Dict[str, Any], str]: + cache_key = f"{operation}:{self.market}:{timeframe}" + cached = self._discovery_cache.get(cache_key) + if cached and cached[0] > time.monotonic(): + return cached[1], cached[2] + + if operation == "kline": + query = ( + f"read-only historical OHLCV candlestick market data API for {self.market}; " + "accept symbol, interval or timeframe, limit, and optional start/end date" + ) + preferred = (os.getenv("QVERIS_KLINE_TOOL_ID") or "").strip() + else: + query = f"read-only latest market price quote API for {self.market} accepting a symbol" + preferred = (os.getenv("QVERIS_TICKER_TOOL_ID") or "").strip() + + payload = self._request_json("/search", json={"query": query, "limit": 10}) + discovery_id = str(payload.get("search_id") or payload.get("discovery_id") or "").strip() + candidates = payload.get("results") or [] + if not discovery_id or not isinstance(candidates, list): + raise QverisError("Qveris discovery response is missing search_id or results") + + tool = self._select_tool(candidates, preferred, operation, timeframe) + expires = time.monotonic() + self.discovery_ttl + self._discovery_cache[cache_key] = (expires, tool, discovery_id) + return tool, discovery_id + + def _select_tool( + self, + candidates: List[Dict[str, Any]], + preferred: str, + operation: str, + timeframe: str, + ) -> Dict[str, Any]: + if preferred: + for candidate in candidates: + if str(candidate.get("tool_id") or "") == preferred: + if self._supports_parameters(self._tool_parameters(candidate), operation, timeframe): + return candidate + raise QverisError(f"Configured Qveris tool {preferred} has unsupported required parameters") + raise QverisError(f"Configured Qveris tool {preferred} was not returned by discovery") + + compatible = [ + candidate + for candidate in candidates + if isinstance(candidate, dict) + and candidate.get("tool_id") + and self._supports_parameters(self._tool_parameters(candidate), operation, timeframe) + ] + if not compatible: + raise QverisError("Qveris discovery returned no compatible tool") + + terms = ( + ("ohlcv", "candlestick", "historical", "time series") + if operation == "kline" + else ("quote", "price", "ticker") + ) + + def rank(candidate: Dict[str, Any]) -> tuple[int, int, float, float]: + text = f"{candidate.get('name', '')} {candidate.get('description', '')}".lower() + relevance = sum(1 for term in terms if term in text) + positive, negative = _MARKET_RANK_TERMS.get(self.market, (set(), set())) + market_affinity = sum(1 for term in positive if term in text) + market_affinity -= 2 * sum(1 for term in negative if term in text) + stats = candidate.get("stats") if isinstance(candidate.get("stats"), dict) else {} + success_rate = _float(candidate.get("success_rate")) + if success_rate is None: + success_rate = _float(stats.get("success_rate")) or 0.0 + execution_time = _float(candidate.get("avg_execution_time_ms")) + if execution_time is None: + execution_time = _float(stats.get("avg_execution_time_ms")) + return relevance, market_affinity, success_rate, -(execution_time or float("inf")) + + compatible.sort(key=rank, reverse=True) + return compatible[0] + + @staticmethod + def _tool_parameters(tool: Dict[str, Any]) -> Any: + """Read current Qveris metadata while accepting older cached results.""" + return tool.get("parameters") or tool.get("params") or [] + + @staticmethod + def _supports_parameters(specs: Any, operation: str, timeframe: str = "") -> bool: + if not isinstance(specs, list) or not specs: + return False + known = set().union(*_PARAM_ALIASES.values()) + normalized = {_key(spec.get("name")) for spec in specs if isinstance(spec, dict)} + if not any(name in _PARAM_ALIASES["symbol"] for name in normalized): + return False + if operation == "kline" and not any(name in _PARAM_ALIASES["timeframe"] for name in normalized): + return False + if operation == "kline": + candidates = _TIMEFRAME_OPTIONS.get(timeframe, (timeframe,)) + for spec in specs: + if not isinstance(spec, dict): + continue + if _key(spec.get("name")) not in _PARAM_ALIASES["timeframe"]: + continue + options = spec.get("enum") or spec.get("options") or [] + option_values = [str(option.get("value") if isinstance(option, dict) else option) for option in options] + if option_values and not any( + _key(candidate) == _key(option) for candidate in candidates for option in option_values + ): + return False + for spec in specs: + if not isinstance(spec, dict) or not spec.get("required"): + continue + options = spec.get("enum") or spec.get("options") or [] + if _key(spec.get("name")) not in known and len(options) != 1: + return False + return True + + def _build_parameters(self, specs: Any, context: Dict[str, Any]) -> Dict[str, Any]: + parameters: Dict[str, Any] = {} + if not isinstance(specs, list): + raise QverisError("Qveris tool parameter metadata is missing") + for spec in specs: + if not isinstance(spec, dict): + continue + name = str(spec.get("name") or "").strip() + normalized = _key(name) + value = self._parameter_value(normalized, spec, context) + if value is None: + if spec.get("required"): + raise QverisError(f"Cannot map required Qveris tool parameter: {name}") + continue + parameters[name] = value + return parameters + + def _parameter_value(self, name: str, spec: Dict[str, Any], context: Dict[str, Any]) -> Any: + if name in _PARAM_ALIASES["symbol"]: + symbol = str(context.get("symbol") or "").strip() + param_type = str(spec.get("type") or "").lower() + description = str(spec.get("description") or "").lower() + if ( + self.market == "USStock" + and "." not in symbol + and (name == "symbolexchange" or "exchange suffix" in description or ".us" in description) + ): + symbol = f"{symbol}.US" + return [symbol] if "array" in param_type or name in {"symbols", "tickers"} else symbol + if name in _PARAM_ALIASES["market"]: + return self.market + if name in _PARAM_ALIASES["timeframe"]: + return self._timeframe_value(str(context.get("timeframe") or "1D"), spec) + if name in _PARAM_ALIASES["limit"]: + return int(context.get("limit") or 300) + if name in _PARAM_ALIASES["start"]: + start = context.get("after_time") + if start is None: + end = int(context.get("before_time") or time.time()) + start = end - self.calculate_time_range( + str(context.get("timeframe") or "1D"), + int(context.get("limit") or 300), + buffer_ratio=1.5, + ) + return self._time_parameter(int(start), name, spec) + if name in _PARAM_ALIASES["end"]: + end = int(context.get("before_time") or time.time()) + return self._time_parameter(end, name, spec) + options = spec.get("enum") or spec.get("options") or [] + option_values = [str(option.get("value") if isinstance(option, dict) else option) for option in options] + if name in {"fmt", "format", "datatype"}: + for option in option_values: + if option.lower() == "json": + return option + if len(option_values) == 1: + return option_values[0] + return None + + @staticmethod + def _time_parameter(timestamp: int, name: str, spec: Dict[str, Any]) -> Any: + description = str(spec.get("description") or "").lower() + if "date" in name or "yyyy-mm-dd" in description: + return datetime.fromtimestamp(timestamp, tz=timezone.utc).date().isoformat() + return timestamp + + @staticmethod + def _timeframe_value(timeframe: str, spec: Dict[str, Any]) -> str: + candidates = _TIMEFRAME_OPTIONS.get(timeframe, (timeframe,)) + options = spec.get("enum") or spec.get("options") or [] + option_values = [str(option.get("value") if isinstance(option, dict) else option) for option in options] + for candidate in candidates: + for option in option_values: + if _key(candidate) == _key(option): + return option + return candidates[0] + + def _request_json(self, path: str, *, params: Optional[dict] = None, json: Optional[dict] = None) -> Dict[str, Any]: + try: + response = self.session.post( + f"{self.base_url}{path}", + params=params or {}, + json=json or {}, + headers={ + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + }, + timeout=self.timeout, + ) + response.raise_for_status() + payload = response.json() + except requests.RequestException as exc: + raise QverisError(f"Qveris HTTP request failed: {exc}") from exc + except ValueError as exc: + raise QverisError("Qveris returned invalid JSON") from exc + if not isinstance(payload, dict): + raise QverisError("Qveris response must be a JSON object") + return payload + + def _normalize_klines(self, payload: Any) -> List[Dict[str, Any]]: + candidates: List[List[Dict[str, Any]]] = [] + self._collect_row_lists(payload, candidates, depth=0) + for dated_rows in self._dated_row_dicts(payload, depth=0): + candidates.append(dated_rows) + normalized = [self._normalize_row(row) for rows in candidates for row in rows] + valid = [row for row in normalized if row is not None] + deduped = {row["time"]: row for row in valid} + return [deduped[timestamp] for timestamp in sorted(deduped)] + + def _collect_row_lists(self, value: Any, output: List[List[Dict[str, Any]]], *, depth: int) -> None: + if depth > 7: + return + if isinstance(value, list): + dict_rows = [row for row in value if isinstance(row, dict)] + if dict_rows: + output.append(dict_rows) + for item in value: + self._collect_row_lists(item, output, depth=depth + 1) + elif isinstance(value, dict): + column_rows = self._column_rows(value) + if column_rows: + output.append(column_rows) + for item in value.values(): + self._collect_row_lists(item, output, depth=depth + 1) + + def _dated_row_dicts(self, value: Any, *, depth: int) -> List[List[Dict[str, Any]]]: + if depth > 7: + return [] + output: List[List[Dict[str, Any]]] = [] + if isinstance(value, dict): + rows: List[Dict[str, Any]] = [] + for key, item in value.items(): + if isinstance(item, dict) and _timestamp(key) is not None: + rows.append({"time": key, **item}) + if rows: + output.append(rows) + for item in value.values(): + output.extend(self._dated_row_dicts(item, depth=depth + 1)) + elif isinstance(value, list): + for item in value: + output.extend(self._dated_row_dicts(item, depth=depth + 1)) + return output + + @staticmethod + def _column_rows(value: Dict[str, Any]) -> List[Dict[str, Any]]: + normalized = {_key(name): item for name, item in value.items()} + + def column(field: str) -> Optional[List[Any]]: + for alias in _ROW_ALIASES[field]: + item = normalized.get(alias) + if isinstance(item, list): + return item + return None + + times = column("time") + opens = column("open") + highs = column("high") + lows = column("low") + closes = column("close") + volumes = column("volume") or [] + if not all((times, opens, highs, lows, closes)): + return [] + length = min(len(times), len(opens), len(highs), len(lows), len(closes)) + return [ + { + "time": times[index], + "open": opens[index], + "high": highs[index], + "low": lows[index], + "close": closes[index], + "volume": volumes[index] if index < len(volumes) else 0, + } + for index in range(length) + ] + + def _normalize_row(self, row: Dict[str, Any]) -> Optional[Dict[str, Any]]: + timestamp = _timestamp(_field(row, _ROW_ALIASES["time"])) + open_price = _float(_field(row, _ROW_ALIASES["open"])) + high = _float(_field(row, _ROW_ALIASES["high"])) + low = _float(_field(row, _ROW_ALIASES["low"])) + close = _float(_field(row, _ROW_ALIASES["close"])) + volume = _float(_field(row, _ROW_ALIASES["volume"])) or 0.0 + if timestamp is None or None in (open_price, high, low, close): + return None + return self.format_kline(timestamp, open_price, high, low, close, volume) + + def _normalize_ticker(self, payload: Any, *, depth: int = 0) -> Optional[Dict[str, Any]]: + if depth > 7: + return None + if isinstance(payload, dict): + last = _float(_field(payload, _QUOTE_ALIASES["last"])) + if last is not None and last > 0: + result: Dict[str, Any] = {"last": last} + for output_name, aliases in _QUOTE_ALIASES.items(): + if output_name == "last": + continue + value = _float(_field(payload, aliases)) + result[output_name] = value if value is not None else 0 + return result + for item in payload.values(): + quote = self._normalize_ticker(item, depth=depth + 1) + if quote: + return quote + elif isinstance(payload, list): + for item in payload: + quote = self._normalize_ticker(item, depth=depth + 1) + if quote: + return quote + return None diff --git a/backend_api_python/env.example b/backend_api_python/env.example index 18ea79594..ca9dd56af 100644 --- a/backend_api_python/env.example +++ b/backend_api_python/env.example @@ -501,6 +501,20 @@ MINIMAX_BASE_URL=https://api.minimax.io/v1 DATA_SOURCE_TIMEOUT=30 DATA_SOURCE_RETRY=3 DATA_SOURCE_RETRY_BACKOFF=0.5 + +# Optional Qveris unified data source. It is disabled unless both the API key +# and at least one market are configured. Existing providers remain the fallback. +# Markets: Crypto, Forex, Futures, USStock, CNStock, HKStock, MOEX, or *. +QVERIS_API_KEY= +QVERIS_DATA_SOURCE_MARKETS= +QVERIS_BASE_URL=https://qveris.ai/api/v1 +QVERIS_TIMEOUT=30 +QVERIS_DISCOVERY_TTL_SECONDS=3600 +# Optional: pin a discovered tool instead of accepting the highest-ranked +# compatible result. The tool still needs to appear in a fresh discovery. +QVERIS_KLINE_TOOL_ID= +QVERIS_TICKER_TOOL_ID= + FINNHUB_API_KEY= # Optional free-tier Finnhub helper for US stock quotes, company profile and news. # Keep FINNHUB_FREE_ONLY=true unless your plan includes paid endpoints such as Economic Calendar/Social Sentiment. diff --git a/backend_api_python/tests/test_qveris_data_source.py b/backend_api_python/tests/test_qveris_data_source.py new file mode 100644 index 000000000..4e2ac23ab --- /dev/null +++ b/backend_api_python/tests/test_qveris_data_source.py @@ -0,0 +1,305 @@ +"""Qveris market-data adapter tests.""" + +from __future__ import annotations + +import requests + +from app.data_sources.base import BaseDataSource +from app.data_sources.factory import DataSourceFactory +from app.data_sources.qveris import QverisDataSource + + +class FakeResponse: + def __init__(self, payload): + self.payload = payload + + def raise_for_status(self): + return None + + def json(self): + return self.payload + + +class FakeSession: + def __init__(self, payloads=None, error=None): + self.payloads = list(payloads or []) + self.error = error + self.calls = [] + + def post(self, url, **kwargs): + self.calls.append({"url": url, **kwargs}) + if self.error: + raise self.error + return FakeResponse(self.payloads.pop(0)) + + +class FallbackDataSource(BaseDataSource): + name = "fallback" + + def __init__(self): + self.kline_calls = 0 + self.ticker_calls = 0 + + def get_kline(self, symbol, timeframe, limit, before_time=None, after_time=None): + self.kline_calls += 1 + return [{"time": 1, "open": 1, "high": 1, "low": 1, "close": 1, "volume": 1}] + + def get_ticker(self, symbol): + self.ticker_calls += 1 + return {"last": 42.0, "symbol": symbol} + + +def kline_discovery(): + return { + "search_id": "search-kline", + "results": [ + { + "tool_id": "market.history.execute.v1", + "name": "Historical OHLCV", + "description": "Historical candlestick time series", + "parameters": [ + {"name": "symbol", "required": True, "type": "string"}, + {"name": "interval", "required": True, "type": "string", "enum": ["1h", "1d"]}, + {"name": "limit", "required": False, "type": "integer"}, + ], + } + ], + } + + +def ticker_discovery(): + return { + "search_id": "search-ticker", + "results": [ + { + "tool_id": "market.quote.execute.v1", + "name": "Latest quote", + "description": "Current market price quote", + "parameters": [{"name": "ticker", "required": True, "type": "string"}], + } + ], + } + + +def test_qveris_normalizes_row_oriented_klines(): + session = FakeSession( + [ + kline_discovery(), + { + "success": True, + "result": { + "data": [ + { + "datetime": "2026-08-01T00:00:00Z", + "open": "100", + "high": 105, + "low": 99, + "close": 104, + "volume": "1,200", + }, + { + "datetime": "2026-08-02T00:00:00Z", + "open": 104, + "high": 108, + "low": 103, + "close": 107, + "volume": 900, + }, + ] + }, + }, + ] + ) + fallback = FallbackDataSource() + source = QverisDataSource("USStock", fallback, api_key="test-key", session=session) + + rows = source.get_kline("AAPL", "1D", 2) + + assert [row["close"] for row in rows] == [104.0, 107.0] + assert rows[0]["volume"] == 1200.0 + assert fallback.kline_calls == 0 + assert session.calls[1]["params"] == {"tool_id": "market.history.execute.v1"} + assert session.calls[1]["json"]["parameters"] == { + "symbol": "AAPL", + "interval": "1d", + "limit": 2, + } + assert session.calls[1]["headers"]["Authorization"] == "Bearer test-key" + + +def test_qveris_normalizes_column_oriented_klines(): + session = FakeSession( + [ + kline_discovery(), + { + "success": True, + "result": { + "timestamps": [1785542400, 1785628800], + "open": [10, 11], + "high": [12, 13], + "low": [9, 10], + "close": [11, 12], + "volume": [100, 200], + }, + }, + ] + ) + source = QverisDataSource("USStock", FallbackDataSource(), api_key="test-key", session=session) + + rows = source.get_kline("MSFT", "1D", 2) + + assert [row["time"] for row in rows] == [1785542400, 1785628800] + assert [row["close"] for row in rows] == [11.0, 12.0] + + +def test_qveris_prefers_market_match_and_maps_eodhd_stock_parameters(): + discovery = { + "search_id": "search-eod", + "results": [ + { + "tool_id": "crypto-history", + "name": "Historical Data for Cryptocurrency", + "description": "Historical OHLCV data", + "params": [ + {"name": "symbol", "required": True, "type": "string"}, + {"name": "period", "required": False, "enum": ["d", "w", "m"]}, + ], + "stats": {"success_rate": 1.0}, + }, + { + "tool_id": "stock-history", + "name": "Historical Stock Market Data", + "description": "Historical end-of-day OHLCV for equities", + "params": [ + { + "name": "symbol_exchange", + "required": True, + "type": "string", + "description": "Ticker with exchange suffix, for example AAPL.US", + }, + {"name": "fmt", "required": False, "enum": ["json", "csv"]}, + {"name": "period", "required": False, "enum": ["d", "w", "m"]}, + {"name": "from", "required": False, "description": "Start date in YYYY-MM-DD format"}, + {"name": "to", "required": False, "description": "End date in YYYY-MM-DD format"}, + ], + "stats": {"success_rate": 0.8}, + }, + ], + } + session = FakeSession( + [ + discovery, + { + "success": True, + "result": [{"date": "2026-08-01", "open": 200, "high": 205, "low": 199, "close": 204, "volume": 1000}], + }, + ] + ) + source = QverisDataSource("USStock", FallbackDataSource(), api_key="test-key", session=session) + + rows = source.get_kline("AAPL", "1D", 1, before_time=1785715200) + + assert rows[0]["close"] == 204.0 + assert session.calls[1]["params"] == {"tool_id": "stock-history"} + assert session.calls[1]["json"]["parameters"] == { + "symbol_exchange": "AAPL.US", + "fmt": "json", + "period": "d", + "from": "2026-08-01", + "to": "2026-08-03", + } + + +def test_qveris_normalizes_dated_time_series(): + session = FakeSession( + [ + kline_discovery(), + { + "success": True, + "result": { + "Time Series (Daily)": { + "2026-08-01": { + "1. open": "20", + "2. high": "22", + "3. low": "19", + "4. close": "21", + "5. volume": "300", + } + } + }, + }, + ] + ) + source = QverisDataSource("USStock", FallbackDataSource(), api_key="test-key", session=session) + + rows = source.get_kline("NVDA", "1D", 1) + + assert rows[0]["close"] == 21.0 + assert rows[0]["volume"] == 300.0 + + +def test_qveris_normalizes_ticker(): + session = FakeSession( + [ + ticker_discovery(), + { + "success": True, + "result": { + "quote": { + "regularMarketPrice": 215.5, + "regularMarketPreviousClose": 210, + "regularMarketDayHigh": 217, + "regularMarketDayLow": 209, + } + }, + }, + ] + ) + fallback = FallbackDataSource() + source = QverisDataSource("USStock", fallback, api_key="test-key", session=session) + + quote = source.get_ticker("AAPL") + + assert quote["last"] == 215.5 + assert quote["previousClose"] == 210.0 + assert quote["symbol"] == "AAPL" + assert fallback.ticker_calls == 0 + + +def test_qveris_falls_back_without_exposing_failure(): + fallback = FallbackDataSource() + session = FakeSession(error=requests.ConnectionError("offline")) + source = QverisDataSource("USStock", fallback, api_key="secret-key", session=session) + + rows = source.get_kline("AAPL", "1D", 1) + quote = source.get_ticker("AAPL") + + assert rows[0]["close"] == 1 + assert quote["last"] == 42.0 + assert fallback.kline_calls == 1 + assert fallback.ticker_calls == 1 + + +def test_qveris_is_disabled_until_key_and_market_are_configured(monkeypatch): + monkeypatch.delenv("QVERIS_API_KEY", raising=False) + monkeypatch.setenv("QVERIS_DATA_SOURCE_MARKETS", "USStock") + assert not QverisDataSource.is_enabled_for("USStock") + + monkeypatch.setenv("QVERIS_API_KEY", "configured") + assert QverisDataSource.is_enabled_for("USStock") + assert not QverisDataSource.is_enabled_for("Crypto") + + monkeypatch.setenv("QVERIS_DATA_SOURCE_MARKETS", "*") + assert QverisDataSource.is_enabled_for("Crypto") + + +def test_factory_wraps_existing_source_only_when_enabled(monkeypatch): + fallback = FallbackDataSource() + monkeypatch.delenv("QVERIS_API_KEY", raising=False) + monkeypatch.setenv("QVERIS_DATA_SOURCE_MARKETS", "USStock") + assert DataSourceFactory._wrap_optional_source("USStock", fallback) is fallback + + monkeypatch.setenv("QVERIS_API_KEY", "configured") + wrapped = DataSourceFactory._wrap_optional_source("USStock", fallback) + assert isinstance(wrapped, QverisDataSource) + assert wrapped.fallback is fallback diff --git a/docs/README_CN.md b/docs/README_CN.md index 6db2326c7..22db9d05c 100644 --- a/docs/README_CN.md +++ b/docs/README_CN.md @@ -218,6 +218,12 @@ docker compose ps Windows、国内镜像、数据库迁移等问题见 [安装故障排查](deployment/INSTALL_TROUBLESHOOTING.md)和[云部署指南](deployment/CLOUD_DEPLOYMENT_CN.md)。 +### 可选 Qveris 数据源 + +QuantDinger 可以按需启用 Qveris 作为统一行情数据层，并在调用失败时自动回退到 +现有数据源。配置方式、支持市场、安全行为与验证步骤见 +[Qveris 数据源接入指南](integrations/QVERIS_DATA_SOURCE.md)。 + ## 生产部署 启动前校验全部生产密钥： diff --git a/docs/integrations/QVERIS_DATA_SOURCE.md b/docs/integrations/QVERIS_DATA_SOURCE.md new file mode 100644 index 000000000..92d306b36 --- /dev/null +++ b/docs/integrations/QVERIS_DATA_SOURCE.md @@ -0,0 +1,104 @@ +# Qveris Data Source + +Qveris can be enabled as an optional, unified market-data layer in front of QuantDinger's existing providers. The integration discovers a compatible read-only Qveris tool, executes it, normalizes common OHLCV or quote response shapes, and falls back to the existing market source when discovery, execution, or normalization fails. + +No existing provider is replaced by default. + +## Configuration + +Add the following values to `backend_api_python/.env`: + +```dotenv +QVERIS_API_KEY=your-qveris-api-key +QVERIS_DATA_SOURCE_MARKETS=USStock,CNStock,HKStock +``` + +Supported market names are `Crypto`, `Forex`, `Futures`, `USStock`, `CNStock`, `HKStock`, and `MOEX`. Use `*` to enable every market. + +Optional settings: + +```dotenv +QVERIS_BASE_URL=https://qveris.ai/api/v1 +QVERIS_TIMEOUT=30 +QVERIS_DISCOVERY_TTL_SECONDS=3600 +QVERIS_KLINE_TOOL_ID= +QVERIS_TICKER_TOOL_ID= +``` + +Leave the tool IDs empty to accept the highest-ranked compatible discovery result. Set them when you want to pin a specific tool. A pinned tool must still appear in a fresh Qveris discovery response so that the adapter receives a valid discovery ID. + +Restart the backend after changing the environment: + +```bash +docker compose up -d --build backend +``` + +## How Requests Flow + +1. `DataSourceFactory` creates the existing QuantDinger provider for the requested market. +2. When a Qveris API key and that market are explicitly configured, the factory wraps the existing provider with `QverisDataSource`. +3. The adapter searches Qveris for a compatible read-only OHLCV or quote tool. +4. It maps common parameters such as symbol, timeframe, limit, and start/end time. +5. It executes the selected tool and normalizes row-oriented, column-oriented, and dated time-series responses. +6. If any step fails, the original provider handles the request. + +Discovery results are cached in memory. API keys stay in the backend environment and are never returned to clients or included in logs. + +## Verify The Adapter + +Run the isolated unit tests, which cover tool discovery, market-aware tool selection, parameter mapping, three common OHLCV response layouts, quote normalization, opt-in behavior, and fallback: + +```bash +cd backend_api_python +python -m pytest tests/test_qveris_data_source.py -q +``` + +Expected result: + +```text +8 passed +``` + +For a live smoke test, start QuantDinger with a valid Qveris key and request data through the existing Agent Gateway: + +```bash +curl -sS -G http://localhost:5000/api/agent/v1/klines \ + -H "Authorization: Bearer ${QUANTDINGER_AGENT_TOKEN}" \ + --data-urlencode market=USStock \ + --data-urlencode symbol=AAPL \ + --data-urlencode timeframe=1D \ + --data-urlencode limit=5 +``` + +The public contract remains the existing normalized QuantDinger format: + +```json +{ + "code": 0, + "message": "ok", + "data": { + "market": "USStock", + "symbol": "AAPL", + "timeframe": "1D", + "count": 1, + "klines": [ + { + "time": 1785542400, + "open": 100.0, + "high": 105.0, + "low": 99.0, + "close": 104.0, + "volume": 1200.0 + } + ] + } +} +``` + +The exact upstream tool depends on Qveris discovery and the configured market. Pin a tool ID when reproducibility matters. + +## Live Result + +The adapter was smoke-tested with `USStock`, `AAPL`, `1D`, and a limit of five. Qveris discovered an end-of-day stock-data tool and returned five rows in QuantDinger's normalized OHLCV format without using the fallback. No API key is included in the image. + +![Qveris live smoke test](../screenshots/qveris-live-smoke-test.png) diff --git a/docs/screenshots/qveris-live-smoke-test.png b/docs/screenshots/qveris-live-smoke-test.png new file mode 100644 index 000000000..faef05c76 Binary files /dev/null and b/docs/screenshots/qveris-live-smoke-test.png differ