using System.Globalization; using System.Runtime.CompilerServices; using System.Runtime.InteropServices; namespace QuanTAlib; /// /// Parsed OHLCV data from a CSV line. /// [StructLayout(LayoutKind.Auto)] internal readonly record struct ParsedOhlcv(long Time, double Open, double High, double Low, double Close, double Volume); /// /// Mutable state for parsing OHLCV columns. Used as ref parameter to reduce method signature size. /// [StructLayout(LayoutKind.Auto)] internal ref struct OhlcvParseState { internal long Time; internal double Open; internal double High; internal double Low; internal double Close; internal double Volume; } /// /// CSV file feed for loading historical OHLCV data. /// Loads data in constructor and streams through it with Next() or returns batches with Fetch(). /// CSV format: timestamp,open,high,low,close,volume (header required) /// Timestamp format: YYYY-MM-DD (UTC midnight assumed) /// [SkipLocalsInit] public sealed class CsvFeed : IFeed { private int _currentIndex; private TBar _currentBar; private bool _hasCurrentBar; /// /// Gets the total number of bars available in the CSV file. /// public int Count { get; } /// /// Gets the file path of the loaded CSV. /// public string FilePath { get; } /// /// Gets whether there are more bars to stream. /// public bool HasMore => _currentIndex < Count; /// /// Gets the current streaming position (0-based index). /// public int CurrentIndex => _currentIndex; /// /// Gets whether the feed has a current bar in progress. /// public bool HasCurrentBar => _hasCurrentBar; /// /// Loads CSV file and prepares data for streaming. /// Data is reversed to chronological order (oldest first). /// /// Path to CSV file /// Thrown when filePath is null or empty /// Thrown when the specified file does not exist /// Thrown when CSV file is empty or contains only header /// Thrown when CSV format is invalid public CsvFeed(string filePath) { if (string.IsNullOrWhiteSpace(filePath)) { throw new ArgumentException("File path cannot be null or empty", nameof(filePath)); } if (!File.Exists(filePath)) { throw new FileNotFoundException($"CSV file not found: {filePath}", filePath); } FilePath = filePath; Data = LoadFromCsv(filePath); Count = Data.Count; _currentIndex = 0; } /// /// Parses CSV file into TBarSeries. /// Expected format: timestamp,open,high,low,close,volume /// Memory-efficient: reads lines into list, reverses in-place (no LINQ allocations). /// private static TBarSeries LoadFromCsv(string filePath) { var dataLines = new List(); using (var reader = new StreamReader(filePath)) { var header = reader.ReadLine(); if (header is null) { throw new InvalidDataException("CSV file is empty"); } while (!reader.EndOfStream) { var line = reader.ReadLine(); if (!string.IsNullOrWhiteSpace(line)) { dataLines.Add(line); } } } if (dataLines.Count == 0) { throw new InvalidDataException("CSV file contains only header, no data"); } // Reverse in-place to chronological order (oldest first) dataLines.Reverse(); var series = new TBarSeries(dataLines.Count); // Pre-allocate arrays for bulk loading (SoA layout) long[] t = new long[dataLines.Count]; double[] o = new double[dataLines.Count]; double[] h = new double[dataLines.Count]; double[] l = new double[dataLines.Count]; double[] c = new double[dataLines.Count]; double[] v = new double[dataLines.Count]; for (int i = 0; i < dataLines.Count; i++) { var line = dataLines[i]; // After Reverse(), index i corresponds to original index (dataLines.Count - 1 - i) int originalIndex = dataLines.Count - 1 - i; int originalLineNumber = originalIndex + 2; // +2 for header and 1-based line numbers var parsed = ParseCsvLine(line, originalLineNumber); t[i] = parsed.Time; o[i] = parsed.Open; h[i] = parsed.High; l[i] = parsed.Low; c[i] = parsed.Close; v[i] = parsed.Volume; } // Bulk add to series series.Add(t, o, h, l, c, v); return series; } /// /// Parses a single CSV line into OHLCV components. /// [MethodImpl(MethodImplOptions.AggressiveInlining)] private static ParsedOhlcv ParseCsvLine(string line, int lineNumber) { // Use Span-based splitting for reduced allocations ReadOnlySpan lineSpan = line.AsSpan(); int col = 0; int start = 0; OhlcvParseState state = default; for (int i = 0; i < lineSpan.Length; i++) { if (lineSpan[i] == ',') { var segment = lineSpan[start..i].Trim(); ParseColumn(segment, col, lineNumber, line, ref state); col++; start = i + 1; } } // Process the last segment after the final comma if (start <= lineSpan.Length) { var segment = lineSpan[start..].Trim(); ParseColumn(segment, col, lineNumber, line, ref state); col++; } if (col != 6) { throw new FormatException($"Invalid CSV format at line {lineNumber}. Expected 6 columns, found {col}"); } return new ParsedOhlcv(state.Time, state.Open, state.High, state.Low, state.Close, state.Volume); } /// /// Parses a single column value into the appropriate OHLCV field. /// [MethodImpl(MethodImplOptions.AggressiveInlining)] private static void ParseColumn( ReadOnlySpan segment, int col, int lineNumber, string line, ref OhlcvParseState state) { switch (col) { case 0: // Timestamp if (!DateTime.TryParseExact(segment, "yyyy-MM-dd", CultureInfo.InvariantCulture, DateTimeStyles.AssumeUniversal | DateTimeStyles.AdjustToUniversal, out var timestamp)) { throw new FormatException($"Failed to parse timestamp at line {lineNumber}: {line}"); } state.Time = timestamp.Ticks; break; case 1: // Open if (!double.TryParse(segment, NumberStyles.Float, CultureInfo.InvariantCulture, out state.Open)) { throw new FormatException($"Failed to parse open price at line {lineNumber}: {line}"); } break; case 2: // High if (!double.TryParse(segment, NumberStyles.Float, CultureInfo.InvariantCulture, out state.High)) { throw new FormatException($"Failed to parse high price at line {lineNumber}: {line}"); } break; case 3: // Low if (!double.TryParse(segment, NumberStyles.Float, CultureInfo.InvariantCulture, out state.Low)) { throw new FormatException($"Failed to parse low price at line {lineNumber}: {line}"); } break; case 4: // Close if (!double.TryParse(segment, NumberStyles.Float, CultureInfo.InvariantCulture, out state.Close)) { throw new FormatException($"Failed to parse close price at line {lineNumber}: {line}"); } break; case 5: // Volume if (!double.TryParse(segment, NumberStyles.Float, CultureInfo.InvariantCulture, out state.Volume)) { throw new FormatException($"Failed to parse volume at line {lineNumber}: {line}"); } break; default: // Extra columns are ignored - this handles the default case requirement break; } } /// /// Gets the next bar with full bidirectional control. /// When end of data reached, returns last bar and sets isNew=false. /// [MethodImpl(MethodImplOptions.AggressiveInlining)] public TBar Next(ref bool isNew) { if (Count == 0) { isNew = false; return default; } if (isNew || !_hasCurrentBar) { if (_currentIndex >= Count) { isNew = false; return _currentBar; } _currentBar = Data[_currentIndex]; _currentIndex++; _hasCurrentBar = true; } return _currentBar; } /// /// Gets the next bar with simple control. /// WARNING: This overload discards changes to isNew made by the internal implementation. /// Callers will not observe when streaming ends. Use Next(ref bool isNew) or check HasMore instead. /// /// Whether to advance to the next bar (true) or replay current bar (false). /// The current or next bar. /// /// Retained for backward compatibility with existing code. Deprecation is intentional to guide /// users toward the ref overload which properly signals end-of-stream conditions. /// #pragma warning disable S1133 // Deprecated code kept for backward compatibility; removal would be breaking change [Obsolete("Use Next(ref bool isNew) to observe end-of-stream, or check HasMore before calling. This overload discards the modified isNew value.")] #pragma warning restore S1133 [MethodImpl(MethodImplOptions.AggressiveInlining)] public TBar Next(bool isNew = true) { return Next(ref isNew); } /// /// Returns a filtered subset of data matching the criteria. /// Resets streaming position to start of returned data. /// /// Number of bars to retrieve (must be positive) /// Starting timestamp in ticks /// Time interval between bars /// A TBarSeries containing the matched bars /// Thrown when count is not positive public TBarSeries Fetch(int count, long startTime, TimeSpan interval) { if (count <= 0) { throw new ArgumentException("Count must be positive", nameof(count)); } var result = new TBarSeries(count); // Find starting index using binary search for better performance int startIndex = FindStartIndex(startTime); if (startIndex == -1) { return result; } // Collect bars matching interval long expectedTime = startTime; int collected = 0; long tolerance = interval.Ticks / 2; for (int i = startIndex; i < Count && collected < count; i++) { var bar = Data[i]; // Check if bar time matches expected time (within tolerance) long timeDiff = Math.Abs(bar.Time - expectedTime); if (timeDiff <= tolerance) { result.Add(bar, isNew: true); collected++; expectedTime += interval.Ticks; } else if (bar.Time > expectedTime) { // Gap in data - skip forward long gaps = (bar.Time - expectedTime) / interval.Ticks; expectedTime += gaps * interval.Ticks; if (Math.Abs(bar.Time - expectedTime) <= tolerance) { result.Add(bar, isNew: true); collected++; expectedTime += interval.Ticks; } } } // Reset streaming to start of returned data _currentIndex = startIndex; _hasCurrentBar = false; return result; } /// /// Finds the starting index for the given start time using binary search. /// [MethodImpl(MethodImplOptions.AggressiveInlining)] private int FindStartIndex(long startTime) { if (Count == 0) { return -1; } if (Data[0].Time >= startTime) { return 0; } if (Data[Count - 1].Time < startTime) { return -1; } int left = 0; int right = Count - 1; while (left < right) { int mid = left + (right - left) / 2; if (Data[mid].Time < startTime) { left = mid + 1; } else { right = mid; } } return left; } /// /// Resets the streaming position to the beginning. /// public void Reset() { _currentIndex = 0; _hasCurrentBar = false; _currentBar = default; } /// /// Resets the streaming position to a specific index. /// /// The index to reset to (must be valid) /// Thrown when index is out of range public void Reset(int index) { if (index < 0 || index > Count) { throw new ArgumentOutOfRangeException(nameof(index), index, $"Index must be between 0 and {Count}"); } _currentIndex = index; _hasCurrentBar = false; _currentBar = default; } /// /// Gets the bar at the specified index without affecting streaming position. /// /// The index of the bar to retrieve /// The bar at the specified index /// Thrown when index is out of range public TBar GetBar(int index) { if (index < 0 || index >= Count) { throw new ArgumentOutOfRangeException(nameof(index), index, $"Index must be between 0 and {Count - 1}"); } return Data[index]; } /// /// Gets the underlying data series (read-only access). /// public TBarSeries Data { get; } }