init

src/main.rs

@@ -0,0 +1,258 @@
+mod records;
+
+use std::io::Read;
+use std::path::{Path, PathBuf};
+use tokio::fs;
+use zip::ZipArchive;
+use bytemuck::cast_slice;
+use crate::records::{M1Record, TickRecord, OhlcItem, DataPoint};
+
+use std::collections::HashMap;
+use std::sync::{Arc, RwLock};
+use std::time::Instant;
+use sysinfo::System;
+use chrono::Local;
+
+#[derive(Debug, Clone)]
+pub struct DataFile {
+    pub path: PathBuf,
+    pub symbol: String,
+    pub year: i32,
+    pub month: i32,
+}
+
+struct CacheEntry<T> {
+    data: Arc<Vec<T>>,
+    last_accessed: Instant,
+}
+
+pub struct DataServer {
+    base_path: PathBuf,
+    symbols: RwLock<HashMap<String, Vec<DataFile>>>,
+    m1_cache: RwLock<HashMap<PathBuf, CacheEntry<DataPoint<OhlcItem>>>>,
+    tick_cache: RwLock<HashMap<PathBuf, CacheEntry<TickRecord>>>,
+    sys: RwLock<System>,
+}
+
+fn log(msg: &str) {
+    println!("[{}] {}", Local::now().format("%H:%M:%S%.3f"), msg);
+}
+
+impl DataServer {
+    pub fn new(path: impl AsRef<Path>) -> Self {
+        let mut sys = System::new_all();
+        sys.refresh_memory();
+        Self {
+            base_path: path.as_ref().to_path_buf(),
+            symbols: RwLock::new(HashMap::new()),
+            m1_cache: RwLock::new(HashMap::new()),
+            tick_cache: RwLock::new(HashMap::new()),
+            sys: RwLock::new(sys),
+        }
+    }
+
+    pub async fn update_symbols(&self) -> tokio::io::Result<()> {
+        let mut entries = fs::read_dir(&self.base_path).await?;
+        let mut temp_map: HashMap<String, Vec<DataFile>> = HashMap::new();
+
+        while let Some(entry) = entries.next_entry().await? {
+            let path = entry.path();
+            if let Some(file_name) = path.file_name().and_then(|n| n.to_str()) {
+                if let Some(data_file) = self.parse_file_name(&path, file_name) {
+                    temp_map.entry(data_file.symbol.clone()).or_default().push(data_file);
+                }
+            }
+        }
+
+        // Sort files within each symbol by year/month
+        for files in temp_map.values_mut() {
+            files.sort_by(|a, b| a.year.cmp(&b.year).then(a.month.cmp(&b.month)));
+        }
+
+        let mut symbols = self.symbols.write().unwrap();
+        *symbols = temp_map;
+        Ok(())
+    }
+
+    pub fn enumerate_symbols(&self) -> Vec<String> {
+        let symbols = self.symbols.read().unwrap();
+        let mut keys: Vec<String> = symbols.keys().cloned().collect();
+        keys.sort();
+        keys
+    }
+
+    fn parse_file_name(&self, path: &Path, file_name: &str) -> Option<DataFile> {
+        let stem = Path::new(file_name).file_stem()?.to_str()?;
+        let ext = Path::new(file_name).extension()?.to_str()?;
+        
+        let parts: Vec<&str> = stem.split('_').collect();
+        if parts.len() < 3 { return None; }
+
+        let month = parts.last()?.parse::<i32>().ok()?;
+        let year = parts[parts.len() - 2].parse::<i32>().ok()?;
+        let symbol = format!("{}.{}", parts[..parts.len() - 2].join("_"), ext);
+
+        Some(DataFile {
+            path: path.to_path_buf(),
+            symbol,
+            year,
+            month,
+        })
+    }
+
+    fn check_memory_and_prune(&self) {
+        let mut sys = self.sys.write().unwrap();
+        sys.refresh_memory();
+        
+        let available = sys.available_memory();
+        let total = sys.total_memory();
+        let threshold = (total / 10).min(2 * 1024 * 1024 * 1024); 
+
+        if available < threshold {
+            self.prune_oldest();
+        }
+    }
+
+    fn prune_oldest(&self) {
+        let mut m1 = self.m1_cache.write().unwrap();
+        let mut tick = self.tick_cache.write().unwrap();
+        
+        let total_before = m1.len() + tick.len();
+        let mut entries: Vec<(PathBuf, Instant, bool)> = m1.iter()
+            .map(|(p, e)| (p.clone(), e.last_accessed, true))
+            .chain(tick.iter().map(|(p, e)| (p.clone(), e.last_accessed, false)))
+            .collect();
+
+        entries.sort_by_key(|e| e.1);
+        let to_remove = (entries.len() / 10).max(1);
+
+        for i in 0..to_remove.min(entries.len()) {
+            let (path, _, is_m1) = &entries[i];
+            if *is_m1 { m1.remove(path); } else { tick.remove(path); }
+        }
+        println!("\n[Cache] Memory low! Pruned {} items. ({} -> {})", to_remove, total_before, m1.len() + tick.len());
+    }
+
+    pub async fn load_m1_data(&self, data_file: &DataFile) -> anyhow::Result<Arc<Vec<DataPoint<OhlcItem>>>> {
+        {
+            let mut cache = self.m1_cache.write().unwrap();
+            if let Some(entry) = cache.get_mut(&data_file.path) {
+                entry.last_accessed = Instant::now();
+                return Ok(Arc::clone(&entry.data));
+            }
+        }
+
+        self.check_memory_and_prune();
+        
+        let raw_data = self.read_from_zip::<M1Record>(&data_file.path).await?;
+        let mapped_data: Vec<DataPoint<OhlcItem>> = raw_data.into_iter()
+            .map(|r| r.to_ohlc())
+            .collect();
+            
+        let shared_data = Arc::new(mapped_data);
+        
+        let mut cache = self.m1_cache.write().unwrap();
+        cache.insert(data_file.path.clone(), CacheEntry {
+            data: Arc::clone(&shared_data),
+            last_accessed: Instant::now(),
+        });
+        
+        Ok(shared_data)
+    }
+
+    pub async fn load_tick_data(&self, data_file: &DataFile) -> anyhow::Result<Arc<Vec<TickRecord>>> {
+        {
+            let mut cache = self.tick_cache.write().unwrap();
+            if let Some(entry) = cache.get_mut(&data_file.path) {
+                entry.last_accessed = Instant::now();
+                return Ok(Arc::clone(&entry.data));
+            }
+        }
+
+        self.check_memory_and_prune();
+        let data = self.read_from_zip::<TickRecord>(&data_file.path).await?;
+        let shared_data = Arc::new(data);
+        
+        let mut cache = self.tick_cache.write().unwrap();
+        cache.insert(data_file.path.clone(), CacheEntry {
+            data: Arc::clone(&shared_data),
+            last_accessed: Instant::now(),
+        });
+        
+        Ok(shared_data)
+    }
+
+    async fn read_from_zip<R: bytemuck::Pod>(&self, path: &Path) -> anyhow::Result<Vec<R>> {
+        let file = std::fs::File::open(path)?;
+        let mut archive = ZipArchive::new(file)?;
+
+        for i in 0..archive.len() {
+            let mut file = archive.by_index(i)?;
+            if file.name().ends_with(".bin") {
+                let size = file.size() as usize;
+                let record_size = std::mem::size_of::<R>();
+                if size % record_size != 0 {
+                    return Err(anyhow::anyhow!("Invalid file size for record type"));
+                }
+                
+                let mut bin_content = vec![0u8; size];
+                file.read_exact(&mut bin_content)?;
+                let records: &[R] = cast_slice(&bin_content);
+                return Ok(records.to_vec());
+            }
+        }
+        Err(anyhow::anyhow!("No .bin file found"))
+    }
+
+    pub async fn process_symbol_m1(&self, symbol: &str) -> anyhow::Result<()> {
+        let files = {
+            let symbols = self.symbols.read().unwrap();
+            symbols.get(symbol).cloned().ok_or_else(|| anyhow::anyhow!("Symbol not found"))?
+        };
+
+        log(&format!(">>> Loading Symbol: {} ({} files)", symbol, files.len()));
+
+        for file in files {
+            if file.symbol.ends_with(".m1") {
+                let _ = self.load_m1_data(&file).await?;
+            }
+        }
+        Ok(())
+    }
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let server = Arc::new(DataServer::new(r"\\COFFEE\TickData\Pepperstone"));
+    
+    log("Indexing symbols...");
+    server.update_symbols().await?;
+    
+    let symbols = server.enumerate_symbols();
+    log(&format!("Found {} symbols.", symbols.len()));
+
+    let start_all = Instant::now();
+    let mut set = tokio::task::JoinSet::new();
+    let max_concurrency = 8; // Lower concurrency since we load full symbols now
+
+    for symbol in symbols {
+        let s = Arc::clone(&server);
+        
+        while set.len() >= max_concurrency {
+            set.join_next().await;
+        }
+
+        set.spawn(async move {
+            let _ = s.process_symbol_m1(&symbol).await;
+        });
+    }
+
+    while let Some(_) = set.join_next().await {}
+
+    log(&format!("Finished loading all symbols in {:?}", start_all.elapsed()));
+    
+    let m1_count = server.m1_cache.read().unwrap().len();
+    log(&format!("Final Cache State: {} M1 files in RAM.", m1_count));
+
+    Ok(())
+}

src/records.rs

@@ -0,0 +1,66 @@
+use bytemuck::{Pod, Zeroable};
+
+// Match Delphi TM1Record (Pack = 1, 48 Bytes)
+#[repr(C, packed)]
+#[derive(Debug, Copy, Clone, Pod, Zeroable)]
+pub struct M1Record {
+    pub time: f64,
+    pub open: f64,
+    pub high: f64,
+    pub low: f64,
+    pub close: f64,
+    pub spread: f32,
+    pub volume: i32,
+}
+
+// Match Delphi TTickRecord (Pack = 1, 24 Bytes)
+#[repr(C, packed)]
+#[derive(Debug, Copy, Clone, Pod, Zeroable)]
+pub struct TickRecord {
+    pub time: f64,
+    pub ask: f64,
+    pub bid: f64,
+}
+
+// Match Delphi TOhlcItem (Pack = 1)
+#[repr(C, packed)]
+#[derive(Debug, Copy, Clone, Pod, Zeroable)]
+pub struct OhlcItem {
+    pub open: f64,
+    pub high: f64,
+    pub low: f64,
+    pub close: f64,
+    pub volume: i32,
+}
+
+// Internal data point structure
+#[repr(C, packed)]
+#[derive(Debug, Copy, Clone, Pod, Zeroable)]
+pub struct DataPoint<T> {
+    pub time: f64,
+    pub data: T,
+}
+
+impl M1Record {
+    pub fn to_ohlc(self) -> DataPoint<OhlcItem> {
+        DataPoint {
+            time: self.time,
+            data: OhlcItem {
+                open: self.open,
+                high: self.high,
+                low: self.low,
+                close: self.close,
+                volume: self.volume,
+            },
+        }
+    }
+}
+
+impl TickRecord {
+    pub fn to_data_point(self) -> DataPoint<TickRecord> {
+        DataPoint {
+            time: self.time,
+            data: self,
+        }
+    }
+}