///|
pub struct StreamingMoments {
  mut count : Int
  mut mean : Double
  mut m2 : Double
  mut m3 : Double
  mut m4 : Double
}

///|
pub fn StreamingMoments::new() -> StreamingMoments {
  { count: 0, mean: 0.0, m2: 0.0, m3: 0.0, m4: 0.0 }
}

///|
pub fn StreamingMoments::push(self : StreamingMoments, value : Double) -> Unit {
  let previous_count = self.count
  self.count += 1
  let n = self.count.to_double()
  let delta = value - self.mean
  let delta_n = delta / n
  let delta_n2 = delta_n * delta_n
  let term1 = delta * delta_n * previous_count.to_double()
  self.mean += delta_n
  self.m4 += term1 * delta_n2 * (n * n - 3.0 * n + 3.0) +
    6.0 * delta_n2 * self.m2 -
    4.0 * delta_n * self.m3
  self.m3 += term1 * delta_n * (n - 2.0) - 3.0 * delta_n * self.m2
  self.m2 += term1
}

///|
pub fn StreamingMoments::merge(
  self : StreamingMoments,
  other : StreamingMoments,
) -> Unit {
  if other.count == 0 {
    return
  }
  if self.count == 0 {
    self.count = other.count
    self.mean = other.mean
    self.m2 = other.m2
    self.m3 = other.m3
    self.m4 = other.m4
    return
  }
  let left_count = self.count.to_double()
  let right_count = other.count.to_double()
  let total_count = left_count + right_count
  let delta = other.mean - self.mean
  let delta2 = delta * delta
  let delta3 = delta2 * delta
  let delta4 = delta3 * delta
  let combined_m2 = self.m2 +
    other.m2 +
    delta2 * left_count * right_count / total_count
  let combined_m3 = self.m3 +
    other.m3 +
    delta3 *
    left_count *
    right_count *
    (left_count - right_count) /
    (total_count * total_count) +
    3.0 * delta * (left_count * other.m2 - right_count * self.m2) / total_count
  let combined_m4 = self.m4 +
    other.m4 +
    delta4 *
    left_count *
    right_count *
    (
      left_count * left_count -
      left_count * right_count +
      right_count * right_count
    ) /
    (total_count * total_count * total_count) +
    6.0 *
    delta2 *
    (left_count * left_count * other.m2 + right_count * right_count * self.m2) /
    (total_count * total_count) +
    4.0 * delta * (left_count * other.m3 - right_count * self.m3) / total_count
  self.mean = (left_count * self.mean + right_count * other.mean) / total_count
  self.count += other.count
  self.m2 = combined_m2
  self.m3 = combined_m3
  self.m4 = combined_m4
}

///|
pub fn StreamingMoments::variance(
  self : StreamingMoments,
  sample? : Bool = true,
) -> Double {
  if self.count == 0 {
    return 0.0
  }
  let denominator = if sample && self.count > 1 {
    self.count - 1
  } else {
    self.count
  }
  self.m2 / denominator.to_double()
}

///|
pub fn StreamingMoments::stddev(
  self : StreamingMoments,
  sample? : Bool = true,
) -> Double {
  self.variance(sample~).sqrt()
}

///|
pub fn StreamingMoments::skewness(self : StreamingMoments) -> Double {
  if self.m2 == 0.0 || self.count == 0 {
    0.0
  } else {
    self.m3 /
    self.count.to_double() /
    (self.m2 / self.count.to_double()).sqrt() /
    (self.m2 / self.count.to_double()).sqrt() /
    (self.m2 / self.count.to_double()).sqrt()
  }
}

///|
pub fn StreamingMoments::kurtosis(self : StreamingMoments) -> Double {
  if self.m2 == 0.0 || self.count == 0 {
    0.0
  } else {
    self.m4 /
    self.count.to_double() /
    (self.m2 / self.count.to_double()) /
    (self.m2 / self.count.to_double())
  }
}

///|
pub struct StreamingCovariance {
  mut count : Int
  mut mean_x : Double
  mut mean_y : Double
  mut co_moment : Double
  mut m2_x : Double
  mut m2_y : Double
}

///|
pub fn StreamingCovariance::new() -> StreamingCovariance {
  { count: 0, mean_x: 0.0, mean_y: 0.0, co_moment: 0.0, m2_x: 0.0, m2_y: 0.0 }
}

///|
pub fn StreamingCovariance::push(
  self : StreamingCovariance,
  x : Double,
  y : Double,
) -> Unit {
  let old_count = self.count
  self.count += 1
  let n = self.count.to_double()
  let delta_x = x - self.mean_x
  let delta_y = y - self.mean_y
  self.mean_x += delta_x / n
  self.mean_y += delta_y / n
  self.co_moment += delta_x * (y - self.mean_y)
  self.m2_x += delta_x * (x - self.mean_x)
  self.m2_y += delta_y * (y - self.mean_y)
  let _ = old_count
}

///|
pub fn StreamingCovariance::covariance(
  self : StreamingCovariance,
  sample? : Bool = true,
) -> Double {
  if self.count <= 1 && sample {
    0.0
  } else {
    let denominator = if sample { self.count - 1 } else { self.count }
    if denominator <= 0 {
      0.0
    } else {
      self.co_moment / denominator.to_double()
    }
  }
}

///|
pub fn StreamingCovariance::correlation(self : StreamingCovariance) -> Double {
  let denominator = (self.m2_x * self.m2_y).sqrt()
  if denominator == 0.0 {
    0.0
  } else {
    self.co_moment / denominator
  }
}

///|
pub struct StreamingWindow {
  capacity : Int
  mut cursor : Int
  mut size : Int
  mut buffer : Array[Double]
}

///|
pub fn StreamingWindow::new(capacity : Int) -> StreamingWindow {
  if capacity <= 0 {
    abort("capacity must be positive")
  }
  { capacity, cursor: 0, size: 0, buffer: [] }
}

///|
pub fn StreamingWindow::push(self : StreamingWindow, value : Double) -> Unit {
  if self.size < self.capacity {
    self.buffer.push(value)
    self.size += 1
  } else {
    self.buffer[self.cursor] = value
  }
  self.cursor = (self.cursor + 1) % self.capacity
}

///|
pub fn StreamingWindow::len(self : StreamingWindow) -> Int {
  self.size
}

///|
pub fn StreamingWindow::is_full(self : StreamingWindow) -> Bool {
  self.size == self.capacity
}

///|
pub fn StreamingWindow::values(self : StreamingWindow) -> Array[Double] {
  let result = []
  if self.size < self.capacity {
    for value in self.buffer {
      result.push(value)
    }
  } else {
    for offset = 0; offset < self.capacity; offset = offset + 1 {
      result.push(self.buffer[(self.cursor + offset) % self.capacity])
    }
  }
  result
}

///|
pub fn StreamingWindow::mean(self : StreamingWindow) -> Double {
  mean(self.values())
}

///|
pub fn StreamingWindow::median(self : StreamingWindow) -> Double {
  median(self.values())
}

///|
pub fn StreamingWindow::mad(self : StreamingWindow) -> Double {
  mad(self.values())
}

///|
pub fn StreamingWindow::robust_z(
  self : StreamingWindow,
  value : Double,
) -> Double {
  let values = self.values()
  robust_z_score(value, median(values), mad(values))
}

///|
pub fn StreamingWindow::trimmed_mean(
  self : StreamingWindow,
  trim_percent : Double,
) -> Double {
  trimmed_mean(self.values(), trim_percent)
}

///|
pub fn StreamingWindow::clear(self : StreamingWindow) -> Unit {
  self.cursor = 0
  self.size = 0
  self.buffer = []
}

///|
pub struct StreamingQuantileSketch {
  capacity : Int
  values_buffer : Array[Double]
  mut seen : Int
  mut state : Int
}

///|
pub fn StreamingQuantileSketch::new(capacity : Int) -> StreamingQuantileSketch {
  if capacity < 4 {
    abort("capacity must be at least four")
  }
  { capacity, values_buffer: [], seen: 0, state: 0x9E3779B9 }
}

///|
fn StreamingQuantileSketch::next_random(self : StreamingQuantileSketch) -> Int {
  self.state = self.state * 1664525 + 1013904223
  if self.state < 0 {
    -self.state
  } else {
    self.state
  }
}

///|
pub fn StreamingQuantileSketch::push(
  self : StreamingQuantileSketch,
  value : Double,
) -> Unit {
  self.seen += 1
  if self.values_buffer.length() < self.capacity {
    self.values_buffer.push(value)
  } else {
    let choice = self.next_random() % self.seen
    if choice < self.capacity {
      self.values_buffer[choice] = value
    }
  }
}

///|
pub fn StreamingQuantileSketch::count(self : StreamingQuantileSketch) -> Int {
  self.seen
}

///|
pub fn StreamingQuantileSketch::quantile(
  self : StreamingQuantileSketch,
  probability : Double,
) -> Double {
  quantile(self.values_buffer, probability)
}

///|
pub fn StreamingQuantileSketch::median(
  self : StreamingQuantileSketch,
) -> Double {
  median(self.values_buffer)
}

///|
pub fn StreamingQuantileSketch::mad(self : StreamingQuantileSketch) -> Double {
  mad(self.values_buffer)
}

///|
pub fn StreamingQuantileSketch::snapshot(
  self : StreamingQuantileSketch,
) -> Array[Double] {
  copy_array(self.values_buffer)
}

///|
pub fn StreamingRobustStats::push_many(
  self : StreamingRobustStats,
  values : Array[Double],
) -> Unit {
  for value in values {
    self.push(value)
  }
}

///|
pub fn StreamingRobustStats::count(self : StreamingRobustStats) -> Int {
  self.count
}

///|
pub fn StreamingRobustStats::reset(self : StreamingRobustStats) -> Unit {
  self.count = 0
  self.mu = 0.0
  self.m2 = 0.0
}

///|
pub fn StreamingRobustStats::standard_deviation(
  self : StreamingRobustStats,
) -> Double {
  self.variance().sqrt()
}

///|
pub fn StreamingRobustStats::merge(
  self : StreamingRobustStats,
  other : StreamingRobustStats,
) -> Unit {
  if other.count == 0 {
    return
  }
  if self.count == 0 {
    self.count = other.count
    self.mu = other.mu
    self.m2 = other.m2
    return
  }
  let left = self.count.to_double()
  let right = other.count.to_double()
  let total = left + right
  let delta = other.mu - self.mu
  self.m2 += other.m2 + delta * delta * left * right / total
  self.mu = (left * self.mu + right * other.mu) / total
  self.count += other.count
}