Brummel/Rust-Tests
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix M1 and tick data loading order

Browse Source
This commit is contained in:
Michael Schimmel
2026-02-16 18:00:00 +01:00
parent f8efc672a0
commit 6974bbaf7f
1 changed files with 133 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main.rs
+133 -44
View File
@@ -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<T: bytemuck::Pod + Send + Sync + 'static> DataServer<T> {
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<T> = raw_data
.iter()
.map(|&r| r.to_data_point().into())
.collect();
let mapped_data: Vec<T> =
raw_data.iter().map(|&r| r.to_data_point().into()).collect();
return Ok(Arc::new(mapped_data));
}
}
@@ -156,7 +154,10 @@ impl<T: bytemuck::Pod + Send + Sync + 'static> DataServer<T> {
Ok(shared_data)
}
async fn read_from_zip<R: bytemuck::Pod + Send + 'static>(&self, path: &Path) -> anyhow::Result<Vec<R>> {
async fn read_from_zip<R: bytemuck::Pod + Send + 'static>(
&self,
path: &Path,
) -> anyhow::Result<Vec<R>> {
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<tokio::task::JoinHandle<anyhow::Result<Arc<Vec<DataPoint<OhlcItem>>>>>> = None;
let mut next_load: Option<
tokio::task::JoinHandle<anyhow::Result<Arc<Vec<DataPoint<OhlcItem>>>>>,
> = 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::<M1Record>(&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::<M1Record>(&next_file).await }));
next_load = Some(tokio::spawn(async move {
s.load_data::<M1Record>(&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<tokio::task::JoinHandle<anyhow::Result<Arc<Vec<DataPoint<TickRecord>>>>>> = None;
let mut next_load: Option<
tokio::task::JoinHandle<anyhow::Result<Arc<Vec<DataPoint<TickRecord>>>>>,
> = 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::<TickRecord>(&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::<TickRecord>(&next_file).await }));
next_load = Some(tokio::spawn(async move {
s.load_data::<TickRecord>(&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));