Coverage for src/pyhrp/covariance.py: 100%

1"""Covariance and correlation estimation from returns. 

2 

3This module isolates the second-moment estimators used by the HRP allocation 

4entry points: 

5- compute_returns: Simple returns from a DataFrame of prices 

6- compute_cov: Covariance matrix from a DataFrame of returns 

7- compute_corr: Correlation matrix from a DataFrame of returns 

8- check_finite_matrix: Guard that rejects matrices containing NaN/null entries 

9""" 

10 

11from __future__ import annotations 

12 

13import warnings 

14 

15import numpy as np 

16import polars as pl 

17 

18__all__ = ["check_finite_matrix", "compute_corr", "compute_cov", "compute_returns"] 

19 

20 

21def _warn_on_missing_returns(returns: pl.DataFrame) -> None: 

22 """Emit a warning when nulls or NaNs survive into the returns frame. 

23 

24 Args: 

25 returns (pl.DataFrame): Returns frame straight out of ``pct_change`` 

26 """ 

27 null_counts = returns.null_count().row(0, named=True) 

28 nan_counts = {c: int(n.is_nan().sum()) for c, n in ((col, returns[col]) for col in returns.columns)} 

29 affected = { 

30 asset: count + nan_counts[asset] for asset, count in null_counts.items() if count + nan_counts[asset] > 0 

31 } 

32 if affected: 

33 detail = ", ".join(f"{a} ({n} rows)" for a, n in sorted(affected.items())) 

34 warnings.warn( 

35 f"Missing prices detected for: {detail}. " 

36 "They are filled with zero returns, which biases covariance estimates. " 

37 "Clean the data upstream or drop affected rows/assets.", 

38 stacklevel=3, 

39 ) 

40 

41 

42def compute_returns(prices: pl.DataFrame) -> pl.DataFrame: 

43 r"""Compute simple returns from prices. 

44 

45 Drops leading all-null rows produced by pct_change and fills remaining 

46 nulls/NaNs (e.g. from missing prices) with zero returns. 

47 

48 Warning: 

49 Filling with zero is a dirty helper, not a data-cleaning strategy: a 

50 suspended asset or an asset listed mid-sample contributes fabricated 

51 zero returns that bias variance down and correlations toward zero. 

52 A :class:`UserWarning` is emitted whenever this path fires. Callers 

53 working with messy universes should clean the data first and compose 

54 the pipeline manually:: 

55 

56 from pyhrp import build_tree, risk_parity 

57 from pyhrp import compute_corr, compute_cov 

58 

59 returns = prices.select(pl.all().pct_change()).drop_nulls() 

60 root = risk_parity(root=build_tree(compute_corr(returns)).root, 

61 cov=compute_cov(returns)) 

62 

63 Args: 

64 prices (pl.DataFrame): Asset price time series (columns are assets, rows are dates) 

65 

66 Returns: 

67 pl.DataFrame: Simple returns, one row shorter than ``prices`` 

68 

69 Examples: 

70 >>> import polars as pl 

71 >>> from pyhrp.covariance import compute_returns 

72 >>> prices = pl.DataFrame({"A": [100.0, 110.0, 99.0]}) 

73 >>> compute_returns(prices)['A'].to_list() 

74 [0.1, -0.1] 

75 """ 

76 raw = prices.select(pl.all().pct_change()).filter(pl.any_horizontal(pl.all().is_not_null())) 

77 _warn_on_missing_returns(raw) 

78 return raw.fill_null(0.0).fill_nan(0.0) 

79 

80 

81def check_finite_matrix(matrix: pl.DataFrame, name: str = "matrix") -> pl.DataFrame: 

82 """Raise if a matrix contains NaN or null entries. 

83 

84 A covariance matrix with non-finite entries makes every downstream weight 

85 meaningless; this guard fails loudly instead of propagating garbage. 

86 

87 Args: 

88 matrix (pl.DataFrame): Square matrix (columns are assets) 

89 name (str): Human-readable name used in the error message 

90 

91 Returns: 

92 pl.DataFrame: The input matrix, unchanged 

93 

94 Raises: 

95 ValueError: If any entry of ``matrix`` is NaN or null. 

96 """ 

97 bad = { 

98 col: int(count) + int(matrix[col].is_nan().sum()) 

99 for col, count in matrix.null_count().row(0, named=True).items() 

100 if count > 0 or bool(matrix[col].is_nan().any()) 

101 } 

102 if bad: 

103 detail = ", ".join(f"{c} ({n})" for c, n in sorted(bad.items())) 

104 msg = f"{name} contains NaN/null entries in column(s): {detail}" 

105 raise ValueError(msg) 

106 return matrix 

107 

108 

109def compute_cov(df: pl.DataFrame) -> pl.DataFrame: 

110 """Compute covariance matrix from a DataFrame of returns.""" 

111 cols = df.columns 

112 cov = np.atleast_2d(np.cov(df.to_numpy().T)) 

113 return pl.DataFrame(dict(zip(cols, cov, strict=True))) 

114 

115 

116def compute_corr(df: pl.DataFrame) -> pl.DataFrame: 

117 """Compute correlation matrix from a DataFrame of returns.""" 

118 cols = df.columns 

119 corr = np.atleast_2d(np.corrcoef(df.to_numpy().T)) 

120 return pl.DataFrame(dict(zip(cols, corr, strict=True)))