Coverage for adrs/execution/runner.py: 59%

1import asyncio 

2from datetime import timedelta 

3 

4import polars as pl 

5 

6from adrs.alpha import Alpha 

7from adrs.portfolio import Portfolio 

8from adrs.data import DataLoader, MetricStream, DatasourceStream 

9from adrs.execution.executor import PortfolioExecutor, AlphaExecutor 

10from adrs.types import Topic 

11 

12 

13def _remap_portfolio_ids(portfolio: Portfolio, alpha_id_map: dict[str, str]) -> None: 

14 """Rewrite the portfolio's alpha ids in place using alpha_id_map. 

15 

16 run_portfolio renames the Alpha objects' ids (publish side) but the 

17 portfolio's frames are keyed by the original ids. Without this, the live 

18 signal an alpha publishes (under its renamed id) never matches the 

19 portfolio's lookup and any "is this my alpha?" guard rejects every signal. 

20 

21 metadata_df/weight_df key by a `custom_id` column; signal_df keys each alpha 

22 by a column name. Unmapped values are left unchanged.""" 

23 portfolio.metadata_df = portfolio.metadata_df.with_columns( 

24 pl.col("custom_id").replace(alpha_id_map) 

25 ) 

26 portfolio.weight_df = portfolio.weight_df.with_columns( 

27 pl.col("custom_id").replace(alpha_id_map) 

28 ) 

29 portfolio.signal_df = portfolio.signal_df.rename( 

30 { 

31 old: new 

32 for old, new in alpha_id_map.items() 

33 if old in portfolio.signal_df.columns 

34 } 

35 ) 

36 

37 

38def _finest_interval(alphas: list[Alpha]) -> timedelta | None: 

39 """Smallest signal interval across all alphas' data topics. Used to default 

40 the signal-freshness window so a daily portfolio isn't rejected by a flat 

41 2h rule. Returns None if no topic declares an interval.""" 

42 intervals: list[timedelta] = [] 

43 for alpha in alphas: 

44 for di in getattr(alpha, "data_infos", []): 

45 try: 

46 iv = Topic.from_str(di.topic).interval() 

47 except Exception: 

48 iv = None 

49 if iv is not None: 

50 intervals.append(iv) 

51 return min(intervals) if intervals else None 

52 

53 

54async def run_portfolio( 

55 portfolio: Portfolio, 

56 alphas: list[Alpha], 

57 dataloader: DataLoader, 

58 metric_stream: MetricStream, 

59 datasource_stream: DatasourceStream, 

60 run_alphas: bool = True, 

61 alpha_id_map: dict[str, str] = {}, 

62 max_signal_age: timedelta | None = None, 

63 signal_namespace: str | None = None, 

64 insert_prefix: str = "public_ts", 

65 resync_interval: timedelta | None = None, 

66): 

67 for alpha in alphas: 

68 if alpha.id in alpha_id_map: 

69 alpha.id = alpha_id_map[alpha.id] 

70 

71 # The rename above only touches the Alpha objects (publish side); remap the 

72 # portfolio's frames with the same map too (see _remap_portfolio_ids). 

73 if alpha_id_map: 

74 _remap_portfolio_ids(portfolio, alpha_id_map) 

75 

76 # Freshness window for PortfolioExecutor's staleness guard. When unset, 

77 # default to twice the finest signal interval (daily portfolio → 2d) but 

78 # never tighter than the legacy 2h — so intraday strategies are unaffected 

79 # and slow strategies aren't spuriously rejected. 

80 if max_signal_age is None: 

81 finest = _finest_interval(alphas) 

82 max_signal_age = ( 

83 max(timedelta(hours=2), finest * 2) 

84 if finest is not None 

85 else timedelta(hours=2) 

86 ) 

87 

88 portfolio_executor = PortfolioExecutor( 

89 portfolio=portfolio, 

90 metric_stream=metric_stream, 

91 max_signal_age=max_signal_age, 

92 signal_namespace=signal_namespace, 

93 insert_prefix=insert_prefix, 

94 ) 

95 

96 if run_alphas: 

97 alpha_executor = AlphaExecutor( 

98 alphas=alphas, 

99 dataloader=dataloader, 

100 metric_stream=metric_stream, 

101 datasource_stream=datasource_stream, 

102 signal_namespace=signal_namespace, 

103 insert_prefix=insert_prefix, 

104 resync_interval=resync_interval, 

105 ) 

106 await asyncio.gather(alpha_executor.start(), portfolio_executor.start()) 

107 else: 

108 await portfolio_executor.start()