Coverage for adrs/data/connector.py: 90%
40 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-29 05:07 +0800
1import json
2import time
4from decimal import Decimal
5from typing import Protocol, AsyncIterator, Callable, Awaitable
6from nats_client import Msg
8from adrs.types import Topic, Message
11class MetricStream(Protocol):
12 async def subscribe(
13 self, subject: str, callback: Callable[[Msg], Awaitable[None]] | None
14 ): ...
16 async def publish(self, subject: str, payload: bytes, **kwargs) -> None: ...
19class DatasourceStream(Protocol):
20 async def connect(self, topics: list[Topic]) -> AsyncIterator[Message] | None: ...
23DEFAULT_METRIC_NAMESPACE = "public_ts"
26class MetricBuilder:
27 """Publishes dashboard metrics under a namespace token (the JetStream root
28 the dashboard consumes, e.g. `public_ts`)."""
30 def __init__(
31 self,
32 metric_stream: MetricStream,
33 insert_prefix: str = DEFAULT_METRIC_NAMESPACE,
34 ):
35 self.metric_stream = metric_stream
36 self.insert_prefix = insert_prefix
38 def _metric_subject(self, name: str) -> str:
39 return f"{self.insert_prefix}.{name}"
41 @staticmethod
42 def _dedup_headers(msg_id: str) -> dict[str, str]:
43 """
44 Nats-Msg-Id for JetStream's server-side dedup: if two OMS replicas
45 (e.g. old + new pod overlapping during a rolling deploy) publish the
46 same logical record, the server collapses the second one instead of
47 it landing twice downstream.
48 """
49 return {"Nats-Msg-Id": msg_id}
51 async def create_alpha_trigger(self, alpha_id: str, topic: str | None = None):
52 return await self.metric_stream.publish(
53 self._metric_subject("alpha_trigger"),
54 json.dumps(
55 {
56 "alpha_id": alpha_id,
57 "topic": topic,
58 "timestamp": time.time_ns(),
59 }
60 ).encode(),
61 use_jetstream=True,
62 )
64 async def create_alpha_signal(
65 self,
66 alpha_id: str,
67 signal: float | Decimal | str,
68 timestamp: int,
69 ):
70 """`timestamp` (unix ns) is the time of the bar the signal was computed
71 on, so the dashboard series lines up with the signal's own candle rather
72 than the wall clock at insert time. Required — the caller always knows
73 which bar it emitted, and defaulting to now() silently misdates the
74 series whenever the emit lags the close."""
75 return await self.metric_stream.publish(
76 self._metric_subject("alpha_signal"),
77 json.dumps(
78 {
79 "alpha_id": alpha_id,
80 "signal": str(signal),
81 "timestamp": timestamp,
82 }
83 ).encode(),
84 use_jetstream=True,
85 )
87 async def create_portfolio_signal(
88 self, portfolio_id: str, signals: dict[str, Decimal]
89 ):
90 for asset, signal in signals.items():
91 await self.metric_stream.publish(
92 self._metric_subject("portfolio_signal"),
93 json.dumps(
94 {
95 "portfolio_id": portfolio_id,
96 "asset": asset,
97 "signal": str(signal),
98 "timestamp": time.time_ns(),
99 }
100 ).encode(),
101 use_jetstream=True,
102 )
104 async def create_alpha_alert(
105 self,
106 alpha_id: str,
107 title: str,
108 description: str = "",
109 priority: int = 3,
110 ):
111 return await self.metric_stream.publish(
112 self._metric_subject("alpha_alert"),
113 json.dumps(
114 {
115 "alpha_id": alpha_id,
116 "title": title,
117 "description": description,
118 "priority": priority,
119 "timestamp": time.time_ns(),
120 }
121 ).encode(),
122 use_jetstream=True,
123 )
125 async def create_portfolio_alert(
126 self,
127 portfolio_id: str,
128 title: str,
129 description: str = "",
130 priority: int = 3,
131 ):
132 return await self.metric_stream.publish(
133 self._metric_subject("portfolio_alert"),
134 json.dumps(
135 {
136 "portfolio_id": portfolio_id,
137 "title": title,
138 "description": description,
139 "priority": priority,
140 "timestamp": time.time_ns(),
141 }
142 ).encode(),
143 use_jetstream=True,
144 )
146 async def create_trade(
147 self,
148 oms_id: str,
149 package_id: str,
150 client_order_id: str,
151 asset: str,
152 symbol: str,
153 exchange: str,
154 executed_quantity: str,
155 executed_price: str,
156 executed_time: int,
157 start_quantity: str,
158 start_price: str,
159 start_time: int,
160 ):
161 return await self.metric_stream.publish(
162 self._metric_subject("trade"),
163 json.dumps(
164 {
165 "oms_id": oms_id,
166 "package_id": package_id,
167 "client_order_id": client_order_id,
168 "asset": asset,
169 "symbol": symbol,
170 "exchange": exchange,
171 "executed_quantity": executed_quantity,
172 "executed_price": executed_price,
173 "executed_time": executed_time,
174 "start_quantity": start_quantity,
175 "start_price": start_price,
176 "start_time": start_time,
177 }
178 ).encode(),
179 use_jetstream=True,
180 # No time-bucketing: a trade for one client_order_id is a
181 # one-time event, so the key never needs to expire.
182 headers=self._dedup_headers(f"trade:{oms_id}:{client_order_id}"),
183 )
185 async def create_position(
186 self,
187 oms_id: str,
188 asset: str,
189 symbol: str,
190 exchange: str,
191 quantity: str,
192 price: str,
193 updated_time: int,
194 ):
195 # Wall-clock minute bucket, not `updated_time`: the exchange only
196 # bumps updated_time when the position actually changes, so keying
197 # on it would collapse every legitimate heartbeat tick for a
198 # position that's simply flat - breaking staleness monitoring
199 # downstream. Bucketing on the publish minute only collapses
200 # genuine same-tick duplicates (e.g. two replicas overlapping during
201 # a rolling deploy) and still lets each new minute's tick through.
202 minute_bucket = int(time.time() // 60)
203 return await self.metric_stream.publish(
204 self._metric_subject("position"),
205 json.dumps(
206 {
207 "oms_id": oms_id,
208 "asset": asset,
209 "symbol": symbol,
210 "exchange": exchange,
211 "quantity": quantity,
212 "price": price,
213 "updated_time": updated_time,
214 "timestamp": time.time_ns(),
215 }
216 ).encode(),
217 use_jetstream=True,
218 headers=self._dedup_headers(f"position:{oms_id}:{symbol}:{minute_bucket}"),
219 )
221 async def create_equity(
222 self,
223 oms_id: str,
224 equity: str,
225 ):
226 # See create_position for why this buckets on the publish minute
227 # rather than any field in the payload.
228 minute_bucket = int(time.time() // 60)
229 return (
230 await self.metric_stream.publish(
231 self._metric_subject("equity"),
232 json.dumps(
233 {
234 "oms_id": oms_id,
235 "equity": equity,
236 "timestamp": time.time_ns(),
237 }
238 ).encode(),
239 use_jetstream=True,
240 headers=self._dedup_headers(f"equity:{oms_id}:{minute_bucket}"),
241 ),
242 )