//! Causal, typed indicator/event expressions shared by research and trading.
use chrono::{DateTime, FixedOffset};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::collections::BTreeMap;
use ta_lib::{
    Core,
    abstract_api::{self, InputType, OptInputType, OutputType},
};

pub const CONTRACT: &str = "fidc_factor_event_expression_v1";
pub const TA_REV: &str = "dd5a90259a3f9e04e2da9f38bf0719a841b40108";

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum Expr {
    Number {
        value: f64,
    },
    Field {
        name: String,
    },
    Indicator {
        name: String,
        #[serde(default)]
        inputs: Vec<Expr>,
        #[serde(default)]
        parameters: BTreeMap<String, Value>,
        #[serde(default)]
        output: usize,
    },
    Operator {
        name: String,
        args: Vec<Expr>,
        #[serde(default)]
        window: Option<usize>,
    },
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Frame {
    pub symbol: String,
    pub frequency: String,
    pub decision_at: DateTime<FixedOffset>,
    pub timestamps: Vec<DateTime<FixedOffset>>,
    pub available_at: Vec<DateTime<FixedOffset>>,
    pub fields: BTreeMap<String, Vec<Option<f64>>>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ValueType {
    Number,
    Boolean,
}

#[derive(Debug, Clone, Serialize)]
pub struct Series {
    pub value_type: ValueType,
    pub values: Vec<Option<f64>>,
}

const OPERATORS: &[&str] = &[
    "GT",
    "GTE",
    "LT",
    "LTE",
    "EQ",
    "NEQ",
    "BETWEEN",
    "OUTSIDE",
    "CROSS_ABOVE",
    "CROSS_BELOW",
    "BREAK_ABOVE",
    "BREAK_BELOW",
    "BREAK_HIGH",
    "BREAK_LOW",
    "CHANGE",
    "DIFF",
    "DELTA",
    "PCT_CHANGE",
    "LOG_RETURN",
    "RISING",
    "FALLING",
    "NON_DECREASING",
    "NON_INCREASING",
    "TURN_UP",
    "TURN_DOWN",
    "BOTTOM_REVERSAL",
    "TOP_REVERSAL",
    "SLOPE",
    "SLOPE_CHANGE",
    "ACCELERATION",
    "HHV",
    "LLV",
    "ARGMAX",
    "ARGMIN",
    "DISTANCE_TO_HIGH",
    "DISTANCE_TO_LOW",
    "NEW_HIGH",
    "NEW_LOW",
    "NEAR_HIGH",
    "NEAR_LOW",
    "BULLISH_DIVERGENCE",
    "BEARISH_DIVERGENCE",
    "ZSCORE",
    "MINMAX",
    "STANDARDIZE",
    "NORMALIZE",
    "COUNT",
    "COUNT_TRUE",
    "CONSECUTIVE",
    "BARS_SINCE",
    "DURATION",
    "DAYS_SINCE",
    "TIME_SINCE",
    "REF",
    "LAG",
    "PREV",
    "SHIFT",
    "ROLLING_MEAN",
    "ROLLING_SUM",
    "ROLLING_STD",
    "ROLLING_MAX",
    "ROLLING_MIN",
    "ROLLING_MEDIAN",
    "ROLLING_CORR",
    "ROLLING_COV",
    "AND",
    "OR",
    "NOT",
    "XOR",
    "ADD",
    "SUB",
    "MUL",
    "DIV",
    "ABS",
    "MAX",
    "MIN",
    "LOG",
    "SQRT",
    "POWER",
    "CUMMAX",
    "CUMMIN",
    "SIGN",
    "IF",
];

pub fn catalog() -> Value {
    let indicators: Vec<Value> = abstract_api::funcs().map(|f| json!({
        "name":f.name, "group":format!("{:?}",f.group), "description":f.hint,
        "inputs":f.inputs.iter().map(|p|json!({"name":p.param_name,"kind":format!("{:?}",p.kind),"flags":p.flags.0})).collect::<Vec<_>>(),
        "parameters":f.opt_inputs.iter().map(|p|json!({"name":p.param_name,"label":p.display_name,"description":p.hint,"domain":format!("{:?}",p.kind)})).collect::<Vec<_>>(),
        "outputs":f.outputs.iter().enumerate().map(|(i,p)|json!({"index":i,"name":p.param_name,"kind":format!("{:?}",p.kind)})).collect::<Vec<_>>(),
        "unstable_period":format!("{:?}",f.unst_id), "production_eligible":false,
    })).collect();
    json!({"contract":CONTRACT,"library":{"name":"TA-Lib native Rust","revision":TA_REV,"license":"BSD-3-Clause"},
        "indicators":indicators,"operators":OPERATORS,"cross_section_operators":crate::factor_cross_section::OPERATORS,"read_only":true,"live_routing":false,
        "policies":{"null":"unknown_not_false","warmup":"null_until_full_history","recursive_seed":"frozen_input_start",
        "breakout":"previous_window_excludes_current","boolean":"three_valued_logic","daily_execution":"next_completed_session",
        "minute_execution":"strictly_after_completed_bar","cross_section":"requires_separate_complete_universe_contract"}})
}

impl Frame {
    pub fn validate(&self) -> Result<(), String> {
        let n = self.timestamps.len();
        if self.symbol.is_empty()
            || n == 0
            || n > 200_000
            || self.available_at.len() != n
            || self.fields.len() > 100
            || n.saturating_mul(self.fields.len()) > 1_000_000
        {
            return Err("factor_frame_invalid: identity/shape/limit".into());
        }
        if !["1d", "1w", "1m", "5m", "15m", "30m", "60m"].contains(&self.frequency.as_str()) {
            return Err("factor_frame_invalid: unsupported_frequency".into());
        }
        for i in 0..n {
            if (i > 0 && self.timestamps[i] <= self.timestamps[i - 1])
                || self.available_at[i] < self.timestamps[i]
                || self.available_at[i] > self.decision_at
            {
                return Err(format!(
                    "factor_input_not_visible: {} index={i}",
                    self.symbol
                ));
            }
        }
        for (field, values) in &self.fields {
            if values.len() != n || values.iter().flatten().any(|v| !v.is_finite()) {
                return Err(format!("factor_field_invalid: {} {field}", self.symbol));
            }
        }
        Ok(())
    }
}

pub fn evaluate(expr: &Expr, frame: &Frame) -> Result<Series, String> {
    frame.validate()?;
    fn cost(expr: &Expr, depth: usize, nodes: &mut usize) -> Result<usize, String> {
        *nodes += 1;
        if depth > 24 || *nodes > 256 {
            return Err("factor_expression_size_exceeded".into());
        }
        let (children, own) = match expr {
            Expr::Indicator {
                inputs, parameters, ..
            } => (
                inputs.as_slice(),
                parameters
                    .values()
                    .filter_map(Value::as_u64)
                    .max()
                    .unwrap_or(30)
                    .min(1_000_000) as usize,
            ),
            Expr::Operator { args, window, .. } => (args.as_slice(), window.unwrap_or(1)),
            _ => (&[][..], 1),
        };
        children.iter().try_fold(own, |total, child| {
            Ok(total.saturating_add(cost(child, depth + 1, nodes)?))
        })
    }
    if frame
        .timestamps
        .len()
        .saturating_mul(cost(expr, 0, &mut 0)?)
        > 20_000_000
    {
        return Err("factor_expression_compute_budget_exceeded".into());
    }
    evaluate_inner(expr, frame, 0)
}

fn evaluate_inner(expr: &Expr, frame: &Frame, depth: usize) -> Result<Series, String> {
    if depth > 24 {
        return Err("factor_expression_too_deep".into());
    }
    match expr {
        Expr::Number { value } if value.is_finite() => Ok(Series {
            value_type: ValueType::Number,
            values: vec![Some(*value); frame.timestamps.len()],
        }),
        Expr::Number { .. } => Err("factor_constant_nonfinite".into()),
        Expr::Field { name } => Ok(Series {
            value_type: ValueType::Number,
            values: frame
                .fields
                .get(name)
                .ok_or_else(|| format!("factor_source_field_missing: {} {name}", frame.symbol))?
                .clone(),
        }),
        Expr::Indicator {
            name,
            inputs,
            parameters,
            output,
        } => indicator(name, inputs, parameters, *output, frame, depth),
        Expr::Operator { name, args, window } => {
            if args.len() > 16 {
                return Err("factor_operator_arity_exceeded".into());
            }
            let args = args
                .iter()
                .map(|a| evaluate_inner(a, frame, depth + 1))
                .collect::<Result<Vec<_>, _>>()?;
            operator(name, &args, *window, frame)
        }
    }
}

fn indicator(
    name: &str,
    inputs: &[Expr],
    parameters: &BTreeMap<String, Value>,
    output: usize,
    frame: &Frame,
    depth: usize,
) -> Result<Series, String> {
    let id =
        abstract_api::get_func_handle(name).ok_or_else(|| format!("indicator_unknown: {name}"))?;
    let info = id.info();
    if output >= info.outputs.len() {
        return Err("indicator_output_invalid".into());
    }
    let real_count = info
        .inputs
        .iter()
        .filter(|i| i.kind == InputType::Real)
        .count();
    if inputs.len() != real_count || info.inputs.iter().any(|i| i.kind == InputType::Integer) {
        return Err(format!(
            "indicator_inputs_invalid: {name} expects {real_count} real series"
        ));
    }
    let mut data = inputs
        .iter()
        .map(|a| evaluate_inner(a, frame, depth + 1))
        .collect::<Result<Vec<_>, _>>()?;
    if data.iter().any(|s| s.value_type != ValueType::Number) {
        return Err("indicator_requires_numeric_input".into());
    }
    let price_names = ["open", "high", "low", "close", "volume", "open_interest"];
    let flags = info
        .inputs
        .iter()
        .filter(|i| i.kind == InputType::Price)
        .fold(0, |v, i| v | i.flags.0);
    let mut price_indices = [None; 6];
    for (i, field) in price_names.iter().enumerate() {
        if flags & (1 << i) != 0 {
            price_indices[i] = Some(data.len());
            data.push(evaluate_inner(
                &Expr::Field {
                    name: (*field).into(),
                },
                frame,
                depth + 1,
            )?);
        }
    }
    let core = Core::new();
    let mut validation = id.new_call(&core);
    for (key, v) in parameters {
        let slot = info
            .opt_inputs
            .iter()
            .position(|p| p.param_name == key)
            .ok_or_else(|| format!("indicator_parameter_unknown: {name}.{key}"))?;
        match info.opt_inputs[slot].kind {
            OptInputType::IntegerRange { .. } | OptInputType::IntegerList { .. } => {
                let v = v
                    .as_i64()
                    .and_then(|v| i32::try_from(v).ok())
                    .ok_or("indicator_parameter_requires_integer")?;
                validation.set_opt(slot, v).map_err(|e| format!("{e:?}"))?;
            }
            _ => {
                validation
                    .set_opt(
                        slot,
                        v.as_f64()
                            .filter(|v| v.is_finite())
                            .ok_or("indicator_parameter_requires_finite_number")?,
                    )
                    .map_err(|e| format!("{e:?}"))?;
            }
        }
    }
    let lookback = validation
        .lookback()
        .map_err(|e| format!("indicator_parameter_invalid: {name} {e:?}"))?;
    let n = frame.timestamps.len();
    let mut result = vec![None; n];
    let mut start = 0;
    // Never bridge missing source observations. Recursive indicators rewarm after a gap.
    while start < n {
        if data.iter().any(|s| s.values[start].is_none()) {
            start += 1;
            continue;
        }
        let mut end = start + 1;
        while end < n && data.iter().all(|s| s.values[end].is_some()) {
            end += 1;
        }
        if end - start <= lookback {
            start = end;
            continue;
        }
        let arrays = data
            .iter()
            .map(|s| {
                s.values[start..end]
                    .iter()
                    .map(|v| v.unwrap())
                    .collect::<Vec<_>>()
            })
            .collect::<Vec<_>>();
        let mut float_out = (0..info.outputs.len())
            .map(|_| vec![0.0; end - start])
            .collect::<Vec<_>>();
        let mut int_out = (0..info.outputs.len())
            .map(|_| vec![0i32; end - start])
            .collect::<Vec<_>>();
        let mut call = id.new_call(&core);
        for (key, v) in parameters {
            let slot = info
                .opt_inputs
                .iter()
                .position(|p| p.param_name == key)
                .unwrap();
            match info.opt_inputs[slot].kind {
                OptInputType::IntegerRange { .. } | OptInputType::IntegerList { .. } => {
                    call.set_opt(slot, v.as_i64().unwrap() as i32)
                        .map_err(|e| format!("{e:?}"))?;
                }
                _ => {
                    call.set_opt(slot, v.as_f64().unwrap())
                        .map_err(|e| format!("{e:?}"))?;
                }
            }
        }
        let mut real_slot = 0;
        for (slot, i) in info.inputs.iter().enumerate() {
            if i.kind == InputType::Real {
                call.set_input(slot, &arrays[real_slot])
                    .map_err(|e| format!("{e:?}"))?;
                real_slot += 1;
            } else {
                let p = price_indices.map(|i| i.map(|i| arrays[i].as_slice()));
                call.set_price_input(slot, p[0], p[1], p[2], p[3], p[4], p[5])
                    .map_err(|e| format!("{e:?}"))?;
            }
        }
        for (slot, (floats, ints)) in float_out.iter_mut().zip(int_out.iter_mut()).enumerate() {
            if info.outputs[slot].kind == OutputType::Real {
                call.set_output(slot, floats)
                    .map_err(|e| format!("{e:?}"))?;
            } else {
                call.set_int_output(slot, ints)
                    .map_err(|e| format!("{e:?}"))?;
            }
        }
        let range = call
            .call(0, end - start - 1)
            .map_err(|e| format!("indicator_failed: {name} {e:?}"))?;
        drop(call);
        for j in 0..range.count {
            let value = if info.outputs[output].kind == OutputType::Real {
                float_out[output][j]
            } else {
                int_out[output][j] as f64
            };
            if !value.is_finite() {
                return Err(format!(
                    "indicator_nonfinite: {name} index={}",
                    start + range.beg_idx + j
                ));
            }
            result[start + range.beg_idx + j] = Some(value);
        }
        start = end;
    }
    Ok(Series {
        value_type: ValueType::Number,
        values: result,
    })
}

fn average(v: &[f64]) -> f64 {
    v[0] + v
        .iter()
        .skip(1)
        .map(|x| (x - v[0]) / v.len() as f64)
        .sum::<f64>()
}
fn slope(v: &[f64]) -> f64 {
    let x = (v.len() - 1) as f64 / 2.0;
    let y = average(v);
    let num = v
        .iter()
        .enumerate()
        .map(|(i, v)| (i as f64 - x) * (v - y))
        .sum::<f64>();
    let den = (0..v.len()).map(|i| (i as f64 - x).powi(2)).sum::<f64>();
    num / den
}
fn boolean(v: bool) -> Option<f64> {
    Some(if v { 1.0 } else { 0.0 })
}

fn operator(
    name: &str,
    args: &[Series],
    window: Option<usize>,
    frame: &Frame,
) -> Result<Series, String> {
    if !OPERATORS.contains(&name) {
        return Err(format!("operator_not_registered: {name}"));
    }
    let bool_input = matches!(
        name,
        "AND"
            | "OR"
            | "NOT"
            | "XOR"
            | "COUNT"
            | "COUNT_TRUE"
            | "CONSECUTIVE"
            | "BARS_SINCE"
            | "DURATION"
            | "DAYS_SINCE"
            | "TIME_SINCE"
    );
    let lag = matches!(name, "REF" | "LAG" | "PREV" | "SHIFT");
    if args.is_empty()
        || (name == "IF"
            && (args.len() != 3
                || args[0].value_type != ValueType::Boolean
                || args[1].value_type != args[2].value_type))
        || (!lag
            && name != "IF"
            && args
                .iter()
                .any(|a| (a.value_type == ValueType::Boolean) != bool_input))
    {
        return Err(format!("operator_input_type_invalid: {name}"));
    }
    let arity = match name {
        "BETWEEN" | "OUTSIDE" | "IF" => 3,
        "GT" | "GTE" | "LT" | "LTE" | "EQ" | "NEQ" | "CROSS_ABOVE" | "CROSS_BELOW"
        | "BREAK_ABOVE" | "BREAK_BELOW" | "ADD" | "SUB" | "MUL" | "DIV" | "MAX" | "MIN"
        | "POWER" | "XOR" | "ROLLING_CORR" | "ROLLING_COV" | "NEAR_HIGH" | "NEAR_LOW"
        | "BULLISH_DIVERGENCE" | "BEARISH_DIVERGENCE" => 2,
        "AND" | "OR" => args.len(),
        _ => 1,
    };
    if args.len() != arity {
        return Err(format!("operator_arity_invalid: {name}"));
    }
    let windowed = matches!(
        name,
        "BREAK_HIGH"
            | "BREAK_LOW"
            | "RISING"
            | "FALLING"
            | "NON_DECREASING"
            | "NON_INCREASING"
            | "SLOPE"
            | "SLOPE_CHANGE"
            | "HHV"
            | "LLV"
            | "ARGMAX"
            | "ARGMIN"
            | "DISTANCE_TO_HIGH"
            | "DISTANCE_TO_LOW"
            | "NEW_HIGH"
            | "NEW_LOW"
            | "NEAR_HIGH"
            | "NEAR_LOW"
            | "BULLISH_DIVERGENCE"
            | "BEARISH_DIVERGENCE"
            | "ZSCORE"
            | "STANDARDIZE"
            | "MINMAX"
            | "NORMALIZE"
            | "COUNT"
            | "COUNT_TRUE"
    ) || name.starts_with("ROLLING_");
    let n = window.unwrap_or(1);
    if n == 0
        || n > 10_000
        || (windowed && window.is_none())
        || (matches!(
            name,
            "SLOPE"
                | "SLOPE_CHANGE"
                | "ZSCORE"
                | "STANDARDIZE"
                | "ROLLING_STD"
                | "ROLLING_CORR"
                | "ROLLING_COV"
        ) && n < 2)
    {
        return Err(format!("operator_window_invalid: {name}"));
    }
    let returns_bool = matches!(
        name,
        "GT" | "GTE"
            | "LT"
            | "LTE"
            | "EQ"
            | "NEQ"
            | "BETWEEN"
            | "OUTSIDE"
            | "CROSS_ABOVE"
            | "CROSS_BELOW"
            | "BREAK_ABOVE"
            | "BREAK_BELOW"
            | "BREAK_HIGH"
            | "BREAK_LOW"
            | "RISING"
            | "FALLING"
            | "NON_DECREASING"
            | "NON_INCREASING"
            | "TURN_UP"
            | "TURN_DOWN"
            | "BOTTOM_REVERSAL"
            | "TOP_REVERSAL"
            | "NEW_HIGH"
            | "NEW_LOW"
            | "NEAR_HIGH"
            | "NEAR_LOW"
            | "BULLISH_DIVERGENCE"
            | "BEARISH_DIVERGENCE"
            | "AND"
            | "OR"
            | "NOT"
            | "XOR"
    );
    let len = frame.timestamps.len();
    let mut out = vec![None; len];
    let mut last_true = None;
    let mut consecutive = Some(0usize);
    let mut extreme: Option<f64> = None;
    let mut cumulative_complete = true;
    for i in 0..len {
        let a = args[0].values[i];
        let b = args.get(1).and_then(|a| a.values[i]);
        let at = |j: usize| args[0].values.get(j).copied().flatten();
        let history = |end: usize, count: usize| -> Option<Vec<f64>> {
            if end < count {
                None
            } else {
                args[0].values[end - count..end].iter().copied().collect()
            }
        };
        out[i] = match name {
            "IF" => a.and_then(|a| {
                if a == 1.0 {
                    args[1].values[i]
                } else {
                    args[2].values[i]
                }
            }),
            "SIGN" => a.map(|v| {
                if v == 0.0 {
                    0.0
                } else if v > 0.0 {
                    1.0
                } else {
                    -1.0
                }
            }),
            "CUMMAX" | "CUMMIN" => {
                cumulative_complete &= a.is_some();
                extreme = a.filter(|_| cumulative_complete).map(|v| {
                    extreme.map_or(v, |p| if name == "CUMMAX" { p.max(v) } else { p.min(v) })
                });
                extreme
            }
            "AND" => {
                if args.iter().any(|a| a.values[i] == Some(0.0)) {
                    Some(0.0)
                } else if args.iter().any(|a| a.values[i].is_none()) {
                    None
                } else {
                    Some(1.0)
                }
            }
            "OR" => {
                if args.iter().any(|a| a.values[i] == Some(1.0)) {
                    Some(1.0)
                } else if args.iter().any(|a| a.values[i].is_none()) {
                    None
                } else {
                    Some(0.0)
                }
            }
            "NOT" => a.map(|v| 1.0 - v),
            "XOR" => a.zip(b).and_then(|(a, b)| boolean(a != b)),
            "GT" | "GTE" | "LT" | "LTE" | "EQ" | "NEQ" => a.zip(b).and_then(|(a, b)| {
                boolean(match name {
                    "GT" => a > b,
                    "GTE" => a >= b,
                    "LT" => a < b,
                    "LTE" => a <= b,
                    "EQ" => a == b,
                    _ => a != b,
                })
            }),
            "BETWEEN" | "OUTSIDE" => a.zip(b).zip(args[2].values[i]).and_then(|((a, b), c)| {
                if b > c {
                    None
                } else {
                    boolean((a >= b && a <= c) == (name == "BETWEEN"))
                }
            }),
            "CROSS_ABOVE" | "CROSS_BELOW" | "BREAK_ABOVE" | "BREAK_BELOW" => {
                if i == 0 {
                    None
                } else {
                    a.zip(b).zip(at(i - 1).zip(args[1].values[i - 1])).and_then(
                        |((a, b), (p, q))| {
                            boolean(if name.ends_with("ABOVE") {
                                p <= q && a > b
                            } else {
                                p >= q && a < b
                            })
                        },
                    )
                }
            }
            "REF" | "LAG" | "PREV" | "SHIFT" => i.checked_sub(n).and_then(at),
            "CHANGE" | "DIFF" | "DELTA" | "PCT_CHANGE" | "LOG_RETURN" => a
                .zip(i.checked_sub(n).and_then(at))
                .and_then(|(a, p)| match name {
                    "PCT_CHANGE" => {
                        if p == 0.0 {
                            None
                        } else {
                            Some(a / p - 1.0)
                        }
                    }
                    "LOG_RETURN" => {
                        if a <= 0.0 || p <= 0.0 {
                            None
                        } else {
                            Some((a / p).ln())
                        }
                    }
                    _ => Some(a - p),
                }),
            "ACCELERATION" => a
                .zip(i.checked_sub(n).and_then(at))
                .zip(i.checked_sub(n * 2).and_then(at))
                .map(|((a, p), q)| a - 2.0 * p + q),
            "BULLISH_DIVERGENCE" | "BEARISH_DIVERGENCE" => {
                if i < n || n < 4 {
                    None
                } else {
                    let price: Option<Vec<f64>> =
                        args[0].values[i - n..=i].iter().copied().collect();
                    let indicator: Option<Vec<f64>> =
                        args[1].values[i - n..=i].iter().copied().collect();
                    price.zip(indicator).and_then(|(price, indicator)| {
                        let low = name == "BULLISH_DIVERGENCE";
                        let pivots = (1..n)
                            .filter(|&j| {
                                if low {
                                    price[j] < price[j - 1] && price[j] < price[j + 1]
                                } else {
                                    price[j] > price[j - 1] && price[j] > price[j + 1]
                                }
                            })
                            .collect::<Vec<_>>();
                        if pivots.last() != Some(&(n - 1)) || pivots.len() < 2 {
                            return boolean(false);
                        }
                        let a = pivots[pivots.len() - 2];
                        let b = n - 1;
                        boolean(if low {
                            price[b] < price[a] && indicator[b] > indicator[a]
                        } else {
                            price[b] > price[a] && indicator[b] < indicator[a]
                        })
                    })
                }
            }
            "TURN_UP" | "TURN_DOWN" | "BOTTOM_REVERSAL" | "TOP_REVERSAL" => {
                if i < 2 {
                    None
                } else {
                    a.zip(at(i - 1)).zip(at(i - 2)).and_then(|((a, p), q)| {
                        if name == "ACCELERATION" {
                            Some(a - 2.0 * p + q)
                        } else {
                            boolean(if matches!(name, "TURN_UP" | "BOTTOM_REVERSAL") {
                                p < q && a > p
                            } else {
                                p > q && a < p
                            })
                        }
                    })
                }
            }
            "ABS" => a.map(f64::abs),
            "LOG" => a.filter(|v| *v > 0.0).map(f64::ln),
            "SQRT" => a.filter(|v| *v >= 0.0).map(f64::sqrt),
            "ADD" => a.zip(b).map(|(a, b)| a + b),
            "SUB" => a.zip(b).map(|(a, b)| a - b),
            "MUL" => a.zip(b).map(|(a, b)| a * b),
            "DIV" => a.zip(b).filter(|(_, b)| *b != 0.0).map(|(a, b)| a / b),
            "MAX" => a.zip(b).map(|(a, b)| a.max(b)),
            "MIN" => a.zip(b).map(|(a, b)| a.min(b)),
            "POWER" => a.zip(b).map(|(a, b)| a.powf(b)),
            "BARS_SINCE" | "DAYS_SINCE" | "TIME_SINCE" => {
                if a == Some(1.0) {
                    last_true = Some(i);
                }
                if a.is_none() {
                    last_true = None;
                }
                last_true.map(|t| {
                    if name == "BARS_SINCE" {
                        (i - t) as f64
                    } else {
                        let secs = (frame.timestamps[i] - frame.timestamps[t]).num_seconds() as f64;
                        if name == "DAYS_SINCE" {
                            secs / 86400.0
                        } else {
                            secs
                        }
                    }
                })
            }
            "CONSECUTIVE" | "DURATION" => {
                consecutive = match a {
                    Some(1.0) => consecutive.map(|v| v + 1),
                    Some(_) => Some(0),
                    None => None,
                };
                consecutive.map(|v| v as f64)
            }
            "BREAK_HIGH" | "NEW_HIGH" | "BREAK_LOW" | "NEW_LOW" => {
                a.zip(history(i, n)).and_then(|(a, v)| {
                    boolean(if matches!(name, "BREAK_HIGH" | "NEW_HIGH") {
                        a > v.into_iter().fold(f64::NEG_INFINITY, f64::max)
                    } else {
                        a < v.into_iter().fold(f64::INFINITY, f64::min)
                    })
                })
            }
            "RISING" | "FALLING" | "NON_DECREASING" | "NON_INCREASING" => history(i + 1, n + 1)
                .and_then(|v| {
                    boolean(v.windows(2).all(|p| match name {
                        "RISING" => p[1] > p[0],
                        "FALLING" => p[1] < p[0],
                        "NON_DECREASING" => p[1] >= p[0],
                        _ => p[1] <= p[0],
                    }))
                }),
            "SLOPE_CHANGE" => history(i + 1, n)
                .zip(history(i, n))
                .map(|(a, b)| slope(&a) - slope(&b)),
            _ => history(i + 1, n).and_then(|mut v| {
                let mean = average(&v);
                let lo = v.iter().copied().fold(f64::INFINITY, f64::min);
                let hi = v.iter().copied().fold(f64::NEG_INFINITY, f64::max);
                let variance = v.iter().map(|v| (v - mean).powi(2)).sum::<f64>() / n as f64;
                match name {
                    "HHV" | "ROLLING_MAX" => Some(hi),
                    "LLV" | "ROLLING_MIN" => Some(lo),
                    "ARGMAX" => v.iter().rposition(|x| *x == hi).map(|p| (n - 1 - p) as f64),
                    "ARGMIN" => v.iter().rposition(|x| *x == lo).map(|p| (n - 1 - p) as f64),
                    "DISTANCE_TO_HIGH" => {
                        if hi == 0.0 {
                            None
                        } else {
                            Some(v[n - 1] / hi - 1.0)
                        }
                    }
                    "DISTANCE_TO_LOW" => {
                        if lo == 0.0 {
                            None
                        } else {
                            Some(v[n - 1] / lo - 1.0)
                        }
                    }
                    "NEAR_HIGH" | "NEAR_LOW" => b.filter(|b| *b >= 0.0).and_then(|b| {
                        let base = if name == "NEAR_HIGH" { hi } else { lo };
                        if base == 0.0 {
                            None
                        } else {
                            boolean((v[n - 1] / base - 1.0).abs() <= b)
                        }
                    }),
                    "ZSCORE" | "STANDARDIZE" => {
                        if variance == 0.0 {
                            None
                        } else {
                            Some((v[n - 1] - mean) / variance.sqrt())
                        }
                    }
                    "MINMAX" | "NORMALIZE" => {
                        if hi == lo {
                            None
                        } else {
                            Some((v[n - 1] - lo) / (hi - lo))
                        }
                    }
                    "ROLLING_MEAN" => Some(mean),
                    "ROLLING_SUM" | "COUNT" | "COUNT_TRUE" => Some(v.iter().sum()),
                    "ROLLING_STD" => Some(variance.sqrt()),
                    "ROLLING_MEDIAN" => {
                        v.sort_by(f64::total_cmp);
                        Some(if n % 2 == 1 {
                            v[n / 2]
                        } else {
                            (v[n / 2 - 1] + v[n / 2]) / 2.0
                        })
                    }
                    "SLOPE" => Some(slope(&v)),
                    "ROLLING_CORR" | "ROLLING_COV" => {
                        let b: Option<Vec<f64>> =
                            args[1].values[i + 1 - n..=i].iter().copied().collect();
                        b.and_then(|b| {
                            let bm = average(&b);
                            let cov = v
                                .iter()
                                .zip(&b)
                                .map(|(a, b)| (a - mean) * (b - bm))
                                .sum::<f64>()
                                / n as f64;
                            if name == "ROLLING_COV" {
                                Some(cov)
                            } else {
                                let bv = b.iter().map(|b| (b - bm).powi(2)).sum::<f64>() / n as f64;
                                let d = (variance * bv).sqrt();
                                if d == 0.0 { None } else { Some(cov / d) }
                            }
                        })
                    }
                    _ => None,
                }
            }),
        }
        .filter(|v| v.is_finite());
    }
    Ok(Series {
        value_type: if name == "IF" {
            args[1].value_type
        } else if lag {
            args[0].value_type
        } else if returns_bool {
            ValueType::Boolean
        } else {
            ValueType::Number
        },
        values: out,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    fn frame(values: Vec<Option<f64>>) -> Frame {
        let start = DateTime::parse_from_rfc3339("2026-09-01T15:30:00+08:00").unwrap();
        let times = (0..values.len())
            .map(|i| start + chrono::Duration::days(i as i64))
            .collect::<Vec<_>>();
        Frame {
            symbol: "TEST".into(),
            frequency: "1d".into(),
            decision_at: *times.last().unwrap(),
            available_at: times.clone(),
            timestamps: times,
            fields: BTreeMap::from([("close".into(), values)]),
        }
    }
    fn expr(v: Value) -> Expr {
        serde_json::from_value(v).unwrap()
    }
    #[test]
    fn ta_sma_real_values_and_parameter_validation() {
        let frame = frame(vec![Some(1.0), Some(2.0), Some(3.0), Some(4.0)]);
        let e = expr(
            json!({"kind":"indicator","name":"SMA","inputs":[{"kind":"field","name":"close"}],"parameters":{"optInTimePeriod":3}}),
        );
        assert_eq!(
            evaluate(&e, &frame).unwrap().values,
            vec![None, None, Some(2.0), Some(3.0)]
        );
        let bad = expr(
            json!({"kind":"indicator","name":"SMA","inputs":[{"kind":"field","name":"close"}],"parameters":{"period":3}}),
        );
        assert!(
            evaluate(&bad, &frame)
                .unwrap_err()
                .contains("parameter_unknown")
        );
    }
    #[test]
    fn cross_is_event_not_state_and_never_uses_future() {
        let f = frame(vec![
            Some(9.0),
            Some(10.0),
            Some(11.0),
            Some(12.0),
            Some(8.0),
        ]);
        let e = expr(
            json!({"kind":"operator","name":"CROSS_ABOVE","args":[{"kind":"field","name":"close"},{"kind":"number","value":10.0}]}),
        );
        assert_eq!(
            evaluate(&e, &f).unwrap().values,
            vec![None, Some(0.0), Some(1.0), Some(0.0), Some(0.0)]
        );
        let mut invalid = f.clone();
        invalid.available_at[4] = invalid.decision_at + chrono::Duration::seconds(1);
        assert!(evaluate(&e, &invalid).is_err());
    }
    #[test]
    fn missing_is_not_zero_and_breakout_excludes_current() {
        let f = frame(vec![Some(1.0), Some(2.0), Some(3.0), None, Some(5.0)]);
        let e = expr(
            json!({"kind":"operator","name":"BREAK_HIGH","window":2,"args":[{"kind":"field","name":"close"}]}),
        );
        assert_eq!(
            evaluate(&e, &f).unwrap().values,
            vec![None, None, Some(1.0), None, None]
        );
        let zero = expr(
            json!({"kind":"operator","name":"DIV","args":[{"kind":"field","name":"close"},{"kind":"number","value":0}]}),
        );
        assert!(
            evaluate(&zero, &f)
                .unwrap()
                .values
                .iter()
                .all(Option::is_none)
        );
    }
    #[test]
    fn ta_rewarms_after_gap_and_const_zscore_is_unknown() {
        let f = frame(vec![Some(1.0), Some(1.0), None, Some(2.0), Some(2.0)]);
        let e = expr(
            json!({"kind":"indicator","name":"SMA","inputs":[{"kind":"field","name":"close"}],"parameters":{"optInTimePeriod":2}}),
        );
        assert_eq!(
            evaluate(&e, &f).unwrap().values,
            vec![None, Some(1.0), None, None, Some(2.0)]
        );
        let e = expr(
            json!({"kind":"operator","name":"ZSCORE","window":2,"args":[{"kind":"field","name":"close"}]}),
        );
        assert!(evaluate(&e, &f).unwrap().values.iter().all(Option::is_none));
    }
    #[test]
    fn no_event_has_no_bars_since_and_type_errors_reject() {
        let f = frame(vec![Some(1.0), Some(1.0), Some(1.0)]);
        let state = json!({"kind":"operator","name":"GT","args":[{"kind":"field","name":"close"},{"kind":"number","value":5}]});
        let e = expr(json!({"kind":"operator","name":"BARS_SINCE","args":[state]}));
        assert!(evaluate(&e, &f).unwrap().values.iter().all(Option::is_none));
        assert!(
            evaluate(
                &expr(
                    json!({"kind":"operator","name":"NOT","args":[{"kind":"field","name":"close"}]})
                ),
                &f
            )
            .is_err()
        );
    }
    #[test]
    fn literal_unknown_fields_reject_and_catalog_is_not_trading_permission() {
        assert!(
            serde_json::from_value::<Expr>(json!({"kind":"number","value":1,"account_id":2}))
                .is_err()
        );
        let c = catalog();
        assert!(c["indicators"].as_array().unwrap().len() > 190);
        assert_eq!(c["live_routing"], false);
    }
}