Coverage for adrs/portfolio.py: 42%
86 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-29 05:07 +0800
1import polars as pl
3from typing import Protocol, Any
4from datetime import datetime
5from pydantic import BaseModel, ConfigDict
7from adrs.performance.metric import Ratio, Drawdown, Trade
10class TradePerformance(BaseModel):
11 largest_loss: float
12 num_datapoints: int
13 num_trades: int
14 avg_holding_time_in_seconds: float
15 long_trades: int
16 short_trades: int
17 win_trades: int
18 lose_trades: int
19 win_streak: int
20 lose_streak: int
21 win_rate: float
23 model_config = ConfigDict(extra="allow")
26class PortfolioPerformance(BaseModel):
27 sharpe_ratio: float
28 calmar_ratio: float
29 sortino_ratio: float
30 cagr: float
31 annualized_return: float
32 total_return: float
33 min_cumu: float
34 start_time: datetime
35 end_time: datetime
36 max_drawdown: float
37 max_drawdown_percentage: float
38 max_drawdown_start_date: datetime
39 max_drawdown_end_date: datetime
40 max_drawdown_recover_date: datetime
41 max_drawdown_max_duration_in_days: float
42 trades: dict[str, TradePerformance]
43 metadata: dict[str, Any]
45 model_config = ConfigDict(extra="allow")
48class WeightAllocator(Protocol):
49 def weights(self) -> pl.DataFrame: ...
52class MeanWeightAllocator:
53 def __init__(
54 self,
55 signal_df: pl.DataFrame,
56 metadata_df: pl.DataFrame,
57 ) -> None:
58 self.signal_df = signal_df
59 self.metadata_df = metadata_df
61 def weights(self) -> pl.DataFrame:
62 n = len(self.metadata_df)
63 return self.metadata_df.select("custom_id").with_columns(
64 pl.lit(1.0 / n).alias("weights")
65 )
68class Portfolio:
69 def __init__(
70 self,
71 id: str,
72 signal_df: pl.DataFrame,
73 metadata_df: pl.DataFrame,
74 weight_df: pl.DataFrame,
75 ) -> None:
76 self.id = id
77 self.signal_df = signal_df
78 self.metadata_df = metadata_df
79 self.weight_df = weight_df
80 self.lastest_signal: dict[str, float] = {}
82 def update_signal(self, id: str, signal: float):
83 self.lastest_signal[id] = signal
85 def update_weights(self, weight_df: pl.DataFrame):
86 self.weight_df = weight_df
88 def get_signal(self) -> pl.DataFrame:
89 rows = []
90 for custom_id in self.metadata_df["custom_id"].to_list():
91 if custom_id in self.lastest_signal:
92 signal = self.lastest_signal[custom_id]
93 else:
94 col = self.signal_df[custom_id]
95 if col.is_empty():
96 raise Exception(
97 f"Alpha ID {custom_id} was not initialized in portfolio"
98 )
99 signal = col[-1]
101 rows.append({"custom_id": custom_id, "signal": signal})
103 signal = pl.DataFrame(rows)
104 return (
105 self.metadata_df.join(signal, on="custom_id", how="left")
106 .join(self.weight_df, on="custom_id", how="left")
107 .with_columns(
108 (pl.col("signal") * pl.col("weights")).alias("weighted_signal")
109 )
110 .group_by("base_asset")
111 .agg(pl.col("weighted_signal").sum())
112 )
114 def backtest(
115 self, prices_df: pl.DataFrame
116 ) -> tuple[PortfolioPerformance, pl.DataFrame]:
117 # check if prices include all base asset
118 base_assets = self.metadata_df["base_asset"].unique().to_list()
119 price_cols = [c for c in prices_df.columns if c != "start_time"]
120 missing_cols = [a for a in base_assets if a not in price_cols]
121 if missing_cols:
122 raise ValueError(
123 f"Price DF did not include the following asset prices {missing_cols}"
124 )
126 merged_pnl: pl.DataFrame | None = None
127 asset_merged: dict[str, pl.DataFrame | None] = {}
128 for alpha in self.metadata_df.iter_rows(named=True):
129 custom_id: str = alpha["custom_id"]
130 fees: float = alpha["fees"]
131 price_shift: int = alpha["shift_backtest_candle_minute"]
132 base_asset: str = alpha["base_asset"]
133 alpha_weight: float = float(
134 self.weight_df.filter(pl.col("custom_id") == custom_id)["weights"][0]
135 )
137 alpha_prices = prices_df.select(
138 pl.col("start_time"),
139 pl.col(base_asset).shift(-price_shift).alias("price"),
140 ).drop_nulls()
142 alpha_signals = self.signal_df.select(
143 pl.col("start_time"),
144 pl.col(custom_id).alias("signal"),
145 )
147 pnl_col = f"{custom_id}_pnl"
148 sig_col = f"{custom_id}_sig"
149 alpha_full_df = (
150 alpha_prices.join(alpha_signals, on="start_time", how="left")
151 .with_columns(
152 pl.col("signal").forward_fill().fill_null(strategy="zero"),
153 pl.col("signal")
154 .shift(1)
155 .alias("prev_signal")
156 .forward_fill()
157 .fill_null(strategy="zero"),
158 pl.col("price")
159 .pct_change()
160 .alias("returns")
161 .fill_null(strategy="zero"),
162 )
163 .with_columns(
164 (pl.col("signal") - pl.col("prev_signal"))
165 .alias("trade")
166 .fill_null(strategy="zero")
167 )
168 .with_columns(
169 (
170 pl.col("prev_signal") * pl.col("returns")
171 - pl.col("trade").abs() * fees / 100
172 ).alias("pnl")
173 )
174 )
176 # accumulate per-asset weighted signal + price for Trade stats
177 alpha_asset_df = alpha_full_df.select(
178 pl.col("start_time"),
179 pl.col("price"),
180 (pl.col("signal") * alpha_weight).alias(sig_col),
181 (pl.col("pnl") * alpha_weight).alias(pnl_col),
182 )
183 if base_asset not in asset_merged or asset_merged[base_asset] is None:
184 asset_merged[base_asset] = alpha_asset_df
185 else:
186 asset_merged[base_asset] = (
187 asset_merged[base_asset]
188 .join(
189 alpha_asset_df.drop("price"),
190 on="start_time",
191 how="full",
192 coalesce=True,
193 )
194 .with_columns(
195 pl.col(sig_col).forward_fill(),
196 pl.col(pnl_col).forward_fill(),
197 )
198 )
200 alpha_df = alpha_full_df.select(
201 pl.col("start_time"),
202 (pl.col("pnl") * alpha_weight).alias(pnl_col),
203 )
205 if merged_pnl is None:
206 merged_pnl = alpha_df
207 else:
208 merged_pnl = merged_pnl.join(
209 alpha_df, on="start_time", how="full", coalesce=True
210 ).with_columns(pl.col(pnl_col).forward_fill())
212 trade_performances: dict[str, TradePerformance] = {}
213 for base_asset, asset_df in asset_merged.items():
214 assert asset_df is not None
215 sig_cols = [c for c in asset_df.columns if c.endswith("_sig")]
216 pnl_cols_asset = [c for c in asset_df.columns if c.endswith("_pnl")]
217 asset_trade_df = (
218 asset_df.select(
219 pl.col("start_time"),
220 pl.col("price"),
221 pl.sum_horizontal(sig_cols).alias("signal"),
222 pl.sum_horizontal(pnl_cols_asset).alias("pnl"),
223 )
224 .sort("start_time")
225 .with_columns(
226 pl.col("signal")
227 .shift(1)
228 .alias("prev_signal")
229 .forward_fill()
230 .fill_null(strategy="zero"),
231 pl.col("signal").diff().alias("trade").fill_null(strategy="zero"),
232 )
233 )
234 trade_result = Trade().compute(asset_trade_df)
235 trade_performances[base_asset] = TradePerformance(**trade_result)
237 assert merged_pnl is not None
238 pnl_cols = [c for c in merged_pnl.columns if c.endswith("_pnl")]
239 performance_df = (
240 merged_pnl.select(
241 pl.col("start_time"),
242 pl.sum_horizontal(pnl_cols).alias("pnl"),
243 *[pl.col(c) for c in pnl_cols],
244 )
245 .sort("start_time")
246 .with_columns(
247 pl.col("pnl").cum_sum().alias("equity").fill_null(strategy="zero"),
248 )
249 )
251 start_time = performance_df["start_time"].min()
252 end_time = performance_df["start_time"].max()
253 assert isinstance(start_time, datetime) and isinstance(end_time, datetime)
254 performance: dict[str, Any] = {
255 "start_time": start_time,
256 "end_time": end_time,
257 "trades": trade_performances,
258 "metadata": {},
259 }
260 for metric in [Ratio(), Drawdown()]:
261 performance = {**performance, **metric.compute(performance_df)}
263 return PortfolioPerformance.model_validate(performance), performance_df