Coverage for adrs/subjects.py: 88%

1"""NATS subject builders for inter-component routing. 

2 

3Two routing channels connect the live pipeline: 

4 

5 AlphaExecutor --alpha_signal--> PortfolioExecutor --portfolio_signal--> OMS 

6 

7Both roots optionally carry a `namespace` token (e.g. a user id) so that, on a 

8shared NATS server, a consumer only receives its own tenant's traffic. NATS 

9wildcards match whole tokens, so the namespace must be its own token 

10(`alpha_signal.<ns>.<id>`) — it cannot be a within-token prefix. 

11 

12These are separate from the dashboard-metric subjects MetricBuilder publishes 

13(those carry an `insert_prefix` namespace like `aegis_ts.*`). 

14""" 

15 

16from __future__ import annotations 

17 

18ALPHA_SIGNAL_ROOT = "alpha_signal" 

19PORTFOLIO_SIGNAL_ROOT = "portfolio_signal" 

20OMS_COMMAND_ROOT = "oms_command" 

21 

22 

23def alpha_signal_subject(alpha_id: str, namespace: str | None = None) -> str: 

24 """Subject an alpha publishes its signal on, for portfolio routing. 

25 

26 With a `namespace` the id becomes its own token 

27 (`alpha_signal.<namespace>.<alpha_id>`) so a PortfolioExecutor can subscribe 

28 to only its own namespace; without one, the legacy flat subject is kept.""" 

29 if namespace: 

30 return f"{ALPHA_SIGNAL_ROOT}.{namespace}.{alpha_id}" 

31 return f"{ALPHA_SIGNAL_ROOT}.{alpha_id}" 

32 

33 

34def alpha_signal_subscription(namespace: str | None = None) -> str: 

35 """Subscription a PortfolioExecutor uses to receive alpha signals. Scoped to 

36 `namespace` when set (`alpha_signal.<namespace>.>`), else `alpha_signal.*`.""" 

37 if namespace: 

38 return f"{ALPHA_SIGNAL_ROOT}.{namespace}.>" 

39 return f"{ALPHA_SIGNAL_ROOT}.*" 

40 

41 

42def parse_alpha_id(subject: str, namespace: str | None = None) -> str: 

43 """Recover the alpha id from a signal subject (inverse of 

44 `alpha_signal_subject`). With a namespace the id is everything after the 

45 `alpha_signal.<namespace>.` prefix (rejoined so ids containing dots survive); 

46 otherwise it's the single token after `alpha_signal.`.""" 

47 parts = subject.split(".") 

48 if namespace: 

49 return ".".join(parts[2:]) 

50 return parts[1] 

51 

52 

53def portfolio_signal_subject(portfolio_id: str, namespace: str | None = None) -> str: 

54 """Subject a PortfolioExecutor publishes its aggregated target on, which the 

55 OMS subscribes to by exact portfolio id. Namespaced symmetrically with 

56 `alpha_signal_subject` so both ends scope to the same tenant token.""" 

57 if namespace: 

58 return f"{PORTFOLIO_SIGNAL_ROOT}.{namespace}.{portfolio_id}" 

59 return f"{PORTFOLIO_SIGNAL_ROOT}.{portfolio_id}" 

60 

61 

62def oms_command_subject(oms_id: str, namespace: str | None = None) -> str: 

63 """Control-plane subject an operator publishes commands on for a specific 

64 OMS instance (e.g. rebalance). 

65 

66 This is a third purpose distinct from the routing roots above (flow = 

67 data) and the MetricBuilder dashboard subjects (aegis = metrics): here we 

68 deliver control. Namespaced symmetrically so a shared broker scopes per 

69 tenant.""" 

70 if namespace: 

71 return f"{OMS_COMMAND_ROOT}.{namespace}.{oms_id}" 

72 return f"{OMS_COMMAND_ROOT}.{oms_id}"