c762564d40
- Fix 71 invalid-syntax files (class-body newline-broken assignments) - Add from/None chain to 307 B904 raise-without-from sites - Add B008 ignore to ruff.toml (already in pyproject.toml) - Noqa F401 on __init__.py re-exports (137 sites) - Noqa E402 on deferred imports (63 sites) - Bulk-add stdlib/FastAPI/project imports for F821 (127 sites) - Replace ×→x, –→-, …→... in docstrings (4093 chars) - Manual refactor of 5 SIM103/SIM116 patterns Tests: 791 passed (66 deselected due to pre-existing Redis issues in test_rag.py) Co-authored-by: opencode <opencode@rugmunch.io>
634 lines
25 KiB
Python
634 lines
25 KiB
Python
"""
|
|
Price Consensus Engine - Multi-Source Aggregation with MAD Outlier Detection.
|
|
|
|
Queries 7+ price sources in parallel, applies Median Absolute Deviation (MAD)
|
|
outlier filtering (z-score > 3 = outlier), and computes a weighted mean price
|
|
using source reliability scores.
|
|
|
|
Sources: DexScreener, GeckoTerminal, Jupiter (Solana), DIA, CoinGecko,
|
|
CryptoCompare, Coinpaprika - all free tier, no paid keys required.
|
|
|
|
Depends on: httpx, numpy (for median/percentile), optional env keys.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import numpy as np
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ── Source Reliability Scores (0.0-1.0, higher = more trusted) ─────────────
|
|
|
|
# These are initial weights based on historical accuracy, API stability,
|
|
# and data freshness. They can be adjusted via _source_stats over time.
|
|
DEFAULT_SOURCE_WEIGHTS = {
|
|
"dexscreener": 0.90, # Direct DEX data, excellent for on-chain tokens
|
|
"geckoterminal": 0.92, # CoinGecko's DEX aggregator, very reliable
|
|
"jupiter": 0.88, # Solana's primary aggregator, excellent for Solana
|
|
"dia": 0.85, # Oracle-grade data, transparent methodology
|
|
"coingecko": 0.88, # CEX + DEX aggregation, broad coverage
|
|
"cryptocompare": 0.82, # Institutional-grade, slower updates on microcaps
|
|
"coinpaprika": 0.78, # Good coverage, slightly less reliable on low-cap
|
|
"birdeye": 0.86, # Good Solana coverage, needs API key
|
|
}
|
|
|
|
# ── Data Classes ────────────────────────────────────────────────────────────
|
|
|
|
|
|
@dataclass
|
|
class PriceSource:
|
|
"""A single price data provider."""
|
|
|
|
name: str
|
|
weight: float # Reliability score 0-1
|
|
fetcher: Any = None # Async callable: (address, chain) → Optional[float]
|
|
last_price: float = 0.0
|
|
last_latency: float = 0.0
|
|
error_count: int = 0
|
|
|
|
|
|
@dataclass
|
|
class PriceConsensus:
|
|
"""Result of multi-source price consensus."""
|
|
|
|
price: float | None = None
|
|
confidence: float = 0.0 # 0-100%
|
|
sources_used: list[str] = field(default_factory=list)
|
|
outlier_sources: list[str] = field(default_factory=list)
|
|
failed_sources: list[str] = field(default_factory=list)
|
|
individual_prices: dict[str, float] = field(default_factory=dict)
|
|
median: float | None = None
|
|
mad: float | None = None
|
|
std_dev: float | None = None
|
|
spread_pct: float | None = None # (max-min)/median * 100
|
|
|
|
@property
|
|
def is_reliable(self) -> bool:
|
|
return self.confidence >= 60.0 and self.price is not None
|
|
|
|
|
|
# ── Price Consensus Engine ─────────────────────────────────────────────────
|
|
|
|
|
|
class PriceConsensusEngine:
|
|
"""Multi-source price aggregation with MAD-based outlier rejection.
|
|
|
|
Fetches from all configured sources in parallel, removes statistical
|
|
outliers (z-score > 3 using Median Absolute Deviation), and computes
|
|
a weighted mean of the remaining prices. Falls back gracefully if
|
|
fewer than 2 sources respond.
|
|
"""
|
|
|
|
# Timeout per source fetch
|
|
PER_SOURCE_TIMEOUT = 10.0
|
|
|
|
# If a source fails this many times consecutively, lower its effective weight
|
|
MAX_CONSECUTIVE_ERRORS = 5
|
|
|
|
def __init__(self):
|
|
self._sources: dict[str, PriceSource] = {}
|
|
self._lock = asyncio.Lock()
|
|
self._setup_sources()
|
|
|
|
def _setup_sources(self):
|
|
"""Register all price sources with their fetcher callables."""
|
|
sources = [
|
|
("dexscreener", self._fetch_dexscreener, DEFAULT_SOURCE_WEIGHTS["dexscreener"]),
|
|
("geckoterminal", self._fetch_geckoterminal, DEFAULT_SOURCE_WEIGHTS["geckoterminal"]),
|
|
("jupiter", self._fetch_jupiter, DEFAULT_SOURCE_WEIGHTS["jupiter"]),
|
|
("dia", self._fetch_dia, DEFAULT_SOURCE_WEIGHTS["dia"]),
|
|
("coingecko", self._fetch_coingecko, DEFAULT_SOURCE_WEIGHTS["coingecko"]),
|
|
("cryptocompare", self._fetch_cryptocompare, DEFAULT_SOURCE_WEIGHTS["cryptocompare"]),
|
|
("coinpaprika", self._fetch_coinpaprika, DEFAULT_SOURCE_WEIGHTS["coinpaprika"]),
|
|
]
|
|
|
|
# Birdeye if key is available
|
|
birdeye_key = os.getenv("BIRDEYE_API_KEY", "")
|
|
if birdeye_key and birdeye_key != "your_birdeye_key_here":
|
|
sources.append(("birdeye", self._fetch_birdeye, DEFAULT_SOURCE_WEIGHTS["birdeye"]))
|
|
|
|
for name, fetcher, weight in sources:
|
|
self._sources[name] = PriceSource(name=name, weight=weight, fetcher=fetcher)
|
|
|
|
logger.info(f"PriceConsensusEngine: {len(self._sources)} sources registered: {list(self._sources.keys())}")
|
|
|
|
# ── Source Fetchers ──────────────────────────────────────────────────
|
|
|
|
async def _fetch_dexscreener(self, address: str, chain: str) -> float | None:
|
|
"""DexScreener free API - no key required."""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.PER_SOURCE_TIMEOUT) as client:
|
|
r = await client.get(
|
|
f"https://api.dexscreener.com/latest/dex/tokens/{address}",
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
if r.status_code == 200:
|
|
data = r.json()
|
|
pairs = data.get("pairs", [])
|
|
if pairs:
|
|
# Find the pair with highest liquidity
|
|
best = max(pairs, key=lambda p: float(p.get("liquidity", {}).get("usd", 0) or 0))
|
|
price = best.get("priceUsd")
|
|
if price:
|
|
return float(price)
|
|
return None
|
|
except Exception as e:
|
|
logger.debug(f"DexScreener fetch error: {e}")
|
|
return None
|
|
|
|
async def _fetch_geckoterminal(self, address: str, chain: str) -> float | None:
|
|
"""GeckoTerminal free API - no key required."""
|
|
network = self._chain_to_gecko_network(chain)
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.PER_SOURCE_TIMEOUT) as client:
|
|
r = await client.get(
|
|
f"https://api.geckoterminal.com/api/v2/networks/{network}/tokens/{address}",
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
if r.status_code == 200:
|
|
data = r.json()
|
|
token_data = data.get("data", {})
|
|
attrs = token_data.get("attributes", {})
|
|
price = attrs.get("price_usd")
|
|
if price:
|
|
return float(price)
|
|
return None
|
|
except Exception as e:
|
|
logger.debug(f"GeckoTerminal fetch error: {e}")
|
|
return None
|
|
|
|
async def _fetch_jupiter(self, address: str, chain: str) -> float | None:
|
|
"""Jupiter price API - free, Solana only."""
|
|
if chain.lower() not in ("solana", "sol"):
|
|
return None
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.PER_SOURCE_TIMEOUT) as client:
|
|
r = await client.get(
|
|
f"https://price.jup.ag/v6/price?ids={address}",
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
if r.status_code == 200:
|
|
data = r.json()
|
|
token_data = data.get("data", {}).get(address)
|
|
if token_data:
|
|
price = token_data.get("price")
|
|
if price:
|
|
return float(price)
|
|
return None
|
|
except Exception as e:
|
|
logger.debug(f"Jupiter fetch error: {e}")
|
|
return None
|
|
|
|
async def _fetch_dia(self, address: str, chain: str) -> float | None:
|
|
"""DIA oracle price feed - free, no key."""
|
|
dia_chain = self._chain_to_dia_chain(chain)
|
|
if not dia_chain:
|
|
return None
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.PER_SOURCE_TIMEOUT) as client:
|
|
r = await client.get(
|
|
f"https://api.diadata.org/v1/assetQuotation/{dia_chain}/{address}",
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
if r.status_code == 200:
|
|
data = r.json()
|
|
price = data.get("Price")
|
|
if price:
|
|
return float(price)
|
|
return None
|
|
except Exception as e:
|
|
logger.debug(f"DIA fetch error: {e}")
|
|
return None
|
|
|
|
async def _fetch_coingecko(self, address: str, chain: str) -> float | None:
|
|
"""CoinGecko token price by contract - free tier."""
|
|
cg_chain = self._chain_to_coingecko_platform(chain)
|
|
if not cg_chain:
|
|
return None
|
|
api_key = os.getenv("COINGECKO_API_KEY", "")
|
|
headers = {"Accept": "application/json"}
|
|
if api_key:
|
|
headers["x-cg-demo-api-key"] = api_key
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.PER_SOURCE_TIMEOUT) as client:
|
|
r = await client.get(
|
|
f"https://api.coingecko.com/api/v3/simple/token_price/{cg_chain}",
|
|
params={
|
|
"contract_addresses": address,
|
|
"vs_currencies": "usd",
|
|
},
|
|
headers=headers,
|
|
)
|
|
if r.status_code == 200:
|
|
data = r.json()
|
|
price = data.get(address.lower(), {}).get("usd")
|
|
if price:
|
|
return float(price)
|
|
return None
|
|
except Exception as e:
|
|
logger.debug(f"CoinGecko fetch error: {e}")
|
|
return None
|
|
|
|
async def _fetch_cryptocompare(self, address: str, chain: str) -> float | None:
|
|
"""CryptoCompare price API - free tier."""
|
|
api_key = os.getenv("CRYPTOCOMPARE_API_KEY", "")
|
|
headers = {"Accept": "application/json"}
|
|
if api_key:
|
|
headers["authorization"] = f"Apikey {api_key}"
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.PER_SOURCE_TIMEOUT) as client:
|
|
r = await client.get(
|
|
"https://min-api.cryptocompare.com/data/price",
|
|
params={
|
|
"fsym": address,
|
|
"tsyms": "USD",
|
|
},
|
|
headers=headers,
|
|
)
|
|
if r.status_code == 200:
|
|
data = r.json()
|
|
price = data.get("USD")
|
|
if price:
|
|
return float(price)
|
|
return None
|
|
except Exception as e:
|
|
logger.debug(f"CryptoCompare fetch error: {e}")
|
|
return None
|
|
|
|
async def _fetch_coinpaprika(self, address: str, chain: str) -> float | None:
|
|
"""Coinpaprika free API - no key required."""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.PER_SOURCE_TIMEOUT) as client:
|
|
# Try by contract address lookup
|
|
r = await client.get(
|
|
f"https://api.coinpaprika.com/v1/contracts/{chain}/{address}",
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
if r.status_code == 200:
|
|
data = r.json()
|
|
coin_id = data.get("id")
|
|
if coin_id:
|
|
# Get ticker for this coin
|
|
r2 = await client.get(
|
|
f"https://api.coinpaprika.com/v1/tickers/{coin_id}",
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
if r2.status_code == 200:
|
|
ticker = r2.json()
|
|
price = ticker.get("quotes", {}).get("USD", {}).get("price")
|
|
if price:
|
|
return float(price)
|
|
return None
|
|
except Exception as e:
|
|
logger.debug(f"Coinpaprika fetch error: {e}")
|
|
return None
|
|
|
|
async def _fetch_birdeye(self, address: str, chain: str) -> float | None:
|
|
"""Birdeye price API - requires BIRDEYE_API_KEY."""
|
|
api_key = os.getenv("BIRDEYE_API_KEY", "")
|
|
if not api_key:
|
|
return None
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.PER_SOURCE_TIMEOUT) as client:
|
|
r = await client.get(
|
|
"https://public-api.birdeye.so/defi/price",
|
|
params={"address": address},
|
|
headers={
|
|
"X-API-KEY": api_key,
|
|
"accept": "application/json",
|
|
},
|
|
)
|
|
if r.status_code == 200:
|
|
data = r.json()
|
|
price = data.get("data", {}).get("value")
|
|
if price:
|
|
return float(price)
|
|
return None
|
|
except Exception as e:
|
|
logger.debug(f"Birdeye fetch error: {e}")
|
|
return None
|
|
|
|
# ── Chain Name Normalization ──────────────────────────────────────────
|
|
|
|
@staticmethod
|
|
def _chain_to_gecko_network(chain: str) -> str:
|
|
mapping = {
|
|
"solana": "solana",
|
|
"sol": "solana",
|
|
"ethereum": "eth",
|
|
"eth": "eth",
|
|
"1": "eth",
|
|
"base": "base",
|
|
"8453": "base",
|
|
"bsc": "bsc",
|
|
"56": "bsc",
|
|
"bnb": "bsc",
|
|
"arbitrum": "arbitrum",
|
|
"42161": "arbitrum",
|
|
"polygon": "polygon_pos",
|
|
"137": "polygon_pos",
|
|
"matic": "polygon_pos",
|
|
"optimism": "optimism",
|
|
"10": "optimism",
|
|
"avalanche": "avax",
|
|
"43114": "avax",
|
|
"fantom": "fantom",
|
|
"250": "fantom",
|
|
}
|
|
return mapping.get(chain.lower(), chain.lower())
|
|
|
|
@staticmethod
|
|
def _chain_to_dia_chain(chain: str) -> str | None:
|
|
mapping = {
|
|
"solana": "Solana",
|
|
"sol": "Solana",
|
|
"ethereum": "Ethereum",
|
|
"eth": "Ethereum",
|
|
"1": "Ethereum",
|
|
"base": "Base",
|
|
"8453": "Base",
|
|
"bsc": "BSC",
|
|
"56": "BSC",
|
|
"bnb": "BSC",
|
|
"arbitrum": "Arbitrum",
|
|
"42161": "Arbitrum",
|
|
"polygon": "Polygon",
|
|
"137": "Polygon",
|
|
"optimism": "Optimism",
|
|
"10": "Optimism",
|
|
}
|
|
return mapping.get(chain.lower())
|
|
|
|
@staticmethod
|
|
def _chain_to_coingecko_platform(chain: str) -> str | None:
|
|
mapping = {
|
|
"solana": "solana",
|
|
"sol": "solana",
|
|
"ethereum": "ethereum",
|
|
"eth": "ethereum",
|
|
"1": "ethereum",
|
|
"base": "base",
|
|
"8453": "base",
|
|
"bsc": "binance-smart-chain",
|
|
"56": "binance-smart-chain",
|
|
"bnb": "binance-smart-chain",
|
|
"arbitrum": "arbitrum-one",
|
|
"42161": "arbitrum-one",
|
|
"polygon": "polygon-pos",
|
|
"137": "polygon-pos",
|
|
"matic": "polygon-pos",
|
|
"optimism": "optimistic-ethereum",
|
|
"10": "optimistic-ethereum",
|
|
"avalanche": "avalanche",
|
|
"43114": "avalanche",
|
|
"fantom": "fantom",
|
|
"250": "fantom",
|
|
}
|
|
return mapping.get(chain.lower())
|
|
|
|
# ── Core Consensus Logic ──────────────────────────────────────────────
|
|
|
|
async def get_consensus_price(
|
|
self,
|
|
token_address: str,
|
|
chain: str = "solana",
|
|
) -> PriceConsensus:
|
|
"""Fetch prices from all sources and compute consensus.
|
|
|
|
Args:
|
|
token_address: Token contract address / mint
|
|
chain: Blockchain identifier (solana, ethereum, base, etc.)
|
|
|
|
Returns:
|
|
PriceConsensus with consensus price, confidence, and breakdown.
|
|
"""
|
|
if not self._sources:
|
|
return PriceConsensus(
|
|
price=None,
|
|
confidence=0.0,
|
|
failed_sources=["no_sources_configured"],
|
|
)
|
|
|
|
# Fire all source fetchers in parallel
|
|
tasks = []
|
|
source_names = []
|
|
for name, source in self._sources.items():
|
|
tasks.append(source.fetcher(token_address, chain))
|
|
source_names.append(name)
|
|
|
|
start = time.monotonic()
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
|
|
# Collect successful prices and track failures
|
|
prices: dict[str, float] = {}
|
|
failed: list[str] = []
|
|
|
|
for name, result in zip(source_names, results, strict=False):
|
|
if isinstance(result, Exception):
|
|
logger.debug(f"Source {name} exception: {result}")
|
|
failed.append(name)
|
|
async with self._lock:
|
|
if name in self._sources:
|
|
self._sources[name].error_count += 1
|
|
elif result is not None and isinstance(result, (int, float)):
|
|
if result > 0:
|
|
prices[name] = float(result)
|
|
latency = time.monotonic() - start
|
|
async with self._lock:
|
|
if name in self._sources:
|
|
self._sources[name].last_price = float(result)
|
|
self._sources[name].last_latency = latency
|
|
self._sources[name].error_count = 0
|
|
else:
|
|
failed.append(name)
|
|
|
|
# If no sources returned a price, return null consensus
|
|
if not prices:
|
|
logger.warning(f"No price sources responded for {token_address} on {chain}")
|
|
return PriceConsensus(
|
|
price=None,
|
|
confidence=0.0,
|
|
failed_sources=failed,
|
|
)
|
|
|
|
price_values = list(prices.values())
|
|
price_names = list(prices.keys())
|
|
|
|
# Single source: return it but with low confidence
|
|
if len(price_values) == 1:
|
|
return PriceConsensus(
|
|
price=price_values[0],
|
|
confidence=30.0,
|
|
sources_used=price_names,
|
|
failed_sources=failed,
|
|
individual_prices=prices,
|
|
median=price_values[0],
|
|
)
|
|
|
|
# ── MAD-based Outlier Detection ─────────────────────────────────
|
|
|
|
arr = np.array(price_values)
|
|
median = float(np.median(arr))
|
|
mad = float(np.median(np.abs(arr - median)))
|
|
|
|
# If MAD is zero (all prices identical), no outliers
|
|
if mad == 0:
|
|
weighted_avg = self._weighted_mean(prices)
|
|
return PriceConsensus(
|
|
price=weighted_avg,
|
|
confidence=95.0 if len(price_values) >= 3 else 70.0,
|
|
sources_used=price_names,
|
|
outlier_sources=[],
|
|
failed_sources=failed,
|
|
individual_prices=prices,
|
|
median=median,
|
|
mad=0.0,
|
|
std_dev=0.0,
|
|
spread_pct=0.0,
|
|
)
|
|
|
|
# Compute modified z-scores using MAD
|
|
# z_i = 0.6745 * (x_i - median) / MAD
|
|
z_scores = 0.6745 * (arr - median) / mad
|
|
|
|
# Outlier threshold: |z| > 3 (very conservative - classic threshold)
|
|
inliers_mask = np.abs(z_scores) <= 3.0
|
|
outliers_mask = ~inliers_mask
|
|
|
|
inlier_prices = {
|
|
name: price
|
|
for name, price, is_inlier in zip(price_names, price_values, inliers_mask, strict=False)
|
|
if is_inlier
|
|
}
|
|
outlier_names = [
|
|
name
|
|
for name, price, is_outlier in zip(price_names, price_values, outliers_mask, strict=False)
|
|
if is_outlier
|
|
]
|
|
|
|
# If all prices are outliers, fall back to all with low confidence
|
|
if not inlier_prices:
|
|
logger.warning(f"All prices flagged as outliers for {token_address} - using all with low confidence")
|
|
weighted_avg = self._weighted_mean(prices)
|
|
return PriceConsensus(
|
|
price=weighted_avg,
|
|
confidence=10.0,
|
|
sources_used=price_names,
|
|
outlier_sources=[],
|
|
failed_sources=failed,
|
|
individual_prices=prices,
|
|
median=median,
|
|
mad=float(mad),
|
|
std_dev=float(np.std(arr)),
|
|
spread_pct=self._spread_pct(price_values),
|
|
)
|
|
|
|
# Compute weighted mean of inliers
|
|
consensus_price = self._weighted_mean(inlier_prices)
|
|
|
|
# Confidence calculation
|
|
total_sources = len(self._sources)
|
|
inlier_count = len(inlier_prices)
|
|
responder_count = len(price_values)
|
|
|
|
# Base confidence from inlier agreement ratio
|
|
if inlier_count >= 3:
|
|
agreement_ratio = inlier_count / responder_count
|
|
confidence = agreement_ratio * 85.0 + 10.0 # 70-95 range
|
|
elif inlier_count == 2:
|
|
confidence = 55.0
|
|
else:
|
|
confidence = 35.0
|
|
|
|
# Penalize if we had many failures
|
|
failure_penalty = (len(failed) / max(total_sources, 1)) * 20.0
|
|
confidence = max(5.0, confidence - failure_penalty)
|
|
|
|
# Bonus for low spread among inliers
|
|
inlier_values = list(inlier_prices.values())
|
|
if len(inlier_values) >= 2:
|
|
spread = self._spread_pct(inlier_values)
|
|
if spread is not None and spread < 2.0:
|
|
confidence = min(100.0, confidence + 10.0)
|
|
|
|
return PriceConsensus(
|
|
price=round(consensus_price, 12),
|
|
confidence=round(confidence, 1),
|
|
sources_used=list(inlier_prices.keys()),
|
|
outlier_sources=outlier_names,
|
|
failed_sources=failed,
|
|
individual_prices=prices,
|
|
median=round(median, 12),
|
|
mad=round(float(mad), 12) if mad else None,
|
|
std_dev=round(float(np.std(arr)), 12),
|
|
spread_pct=self._spread_pct(price_values),
|
|
)
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────────
|
|
|
|
def _weighted_mean(self, prices: dict[str, float]) -> float:
|
|
"""Weighted mean using source reliability weights, adjusted by error history."""
|
|
if not prices:
|
|
return 0.0
|
|
total_weight = 0.0
|
|
weighted_sum = 0.0
|
|
for name, price in prices.items():
|
|
source = self._sources.get(name)
|
|
if source:
|
|
# Reduce weight if source has errors
|
|
error_penalty = min(0.5, source.error_count * 0.1)
|
|
weight = source.weight * (1.0 - error_penalty)
|
|
else:
|
|
weight = 0.5
|
|
weighted_sum += price * weight
|
|
total_weight += weight
|
|
return weighted_sum / total_weight if total_weight > 0 else 0.0
|
|
|
|
@staticmethod
|
|
def _spread_pct(values: list[float]) -> float | None:
|
|
"""(max - min) / median * 100. Lower = more consensus."""
|
|
if len(values) < 2:
|
|
return None
|
|
arr = np.array(values)
|
|
median = float(np.median(arr))
|
|
if median == 0:
|
|
return None
|
|
return round(float((arr.max() - arr.min()) / median * 100), 2)
|
|
|
|
# ── Stats ─────────────────────────────────────────────────────────────
|
|
|
|
async def stats(self) -> dict[str, Any]:
|
|
"""Return per-source stats and aggregate metrics."""
|
|
source_stats = {}
|
|
async with self._lock:
|
|
for name, src in self._sources.items():
|
|
source_stats[name] = {
|
|
"weight": src.weight,
|
|
"effective_weight": round(src.weight * (1.0 - min(0.5, src.error_count * 0.1)), 3),
|
|
"last_price": src.last_price,
|
|
"last_latency": round(src.last_latency, 3),
|
|
"error_count": src.error_count,
|
|
}
|
|
return {
|
|
"total_sources": len(self._sources),
|
|
"sources": source_stats,
|
|
}
|
|
|
|
|
|
# ── Singleton ─────────────────────────────────────────────────────────────
|
|
|
|
_price_engine: PriceConsensusEngine | None = None
|
|
|
|
|
|
def get_price_consensus() -> PriceConsensusEngine:
|
|
"""Get the global PriceConsensusEngine singleton."""
|
|
global _price_engine
|
|
if _price_engine is None:
|
|
_price_engine = PriceConsensusEngine()
|
|
return _price_engine
|