Coverage for src/stable_yield_lab/risk_scoring.py: 100%

1from __future__ import annotations 

2 

3"""Heuristic risk scoring for stablecoin yield pools.""" 

4 

5from dataclasses import replace 

6from typing import Mapping, TYPE_CHECKING 

7 

8if TYPE_CHECKING: # pragma: no cover - imported only for type checking 

9 from stable_yield_lab.core import Pool 

10 

11# Simple reputation mapping per chain. Values range from 0 (unknown) to 1 (blue chip). 

12CHAIN_REPUTATION: Mapping[str, float] = { 

13 "Ethereum": 1.0, 

14 "Arbitrum": 0.9, 

15 "Polygon": 0.7, 

16 "BSC": 0.6, 

17 "Tron": 0.4, 

18} 

19 

20 

21def calculate_risk_score(chain_rep: float, audits: int, yield_volatility: float) -> float: 

22 """Combine factors into a normalized risk score in the range [1, 3]. 

23 

24 Parameters 

25 ---------- 

26 chain_rep: 

27 Reputation of the underlying chain, ``0`` (unknown) to ``1`` (established). 

28 audits: 

29 Number of protocol security audits. Values above ``5`` are capped. 

30 yield_volatility: 

31 Normalized measure of historical yield volatility (``0`` stable, ``1`` erratic). 

32 """ 

33 

34 # Clamp inputs to expected ranges 

35 chain_rep = max(0.0, min(chain_rep, 1.0)) 

36 audits = max(0, min(audits, 5)) 

37 yield_volatility = max(0.0, min(yield_volatility, 1.0)) 

38 

39 chain_component = 1.0 - chain_rep # higher reputation lowers risk 

40 audit_component = 1.0 - audits / 5.0 # more audits lower risk 

41 vol_component = yield_volatility # higher volatility increases risk 

42 

43 raw = (chain_component + audit_component + vol_component) / 3.0 

44 return 1.0 + 2.0 * raw 

45 

46 

47def score_pool( 

48 pool: "Pool", 

49 *, 

50 chain_reputation: Mapping[str, float] | None = None, 

51 protocol_audits: Mapping[str, int] | None = None, 

52 yield_volatility: Mapping[str, float] | None = None, 

53) -> "Pool": 

54 """Return a new :class:`~stable_yield_lab.Pool` with an updated ``risk_score``. 

55 

56 Missing mapping entries default to neutral values (``0.5`` reputation, 

57 ``0`` audits and ``0`` volatility). 

58 """ 

59 

60 chain_rep_map = chain_reputation or CHAIN_REPUTATION 

61 audits_map = protocol_audits or {} 

62 vol_map = yield_volatility or {} 

63 

64 chain_rep = chain_rep_map.get(pool.chain, 0.5) 

65 audits = audits_map.get(pool.name, 0) 

66 vol = vol_map.get(pool.name, 0.0) 

67 

68 score = calculate_risk_score(chain_rep, audits, vol) 

69 return replace(pool, risk_score=score)