use super::super::region_intersects_source_range;
use super::super::scripting::ScriptIndicatorRuntime;
use super::input::IndicatorInputSource;
use super::{
BaseSeriesSource, IndicatorConfig, IndicatorConfigError, IndicatorId, IndicatorOutputRef,
OutputValue,
};
use crate::chart::kline::{
IndicatorInstance, IndicatorInstanceId, IndicatorInstanceStore, SourceBindingProblem,
};
use indicator_engine::{IndicatorFrame, SeriesInputValue};
use rustc_hash::{FxHashMap, FxHashSet};
use std::collections::{BTreeMap, BTreeSet};
use std::ops::{ControlFlow, Range};
struct IndicatorRuntimeEntry {
id: IndicatorInstanceId,
kind: IndicatorId,
config: IndicatorConfig,
source_inputs: Vec<String>,
source_bindings: BTreeMap<String, IndicatorOutputRef>,
base_source_selections: BTreeMap<String, BaseSeriesSource>,
runtime: ScriptIndicatorRuntime,
output_indices: Option<OutputIndexMap>,
}
enum OutputIndexMap {
Prefix(usize),
Explicit(Vec<Option<usize>>),
}
impl OutputIndexMap {
fn output_index(&self, index: usize) -> Option<usize> {
match self {
Self::Prefix(prefix) => index.checked_sub(*prefix),
Self::Explicit(indices) => indices.get(index).copied().flatten(),
}
}
fn current_index(&self, index: usize) -> Option<usize> {
match self {
Self::Prefix(prefix) => index.checked_add(*prefix),
Self::Explicit(indices) => indices.iter().position(|previous| *previous == Some(index)),
}
}
}
impl IndicatorRuntimeEntry {
fn new(instance: &IndicatorInstance) -> Result<Option<Self>, IndicatorConfigError> {
let Some(registration) = super::catalog().get(&instance.kind) else {
return Ok(None);
};
let runtime = registration.create_runtime(instance.config.clone())?;
Ok(Some(Self {
id: instance.id,
kind: instance.kind.clone(),
config: instance.config.clone(),
source_inputs: registration
.definition
.source_inputs
.iter()
.map(|input| input.id.clone())
.collect(),
source_bindings: instance.source_bindings.clone(),
base_source_selections: instance.base_source_selections.clone(),
runtime,
output_indices: None,
}))
}
fn output_index(&self, index: usize) -> Option<usize> {
self.output_indices
.as_ref()
.map_or(Some(index), |indices| indices.output_index(index))
}
fn current_index(&self, index: usize) -> Option<usize> {
self.output_indices
.as_ref()
.map_or(Some(index), |indices| indices.current_index(index))
}
fn runtime_range_for_source_range(&self, source_range: &Range<usize>) -> Option<Range<usize>> {
match &self.output_indices {
None => Some(source_range.clone()),
Some(OutputIndexMap::Prefix(prefix)) => Some(
source_range.start.saturating_sub(*prefix)
..source_range.end.saturating_sub(*prefix),
),
Some(OutputIndexMap::Explicit(_)) => None,
}
}
}
enum IndicatorGraphDiagnostic {
AttachedHasNoNumericPlotLayer,
InvalidSourceBinding(SourceBindingProblem),
DependencyCycle,
}
impl std::fmt::Display for IndicatorGraphDiagnostic {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::AttachedHasNoNumericPlotLayer => {
formatter.write_str("attached indicator has no numeric plot layer")
}
Self::InvalidSourceBinding(problem) => write!(formatter, "{problem}"),
Self::DependencyCycle => formatter.write_str("indicator dependency cycle"),
}
}
}
#[derive(Default)]
pub struct IndicatorHost {
entries: Vec<IndicatorRuntimeEntry>,
entry_indices: FxHashMap<IndicatorInstanceId, usize>,
execution_order: Vec<IndicatorInstanceId>,
dependencies: FxHashMap<IndicatorInstanceId, Vec<IndicatorInstanceId>>,
dependents: FxHashMap<IndicatorInstanceId, Vec<IndicatorInstanceId>>,
graph_diagnostics: FxHashMap<IndicatorInstanceId, IndicatorGraphDiagnostic>,
runtime_diagnostics: FxHashMap<IndicatorInstanceId, IndicatorConfigError>,
blocked_nodes: FxHashSet<IndicatorInstanceId>,
}
impl IndicatorHost {
fn record_runtime_diagnostic(
&mut self,
instance: &IndicatorInstance,
error: IndicatorConfigError,
) {
log::error!(
"indicator {} (instance {:?}) runtime rebuild failure: {error}",
instance.kind.id(),
instance.id
);
self.runtime_diagnostics.insert(instance.id, error);
}
pub fn remap_outputs(&mut self, previous_indices: &[Option<usize>]) {
for entry in &mut self.entries {
entry.output_indices = Some(OutputIndexMap::Explicit(
previous_indices
.iter()
.map(|index| index.and_then(|index| entry.output_index(index)))
.collect(),
));
}
}
pub fn remap_outputs_prefix(&mut self, prefix: usize) {
if prefix == 0 {
return;
}
for entry in &mut self.entries {
let output_indices = match entry.output_indices.take() {
None => OutputIndexMap::Prefix(prefix),
Some(OutputIndexMap::Prefix(previous_prefix)) => {
OutputIndexMap::Prefix(previous_prefix.saturating_add(prefix))
}
Some(OutputIndexMap::Explicit(previous_indices)) => {
let mut indices =
Vec::with_capacity(prefix.saturating_add(previous_indices.len()));
indices.resize(prefix, None);
indices.extend(previous_indices);
OutputIndexMap::Explicit(indices)
}
};
entry.output_indices = Some(output_indices);
}
}
pub fn has_entries(&self, ids: &[IndicatorInstanceId]) -> bool {
ids.iter().all(|id| self.entry_indices.contains_key(id))
}
pub fn can_append_from(&self, ids: &[IndicatorInstanceId], start: usize) -> bool {
ids.iter().all(|id| {
self.entry_indices.get(id).is_some_and(|index| {
let entry = &self.entries[*index];
entry.output_indices.is_none()
&& entry
.runtime
.committed_index()
.map_or(start == 0, |index| index.checked_add(1) == Some(start))
})
})
}
pub fn dependent_ids(&self, roots: &[IndicatorInstanceId]) -> Vec<IndicatorInstanceId> {
let mut affected = roots.iter().copied().collect::<FxHashSet<_>>();
let mut pending = roots.to_vec();
while let Some(id) = pending.pop() {
for dependent in self.dependents.get(&id).into_iter().flatten().copied() {
if affected.insert(dependent) {
pending.push(dependent);
}
}
}
let mut result = Vec::new();
let mut returned = FxHashSet::default();
for id in roots
.iter()
.copied()
.chain(self.execution_order.iter().copied())
{
if affected.contains(&id) && returned.insert(id) {
result.push(id);
}
}
result
}
pub fn calculation_ids(&self, roots: &[IndicatorInstanceId]) -> Vec<IndicatorInstanceId> {
let mut included = roots.iter().copied().collect::<FxHashSet<_>>();
let mut discovered = roots.to_vec();
let mut pending = roots.to_vec();
while let Some(id) = pending.pop() {
for dependency in self.dependencies.get(&id).into_iter().flatten().copied() {
if included.insert(dependency) {
discovered.push(dependency);
pending.push(dependency);
}
}
}
let mut result = self
.execution_order
.iter()
.copied()
.filter(|id| included.contains(id))
.collect::<Vec<_>>();
for id in discovered {
if !result.contains(&id) {
result.push(id);
}
}
result
}
pub fn merge_rebuilt(
&mut self,
mut rebuilt: Self,
replaced_ids: &[IndicatorInstanceId],
active: &[IndicatorInstanceId],
instances: &IndicatorInstanceStore,
) {
let replaced = replaced_ids.iter().copied().collect::<FxHashSet<_>>();
self.entries.retain(|entry| !replaced.contains(&entry.id));
self.entries.append(&mut rebuilt.entries);
self.runtime_diagnostics
.retain(|id, _| !replaced.contains(id));
self.runtime_diagnostics.extend(rebuilt.runtime_diagnostics);
self.entries.sort_by_key(|entry| {
active
.iter()
.position(|id| *id == entry.id)
.unwrap_or(usize::MAX)
});
self.rebuild_entry_indices();
self.rebuild_graph(active, instances);
}
pub fn recalculate_interruptible(
&mut self,
active: &[IndicatorInstanceId],
instances: &IndicatorInstanceStore,
frames: &[IndicatorFrame],
mut control: impl FnMut() -> ControlFlow<()>,
) -> ControlFlow<()> {
let mut rebuilt = Self::default();
for instance in active.iter().filter_map(|id| instances.get(*id)) {
control()?;
match IndicatorRuntimeEntry::new(instance) {
Ok(Some(entry)) => rebuilt.entries.push(entry),
Ok(None) => {}
Err(error) => rebuilt.record_runtime_diagnostic(instance, error),
};
}
rebuilt.rebuild_entry_indices();
rebuilt.rebuild_graph(active, instances);
rebuilt.replay_closed_interruptible(frames, control)?;
*self = rebuilt;
ControlFlow::Continue(())
}
pub fn replay_closed_interruptible(
&mut self,
frames: &[IndicatorFrame],
mut control: impl FnMut() -> ControlFlow<()>,
) -> ControlFlow<()> {
for chunk in frames.chunks(64) {
for frame in chunk {
control()?;
for id in self.execution_order.clone() {
control()?;
self.evaluate_entry(id, frame, false);
}
}
std::thread::yield_now();
}
ControlFlow::Continue(())
}
pub fn set_active(
&mut self,
active: &[IndicatorInstanceId],
instances: &IndicatorInstanceStore,
) -> Vec<IndicatorInstanceId> {
let changed = active
.iter()
.copied()
.filter(|id| {
instances.get(*id).is_some_and(|instance| {
instance.kind.is_registered()
&& self.entry_indices.get(id).is_none_or(|index| {
let entry = &self.entries[*index];
entry.kind != instance.kind
|| entry.config != instance.config
|| entry.source_bindings != instance.source_bindings
|| entry.base_source_selections != instance.base_source_selections
})
})
})
.collect::<Vec<_>>();
self.entries.retain(|entry| {
active.contains(&entry.id)
&& !changed.contains(&entry.id)
&& instances
.get(entry.id)
.is_some_and(|instance| instance.kind.is_registered())
});
self.runtime_diagnostics.retain(|id, _| {
active.contains(id)
&& instances
.get(*id)
.is_some_and(|instance| instance.kind.is_registered())
});
self.entries.sort_by_key(|entry| {
active
.iter()
.position(|id| *id == entry.id)
.unwrap_or(usize::MAX)
});
self.rebuild_entry_indices();
self.rebuild_graph(active, instances);
changed
}
pub fn evaluate_closed(&mut self, frame: &IndicatorFrame) {
let order = self.execution_order.clone();
for id in order {
self.evaluate_entry(id, frame, false);
}
}
pub fn evaluate_closed_selected(
&mut self,
ids: &[IndicatorInstanceId],
frame: &IndicatorFrame,
) {
let selected = ids.iter().copied().collect::<FxHashSet<_>>();
self.evaluate_nodes(&selected, frame, false);
}
pub fn preview_open_selected(&mut self, ids: &[IndicatorInstanceId], frame: &IndicatorFrame) {
let selected = ids.iter().copied().collect::<FxHashSet<_>>();
self.evaluate_nodes(&selected, frame, true);
}
fn evaluate_nodes(
&mut self,
selected: &FxHashSet<IndicatorInstanceId>,
frame: &IndicatorFrame,
preview: bool,
) {
let order = self.execution_order.clone();
for id in order {
if selected.contains(&id) {
self.evaluate_entry(id, frame, preview);
}
}
}
fn evaluate_entry(
&mut self,
id: IndicatorInstanceId,
base_frame: &IndicatorFrame,
preview: bool,
) {
if self.blocked_nodes.contains(&id) {
return;
}
let Some(entry_index) = self.entry_indices.get(&id).copied() else {
return;
};
let frame = self.frame_for(entry_index, base_frame);
if let Some(entry) = self.entries.get_mut(entry_index) {
if preview {
entry.runtime.preview_open(&frame);
} else {
entry.runtime.evaluate_closed(&frame);
}
}
}
fn frame_for(&self, entry_index: usize, base_frame: &IndicatorFrame) -> IndicatorFrame {
let Some(entry) = self.entries.get(entry_index) else {
return base_frame.clone();
};
let values = entry
.source_inputs
.iter()
.map(|input_id| match entry.source_bindings.get(input_id) {
None => entry
.base_source_selections
.get(input_id)
.copied()
.map_or(SeriesInputValue::Default, SeriesInputValue::Base),
Some(_)
if !entry
.kind
.definition()
.source_inputs
.iter()
.any(|input| input.id == *input_id && input.allow_indicator) =>
{
SeriesInputValue::Value(None)
}
Some(reference) if reference.instance_id == entry.id => {
SeriesInputValue::Value(None)
}
Some(reference) => self
.entry_indices
.get(&reference.instance_id)
.and_then(|target_index| self.entries.get(*target_index))
.filter(|target| {
!self.blocked_nodes.contains(&target.id)
&& target
.kind
.definition()
.sourceable_numeric_output(&reference.channel_id)
.is_some()
})
.and_then(|_| {
self.output_value_for(
reference.instance_id,
&reference.channel_id,
base_frame.index,
)
})
.map_or(SeriesInputValue::Value(None), |value| {
SeriesInputValue::Value(value.as_number())
}),
})
.collect::<Vec<_>>();
let mut frame = base_frame.clone();
frame.series_inputs = values.into();
frame
}
fn rebuild_graph(
&mut self,
active_ids: &[IndicatorInstanceId],
instances: &IndicatorInstanceStore,
) {
self.execution_order.clear();
self.dependencies.clear();
self.dependents.clear();
self.graph_diagnostics.clear();
self.blocked_nodes.clear();
let ids = active_ids
.iter()
.copied()
.filter(|id| {
instances
.get(*id)
.is_some_and(|instance| instance.kind.is_registered())
})
.collect::<Vec<_>>();
let active = ids.iter().copied().collect::<BTreeSet<_>>();
for id in &ids {
let Some(instance) = instances.get(*id) else {
continue;
};
if !instance.source_bindings.is_empty()
&& !instance.kind.definition().has_numeric_plot_layers()
{
self.graph_diagnostics
.insert(*id, IndicatorGraphDiagnostic::AttachedHasNoNumericPlotLayer);
self.blocked_nodes.insert(*id);
continue;
}
for source_id in instance.source_bindings.keys() {
let target = match instances.dependency_target(instance, source_id, &active) {
Ok(Some(target)) => target,
Ok(None) => continue,
Err(problem) => {
self.graph_diagnostics
.entry(*id)
.or_insert(IndicatorGraphDiagnostic::InvalidSourceBinding(problem));
continue;
}
};
let dependencies = self.dependencies.entry(*id).or_default();
if !dependencies.contains(&target.id) {
dependencies.push(target.id);
self.dependents.entry(target.id).or_default().push(*id);
}
}
}
let mut indegree = ids
.iter()
.map(|id| (*id, self.dependencies.get(id).map_or(0, Vec::len)))
.collect::<FxHashMap<_, _>>();
let mut emitted = FxHashSet::default();
while let Some(id) = ids
.iter()
.copied()
.find(|id| !emitted.contains(id) && indegree.get(id).copied() == Some(0))
{
emitted.insert(id);
self.execution_order.push(id);
for dependent in self.dependents.get(&id).into_iter().flatten() {
if let Some(degree) = indegree.get_mut(dependent) {
*degree = degree.saturating_sub(1);
}
}
}
for id in ids {
if emitted.contains(&id) {
continue;
}
self.blocked_nodes.insert(id);
self.graph_diagnostics
.entry(id)
.or_insert(IndicatorGraphDiagnostic::DependencyCycle);
self.execution_order.push(id);
}
}
pub fn has_input_consumer(
&self,
source: IndicatorInputSource,
instances: &IndicatorInstanceStore,
) -> bool {
self.execution_order.iter().any(|id| {
instances
.get(*id)
.is_some_and(|instance| instance.kind.definition().uses(source))
})
}
pub fn output_value_for(
&self,
id: IndicatorInstanceId,
channel_id: &str,
index: usize,
) -> Option<OutputValue> {
self.entry_indices
.get(&id)
.and_then(|entry_index| self.entries.get(*entry_index))
.and_then(|entry| {
entry
.runtime
.output_value(channel_id, entry.output_index(index)?)
})
}
pub fn output_color_for(
&self,
id: IndicatorInstanceId,
channel_id: &str,
index: usize,
) -> Option<indicator_engine::Color> {
self.entry_indices
.get(&id)
.and_then(|entry_index| self.entries.get(*entry_index))
.and_then(|entry| {
entry
.runtime
.output_color(channel_id, entry.output_index(index)?)
})
}
pub fn marker_outputs_for(
&self,
id: IndicatorInstanceId,
index: usize,
) -> Vec<indicator_engine::MarkerOutput> {
self.entry_indices
.get(&id)
.and_then(|entry_index| self.entries.get(*entry_index))
.and_then(|entry| Some(entry.runtime.marker_outputs_at(entry.output_index(index)?)))
.unwrap_or_default()
}
pub fn segment_outputs_for(
&self,
id: IndicatorInstanceId,
source_range: Range<usize>,
mut intersects_viewport: impl FnMut(usize, usize) -> bool,
) -> Vec<indicator_engine::SegmentOutput> {
if source_range.is_empty() {
return Vec::new();
}
self.entry_indices
.get(&id)
.and_then(|entry_index| self.entries.get(*entry_index))
.map(|entry| {
let inverse = match &entry.output_indices {
Some(OutputIndexMap::Explicit(indices)) => Some(
indices
.iter()
.enumerate()
.filter_map(|(current, previous)| {
previous.map(|previous| (previous, current))
})
.collect::<FxHashMap<_, _>>(),
),
_ => None,
};
let current_index = |index| match &inverse {
Some(indices) => indices.get(&index).copied(),
None => entry.current_index(index),
};
entry
.runtime
.segment_outputs()
.filter_map(|segment| {
let start = current_index(segment.start_index)?;
let end = current_index(segment.end_index)?;
if start.min(end) >= source_range.end
|| start.max(end) < source_range.start
|| !intersects_viewport(start, end)
{
return None;
}
let mut segment = segment.clone();
segment.start_index = start;
segment.end_index = end;
Some(segment)
})
.collect()
})
.unwrap_or_default()
}
pub fn level_outputs_for(
&self,
id: IndicatorInstanceId,
) -> Vec<indicator_engine::HorizontalLevelOutput> {
self.entry_indices
.get(&id)
.and_then(|entry_index| self.entries.get(*entry_index))
.map(|entry| entry.runtime.level_outputs())
.unwrap_or_default()
}
pub fn ohlc_output_for(
&self,
id: IndicatorInstanceId,
channel_id: &str,
index: usize,
) -> Option<indicator_engine::OhlcOutput> {
self.entry_indices
.get(&id)
.and_then(|entry_index| self.entries.get(*entry_index))
.and_then(|entry| {
entry
.runtime
.ohlc_output(channel_id, entry.output_index(index)?)
})
}
pub fn range_output_for(
&self,
id: IndicatorInstanceId,
channel_id: &str,
index: usize,
) -> Option<indicator_engine::RangeOutput> {
self.entry_indices
.get(&id)
.and_then(|entry_index| self.entries.get(*entry_index))
.and_then(|entry| {
entry
.runtime
.range_output(channel_id, entry.output_index(index)?)
})
}
pub fn region_outputs_for_range(
&self,
id: IndicatorInstanceId,
source_range: Option<Range<usize>>,
) -> Vec<indicator_engine::RegionOutput> {
self.entry_indices
.get(&id)
.and_then(|entry_index| self.entries.get(*entry_index))
.map(|entry| {
let runtime_range = source_range
.as_ref()
.and_then(|range| entry.runtime_range_for_source_range(range));
entry
.runtime
.region_outputs_in_range(runtime_range)
.into_iter()
.filter_map(|mut region| {
region.left_index = entry.current_index(region.left_index)?;
if !region.extend_right {
region.right_index = entry.current_index(region.right_index)?;
} else if let Some(right_index) = entry.current_index(region.right_index) {
region.right_index = right_index;
}
if let Some(source_range) = &source_range {
region_intersects_source_range(®ion, source_range).then_some(region)
} else {
Some(region)
}
})
.collect()
})
.unwrap_or_default()
}
pub fn diagnostic_for(&self, id: IndicatorInstanceId) -> Option<String> {
if let Some(diagnostic) = self.graph_diagnostics.get(&id) {
return Some(diagnostic.to_string());
}
if let Some(diagnostic) = self.runtime_diagnostics.get(&id) {
return Some(format!("rebuild: {diagnostic}"));
}
self.entry_indices
.get(&id)
.and_then(|entry_index| self.entries.get(*entry_index))
.and_then(|entry| entry.runtime.diagnostic().map(str::to_owned))
}
fn rebuild_entry_indices(&mut self) {
self.entry_indices.clear();
self.entry_indices.extend(
self.entries
.iter()
.enumerate()
.map(|(index, entry)| (entry.id, index)),
);
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn source_index_remapping_supports_region_endpoints() {
let explicit = OutputIndexMap::Explicit(vec![Some(10), None, Some(12)]);
assert_eq!(explicit.current_index(10), Some(0));
assert_eq!(explicit.current_index(12), Some(2));
assert_eq!(explicit.current_index(11), None);
let prefixed = OutputIndexMap::Prefix(3);
assert_eq!(prefixed.current_index(7), Some(10));
}
#[test]
fn region_culling_keeps_intersecting_finite_and_right_extending_zones() {
let finite = |left, right| indicator_engine::RegionOutput {
id: "zones".to_owned(),
key: left as i64,
left_index: left,
right_index: right,
upper: 10.0,
lower: 5.0,
border_color: None,
fill_color: None,
border_color_slot: None,
fill_color_slot: None,
extend_right: false,
width: 1.0,
};
let extending = |left| indicator_engine::RegionOutput {
extend_right: true,
..finite(left, left)
};
let visible = 5..10;
assert!(!region_intersects_source_range(&finite(0, 4), &visible));
assert!(region_intersects_source_range(&finite(0, 5), &visible));
assert!(region_intersects_source_range(&finite(8, 12), &visible));
assert!(!region_intersects_source_range(&finite(10, 12), &visible));
assert!(region_intersects_source_range(&extending(0), &visible));
assert!(region_intersects_source_range(&extending(9), &visible));
assert!(!region_intersects_source_range(&extending(10), &visible));
}
#[test]
fn interrupted_replay_preserves_the_previous_host() {
let id = IndicatorInstanceId::default_for(&IndicatorId::parse("core.volume").unwrap());
let mut host = IndicatorHost {
execution_order: vec![id],
..IndicatorHost::default()
};
let frames = (0..130)
.map(|index| IndicatorFrame {
index,
time: index as i64,
basis: indicator_engine::TimeOrTick::Tick,
instrument: indicator_engine::InstrumentContext::default(),
chart: indicator_engine::ChartFrame {
open: Some(1.0),
high: Some(1.0),
low: Some(1.0),
close: Some(1.0),
volume: Some(1.0),
directional_volume_available: false,
buy_qty: 0.0,
sell_qty: 0.0,
trade: None,
},
open_interest: None,
series_inputs: std::sync::Arc::from([]),
})
.collect::<Vec<_>>();
let mut checks = 0;
assert!(
host.replay_closed_interruptible(&frames, || {
checks += 1;
if checks == 5 {
ControlFlow::Break(())
} else {
ControlFlow::Continue(())
}
})
.is_break()
);
assert_eq!(checks, 5);
assert!(
host.recalculate_interruptible(
&[],
&IndicatorInstanceStore::default(),
&frames,
|| ControlFlow::Break(())
)
.is_break()
);
assert_eq!(host.execution_order, vec![id]);
assert!(
host.recalculate_interruptible(
&[],
&IndicatorInstanceStore::default(),
&frames,
|| ControlFlow::Continue(())
)
.is_continue()
);
assert!(host.execution_order.is_empty());
}
}