Jonathan
a2ca82062e
When the CoinGecko API fails, fall back to the last successful price instead of 0.0, and surface a warning indicator on the price card. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
76 lines
2.5 KiB
Python
76 lines
2.5 KiB
Python
import logging
|
|
import requests
|
|
from datetime import datetime, timezone
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_btc_history_eur() -> list:
|
|
try:
|
|
resp = requests.get(
|
|
"https://api.coingecko.com/api/v3/coins/bitcoin/market_chart",
|
|
params={"vs_currency": "eur", "days": "365", "interval": "daily"},
|
|
timeout=15,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json().get("prices", []) # [[timestamp_ms, price], ...]
|
|
except Exception as e:
|
|
logger.error(f"Failed to fetch BTC history: {e}")
|
|
return []
|
|
|
|
|
|
def get_btc_ohlc_eur(days: int) -> list:
|
|
"""Fetch OHLC candles from CoinGecko. Returns [[ts_ms, open, high, low, close], ...]."""
|
|
try:
|
|
resp = requests.get(
|
|
"https://api.coingecko.com/api/v3/coins/bitcoin/ohlc",
|
|
params={"vs_currency": "eur", "days": str(days)},
|
|
timeout=20,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json() # [[timestamp_ms, open, high, low, close], ...]
|
|
except Exception as e:
|
|
logger.error(f"Failed to fetch BTC OHLC: {e}")
|
|
return []
|
|
|
|
|
|
def aggregate_to_daily(raw: list) -> dict:
|
|
"""Collapse intraday OHLC rows into one candle per UTC calendar date.
|
|
|
|
Returns {date_str: {open, high, low, close}} using first open, max high,
|
|
min low, last close per day.
|
|
"""
|
|
by_date: dict = {}
|
|
for row in raw:
|
|
ts_ms, o, h, l, c = row
|
|
date = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc).strftime("%Y-%m-%d")
|
|
if date not in by_date:
|
|
by_date[date] = {"open": o, "high": h, "low": l, "close": c}
|
|
else:
|
|
existing = by_date[date]
|
|
existing["high"] = max(existing["high"], h)
|
|
existing["low"] = min(existing["low"], l)
|
|
existing["close"] = c # last close wins
|
|
return by_date
|
|
|
|
|
|
_last_known_price: float = 0.0
|
|
|
|
|
|
def get_btc_price_eur() -> tuple[float, bool]:
|
|
"""Returns (price, is_cached). is_cached=True when using a stale fallback."""
|
|
global _last_known_price
|
|
try:
|
|
resp = requests.get(
|
|
"https://api.coingecko.com/api/v3/simple/price",
|
|
params={"ids": "bitcoin", "vs_currencies": "eur"},
|
|
timeout=10,
|
|
)
|
|
resp.raise_for_status()
|
|
price = float(resp.json()["bitcoin"]["eur"])
|
|
_last_known_price = price
|
|
return price, False
|
|
except Exception as e:
|
|
logger.error(f"Failed to fetch BTC price: {e}")
|
|
return _last_known_price, True
|