100 lines
3.5 KiB
Python
100 lines
3.5 KiB
Python
"""
|
|
External data signals for Bayesian probability estimation.
|
|
Sources: CoinGecko (crypto prices), Alternative.me (Fear&Greed), Polymarket trends
|
|
"""
|
|
import logging
|
|
from dataclasses import dataclass
|
|
import httpx
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class ExternalSignals:
|
|
btc_price: float = 0.0
|
|
btc_change_24h: float = 0.0 # % change
|
|
eth_price: float = 0.0
|
|
eth_change_24h: float = 0.0
|
|
btc_dominance: float = 50.0 # BTC market dominance %
|
|
fear_greed_index: int = 50 # 0=extreme fear, 100=extreme greed
|
|
fear_greed_label: str = "neutral"
|
|
total_market_cap_change: float = 0.0
|
|
valid: bool = False
|
|
|
|
|
|
class ExternalDataClient:
|
|
"""Fetches external market signals used to calibrate probability estimates."""
|
|
|
|
def __init__(self) -> None:
|
|
self._client = httpx.AsyncClient(timeout=15)
|
|
|
|
async def get_all_signals(self) -> ExternalSignals:
|
|
"""Aggregate all external signals. Returns best-effort (partial ok)."""
|
|
signals = ExternalSignals()
|
|
|
|
try:
|
|
prices = await self._get_crypto_prices()
|
|
signals.btc_price = prices.get("bitcoin", {}).get("usd", 0)
|
|
signals.btc_change_24h = prices.get("bitcoin", {}).get("usd_24h_change", 0)
|
|
signals.eth_price = prices.get("ethereum", {}).get("usd", 0)
|
|
signals.eth_change_24h = prices.get("ethereum", {}).get("usd_24h_change", 0)
|
|
except Exception as e:
|
|
log.warning("CoinGecko fetch failed: %s", e)
|
|
|
|
try:
|
|
fg = await self._get_fear_greed()
|
|
signals.fear_greed_index = fg["value"]
|
|
signals.fear_greed_label = fg["label"]
|
|
except Exception as e:
|
|
log.warning("Fear&Greed fetch failed: %s", e)
|
|
|
|
try:
|
|
global_data = await self._get_global_market()
|
|
signals.btc_dominance = global_data.get("btc_dominance", 50)
|
|
signals.total_market_cap_change = global_data.get("market_cap_change_24h", 0)
|
|
except Exception as e:
|
|
log.warning("Global market data fetch failed: %s", e)
|
|
|
|
signals.valid = signals.btc_price > 0
|
|
log.info(
|
|
"External signals: BTC=$%.0f (%.1f%%) F&G=%d/%s",
|
|
signals.btc_price,
|
|
signals.btc_change_24h,
|
|
signals.fear_greed_index,
|
|
signals.fear_greed_label,
|
|
)
|
|
return signals
|
|
|
|
async def _get_crypto_prices(self) -> dict:
|
|
resp = await self._client.get(
|
|
"https://api.coingecko.com/api/v3/simple/price",
|
|
params={
|
|
"ids": "bitcoin,ethereum",
|
|
"vs_currencies": "usd",
|
|
"include_24hr_change": True,
|
|
},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def _get_fear_greed(self) -> dict:
|
|
resp = await self._client.get("https://api.alternative.me/fng/?limit=1")
|
|
resp.raise_for_status()
|
|
data = resp.json()["data"][0]
|
|
return {
|
|
"value": int(data["value"]),
|
|
"label": data["value_classification"],
|
|
}
|
|
|
|
async def _get_global_market(self) -> dict:
|
|
resp = await self._client.get("https://api.coingecko.com/api/v3/global")
|
|
resp.raise_for_status()
|
|
data = resp.json()["data"]
|
|
return {
|
|
"btc_dominance": data.get("market_cap_percentage", {}).get("btc", 50),
|
|
"market_cap_change_24h": data.get("market_cap_change_percentage_24h_usd", 0),
|
|
}
|
|
|
|
async def close(self) -> None:
|
|
await self._client.aclose()
|