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

1"""Beefy Finance adapter for :mod:`stable_yield_lab`.""" 

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, STABLE_TOKENS 

13 

14logger = logging.getLogger(__name__) 

15 

16 

17class BeefySource: 

18 """HTTP client for Beefy vault data.""" 

19 

20 VAULTS_URL = "https://api.beefy.finance/vaults" 

21 APY_URL = "https://api.beefy.finance/apy" 

22 TVL_URL = "https://api.beefy.finance/tvl" 

23 

24 CHAIN_IDS = { 

25 "ethereum": "1", 

26 "bsc": "56", 

27 "polygon": "137", 

28 "arbitrum": "42161", 

29 "optimism": "10", 

30 } 

31 

32 def __init__(self, cache_dir: str | None = None) -> None: 

33 self.cache_dir = Path(cache_dir) if cache_dir else None 

34 

35 def _get_json(self, name: str, url: str) -> Any: 

36 if self.cache_dir: 

37 path = self.cache_dir / f"{name}.json" 

38 if path.exists(): 

39 with path.open() as f: 

40 return json.load(f) 

41 with urllib.request.urlopen(url) as resp: # pragma: no cover - network path 

42 data = json.load(resp) 

43 if self.cache_dir: 

44 self.cache_dir.mkdir(parents=True, exist_ok=True) 

45 with (self.cache_dir / f"{name}.json").open("w") as f: 

46 json.dump(data, f) 

47 return data 

48 

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

50 try: 

51 vaults = self._get_json("vaults", self.VAULTS_URL) 

52 apy = self._get_json("apy", self.APY_URL) 

53 tvl = self._get_json("tvl", self.TVL_URL) 

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

55 logger.warning("Beefy request failed: %s", exc) 

56 return [] 

57 pools: list[Pool] = [] 

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

59 for v in vaults: 

60 if v.get("status") != "active": 60 ↛ 61line 60 didn't jump to line 61 because the condition on line 60 was never true

61 continue 

62 assets = v.get("assets") or [] 

63 if not assets or not all( 63 ↛ 66line 63 didn't jump to line 66 because the condition on line 63 was never true

64 a.upper() in STABLE_TOKENS or "USD" in a.upper() for a in assets 

65 ): 

66 continue 

67 chain = str(v.get("chain", "")) 

68 chain_id = self.CHAIN_IDS.get(chain.lower(), chain) 

69 tvl_usd = float(tvl.get(str(chain_id), {}).get(v["id"], 0.0)) 

70 base = float(apy.get(v["id"], 0.0)) 

71 pools.append( 

72 Pool( 

73 name=str(v.get("name", v["id"])), 

74 chain=chain, 

75 stablecoin=str(assets[0]), 

76 tvl_usd=tvl_usd, 

77 base_apy=base, 

78 reward_apy=0.0, 

79 is_auto=True, 

80 source="beefy", 

81 timestamp=now, 

82 ) 

83 ) 

84 return pools 

85 

86 

87__all__ = ["BeefySource"]