Coverage for /opt/conda/lib/python3.14/site-packages/medil/evaluate.py: 100%

1"""Evaluation metrics for learned MeDIL causal model structures.""" 

2 

3import numpy as np 

4import numpy.typing as npt 

5 

6 

7def sfd( 

8 true_biadj: npt.NDArray, 

9 predicted_biadj: npt.NDArray, 

10 to_return: str = "raw", 

11) -> int | float | tuple[int, float]: 

12 """Structural Frobenius distance sums difference of latent parents. 

13 

14 For a binary biadjacency matrix B, consider U = B'B, where U_ij 

15 counts the number of parents nodes i and j have in common (so U_ii 

16 is the assignment number, and diag(U) is a sufficient statistic 

17 for the graph under the 1-pure-child assumption). sfd(B_1, B_2) is 

18 the sum of the differences between U_ij for B_1 and B_2 (without 

19 double-counting for U_ji). 

20 

21 Parameters 

22 ---------- 

23 true_biadj : ndarray 

24 Ground-truth biadjacency matrix. 

25 predicted_biadj : ndarray 

26 Learned biadjacency matrix. 

27 to_return : str, optional 

28 ``"raw"`` (default), ``"normalized"``, or ``"both"``. 

29 

30 Returns 

31 ------- 

32 int or float or tuple 

33 Raw sfd (int) if ``to_return='raw'``, normalized nsfd (float) if 

34 ``to_return='normalized'``, or ``(sfd, nsfd)`` if ``to_return='both'``. 

35 """ 

36 true_biadj = true_biadj.astype(int) 

37 true_wtd_ug = true_biadj.T @ true_biadj 

38 

39 predicted_biadj = predicted_biadj.astype(int) 

40 predicted_wtd_ug = predicted_biadj.T @ predicted_biadj 

41 

42 sfd = np.abs(np.triu(true_wtd_ug - predicted_wtd_ug)).sum() 

43 

44 if to_return == "raw": 

45 return sfd 

46 

47 true_zeros = np.where(true_wtd_ug == 0) 

48 true_wtd_ug[true_zeros] = -1 

49 

50 predicted_zeros = np.where(predicted_wtd_ug == 0) 

51 predicted_wtd_ug[predicted_zeros] = -1 

52 

53 similarity = np.sum(true_wtd_ug * predicted_wtd_ug) 

54 

55 cosin_normalizer = np.sqrt((true_wtd_ug**2).sum()) * np.sqrt( 

56 (predicted_wtd_ug**2).sum() 

57 ) 

58 

59 nsfd = np.arccos(similarity / cosin_normalizer) / np.pi 

60 

61 match to_return: 

62 case "normalized": 

63 return nsfd 

64 case "both": 

65 return sfd, nsfd 

66 case _: 

67 raise ValueError("`to_return` should be 'raw', 'normalized', or 'both'") 

68 

69 

70def _shd( 

71 true_biadj: npt.NDArray, 

72 *, 

73 predicted_biadj: npt.NDArray = np.array([]), 

74 predicted_adj: npt.NDArray = np.array([]), 

75 to_return: str = "raw", 

76) -> int | float | tuple[int, float]: 

77 """Structural Hamming distance counts number of incorrect arrowheads/tails. 

78 

79 Parameters 

80 ---------- 

81 true_biadj: true bipartite directed graph 

82 predicted_biadj: learned bipartite directed graph 

83 predicted_adj: learned mixed graph 

84 

85 Returns 

86 ------- 

87 nshd: normalized structural Hamming distance 

88 """ 

89 if bool(len(predicted_biadj)) == bool(len(predicted_adj)): 

90 raise ValueError( 

91 "Must provide `predicted_biadj` or `predicted_adj` but not both." 

92 ) 

93 elif bool(len(predicted_biadj)): 

94 predicted_adj = _recover_ug(predicted_biadj) 

95 

96 ug = _recover_ug(true_biadj) 

97 

98 shd = np.logical_xor(ug, predicted_adj).sum() 

99 

100 if to_return == "raw": 

101 return shd 

102 

103 n = len(ug) 

104 nshd = shd / (n**2 - n) 

105 

106 match to_return: 

107 case "normalized": 

108 return nshd 

109 case "both": 

110 return shd, nshd 

111 case _: 

112 raise ValueError("`to_return` should be 'raw', 'normalized', or 'both'") 

113 

114 

115def _recover_ug(biadj_mat: npt.NDArray) -> npt.NDArray: 

116 ug = biadj_mat.T @ biadj_mat 

117 np.fill_diagonal(ug, False) 

118 return ug