Coverage for adrs/execution/backtest.py: 28%

1import polars as pl 

2 

3from pathlib import Path 

4from datetime import datetime, timedelta, timezone 

5 

6from adrs import Alpha 

7from adrs.types import Topic 

8from adrs.performance.evaluator import Evaluator, Datamap 

9 

10 

11class SignalCache: 

12 def __init__(self, cache_dir: Path): 

13 self._dir = cache_dir 

14 self._memory: dict[str, pl.DataFrame] = {} 

15 

16 def _path(self, key: str) -> Path: 

17 return self._dir / f"{key}.parquet" 

18 

19 def get(self, key: str) -> pl.DataFrame | None: 

20 if key in self._memory: 

21 return self._memory[key] 

22 path = self._path(key) 

23 if path.exists(): 

24 df = pl.read_parquet(path) 

25 self._memory[key] = df 

26 return df 

27 return None 

28 

29 def set(self, key: str, df: pl.DataFrame) -> None: 

30 self._dir.mkdir(parents=True, exist_ok=True) 

31 df.write_parquet(self._path(key)) 

32 self._memory[key] = df 

33 

34 

35_cache = SignalCache(Path("output/signal_cache")) 

36 

37 

38def _min_interval(alphas: list[Alpha]) -> timedelta: 

39 intervals = [ 

40 iv 

41 for alpha in alphas 

42 for info in alpha.data_infos 

43 if (iv := Topic.from_str(info.topic).interval()) is not None 

44 ] 

45 return min(intervals) 

46 

47 

48def _floor_to_interval(dt: datetime, interval: timedelta) -> datetime: 

49 interval_s = interval.total_seconds() 

50 ts = dt.timestamp() 

51 return datetime.fromtimestamp(ts - (ts % interval_s), tz=timezone.utc) 

52 

53 

54def _make_key(cache_id: str, start: datetime, end: datetime) -> str: 

55 return f"{cache_id}__{start.strftime('%Y%m%dT%H%M%S')}__{end.strftime('%Y%m%dT%H%M%S')}" 

56 

57 

58def generate_signal_df( 

59 alphas: list[Alpha], 

60 metadata_df: pl.DataFrame, 

61 evaluator: Evaluator, 

62 datamap: Datamap, 

63 start_time: datetime, 

64 end_time: datetime, 

65 cache_id: str = "signals", 

66 forward_fill_to_end: bool = False, 

67) -> pl.DataFrame | None: 

68 """Build the per-alpha signal matrix. 

69 

70 `forward_fill_to_end` (live/OMS only — keep False for backtests so results 

71 are unchanged): after assembling, forward-fill EVERY alpha column across all 

72 rows (the incremental join otherwise leaves earlier alphas null at 

73 timestamps a later alpha introduced) and carry the last decision forward to 

74 `snapped_end` (~now). This lets a mixed-cadence portfolio (e.g. 1h + 24h) 

75 present a signal_df whose last row reaches current time, so the OMS isn't 

76 rejected by the staleness guard while it holds the most recent signal.""" 

77 interval = _min_interval(alphas) 

78 snapped_start = _floor_to_interval(start_time, interval) 

79 snapped_end = _floor_to_interval(end_time, interval) 

80 key = _make_key(cache_id, snapped_start, snapped_end) 

81 if forward_fill_to_end: 

82 key += "__ffe" # distinct cache namespace so it never bleeds into backtests 

83 cached = _cache.get(key) 

84 if cached is not None: 

85 return cached 

86 

87 signal_df: pl.DataFrame | None = None 

88 for alpha in alphas: 

89 alpha_meta = metadata_df.filter(pl.col("custom_id") == alpha.id) 

90 base_asset = alpha_meta["base_asset"][0] 

91 _, df = alpha.backtest( 

92 evaluator=evaluator, 

93 base_asset=base_asset, 

94 datamap=datamap, 

95 start_time=snapped_start, 

96 end_time=snapped_end, 

97 fees=alpha_meta["fees"][0], 

98 price_shift=alpha_meta["shift_backtest_candle_minute"][0], 

99 ) 

100 alpha_signal = df.select( 

101 pl.col("start_time"), 

102 pl.col("signal").alias(alpha.id), 

103 ) 

104 if signal_df is None: 

105 signal_df = alpha_signal 

106 else: 

107 signal_df = ( 

108 signal_df.join(alpha_signal, on="start_time", how="full", coalesce=True) 

109 .sort("start_time") 

110 .with_columns(pl.col(alpha.id).forward_fill()) 

111 ) 

112 if signal_df is not None and forward_fill_to_end: 

113 alpha_cols = [a.id for a in alphas] 

114 # 1) carry every alpha's last signal across ALL rows (the per-join 

115 # forward_fill above leaves earlier alphas null at timestamps a later 

116 # alpha added). 

117 signal_df = signal_df.sort("start_time").with_columns( 

118 [pl.col(c).forward_fill() for c in alpha_cols] 

119 ) 

120 # 2) extend the last row to snapped_end (~now) so a slow topic whose last 

121 # bar is older doesn't make signal_df stale — hold the last decision. 

122 last_ts = signal_df["start_time"][-1] 

123 if last_ts is not None and last_ts < snapped_end: 

124 tail = pl.DataFrame({"start_time": [snapped_end]}).select( 

125 pl.col("start_time").cast(signal_df["start_time"].dtype) 

126 ) 

127 signal_df = ( 

128 pl.concat([signal_df, tail], how="diagonal") 

129 .sort("start_time") 

130 .with_columns([pl.col(c).forward_fill() for c in alpha_cols]) 

131 ) 

132 

133 if signal_df is not None: 

134 _cache.set(key, signal_df) 

135 

136 return signal_df