Coverage for adrs/report/portfolio.py: 0%

1import io 

2import polars as pl 

3from typing import Self 

4from pydantic import BaseModel 

5 

6from adrs import json_utils as json 

7from adrs.portfolio import ( 

8 Portfolio, 

9 PortfolioPerformance, 

10) 

11 

12 

13class PortfolioReportV1Performance(BaseModel): 

14 performance: PortfolioPerformance 

15 performance_df: pl.DataFrame 

16 

17 class Config: 

18 arbitrary_types_allowed = True 

19 

20 

21class PortfolioReportV1(BaseModel): 

22 portfolio_id: str 

23 back: PortfolioReportV1Performance 

24 forward: PortfolioReportV1Performance 

25 

26 @staticmethod 

27 def version() -> str: 

28 return "portfolio-report-v1" 

29 

30 @classmethod 

31 def compute( 

32 cls, 

33 id: str, 

34 signal_df: pl.DataFrame, 

35 metadata_df: pl.DataFrame, 

36 weight_df: pl.DataFrame, 

37 prices_df: pl.DataFrame, 

38 back_pct: float = 0.7, 

39 ) -> Self: 

40 timestamps = signal_df["start_time"].sort() 

41 n = len(timestamps) 

42 split_idx = int(n * back_pct) 

43 split_time = timestamps[split_idx] 

44 

45 def make_portfolio( 

46 lf: pl.DataFrame, pf: pl.DataFrame 

47 ) -> tuple[Portfolio, pl.DataFrame]: 

48 return Portfolio( 

49 id=id, 

50 signal_df=lf, 

51 metadata_df=metadata_df, 

52 weight_df=weight_df, 

53 ), pf 

54 

55 back_portfolio, back_prices = make_portfolio( 

56 signal_df.filter(pl.col("start_time") < split_time), 

57 prices_df.filter(pl.col("start_time") < split_time), 

58 ) 

59 forward_portfolio, forward_prices = make_portfolio( 

60 signal_df.filter(pl.col("start_time") >= split_time), 

61 prices_df.filter(pl.col("start_time") >= split_time), 

62 ) 

63 

64 back_perf, back_df = back_portfolio.backtest(back_prices) 

65 forward_perf, forward_df = forward_portfolio.backtest(forward_prices) 

66 

67 return cls( 

68 portfolio_id=id, 

69 back=PortfolioReportV1Performance( 

70 performance=back_perf, 

71 performance_df=back_df, 

72 ), 

73 forward=PortfolioReportV1Performance( 

74 performance=forward_perf, 

75 performance_df=forward_df, 

76 ), 

77 ) 

78 

79 def __eq__(self, other: object) -> bool: 

80 if not isinstance(other, PortfolioReportV1): 

81 return False 

82 return ( 

83 self.portfolio_id == other.portfolio_id 

84 and self.back.model_dump(exclude={"performance_df"}) 

85 == other.back.model_dump(exclude={"performance_df"}) 

86 and self.back.performance_df.equals(other.back.performance_df) 

87 and self.forward.model_dump(exclude={"performance_df"}) 

88 == other.forward.model_dump(exclude={"performance_df"}) 

89 and self.forward.performance_df.equals(other.forward.performance_df) 

90 ) 

91 

92 def serialize(self) -> bytes: 

93 back_pdf_buf = io.BytesIO() 

94 self.back.performance_df.write_parquet(back_pdf_buf) 

95 forward_pdf_buf = io.BytesIO() 

96 self.forward.performance_df.write_parquet(forward_pdf_buf) 

97 

98 buf = io.BytesIO() 

99 df = pl.DataFrame( 

100 { 

101 "portfolio_id": [self.portfolio_id], 

102 "back": [self.back.model_dump_json(exclude={"performance_df"})], 

103 "back_pdf_buf": [back_pdf_buf.getvalue()], 

104 "forward": [self.forward.model_dump_json(exclude={"performance_df"})], 

105 "forward_pdf_buf": [forward_pdf_buf.getvalue()], 

106 } 

107 ) 

108 df.write_parquet(buf) 

109 

110 return buf.getvalue() 

111 

112 @classmethod 

113 def deserialize(cls, buf: bytes) -> Self: 

114 df = pl.read_parquet(buf) 

115 

116 back_pdf = pl.read_parquet(df["back_pdf_buf"][0]) 

117 forward_pdf = pl.read_parquet(df["forward_pdf_buf"][0]) 

118 

119 return cls( 

120 portfolio_id=df["portfolio_id"][0], 

121 back=PortfolioReportV1Performance( 

122 **json.loads(df["back"][0]), 

123 performance_df=back_pdf, 

124 ), 

125 forward=PortfolioReportV1Performance( 

126 **json.loads(df["forward"][0]), 

127 performance_df=forward_pdf, 

128 ), 

129 )