(
&mut self,
event: StreamEvent,
)
| 283 | } |
| 284 | |
| 285 | pub fn on_event( |
| 286 | &mut self, |
| 287 | event: StreamEvent, |
| 288 | ) -> Result<EarlyWarningSnapshot, StreamingHpcError> { |
| 289 | validate_event(event)?; |
| 290 | let vpin = self.vpin_state.update(event.buy_volume, event.sell_volume)?; |
| 291 | let hhi = self.hhi_state.update(event.venue_id); |
| 292 | let normalized_risk_score = match (vpin, hhi) { |
| 293 | (Some(v), Some(h)) => { |
| 294 | Some(0.5 * (v / self.cfg.thresholds.vpin + h / self.cfg.thresholds.hhi)) |
| 295 | } |
| 296 | _ => None, |
| 297 | }; |
| 298 | let is_alert = match (vpin, hhi) { |
| 299 | (Some(v), Some(h)) => v >= self.cfg.thresholds.vpin && h >= self.cfg.thresholds.hhi, |
| 300 | _ => false, |
| 301 | }; |
| 302 | Ok(EarlyWarningSnapshot { |
| 303 | timestamp_ns: event.timestamp_ns, |
| 304 | price: event.price, |
| 305 | vpin, |
| 306 | hhi, |
| 307 | normalized_risk_score, |
| 308 | is_alert, |
| 309 | }) |
| 310 | } |
| 311 | } |
| 312 | |
| 313 | pub fn run_streaming_pipeline( |
no test coverage detected