navi_vm/instance.rs
1use std::{collections::HashMap, sync::Arc};
2
3use gc_arena::Arena;
4use navi_program::Program;
5use serde::{Deserialize, Serialize};
6
7use crate::{
8 BarState, Candlestick, Error, ExecutionLimits, InputSessions, OutputMode, RunHandle, Series,
9 StrategyConfig, StrategyConfigOverride, StrategyState, SymbolInfo, TimeFrame,
10 aux_series::{AuxSeriesBuffer, AuxSeriesKey},
11 context::{ExecuteContext, SecurityCapture},
12 data_provider::InternalProvider,
13 events::{BarStartEvent, Event},
14 executor::Interrupt,
15 gc_serde::{SerializedObject, SerializedRawValue},
16 script_info::{ScriptInfo, ScriptType},
17 security::{CandlestickBuffer, SecurityLowerTfSubState, SecurityStreamKey, SecuritySubState},
18 snapshot::save::serialize_value_in_arena,
19 state::{ArenaType, State},
20 strategy::report::StrategyReport,
21 visuals::{Chart, DrawEvent},
22};
23
24const COLLECTOR_GRANULARITY: f64 = 1024.0;
25
26/// Determines how a candlestick is processed by [`Instance::execute()`].
27#[derive(Debug, Copy, Clone)]
28pub(crate) enum ExecuteMode {
29 /// Realtime bar update.
30 Realtime(Candlestick),
31 /// Confirmed (historical) bar execution.
32 Confirmed(Candlestick),
33 /// Confirms the last realtime bar.
34 Confirm,
35}
36
37/// The last information about the input candlesticks.
38/// Metadata about the last available input bar.
39#[derive(Debug, Copy, Clone, Serialize, Deserialize)]
40#[serde(rename_all = "camelCase")]
41pub struct LastInfo {
42 /// The index of the last candlestick (base 0).
43 pub index: usize,
44 /// The time of the last candlestick (in milliseconds since UNIX epoch).
45 pub time: i64,
46}
47
48impl LastInfo {
49 /// Creates a new `LastInfo`.
50 #[inline]
51 pub const fn new(index: usize, time: i64) -> Self {
52 Self { index, time }
53 }
54}
55
56/// An instance used to execute a compiled program.
57pub struct Instance {
58 pub(crate) program: Program,
59 pub(crate) locale: String,
60 pub(crate) arena: ArenaType,
61 pub(crate) timeframe: TimeFrame,
62 pub(crate) main_timeframe: TimeFrame,
63 pub(crate) symbol_info: SymbolInfo,
64 pub(crate) main_symbol: String,
65 pub(crate) candlesticks: Series<Candlestick>,
66 pub(crate) last_info: Option<LastInfo>,
67 pub(crate) bar_index: usize,
68 pub(crate) input_index: usize,
69 pub(crate) script_info: Arc<ScriptInfo>,
70 pub(crate) chart: Chart,
71 pub(crate) events: Vec<Event>,
72 pub(crate) input_sessions: InputSessions,
73 pub(crate) strategy_state: Option<Box<StrategyState>>,
74 /// Optional overrides for strategy execution parameters, applied on top of
75 /// the `strategy()` call arguments each time the state is (re)initialized.
76 pub(crate) strategy_config_override: Option<StrategyConfigOverride>,
77 /// Auxiliary data series buffers (currency rates, financials, etc.).
78 pub(crate) aux_series_buffers: HashMap<AuxSeriesKey, AuxSeriesBuffer>,
79 pub(crate) execution_limits: ExecutionLimits,
80 /// Provider for symbol metadata and `request.security()` data.
81 pub(crate) data_provider: Option<Box<dyn InternalProvider>>,
82 /// Shared candlestick buffers keyed by request stream identity.
83 pub(crate) candlestick_buffers: HashMap<SecurityStreamKey, CandlestickBuffer>,
84 /// Per-call-site execution states for `request.security()`.
85 pub(crate) security_sub_states: HashMap<usize, Box<SecuritySubState>>,
86 /// Per-call-site execution states for `request.security_lower_tf()`.
87 pub(crate) security_lower_tf_sub_states: HashMap<usize, Box<SecurityLowerTfSubState>>,
88 /// Whether the last executed bar has been confirmed.
89 ///
90 /// History bars are always confirmed immediately after execution.
91 /// A realtime bar becomes confirmed only after `RealtimeConfirmed` runs
92 /// (either explicitly via [`ExecuteMode::Confirm`] or automatically when
93 /// a new realtime timestamp arrives). This flag prevents the auto-confirm
94 /// step from firing — and using an already-advanced `bar_index` — after
95 /// a sequence of confirmed bars.
96 pub(crate) last_bar_confirmed: bool,
97 /// Pending security capture set by `execute_security_bar`.
98 ///
99 /// `do_execute` reads this instead of a hardcoded `None` so that
100 /// sub-instance executions triggered by `request.security()` can capture
101 /// their expression value without going through the full `execute()` path.
102 pub(crate) pending_security_capture: Option<SecurityCapture>,
103 /// Controls whether series graph data is written to the chart or emitted
104 /// as [`DrawEvent`](crate::DrawEvent) events.
105 pub(crate) output_mode: OutputMode,
106 /// Number of candlestick entries removed from the front of the series.
107 ///
108 /// Used to translate absolute `bar_index` values into buffer-relative
109 /// indices: `candlesticks[bar_index - candlestick_offset]`.
110 pub(crate) candlestick_offset: usize,
111 /// Script-level `max_bars_back` from `indicator(max_bars_back=N)` or
112 /// `strategy(max_bars_back=N)`. Capped by `execution_limits.max_bars_back`.
113 pub(crate) script_max_bars_back: Option<usize>,
114 /// Per-candlestick-series override set by `max_bars_back(close/open/…, N)`.
115 /// Multiple calls take the max. Capped by `execution_limits.max_bars_back`.
116 pub(crate) candlestick_max_bars_back: Option<usize>,
117 /// The maximum lookback depth actually observed at runtime for
118 /// candlestick buffer accesses. Starts at 1 (current bar only).
119 pub(crate) candlestick_observed_max_lookback: usize,
120 /// Global minimum lookback depth derived from history reference
121 /// instructions (`A[N]`). Updated even when `bar_index - N` underflows,
122 /// ensuring buffers are not truncated before lookback requirements are
123 /// observed by `get()`.
124 pub(crate) global_observed_max_lookback: usize,
125 /// Epoch-millisecond timestamp of the last confirmed bar, used as the
126 /// starting point when fetching candlesticks from the data provider.
127 ///
128 /// Set from the snapshot on restore; `0` for fresh instances (meaning
129 /// "start from the beginning").
130 pub(crate) last_bar_time: i64,
131 /// Maximum time to wait for realtime MTF sub-streams to catch up before
132 /// executing each realtime bar.
133 ///
134 /// Set via [`InstanceBuilder::with_mtf_wait`](crate::InstanceBuilder::with_mtf_wait).
135 /// Defaults to 3 seconds for new instances; `Duration::ZERO` disables
136 /// waiting.
137 pub(crate) mtf_wait: std::time::Duration,
138 /// Pool of reusable [`ExecState`] instances shared across all bars.
139 ///
140 /// Holds at most one entry per active `execute` call on the stack (i.e.
141 /// the main script plus one per `request.security` nesting level). Avoids
142 /// per-bar Vec allocations after warm-up.
143 pub(crate) exec_state_pool: Vec<crate::executor::ExecState>,
144 /// JIT-compiled native code (when `jit` feature is enabled).
145 #[cfg(feature = "jit")]
146 pub(crate) jit_code: Option<crate::jit::JitCode>,
147 /// Per-execution JIT state (stack, resume point, etc.).
148 #[cfg(feature = "jit")]
149 pub(crate) jit_state: Option<crate::jit::JitExecState>,
150}
151
152impl Instance {
153 /// Returns the compiled program held by this instance.
154 ///
155 /// The [`Program`] is reference-counted (`Arc` internally) so cloning it
156 /// is cheap. Used by providers to cache compiled programs across stream
157 /// restarts.
158 pub fn program(&self) -> &Program {
159 &self.program
160 }
161
162 /// Waits `mtf_wait` then drains all sub-stream buffers before executing a
163 /// realtime bar.
164 ///
165 /// Sleeping a fixed duration gives every `request.security()` sub-stream
166 /// time to deliver its latest data regardless of timeframe relationship
167 /// (higher, equal, or lower than the main chart). After the sleep, all
168 /// buffers are drained non-blockingly to pick up whatever arrived.
169 ///
170 /// Returns immediately with `Ok(())` if both maps are empty (no
171 /// `request.*` calls in the script) — zero overhead.
172 pub(crate) async fn pre_drain_sub_streams(
173 &mut self,
174 ) -> Result<(), crate::data_provider::InternalProviderError> {
175 if self.candlestick_buffers.is_empty() && self.aux_series_buffers.is_empty() {
176 return Ok(());
177 }
178 tokio::time::sleep(self.mtf_wait).await;
179 for buf in self.candlestick_buffers.values_mut() {
180 buf.drain_ready().await;
181 if let Some(e) = buf.take_stream_error() {
182 return Err(e);
183 }
184 }
185 for buf in self.aux_series_buffers.values_mut() {
186 buf.drain_ready().await;
187 }
188 Ok(())
189 }
190
191 fn trim_security_streams(&mut self) {
192 for stream in self.candlestick_buffers.values_mut() {
193 stream.trim_registered_consumers();
194 }
195 }
196
197 /// Resets the instance to its post-build state, as if no bars had been
198 /// executed.
199 ///
200 /// All runtime state — variable series, chart visuals, events, strategy
201 /// positions, security sub-states, and bar counters — is discarded and
202 /// re-initialised from scratch. The compiled program, symbol/timeframe
203 /// configuration, data provider, execution limits, and output mode are
204 /// preserved unchanged.
205 ///
206 /// > **Note:** Input values that were overridden via
207 /// > [`InstanceBuilder::with_input_value`](crate::InstanceBuilder::with_input_value) are **not** preserved across a
208 /// > reset; inputs revert to their script-default values.
209 ///
210 /// # Examples
211 ///
212 /// ```no_run
213 /// # use navi_vm::*;
214 /// # async fn example() -> Result<(), Error> {
215 /// # let provider = Vec::<Candlestick>::new();
216 /// # let instance = Instance::builder(provider, "indicator(\"\")\nplot(close)", TimeFrame::days(1), "NASDAQ:AAPL").build().await?;
217 /// let mut instance = instance
218 /// .run_to_end("NASDAQ:AAPL", TimeFrame::days(1))
219 /// .await?;
220 /// instance.reset();
221 /// // instance is now clean and ready to re-run
222 /// # Ok(())
223 /// # }
224 /// ```
225 pub fn reset(&mut self) {
226 let background_color = self.chart.background_color();
227
228 self.arena = Arena::new(|mc| State::new(mc, &self.program, Some(&self.script_info)));
229 self.candlesticks = Series::new();
230 self.bar_index = 0;
231 self.input_index = 0;
232 self.candlestick_offset = 0;
233 self.script_max_bars_back = self.script_info.script_type.max_bars_back();
234 self.candlestick_max_bars_back = None;
235 self.candlestick_observed_max_lookback = 1;
236 self.global_observed_max_lookback = 0;
237 self.last_bar_time = 0;
238 self.last_bar_confirmed = true;
239 self.pending_security_capture = None;
240
241 self.chart = Chart::default();
242 self.chart.set_background_color(background_color);
243 self.events.clear();
244
245 self.candlestick_buffers.clear();
246 self.security_sub_states.clear();
247 self.security_lower_tf_sub_states.clear();
248
249 self.strategy_state = if let ScriptType::Strategy(strategy) = &self.script_info.script_type
250 {
251 let mut config = StrategyConfig::new(strategy);
252 if let Some(ref ov) = self.strategy_config_override {
253 config = config.apply_override(ov);
254 }
255 let mintick = self.symbol_info.min_tick();
256 let min_contract = self.symbol_info.min_contract().unwrap_or(0.0);
257 Some(Box::new(StrategyState::new(config, mintick, min_contract)))
258 } else {
259 None
260 };
261
262 #[cfg(feature = "jit")]
263 if self.jit_code.is_some() {
264 self.jit_state = Some(crate::jit::JitExecState::new());
265 }
266 }
267
268 /// Returns the warnings generated during compilation, bundled with
269 /// source files so the caller can render them via `Warnings::display()`.
270 pub fn warnings(&self) -> navi_types::Warnings {
271 let diagnostics = self.program.warnings().to_vec();
272 let source_files = self
273 .program
274 .source_files()
275 .iter()
276 .enumerate()
277 .map(|(i, sf)| (i as u16, sf.clone()))
278 .collect();
279 navi_types::Warnings::new(diagnostics, source_files)
280 }
281
282 fn gc_collect(&mut self) {
283 if self.arena.metrics().allocation_debt() > COLLECTOR_GRANULARITY {
284 self.arena.collect_debt();
285 }
286 }
287
288 /// Runs a full garbage collection cycle on the GC arena.
289 ///
290 /// Normally the VM collects incrementally during execution. Call this
291 /// after a batch of bars to reclaim all unreachable GC objects at once.
292 pub fn gc_collect_all(&mut self) {
293 self.arena.finish_cycle();
294 }
295
296 /// Emits a `BarStart` event using the current candlestick.
297 fn emit_bar_start(&mut self, bar_state: BarState) {
298 let candlestick = self.candlesticks.last().copied().unwrap_or_default();
299 self.emit_bar_start_with_candlestick(candlestick, bar_state);
300 }
301
302 /// Emits a `BarStart` event with an explicit candlestick.
303 fn emit_bar_start_with_candlestick(&mut self, candlestick: Candlestick, bar_state: BarState) {
304 self.events.push(Event::BarStart(BarStartEvent {
305 bar_index: self.bar_index,
306 candlestick,
307 bar_state,
308 }));
309 }
310
311 /// Returns the effective max_bars_back for series variables, considering
312 /// the script-level default and the hard limit from ExecutionLimits.
313 fn effective_max_bars_back(&self) -> usize {
314 let hard_limit = self.execution_limits.max_bars_back;
315 self.script_max_bars_back
316 .map(|n| n.min(hard_limit))
317 .unwrap_or(hard_limit)
318 }
319
320 /// Returns the effective max_bars_back for the candlestick buffer.
321 fn effective_candlestick_max_bars_back(&self) -> usize {
322 let base = self.effective_max_bars_back();
323 self.candlestick_max_bars_back
324 .map(|n| n.min(self.execution_limits.max_bars_back))
325 .unwrap_or(base)
326 }
327
328 pub(crate) fn before_execute(&mut self, bar_state: BarState) {
329 match bar_state {
330 BarState::RealtimeNew => {
331 let max_bars_back = self.effective_max_bars_back();
332 let candlestick_max = self.effective_candlestick_max_bars_back();
333 self.arena.mutate_root(|_, state| {
334 for variable_value in state.variables.iter_mut() {
335 variable_value.append_new(max_bars_back);
336 }
337 });
338 self.candlesticks.append_new();
339 self.candlestick_offset += self.candlesticks.truncate(candlestick_max);
340 if self.output_mode == OutputMode::Chart {
341 self.chart.append_new();
342 } else if self.output_mode == OutputMode::Stream {
343 self.events.push(Event::Draw(DrawEvent::NewBar {
344 bar_index: self.bar_index,
345 }));
346 }
347 }
348 BarState::RealtimeUpdate => {
349 self.arena
350 .mutate_root(|_, state| state.do_rollback_actions(&mut self.chart));
351 }
352 BarState::History => {
353 let max_bars_back = self.effective_max_bars_back();
354 let candlestick_max = self.effective_candlestick_max_bars_back();
355 self.arena.mutate_root(|_, state| {
356 for variable_value in state.variables.iter_mut() {
357 variable_value.append_new(max_bars_back);
358 }
359 state.rollback_actions.clear();
360 });
361 self.candlesticks.append_new();
362 self.candlestick_offset += self.candlesticks.truncate(candlestick_max);
363 if self.output_mode == OutputMode::Chart {
364 self.chart.append_new();
365 } else if self.output_mode == OutputMode::Stream {
366 self.events.push(Event::Draw(DrawEvent::NewBar {
367 bar_index: self.bar_index,
368 }));
369 }
370 }
371 BarState::RealtimeConfirmed => self
372 .arena
373 .mutate_root(|_, state| state.do_rollback_actions(&mut self.chart)),
374 }
375 }
376
377 pub(crate) fn after_execute(&mut self, bar_state: BarState) {
378 // Shrink series buffers after execution on bars that grew them.
379 // This must happen AFTER execution so the script has access to
380 // all data during the current bar, but BEFORE the next bar to
381 // free memory from variables whose observed lookback is smaller
382 // than the global max_bars_back.
383 if matches!(bar_state, BarState::History | BarState::RealtimeNew) {
384 let max_bars_back = self.effective_max_bars_back();
385 let global_min = self.global_observed_max_lookback;
386 self.arena.mutate_root(|_, state| {
387 for variable_value in state.variables.iter_mut() {
388 variable_value.shrink_to_observed(max_bars_back, global_min);
389 }
390 });
391 let candlestick_max = self.effective_candlestick_max_bars_back();
392 let effective_candle = self.candlestick_observed_max_lookback.max(global_min);
393 let candlestick_cap = effective_candle.min(candlestick_max);
394 if self.candlesticks.len() > candlestick_cap {
395 self.candlestick_offset += self.candlesticks.truncate(candlestick_cap);
396 }
397 }
398
399 match bar_state {
400 BarState::History => {
401 self.bar_index += 1;
402 self.input_index += 1;
403 self.last_bar_confirmed = true;
404 }
405 BarState::RealtimeNew | BarState::RealtimeUpdate => {
406 self.input_index += 1;
407 self.last_bar_confirmed = false;
408 }
409 BarState::RealtimeConfirmed => {
410 self.bar_index += 1;
411 self.last_bar_confirmed = true;
412 }
413 }
414 }
415
416 fn process_strategy(&mut self) {
417 if let Some(strategy_state) = &mut self.strategy_state {
418 strategy_state.step(&self.candlesticks[self.bar_index - self.candlestick_offset]);
419 strategy_state.process_pending_orders(&mut self.chart, self.bar_index);
420 strategy_state.process_exit_orders(&mut self.chart, self.bar_index);
421 strategy_state.check_margin_call(&mut self.chart, self.bar_index);
422 strategy_state.step_update_trades();
423 strategy_state.update_equity_extremes();
424 strategy_state.record_equity_curve();
425 strategy_state.check_risk_limits(&mut self.chart, self.bar_index);
426 }
427 self.drain_strategy_events();
428 }
429
430 /// When `process_orders_on_close = true`, fills any pending orders placed
431 /// during `do_execute()` at the current bar's close price and updates
432 /// equity extremes accordingly.
433 fn process_strategy_on_close(&mut self) {
434 if let Some(strategy_state) = &mut self.strategy_state
435 && strategy_state.should_process_on_close()
436 {
437 strategy_state.process_pending_orders_on_close(&mut self.chart, self.bar_index);
438 strategy_state.update_equity_extremes();
439 strategy_state.check_risk_limits(&mut self.chart, self.bar_index);
440 strategy_state.clear_immediate_flag();
441 }
442 self.drain_strategy_events();
443 }
444
445 /// Drains strategy events from `StrategyState` and appends them to the
446 /// main event stream (only in Stream mode).
447 fn drain_strategy_events(&mut self) {
448 if self.output_mode != OutputMode::Stream {
449 return;
450 }
451 if let Some(strategy_state) = &mut self.strategy_state {
452 self.events.extend(strategy_state.drain_events());
453 }
454 }
455
456 /// When `calc_on_order_fills` is enabled, simulates intrabar ticks
457 /// (O, H, L, C or O, L, H, C) and re-executes the script whenever an
458 /// order fills at a tick. Each re-execution rolls back variable state
459 /// (like a realtime update) so the script sees the updated strategy
460 /// state. Up to 3 re-executions per bar (4 total including the initial).
461 async fn process_intrabar_fills(&mut self, bar_state: BarState) -> Result<(), Error> {
462 let should = self
463 .strategy_state
464 .as_ref()
465 .is_some_and(|s| s.should_calc_on_order_fills());
466 if !should {
467 return Ok(());
468 }
469
470 let tick_prices = self.strategy_state.as_ref().unwrap().intrabar_tick_prices();
471
472 // We allow up to 3 re-executions (ticks after the first one can
473 // trigger fills). The first tick (open) is included because the
474 // script may have placed new orders during the initial do_execute().
475 let max_reexecutions = 3;
476 let mut reexecutions = 0;
477
478 for &tick_price in &tick_prices {
479 if reexecutions >= max_reexecutions {
480 break;
481 }
482
483 let had_fill = self
484 .strategy_state
485 .as_mut()
486 .unwrap()
487 .process_orders_at_tick(&mut self.chart, self.bar_index, tick_price);
488 self.drain_strategy_events();
489
490 if had_fill {
491 reexecutions += 1;
492
493 // Roll back variable state (same mechanism as RealtimeUpdate)
494 // so the script re-executes with updated strategy state.
495 self.arena
496 .mutate_root(|_, state| state.do_rollback_actions(&mut self.chart));
497
498 let provider_ptr: Option<*const dyn InternalProvider> =
499 self.data_provider.as_deref().map(|p| p as *const _);
500 let provider = provider_ptr.map(|p| unsafe { &*p as &dyn InternalProvider });
501 self.do_execute(bar_state, provider, 0).await?;
502 self.drain_strategy_events();
503 }
504 }
505
506 Ok(())
507 }
508
509 /// Executes the compiled program body for the current bar state.
510 ///
511 /// Unlike [`execute`](Self::execute), this does **not** update
512 /// `bar_index`, run strategy logic, or collect GC debt. It is the raw
513 /// execution primitive used by both the public API and by sub-instance
514 /// executions for `request.security()`.
515 ///
516 /// The `security_provider` and `security_depth` parameters are forwarded
517 /// into the [`ExecuteContext`] so that nested `request.security()` calls
518 /// receive the correct provider and depth guard.
519 pub(crate) async fn do_execute(
520 &mut self,
521 bar_state: BarState,
522 security_provider: Option<&dyn InternalProvider>,
523 security_depth: usize,
524 ) -> Result<(), Error> {
525 let execution_limits = self.execution_limits;
526 let program = self.program.clone();
527 let security_capture = self.pending_security_capture.take();
528 let mut ctx = ExecuteContext {
529 program: &program,
530 locale: &self.locale,
531 arena: &mut self.arena,
532 candlesticks: &self.candlesticks,
533 last_info: self.last_info.as_ref(),
534 bar_state,
535 bar_index: self.bar_index,
536 candlestick_offset: self.candlestick_offset,
537 input_index: self.input_index,
538 partial_script_info: None,
539 script_info: Some(&self.script_info),
540 chart: &mut self.chart,
541 events: &mut self.events,
542 current_span: None,
543 module_stack: vec![],
544 timeframe: &self.timeframe,
545 main_timeframe: &self.main_timeframe,
546 symbol_info: &self.symbol_info,
547 main_symbol: &self.main_symbol,
548 input_sessions: self.input_sessions,
549 strategy_state: self.strategy_state.as_deref_mut(),
550 aux_series_buffers: &mut self.aux_series_buffers,
551 loop_iterations_remaining: execution_limits.max_loop_iterations_per_bar,
552 security_provider,
553 candlestick_buffers: &mut self.candlestick_buffers,
554 security_sub_states: &mut self.security_sub_states,
555 security_lower_tf_sub_states: &mut self.security_lower_tf_sub_states,
556 security_depth,
557 execution_limits,
558 security_capture,
559 output_mode: self.output_mode,
560 candlestick_max_bars_back_override: None,
561 candlestick_observed_max_lookback: &mut self.candlestick_observed_max_lookback,
562 global_observed_max_lookback: &mut self.global_observed_max_lookback,
563 exec_state_pool: &mut self.exec_state_pool,
564 };
565 #[cfg(feature = "jit")]
566 let res = if let (Some(jit_code), Some(jit_state)) = (&self.jit_code, &mut self.jit_state) {
567 crate::jit::execute_jit(jit_code, jit_state, &mut ctx).await
568 } else {
569 crate::executor::execute(program.ops(), &mut ctx).await
570 };
571 #[cfg(not(feature = "jit"))]
572 let res = crate::executor::execute(program.ops(), &mut ctx).await;
573 // Write back the captured result so `execute_security_bar` can read it.
574 self.pending_security_capture = ctx.security_capture.take();
575
576 // Write back candlestick max_bars_back override from max_bars_back().
577 if let Some(n) = ctx.candlestick_max_bars_back_override {
578 let current = self.candlestick_max_bars_back.unwrap_or(0);
579 self.candlestick_max_bars_back = Some(current.max(n));
580 }
581
582 match res {
583 Ok(_) => {
584 self.trim_security_streams();
585 Ok(())
586 }
587 Err(Interrupt::RuntimeError { error, backtrace }) => {
588 Err(Error::Exception(crate::error::build_runtime_error(
589 error.value,
590 error.span,
591 backtrace,
592 self.program.source_files(),
593 )))
594 }
595 }
596 }
597
598 fn check_input_candlestick(&self, candlestick: &Candlestick) -> Result<(), Error> {
599 if !self.input_sessions.allow(candlestick.trade_session) {
600 return Err(Error::SessionNotAllowed {
601 session: candlestick.trade_session,
602 });
603 }
604 Ok(())
605 }
606
607 /// Executes one step using the given mode and input values.
608 ///
609 /// For realtime execution this may perform a bar confirmation step when the
610 /// incoming candlestick timestamp advances.
611 ///
612 /// # Execution modes
613 ///
614 /// - [`ExecuteMode::Confirmed`] — Feed a confirmed (historical) bar. Each
615 /// call advances `bar_index` by one.
616 /// - [`ExecuteMode::Realtime`] — Feed a live bar. If the timestamp matches
617 /// the previous bar, it updates in-place; if the timestamp advances, the
618 /// previous bar is rolled over (auto-confirmed) and a new bar begins.
619 /// - [`ExecuteMode::Confirm`] — Explicitly confirm the current realtime bar
620 /// without providing new data.
621 ///
622 /// Prefer [`run()`](Self::run) over calling this method directly.
623 pub(crate) async fn execute(&mut self, mode: ExecuteMode) -> Result<(), Error> {
624 // Use a raw pointer to avoid holding an immutable borrow across mutable
625 // method calls. Safety: `data_provider` outlives all calls within this
626 // method and is not replaced during execution.
627 let provider_ptr: Option<*const dyn InternalProvider> =
628 self.data_provider.as_deref().map(|p| p as *const _);
629 let provider = |ptr: Option<*const dyn InternalProvider>| -> Option<&dyn InternalProvider> {
630 ptr.map(|p| unsafe { &*p })
631 };
632 match mode {
633 ExecuteMode::Realtime(candlestick) => {
634 self.check_input_candlestick(&candlestick)?;
635
636 let bar_state = match self.candlesticks.last() {
637 Some(last_candlestick) if last_candlestick.time == candlestick.time => {
638 if self.last_bar_confirmed {
639 return Err(Error::ConfirmedBarUpdate);
640 }
641 BarState::RealtimeUpdate
642 }
643 Some(_) => {
644 // Rollover: a later timestamp means the previous realtime
645 // bar is confirmed. Skip if already confirmed — history
646 // bars are pre-confirmed so this avoids using an
647 // out-of-bounds `bar_index` past the last history bar.
648 if !self.last_bar_confirmed {
649 let bar_state = BarState::RealtimeConfirmed;
650 self.emit_bar_start(bar_state);
651 self.before_execute(bar_state);
652 self.process_strategy();
653 self.do_execute(bar_state, provider(provider_ptr), 0)
654 .await?;
655 self.process_strategy_on_close();
656 self.after_execute(bar_state);
657 self.events.push(Event::BarEnd);
658 }
659 BarState::RealtimeNew
660 }
661 None => BarState::RealtimeNew,
662 };
663
664 // calculate for new bar
665 self.emit_bar_start_with_candlestick(candlestick, bar_state);
666 self.before_execute(bar_state);
667 self.process_strategy();
668 self.candlesticks.update(candlestick);
669 self.do_execute(bar_state, provider(provider_ptr), 0)
670 .await?;
671 self.process_strategy_on_close();
672 self.after_execute(bar_state);
673 self.events.push(Event::BarEnd);
674 }
675 ExecuteMode::Confirmed(candlestick) => {
676 // calculate for confirmed (historical) bar
677 self.check_input_candlestick(&candlestick)?;
678 let bar_state = BarState::History;
679 self.emit_bar_start_with_candlestick(candlestick, bar_state);
680 self.before_execute(bar_state);
681 self.candlesticks.update(candlestick);
682 self.process_strategy();
683 self.do_execute(bar_state, provider(provider_ptr), 0)
684 .await?;
685 self.process_intrabar_fills(bar_state).await?;
686 self.process_strategy_on_close();
687 self.after_execute(bar_state);
688 self.events.push(Event::BarEnd);
689 }
690 ExecuteMode::Confirm => {
691 // confirm the last realtime bar
692 let bar_state = BarState::RealtimeConfirmed;
693 self.emit_bar_start(bar_state);
694 self.before_execute(bar_state);
695 self.process_strategy();
696 self.do_execute(bar_state, provider(provider_ptr), 0)
697 .await?;
698 self.process_strategy_on_close();
699 self.after_execute(bar_state);
700 self.events.push(Event::BarEnd);
701 }
702 }
703
704 self.gc_collect();
705 Ok(())
706 }
707
708 /// Runs the script against the candlestick stream from the injected
709 /// [`DataProvider`](crate::DataProvider), returning a
710 /// [`RunHandle`](crate::RunHandle) for iterating events.
711 ///
712 /// The data provider's `from_time` is determined automatically:
713 /// - For fresh instances: `0` (beginning of available data).
714 /// - For restored instances: `last_bar_time + 1` (resume after the last
715 /// confirmed bar in the snapshot).
716 ///
717 /// - [`CandlestickItem::Bar`](crate::CandlestickItem::Bar) before
718 /// `HistoryEnd` — treated as a confirmed bar, advances `bar_index`.
719 /// - [`CandlestickItem::Bar`](crate::CandlestickItem::Bar) after
720 /// `HistoryEnd` — treated as a realtime bar; same-timestamp ticks become
721 /// `RealtimeUpdate` automatically; confirmed at stream end.
722 /// - [`CandlestickItem::HistoryEnd`](crate::CandlestickItem::HistoryEnd) —
723 /// emits [`Event::HistoryEnd`](crate::Event::HistoryEnd).
724 ///
725 /// Returns a handle that immediately yields `None` if no
726 /// [`DataProvider`](crate::DataProvider) was injected.
727 ///
728 /// The returned [`RunHandle`](crate::RunHandle) owns this instance and
729 /// implements [`Deref`](std::ops::Deref)/[`DerefMut`](std::ops::DerefMut)
730 /// targeting `Instance`, so instance methods such as `save_state()` can be
731 /// called directly through the handle while iterating events. Recover the
732 /// instance afterwards with
733 /// [`RunHandle::into_instance`](crate::RunHandle::into_instance).
734 pub fn run(self, symbol: &str, timeframe: TimeFrame) -> RunHandle {
735 let from_time = if self.last_bar_time > 0 {
736 self.last_bar_time + 1
737 } else {
738 0
739 };
740 RunHandle::new(self, symbol.to_owned(), from_time, timeframe)
741 }
742
743 /// Runs the script to completion, discarding all events.
744 ///
745 /// This is a convenience wrapper around [`run`](Self::run) for callers
746 /// that only need the final [`chart`](Self::chart) state.
747 ///
748 /// Because [`run`](Self::run) transfers ownership into the
749 /// [`RunHandle`](crate::RunHandle), this method also consumes the instance
750 /// and returns it after execution finishes.
751 pub async fn run_to_end(self, symbol: &str, timeframe: TimeFrame) -> Result<Self, Error> {
752 let mut handle = self.run(symbol, timeframe);
753 while let Some(result) = handle.next_event().await {
754 result?;
755 }
756 Ok(handle.into_instance())
757 }
758
759 /// Returns a [`PlotRowStream`](crate::PlotRowStream) for iterating
760 /// `plot()` outputs one bar at a time.
761 ///
762 /// Call [`PlotRowStream::columns`](crate::PlotRowStream::columns) to
763 /// obtain the column titles, then drive the stream with
764 /// [`PlotRowStream::next_row`](crate::PlotRowStream::next_row).
765 /// Only `plot()` outputs are included; other series such as `bgcolor()` or
766 /// `fill()` are ignored.
767 ///
768 /// # Panics
769 ///
770 /// Panics if the instance is not configured with
771 /// [`OutputMode::Stream`](crate::OutputMode::Stream).
772 pub fn plot_rows(self, symbol: &str, timeframe: TimeFrame) -> crate::PlotRowStream {
773 assert!(
774 self.output_mode == OutputMode::Stream,
775 "Instance::plot_rows requires OutputMode::Stream"
776 );
777
778 let mut columns = Vec::<Option<String>>::new();
779 let mut plot_indexes = vec![None::<usize>; self.script_info().series_declarations.len()];
780 for declaration in &self.script_info().series_declarations {
781 if let crate::script_info::SeriesDeclaration::Plot(plot) = declaration {
782 let index = columns.len();
783 columns.push(plot.title.clone());
784 plot_indexes[plot.id as usize] = Some(index);
785 }
786 }
787
788 let col_count = columns.len();
789 let handle = self.run(symbol, timeframe);
790
791 crate::plot_row_stream::PlotRowStream {
792 columns,
793 handle,
794 plot_indexes,
795 row: vec![None; col_count],
796 row_open: false,
797 needs_reset: false,
798 }
799 }
800
801 /// Returns a reference to the chart containing the visuals.
802 ///
803 /// The [`Chart`] holds all plots, lines, labels, boxes, and other visuals
804 /// produced by the script. Query it after calling
805 /// `execute()`.
806 ///
807 /// # Examples
808 ///
809 /// ```no_run
810 /// # use navi_vm::*;
811 /// # async fn example() {
812 /// # let source = "indicator(\"\")\nplot(close)";
813 /// # let provider = Vec::<Candlestick>::new();
814 /// # let instance = Instance::builder(provider, source, TimeFrame::days(1),
815 /// # "NASDAQ:AAPL").build().await.unwrap();
816 /// # let instance = instance.run_to_end("NASDAQ:AAPL", TimeFrame::days(1)).await.unwrap();
817 /// let chart = instance.chart();
818 ///
819 /// // Iterate over per-bar series graphs (plot, bgcolor, fill, etc.)
820 /// for (id, series_graph) in chart.series_graphs() {
821 /// if let Some(plot) = series_graph.as_plot() {
822 /// println!("Plot '{:?}': {} bars", plot.title, plot.series.len());
823 /// }
824 /// }
825 ///
826 /// // Iterate over non-series graphs (label, line, box, table, etc.)
827 /// for (id, graph) in chart.graphs() {
828 /// if let Some(label) = graph.as_label() {
829 /// println!("Label at ({}, {}): {:?}", label.x, label.y, label.text);
830 /// }
831 /// }
832 /// # }
833 /// ```
834 /// Returns the symbol info for the current chart symbol.
835 #[inline]
836 pub fn symbol_info(&self) -> &SymbolInfo {
837 &self.symbol_info
838 }
839
840 /// Returns the chart state produced by this instance.
841 #[inline]
842 pub fn chart(&self) -> &Chart {
843 &self.chart
844 }
845
846 /// Returns a reference to the script info (script type, inputs, etc.).
847 #[inline]
848 pub fn script_info(&self) -> &ScriptInfo {
849 self.script_info.as_ref()
850 }
851
852 /// Returns the total number of bars processed since execution started,
853 /// including bars that have been truncated from memory.
854 #[inline]
855 pub fn total_bars(&self) -> usize {
856 self.candlestick_offset + self.candlesticks.len()
857 }
858
859 /// Returns whether the current bar has been confirmed.
860 ///
861 /// Historical bars are always confirmed immediately after execution.
862 /// Realtime bars become confirmed only after `RealtimeConfirmed` runs,
863 /// either explicitly at stream end or automatically when a later realtime
864 /// timestamp arrives.
865 #[inline]
866 pub fn current_bar_confirmed(&self) -> bool {
867 self.last_bar_confirmed
868 }
869
870 /// Returns the strategy backtest report, or `None` if the script is
871 /// not a strategy.
872 pub fn strategy_report(&self) -> Option<StrategyReport> {
873 self.strategy_state.as_ref().map(|s| s.to_report())
874 }
875
876 /// Consumes the instance and returns the chart data.
877 ///
878 /// This avoids cloning the chart when the instance is no longer needed
879 /// (e.g. in a WASM playground that only needs the chart for rendering).
880 #[inline]
881 pub fn into_chart(self) -> Chart {
882 self.chart
883 }
884
885 /// Creates a sub-instance for `request.security()` expression evaluation.
886 ///
887 /// The sub-instance shares the compiled program but has its own GC arena,
888 /// candlestick series, and chart (for isolation). No strategy state.
889 #[allow(clippy::too_many_arguments)]
890 pub(crate) fn new_for_security(
891 program: Program,
892 locale: String,
893 timeframe: TimeFrame,
894 main_timeframe: TimeFrame,
895 symbol_info: SymbolInfo,
896 main_symbol: String,
897 script_info: Arc<ScriptInfo>,
898 execution_limits: ExecutionLimits,
899 ) -> Self {
900 let arena = Arena::new(|mc| State::new(mc, &program, Some(script_info.as_ref())));
901 Instance {
902 program,
903 locale,
904 arena,
905 timeframe,
906 main_timeframe,
907 symbol_info,
908 main_symbol,
909 candlesticks: Series::new(),
910 last_info: None,
911 bar_index: 0,
912 input_index: 0,
913 script_info,
914 chart: Chart::default(),
915 events: Vec::new(),
916 input_sessions: InputSessions::ALL,
917 strategy_state: None,
918 strategy_config_override: None,
919 aux_series_buffers: HashMap::new(),
920 execution_limits,
921 data_provider: None,
922 candlestick_buffers: HashMap::new(),
923 security_sub_states: HashMap::new(),
924 security_lower_tf_sub_states: HashMap::new(),
925 last_bar_confirmed: true,
926 pending_security_capture: None,
927
928 output_mode: OutputMode::default(),
929 candlestick_offset: 0,
930 script_max_bars_back: None,
931 candlestick_max_bars_back: None,
932 candlestick_observed_max_lookback: 1,
933 global_observed_max_lookback: 0,
934 last_bar_time: 0,
935 mtf_wait: std::time::Duration::ZERO,
936 exec_state_pool: Vec::new(),
937 #[cfg(feature = "jit")]
938 jit_code: None,
939 #[cfg(feature = "jit")]
940 jit_state: None,
941 }
942 }
943
944 /// Executes the full program body for the given MTF bar in capture mode,
945 /// returning the value produced by the `request.security` call identified
946 /// by `call_key`.
947 ///
948 /// Unlike the previous instruction-block approach, this runs the complete
949 /// program so that `var`/`varip` state is correctly accumulated before the
950 /// expression is evaluated. When the matching `request.security` call is
951 /// reached during execution, `request_security` stores the expression
952 /// value in `ExecuteContext::security_capture` and returns `NA`; all other
953 /// `request.security` calls also return `NA` to prevent recursion.
954 ///
955 /// `bar_state` must be computed by the caller (`request_security.rs`)
956 /// based on whether this is a confirmed, new-realtime, or
957 /// update-realtime execution.
958 ///
959 /// Returns the captured value (serialised for arena-independence) together
960 /// with its object table.
961 #[allow(clippy::too_many_arguments)]
962 pub(crate) async fn execute_security_bar(
963 &mut self,
964 bar: Candlestick,
965 bar_state: BarState,
966 call_key: usize,
967 return_type: &navi_program::TypeDescriptor,
968 security_depth: usize,
969 _execution_limits: ExecutionLimits,
970 provider: Option<&dyn InternalProvider>,
971 ) -> Result<(SerializedRawValue, Vec<SerializedObject>), Error> {
972 use crate::raw_value::RawValue;
973
974 self.before_execute(bar_state);
975 self.candlesticks.update(bar);
976
977 self.pending_security_capture = Some(SecurityCapture {
978 call_key,
979 result: None,
980 });
981
982 self.do_execute(bar_state, provider, security_depth).await?;
983
984 self.after_execute(bar_state);
985
986 // Retrieve the captured result written back by do_execute.
987 let captured = self.pending_security_capture.take().and_then(|c| c.result);
988
989 // `return_type` is `TypeDescriptor` for `instructions<T>`, so extract
990 // the inner type `T` to determine whether the result value is a
991 // reference type and to pass the correct descriptor to the serializer.
992 let inner_type = return_type.instructions_return_type().into_owned();
993 let is_ref = inner_type.is_reference_type();
994 let raw_value = captured.unwrap_or(RawValue::NA);
995 let serialized = serialize_value_in_arena(&mut self.arena, raw_value, is_ref, &inner_type);
996 Ok(serialized)
997 }
998}