Coverage for tests / test_creditmetrics.py: 98%

66 statements  

« prev     ^ index     » next       coverage.py v7.12.0, created at 2025-11-27 14:36 +0000

1import unittest 

2import numpy as np 

3import pandas as pd 

4from rivapy.credit.creditmetrics import CreditMetricsModel 

5from rivapy.instruments.components import Issuer 

6 

7 

8class TestCreditMetricsModel(unittest.TestCase): 

9 def setUp(self): 

10 # Minimal Testdaten 

11 self.n_simulation = 100 

12 self.transition_matrix = ( 

13 np.matrix( 

14 """ 

15 90.81, 8.33, 0.68, 0.06, 0.08, 0.02, 0.01, 0.01; 

16 0.70, 90.65, 7.79, 0.64, 0.06, 0.13, 0.02, 0.01; 

17 0.09, 2.27, 91.05, 5.52, 0.74, 0.26, 0.01, 0.06; 

18 0.02, 0.33, 5.95, 85.93, 5.30, 1.17, 1.12, 0.18; 

19 0.03, 0.14, 0.67, 7.73, 80.53, 8.84, 1.00, 1.06; 

20 0.01, 0.11, 0.24, 0.43, 6.48, 83.46, 4.07, 5.20; 

21 0.21, 0, 0.22, 1.30, 2.38, 11.24, 64.86, 19.79""" 

22 ) 

23 / 100 

24 ) 

25 self.position_data = pd.DataFrame({"IssuerID": ["A", "B"], "Exposure": [100, 200], "RecoveryRate": [0.4, 0.5]}) 

26 self.issuer_data = [ 

27 Issuer(obj_id="A", name="IssuerA", rating="AAA", country="DE", sector="Industrials", esg_rating="AAA"), 

28 Issuer(obj_id="B", name="IssuerB", rating="BBB", country="DE", sector="Industrials", esg_rating="AAA"), 

29 ] 

30 self.stock_data = pd.DataFrame( 

31 { 

32 "Date": pd.date_range("2020-01-01", periods=5), 

33 "Dax": [100, 101, 102, 103, 104], 

34 "IssuerA": [50, 51, 52, 53, 54], 

35 "IssuerB": [60, 61, 62, 63, 64], 

36 } 

37 ) 

38 self.r = 0.01 

39 self.t = 1 

40 self.confidencelevel = 5 

41 self.seed = 42 

42 

43 self.model = CreditMetricsModel( 

44 n_simulation=self.n_simulation, 

45 transition_matrix=self.transition_matrix, 

46 position_data=self.position_data, 

47 issuer_data=self.issuer_data, 

48 stock_data=self.stock_data, 

49 r=self.r, 

50 t=self.t, 

51 confidencelevel=self.confidencelevel, 

52 seed=self.seed, 

53 list_of_indices=["Dax"], 

54 mapping_countries_on_indices={"DE": "Dax", "US": "SP"}, 

55 ) 

56 

57 def test_merge_positions_issuer(self): 

58 merged = self.model.merge_positions_issuer() 

59 self.assertIn("IssuerName", merged.columns) 

60 self.assertIn("RatingID", merged.columns) 

61 self.assertEqual(len(merged), 2) 

62 

63 def test_get_correlation(self): 

64 # get_correlation now returns (indices_correlation, corr_pairs) 

65 indices_corr, corr_pairs = self.model.get_correlation() 

66 # indices_corr should be a DataFrame containing the configured index names 

67 for idx_name in self.model.list_of_indices: 

68 self.assertIn(idx_name, indices_corr.columns) 

69 # corr_pairs should contain correlations for issuer columns 

70 self.assertIn("IssuerA", corr_pairs.index) 

71 self.assertIn("IssuerB", corr_pairs.index) 

72 

73 def test_get_cutoffs_rating(self): 

74 cutoffs = self.model.get_cutoffs_rating() 

75 # Cutoffs shape: (8 target ratings, 8 initial ratings) after transformation 

76 self.assertEqual(cutoffs.shape[0], 8) 

77 

78 def test_get_expected_value(self): 

79 ev = self.model.get_expected_value() 

80 self.assertTrue("EV" in ev.columns) 

81 self.assertTrue("issuer" in ev.index.names or "issuer" in ev.index) 

82 

83 def test_get_states(self): 

84 states = self.model.get_states() 

85 self.assertTrue("AAA" in states.columns) 

86 self.assertTrue("issuer" in states.index.names or "issuer" in states.index) 

87 

88 def test_mc_calculation(self): 

89 Loss, rr_scenarios, issuer_ids, issuer_names = self.model.mc_calculation() 

90 self.assertEqual(Loss.shape, (self.n_simulation, len(issuer_ids))) 

91 self.assertEqual(rr_scenarios.shape, (self.n_simulation, len(issuer_ids))) 

92 self.assertEqual(len(issuer_ids), 2) 

93 self.assertEqual(len(issuer_names), 2) 

94 

95 def test_get_loss_distribution(self): 

96 Loss, _, _, _ = self.model.mc_calculation() 

97 loss_distribution = self.model.get_loss_distribution(Loss) 

98 self.assertEqual(loss_distribution.shape[0], self.n_simulation) 

99 

100 def test_get_portfolio_VaR(self): 

101 Loss, _, _, _ = self.model.mc_calculation() 

102 loss_distribution = self.model.get_loss_distribution(Loss) 

103 expected = np.sort(loss_distribution) 

104 expected = (-1) * np.interp((self.confidencelevel) / 100, np.linspace(0, 1, len(expected)), expected) 

105 var = self.model.get_portfolio_VaR(loss_distribution) 

106 self.assertIsInstance(var, float) 

107 self.assertAlmostEqual(var, expected, places=5) 

108 

109 def test_get_portfolio_ES(self): 

110 Loss, _, _, _ = self.model.mc_calculation() 

111 loss_distribution = self.model.get_loss_distribution(Loss) 

112 # Compute expected shortfall the same way as the model 

113 expected = -1.0 * np.mean(loss_distribution[loss_distribution <= np.percentile(loss_distribution, self.confidencelevel)]) 

114 es = self.model.get_portfolio_ES(loss_distribution) 

115 self.assertIsInstance(es, float) 

116 self.assertAlmostEqual(es, expected, places=5) 

117 

118 

119if __name__ == "__main__": 

120 unittest.main()