Coverage for src/stable_yield_lab/sources/defillama.py: 80%

1"""DefiLlama adapter returning :class:`~stable_yield_lab.core.Pool` instances.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import logging 

7import urllib.request 

8from datetime import UTC, datetime 

9from pathlib import Path 

10from typing import Any 

11 

12from ..core import Pool 

13 

14logger = logging.getLogger(__name__) 

15 

16 

17class DefiLlamaSource: 

18 """HTTP client for https://yields.llama.fi/pools.""" 

19 

20 URL = "https://yields.llama.fi/pools" 

21 

22 def __init__(self, stable_only: bool = True, cache_path: str | None = None) -> None: 

23 self.stable_only = stable_only 

24 self.cache_path = Path(cache_path) if cache_path else None 

25 

26 def _load(self) -> dict[str, Any]: 

27 if self.cache_path and self.cache_path.exists(): 

28 with self.cache_path.open() as f: 

29 return json.load(f) 

30 with urllib.request.urlopen(self.URL) as resp: # pragma: no cover - network path 

31 data = json.load(resp) 

32 if self.cache_path: 

33 self.cache_path.parent.mkdir(parents=True, exist_ok=True) 

34 with self.cache_path.open("w") as f: 

35 json.dump(data, f) 

36 return data 

37 

38 def fetch(self) -> list[Pool]: 

39 try: 

40 raw = self._load() 

41 except Exception as exc: # pragma: no cover - network errors 

42 logger.warning("DefiLlama request failed: %s", exc) 

43 return [] 

44 pools: list[Pool] = [] 

45 now = datetime.now(tz=UTC).timestamp() 

46 for item in raw.get("data", []): 

47 if self.stable_only and not item.get("stablecoin"): 47 ↛ 48line 47 didn't jump to line 48 because the condition on line 47 was never true

48 continue 

49 base_val = item.get("apyBase") 

50 if base_val is None: 

51 base_val = item.get("apy", 0.0) 

52 reward_val = item.get("apyReward") or 0.0 

53 pools.append( 

54 Pool( 

55 name=f"{item.get('project', '')}:{item.get('symbol', '')}", 

56 chain=str(item.get("chain", "")), 

57 stablecoin=str(item.get("symbol", "")), 

58 tvl_usd=float(item.get("tvlUsd", 0.0)), 

59 base_apy=float(base_val) / 100.0, 

60 reward_apy=float(reward_val) / 100.0, 

61 is_auto=False, 

62 source="defillama", 

63 timestamp=now, 

64 ) 

65 ) 

66 return pools 

67 

68 

69__all__ = ["DefiLlamaSource"]