fin-primitives 2.12.0

//! Signal pipeline: applies multiple signals to each OHLCV bar and collects results.

use crate::error::FinError;
use crate::ohlcv::OhlcvBar;
use crate::signals::{Signal, SignalValue};
use rust_decimal::Decimal;
use std::collections::HashMap;

/// A named map of signal output values produced by a single [`SignalPipeline::update`] call.
///
/// Keys are the signal names as returned by [`crate::signals::Signal::name`].
/// Per-signal errors are stored alongside successful values so that one failing
/// indicator does not abort the entire pipeline.
pub struct SignalMap {
    values: HashMap<String, SignalValue>,
    errors: HashMap<String, FinError>,
}

impl SignalMap {
    /// Returns the signal value for `name`, or `None` if the name is not in this map.
    pub fn get(&self, name: &str) -> Option<&SignalValue> {
        self.values.get(name)
    }

    /// Returns the per-signal error for `name`, if that signal errored this cycle.
    pub fn error(&self, name: &str) -> Option<&FinError> {
        self.errors.get(name)
    }

    /// Returns `true` if any signal produced an error this update cycle.
    pub fn has_errors(&self) -> bool {
        !self.errors.is_empty()
    }

    /// Returns an iterator over the names of signals that errored this cycle.
    pub fn error_names(&self) -> impl Iterator<Item = &str> {
        self.errors.keys().map(String::as_str)
    }

    /// Returns an iterator over all `(name, value)` pairs in this map.
    pub fn values(&self) -> impl Iterator<Item = (&str, &SignalValue)> {
        self.values.iter().map(|(k, v)| (k.as_str(), v))
    }

    /// Returns all signal names in this map (order unspecified).
    pub fn names(&self) -> Vec<&str> {
        self.values.keys().map(String::as_str).collect()
    }

    /// Returns the number of signal entries in this map.
    pub fn len(&self) -> usize {
        self.values.len()
    }

    /// Returns `true` if this map contains no signal entries.
    pub fn is_empty(&self) -> bool {
        self.values.is_empty()
    }

    /// Returns the scalar value for `name`, or `default` if absent or unavailable.
    ///
    /// Combines `.get(name)` and `.as_decimal()` in one call. Useful in hot paths
    /// where downstream logic needs a numeric fallback rather than an `Option`.
    pub fn scalar_or(&self, name: &str, default: Decimal) -> Decimal {
        self.values
            .get(name)
            .and_then(SignalValue::as_decimal)
            .unwrap_or(default)
    }

    /// Returns the number of signals in this map that produced a `Scalar` (i.e. are available).
    pub fn count_available(&self) -> usize {
        self.values.values().filter(|v| matches!(v, SignalValue::Scalar(_))).count()
    }

    /// Returns `true` if every signal in this map has produced a `Scalar` value (none are `Unavailable`).
    pub fn all_available(&self) -> bool {
        !self.values.is_empty() && self.values.values().all(|v| matches!(v, SignalValue::Scalar(_)))
    }

    /// Returns `true` if at least one signal in this map is `Unavailable`.
    pub fn any_unavailable(&self) -> bool {
        self.values.values().any(|v| matches!(v, SignalValue::Unavailable))
    }

    /// Returns an iterator over `(name, Decimal)` for every signal that produced a `Scalar` value.
    pub fn scalars(&self) -> impl Iterator<Item = (&str, Decimal)> {
        self.values.iter().filter_map(|(k, v)| match v {
            SignalValue::Scalar(d) => Some((k.as_str(), *d)),
            SignalValue::Unavailable => None,
        })
    }

    /// Returns the scalar value for `name` if it exists and is ready, or `None` otherwise.
    pub fn get_scalar(&self, name: &str) -> Option<Decimal> {
        self.values.get(name)?.as_decimal()
    }

    /// Returns the `(name, value)` pair with the smallest scalar value, or `None` if no scalars.
    pub fn min_scalar(&self) -> Option<(&str, Decimal)> {
        self.scalars()
            .reduce(|acc, item| if item.1 < acc.1 { item } else { acc })
    }

    /// Returns the `(name, value)` pair with the largest scalar value, or `None` if no scalars.
    pub fn max_scalar(&self) -> Option<(&str, Decimal)> {
        self.scalars()
            .reduce(|acc, item| if item.1 > acc.1 { item } else { acc })
    }

    /// Returns the name of the signal with the highest scalar value, or `None` if no scalars.
    pub fn name_of_max(&self) -> Option<&str> {
        self.max_scalar().map(|(name, _)| name)
    }

    /// Returns the name of the signal with the lowest scalar value, or `None` if no scalars.
    pub fn name_of_min(&self) -> Option<&str> {
        self.min_scalar().map(|(name, _)| name)
    }

    /// Returns the sum of all ready scalar values in this map.
    pub fn sum_scalars(&self) -> Decimal {
        self.scalars().map(|(_, v)| v).sum()
    }

    /// Returns the arithmetic mean of all ready scalar values, or `None` if there are none.
    pub fn avg_scalar(&self) -> Option<Decimal> {
        let mut count = 0u32;
        let mut sum = Decimal::ZERO;
        for (_, v) in self.scalars() {
            sum += v;
            count += 1;
        }
        if count == 0 { None } else { Some(sum / Decimal::from(count)) }
    }

    /// Returns names of all signals whose scalar value strictly exceeds `threshold`.
    pub fn above_threshold(&self, threshold: Decimal) -> Vec<&str> {
        self.scalars()
            .filter(|(_, v)| *v > threshold)
            .map(|(name, _)| name)
            .collect()
    }

    /// Returns the sum of all scalar values strictly above `threshold`.
    pub fn sum_above(&self, threshold: Decimal) -> Decimal {
        self.scalars()
            .filter(|(_, v)| *v > threshold)
            .map(|(_, v)| v)
            .sum()
    }

    /// Returns the count of scalar values strictly above `threshold`.
    pub fn count_above(&self, threshold: Decimal) -> usize {
        self.scalars().filter(|(_, v)| *v > threshold).count()
    }

    /// Returns the count of scalar values strictly below `threshold`.
    pub fn count_below(&self, threshold: Decimal) -> usize {
        self.scalars().filter(|(_, v)| *v < threshold).count()
    }

    /// Returns a sorted `Vec` of names of all signals that have a ready scalar value.
    pub fn scalar_names(&self) -> Vec<&str> {
        let mut names: Vec<&str> = self.scalars().map(|(name, _)| name).collect();
        names.sort_unstable();
        names
    }

    /// Returns names of all signals whose scalar value is strictly below `threshold`.
    pub fn below_threshold(&self, threshold: Decimal) -> Vec<&str> {
        self.scalars()
            .filter(|(_, v)| *v < threshold)
            .map(|(name, _)| name)
            .collect()
    }

    /// Returns the percentile rank (0–100) of signal `name` among all current scalar values.
    ///
    /// Returns `None` if the signal is not found, is `Unavailable`, or there are no scalars.
    pub fn percentile_rank_of(&self, name: &str) -> Option<Decimal> {
        let target = self.get_scalar(name)?;
        let mut all: Vec<Decimal> = self.scalars().map(|(_, v)| v).collect();
        if all.is_empty() { return None; }
        all.sort();
        let below = all.iter().filter(|&&v| v < target).count();
        #[allow(clippy::cast_possible_truncation)]
        Some(Decimal::from(below as u32) / Decimal::from(all.len() as u32) * Decimal::ONE_HUNDRED)
    }

    /// Collects all ready scalar values into an owned `HashMap<String, Decimal>`.
    ///
    /// Useful when the caller needs an owned snapshot of all current signal values,
    /// e.g. to send across a channel or serialize.
    pub fn get_all_scalars(&self) -> std::collections::HashMap<String, Decimal> {
        self.scalars()
            .map(|(name, val)| (name.to_owned(), val))
            .collect()
    }

    /// Returns a `Vec` of names of all signals that errored in this update cycle.
    pub fn names_with_errors(&self) -> Vec<&str> {
        self.errors.keys().map(String::as_str).collect()
    }

    /// Returns the number of signals that produced a `Scalar` value this update cycle.
    pub fn count_ready(&self) -> usize {
        self.scalars().count()
    }

    /// Returns the arithmetic mean of all ready scalar values, or `None` if there are none.
    pub fn average_scalar(&self) -> Option<Decimal> {
        let mut count = 0usize;
        let mut sum = Decimal::ZERO;
        for (_, v) in self.scalars() {
            sum += v;
            count += 1;
        }
        if count == 0 {
            None
        } else {
            #[allow(clippy::cast_possible_truncation)]
            Some(sum / Decimal::from(count as u64))
        }
    }

    /// Returns a `HashMap` of signal names to scalar values for all signals whose scalar
    /// value is strictly greater than `threshold`.
    ///
    /// Signals that are `Unavailable` or whose value does not exceed the threshold are excluded.
    pub fn filter_scalars_above(&self, threshold: Decimal) -> std::collections::HashMap<&str, Decimal> {
        self.scalars()
            .filter(|(_, v)| *v > threshold)
            .collect()
    }

    /// Returns a `HashMap` of signal names to scalar values for all signals whose scalar
    /// value is strictly less than `threshold`.
    pub fn filter_scalars_below(&self, threshold: Decimal) -> std::collections::HashMap<&str, Decimal> {
        self.scalars()
            .filter(|(_, v)| *v < threshold)
            .collect()
    }

    /// Returns a `HashMap` of signal names to scalar values for signals whose value
    /// falls within `[lo, hi]` (inclusive on both ends).
    pub fn scalars_in_range(&self, lo: Decimal, hi: Decimal) -> std::collections::HashMap<&str, Decimal> {
        self.scalars()
            .filter(|(_, v)| *v >= lo && *v <= hi)
            .collect()
    }

    /// Returns the number of scalars strictly above `threshold`.
    pub fn above_count(&self, threshold: Decimal) -> usize {
        self.scalars().filter(|(_, v)| *v > threshold).count()
    }

    /// Returns the number of scalars strictly below `threshold`.
    pub fn below_count(&self, threshold: Decimal) -> usize {
        self.scalars().filter(|(_, v)| *v < threshold).count()
    }

    /// Returns the median scalar value across all available scalars.
    ///
    /// Returns `None` if there are no scalar values.
    pub fn median_scalar(&self) -> Option<Decimal> {
        let mut vals: Vec<Decimal> = self.scalars().map(|(_, v)| v).collect();
        if vals.is_empty() {
            return None;
        }
        vals.sort();
        let mid = vals.len() / 2;
        if vals.len() % 2 == 0 {
            Some((vals[mid - 1] + vals[mid]) / Decimal::TWO)
        } else {
            Some(vals[mid])
        }
    }

    /// Returns the population standard deviation of all scalar signal values.
    ///
    /// Returns `None` if there are fewer than 2 scalar values.
    pub fn std_dev(&self) -> Option<Decimal> {
        let vals: Vec<Decimal> = self.scalars().map(|(_, v)| v).collect();
        if vals.len() < 2 {
            return None;
        }
        let n = Decimal::from(vals.len() as u64);
        let mean = vals.iter().sum::<Decimal>() / n;
        let variance = vals.iter().map(|v| {
            let diff = v - mean;
            diff * diff
        }).sum::<Decimal>() / n;
        // Approximate sqrt via f64
        use rust_decimal::prelude::ToPrimitive;
        let var_f = variance.to_f64()?;
        Decimal::try_from(var_f.sqrt()).ok()
    }

    /// Returns the min-max normalized value of the named signal relative to all scalars.
    ///
    /// `(value - min_scalar) / (max_scalar - min_scalar)` — in `[0, 1]`.
    ///
    /// Returns `None` if the signal is not found, is `Unavailable`, or all scalars are equal.
    pub fn normalize_scalar(&self, name: &str) -> Option<Decimal> {
        let v = self.get_scalar(name)?;
        let (_, min) = self.min_scalar()?;
        let (_, max) = self.max_scalar()?;
        let range = max - min;
        if range.is_zero() {
            return None;
        }
        Some((v - min) / range)
    }

    /// Returns `max_scalar - min_scalar`. Returns `None` if there are no scalar values.
    pub fn range(&self) -> Option<Decimal> {
        let (_, max) = self.max_scalar()?;
        let (_, min) = self.min_scalar()?;
        Some(max - min)
    }

    /// Returns the top `n` signals by scalar value (descending), as `(name, value)` pairs.
    ///
    /// If fewer than `n` scalars exist, returns all available. Returns an empty vec if none.
    pub fn top_n(&self, n: usize) -> Vec<(&str, Decimal)> {
        let mut pairs: Vec<(&str, Decimal)> = self.scalars().collect();
        pairs.sort_by(|a, b| b.1.cmp(&a.1));
        pairs.truncate(n);
        pairs
    }

    /// Returns the weighted average of scalar values using the provided weight map.
    ///
    /// Only signals present in both the map and `weights` contribute. Signals with
    /// non-positive or zero weight are skipped. Returns `None` if total weight is zero.
    pub fn weighted_average(&self, weights: &std::collections::HashMap<&str, Decimal>) -> Option<Decimal> {
        let mut numerator = Decimal::ZERO;
        let mut denominator = Decimal::ZERO;
        for (name, value) in self.scalars() {
            if let Some(&w) = weights.get(name) {
                if w > Decimal::ZERO {
                    numerator += value * w;
                    denominator += w;
                }
            }
        }
        if denominator.is_zero() { None } else { Some(numerator / denominator) }
    }

    /// Returns the bottom `n` scalar signals by value, sorted ascending.
    pub fn bottom_n(&self, n: usize) -> Vec<(&str, Decimal)> {
        let mut items: Vec<(&str, Decimal)> = self.scalars().collect();
        items.sort_by(|a, b| a.1.cmp(&b.1));
        items.truncate(n);
        items
    }

    /// Sum of all scalar values strictly below `threshold`.
    pub fn sum_below(&self, threshold: Decimal) -> Decimal {
        self.scalars()
            .filter(|(_, v)| *v < threshold)
            .map(|(_, v)| v)
            .fold(Decimal::ZERO, |acc, v| acc + v)
    }

    /// Signal names sorted by scalar value descending.
    pub fn names_sorted_desc(&self) -> Vec<&str> {
        let mut items: Vec<(&str, Decimal)> = self.scalars().collect();
        items.sort_by(|a, b| b.1.cmp(&a.1));
        items.into_iter().map(|(n, _)| n).collect()
    }

    /// Z-score of each scalar relative to the set's mean and sample std dev.
    ///
    /// Returns an empty map if there are fewer than 2 scalar values or std dev is zero.
    pub fn z_scores(&self) -> HashMap<String, f64> {
        let pairs: Vec<(&str, Decimal)> = self.scalars().collect();
        if pairs.len() < 2 {
            return HashMap::new();
        }
        use rust_decimal::prelude::ToPrimitive;
        let floats: Vec<f64> = pairs.iter()
            .map(|(_, v)| v.to_f64().unwrap_or(0.0))
            .collect();
        let n = floats.len() as f64;
        let mean = floats.iter().sum::<f64>() / n;
        let variance = floats.iter().map(|&x| { let d = x - mean; d * d }).sum::<f64>() / (n - 1.0);
        let std_dev = variance.sqrt();
        if std_dev == 0.0 {
            return pairs.into_iter().map(|(name, _)| (name.to_string(), 0.0)).collect();
        }
        pairs.into_iter().zip(floats)
            .map(|((name, _), x)| (name.to_string(), (x - mean) / std_dev))
            .collect()
    }

    /// Names of the top `n` scalar signals by value (descending).
    pub fn top_n_names(&self, n: usize) -> Vec<&str> {
        let mut items: Vec<(&str, Decimal)> = self.scalars().collect();
        items.sort_by(|a, b| b.1.cmp(&a.1));
        items.truncate(n);
        items.into_iter().map(|(name, _)| name).collect()
    }

    /// Min-max normalize all scalar values to `[0, 1]`.
    ///
    /// Returns an empty map if there are fewer than 2 scalars or all values are equal.
    pub fn normalize_all(&self) -> HashMap<String, Decimal> {
        let pairs: Vec<(&str, Decimal)> = self.scalars().collect();
        if pairs.len() < 2 { return HashMap::new(); }
        let min = pairs.iter().map(|(_, v)| *v).min().unwrap_or(Decimal::ZERO);
        let max = pairs.iter().map(|(_, v)| *v).max().unwrap_or(Decimal::ZERO);
        let range = max - min;
        if range.is_zero() { return HashMap::new(); }
        pairs.into_iter()
            .map(|(name, v)| (name.to_string(), (v - min) / range))
            .collect()
    }

    /// Count of scalar values strictly greater than zero.
    pub fn count_positive(&self) -> usize {
        self.scalars().filter(|(_, v)| v > &Decimal::ZERO).count()
    }

    /// Population variance of all scalar values.
    ///
    /// Returns `None` if there are fewer than 2 scalars.
    pub fn variance(&self) -> Option<Decimal> {
        let vals: Vec<Decimal> = self.scalars().map(|(_, v)| v).collect();
        if vals.len() < 2 { return None; }
        let n = Decimal::from(vals.len() as u32);
        let mean = vals.iter().sum::<Decimal>() / n;
        let var = vals.iter().map(|v| { let d = v - mean; d * d }).sum::<Decimal>() / n;
        Some(var)
    }

    /// Sum of all scalar values strictly less than zero.
    pub fn sum_negative(&self) -> Decimal {
        self.scalars()
            .filter(|(_, v)| *v < Decimal::ZERO)
            .map(|(_, v)| v)
            .fold(Decimal::ZERO, |acc, v| acc + v)
    }

    /// Count of scalar values strictly less than zero.
    pub fn count_negative(&self) -> usize {
        self.scalars().filter(|(_, v)| *v < Decimal::ZERO).count()
    }

    /// Returns `true` if all scalar values are strictly positive.
    pub fn all_positive(&self) -> bool {
        let mut has_scalars = false;
        for (_, v) in self.scalars() {
            has_scalars = true;
            if v <= Decimal::ZERO { return false; }
        }
        has_scalars
    }

    /// Returns `true` if all scalar values are strictly negative.
    pub fn all_negative(&self) -> bool {
        let mut has_scalars = false;
        for (_, v) in self.scalars() {
            has_scalars = true;
            if v >= Decimal::ZERO { return false; }
        }
        has_scalars
    }

    /// Count of scalar values that are exactly zero.
    pub fn count_zero(&self) -> usize {
        self.scalars().filter(|(_, v)| v.is_zero()).count()
    }

    /// Returns `true` if the named signal has a scalar (non-Unavailable) value.
    pub fn has_scalar(&self, name: &str) -> bool {
        self.get_scalar(name).is_some()
    }

    /// All scalar values as a `HashMap<String, Decimal>` (Unavailable entries excluded).
    pub fn remove_unavailable(&self) -> HashMap<String, Decimal> {
        self.scalars().map(|(name, v)| (name.to_string(), v)).collect()
    }

    /// Ratio of two named scalar signals: `a / b`.
    ///
    /// Returns `None` if either signal is unavailable, not found, or `b` is zero.
    pub fn signal_ratio(&self, a: &str, b: &str) -> Option<Decimal> {
        let va = self.get_scalar(a)?;
        let vb = self.get_scalar(b)?;
        if vb.is_zero() { return None; }
        Some(va / vb)
    }

    /// Spread of scalar values as a percentage of the minimum: `(max - min) / min * 100`.
    ///
    /// Returns `None` if fewer than 2 scalars or min is zero.
    pub fn spread_pct(&self) -> Option<Decimal> {
        let (_, min) = self.min_scalar()?;
        let (_, max) = self.max_scalar()?;
        if min.is_zero() { return None; }
        Some((max - min) / min.abs() * Decimal::ONE_HUNDRED)
    }

    /// Weighted sum: `Σ value_i * weight_i` for signals found in `weights`.
    ///
    /// Returns `Decimal::ZERO` if no matching scalar signals are found.
    pub fn weighted_sum(&self, weights: &std::collections::HashMap<&str, Decimal>) -> Decimal {
        self.scalars()
            .filter_map(|(name, v)| weights.get(name).map(|&w| v * w))
            .fold(Decimal::ZERO, |acc, x| acc + x)
    }

    /// Multiply all scalar values by `factor`, returning a new `HashMap`.
    pub fn scale_all(&self, factor: Decimal) -> HashMap<String, Decimal> {
        self.scalars().map(|(name, v)| (name.to_string(), v * factor)).collect()
    }

    /// Shannon entropy (bits) of the scalar distribution, treating each value's share of the
    /// absolute total as a probability. Returns `None` if there are fewer than 2 scalars or the
    /// total is zero.
    pub fn entropy(&self) -> Option<f64> {
        use rust_decimal::prelude::ToPrimitive;
        let vals: Vec<f64> = self.scalars()
            .map(|(_, v)| v.abs().to_f64().unwrap_or(0.0))
            .collect();
        if vals.len() < 2 { return None; }
        let total: f64 = vals.iter().sum();
        if total == 0.0 { return None; }
        let entropy = vals.iter()
            .filter(|&&x| x > 0.0)
            .map(|&x| { let p = x / total; -p * p.log2() })
            .sum::<f64>();
        Some(entropy)
    }

    /// Gini coefficient of the scalar absolute values: 0 = perfect equality, 1 = maximum
    /// concentration. Returns `None` if there are fewer than 2 scalars.
    pub fn gini_coefficient(&self) -> Option<f64> {
        use rust_decimal::prelude::ToPrimitive;
        let mut vals: Vec<f64> = self.scalars()
            .map(|(_, v)| v.abs().to_f64().unwrap_or(0.0))
            .collect();
        if vals.len() < 2 { return None; }
        vals.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
        let n = vals.len() as f64;
        let total: f64 = vals.iter().sum();
        if total == 0.0 { return Some(0.0); }
        let numerator: f64 = vals.iter().enumerate()
            .map(|(i, &x)| (2.0 * (i as f64 + 1.0) - n - 1.0) * x)
            .sum();
        Some(numerator / (n * total))
    }
}

/// A pipeline that applies a sequence of signals to each incoming OHLCV bar.
pub struct SignalPipeline {
    signals: Vec<Box<dyn Signal>>,
}

impl SignalPipeline {
    /// Creates an empty `SignalPipeline`.
    pub fn new() -> Self {
        Self {
            signals: Vec::new(),
        }
    }

    /// Adds a signal to the pipeline (builder pattern).
    #[must_use]
    #[allow(clippy::should_implement_trait)]
    pub fn add(mut self, signal: impl Signal + 'static) -> Self {
        self.signals.push(Box::new(signal));
        self
    }

    /// Updates all signals with `bar` and returns a [`SignalMap`] of all results.
    ///
    /// Each signal is evaluated independently. Per-signal arithmetic errors are
    /// stored in the returned `SignalMap` (accessible via `map.error(name)`) rather
    /// than aborting the pipeline. Erroring signals appear as `Unavailable` in the map.
    pub fn update(&mut self, bar: &OhlcvBar) -> SignalMap {
        self.update_bar_input(&crate::signals::BarInput::from(bar))
    }

    /// Update all signals from a [`BarInput`] directly, without requiring an [`OhlcvBar`].
    ///
    /// Use this variant when working with synthetic or non-OHLCV data sources that
    /// already produce [`BarInput`] (e.g. custom tick aggregators, external feeds).
    pub fn update_bar_input(&mut self, bar: &crate::signals::BarInput) -> SignalMap {
        let mut values = HashMap::with_capacity(self.signals.len());
        let mut errors = HashMap::new();
        for signal in &mut self.signals {
            let name = signal.name().to_owned();
            match signal.update(bar) {
                Ok(value) => {
                    values.insert(name, value);
                }
                Err(e) => {
                    values.insert(name.clone(), SignalValue::Unavailable);
                    errors.insert(name, e);
                }
            }
        }
        SignalMap { values, errors }
    }

    /// Returns an iterator over the names of all registered signals in insertion order.
    pub fn signal_names(&self) -> impl Iterator<Item = &str> {
        self.signals.iter().map(|s| s.name())
    }

    /// Returns the total number of registered signals.
    pub fn signal_count(&self) -> usize {
        self.signals.len()
    }

    /// Returns the number of registered signals (alias for `signal_count`).
    pub fn len(&self) -> usize {
        self.signals.len()
    }

    /// Returns `true` if no signals are registered.
    pub fn is_empty(&self) -> bool {
        self.signals.is_empty()
    }

    /// Returns the number of signals that are currently ready.
    pub fn ready_count(&self) -> usize {
        self.signals.iter().filter(|s| s.is_ready()).count()
    }

    /// Returns the number of signals that are still warming up (not yet ready).
    pub fn not_ready_count(&self) -> usize {
        self.signals.iter().filter(|s| !s.is_ready()).count()
    }

    /// Returns `true` if every registered signal is ready to produce values.
    ///
    /// Useful as a gate before using pipeline output in production logic:
    /// only act on signals once `all_ready()` returns `true`.
    pub fn all_ready(&self) -> bool {
        !self.signals.is_empty() && self.signals.iter().all(|s| s.is_ready())
    }

    /// Returns an iterator over the names of signals that are currently ready.
    ///
    /// Useful for selectively reading only warmed-up signals from the output map.
    pub fn names_ready(&self) -> impl Iterator<Item = &str> {
        self.signals
            .iter()
            .filter(|s| s.is_ready())
            .map(|s| s.name())
    }

    /// Returns a reference to the signal with the given `name`, or `None` if not registered.
    pub fn get_signal(&self, name: &str) -> Option<&dyn Signal> {
        self.signals
            .iter()
            .find(|s| s.name() == name)
            .map(|s| s.as_ref())
    }

    /// Resets all registered signals to their initial (warm-up) state.
    ///
    /// Equivalent to calling `signal.reset()` on each registered signal.
    /// Useful for walk-forward backtesting without rebuilding the pipeline.
    pub fn reset(&mut self) {
        for signal in &mut self.signals {
            signal.reset();
        }
    }

    /// Updates all signals for every bar in `series`, returning one [`SignalMap`] per bar.
    ///
    /// The output vector has the same length as `series`. Useful for batch-processing
    /// a historical series in one call before inspecting the final state.
    pub fn update_series(&mut self, series: &crate::ohlcv::OhlcvSeries) -> Vec<SignalMap> {
        series.bars().iter().map(|bar| self.update(bar)).collect()
    }

    /// Runs all bars in `series` through the pipeline discarding outputs.
    ///
    /// Use this to warm up signal state before starting live updates,
    /// without allocating a `Vec<SignalMap>`.
    pub fn warm_up_bars(&mut self, series: &crate::ohlcv::OhlcvSeries) {
        for bar in series.bars() {
            self.update(bar);
        }
    }

    /// Resets all signals in the pipeline to their initial (un-warmed) state.
    pub fn reset_all(&mut self) {
        for signal in &mut self.signals {
            signal.reset();
        }
    }

    /// Returns the names of signals that are currently ready (producing values).
    ///
    /// A signal is "ready" when it has accumulated enough bars.
    pub fn ready_signal_names(&self) -> Vec<&str> {
        self.signals
            .iter()
            .filter(|s| s.is_ready())
            .map(|s| s.name())
            .collect()
    }

    /// Retains only the signals for which `predicate` returns `true`.
    ///
    /// Signals for which `predicate` returns `false` are dropped in place.
    /// Useful for culling a pipeline by period, name pattern, or readiness.
    pub fn retain<F>(&mut self, mut predicate: F)
    where
        F: FnMut(&dyn Signal) -> bool,
    {
        self.signals.retain(|s| predicate(s.as_ref()));
    }

    /// Returns the `(name, period)` pairs for every registered signal, in insertion order.
    pub fn signal_periods(&self) -> Vec<(&str, usize)> {
        self.signals.iter().map(|s| (s.name(), s.period())).collect()
    }

    /// Returns the names of all registered signals in insertion order.
    pub fn names(&self) -> Vec<&str> {
        self.signals.iter().map(|s| s.name()).collect()
    }

    /// Returns `(name, bars_remaining)` for each signal not yet ready.
    ///
    /// `bars_remaining` is an estimate: `signal.period().saturating_sub(ready_count)`.
    /// Returns an empty `Vec` once all signals are ready.
    pub fn warmup_periods_remaining(&self) -> Vec<(&str, usize)> {
        self.signals
            .iter()
            .filter(|s| !s.is_ready())
            .map(|s| (s.name(), s.period()))
            .collect()
    }

    /// Returns a sorted `Vec<&str>` of all registered signal names.
    pub fn names_sorted(&self) -> Vec<&str> {
        let mut names: Vec<&str> = self.signals.iter().map(|s| s.name()).collect();
        names.sort_unstable();
        names
    }

    /// Returns the maximum `period()` across all registered signals, or `0` if the pipeline is empty.
    pub fn longest_period(&self) -> usize {
        self.signals.iter().map(|s| s.period()).max().unwrap_or(0)
    }

    /// Returns the minimum `period()` across all registered signals, or `0` if the pipeline is empty.
    pub fn shortest_period(&self) -> usize {
        self.signals.iter().map(|s| s.period()).min().unwrap_or(0)
    }

    /// Removes the signal with the given `name` from the pipeline, returning `true` if found.
    ///
    /// If multiple signals share the same name (not recommended), only the first is removed.
    pub fn remove(&mut self, name: &str) -> bool {
        if let Some(pos) = self.signals.iter().position(|s| s.name() == name) {
            self.signals.remove(pos);
            true
        } else {
            false
        }
    }

    /// Returns the fraction of signals that are currently ready, in `[0.0, 1.0]`.
    ///
    /// Returns `0.0` if the pipeline is empty.
    pub fn pct_ready(&self) -> f64 {
        if self.signals.is_empty() {
            return 0.0;
        }
        self.ready_count() as f64 / self.signals.len() as f64
    }

}

impl Default for SignalPipeline {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::ohlcv::{OhlcvBar, OhlcvSeries};
    use crate::signals::indicators::{Ema, Rsi, Sma};
    use crate::types::{NanoTimestamp, Price, Quantity, Symbol};
    use rust_decimal_macros::dec;

    fn bar(close: &str) -> OhlcvBar {
        let p = Price::new(close.parse().unwrap()).unwrap();
        OhlcvBar {
            symbol: Symbol::new("X").unwrap(),
            open: p,
            high: p,
            low: p,
            close: p,
            volume: Quantity::zero(),
            ts_open: NanoTimestamp::new(0),
            ts_close: NanoTimestamp::new(1),
            tick_count: 1,
        }
    }

    #[test]
    fn test_signal_pipeline_update_all() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma3", 3).unwrap())
            .add(Ema::new("ema3", 3).unwrap())
            .add(Rsi::new("rsi3", 3).unwrap());

        let prices = ["100", "102", "104", "106"];
        let mut last_map = None;
        for p in &prices {
            last_map = Some(pipeline.update(&bar(p)));
        }
        let map = last_map.unwrap();
        assert!(map.get("sma3").is_some());
        assert!(map.get("ema3").is_some());
        assert!(map.get("rsi3").is_some());
        assert!(!map.has_errors());
        assert_eq!(pipeline.ready_count(), 3);
    }

    #[test]
    fn test_signal_pipeline_ready_count_zero_initially() {
        let pipeline = SignalPipeline::new()
            .add(Sma::new("sma5", 5).unwrap())
            .add(Ema::new("ema5", 5).unwrap());
        assert_eq!(pipeline.ready_count(), 0);
    }

    #[test]
    fn test_signal_pipeline_empty_map_for_empty_pipeline() {
        let mut pipeline = SignalPipeline::new();
        let map = pipeline.update(&bar("100"));
        assert!(map.get("any").is_none());
        assert!(!map.has_errors());
    }

    #[test]
    fn test_signal_pipeline_signal_names() {
        let pipeline = SignalPipeline::new()
            .add(Sma::new("sma3", 3).unwrap())
            .add(Ema::new("ema5", 5).unwrap());
        let names: Vec<&str> = pipeline.signal_names().collect();
        assert_eq!(names, vec!["sma3", "ema5"]);
    }

    #[test]
    fn test_signal_pipeline_signal_count() {
        let pipeline = SignalPipeline::new()
            .add(Sma::new("a", 2).unwrap())
            .add(Rsi::new("b", 3).unwrap());
        assert_eq!(pipeline.signal_count(), 2);
    }

    #[test]
    fn test_signal_pipeline_no_errors_on_normal_input() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma3", 3).unwrap())
            .add(Rsi::new("rsi3", 3).unwrap());
        for p in &["100", "101", "102", "103"] {
            let map = pipeline.update(&bar(p));
            assert!(!map.has_errors());
        }
    }

    #[test]
    fn test_signal_map_scalars_yields_ready_values() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma3", 3).unwrap())
            .add(Ema::new("ema3", 3).unwrap());
        pipeline.update(&bar("100"));
        pipeline.update(&bar("102"));
        let map = pipeline.update(&bar("104"));
        let scalars: Vec<_> = map.scalars().collect();
        assert_eq!(scalars.len(), 2);
        let names: Vec<_> = scalars.iter().map(|(k, _)| *k).collect();
        assert!(names.contains(&"sma3"));
        assert!(names.contains(&"ema3"));
    }

    #[test]
    fn test_signal_map_scalars_empty_before_warmup() {
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma5", 5).unwrap());
        let map = pipeline.update(&bar("100")); // only 1 bar
        let scalars: Vec<_> = map.scalars().collect();
        assert!(scalars.is_empty());
    }

    #[test]
    fn test_pipeline_get_signal_found() {
        let pipeline = SignalPipeline::new()
            .add(Sma::new("sma3", 3).unwrap())
            .add(Ema::new("ema5", 5).unwrap());
        assert!(pipeline.get_signal("sma3").is_some());
        assert_eq!(pipeline.get_signal("sma3").unwrap().name(), "sma3");
    }

    #[test]
    fn test_pipeline_get_signal_not_found() {
        let pipeline = SignalPipeline::new().add(Sma::new("sma3", 3).unwrap());
        assert!(pipeline.get_signal("nonexistent").is_none());
    }

    #[test]
    fn test_pipeline_get_signal_returns_correct_period() {
        let pipeline = SignalPipeline::new()
            .add(Sma::new("sma10", 10).unwrap())
            .add(Ema::new("ema20", 20).unwrap());
        assert_eq!(pipeline.get_signal("ema20").unwrap().period(), 20);
    }

    #[test]
    fn test_signal_map_get_scalar_returns_value_when_ready() {
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma3", 3).unwrap());
        pipeline.update(&bar("100"));
        pipeline.update(&bar("102"));
        let map = pipeline.update(&bar("104"));
        let v = map.get_scalar("sma3").unwrap();
        assert_eq!(v, dec!(102)); // (100 + 102 + 104) / 3
    }

    #[test]
    fn test_signal_map_get_scalar_returns_none_before_warmup() {
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma5", 5).unwrap());
        let map = pipeline.update(&bar("100"));
        assert!(map.get_scalar("sma5").is_none());
    }

    #[test]
    fn test_signal_map_get_scalar_missing_name() {
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma3", 3).unwrap());
        let map = pipeline.update(&bar("100"));
        assert!(map.get_scalar("nonexistent").is_none());
    }

    #[test]
    fn test_signal_map_min_max_scalar() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma2", 2).unwrap())
            .add(Sma::new("sma3", 3).unwrap());
        pipeline.update(&bar("100"));
        pipeline.update(&bar("102"));
        let map = pipeline.update(&bar("106"));
        // sma2 = (102+106)/2 = 104; sma3 = (100+102+106)/3 = 102.666...
        let (min_name, min_val) = map.min_scalar().unwrap();
        let (max_name, max_val) = map.max_scalar().unwrap();
        assert!(min_val < max_val);
        assert_ne!(min_name, max_name);
    }

    #[test]
    fn test_signal_map_min_max_scalar_empty() {
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma5", 5).unwrap());
        let map = pipeline.update(&bar("100"));
        assert!(map.min_scalar().is_none());
        assert!(map.max_scalar().is_none());
    }

    #[test]
    fn test_signal_map_sum_scalars() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma2", 2).unwrap())
            .add(Sma::new("sma3", 3).unwrap());
        pipeline.update(&bar("100"));
        pipeline.update(&bar("100"));
        let map = pipeline.update(&bar("100"));
        // Both SMAs = 100
        assert_eq!(map.sum_scalars(), dec!(200));
    }

    #[test]
    fn test_signal_map_sum_scalars_before_warmup() {
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma5", 5).unwrap());
        let map = pipeline.update(&bar("100"));
        assert_eq!(map.sum_scalars(), dec!(0));
    }

    #[test]
    fn test_signal_pipeline_update_series_length_matches() {
        use crate::ohlcv::{OhlcvBar, OhlcvSeries};
        let bars: Vec<OhlcvBar> = ["100", "102", "104", "106", "108"]
            .iter()
            .map(|p| bar(p))
            .collect();
        let series = OhlcvSeries::from_bars(bars).unwrap();
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma3", 3).unwrap());
        let maps = pipeline.update_series(&series);
        assert_eq!(maps.len(), 5);
    }

    #[test]
    fn test_signal_map_get_all_scalars_returns_owned_map() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma3", 3).unwrap())
            .add(Ema::new("ema3", 3).unwrap());
        pipeline.update(&bar("100"));
        pipeline.update(&bar("102"));
        let map = pipeline.update(&bar("104"));
        let scalars = map.get_all_scalars();
        assert_eq!(scalars.len(), 2);
        assert!(scalars.contains_key("sma3"));
        assert!(scalars.contains_key("ema3"));
    }

    #[test]
    fn test_signal_map_get_all_scalars_empty_before_warmup() {
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma5", 5).unwrap());
        let map = pipeline.update(&bar("100"));
        assert!(map.get_all_scalars().is_empty());
    }

    #[test]
    fn test_signal_pipeline_update_series_last_map_has_value() {
        use crate::ohlcv::{OhlcvBar, OhlcvSeries};
        let bars: Vec<OhlcvBar> = ["100", "100", "100", "100"]
            .iter()
            .map(|p| bar(p))
            .collect();
        let series = OhlcvSeries::from_bars(bars).unwrap();
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma3", 3).unwrap());
        let maps = pipeline.update_series(&series);
        assert_eq!(maps.last().unwrap().get_scalar("sma3"), Some(dec!(100)));
    }

    #[test]
    fn test_signal_map_names_with_errors_empty_when_no_errors() {
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma3", 3).unwrap());
        let map = pipeline.update(&bar("100"));
        assert!(map.names_with_errors().is_empty());
    }

    #[test]
    fn test_signal_pipeline_warm_up_bars_advances_state() {
        let bars: Vec<OhlcvBar> = ["100", "101", "102", "103", "104"]
            .iter()
            .map(|p| bar(p))
            .collect();
        let series = OhlcvSeries::from_bars(bars).unwrap();
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma3", 3).unwrap());
        pipeline.warm_up_bars(&series);
        // After 5 bars the SMA(3) should be ready; next update should yield a scalar
        let map = pipeline.update(&bar("100"));
        assert!(map.get_scalar("sma3").is_some());
    }

    #[test]
    fn test_signal_pipeline_warm_up_bars_fewer_bars_than_period() {
        let bars: Vec<OhlcvBar> = vec![bar("100")];
        let series = OhlcvSeries::from_bars(bars).unwrap();
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma3", 3).unwrap());
        // Should not panic even when series has fewer bars than period
        pipeline.warm_up_bars(&series);
        let map = pipeline.update(&bar("100"));
        // Only 2 bars total — still below period of 3
        assert!(map.get_scalar("sma3").is_none());
    }

    #[test]
    fn test_signal_pipeline_reset_all_clears_state() {
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma3", 3).unwrap());
        pipeline.update(&bar("100"));
        pipeline.update(&bar("101"));
        pipeline.update(&bar("102")); // now ready
        pipeline.reset_all();
        // After reset, signal should not be ready
        let map = pipeline.update(&bar("103"));
        assert!(map.get_scalar("sma3").is_none());
    }

    #[test]
    fn test_signal_pipeline_ready_signal_names_empty_before_warmup() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma3", 3).unwrap())
            .add(Sma::new("sma5", 5).unwrap());
        pipeline.update(&bar("100"));
        assert!(pipeline.ready_signal_names().is_empty());
    }

    #[test]
    fn test_signal_pipeline_ready_signal_names_after_warmup() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma3", 3).unwrap())
            .add(Sma::new("sma5", 5).unwrap());
        for i in 0..3 {
            let p = format!("{}", 100 + i);
            pipeline.update(&bar(&p));
        }
        let names = pipeline.ready_signal_names();
        assert_eq!(names, vec!["sma3"]);
    }

    #[test]
    fn test_signal_pipeline_remove_existing_signal() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma3", 3).unwrap())
            .add(Ema::new("ema5", 5).unwrap());
        assert!(pipeline.remove("sma3"));
        assert_eq!(pipeline.signal_count(), 1);
        assert!(pipeline.get_signal("sma3").is_none());
        assert!(pipeline.get_signal("ema5").is_some());
    }

    #[test]
    fn test_signal_pipeline_remove_nonexistent_returns_false() {
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma3", 3).unwrap());
        assert!(!pipeline.remove("nonexistent"));
        assert_eq!(pipeline.signal_count(), 1);
    }

    #[test]
    fn test_signal_pipeline_remove_then_update_only_remaining() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma3", 3).unwrap())
            .add(Ema::new("ema3", 3).unwrap());
        pipeline.remove("sma3");
        pipeline.update(&bar("100"));
        pipeline.update(&bar("102"));
        let map = pipeline.update(&bar("104"));
        assert!(map.get("sma3").is_none());
        assert!(map.get("ema3").is_some());
    }

    #[test]
    fn test_signal_map_filter_scalars_above_returns_matching() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma2", 2).unwrap())
            .add(Sma::new("sma3", 3).unwrap());
        pipeline.update(&bar("100"));
        pipeline.update(&bar("110"));
        let map = pipeline.update(&bar("120"));
        // sma2 = (110+120)/2 = 115; sma3 = (100+110+120)/3 = 110
        let above = map.filter_scalars_above(dec!(112));
        assert_eq!(above.len(), 1);
        assert!(above.contains_key("sma2"));
    }

    #[test]
    fn test_signal_map_filter_scalars_above_empty_when_none_qualify() {
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma3", 3).unwrap());
        pipeline.update(&bar("100"));
        pipeline.update(&bar("100"));
        let map = pipeline.update(&bar("100"));
        let above = map.filter_scalars_above(dec!(200));
        assert!(above.is_empty());
    }

    #[test]
    fn test_signal_map_filter_scalars_above_excludes_unavailable() {
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma5", 5).unwrap());
        let map = pipeline.update(&bar("100")); // not yet ready
        let above = map.filter_scalars_above(dec!(0));
        assert!(above.is_empty());
    }

    #[test]
    fn test_signal_map_count_ready_zero_before_warmup() {
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma5", 5).unwrap());
        let map = pipeline.update(&bar("100"));
        assert_eq!(map.count_ready(), 0);
    }

    #[test]
    fn test_signal_map_count_ready_after_warmup() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma3", 3).unwrap())
            .add(Ema::new("ema3", 3).unwrap());
        pipeline.update(&bar("100"));
        pipeline.update(&bar("101"));
        let map = pipeline.update(&bar("102"));
        assert_eq!(map.count_ready(), 2);
    }

    #[test]
    fn test_signal_map_average_scalar_returns_none_when_empty() {
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma5", 5).unwrap());
        let map = pipeline.update(&bar("100"));
        assert!(map.average_scalar().is_none());
    }

    #[test]
    fn test_signal_map_average_scalar_single_value() {
        let mut pipeline = SignalPipeline::new().add(Sma::new("sma3", 3).unwrap());
        pipeline.update(&bar("100"));
        pipeline.update(&bar("100"));
        let map = pipeline.update(&bar("100"));
        assert_eq!(map.average_scalar(), Some(dec!(100)));
    }

    #[test]
    fn test_signal_map_average_scalar_multiple_values() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma2", 2).unwrap())
            .add(Sma::new("sma3", 3).unwrap());
        pipeline.update(&bar("100"));
        pipeline.update(&bar("100"));
        let map = pipeline.update(&bar("100")); // both SMAs = 100
        assert_eq!(map.average_scalar(), Some(dec!(100)));
    }

    #[test]
    fn test_signal_pipeline_retain_by_period() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma3", 3).unwrap())
            .add(Sma::new("sma5", 5).unwrap())
            .add(Ema::new("ema10", 10).unwrap());
        pipeline.retain(|s| s.period() <= 5);
        assert_eq!(pipeline.signal_count(), 2);
        assert!(pipeline.get_signal("ema10").is_none());
    }

    #[test]
    fn test_signal_pipeline_retain_all_pass() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma3", 3).unwrap())
            .add(Ema::new("ema5", 5).unwrap());
        pipeline.retain(|_| true);
        assert_eq!(pipeline.signal_count(), 2);
    }

    #[test]
    fn test_signal_pipeline_retain_none_pass() {
        let mut pipeline = SignalPipeline::new()
            .add(Sma::new("sma3", 3).unwrap())
            .add(Ema::new("ema5", 5).unwrap());
        pipeline.retain(|_| false);
        assert_eq!(pipeline.signal_count(), 0);
    }

    #[test]
    fn test_signal_pipeline_signal_periods() {
        let pipeline = SignalPipeline::new()
            .add(Sma::new("sma3", 3).unwrap())
            .add(Ema::new("ema5", 5).unwrap())
            .add(Rsi::new("rsi14", 14).unwrap());
        let periods = pipeline.signal_periods();
        assert_eq!(periods.len(), 3);
        assert_eq!(periods[0], ("sma3", 3));
        assert_eq!(periods[1], ("ema5", 5));
        assert_eq!(periods[2], ("rsi14", 14));
    }

    #[test]
    fn test_signal_pipeline_signal_periods_empty() {
        let pipeline = SignalPipeline::new();
        assert!(pipeline.signal_periods().is_empty());
    }

    #[test]
    fn test_signal_pipeline_names_sorted() {
        let pipeline = SignalPipeline::new()
            .add(Sma::new("zzz", 3).unwrap())
            .add(Ema::new("aaa", 5).unwrap())
            .add(Rsi::new("mmm", 7).unwrap());
        let names = pipeline.names_sorted();
        assert_eq!(names, vec!["aaa", "mmm", "zzz"]);
    }

    #[test]
    fn test_signal_pipeline_longest_shortest_period() {
        let pipeline = SignalPipeline::new()
            .add(Sma::new("s3", 3).unwrap())
            .add(Ema::new("e10", 10).unwrap())
            .add(Rsi::new("r7", 7).unwrap());
        assert_eq!(pipeline.longest_period(), 10);
        assert_eq!(pipeline.shortest_period(), 3);
    }

    #[test]
    fn test_signal_pipeline_longest_shortest_empty() {
        let pipeline = SignalPipeline::new();
        assert_eq!(pipeline.longest_period(), 0);
        assert_eq!(pipeline.shortest_period(), 0);
    }

    #[test]
    fn test_signal_pipeline_signal_count_v2() {
        let p = SignalPipeline::new()
            .add(Sma::new("s3", 3).unwrap())
            .add(Ema::new("e5", 5).unwrap());
        assert_eq!(p.signal_count(), 2);
        let empty = SignalPipeline::new();
        assert_eq!(empty.signal_count(), 0);
    }

    #[test]
    fn test_signal_pipeline_reset_all() {
        let mut p = SignalPipeline::new().add(Sma::new("s3", 3).unwrap());
        for price in &["100", "101", "102"] {
            p.update(&bar(price));
        }
        assert_eq!(p.ready_count(), 1);
        p.reset_all();
        // After reset, the SMA needs 3 bars again
        let map = p.update(&bar("100"));
        assert!(matches!(map.get("s3"), Some(crate::signals::SignalValue::Unavailable)));
    }

    #[test]
    fn test_signal_map_std_dev_two_values() {
        let mut p = SignalPipeline::new()
            .add(Sma::new("s1", 1).unwrap())
            .add(Sma::new("s2", 1).unwrap());
        // Feed same bar to both — but s1 and s2 are different signals tracking the same close
        // We need different values; use a trick: compute manually
        // Actually both SMA(1) return the same close. Use a different approach.
        // We can't easily create a SignalMap with different values through the pipeline
        // since both SMA(1)s produce the same close. Just verify it returns Some for 2+ values.
        let map = p.update(&bar("100"));
        // both s1 and s2 are SMA(1), both return 100 => std_dev = 0, but Some
        // (population std dev of identical values is 0)
        let sd = map.std_dev();
        assert!(sd.is_some(), "expected Some for 2 scalar values");
    }

    #[test]
    fn test_signal_map_std_dev_single_value() {
        let mut p = SignalPipeline::new().add(Sma::new("s1", 1).unwrap());
        let map = p.update(&bar("100"));
        assert!(map.std_dev().is_none(), "std_dev needs at least 2 values");
    }

    #[test]
    fn test_signal_map_normalize_scalar_basic() {
        let mut p = SignalPipeline::new()
            .add(Sma::new("lo", 1).unwrap())
            .add(Sma::new("hi", 1).unwrap());
        // Both are SMA(1), so both = 100. Range is zero → None
        let map = p.update(&bar("100"));
        assert!(map.normalize_scalar("lo").is_none(), "zero range → None");
    }

    #[test]
    fn test_signal_map_normalize_scalar_not_found() {
        let mut p = SignalPipeline::new().add(Sma::new("s1", 1).unwrap());
        let map = p.update(&bar("100"));
        assert!(map.normalize_scalar("nonexistent").is_none());
    }
}