indicators/signal/
cvd.rs

1//! Layer 8 — Cumulative Volume Delta (OHLCV heuristic).
2//!
3//! Estimates buy/sell volume from OHLCV bars and tracks cumulative delta,
4//! slope, and price-CVD divergence.
5
6use std::collections::{HashMap, VecDeque};
7
8use chrono::{NaiveDate, TimeZone, Utc};
9
10use crate::error::IndicatorError;
11use crate::indicator::{Indicator, IndicatorOutput};
12use crate::registry::param_usize;
13use crate::types::Candle;
14
15// ── Params ────────────────────────────────────────────────────────────────────
16
17#[derive(Debug, Clone)]
18pub struct CvdParams {
19    pub slope_bars: usize,
20    pub div_lookback: usize,
21}
22
23impl Default for CvdParams {
24    fn default() -> Self {
25        Self {
26            slope_bars: 10,
27            div_lookback: 20,
28        }
29    }
30}
31
32// ── Indicator wrapper ─────────────────────────────────────────────────────────
33
34/// Batch `Indicator` adapter for [`CVDTracker`].
35#[derive(Debug, Clone)]
36pub struct CvdIndicator {
37    pub params: CvdParams,
38}
39
40impl CvdIndicator {
41    pub fn new(params: CvdParams) -> Self {
42        Self { params }
43    }
44}
45
46impl Indicator for CvdIndicator {
47    fn name(&self) -> &'static str {
48        "CVD"
49    }
50    fn required_len(&self) -> usize {
51        self.params.div_lookback + 1
52    }
53    fn required_columns(&self) -> &[&'static str] {
54        &["open", "high", "low", "close", "volume"]
55    }
56
57    fn calculate(&self, candles: &[Candle]) -> Result<IndicatorOutput, IndicatorError> {
58        self.check_len(candles)?;
59        let p = &self.params;
60        let mut tracker = CVDTracker::new(p.slope_bars, p.div_lookback);
61        let n = candles.len();
62        let mut cvd_out = vec![f64::NAN; n];
63        let mut slope = vec![f64::NAN; n];
64        let mut div_out = vec![f64::NAN; n];
65        for (i, c) in candles.iter().enumerate() {
66            tracker.update(c);
67            cvd_out[i] = tracker.cvd;
68            slope[i] = tracker.cvd_slope;
69            div_out[i] = tracker.divergence as f64;
70        }
71        Ok(IndicatorOutput::from_pairs([
72            ("cvd", cvd_out),
73            ("cvd_slope", slope),
74            ("cvd_div", div_out),
75        ]))
76    }
77}
78
79// ── Registry factory ──────────────────────────────────────────────────────────
80
81pub fn factory<S: ::std::hash::BuildHasher>(params: &HashMap<String, String, S>) -> Result<Box<dyn Indicator>, IndicatorError> {
82    let slope_bars = param_usize(params, "slope_bars", 10)?;
83    let div_lookback = param_usize(params, "div_lookback", 20)?;
84    Ok(Box::new(CvdIndicator::new(CvdParams {
85        slope_bars,
86        div_lookback,
87    })))
88}
89
90#[derive(Debug)]
91pub struct CVDTracker {
92    slope_bars: usize,
93    div_lookback: usize,
94
95    day_cvd: f64,
96    last_date: Option<NaiveDate>,
97    cvd_hist: VecDeque<f64>,
98    price_hist: VecDeque<f64>,
99
100    pub cvd: f64,
101    pub delta: f64,
102    pub cvd_slope: f64,
103    pub bullish: bool,
104    /// `+1` = bullish divergence, `-1` = bearish divergence, `0` = none.
105    pub divergence: i8,
106}
107
108impl CVDTracker {
109    pub fn new(slope_bars: usize, div_lookback: usize) -> Self {
110        let cap = (div_lookback + 10).max(50);
111        Self {
112            slope_bars,
113            div_lookback,
114            day_cvd: 0.0,
115            last_date: None,
116            cvd_hist: VecDeque::with_capacity(cap),
117            price_hist: VecDeque::with_capacity(cap),
118            cvd: 0.0,
119            delta: 0.0,
120            cvd_slope: 0.0,
121            bullish: false,
122            divergence: 0,
123        }
124    }
125
126    pub fn update(&mut self, candle: &Candle) {
127        let dt = Utc
128            .timestamp_millis_opt(candle.time)
129            .single()
130            .unwrap_or_else(Utc::now);
131        let date = dt.date_naive();
132
133        if Some(date) != self.last_date {
134            self.day_cvd = 0.0;
135            self.last_date = Some(date);
136        }
137
138        let bar_rng = candle.high - candle.low;
139        let buy_vol = if bar_rng > 0.0 {
140            candle.volume * (candle.close - candle.low) / bar_rng
141        } else {
142            candle.volume * 0.5
143        };
144        self.delta = buy_vol - (candle.volume - buy_vol);
145        self.day_cvd += self.delta;
146        self.cvd = self.day_cvd;
147
148        let cap = self.cvd_hist.capacity();
149        if self.cvd_hist.len() == cap {
150            self.cvd_hist.pop_front();
151        }
152        if self.price_hist.len() == cap {
153            self.price_hist.pop_front();
154        }
155        self.cvd_hist.push_back(self.cvd);
156        self.price_hist.push_back(candle.close);
157
158        if self.cvd_hist.len() >= self.slope_bars {
159            let arr: Vec<f64> = self.cvd_hist.iter().copied().collect();
160            self.cvd_slope = arr[arr.len() - 1] - arr[arr.len() - self.slope_bars];
161        }
162        self.bullish = self.cvd_slope > 0.0;
163        self.divergence = self.check_divergence();
164    }
165
166    fn check_divergence(&self) -> i8 {
167        let n = self.cvd_hist.len().min(self.div_lookback);
168        if n < 10 {
169            return 0;
170        }
171        let prices: Vec<f64> = self.price_hist.iter().rev().take(n).copied().collect();
172        let cvds: Vec<f64> = self.cvd_hist.iter().rev().take(n).copied().collect();
173
174        let last_p = prices[0];
175        let last_c = cvds[0];
176
177        // Bullish divergence: price at new low but CVD is not
178        let min_p = prices[1..].iter().copied().fold(f64::INFINITY, f64::min);
179        let min_c = cvds[1..].iter().copied().fold(f64::INFINITY, f64::min);
180        if last_p < min_p && last_c > min_c {
181            return 1;
182        }
183
184        // Bearish divergence: price at new high but CVD is not
185        let max_p = prices[1..]
186            .iter()
187            .copied()
188            .fold(f64::NEG_INFINITY, f64::max);
189        let max_c = cvds[1..].iter().copied().fold(f64::NEG_INFINITY, f64::max);
190        if last_p > max_p && last_c < max_c {
191            return -1;
192        }
193
194        0
195    }
196}
indicators/signal/cvd.rs

indicators/signal/
cvd.rs
