rustrade_data/exchange/massive/
corporate_action.rs

1//! [`StockSplitSource`] implementation for the Massive REST client.
2//!
3//! Adapts the lower-level [`MassiveRestClient::fetch_splits_raw`](super::rest::MassiveRestClient)
4//! raw-split stream into the provider-agnostic
5//! [`CorporateAction`] sourcing descriptor, computing the split ratio via the shared
6//! [`CorporateActionKind::stock_split`] helper so Massive derives ratios identically to every other
7//! provider.
8
9use async_stream::try_stream;
10use futures::{Stream, StreamExt};
11use smol_str::SmolStr;
12use tracing::warn;
13
14use rustrade_instrument::corporate_action::{CorporateAction, CorporateActionKind};
15use rustrade_integration::corporate_action::{CorporateActionFilter, StockSplitSource};
16
17use super::error::MassiveError;
18use super::reference::{SplitQuery, StockSplit};
19use super::rest::MassiveRestClient;
20
21impl StockSplitSource for MassiveRestClient {
22    type Error = MassiveError;
23
24    /// Fetch stock splits for `filter`'s symbols and effective-date range, mapped to
25    /// [`CorporateAction<SmolStr>`] descriptors.
26    ///
27    /// `effective_date` provenance: this impl maps it onto Massive's `execution_date` — the split's
28    /// market execution date, satisfying the [`StockSplitSource::fetch_splits`] `effective_date`
29    /// contract. (`StockSplit` exposes no other date field, so there is no ambiguity to resolve.)
30    /// Massive's `/v3/reference/splits` endpoint filters by a **single** ticker, so a multi-symbol filter
31    /// fans out into one query per symbol, concatenated into a single stream in `symbols` order; an
32    /// **empty** `symbols` list issues one unfiltered query (all splits in the date range, subject
33    /// to provider page limits). Splits whose raw `split_to` / `split_from` yield a degenerate ratio
34    /// (zero, negative, or division by zero) are logged and skipped — see
35    /// [`CorporateActionKind::stock_split`].
36    fn fetch_splits(
37        &self,
38        filter: &CorporateActionFilter,
39    ) -> impl Stream<Item = Result<CorporateAction<SmolStr>, Self::Error>> + Send {
40        // Build owned queries eagerly so the returned stream borrows only `self`, not `filter`.
41        let queries = build_split_queries(filter);
42
43        try_stream! {
44            for query in &queries {
45                // Calls the *inherent* `MassiveRestClient::fetch_splits_raw` (taking `&SplitQuery`),
46                // the lower-level raw-split stream this impl adapts. Its distinct name keeps the
47                // dispatch explicit — no collision with this trait's `fetch_splits`, so no recursion.
48                let raw = self.fetch_splits_raw(query);
49                futures::pin_mut!(raw);
50                while let Some(split) = raw.next().await {
51                    if let Some(action) = map_split(split?) {
52                        yield action;
53                    }
54                }
55            }
56        }
57    }
58}
59
60/// Translate a [`CorporateActionFilter`] into one or more Massive [`SplitQuery`]s (one per symbol,
61/// or a single date-only query when no symbols are given).
62fn build_split_queries(filter: &CorporateActionFilter) -> Vec<SplitQuery> {
63    // These queries only ever set the `execution_date_gte`/`lte` range — never the exact
64    // `execution_date` field. `SplitQuery::validate()` (run inside the inner `fetch_splits_raw` stream)
65    // rejects only the exact+range combination, so queries built here can never trip it.
66    let with_dates = |mut query: SplitQuery| {
67        if let Some(start) = filter.start {
68            query = query.execution_date_gte(start);
69        }
70        if let Some(end) = filter.end {
71            query = query.execution_date_lte(end);
72        }
73        query
74    };
75
76    if filter.symbols.is_empty() {
77        vec![with_dates(SplitQuery::new())]
78    } else {
79        filter
80            .symbols
81            .iter()
82            .map(|symbol| with_dates(SplitQuery::new().ticker(symbol.as_str())))
83            .collect()
84    }
85}
86
87/// Map a raw [`StockSplit`] to a [`CorporateAction`] descriptor via the shared ratio helper.
88/// Returns `None` (and logs) for a degenerate ratio so the caller's stream simply omits it.
89fn map_split(split: StockSplit) -> Option<CorporateAction<SmolStr>> {
90    match CorporateActionKind::stock_split(split.split_to, split.split_from) {
91        // `effective_date` maps to Massive's `execution_date` — the split's market execution date,
92        // per the `StockSplitSource::fetch_splits` contract.
93        Some(kind) => Some(CorporateAction::new(
94            SmolStr::from(split.ticker),
95            kind,
96            Some(split.execution_date),
97        )),
98        None => {
99            warn!(
100                ticker = %split.ticker,
101                split_to = %split.split_to,
102                split_from = %split.split_from,
103                "Massive stock split has a degenerate ratio; skipping",
104            );
105            None
106        }
107    }
108}
109
110#[cfg(test)]
111#[allow(clippy::unwrap_used)] // Tests should panic on unexpected values
112mod tests {
113    use super::*;
114    use chrono::NaiveDate;
115    use rust_decimal::Decimal;
116    use rustrade_instrument::corporate_action::SplitRatio;
117
118    #[test]
119    fn build_split_queries_fans_out_per_symbol() {
120        let filter = CorporateActionFilter::new(
121            vec![SmolStr::new("AAPL"), SmolStr::new("NVDA")],
122            NaiveDate::from_ymd_opt(2020, 1, 1),
123            NaiveDate::from_ymd_opt(2024, 12, 31),
124        );
125
126        let queries = build_split_queries(&filter);
127        assert_eq!(queries.len(), 2);
128        assert_eq!(queries[0].ticker.as_deref(), Some("AAPL"));
129        assert_eq!(queries[1].ticker.as_deref(), Some("NVDA"));
130        // Date bounds are applied to every per-symbol query.
131        assert_eq!(queries[0].execution_date_gte, filter.start);
132        assert_eq!(queries[0].execution_date_lte, filter.end);
133    }
134
135    #[test]
136    fn build_split_queries_empty_symbols_is_one_unfiltered_query() {
137        let filter = CorporateActionFilter::new(vec![], NaiveDate::from_ymd_opt(2023, 1, 1), None);
138
139        let queries = build_split_queries(&filter);
140        assert_eq!(queries.len(), 1);
141        assert_eq!(queries[0].ticker, None);
142        assert_eq!(queries[0].execution_date_gte, filter.start);
143        assert_eq!(queries[0].execution_date_lte, None);
144    }
145
146    #[test]
147    fn map_split_computes_forward_ratio() {
148        let split = StockSplit {
149            ticker: "AAPL".to_string(),
150            execution_date: NaiveDate::from_ymd_opt(2020, 8, 31).unwrap(),
151            split_to: Decimal::from(4),
152            split_from: Decimal::from(1),
153        };
154
155        let action = map_split(split).unwrap();
156        assert_eq!(action.instrument, SmolStr::new("AAPL"));
157        assert_eq!(
158            action.kind,
159            CorporateActionKind::StockSplit {
160                ratio: SplitRatio::new(Decimal::from(4)).unwrap()
161            }
162        );
163        // Pins the provenance: `effective_date` is mapped from the raw split's `execution_date`.
164        assert_eq!(action.effective_date, NaiveDate::from_ymd_opt(2020, 8, 31));
165    }
166
167    #[test]
168    fn map_split_drops_degenerate_ratio() {
169        let split = StockSplit {
170            ticker: "BAD".to_string(),
171            execution_date: NaiveDate::from_ymd_opt(2023, 1, 1).unwrap(),
172            split_to: Decimal::from(1),
173            split_from: Decimal::ZERO,
174        };
175        assert!(map_split(split).is_none());
176    }
177
178    #[tokio::test]
179    async fn fetch_splits_via_trait_object_is_stream_of_descriptors() {
180        // A tiny in-test source confirms the trait shape drives end-to-end without a live client.
181        struct OneSplit;
182        impl StockSplitSource for OneSplit {
183            type Error = std::convert::Infallible;
184            fn fetch_splits(
185                &self,
186                _filter: &CorporateActionFilter,
187            ) -> impl Stream<Item = Result<CorporateAction<SmolStr>, Self::Error>> + Send
188            {
189                futures::stream::iter([Ok(CorporateAction::new(
190                    SmolStr::new("TSLA"),
191                    CorporateActionKind::StockSplit {
192                        ratio: SplitRatio::new(Decimal::from(3)).unwrap(),
193                    },
194                    NaiveDate::from_ymd_opt(2022, 8, 25),
195                ))])
196            }
197        }
198
199        let actions: Vec<_> = OneSplit
200            .fetch_splits(&CorporateActionFilter::default())
201            .collect()
202            .await;
203        assert_eq!(actions.len(), 1);
204        assert_eq!(
205            actions[0].as_ref().unwrap().instrument,
206            SmolStr::new("TSLA")
207        );
208    }
209}
rustrade_data/exchange/massive/corporate_action.rs

rustrade_data/exchange/massive/
corporate_action.rs
