Coverage for src/stable_yield_lab/sources/morpho.py: 72%

1"""Morpho Blue 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 MorphoSource: 

18 """GraphQL client for Morpho Blue markets.""" 

19 

20 URL = "https://blue-api.morpho.org/graphql" 

21 

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

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

24 

25 def _post_json(self) -> dict[str, Any]: 

26 payload = { 

27 "query": ( 

28 "{ markets { items { uniqueKey loanAsset { symbol } " 

29 "collateralAsset { symbol } state { supplyApy supplyAssetsUsd } } } }" 

30 ) 

31 } 

32 req = urllib.request.Request( 

33 self.URL, 

34 data=json.dumps(payload).encode(), 

35 headers={"Content-Type": "application/json"}, 

36 ) 

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

38 return json.load(resp) 

39 

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

41 if self.cache_path and self.cache_path.exists(): 41 ↛ 44line 41 didn't jump to line 44 because the condition on line 41 was always true

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

43 return json.load(f) 

44 data = self._post_json() 

45 if self.cache_path: 

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

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

48 json.dump(data, f) 

49 return data 

50 

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

52 try: 

53 raw = self._load() 

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

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

56 return [] 

57 pools: list[Pool] = [] 

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

59 items = raw.get("data", {}).get("markets", {}).get("items", []) 

60 for item in items: 

61 sym = str(item.get("loanAsset", {}).get("symbol", "")) 

62 if sym.upper() not in STABLE_TOKENS: 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true

63 continue 

64 pools.append( 

65 Pool( 

66 name=f"{sym}-{item.get('collateralAsset', {}).get('symbol', '')}", 

67 chain="Ethereum", 

68 stablecoin=sym, 

69 tvl_usd=float(item.get("state", {}).get("supplyAssetsUsd", 0.0)), 

70 base_apy=float(item.get("state", {}).get("supplyApy", 0.0)) / 100.0, 

71 reward_apy=0.0, 

72 is_auto=True, 

73 source="morpho", 

74 timestamp=now, 

75 ) 

76 ) 

77 return pools 

78 

79 

80__all__ = ["MorphoSource"]