From 6974bbaf7f004ede59aad0a9a4888d6706f8bc8c Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Mon, 16 Feb 2026 18:00:00 +0100 Subject: [PATCH] Fix M1 and tick data loading order --- src/main.rs | 177 +++++++++++++++++++++++++++++++++++++++------------- 1 file changed, 133 insertions(+), 44 deletions(-) diff --git a/src/main.rs b/src/main.rs index c83c13a..c0f871a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,15 +1,15 @@ mod records; -use chrono::{DateTime, Utc}; use crate::records::{DataPoint, M1Record, OhlcItem, TickRecord}; +use chrono::{DateTime, Utc}; fn format_ts(ts: f64) -> String { - // Delphi TDateTime: Days since 1899-12-30. + // Delphi TDateTime: Days since 1899-12-30. // Unix Epoch (1970-01-01) is 25569.0 days after 1899-12-30. let unix_secs = (ts - 25569.0) * 86400.0; let secs = unix_secs.floor() as i64; let nanos = ((unix_secs - unix_secs.floor()) * 1_000_000_000.0) as u32; - + if let Some(dt) = DateTime::from_timestamp(secs, nanos) { dt.format("%Y-%m-%d %H:%M:%S%.6f").to_string() } else { @@ -131,13 +131,11 @@ impl DataServer { let mut bin_content = vec![0u8; file.size() as usize]; file.read_exact(&mut bin_content)?; let raw_data: &[R] = cast_slice(&bin_content); - + // Perform mapping inside blocking task - let mapped_data: Vec = raw_data - .iter() - .map(|&r| r.to_data_point().into()) - .collect(); - + let mapped_data: Vec = + raw_data.iter().map(|&r| r.to_data_point().into()).collect(); + return Ok(Arc::new(mapped_data)); } } @@ -156,7 +154,10 @@ impl DataServer { Ok(shared_data) } - async fn read_from_zip(&self, path: &Path) -> anyhow::Result> { + async fn read_from_zip( + &self, + path: &Path, + ) -> anyhow::Result> { let path = path.to_path_buf(); tokio::task::spawn_blocking(move || { let file = std::fs::File::open(&path)?; @@ -223,10 +224,7 @@ impl DataViewerApp { let m1_clone = Arc::clone(&m1_server); let tick_clone = Arc::clone(&tick_server); let symbols = rt.block_on(async { - let _ = tokio::join!( - m1_clone.update_symbols(), - tick_clone.update_symbols() - ); + let _ = tokio::join!(m1_clone.update_symbols(), tick_clone.update_symbols()); m1_clone.enumerate_symbols() }); @@ -248,9 +246,15 @@ impl DataViewerApp { self.is_loading = true; self.status_msg = format!("Loading {}...", symbol); self.loaded_data = if symbol.ends_with(".m1") { - LoadedData::M1 { chunks: Vec::new(), total_count: 0 } + LoadedData::M1 { + chunks: Vec::new(), + total_count: 0, + } } else { - LoadedData::Ticks { chunks: Vec::new(), total_count: 0 } + LoadedData::Ticks { + chunks: Vec::new(), + total_count: 0, + } }; let m1_server = Arc::clone(&self.m1_server); @@ -264,11 +268,16 @@ impl DataViewerApp { let symbols = m1_server.symbols.read().unwrap(); symbols.get(&sym_clone).cloned().unwrap_or_default() }; - let mut next_load: Option>>>>> = None; + let mut next_load: Option< + tokio::task::JoinHandle>>>>, + > = None; for i in 0..files.len() { let data_res = if let Some(handle) = next_load.take() { - handle.await.map_err(|e| anyhow::anyhow!("Join error: {}", e)).and_then(|res| res) + handle + .await + .map_err(|e| anyhow::anyhow!("Join error: {}", e)) + .and_then(|res| res) } else { m1_server.load_data::(&files[i]).await }; @@ -276,7 +285,9 @@ impl DataViewerApp { if i + 1 < files.len() { let next_file = files[i + 1].clone(); let s = Arc::clone(&m1_server); - next_load = Some(tokio::spawn(async move { s.load_data::(&next_file).await })); + next_load = Some(tokio::spawn(async move { + s.load_data::(&next_file).await + })); } if let Ok(data) = data_res { @@ -288,11 +299,16 @@ impl DataViewerApp { let symbols = tick_server.symbols.read().unwrap(); symbols.get(&sym_clone).cloned().unwrap_or_default() }; - let mut next_load: Option>>>>> = None; + let mut next_load: Option< + tokio::task::JoinHandle>>>>, + > = None; for i in 0..files.len() { let data_res = if let Some(handle) = next_load.take() { - handle.await.map_err(|e| anyhow::anyhow!("Join error: {}", e)).and_then(|res| res) + handle + .await + .map_err(|e| anyhow::anyhow!("Join error: {}", e)) + .and_then(|res| res) } else { tick_server.load_data::(&files[i]).await }; @@ -300,7 +316,9 @@ impl DataViewerApp { if i + 1 < files.len() { let next_file = files[i + 1].clone(); let s = Arc::clone(&tick_server); - next_load = Some(tokio::spawn(async move { s.load_data::(&next_file).await })); + next_load = Some(tokio::spawn(async move { + s.load_data::(&next_file).await + })); } if let Ok(data) = data_res { @@ -325,13 +343,21 @@ impl eframe::App for DataViewerApp { if let Some(chunk) = msg { match chunk { DataChunk::M1(data) => { - if let LoadedData::M1 { chunks, total_count } = &mut self.loaded_data { + if let LoadedData::M1 { + chunks, + total_count, + } = &mut self.loaded_data + { *total_count += data.len(); chunks.push(data); } } DataChunk::Ticks(data) => { - if let LoadedData::Ticks { chunks, total_count } = &mut self.loaded_data { + if let LoadedData::Ticks { + chunks, + total_count, + } = &mut self.loaded_data + { *total_count += data.len(); chunks.push(data); } @@ -381,9 +407,20 @@ impl eframe::App for DataViewerApp { }); match &self.loaded_data { - LoadedData::M1 { chunks, total_count } if *total_count > 0 => { - let last_ts = chunks.last().and_then(|c| c.last()).map(|d| d.time).unwrap_or(0.0); - ui.label(format!("Records: {} | Latest: {}", total_count, format_ts(last_ts))); + LoadedData::M1 { + chunks, + total_count, + } if *total_count > 0 => { + let last_ts = chunks + .last() + .and_then(|c| c.last()) + .map(|d| d.time) + .unwrap_or(0.0); + ui.label(format!( + "Records: {} | Latest: {}", + total_count, + format_ts(last_ts) + )); // Get last 200 points let mut last_points = Vec::with_capacity(200); @@ -395,18 +432,29 @@ impl eframe::App for DataViewerApp { last_points.push(&chunk[i]); } count_needed -= take; - if count_needed == 0 { break; } + if count_needed == 0 { + break; + } } last_points.reverse(); // Render Chart - let (rect, _response) = ui.allocate_at_least(egui::vec2(ui.available_width(), ui.available_height()), egui::Sense::hover()); + let (rect, _response) = ui.allocate_at_least( + egui::vec2(ui.available_width(), ui.available_height()), + egui::Sense::hover(), + ); let painter = ui.painter_at(rect); painter.rect_filled(rect, 0.0, egui::Color32::from_black_alpha(20)); if !last_points.is_empty() { - let min_y = last_points.iter().map(|d| d.data.low).fold(f64::INFINITY, f64::min); - let max_y = last_points.iter().map(|d| d.data.high).fold(f64::NEG_INFINITY, f64::max); + let min_y = last_points + .iter() + .map(|d| d.data.low) + .fold(f64::INFINITY, f64::min); + let max_y = last_points + .iter() + .map(|d| d.data.high) + .fold(f64::NEG_INFINITY, f64::max); let range_y = (max_y - min_y).max(0.00001); let to_screen_y = |y: f64| { @@ -416,21 +464,45 @@ impl eframe::App for DataViewerApp { let x_step = rect.width() / 200.0; for (i, p) in last_points.iter().enumerate() { let x = rect.left() + i as f32 * x_step + x_step * 0.5; - let color = if p.data.close >= p.data.open { egui::Color32::GREEN } else { egui::Color32::RED }; - + let color = if p.data.close >= p.data.open { + egui::Color32::GREEN + } else { + egui::Color32::RED + }; + // Wick - painter.line_segment([egui::pos2(x, to_screen_y(p.data.high)), egui::pos2(x, to_screen_y(p.data.low))], (1.0, color)); + painter.line_segment( + [ + egui::pos2(x, to_screen_y(p.data.high)), + egui::pos2(x, to_screen_y(p.data.low)), + ], + (1.0, color), + ); // Body let y_open = to_screen_y(p.data.open); let y_close = to_screen_y(p.data.close); - let body_rect = egui::Rect::from_x_y_ranges(x - x_step * 0.4..=x + x_step * 0.4, y_open.min(y_close)..=y_open.max(y_close)); + let body_rect = egui::Rect::from_x_y_ranges( + x - x_step * 0.4..=x + x_step * 0.4, + y_open.min(y_close)..=y_open.max(y_close), + ); painter.rect_filled(body_rect, 0.0, color); } } } - LoadedData::Ticks { chunks, total_count } if *total_count > 0 => { - let last_ts = chunks.last().and_then(|c| c.last()).map(|d| d.time).unwrap_or(0.0); - ui.label(format!("Records: {} | Latest: {}", total_count, format_ts(last_ts))); + LoadedData::Ticks { + chunks, + total_count, + } if *total_count > 0 => { + let last_ts = chunks + .last() + .and_then(|c| c.last()) + .map(|d| d.time) + .unwrap_or(0.0); + ui.label(format!( + "Records: {} | Latest: {}", + total_count, + format_ts(last_ts) + )); let mut last_points = Vec::with_capacity(200); let mut count_needed = 200; @@ -441,17 +513,28 @@ impl eframe::App for DataViewerApp { last_points.push(&chunk[i]); } count_needed -= take; - if count_needed == 0 { break; } + if count_needed == 0 { + break; + } } last_points.reverse(); - let (rect, _response) = ui.allocate_at_least(egui::vec2(ui.available_width(), ui.available_height()), egui::Sense::hover()); + let (rect, _response) = ui.allocate_at_least( + egui::vec2(ui.available_width(), ui.available_height()), + egui::Sense::hover(), + ); let painter = ui.painter_at(rect); painter.rect_filled(rect, 0.0, egui::Color32::from_black_alpha(20)); if !last_points.is_empty() { - let min_y = last_points.iter().map(|d| d.data.bid.min(d.data.ask)).fold(f64::INFINITY, f64::min); - let max_y = last_points.iter().map(|d| d.data.bid.max(d.data.ask)).fold(f64::NEG_INFINITY, f64::max); + let min_y = last_points + .iter() + .map(|d| d.data.bid.min(d.data.ask)) + .fold(f64::INFINITY, f64::min); + let max_y = last_points + .iter() + .map(|d| d.data.bid.max(d.data.ask)) + .fold(f64::NEG_INFINITY, f64::max); let range_y = (max_y - min_y).max(0.00001); let to_screen_y = |y: f64| { @@ -468,10 +551,16 @@ impl eframe::App for DataViewerApp { let y_bid = to_screen_y(p.data.bid); if let Some(prev_x) = last_ask { - painter.line_segment([prev_x, egui::pos2(x, y_ask)], (1.0, egui::Color32::LIGHT_BLUE)); + painter.line_segment( + [prev_x, egui::pos2(x, y_ask)], + (1.0, egui::Color32::LIGHT_BLUE), + ); } if let Some(prev_x) = last_bid { - painter.line_segment([prev_x, egui::pos2(x, y_bid)], (1.0, egui::Color32::LIGHT_RED)); + painter.line_segment( + [prev_x, egui::pos2(x, y_bid)], + (1.0, egui::Color32::LIGHT_RED), + ); } last_ask = Some(egui::pos2(x, y_ask)); last_bid = Some(egui::pos2(x, y_bid));