flowsurface_data/chart/kline/
fetch_plan.rs1use super::IndicatorHistoryRequirement;
2use super::series::KlineSeries;
3use crate::chart::Basis;
4use exchange::adapter::Exchange;
5use exchange::{TickerInfo, Timeframe, UnixMs};
6
7const DEFAULT_FETCH_BARS: u64 = 500;
8
9#[derive(Debug, Clone, Copy, PartialEq, Eq)]
10pub enum FetchRange {
11 Kline(UnixMs, UnixMs),
12 ProfileKline {
13 timeframe: Timeframe,
14 from: UnixMs,
15 to: UnixMs,
16 },
17 OpenInterest {
18 from: UnixMs,
19 to: UnixMs,
20 },
21 Trades(UnixMs, UnixMs),
22}
23
24impl FetchRange {
25 pub const fn open_interest(from: UnixMs, to: UnixMs) -> Self {
26 Self::OpenInterest { from, to }
27 }
28}
29
30#[derive(Debug, Clone, PartialEq, Eq)]
31pub struct FetchBatch {
32 pub range: FetchRange,
33 pub tickers: Vec<TickerInfo>,
34}
35
36#[derive(Clone, Copy)]
37pub struct FetchPlanInput<'a> {
38 pub basis: Basis,
39 pub horizontal_offset: f32,
40 pub base_ticker: TickerInfo,
41 pub series: &'a [KlineSeries],
42 pub open_interest_required: bool,
43 pub trade_data_required: bool,
44 pub trade_fetch_enabled: bool,
45 pub visible_points: u64,
46 pub history_requirement: IndicatorHistoryRequirement,
47 pub now: UnixMs,
48}
49
50impl FetchPlanInput<'_> {
51 pub fn desired_fetch_batches(self) -> Vec<FetchBatch> {
52 let Basis::Time(timeframe) = self.basis else {
53 return Vec::new();
54 };
55
56 let dt = timeframe.to_milliseconds().max(1);
57 let visible_points = self.visible_points.max(2);
58
59 let indicator_warmup_bars = self.history_requirement.warmup_bars();
60
61 let required_history_bars =
62 indicator_warmup_bars.max(self.history_requirement.history_bars());
63 let base_fetch_bars = DEFAULT_FETCH_BARS.max(visible_points + required_history_bars);
64 let comparison_fetch_bars = DEFAULT_FETCH_BARS.max(visible_points);
65 let base_span = base_fetch_bars.saturating_mul(dt);
66 let comparison_span = comparison_fetch_bars.saturating_mul(dt);
67 let initial_end = self.now;
68
69 let mut batches = Vec::new();
70 let mut empty_base = false;
71 let mut empty_comparisons = Vec::new();
72
73 for series in self.series {
74 if series.is_empty() {
75 if series.ticker_info == self.base_ticker {
76 empty_base = true;
77 } else {
78 empty_comparisons.push(series.ticker_info);
79 }
80 }
81 }
82
83 if empty_base {
84 let end = initial_end;
85 let start = end.saturating_sub(comparison_span);
86 batches.push(FetchBatch {
87 range: FetchRange::Kline(start, end),
88 tickers: vec![self.base_ticker],
89 });
90 }
91
92 if !empty_comparisons.is_empty() {
93 let end = initial_end;
94 batches.push(FetchBatch {
95 range: FetchRange::Kline(end.saturating_sub(comparison_span), end),
96 tickers: empty_comparisons,
97 });
98 }
99
100 if let Some((window_min, window_max)) = visible_window(
101 self.series,
102 self.horizontal_offset,
103 timeframe,
104 self.visible_points,
105 ) {
106 let indicator_warmup_start = if indicator_warmup_bars > 0 {
107 window_min
108 .saturating_sub(indicator_warmup_bars.saturating_mul(dt))
109 .floor_to(timeframe)
110 } else {
111 window_min
112 };
113 let warmup_start = self
114 .history_requirement
115 .history_start()
116 .map(|start| start.min(indicator_warmup_start))
117 .unwrap_or(indicator_warmup_start);
118
119 let mut base_boundary_fetch_pending = false;
120 for series in self.series {
121 let (target_min, fetch_span, fetching_base_history) =
122 if series.ticker_info == self.base_ticker {
123 if window_min
124 < series
125 .first_point()
126 .map(|point| point.kline.time)
127 .unwrap_or(window_min)
128 {
129 (window_min, comparison_span, false)
130 } else {
131 (warmup_start, base_span, true)
132 }
133 } else {
134 (window_min, comparison_span, false)
135 };
136
137 if let Some(series_min) = series.first_point().map(|point| point.kline.time)
138 && target_min < series_min
139 {
140 let end = series_min.floor_to(timeframe);
141 batches.push(FetchBatch {
142 range: FetchRange::Kline(end.saturating_sub(fetch_span), end),
143 tickers: vec![series.ticker_info],
144 });
145 base_boundary_fetch_pending |= fetching_base_history;
146 }
147 }
148
149 if !base_boundary_fetch_pending
150 && let Some(base_series) = self
151 .series
152 .iter()
153 .find(|series| series.ticker_info == self.base_ticker)
154 && let Some(series_earliest) =
155 base_series.first_point().map(|point| point.kline.time)
156 && let Some(series_latest) =
157 base_series.latest_point().map(|point| point.kline.time)
158 {
159 let visible_span = window_max.as_u64().saturating_sub(window_min.as_u64());
160 let prefetch_earliest = window_min.saturating_sub(visible_span);
161 let check_earliest = warmup_start.min(prefetch_earliest).max(series_earliest);
162 let check_latest = window_max.saturating_add(dt).min(series_latest);
163
164 if let Some((missing_earliest, missing_latest)) =
165 base_series.missing_kline_range(check_earliest, check_latest)
166 {
167 batches.push(FetchBatch {
168 range: FetchRange::Kline(
169 missing_earliest.saturating_sub(dt),
170 missing_latest.saturating_add(dt),
171 ),
172 tickers: vec![self.base_ticker],
173 });
174 }
175 }
176
177 if self.open_interest_required {
178 let Some(base_series) = self
179 .series
180 .iter()
181 .find(|series| series.ticker_info == self.base_ticker)
182 else {
183 return batches;
184 };
185 let Some(kline_latest) = base_series.latest_point().map(|point| point.kline.time)
186 else {
187 return batches;
188 };
189 let visible_window = visible_window(
190 self.series,
191 self.horizontal_offset,
192 timeframe,
193 self.visible_points,
194 );
195 let visible_earliest = visible_window.map(|(start, _)| start).unwrap_or(window_min);
196 let visible_span = visible_window
197 .map(|(start, end)| end.as_u64().saturating_sub(start.as_u64()))
198 .unwrap_or(base_span);
199 let prefetch_earliest = visible_earliest.saturating_sub(visible_span);
200
201 match base_series.open_interest_range() {
202 Some((oi_earliest, oi_latest)) => {
203 if visible_earliest < oi_earliest {
204 batches.push(FetchBatch {
205 range: FetchRange::open_interest(prefetch_earliest, oi_earliest),
206 tickers: vec![self.base_ticker],
207 });
208 }
209
210 if oi_latest < kline_latest {
211 let start = oi_latest.max(prefetch_earliest);
212 if start < kline_latest {
213 batches.push(FetchBatch {
214 range: FetchRange::open_interest(start, kline_latest),
215 tickers: vec![self.base_ticker],
216 });
217 }
218 }
219 }
220 None => {
221 if prefetch_earliest < kline_latest {
222 batches.push(FetchBatch {
223 range: FetchRange::open_interest(prefetch_earliest, kline_latest),
224 tickers: vec![self.base_ticker],
225 });
226 }
227 }
228 }
229 }
230 }
231
232 batches
233 }
234
235 pub fn desired_trade_fetch_range(self, target_ticker: TickerInfo) -> Option<FetchRange> {
236 if target_ticker != self.base_ticker {
237 return None;
238 }
239 let Basis::Time(timeframe) = self.basis else {
240 return None;
241 };
242
243 if !self.trade_data_required || !self.trade_fetch_enabled {
244 return None;
245 }
246
247 let target_series = self
248 .series
249 .iter()
250 .find(|series| series.ticker_info == target_ticker)?;
251 if target_series.is_empty() {
252 return None;
253 }
254
255 let dt = timeframe.to_milliseconds().max(1);
256 let visible_points = self.visible_points.max(2);
257 let (visible_earliest, visible_latest) = visible_window(
258 std::slice::from_ref(target_series),
259 self.horizontal_offset,
260 timeframe,
261 visible_points,
262 )
263 .map(|(earliest, latest)| {
264 let half_interval = dt / 2;
265 let half_visible_span = latest.saturating_diff(earliest) / 2;
266 (
267 earliest.saturating_sub(half_visible_span.saturating_add(half_interval)),
268 latest.saturating_add(half_interval),
269 )
270 })?;
271
272 if visible_earliest >= visible_latest {
273 return None;
274 }
275
276 if let Some((earliest, latest)) =
277 target_series.suggest_trade_fetch_range(visible_earliest, visible_latest)
278 {
279 return Some(FetchRange::Trades(earliest, latest));
280 }
281
282 None
283 }
284}
285
286pub fn visible_window(
287 series: &[KlineSeries],
288 horizontal_offset: f32,
289 timeframe: Timeframe,
290 visible_points: u64,
291) -> Option<(UnixMs, UnixMs)> {
292 let max_seen = series
293 .iter()
294 .filter_map(|series| series.latest_point().map(|point| point.kline.time))
295 .max()?;
296
297 let dt = timeframe.to_milliseconds().max(1);
298 let span_units = visible_points.saturating_sub(1);
299 let right_offset = (f64::from(horizontal_offset) * dt as f64)
300 .round()
301 .clamp(i64::MIN as f64, i64::MAX as f64) as i64;
302 let right = super::ChartCoord::from_unix_ms(max_seen).saturating_add_i64(right_offset);
303 let span_offset = span_units.saturating_mul(dt);
304 let left = right.saturating_sub_i64(i64::try_from(span_offset).unwrap_or(i64::MAX));
305
306 Some((
307 left.to_unix_ms_non_negative(),
308 right.to_unix_ms_non_negative(),
309 ))
310}
311
312pub fn supports_open_interest(ticker_info: TickerInfo, timeframe: Timeframe) -> bool {
314 supports_open_interest_exchange(ticker_info.exchange())
315 && supports_open_interest_timeframe(timeframe)
316}
317
318pub fn supports_open_interest_exchange(exchange: Exchange) -> bool {
319 matches!(
320 exchange,
321 Exchange::BinanceLinear
322 | Exchange::BinanceInverse
323 | Exchange::BybitLinear
324 | Exchange::BybitInverse
325 | Exchange::OkexLinear
326 | Exchange::OkexInverse
327 )
328}
329
330pub fn supports_open_interest_timeframe(timeframe: Timeframe) -> bool {
331 timeframe >= Timeframe::M5 && timeframe <= Timeframe::H4 && timeframe != Timeframe::H2
332}