use std::collections::VecDeque;
use crate::model::Bar;
use super::smoothing::{crossed_over, crossed_under};
use super::{Indicator, IndicatorAlert, IndicatorOutput};
use std::collections::HashMap;
pub struct MoneyFlowProfileEngine {
lookback: usize,
rows: usize,
va_pct: f64,
window: VecDeque<Bar>,
prev_close: Option<f64>,
prev_vah: Option<f64>,
prev_val: Option<f64>,
prev_bull_pct: Option<f64>,
alerts: Vec<IndicatorAlert>,
}
impl MoneyFlowProfileEngine {
pub fn new(lookback: usize, rows: usize, va_pct: f64) -> Self {
let lookback = lookback.max(1);
let rows = rows.max(1);
Self {
lookback,
rows,
va_pct,
window: VecDeque::with_capacity(lookback),
prev_close: None,
prev_vah: None,
prev_val: None,
prev_bull_pct: None,
alerts: Vec::new(),
}
}
pub fn with_defaults() -> Self {
Self::new(200, 25, 0.70)
}
}
impl Indicator for MoneyFlowProfileEngine {
fn name(&self) -> &str {
"money_flow_profile"
}
fn warmup_period(&self) -> usize {
2
}
fn reset(&mut self) {
self.window.clear();
self.prev_close = None;
self.prev_vah = None;
self.prev_val = None;
self.prev_bull_pct = None;
self.alerts.clear();
}
fn on_bar(&mut self, bar: &Bar) -> Option<IndicatorOutput> {
self.alerts.clear();
if self.window.len() == self.lookback {
self.window.pop_front();
}
self.window.push_back(bar.clone());
if self.window.len() < 2 {
return None;
}
let p_lo = self
.window
.iter()
.map(|b| b.low)
.fold(f64::INFINITY, f64::min);
let p_hi = self
.window
.iter()
.map(|b| b.high)
.fold(f64::NEG_INFINITY, f64::max);
if p_hi <= p_lo {
return None;
}
let p_step = (p_hi - p_lo) / self.rows as f64;
let mut total_flow = vec![0.0_f64; self.rows];
let mut bull_flow = vec![0.0_f64; self.rows];
for b in &self.window {
let (h, l, c) = (b.high, b.low, b.close);
if h <= l {
continue;
}
let v = if b.volume > 0.0 { b.volume } else { 1.0 };
let buy_ratio = ((c - l) / (h - l)).clamp(0.0, 1.0);
for r in 0..self.rows {
let row_lo = p_lo + r as f64 * p_step;
let row_hi = row_lo + p_step;
if h < row_lo || l >= row_hi {
continue;
}
let overlap = if l >= row_lo && h > row_hi {
(row_hi - l) / (h - l)
} else if h <= row_hi && l < row_lo {
(h - row_lo) / (h - l)
} else if l >= row_lo && h <= row_hi {
1.0
} else {
p_step / (h - l)
};
let mf_price = p_lo + (r as f64 + 0.5) * p_step; let flow = v * overlap * mf_price; total_flow[r] += flow;
bull_flow[r] += flow * buy_ratio;
}
}
let tot_max = total_flow.iter().cloned().fold(0.0_f64, f64::max);
if tot_max <= 0.0 {
return None;
}
let tot_sum: f64 = total_flow.iter().sum();
let poc_idx = total_flow
.iter()
.enumerate()
.max_by(|a, b| a.1.partial_cmp(b.1).unwrap())
.map(|(i, _)| i)
.unwrap();
let poc_price = p_lo + (poc_idx as f64 + 0.5) * p_step;
let mut delta_poc_idx = 0usize;
let mut delta_poc_abs_max = 0.0_f64;
for r in 0..self.rows {
let d = (2.0 * bull_flow[r] - total_flow[r]).abs();
if d > delta_poc_abs_max {
delta_poc_abs_max = d;
delta_poc_idx = r;
}
}
let delta_poc_price = p_lo + (delta_poc_idx as f64 + 0.5) * p_step;
let mut va_lo = poc_idx;
let mut va_hi = poc_idx;
let mut va_acc = total_flow[poc_idx];
let va_tgt = tot_sum * self.va_pct;
while va_acc < va_tgt {
let add_lo = if va_lo > 0 {
total_flow[va_lo - 1]
} else {
-1.0
};
let add_hi = if va_hi < self.rows - 1 {
total_flow[va_hi + 1]
} else {
-1.0
};
if add_lo < 0.0 && add_hi < 0.0 {
break;
}
if add_lo >= add_hi {
va_lo -= 1;
va_acc += add_lo;
} else {
va_hi += 1;
va_acc += add_hi;
}
}
let vah_price = p_lo + (va_hi + 1) as f64 * p_step;
let val_price = p_lo + va_lo as f64 * p_step;
let bull_pct = bull_flow.iter().sum::<f64>() / tot_sum * 100.0;
let mut vah_breakout = false;
let mut val_breakdown = false;
let mut bull_bias = false;
let mut bear_bias = false;
let mut vah_breakout_strength = 0.0;
let mut val_breakdown_strength = 0.0;
if let (Some(prev_close), Some(prev_vah), Some(prev_val), Some(prev_bull_pct)) = (
self.prev_close,
self.prev_vah,
self.prev_val,
self.prev_bull_pct,
) {
vah_breakout = crossed_over(prev_close, prev_vah, bar.close, vah_price);
val_breakdown = crossed_under(prev_close, prev_val, bar.close, val_price);
bull_bias = crossed_over(prev_bull_pct, 50.0, bull_pct, 50.0);
bear_bias = crossed_under(prev_bull_pct, 50.0, bull_pct, 50.0);
let va_width = vah_price - val_price;
vah_breakout_strength = if va_width > 0.0 {
((bar.close - vah_price) / va_width).clamp(0.0, 1.0)
} else {
1.0
};
val_breakdown_strength = if va_width > 0.0 {
((val_price - bar.close) / va_width).clamp(0.0, 1.0)
} else {
1.0
};
}
if vah_breakout {
self.alerts.push(IndicatorAlert::new(
"vah_breakout",
"Money Flow Profile: close crossed above the Value Area High",
vah_breakout_strength,
));
}
if val_breakdown {
self.alerts.push(IndicatorAlert::new(
"val_breakdown",
"Money Flow Profile: close crossed below the Value Area Low",
val_breakdown_strength,
));
}
if bull_bias {
self.alerts.push(IndicatorAlert::new(
"bull_bias",
"Money Flow Profile: flow bias turned bullish",
1.0,
));
}
if bear_bias {
self.alerts.push(IndicatorAlert::new(
"bear_bias",
"Money Flow Profile: flow bias turned bearish",
1.0,
));
}
self.prev_close = Some(bar.close);
self.prev_vah = Some(vah_price);
self.prev_val = Some(val_price);
self.prev_bull_pct = Some(bull_pct);
let mut extra = HashMap::new();
extra.insert("vah".to_string(), vah_price);
extra.insert("val".to_string(), val_price);
extra.insert("delta_poc".to_string(), delta_poc_price);
extra.insert("bull_pct".to_string(), bull_pct);
Some(IndicatorOutput::with_extra(poc_price, extra))
}
fn alerts(&self) -> Vec<IndicatorAlert> {
self.alerts.clone()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn synthetic_volume_substitutes_for_non_positive_volume() {
let mut mfp = MoneyFlowProfileEngine::new(2, 5, 0.70);
let bars = [
Bar::new(1, 100.0, 101.0, 99.0, 100.0, 0.0),
Bar::new(2, 101.0, 102.0, 100.0, 101.0, 0.0),
];
mfp.on_bar(&bars[0]);
let out = mfp.on_bar(&bars[1]);
assert!(
out.is_some(),
"zero-volume bars must fall back to synthetic volume, not None"
);
}
#[test]
fn reset_clears_window_and_cross_state() {
let mut mfp = MoneyFlowProfileEngine::new(2, 10, 0.70);
mfp.on_bar(&Bar::new(1, 10.0, 10.1, 9.9, 10.0, 1000.0));
mfp.on_bar(&Bar::new(2, 49.9, 50.0, 49.8, 49.9, 300.0));
mfp.reset();
assert!(mfp
.on_bar(&Bar::new(1, 10.0, 10.1, 9.9, 10.0, 1000.0))
.is_none());
}
}