From 45dac305170aa33f57e9af7b09f5d627271b0c01 Mon Sep 17 00:00:00 2001
From: Andrew Sasmito <asasmito1920@gmail.com>
Date: Wed, 22 Jul 2026 17:10:48 -0400
Subject: [PATCH 1/7] Add CsvInputAdapterManager

Signed-off-by: Andrew Sasmito <asasmito1920@gmail.com>

Refactor CsvInputAdapterManager for improved clarity

Add new fields and methods to CsvInputAdapterManager

Fix error
---
 cpp/csp/adapters/csv/CMakeLists.txt           |   0
 .../adapters/csv/CsvInputAdapterManager.cpp   | 358 ++++++++++++++++++
 cpp/csp/adapters/csv/CsvInputAdapterManager.h | 103 +++++
 cpp/csp/python/adapters/csvadapterimpl.cpp    |   0
 4 files changed, 461 insertions(+)
 create mode 100644 cpp/csp/adapters/csv/CMakeLists.txt
 create mode 100644 cpp/csp/adapters/csv/CsvInputAdapterManager.cpp
 create mode 100644 cpp/csp/adapters/csv/CsvInputAdapterManager.h
 create mode 100644 cpp/csp/python/adapters/csvadapterimpl.cpp

diff --git a/cpp/csp/adapters/csv/CMakeLists.txt b/cpp/csp/adapters/csv/CMakeLists.txt
new file mode 100644
index 000000000..e69de29bb
diff --git a/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp b/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp
new file mode 100644
index 000000000..273b690b4
--- /dev/null
+++ b/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp
@@ -0,0 +1,358 @@
+#include "csp/core/Exception.h"
+#include <csp/adapters/csv/CsvInputAdapterManager.h>
+#include <csp/adapters/utils/ValueDispatcher.h>
+#include <csp/engine/AdapterManager.h>
+#include <csp/engine/CspType.h>
+#include <csp/engine/Dictionary.h>
+#include <csp/engine/Struct.h>
+#include <cstddef>
+#include <functional>
+#include <unordered_map>
+#include <set>
+#include <ranges>
+#include <iterator>
+#include <string>
+#include <optional>
+#include <variant>
+#include <vector>
+
+namespace csp::adapters::csv
+{
+
+// Parses exactly "YYYY-MM-DD HH::MM::SS"
+DateTime parseFixed_YmdHMS(std::string_view date) {
+    if (date.size() < 19)
+        CSP_THROW(ValueError, "Timestamp too short");
+
+    auto d2 = [&](size_t i) {
+        return (date[i] - '0') * 10 + (date[i + 1] - '0');
+    };
+    
+    int year            = d2(0) * 100 + d2(2);
+    int month           = d2(5);
+    int day             = d2(8);
+
+    int hour            = d2(11);
+    int minute          = d2(14);
+    int second          = d2(17);
+
+    return DateTime(year, month, day, hour, minute, second);
+}
+
+
+CsvInputAdapterManager::CsvInputAdapterManager( csp::Engine *engine, const Dictionary &properties, std::string filename ) :
+        AdapterManager( engine ),
+        m_filename(filename)
+{
+    CSP_TRUE_OR_THROW_RUNTIME( !m_filename.empty(), "Filename must be provided" );
+    auto tz = properties.get<std::string>( "tz", "UTC" );
+    CSP_TRUE_OR_THROW_RUNTIME( tz == "UTC",
+                               "Only UTC default timezone is supported, got:" << tz );
+
+    properties.tryGet( "start_time", m_startTime );
+    properties.tryGet( "end_time", m_endTime );
+
+    m_delimiter        = properties.get<std::string>( "delimiter", ",");
+    m_hasHeader        = properties.get<bool>       ( "hasHeader", true);
+    m_timeColumn       = properties.get<std::string>( "time_column", "" );
+    m_symbolColumnName = properties.get<std::string>( "symbol_column", "" );
+
+    CSP_TRUE_OR_THROW_RUNTIME( m_timeColumn != "", "Time column can't be empty" );
+
+    properties.tryGet( "time_format", m_timeFormat );
+
+    if (m_timeFormat.empty()) {
+        dateParser = parseFixed_YmdHMS;
+    } else if ( m_timeFormat == "YYYY-MM-DD HH::MM::SS") {
+        dateParser = parseFixed_YmdHMS;
+    } else {
+        CSP_THROW(ValueError, "Time format not supported");
+    }
+}
+
+CsvInputAdapterManager::~CsvInputAdapterManager() = default;
+
+void CsvInputAdapterManager::setupProcessor(
+    const std::vector<std::string> & schema,
+    const std::set<std::string> & neededColumns,
+    std::optional<std::string> symbolColumn,
+    bool subscribeAllOnEmptySymbol )
+{
+    m_schema = {};
+    m_symbolColumn = std::nullopt;
+
+    for( int i = 0; i < std::ssize( schema ); ++i ) {
+        if (neededColumns.contains( schema[i] )) {
+            m_schema.push_back(i);
+        }
+
+        if (symbolColumn && *symbolColumn == schema[i]) [[ unlikely ]] {
+            m_symbolColumn = i;
+        }
+    }
+}
+
+ManagedSimInputAdapter * CsvInputAdapterManager::getInputAdapter(
+    CspTypePtr & type, const Dictionary & properties, PushMode pushMode )
+{
+    // Per-subscription symbol filter. Empty string means "subscribe to every row".
+    std::string symbol = properties.get<std::string>( "symbol", "" );
+
+    auto * adapter = engine() -> createOwnedObject<ManagedSimInputAdapter>(
+        type, this, pushMode );
+
+    Subscriber sub;
+    sub.m_adapter = adapter;
+
+    // Stash the field_map so the subscriber can convert its row at dispatch time (Stage 4).
+    // Shape mirrors parquet: string -> single-column; DictionaryPtr -> struct field map;
+    // absent/None -> whole-row dict.
+    if( properties.exists( "field_map" ) )
+    {
+        auto & fm = properties.getUntypedValue( "field_map" );
+        if( std::holds_alternative<std::string>( fm ) )
+            sub.m_fieldMap = std::get<std::string>( fm );
+        else if( std::holds_alternative<DictionaryPtr>( fm ) )
+            sub.m_fieldMap = std::get<DictionaryPtr>( fm );
+        // else: leave monostate — whole-row dict output
+    }
+
+    if( symbol.empty() )
+        m_subscribers.push_back( std::move( sub ) );
+    else
+        m_subscribersBySymbol[ symbol ].push_back( std::move( sub ) );
+
+    return adapter;
+}
+
+void CsvInputAdapterManager::start( DateTime starttime, DateTime endtime )
+{
+    if( !m_startTime.isNone() )
+    {
+        starttime = std::max( starttime, m_startTime );
+    }
+    AdapterManager::start( starttime, endtime );
+
+    m_file = std::ifstream(m_filename, std::ios::binary);
+    if (!m_file) CSP_THROW(IOError, "Failed to open " << m_filename);
+
+    // Reusable split helper: split a line by m_delimiter into owned strings,
+    // trimming a trailing '\r' from the final field for Windows CSVs.
+    auto splitLine = []( std::string_view line, std::string_view delim ) {
+        std::vector<std::string> parts;
+        for( auto part : std::views::split( line, delim ) ) {
+            auto begin = part.begin();
+            auto len = std::ranges::distance( part );
+            parts.emplace_back( len == 0 ? std::string() : std::string( &*begin, len ) );
+        }
+        if( !parts.empty() && !parts.back().empty() && parts.back().back() == '\r' )
+            parts.back().pop_back();
+        return parts;
+    };
+
+    // --- Parse header ---
+    m_columnNames.clear();
+    if( m_hasHeader )
+    {
+        std::string headerLine;
+        if( !std::getline( m_file, headerLine ) )
+            CSP_THROW( IOError, "Failed to read header from " << m_filename );
+        m_columnNames = splitLine( headerLine, m_delimiter );
+    }
+
+    // --- Collect the set of columns any subscriber cares about ---
+    std::set<std::string> neededColumns;
+    bool needAllColumns = false;
+
+    if( !m_timeColumn.empty() )        neededColumns.insert( m_timeColumn );
+    if( !m_symbolColumnName.empty() )  neededColumns.insert( m_symbolColumnName );
+
+    auto collectFrom = [&]( const Subscriber & sub ) {
+        if( std::holds_alternative<std::monostate>( sub.m_fieldMap ) )
+        {
+            needAllColumns = true;
+        }
+        else if( std::holds_alternative<std::string>( sub.m_fieldMap ) )
+        {
+            neededColumns.insert( std::get<std::string>( sub.m_fieldMap ) );
+        }
+        else if( std::holds_alternative<DictionaryPtr>( sub.m_fieldMap ) )
+        {
+            auto & fm = std::get<DictionaryPtr>( sub.m_fieldMap );
+            for( auto it = fm -> begin(); it != fm -> end(); ++it )
+                neededColumns.insert( it.key() );
+        }
+    };
+
+    for( const auto & sub : m_subscribers ) collectFrom( sub );
+    for( const auto & [ symbol, subs ] : m_subscribersBySymbol )
+        for( const auto & sub : subs ) collectFrom( sub );
+
+    if( needAllColumns )
+        for( const auto & name : m_columnNames )
+            neededColumns.insert( name );
+
+    // --- Do column-name -> index mappings through setupProcessor ---
+    std::optional<std::string> symbolColumnOpt;
+    if( !m_symbolColumnName.empty() )
+        symbolColumnOpt = m_symbolColumnName;
+
+    bool subscribeAllOnEmptySymbol = !m_subscribers.empty();
+    setupProcessor( m_columnNames, neededColumns, symbolColumnOpt, subscribeAllOnEmptySymbol );
+
+    // Locate the time column so processNextSimTimeSlice can extract it O(1).
+    m_timeColumnIndex = -1;
+    for( int i = 0; i < std::ssize( m_columnNames ); ++i )
+    {
+        if( m_columnNames[ i ] == m_timeColumn )
+        {
+            m_timeColumnIndex = i;
+            break;
+        }
+    }
+    
+    CSP_TRUE_OR_THROW_RUNTIME( m_timeColumnIndex >= 0,
+        "Time column '" << m_timeColumn << "' not found in CSV header" );
+
+    if( !m_symbolColumnName.empty() )
+    {
+        CSP_TRUE_OR_THROW_RUNTIME( m_symbolColumn.has_value(),
+            "Symbol column '" << m_symbolColumnName << "' not found in CSV header" );
+    }
+
+    // Bind each subscriber's row -> tick callback now that the schema is known.
+    bindSubscriberDispatchers();
+
+    // Cache the first data row so processNextSimTimeSlice's skip loop has data to compare.
+    if( !std::getline( m_file, m_row ) )
+        m_row.clear();
+}
+
+void CsvInputAdapterManager::bindSubscriberDispatchers()
+{
+    // Column name -> header index (built once).
+    std::unordered_map<std::string_view, size_t> colIndex;
+    colIndex.reserve( m_columnNames.size() );
+    for( size_t i = 0; i < m_columnNames.size(); ++i )
+        colIndex[ m_columnNames[ i ] ] = i;
+
+    auto bind = [&]( Subscriber & sub )
+    {
+        // Whole-row dict: needs Python to build a PyDict
+        if( std::holds_alternative<std::monostate>( sub.m_fieldMap ) )
+            return;
+
+        // Single-column subscription — extract the named column and push it.
+        if( std::holds_alternative<std::string>( sub.m_fieldMap ) )
+        {
+            const auto & colName = std::get<std::string>( sub.m_fieldMap );
+            auto it = colIndex.find( colName );
+            CSP_TRUE_OR_THROW_RUNTIME( it != colIndex.end(),
+                "Column '" << colName << "' not found in CSV header" );
+            size_t idx = it -> second;
+
+            auto * adapter = sub.m_adapter;
+            auto tag = adapter -> dataType() -> type();
+
+            if( tag == CspType::Type::STRING )
+            {
+                // Only pure-string ticks are implementable without Python.
+                sub.m_dispatch = [ adapter, idx ]( const std::vector<std::string_view> & cols )
+                {
+                    adapter -> pushTick<std::string>( std::string( cols[ idx ] ) );
+                };
+            }
+            // else: leave m_dispatch null; Stage 5 will bind richer conversions.
+            return;
+        }
+
+        // Struct field_map: needs csp::Struct construction with per-field type
+    };
+
+    for( auto & sub : m_subscribers ) bind( sub );
+    for( auto & [ symbol, subs ] : m_subscribersBySymbol )
+        for( auto & sub : subs ) bind( sub );
+}
+
+void CsvInputAdapterManager::stop()
+{
+    m_subscribers.clear();
+    m_subscribersBySymbol.clear();
+    m_schema.clear();
+    m_file.close();
+    AdapterManager::stop();
+}
+
+DateTime CsvInputAdapterManager::processNextSimTimeSlice( DateTime time )
+{
+    if( m_row.empty() ) [[unlikely]]
+        return DateTime::NONE();
+
+    // Split m_row once per row into string_views. Views are valid until the
+    // next getline() mutates m_row, so every read of `cols` must precede the
+    // next read from the file.
+    auto splitRow = [ this ]()
+    {
+        std::vector<std::string_view> cols;
+        for( auto part : std::views::split( m_row, m_delimiter ) )
+        {
+            auto begin = part.begin();
+            auto len = std::ranges::distance( part );
+            cols.emplace_back( len == 0 ? std::string_view()
+                                        : std::string_view( &*begin, len ) );
+        }
+        // Trim trailing '\r' on Windows CSVs so the final field parses cleanly.
+        if( !cols.empty() && !cols.back().empty() && cols.back().back() == '\r' )
+            cols.back().remove_suffix( 1 );
+        return cols;
+    };
+
+    // Skip loop: advance until we find a row at or after `time`.
+    std::vector<std::string_view> cols = splitRow();
+    DateTime rowTime = dateParser( cols[ m_timeColumnIndex ] );
+    while( rowTime < time )
+    {
+        if( !std::getline( m_file, m_row ) )
+        {
+            m_row.clear();
+            return DateTime::NONE();
+        }
+        cols = splitRow();
+        rowTime = dateParser( cols[ m_timeColumnIndex ] );
+    }
+
+    if( !m_endTime.isNone() && rowTime > m_endTime )
+        return DateTime::NONE();
+
+    if( rowTime > time )
+        return rowTime;
+
+    // Dispatch every row with this exact timestamp.
+    do
+    {
+        // Subscribe-all subscribers see every row.
+        for( auto & sub : m_subscribers )
+            if( sub.m_dispatch ) sub.m_dispatch( cols );
+
+        // Symbol-filtered subscribers only see rows where their symbol matches.
+        if( m_symbolColumn.has_value() )
+        {
+            std::string sym( cols[ *m_symbolColumn ] );
+            auto it = m_subscribersBySymbol.find( sym );
+            if( it != m_subscribersBySymbol.end() )
+                for( auto & sub : it -> second )
+                    if( sub.m_dispatch ) sub.m_dispatch( cols );
+        }
+
+        if( !std::getline( m_file, m_row ) )
+        {
+            m_row.clear();
+            return DateTime::NONE();
+        }
+        cols = splitRow();
+        rowTime = dateParser( cols[ m_timeColumnIndex ] );
+    } while( rowTime == time );
+
+    return rowTime;
+}
+}
diff --git a/cpp/csp/adapters/csv/CsvInputAdapterManager.h b/cpp/csp/adapters/csv/CsvInputAdapterManager.h
new file mode 100644
index 000000000..f403c487b
--- /dev/null
+++ b/cpp/csp/adapters/csv/CsvInputAdapterManager.h
@@ -0,0 +1,103 @@
+#ifndef _IN_CSP_ADAPTERS_CSV_CsvInputAdapterManager_H
+#define _IN_CSP_ADAPTERS_CSV_CsvInputAdapterManager_H
+
+#include "csp/core/Time.h"
+#include <csp/engine/AdapterManager.h>
+#include <csp/engine/Dictionary.h>
+#include <csp/engine/Struct.h>
+#include <fstream>
+#include <functional>
+#include <unordered_map>
+#include <set>
+#include <string>
+#include <optional>
+#include <variant>
+#include <vector>
+
+
+namespace csp::adapters::csv
+{
+
+
+// Manages all csv input adapters for a single engine run.
+//
+// Lifecycle:
+//   1. Registration:  getInputAdapter() called per subscription (before engine starts)
+//   2. start():       create processors → wire adapters → read first row
+//   3. processNextSimTimeSlice():  skip/dispatch loop per engine tick
+//   4. stop():        tear down all state
+class CsvInputAdapterManager final : public csp::AdapterManager
+{
+public:
+
+    CsvInputAdapterManager( csp::Engine *engine, const Dictionary &properties, std::string filename );
+
+    ~CsvInputAdapterManager();
+
+    const char *name() const override{ return "CsvInputAdapterManager"; }
+
+    void start( DateTime starttime, DateTime endtime ) override;
+    void stop() override;
+    DateTime processNextSimTimeSlice( DateTime time ) override;
+
+    ManagedSimInputAdapter * getInputAdapter( CspTypePtr & type, const Dictionary & properties, PushMode pushMode );
+
+private:
+
+    void setupProcessor( const std::vector<std::string> & schema,
+                         const std::set<std::string> & neededColumns,
+                         std::optional<std::string> symbolColumn,
+                         bool subscribeAllOnEmptySymbol );
+
+    bool readNextRow();
+
+    // Option A: each subscriber owns its own field_map / target type,
+    // so per-subscriber row conversion (Stage 4) can dispatch without
+    // needing the manager to know Python.
+    //   - monostate  -> return whole row as dict
+    //   - string     -> single-column adapter (value from that column)
+    //   - DictionaryPtr -> struct field map (csv column name -> struct field)
+    //
+    // m_dispatch is bound after the schema is known (bindSubscriberDispatchers).
+    // The manager just calls it per row — the callback owns the conversion.
+    // A null m_dispatch means "conversion not supported yet in this stage" and
+    // the manager silently skips it.
+    struct Subscriber {
+        ManagedSimInputAdapter *                                          m_adapter;
+        std::variant<std::monostate, std::string, DictionaryPtr>          m_fieldMap;
+        std::function<void(const std::vector<std::string_view> &)>        m_dispatch;
+    };
+
+    // Walk registered subscribers and bind each one's m_dispatch based on its
+    // field_map + the parsed header. Must be called after m_columnNames is populated.
+    void bindSubscriberDispatchers();
+
+    using dateTimeParserfn = DateTime(*)(std::string_view);
+
+    // Registration-phase state (populated by getInputAdapter before start)
+    std::vector<Subscriber> m_subscribers;
+    std::unordered_map<std::string, std::vector<Subscriber>> m_subscribersBySymbol;
+
+    // Configuration (from properties dict)
+    csp::DateTime                       m_startTime;
+    csp::DateTime                       m_endTime;
+    std::string                         m_timeColumn;
+    std::string                         m_timeFormat;
+    std::string                         m_filename;
+    std::string                         m_delimiter;
+    std::string                         m_symbolColumnName; // configured symbol column name ("" = no symbol column)
+    bool                                m_hasHeader;
+    dateTimeParserfn                    dateParser;
+
+    // Runtime state (initialized in start, used in processNextSimTimeSlice)
+    std::vector<std::string>                    m_columnNames; // full header, in order
+    std::optional<int>                          m_symbolColumn; // Index of symbol column
+    std::vector<int>                            m_schema; // Indices to be used
+    int                                         m_timeColumnIndex;
+    std::ifstream                               m_file;
+    std::string                                 m_row; // Current cached row
+};
+
+}
+
+#endif // _IN_CSP_ADAPTERS_CSV_CsvInputAdapterManager_H
diff --git a/cpp/csp/python/adapters/csvadapterimpl.cpp b/cpp/csp/python/adapters/csvadapterimpl.cpp
new file mode 100644
index 000000000..e69de29bb

From d448879be1c73290ffec2c7e1cda676c5fa0c661 Mon Sep 17 00:00:00 2001
From: Andrew Sasmito <asasmito1920@gmail.com>
Date: Wed, 22 Jul 2026 20:42:30 -0400
Subject: [PATCH 2/7] Create linker and update cmakes

Signed-off-by: Andrew Sasmito <asasmito1920@gmail.com>
---
 CMakeFiles/CMakeSystem.cmake                  | 15 +++++
 CMakeLists.txt                                |  5 ++
 cpp/cmake/modules/FindDepsCsvAdapter.cmake    |  3 +
 cpp/csp/adapters/CMakeLists.txt               |  4 ++
 cpp/csp/adapters/csv/CMakeLists.txt           | 25 +++++++
 .../adapters/csv/CsvInputAdapterManager.cpp   |  7 +-
 cpp/csp/adapters/csv/CsvInputAdapterManager.h |  2 +-
 cpp/csp/python/adapters/CMakeLists.txt        |  6 ++
 cpp/csp/python/adapters/csvadapterimpl.cpp    | 66 +++++++++++++++++++
 setup.py                                      |  1 +
 10 files changed, 130 insertions(+), 4 deletions(-)
 create mode 100644 CMakeFiles/CMakeSystem.cmake
 create mode 100644 cpp/cmake/modules/FindDepsCsvAdapter.cmake

diff --git a/CMakeFiles/CMakeSystem.cmake b/CMakeFiles/CMakeSystem.cmake
new file mode 100644
index 000000000..0473afc46
--- /dev/null
+++ b/CMakeFiles/CMakeSystem.cmake
@@ -0,0 +1,15 @@
+set(CMAKE_HOST_SYSTEM "Darwin-24.6.0")
+set(CMAKE_HOST_SYSTEM_NAME "Darwin")
+set(CMAKE_HOST_SYSTEM_VERSION "24.6.0")
+set(CMAKE_HOST_SYSTEM_PROCESSOR "arm64")
+
+
+
+set(CMAKE_SYSTEM "Darwin-24.6.0")
+set(CMAKE_SYSTEM_NAME "Darwin")
+set(CMAKE_SYSTEM_VERSION "24.6.0")
+set(CMAKE_SYSTEM_PROCESSOR "arm64")
+
+set(CMAKE_CROSSCOMPILING "FALSE")
+
+set(CMAKE_SYSTEM_LOADED 1)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index dbbff2f36..6dadad570 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -74,6 +74,7 @@ option(CSP_ENABLE_UBSAN "Build with undefined behavior sanitizer" OFF)
 option(CSP_BUILD_KAFKA_ADAPTER "Build kafka adapter" ON)
 option(CSP_BUILD_ARROW_ADAPTER "Build arrow adapter" ON)
 option(CSP_BUILD_PARQUET_ADAPTER "Build parquet adapter" ON)
+option(CSP_BUILD_CSV_ADAPTER "Build CSV adapter" ON)
 
 # Parquet adapter depends on arrow adapter
 if(CSP_BUILD_PARQUET_ADAPTER AND NOT CSP_BUILD_ARROW_ADAPTER)
@@ -271,6 +272,10 @@ if(CSP_BUILD_PARQUET_ADAPTER)
     find_package(DepsParquetAdapter REQUIRED)
 endif()
 
+if(CSP_BUILD_CSV_ADAPTER)
+    find_package(DepsCsvAdapter REQUIRED)
+endif()
+
 
 # PYTHON
 if(CSP_MANYLINUX)
diff --git a/cpp/cmake/modules/FindDepsCsvAdapter.cmake b/cpp/cmake/modules/FindDepsCsvAdapter.cmake
new file mode 100644
index 000000000..512d3d4f0
--- /dev/null
+++ b/cpp/cmake/modules/FindDepsCsvAdapter.cmake
@@ -0,0 +1,3 @@
+cmake_minimum_required(VERSION 3.7.2)
+
+set(DepsCsvAdapter_FOUND TRUE)
diff --git a/cpp/csp/adapters/CMakeLists.txt b/cpp/csp/adapters/CMakeLists.txt
index 28584bdb1..26e9e7c53 100644
--- a/cpp/csp/adapters/CMakeLists.txt
+++ b/cpp/csp/adapters/CMakeLists.txt
@@ -15,4 +15,8 @@ if(CSP_BUILD_WS_CLIENT_ADAPTER)
     add_subdirectory(websocket)
 endif()
 
+if(CSP_BUILD_CSV_ADAPTER)
+    add_subdirectory(csv)
+endif()
+
 add_subdirectory(utils)
diff --git a/cpp/csp/adapters/csv/CMakeLists.txt b/cpp/csp/adapters/csv/CMakeLists.txt
index e69de29bb..fa5f4fd4d 100644
--- a/cpp/csp/adapters/csv/CMakeLists.txt
+++ b/cpp/csp/adapters/csv/CMakeLists.txt
@@ -0,0 +1,25 @@
+set(CSV_HEADER_FILES
+    CsvInputAdapterManager.h
+)
+
+set(CSV_SOURCE_FILES
+    CsvInputAdapterManager.cpp
+    ${CSV_HEADER_FILES}
+)
+
+add_library(csp_csv_adapter STATIC ${CSV_SOURCE_FILES})
+
+set_target_properties(csp_csv_adapter PROPERTIES
+    PUBLIC_HEADER "${CSV_HEADER_FILES}"
+)
+
+target_link_libraries(csp_csv_adapter
+    PRIVATE
+        csp_engine
+)
+
+install(TARGETS csp_csv_adapter
+    PUBLIC_HEADER DESTINATION include/csp/adapters/csv
+    RUNTIME DESTINATION ${CSP_RUNTIME_INSTALL_SUBDIR}
+    LIBRARY DESTINATION lib/
+)
\ No newline at end of file
diff --git a/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp b/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp
index 273b690b4..a6c123d48 100644
--- a/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp
+++ b/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp
@@ -40,11 +40,12 @@ DateTime parseFixed_YmdHMS(std::string_view date) {
 }
 
 
-CsvInputAdapterManager::CsvInputAdapterManager( csp::Engine *engine, const Dictionary &properties, std::string filename ) :
-        AdapterManager( engine ),
-        m_filename(filename)
+CsvInputAdapterManager::CsvInputAdapterManager( csp::Engine *engine, const Dictionary &properties ) :
+        AdapterManager( engine )
 {
+    m_filename         = properties.get<std::string>( "filename", "" ); 
     CSP_TRUE_OR_THROW_RUNTIME( !m_filename.empty(), "Filename must be provided" );
+
     auto tz = properties.get<std::string>( "tz", "UTC" );
     CSP_TRUE_OR_THROW_RUNTIME( tz == "UTC",
                                "Only UTC default timezone is supported, got:" << tz );
diff --git a/cpp/csp/adapters/csv/CsvInputAdapterManager.h b/cpp/csp/adapters/csv/CsvInputAdapterManager.h
index f403c487b..e4b5c97dd 100644
--- a/cpp/csp/adapters/csv/CsvInputAdapterManager.h
+++ b/cpp/csp/adapters/csv/CsvInputAdapterManager.h
@@ -30,7 +30,7 @@ class CsvInputAdapterManager final : public csp::AdapterManager
 {
 public:
 
-    CsvInputAdapterManager( csp::Engine *engine, const Dictionary &properties, std::string filename );
+    CsvInputAdapterManager( csp::Engine *engine, const Dictionary &properties );
 
     ~CsvInputAdapterManager();
 
diff --git a/cpp/csp/python/adapters/CMakeLists.txt b/cpp/csp/python/adapters/CMakeLists.txt
index cf1e473fe..828647d96 100644
--- a/cpp/csp/python/adapters/CMakeLists.txt
+++ b/cpp/csp/python/adapters/CMakeLists.txt
@@ -24,3 +24,9 @@ if(CSP_BUILD_WS_CLIENT_ADAPTER)
     target_link_libraries(websocketadapterimpl csp_core csp_engine cspimpl csp_websocket_client_adapter)
     install(TARGETS websocketadapterimpl RUNTIME DESTINATION ${CSP_RUNTIME_INSTALL_SUBDIR})
 endif()
+
+if(CSP_BUILD_CSV_ADAPTER)
+    add_library(csvadapterimpl SHARED csvadapterimpl.cpp)
+    target_link_libraries(csvadapterimpl csp_core csp_engine cspimpl csp_csv_adapter)
+    install(TARGETS csvadapterimpl RUNTIME DESTINATION ${CSP_RUNTIME_INSTALL_SUBDIR})
+endif()
diff --git a/cpp/csp/python/adapters/csvadapterimpl.cpp b/cpp/csp/python/adapters/csvadapterimpl.cpp
index e69de29bb..4005989f5 100644
--- a/cpp/csp/python/adapters/csvadapterimpl.cpp
+++ b/cpp/csp/python/adapters/csvadapterimpl.cpp
@@ -0,0 +1,66 @@
+#include <csp/adapters/csv/CsvInputAdapterManager.h>
+#include <csp/engine/PushInputAdapter.h>
+#include <csp/python/Conversions.h>
+#include <csp/python/Exception.h>
+#include <csp/python/InitHelper.h>
+#include <csp/python/PyAdapterManagerWrapper.h>
+#include <csp/python/PyEngine.h>
+#include <csp/python/PyInputAdapterWrapper.h>
+#include <csp/python/PyOutputAdapterWrapper.h>
+
+using namespace csp::adapters::csv;
+
+namespace csp::python
+{
+
+//AdapterManager
+csp::AdapterManager * create_csv_adapter_manager( PyEngine * engine, const Dictionary & properties )
+{
+    return engine -> engine() -> createOwnedObject<CsvInputAdapterManager>( properties );
+}
+
+static InputAdapter * create_csv_input_adapter( csp::AdapterManager * manager, PyEngine * pyengine, PyObject * pyType, PushMode pushMode, PyObject * args )
+{
+    auto & cspType = pyTypeAsCspType( pyType );
+
+    PyObject * pyProperties;
+    PyObject * type;
+
+    auto * csvManager = dynamic_cast<CsvInputAdapterManager*>( manager );
+    if( !csvManager )
+        CSP_THROW( TypeError, "Expected CsvInputAdapterManager" );
+
+    if( !PyArg_ParseTuple( args, "O!O!",
+                           &PyType_Type, &type,
+                           &PyDict_Type, &pyProperties ) )
+        CSP_THROW( PythonPassthrough, "" );
+
+    return csvManager -> getInputAdapter( cspType, fromPython<Dictionary>( pyProperties ), pushMode );
+}
+
+REGISTER_ADAPTER_MANAGER( _csv_adapter_manager, create_csv_adapter_manager );
+REGISTER_INPUT_ADAPTER(   _csv_input_adapter,   create_csv_input_adapter );
+
+static PyModuleDef _csvadapterimpl_module = {
+        PyModuleDef_HEAD_INIT,
+        "_csvadapterimpl",
+        "_csvadapterimpl c++ module",
+        -1,
+        NULL, NULL, NULL, NULL, NULL
+};
+
+PyMODINIT_FUNC PyInit__csvadapterimpl(void)
+{
+    PyObject* m;
+
+    m = PyModule_Create( &_csvadapterimpl_module);
+    if( m == NULL )
+        return NULL;
+
+    if( !InitHelper::instance().execute( m ) )
+        return NULL;
+
+    return m;
+}
+
+}
diff --git a/setup.py b/setup.py
index c50b60f10..7ec6fd1ab 100644
--- a/setup.py
+++ b/setup.py
@@ -22,6 +22,7 @@
     ("CSP_BUILD_ARROW_ADAPTER", "1"),
     ("CSP_BUILD_KAFKA_ADAPTER", "1"),
     ("CSP_BUILD_PARQUET_ADAPTER", "1"),
+    ("CSP_BUILD_CSV_ADAPTER", "1"),
     ("CSP_BUILD_WS_CLIENT_ADAPTER", "1"),
     ("CSP_ENABLE_ASAN", "0"),
     ("CSP_ENABLE_UBSAN", "0"),

From e06bcb71ea0aecfdba8ad2fa1cd28d8de4a0dab7 Mon Sep 17 00:00:00 2001
From: Andrew Sasmito <asasmito1920@gmail.com>
Date: Sun, 26 Jul 2026 14:49:54 -0400
Subject: [PATCH 3/7] Update python side and add tests

Signed-off-by: Andrew Sasmito <asasmito1920@gmail.com>
---
 compile_commands.json                         |   1 +
 .../adapters/csv/CsvInputAdapterManager.cpp   |  60 ++++-
 cpp/csp/adapters/csv/CsvInputAdapterManager.h |  33 ++-
 csp/adapters/csv.py                           | 168 +++++---------
 csp/tests/adapters/csv_test_data.csv          |  14 +-
 csp/tests/adapters/test_csv.py                | 211 +++++++++++++-----
 .../Write-Historical-Input-Adapters.md        | 211 +-----------------
 7 files changed, 294 insertions(+), 404 deletions(-)
 create mode 120000 compile_commands.json

diff --git a/compile_commands.json b/compile_commands.json
new file mode 120000
index 000000000..25eb4b2b4
--- /dev/null
+++ b/compile_commands.json
@@ -0,0 +1 @@
+build/compile_commands.json
\ No newline at end of file
diff --git a/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp b/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp
index a6c123d48..c25e55eb9 100644
--- a/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp
+++ b/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp
@@ -105,8 +105,8 @@ ManagedSimInputAdapter * CsvInputAdapterManager::getInputAdapter(
     Subscriber sub;
     sub.m_adapter = adapter;
 
-    // Stash the field_map so the subscriber can convert its row at dispatch time (Stage 4).
-    // Shape mirrors parquet: string -> single-column; DictionaryPtr -> struct field map;
+    // Stash the field_map so the subscriber can convert its row at dispatch time.
+    // string -> single-column; DictionaryPtr -> struct field map;
     // absent/None -> whole-row dict.
     if( properties.exists( "field_map" ) )
     {
@@ -239,9 +239,41 @@ void CsvInputAdapterManager::bindSubscriberDispatchers()
 
     auto bind = [&]( Subscriber & sub )
     {
-        // Whole-row dict: needs Python to build a PyDict
-        if( std::holds_alternative<std::monostate>( sub.m_fieldMap ) )
+        // Whole-row dict: Build a struct
+        if(std::holds_alternative<std::monostate>(sub.m_fieldMap))
+        {
+            auto *structType =
+                static_cast<const CspStructType *>(sub.m_adapter->dataType());
+
+            auto meta = structType->meta();
+
+            StructSubscription subscription;
+            subscription.m_adapter = sub.m_adapter;
+            subscription.m_structMeta = meta;
+
+
+            for(size_t i = 0; i < m_columnNames.size(); i++)
+            {
+                auto field = meta->field(m_columnNames[i]);
+
+                if(!field)
+                    continue;
+
+                subscription.m_fieldSetters.push_back(
+                    [i, field](StructPtr &s,
+                            const std::vector<std::string_view> &cols)
+                    {
+                        field->setValue<std::string>(
+                            s.get(),
+                            std::string(cols[i])
+                        );
+                    }
+                );
+            }
+
+            sub.m_structSubscription = std::move(subscription);
             return;
+        }
 
         // Single-column subscription — extract the named column and push it.
         if( std::holds_alternative<std::string>( sub.m_fieldMap ) )
@@ -263,7 +295,7 @@ void CsvInputAdapterManager::bindSubscriberDispatchers()
                     adapter -> pushTick<std::string>( std::string( cols[ idx ] ) );
                 };
             }
-            // else: leave m_dispatch null; Stage 5 will bind richer conversions.
+            // else: leave m_dispatch null;
             return;
         }
 
@@ -332,8 +364,13 @@ DateTime CsvInputAdapterManager::processNextSimTimeSlice( DateTime time )
     do
     {
         // Subscribe-all subscribers see every row.
-        for( auto & sub : m_subscribers )
-            if( sub.m_dispatch ) sub.m_dispatch( cols );
+        for( auto & sub : m_subscribers ) {
+            if(sub.m_structSubscription)
+                sub.m_structSubscription->dispatchValue(cols);
+
+            else if(sub.m_dispatch)
+                sub.m_dispatch(cols);
+        }
 
         // Symbol-filtered subscribers only see rows where their symbol matches.
         if( m_symbolColumn.has_value() )
@@ -341,8 +378,13 @@ DateTime CsvInputAdapterManager::processNextSimTimeSlice( DateTime time )
             std::string sym( cols[ *m_symbolColumn ] );
             auto it = m_subscribersBySymbol.find( sym );
             if( it != m_subscribersBySymbol.end() )
-                for( auto & sub : it -> second )
-                    if( sub.m_dispatch ) sub.m_dispatch( cols );
+                for( auto & sub : it -> second ) {
+                    if(sub.m_structSubscription)
+                        sub.m_structSubscription->dispatchValue(cols);
+                    
+                    else if(sub.m_dispatch)
+                        sub.m_dispatch(cols);
+                }
         }
 
         if( !std::getline( m_file, m_row ) )
diff --git a/cpp/csp/adapters/csv/CsvInputAdapterManager.h b/cpp/csp/adapters/csv/CsvInputAdapterManager.h
index e4b5c97dd..0e0669d73 100644
--- a/cpp/csp/adapters/csv/CsvInputAdapterManager.h
+++ b/cpp/csp/adapters/csv/CsvInputAdapterManager.h
@@ -51,21 +51,32 @@ class CsvInputAdapterManager final : public csp::AdapterManager
 
     bool readNextRow();
 
-    // Option A: each subscriber owns its own field_map / target type,
-    // so per-subscriber row conversion (Stage 4) can dispatch without
-    // needing the manager to know Python.
-    //   - monostate  -> return whole row as dict
-    //   - string     -> single-column adapter (value from that column)
-    //   - DictionaryPtr -> struct field map (csv column name -> struct field)
-    //
-    // m_dispatch is bound after the schema is known (bindSubscriberDispatchers).
-    // The manager just calls it per row — the callback owns the conversion.
-    // A null m_dispatch means "conversion not supported yet in this stage" and
-    // the manager silently skips it.
+    struct StructSubscription
+    {
+        using FieldSetter = std::function<void(StructPtr &, const std::vector<std::string_view> &)>;
+
+        ManagedSimInputAdapter *m_adapter;
+        std::shared_ptr<const StructMeta> m_structMeta;
+
+        std::vector<FieldSetter> m_fieldSetters;
+
+        void dispatchValue(const std::vector<std::string_view> &cols)
+        {
+            StructPtr value = m_structMeta->create();
+
+            for(auto &setter : m_fieldSetters)
+                setter(value, cols);
+
+            m_adapter->pushTick(value);
+        }
+    };
+
     struct Subscriber {
         ManagedSimInputAdapter *                                          m_adapter;
         std::variant<std::monostate, std::string, DictionaryPtr>          m_fieldMap;
+
         std::function<void(const std::vector<std::string_view> &)>        m_dispatch;
+        std::optional<StructSubscription>                                 m_structSubscription;
     };
 
     // Walk registered subscribers and bind each one's m_dispatch based on its
diff --git a/csp/adapters/csv.py b/csp/adapters/csv.py
index 82dc3d347..b5fa67daa 100644
--- a/csp/adapters/csv.py
+++ b/csp/adapters/csv.py
@@ -1,113 +1,61 @@
-import csv as pycsv
-from datetime import datetime
-
-from csp import PushMode, ts
-from csp.impl.adaptermanager import AdapterManagerImpl, ManagedSimInputAdapter
-from csp.impl.wiring import py_managed_adapter_def
-
-
-def time_converter(column, format_string, tz=None):
-    def convert(row):
-        v = row[column]
-        dt = datetime.strptime(v, format_string)
-        if tz is not None:
-            dt = tz.localize(dt)
-        return dt
-
-    return convert
-
-
-def YYYYMMDD_TIME_formatter(column, include_fraction=False, tz=None):
-    format_string = "%Y%m%d %X"
-    if include_fraction:
-        format_string += ".%f"
-
-    return time_converter(column, format_string, tz)
-
-
-# GRAPH TIME
-class CSVReader:
-    ## TODO we might want to support initial snapshot value
-    def __init__(self, filename, time_converter, delimiter=",", symbol_column=None):
-        self._filename = filename
-        self._symbol_column = symbol_column
-        self._delimiter = delimiter
-        self._time_converter = time_converter
-
-    def subscribe(self, symbol, typ, field_map=None, push_mode=PushMode.NON_COLLAPSING):
-        return self._subscribe(symbol, typ, field_map, push_mode)
-
-    def subscribe_all(self, typ, field_map=None, push_mode=PushMode.NON_COLLAPSING):
-        return self._subscribe("", typ, field_map, push_mode)
-
-    def _subscribe(self, symbol, typ, field_map, push_mode):
-        return CSVReadAdapter(self, symbol, typ, field_map, push_mode=push_mode)
+from csp import ts, PushMode
+from csp.impl.wiring import input_adapter_def
+from csp.lib import _csvadapterimpl
+
+
+class CsvAdapterManager:
+    def __init__(
+        self,
+        filename,
+        time_column,
+        symbol_column="",
+        delimiter=",",
+        has_header=True,
+        time_format=None,
+    ):
+        self._properties = {
+            "filename": filename,
+            "time_column": time_column,
+            "symbol_column": symbol_column,
+            "delimiter": delimiter,
+            "hasHeader": has_header,
+        }
+
+        if time_format is not None:
+            self._properties["time_format"] = time_format
+
+    def subscribe(
+        self,
+        ts_type,
+        field_map=None,
+        symbol=None,
+        push_mode=PushMode.LAST_VALUE,
+    ):
+        properties = self._properties.copy()
+
+        if field_map is not None:
+            properties["field_map"] = field_map
+        properties["symbol"] = symbol or ""
+
+        return _csv_input_adapter_def(
+            self,
+            ts_type,
+            properties,
+            push_mode=push_mode,
+        )
 
     def _create(self, engine, memo):
-        return CSVReaderImpl(engine, self)
-
-
-# RUN TIME
-class CSVReaderImpl(AdapterManagerImpl):
-    def __init__(self, engine, adapterRep):
-        super().__init__(engine)
-
-        self._rep = adapterRep
-        self._inputs = {}
-        self._csv_reader = None
-        self._next_row = None
-
-    def start(self, starttime, endtime):
-        self._csv_reader = pycsv.DictReader(open(self._rep._filename, "r"), delimiter=self._rep._delimiter)
-        self._next_row = None
-
-        for row in self._csv_reader:
-            time = self._rep._time_converter(row)
-            self._next_row = row
-            if time >= starttime:
-                break
-
-    def stop(self):
-        self._csv_reader = None
-
-    def register_input_adapter(self, symbol, adapter):
-        if symbol not in self._inputs:
-            self._inputs[symbol] = []
-        self._inputs[symbol].append(adapter)
-
-    def process_next_sim_timeslice(self, now):
-        if not self._next_row:
-            return None
-
-        while True:
-            time = self._rep._time_converter(self._next_row)
-            if time > now:
-                return time
-            self.process_row(self._next_row)
-            try:
-                self._next_row = next(self._csv_reader)
-            except StopIteration:
-                return None
-
-    def process_row(self, row):
-        if self._rep._symbol_column is not None:
-            symbol = row[self._rep._symbol_column]
-
-            if symbol in self._inputs:
-                for input in self._inputs.get(symbol, []):
-                    input.process_dict(row)
-
-        # subscribeAll
-        for input in self._inputs.get("", []):
-            input.process_dict(row)
-
-
-class CSVReadAdapterImpl(ManagedSimInputAdapter):
-    def __init__(self, managerImpl, symbol, typ, field_map):
-        managerImpl.register_input_adapter(symbol, self)
-        super().__init__(typ, field_map)
-
-
-CSVReadAdapter = py_managed_adapter_def(
-    "csvadapter", CSVReadAdapterImpl, ts["T"], CSVReader, symbol=str, typ="T", fieldMap=(object, None)
+        return _csvadapterimpl._csv_adapter_manager(
+            engine,
+            self._properties
+        )
+
+
+_csv_input_adapter_def = input_adapter_def(
+    "csv_input_adapter",
+    _csvadapterimpl._csv_input_adapter,
+    ts["T"],
+    CsvAdapterManager,
+    typ="T",
+    properties=dict,
 )
diff --git a/csp/tests/adapters/csv_test_data.csv b/csp/tests/adapters/csv_test_data.csv
index 20a256b37..9dee96014 100644
--- a/csp/tests/adapters/csv_test_data.csv
+++ b/csp/tests/adapters/csv_test_data.csv
@@ -1,8 +1,8 @@
 TIME|SYMBOL|PRICE|SIZE|SIDE
-20200303 09:30:00|AAPL|500.00|100|BUY
-20200303 09:30:01|IBM|100.00|200|BUY
-20200303 09:30:02|AAPL|400.00|100|BUY
-20200303 09:30:03|IBM|200.00|300|SELL
-20200303 09:30:04|AAPL|300.00|200|SELL
-20200303 09:30:05|AAPL|200.00|400|BUY
-20200303 09:30:06|GM|2.00|1|BUY
+2020-03-03 09:30:00|AAPL|500.00|100|BUY
+2020-03-03 09:30:01|IBM|100.00|200|BUY
+2020-03-03 09:30:02|AAPL|400.00|100|BUY
+2020-03-03 09:30:03|IBM|200.00|300|SELL
+2020-03-03 09:30:04|AAPL|300.00|200|SELL
+2020-03-03 09:30:05|AAPL|200.00|400|BUY
+2020-03-03 09:30:06|GM|2.00|1|BUY
diff --git a/csp/tests/adapters/test_csv.py b/csp/tests/adapters/test_csv.py
index e8aa46f3f..5fe7d08b6 100644
--- a/csp/tests/adapters/test_csv.py
+++ b/csp/tests/adapters/test_csv.py
@@ -1,105 +1,202 @@
 import os
 import unittest
-from datetime import datetime, timedelta
+
+from datetime import datetime
 
 import csp
-from csp import ts
-from csp.adapters.csv import CSVReader, YYYYMMDD_TIME_formatter
+
+from csp.adapters.csv import CsvAdapterManager
 
 
+# Current adapter only supports string fields
 class PriceQuantity(csp.Struct):
-    PRICE: float
-    SIZE: int
+    PRICE: str
+    SIZE: str
     SIDE: str
     SYMBOL: str
 
 
-class PriceQuantity2(csp.Struct):
-    price: float
-    quantity: int
-    side: str
-
-
 class TestCSVReader(unittest.TestCase):
+
     def setUp(self):
-        self._filename = os.path.join(os.path.dirname(__file__), "csv_test_data.csv")
-        self._time_formatter = YYYYMMDD_TIME_formatter("TIME")
+        self._filename = os.path.join(
+            os.path.dirname(__file__),
+            "csv_test_data.csv"
+        )
+
+        self.reader = CsvAdapterManager(
+            self._filename,
+            time_column="TIME",
+            symbol_column="SYMBOL",
+            delimiter="|",
+        )
+
 
     def test_basic(self):
+
         def graph():
-            reader = CSVReader(self._filename, self._time_formatter, symbol_column="SYMBOL", delimiter="|")
 
-            # Struct
-            aapl = reader.subscribe("AAPL", PriceQuantity)
-            ibm = reader.subscribe("IBM", PriceQuantity)
+            # Subscribe AAPL
+            aapl = self.reader.subscribe(
+                PriceQuantity,
+                symbol="AAPL"
+            )
 
-            # Struct with fieldMapping
-            aapl2 = reader.subscribe(
-                "AAPL", PriceQuantity2, field_map={"PRICE": "price", "SIZE": "quantity", "SIDE": "side"}
+            # Subscribe IBM
+            ibm = self.reader.subscribe(
+                PriceQuantity,
+                symbol="IBM"
             )
 
-            # specific field
-            aapl_price = reader.subscribe("AAPL", float, field_map="PRICE")
+            # Specific field (string only)
+            aapl_price = self.reader.subscribe(
+                str,
+                symbol="AAPL",
+                field_map="PRICE"
+            )
+
+            # Subscribe all symbols
+            all_data = self.reader.subscribe(
+                PriceQuantity
+            )
 
-            # all data
-            all = reader.subscribe_all(PriceQuantity)
 
             csp.add_graph_output("aapl", aapl)
             csp.add_graph_output("ibm", ibm)
-            csp.add_graph_output("aapl2", aapl2)
             csp.add_graph_output("aapl_price", aapl_price)
-            csp.add_graph_output("all", all)
+            csp.add_graph_output("all", all_data)
 
-        result = csp.run(graph, starttime=datetime(2020, 3, 3, 9, 30))
-        self.assertEqual(len(result["aapl"]), 4)
-        self.assertTrue(all(v[1].SYMBOL == "AAPL" for v in result["aapl"]))
 
-        self.assertEqual(len(result["ibm"]), 2)
-        self.assertTrue(all(v[1].SYMBOL == "IBM" for v in result["ibm"]))
+        result = csp.run(
+            graph,
+            starttime=datetime(2020, 3, 3, 9, 30)
+        )
+
+
+        # AAPL
+        self.assertEqual(
+            len(result["aapl"]),
+            4
+        )
+
+        self.assertTrue(
+            all(
+                v[1].SYMBOL == "AAPL"
+                for v in result["aapl"]
+            )
+        )
+
 
         self.assertEqual(
             [v[1] for v in result["aapl"]],
             [
-                PriceQuantity(PRICE=500.0, SIZE=100, SIDE="BUY", SYMBOL="AAPL"),
-                PriceQuantity(PRICE=400.0, SIZE=100, SIDE="BUY", SYMBOL="AAPL"),
-                PriceQuantity(PRICE=300.0, SIZE=200, SIDE="SELL", SYMBOL="AAPL"),
-                PriceQuantity(PRICE=200.0, SIZE=400, SIDE="BUY", SYMBOL="AAPL"),
+                PriceQuantity(
+                    PRICE="500.00",
+                    SIZE="100",
+                    SIDE="BUY",
+                    SYMBOL="AAPL",
+                ),
+                PriceQuantity(
+                    PRICE="400.00",
+                    SIZE="100",
+                    SIDE="BUY",
+                    SYMBOL="AAPL",
+                ),
+                PriceQuantity(
+                    PRICE="300.00",
+                    SIZE="200",
+                    SIDE="SELL",
+                    SYMBOL="AAPL",
+                ),
+                PriceQuantity(
+                    PRICE="200.00",
+                    SIZE="400",
+                    SIDE="BUY",
+                    SYMBOL="AAPL",
+                ),
             ],
         )
 
+
+        # IBM
+        self.assertEqual(
+            len(result["ibm"]),
+            2
+        )
+
+        self.assertTrue(
+            all(
+                v[1].SYMBOL == "IBM"
+                for v in result["ibm"]
+            )
+        )
+
+
+        # Single field
         self.assertEqual(
-            [v[1] for v in result["aapl2"]],
+            [v[1] for v in result["aapl_price"]],
             [
-                PriceQuantity2(price=500.0, quantity=100, side="BUY"),
-                PriceQuantity2(price=400.0, quantity=100, side="BUY"),
-                PriceQuantity2(
-                    price=300.0,
-                    quantity=200,
-                    side="SELL",
-                ),
-                PriceQuantity2(price=200.0, quantity=400, side="BUY"),
+                "500.00",
+                "400.00",
+                "300.00",
+                "200.00",
             ],
         )
 
-        self.assertEqual([v[1] for v in result["aapl_price"]], [500.0, 400.0, 300.0, 200.0])
-        self.assertEqual(len(result["all"]), 7)
+
+        # Subscribe all
+        self.assertEqual(
+            len(result["all"]),
+            7
+        )
+
+
 
     def test_starttime(self):
-        reader = CSVReader(self._filename, self._time_formatter, symbol_column="SYMBOL", delimiter="|")
-        aapl = reader.subscribe("AAPL", float, "PRICE")
+
+        aapl = self.reader.subscribe(
+            str,
+            symbol="AAPL",
+            field_map="PRICE"
+        )
+
 
         # Exact hit
-        res = csp.run(aapl, starttime=datetime(2020, 3, 3, 9, 30, 4))[0]
-        self.assertEqual(len(res), 2)
-        self.assertEqual(res[0][0], datetime(2020, 3, 3, 9, 30, 4))
+        res = csp.run(
+            aapl,
+            starttime=datetime(2020, 3, 3, 9, 30, 4)
+        )[0]
 
-        # Missed, should start with first found tick
-        res = csp.run(aapl, starttime=datetime(2020, 3, 3, 9, 30, 3, 2))[0]
-        self.assertEqual(len(res), 2)
-        self.assertEqual(res[0][0], datetime(2020, 3, 3, 9, 30, 4))
 
-        # TBD snapshoting
+        self.assertEqual(
+            len(res),
+            2
+        )
+
+        self.assertEqual(
+            res[0][0],
+            datetime(2020, 3, 3, 9, 30, 4)
+        )
+
+
+        # Missed timestamp:
+        # should start from first available tick
+        res = csp.run(
+            aapl,
+            starttime=datetime(2020, 3, 3, 9, 30, 3, 2)
+        )[0]
+
+
+        self.assertEqual(
+            len(res),
+            2
+        )
+
+        self.assertEqual(
+            res[0][0],
+            datetime(2020, 3, 3, 9, 30, 4)
+        )
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()
\ No newline at end of file
diff --git a/docs/wiki/how-tos/Write-Historical-Input-Adapters.md b/docs/wiki/how-tos/Write-Historical-Input-Adapters.md
index 6f4e015cd..7b4a6c464 100644
--- a/docs/wiki/how-tos/Write-Historical-Input-Adapters.md
+++ b/docs/wiki/how-tos/Write-Historical-Input-Adapters.md
@@ -201,215 +201,6 @@ The **\_create** is the bridge between the *--graph--* time AdapterManager repre
 
 Lets take a look at [`CSVReader`](https://github.com/Point72/csp/blob/main/csp/adapters/csv.py) as an example:
 
-```python
-# GRAPH TIME
-class CSVReader:
-    def __init__(self, filename, time_converter, delimiter=',', symbol_column=None):
-        self._filename = filename
-        self._symbol_column = symbol_column
-        self._delimiter = delimiter
-        self._time_converter = time_converter
-
-    def subscribe(self, symbol, typ, field_map=None):
-        return CSVReadAdapter(self, symbol, typ, field_map)
-
-    def _create(self, engine, memo):
-        return CSVReaderImpl(engine, self)
-```
-
-- **`__init__`**: as you can see, all `__init__` does is keep the parameters that the impl will need.
-- **`subscribe`**: API to create an individual timeseries / edge from this file for the given symbol.
-  typ denotes the type of the timeseries to create (ie `ts[int]`) and field_map is used for mapping columns onto `csp.Struct` types.
-  Note that subscribe returns a `CSVReadAdapter` instance.
-  `CSVReadAdapter` is the *--graph--* time representation of the edge (similar to how we defined `csp.curve` above).
-  We pass it `self` as its first argument, which will be used to create the AdapterManager *--impl--*
-- **`_create`**: the method to create the *--impl--* object from the given *--graph--* time representation of the manager
-
-The `CSVReader` would then be used in graph building code like so:
-
-```python
-reader = CSVReader('my_data.csv', time_formatter, symbol_column='SYMBOL', delimiter='|')
-# aapl will represent a ts[PriceQuantity] edge that will tick with rows from
-# the csv file matching on SYMBOL column AAPL
-aapl = reader.subscribe('AAPL', PriceQuantity)
-```
-
-### AdapterManager - **--impl-- runtime**
-
-The AdapterManager *--impl--* is responsible for opening the data source, parsing and processing through all the data and managing all the adapters it needs to feed.
-The impl class should derive from `csp.impl.adaptermanager.AdapterManagerImpl` and implement the following methods:
-
-- **`start(self,starttime,endtime)`**: this is called when the engine starts up.
-  At this point the impl should open the resource providing the data and seek to starttime.
-  starttime/endtime will be tz-unaware datetime objects in UTC time
-- **`stop(self)`**: this is called at the end of the run, resources should be cleaned up at this point
-- **`process_next_sim_timeslice(self, now)`**: this method will be called multiple times through the run.
-  The initial call will provide now with starttime.
-  The impl's responsibility is to process all data at the given timestamp (more on how to do this below).
-  The method should return the next time in the data source, or None if there is no more data to process.
-  The method will be called again with the provided timestamp as "now" in the next iteration.
-  **NOTE** that process_next_sim_timeslice is required to move ahead in time.
-  In most cases the resource data can be supplied in time order, if not it would have to be sorted up front.
-
-`process_next_sim_timeslice` should parse data for a given time/row of data and then push it through to any registered `ManagedSimInputAdapter` that matches on the given row.
-
-### ManagedSimInputAdapter - **--impl-- runtime**
-
-Users will need to define `ManagedSimInputAdapter` derived types to represent the individual timeseries adapter *--impl--* objects.
-Objects should derive from `csp.impl.adaptermanager.ManagedSimInputAdapter`.
-
-`ManagedSimInputAdapter.__init__` takes two arguments:
-
-- **`typ`**: this is the type of the timeseries, ie int for a `ts[int]`
-- **`field_map`**: Optional, field_map is a dictionary used to map source column names → `csp.Struct` field names.
-
-`ManagedSimInputAdapter` defines a method `push_tick()` which takes the value to feed the input for given timeslice (as defined by "now" at the adapter manager level).
-There is also a convenience method called `process_dict()` which will take a dictionary of `{column : value}` entries and convert it properly into the right value based on the given **field_map.**
-
-### ManagedSimInputAdapter - **--graph-- time**
-
-As with the `csp.curve` example, we need to define a graph-time construct that represents a `ManagedSimInputAdapter` edge.
-In order to define this we use `py_managed_adapter_def`.
-`py_managed_adapter_def` is AdapterManager-"aware" and will properly create the AdapterManager *--impl--* the first time its encountered.
-It will then pass the manager impl as an argument to the `ManagedSimInputAdapter`.
-
-```python
-def py_managed_adapter_def(name, adapterimpl, out_type, manager_type, **kwargs):
-"""
-Create a graph representation of a python managed sim input adapter.
-:param name: string name for the adapter
-:param adapterimpl: a derived implementation of csp.impl.adaptermanager.ManagedSimInputAdapter
-:param out_type: the type of the output, should be a ts[] type. Note this can use tvar types if a subsequent argument defines the tvar
-:param manager_type: the type of the graph time representation of the AdapterManager that will manage this adapter
-:param kwargs: **kwargs will be passed through as arguments to the ManagedSimInputAdapter implementation
-the first argument to the implementation will be the adapter manager impl instance
-"""
-```
-
-### Example - CSVReader
-
-Putting this all together lets take a look at a `CSVReader` implementation
-and step through what's going on:
-
-```python
-import csv as pycsv
-from datetime import datetime
-
-from csp import ts
-from csp.impl.adaptermanager import AdapterManagerImpl, ManagedSimInputAdapter
-from csp.impl.wiring import pymanagedadapterdef
-
-# GRAPH TIME
-class CSVReader:
-    def __init__(self, filename, time_converter, delimiter=',', symbol_column=None):
-        self._filename = filename
-        self._symbol_column = symbol_column
-        self._delimiter = delimiter
-        self._time_converter = time_converter
-
-    def subscribe(self, symbol, typ, field_map=None):
-        return CSVReadAdapter(self, symbol, typ, field_map)
-
-    def _create(self, engine, memo):
-        return CSVReaderImpl(engine, self)
-```
-
-Here we define CSVReader, our AdapterManager *--graph--* time representation.
-It holds the parameters that will be used for the impl, it implements a `subscribe()` call for users to create timeseries and defines a \_create method to create a runtime *--impl–-* instance from the graphtime representation.
-Note how on line 17 we pass self to the CSVReadAdapter, this is what binds the input adapter to this AdapterManager
-
-```python
-# RUN TIME
-class CSVReaderImpl(AdapterManagerImpl):                     # 1
-    def __init__(self, engine, adapterRep):                  # 2
-        super().__init__(engine)                             # 3
-                                                             # 4
-        self._rep = adapterRep                               # 5
-        self._inputs = {}                                    # 6
-        self._csv_reader = None                              # 7
-        self._next_row = None                                # 8
-                                                             # 9
-    def start(self, starttime, endtime):                     # 10
-        self._csv_reader = pycsv.DictReader(                 # 11
-            open(self._rep._filename, 'r'),                  # 12
-            delimiter=self._rep._delimiter                   # 13
-        )                                                    # 14
-        self._next_row = None                                # 15
-                                                             # 16
-        for row in self._csv_reader:                         # 17
-            time = self._rep._time_converter(row)            # 18
-            self._next_row = row                             # 19
-            if time >= starttime:                            # 20
-                break                                        # 21
-                                                             # 22
-    def stop(self):                                          # 23
-        self._csv_reader = None                              # 24
-                                                             # 25
-    def register_input_adapter(self, symbol, adapter):       # 26
-        if symbol not in self._inputs:                       # 27
-            self._inputs[symbol] = []                        # 28
-        self._inputs[symbol].append(adapter)                 # 29
-                                                             # 30
-    def process_next_sim_timeslice(self, now):               # 31
-        if not self._next_row:                               # 32
-            return None                                      # 33
-                                                             # 34
-        while True:                                          # 35
-            time = self._rep._time_converter(self._next_row) # 36
-            if time > now:                                   # 37
-                return time                                  # 38
-            self.process_row(self._next_row)                 # 39
-            try:                                             # 40
-                self._next_row = next(self._csv_reader)      # 41
-            except StopIteration:                            # 42
-                return None                                  # 43
-                                                             # 44
-    def process_row(self, row):                              # 45
-        symbol = row[self._rep._symbol_column]               # 46
-        if symbol in self._inputs:                           # 47
-            for input in self._inputs.get(symbol, []):       # 48
-                input.process_dict(row)                      # 49
-```
-
-`CSVReaderImpl` is the runtime *--impl–-*.
-It gets created when the engine is being built from the described graph.
-
-- **lines 10-21 - start()**: this is the start method that gets called with the time range the graph will be run against.
-  Here we open our resource (`pycsv.DictReader`) and scan t through the data until we reach the requested starttime.
-
-- **lines 23-24 - stop()**: this is the stop call that gets called when the engine is done running and is shutdown, we free our resource here
-
-- **lines 26-29**: the `CSVReader` allows one to subscribe to many symbols from one file.
-  symbols are keyed by a provided `SYMBOL` column.
-  The individual adapters will self-register with the `CSVReaderImpl` when they are created with the requested symbol.
-  `CSVReaderImpl` keeps track of what adapters have been registered for what symbol in its `self._inputs` map.
-
-- **lines 31-43**: this is main method that gets invoked repeatedly throughout the run.
-  For every distinct timestamp in the file, this method will get invoked once and the method is expected to go through the resource data for all points with time now, process the row and push the data to any matching adapters.
-  The method returns the next timestamp when its done processing all data for "now", or None if there is no more data.
-  **NOTE** that the csv impl expects the data to be in time order.
-  `process_next_sim_timeslice` must advance time forward.
-
-- **lines 45-49**: this method takes a row of data (provided as a dict from `DictReader`), extracts the symbol and pushes the row through to all input adapters that match
-
-```python
-class CSVReadAdapterImpl(ManagedSimInputAdapter):            # 1
-    def __init__(self, managerImpl, symbol, typ, field_map): # 2
-        managerImpl.register_input_adapter(symbol, self)     # 3
-        super().__init__(typ, field_map)                     # 4
-                                                             # 5
-CSVReadAdapter = py_managed_adapter_def(                     # 6
-    'csvadapter',
-    CSVReadAdapterImpl,
-    ts['T'],
-    CSVReader,
-    symbol=str,
-    typ='T',
-    fieldMap=(object, None)
-)
-```
-
-- **line 3**: this is where the instance of an adapter *--impl--* registers itself with the `CSVReaderImpl`.
-- **line 6+**: this is where we define `CSVReadAdapter`, the *--graph--* time representation of a CSV adapter, returned from `CSVReader.subscribe`
+TODO
 
 See example [e3_adaptermanager_pullinput.py](https://github.com/Point72/csp/blob/main/examples/04_writing_adapters/e3_adaptermanager_pullinput.py) for another example of how to write a managed sim adapter manager.

From e62190f89c71eb5e0961dbec477410b5ca762d5c Mon Sep 17 00:00:00 2001
From: Andrew Sasmito <asasmito1920@gmail.com>
Date: Mon, 3 Aug 2026 19:50:45 -0400
Subject: [PATCH 4/7] Fix linter issue

Signed-off-by: Andrew Sasmito <asasmito1920@gmail.com>
---
 .../adapters/csv/CsvInputAdapterManager.cpp   | 653 +++++++++---------
 cpp/csp/adapters/csv/CsvInputAdapterManager.h | 163 +++--
 cpp/csp/python/adapters/csvadapterimpl.cpp    |  78 ++-
 csp/adapters/csv.py                           |   5 +-
 csp/tests/adapters/test_csv.py                | 115 +--
 5 files changed, 455 insertions(+), 559 deletions(-)

diff --git a/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp b/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp
index c25e55eb9..be7307abd 100644
--- a/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp
+++ b/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp
@@ -7,395 +7,370 @@
 #include <csp/engine/Struct.h>
 #include <cstddef>
 #include <functional>
-#include <unordered_map>
-#include <set>
-#include <ranges>
 #include <iterator>
-#include <string>
 #include <optional>
+#include <ranges>
+#include <set>
+#include <string>
+#include <unordered_map>
 #include <variant>
 #include <vector>
 
-namespace csp::adapters::csv
-{
+namespace csp::adapters::csv {
 
 // Parses exactly "YYYY-MM-DD HH::MM::SS"
 DateTime parseFixed_YmdHMS(std::string_view date) {
-    if (date.size() < 19)
-        CSP_THROW(ValueError, "Timestamp too short");
-
-    auto d2 = [&](size_t i) {
-        return (date[i] - '0') * 10 + (date[i + 1] - '0');
-    };
-    
-    int year            = d2(0) * 100 + d2(2);
-    int month           = d2(5);
-    int day             = d2(8);
-
-    int hour            = d2(11);
-    int minute          = d2(14);
-    int second          = d2(17);
-
-    return DateTime(year, month, day, hour, minute, second);
-}
+  if (date.size() < 19)
+    CSP_THROW(ValueError, "Timestamp too short");
+
+  auto d2 = [&](size_t i) {
+    return (date[i] - '0') * 10 + (date[i + 1] - '0');
+  };
 
+  int year = d2(0) * 100 + d2(2);
+  int month = d2(5);
+  int day = d2(8);
 
-CsvInputAdapterManager::CsvInputAdapterManager( csp::Engine *engine, const Dictionary &properties ) :
-        AdapterManager( engine )
-{
-    m_filename         = properties.get<std::string>( "filename", "" ); 
-    CSP_TRUE_OR_THROW_RUNTIME( !m_filename.empty(), "Filename must be provided" );
+  int hour = d2(11);
+  int minute = d2(14);
+  int second = d2(17);
 
-    auto tz = properties.get<std::string>( "tz", "UTC" );
-    CSP_TRUE_OR_THROW_RUNTIME( tz == "UTC",
-                               "Only UTC default timezone is supported, got:" << tz );
+  return DateTime(year, month, day, hour, minute, second);
+}
 
-    properties.tryGet( "start_time", m_startTime );
-    properties.tryGet( "end_time", m_endTime );
+CsvInputAdapterManager::CsvInputAdapterManager(csp::Engine *engine,
+                                               const Dictionary &properties)
+    : AdapterManager(engine) {
+  m_filename = properties.get<std::string>("filename", "");
+  CSP_TRUE_OR_THROW_RUNTIME(!m_filename.empty(), "Filename must be provided");
 
-    m_delimiter        = properties.get<std::string>( "delimiter", ",");
-    m_hasHeader        = properties.get<bool>       ( "hasHeader", true);
-    m_timeColumn       = properties.get<std::string>( "time_column", "" );
-    m_symbolColumnName = properties.get<std::string>( "symbol_column", "" );
+  auto tz = properties.get<std::string>("tz", "UTC");
+  CSP_TRUE_OR_THROW_RUNTIME(
+      tz == "UTC", "Only UTC default timezone is supported, got:" << tz);
 
-    CSP_TRUE_OR_THROW_RUNTIME( m_timeColumn != "", "Time column can't be empty" );
+  properties.tryGet("start_time", m_startTime);
+  properties.tryGet("end_time", m_endTime);
 
-    properties.tryGet( "time_format", m_timeFormat );
+  m_delimiter = properties.get<std::string>("delimiter", ",");
+  m_hasHeader = properties.get<bool>("hasHeader", true);
+  m_timeColumn = properties.get<std::string>("time_column", "");
+  m_symbolColumnName = properties.get<std::string>("symbol_column", "");
 
-    if (m_timeFormat.empty()) {
-        dateParser = parseFixed_YmdHMS;
-    } else if ( m_timeFormat == "YYYY-MM-DD HH::MM::SS") {
-        dateParser = parseFixed_YmdHMS;
-    } else {
-        CSP_THROW(ValueError, "Time format not supported");
-    }
+  CSP_TRUE_OR_THROW_RUNTIME(m_timeColumn != "", "Time column can't be empty");
+
+  properties.tryGet("time_format", m_timeFormat);
+
+  if (m_timeFormat.empty()) {
+    dateParser = parseFixed_YmdHMS;
+  } else if (m_timeFormat == "YYYY-MM-DD HH::MM::SS") {
+    dateParser = parseFixed_YmdHMS;
+  } else {
+    CSP_THROW(ValueError, "Time format not supported");
+  }
 }
 
 CsvInputAdapterManager::~CsvInputAdapterManager() = default;
 
 void CsvInputAdapterManager::setupProcessor(
-    const std::vector<std::string> & schema,
-    const std::set<std::string> & neededColumns,
-    std::optional<std::string> symbolColumn,
-    bool subscribeAllOnEmptySymbol )
-{
-    m_schema = {};
-    m_symbolColumn = std::nullopt;
-
-    for( int i = 0; i < std::ssize( schema ); ++i ) {
-        if (neededColumns.contains( schema[i] )) {
-            m_schema.push_back(i);
-        }
+    const std::vector<std::string> &schema,
+    const std::set<std::string> &neededColumns,
+    std::optional<std::string> symbolColumn, bool subscribeAllOnEmptySymbol) {
+  m_schema = {};
+  m_symbolColumn = std::nullopt;
+
+  for (int i = 0; i < std::ssize(schema); ++i) {
+    if (neededColumns.contains(schema[i])) {
+      m_schema.push_back(i);
+    }
 
-        if (symbolColumn && *symbolColumn == schema[i]) [[ unlikely ]] {
-            m_symbolColumn = i;
-        }
+    if (symbolColumn && *symbolColumn == schema[i]) [[unlikely]] {
+      m_symbolColumn = i;
     }
+  }
+}
+
+ManagedSimInputAdapter *CsvInputAdapterManager::getInputAdapter(
+    CspTypePtr &type, const Dictionary &properties, PushMode pushMode) {
+  // Per-subscription symbol filter. Empty string means "subscribe to every
+  // row".
+  std::string symbol = properties.get<std::string>("symbol", "");
+
+  auto *adapter =
+      engine()->createOwnedObject<ManagedSimInputAdapter>(type, this, pushMode);
+
+  Subscriber sub;
+  sub.m_adapter = adapter;
+
+  // Stash the field_map so the subscriber can convert its row at dispatch time.
+  // string -> single-column; DictionaryPtr -> struct field map;
+  // absent/None -> whole-row dict.
+  if (properties.exists("field_map")) {
+    auto &fm = properties.getUntypedValue("field_map");
+    if (std::holds_alternative<std::string>(fm))
+      sub.m_fieldMap = std::get<std::string>(fm);
+    else if (std::holds_alternative<DictionaryPtr>(fm))
+      sub.m_fieldMap = std::get<DictionaryPtr>(fm);
+    // else: leave monostate — whole-row dict output
+  }
+
+  if (symbol.empty())
+    m_subscribers.push_back(std::move(sub));
+  else
+    m_subscribersBySymbol[symbol].push_back(std::move(sub));
+
+  return adapter;
 }
 
-ManagedSimInputAdapter * CsvInputAdapterManager::getInputAdapter(
-    CspTypePtr & type, const Dictionary & properties, PushMode pushMode )
-{
-    // Per-subscription symbol filter. Empty string means "subscribe to every row".
-    std::string symbol = properties.get<std::string>( "symbol", "" );
-
-    auto * adapter = engine() -> createOwnedObject<ManagedSimInputAdapter>(
-        type, this, pushMode );
-
-    Subscriber sub;
-    sub.m_adapter = adapter;
-
-    // Stash the field_map so the subscriber can convert its row at dispatch time.
-    // string -> single-column; DictionaryPtr -> struct field map;
-    // absent/None -> whole-row dict.
-    if( properties.exists( "field_map" ) )
-    {
-        auto & fm = properties.getUntypedValue( "field_map" );
-        if( std::holds_alternative<std::string>( fm ) )
-            sub.m_fieldMap = std::get<std::string>( fm );
-        else if( std::holds_alternative<DictionaryPtr>( fm ) )
-            sub.m_fieldMap = std::get<DictionaryPtr>( fm );
-        // else: leave monostate — whole-row dict output
+void CsvInputAdapterManager::start(DateTime starttime, DateTime endtime) {
+  if (!m_startTime.isNone()) {
+    starttime = std::max(starttime, m_startTime);
+  }
+  AdapterManager::start(starttime, endtime);
+
+  m_file = std::ifstream(m_filename, std::ios::binary);
+  if (!m_file)
+    CSP_THROW(IOError, "Failed to open " << m_filename);
+
+  // Reusable split helper: split a line by m_delimiter into owned strings,
+  // trimming a trailing '\r' from the final field for Windows CSVs.
+  auto splitLine = [](std::string_view line, std::string_view delim) {
+    std::vector<std::string> parts;
+    for (auto part : std::views::split(line, delim)) {
+      auto begin = part.begin();
+      auto len = std::ranges::distance(part);
+      parts.emplace_back(len == 0 ? std::string() : std::string(&*begin, len));
     }
+    if (!parts.empty() && !parts.back().empty() && parts.back().back() == '\r')
+      parts.back().pop_back();
+    return parts;
+  };
+
+  // --- Parse header ---
+  m_columnNames.clear();
+  if (m_hasHeader) {
+    std::string headerLine;
+    if (!std::getline(m_file, headerLine))
+      CSP_THROW(IOError, "Failed to read header from " << m_filename);
+    m_columnNames = splitLine(headerLine, m_delimiter);
+  }
+
+  // --- Collect the set of columns any subscriber cares about ---
+  std::set<std::string> neededColumns;
+  bool needAllColumns = false;
+
+  if (!m_timeColumn.empty())
+    neededColumns.insert(m_timeColumn);
+  if (!m_symbolColumnName.empty())
+    neededColumns.insert(m_symbolColumnName);
+
+  auto collectFrom = [&](const Subscriber &sub) {
+    if (std::holds_alternative<std::monostate>(sub.m_fieldMap)) {
+      needAllColumns = true;
+    } else if (std::holds_alternative<std::string>(sub.m_fieldMap)) {
+      neededColumns.insert(std::get<std::string>(sub.m_fieldMap));
+    } else if (std::holds_alternative<DictionaryPtr>(sub.m_fieldMap)) {
+      auto &fm = std::get<DictionaryPtr>(sub.m_fieldMap);
+      for (auto it = fm->begin(); it != fm->end(); ++it)
+        neededColumns.insert(it.key());
+    }
+  };
+
+  for (const auto &sub : m_subscribers)
+    collectFrom(sub);
+  for (const auto &[symbol, subs] : m_subscribersBySymbol)
+    for (const auto &sub : subs)
+      collectFrom(sub);
+
+  if (needAllColumns)
+    for (const auto &name : m_columnNames)
+      neededColumns.insert(name);
+
+  // --- Do column-name -> index mappings through setupProcessor ---
+  std::optional<std::string> symbolColumnOpt;
+  if (!m_symbolColumnName.empty())
+    symbolColumnOpt = m_symbolColumnName;
+
+  bool subscribeAllOnEmptySymbol = !m_subscribers.empty();
+  setupProcessor(m_columnNames, neededColumns, symbolColumnOpt,
+                 subscribeAllOnEmptySymbol);
+
+  // Locate the time column so processNextSimTimeSlice can extract it O(1).
+  m_timeColumnIndex = -1;
+  for (int i = 0; i < std::ssize(m_columnNames); ++i) {
+    if (m_columnNames[i] == m_timeColumn) {
+      m_timeColumnIndex = i;
+      break;
+    }
+  }
+
+  CSP_TRUE_OR_THROW_RUNTIME(m_timeColumnIndex >= 0,
+                            "Time column '" << m_timeColumn
+                                            << "' not found in CSV header");
 
-    if( symbol.empty() )
-        m_subscribers.push_back( std::move( sub ) );
-    else
-        m_subscribersBySymbol[ symbol ].push_back( std::move( sub ) );
+  if (!m_symbolColumnName.empty()) {
+    CSP_TRUE_OR_THROW_RUNTIME(m_symbolColumn.has_value(),
+                              "Symbol column '" << m_symbolColumnName
+                                                << "' not found in CSV header");
+  }
 
-    return adapter;
+  // Bind each subscriber's row -> tick callback now that the schema is known.
+  bindSubscriberDispatchers();
+
+  // Cache the first data row so processNextSimTimeSlice's skip loop has data to
+  // compare.
+  if (!std::getline(m_file, m_row))
+    m_row.clear();
 }
 
-void CsvInputAdapterManager::start( DateTime starttime, DateTime endtime )
-{
-    if( !m_startTime.isNone() )
-    {
-        starttime = std::max( starttime, m_startTime );
-    }
-    AdapterManager::start( starttime, endtime );
-
-    m_file = std::ifstream(m_filename, std::ios::binary);
-    if (!m_file) CSP_THROW(IOError, "Failed to open " << m_filename);
-
-    // Reusable split helper: split a line by m_delimiter into owned strings,
-    // trimming a trailing '\r' from the final field for Windows CSVs.
-    auto splitLine = []( std::string_view line, std::string_view delim ) {
-        std::vector<std::string> parts;
-        for( auto part : std::views::split( line, delim ) ) {
-            auto begin = part.begin();
-            auto len = std::ranges::distance( part );
-            parts.emplace_back( len == 0 ? std::string() : std::string( &*begin, len ) );
-        }
-        if( !parts.empty() && !parts.back().empty() && parts.back().back() == '\r' )
-            parts.back().pop_back();
-        return parts;
-    };
-
-    // --- Parse header ---
-    m_columnNames.clear();
-    if( m_hasHeader )
-    {
-        std::string headerLine;
-        if( !std::getline( m_file, headerLine ) )
-            CSP_THROW( IOError, "Failed to read header from " << m_filename );
-        m_columnNames = splitLine( headerLine, m_delimiter );
-    }
+void CsvInputAdapterManager::bindSubscriberDispatchers() {
+  // Column name -> header index (built once).
+  std::unordered_map<std::string_view, size_t> colIndex;
+  colIndex.reserve(m_columnNames.size());
+  for (size_t i = 0; i < m_columnNames.size(); ++i)
+    colIndex[m_columnNames[i]] = i;
 
-    // --- Collect the set of columns any subscriber cares about ---
-    std::set<std::string> neededColumns;
-    bool needAllColumns = false;
+  auto bind = [&](Subscriber &sub) {
+    // Whole-row dict: Build a struct
+    if (std::holds_alternative<std::monostate>(sub.m_fieldMap)) {
+      auto *structType =
+          static_cast<const CspStructType *>(sub.m_adapter->dataType());
 
-    if( !m_timeColumn.empty() )        neededColumns.insert( m_timeColumn );
-    if( !m_symbolColumnName.empty() )  neededColumns.insert( m_symbolColumnName );
+      auto meta = structType->meta();
 
-    auto collectFrom = [&]( const Subscriber & sub ) {
-        if( std::holds_alternative<std::monostate>( sub.m_fieldMap ) )
-        {
-            needAllColumns = true;
-        }
-        else if( std::holds_alternative<std::string>( sub.m_fieldMap ) )
-        {
-            neededColumns.insert( std::get<std::string>( sub.m_fieldMap ) );
-        }
-        else if( std::holds_alternative<DictionaryPtr>( sub.m_fieldMap ) )
-        {
-            auto & fm = std::get<DictionaryPtr>( sub.m_fieldMap );
-            for( auto it = fm -> begin(); it != fm -> end(); ++it )
-                neededColumns.insert( it.key() );
-        }
-    };
-
-    for( const auto & sub : m_subscribers ) collectFrom( sub );
-    for( const auto & [ symbol, subs ] : m_subscribersBySymbol )
-        for( const auto & sub : subs ) collectFrom( sub );
-
-    if( needAllColumns )
-        for( const auto & name : m_columnNames )
-            neededColumns.insert( name );
-
-    // --- Do column-name -> index mappings through setupProcessor ---
-    std::optional<std::string> symbolColumnOpt;
-    if( !m_symbolColumnName.empty() )
-        symbolColumnOpt = m_symbolColumnName;
-
-    bool subscribeAllOnEmptySymbol = !m_subscribers.empty();
-    setupProcessor( m_columnNames, neededColumns, symbolColumnOpt, subscribeAllOnEmptySymbol );
-
-    // Locate the time column so processNextSimTimeSlice can extract it O(1).
-    m_timeColumnIndex = -1;
-    for( int i = 0; i < std::ssize( m_columnNames ); ++i )
-    {
-        if( m_columnNames[ i ] == m_timeColumn )
-        {
-            m_timeColumnIndex = i;
-            break;
-        }
-    }
-    
-    CSP_TRUE_OR_THROW_RUNTIME( m_timeColumnIndex >= 0,
-        "Time column '" << m_timeColumn << "' not found in CSV header" );
-
-    if( !m_symbolColumnName.empty() )
-    {
-        CSP_TRUE_OR_THROW_RUNTIME( m_symbolColumn.has_value(),
-            "Symbol column '" << m_symbolColumnName << "' not found in CSV header" );
-    }
+      StructSubscription subscription;
+      subscription.m_adapter = sub.m_adapter;
+      subscription.m_structMeta = meta;
 
-    // Bind each subscriber's row -> tick callback now that the schema is known.
-    bindSubscriberDispatchers();
+      for (size_t i = 0; i < m_columnNames.size(); i++) {
+        auto field = meta->field(m_columnNames[i]);
 
-    // Cache the first data row so processNextSimTimeSlice's skip loop has data to compare.
-    if( !std::getline( m_file, m_row ) )
-        m_row.clear();
-}
+        if (!field)
+          continue;
 
-void CsvInputAdapterManager::bindSubscriberDispatchers()
-{
-    // Column name -> header index (built once).
-    std::unordered_map<std::string_view, size_t> colIndex;
-    colIndex.reserve( m_columnNames.size() );
-    for( size_t i = 0; i < m_columnNames.size(); ++i )
-        colIndex[ m_columnNames[ i ] ] = i;
-
-    auto bind = [&]( Subscriber & sub )
-    {
-        // Whole-row dict: Build a struct
-        if(std::holds_alternative<std::monostate>(sub.m_fieldMap))
-        {
-            auto *structType =
-                static_cast<const CspStructType *>(sub.m_adapter->dataType());
-
-            auto meta = structType->meta();
-
-            StructSubscription subscription;
-            subscription.m_adapter = sub.m_adapter;
-            subscription.m_structMeta = meta;
-
-
-            for(size_t i = 0; i < m_columnNames.size(); i++)
-            {
-                auto field = meta->field(m_columnNames[i]);
-
-                if(!field)
-                    continue;
-
-                subscription.m_fieldSetters.push_back(
-                    [i, field](StructPtr &s,
-                            const std::vector<std::string_view> &cols)
-                    {
-                        field->setValue<std::string>(
-                            s.get(),
-                            std::string(cols[i])
-                        );
-                    }
-                );
-            }
-
-            sub.m_structSubscription = std::move(subscription);
-            return;
-        }
+        subscription.m_fieldSetters.push_back(
+            [i, field](StructPtr &s,
+                       const std::vector<std::string_view> &cols) {
+              field->setValue<std::string>(s.get(), std::string(cols[i]));
+            });
+      }
 
-        // Single-column subscription — extract the named column and push it.
-        if( std::holds_alternative<std::string>( sub.m_fieldMap ) )
-        {
-            const auto & colName = std::get<std::string>( sub.m_fieldMap );
-            auto it = colIndex.find( colName );
-            CSP_TRUE_OR_THROW_RUNTIME( it != colIndex.end(),
-                "Column '" << colName << "' not found in CSV header" );
-            size_t idx = it -> second;
-
-            auto * adapter = sub.m_adapter;
-            auto tag = adapter -> dataType() -> type();
-
-            if( tag == CspType::Type::STRING )
-            {
-                // Only pure-string ticks are implementable without Python.
-                sub.m_dispatch = [ adapter, idx ]( const std::vector<std::string_view> & cols )
-                {
-                    adapter -> pushTick<std::string>( std::string( cols[ idx ] ) );
-                };
-            }
-            // else: leave m_dispatch null;
-            return;
-        }
+      sub.m_structSubscription = std::move(subscription);
+      return;
+    }
+
+    // Single-column subscription — extract the named column and push it.
+    if (std::holds_alternative<std::string>(sub.m_fieldMap)) {
+      const auto &colName = std::get<std::string>(sub.m_fieldMap);
+      auto it = colIndex.find(colName);
+      CSP_TRUE_OR_THROW_RUNTIME(it != colIndex.end(),
+                                "Column '" << colName
+                                           << "' not found in CSV header");
+      size_t idx = it->second;
+
+      auto *adapter = sub.m_adapter;
+      auto tag = adapter->dataType()->type();
+
+      if (tag == CspType::Type::STRING) {
+        // Only pure-string ticks are implementable without Python.
+        sub.m_dispatch = [adapter,
+                          idx](const std::vector<std::string_view> &cols) {
+          adapter->pushTick<std::string>(std::string(cols[idx]));
+        };
+      }
+      // else: leave m_dispatch null;
+      return;
+    }
 
-        // Struct field_map: needs csp::Struct construction with per-field type
-    };
+    // Struct field_map: needs csp::Struct construction with per-field type
+  };
 
-    for( auto & sub : m_subscribers ) bind( sub );
-    for( auto & [ symbol, subs ] : m_subscribersBySymbol )
-        for( auto & sub : subs ) bind( sub );
+  for (auto &sub : m_subscribers)
+    bind(sub);
+  for (auto &[symbol, subs] : m_subscribersBySymbol)
+    for (auto &sub : subs)
+      bind(sub);
 }
 
-void CsvInputAdapterManager::stop()
-{
-    m_subscribers.clear();
-    m_subscribersBySymbol.clear();
-    m_schema.clear();
-    m_file.close();
-    AdapterManager::stop();
+void CsvInputAdapterManager::stop() {
+  m_subscribers.clear();
+  m_subscribersBySymbol.clear();
+  m_schema.clear();
+  m_file.close();
+  AdapterManager::stop();
 }
 
-DateTime CsvInputAdapterManager::processNextSimTimeSlice( DateTime time )
-{
-    if( m_row.empty() ) [[unlikely]]
-        return DateTime::NONE();
-
-    // Split m_row once per row into string_views. Views are valid until the
-    // next getline() mutates m_row, so every read of `cols` must precede the
-    // next read from the file.
-    auto splitRow = [ this ]()
-    {
-        std::vector<std::string_view> cols;
-        for( auto part : std::views::split( m_row, m_delimiter ) )
-        {
-            auto begin = part.begin();
-            auto len = std::ranges::distance( part );
-            cols.emplace_back( len == 0 ? std::string_view()
-                                        : std::string_view( &*begin, len ) );
-        }
-        // Trim trailing '\r' on Windows CSVs so the final field parses cleanly.
-        if( !cols.empty() && !cols.back().empty() && cols.back().back() == '\r' )
-            cols.back().remove_suffix( 1 );
-        return cols;
-    };
-
-    // Skip loop: advance until we find a row at or after `time`.
-    std::vector<std::string_view> cols = splitRow();
-    DateTime rowTime = dateParser( cols[ m_timeColumnIndex ] );
-    while( rowTime < time )
-    {
-        if( !std::getline( m_file, m_row ) )
-        {
-            m_row.clear();
-            return DateTime::NONE();
-        }
-        cols = splitRow();
-        rowTime = dateParser( cols[ m_timeColumnIndex ] );
+DateTime CsvInputAdapterManager::processNextSimTimeSlice(DateTime time) {
+  if (m_row.empty()) [[unlikely]]
+    return DateTime::NONE();
+
+  // Split m_row once per row into string_views. Views are valid until the
+  // next getline() mutates m_row, so every read of `cols` must precede the
+  // next read from the file.
+  auto splitRow = [this]() {
+    std::vector<std::string_view> cols;
+    for (auto part : std::views::split(m_row, m_delimiter)) {
+      auto begin = part.begin();
+      auto len = std::ranges::distance(part);
+      cols.emplace_back(len == 0 ? std::string_view()
+                                 : std::string_view(&*begin, len));
     }
+    // Trim trailing '\r' on Windows CSVs so the final field parses cleanly.
+    if (!cols.empty() && !cols.back().empty() && cols.back().back() == '\r')
+      cols.back().remove_suffix(1);
+    return cols;
+  };
+
+  // Skip loop: advance until we find a row at or after `time`.
+  std::vector<std::string_view> cols = splitRow();
+  DateTime rowTime = dateParser(cols[m_timeColumnIndex]);
+  while (rowTime < time) {
+    if (!std::getline(m_file, m_row)) {
+      m_row.clear();
+      return DateTime::NONE();
+    }
+    cols = splitRow();
+    rowTime = dateParser(cols[m_timeColumnIndex]);
+  }
 
-    if( !m_endTime.isNone() && rowTime > m_endTime )
-        return DateTime::NONE();
+  if (!m_endTime.isNone() && rowTime > m_endTime)
+    return DateTime::NONE();
 
-    if( rowTime > time )
-        return rowTime;
+  if (rowTime > time)
+    return rowTime;
 
-    // Dispatch every row with this exact timestamp.
-    do
-    {
-        // Subscribe-all subscribers see every row.
-        for( auto & sub : m_subscribers ) {
-            if(sub.m_structSubscription)
-                sub.m_structSubscription->dispatchValue(cols);
+  // Dispatch every row with this exact timestamp.
+  do {
+    // Subscribe-all subscribers see every row.
+    for (auto &sub : m_subscribers) {
+      if (sub.m_structSubscription)
+        sub.m_structSubscription->dispatchValue(cols);
 
-            else if(sub.m_dispatch)
-                sub.m_dispatch(cols);
-        }
+      else if (sub.m_dispatch)
+        sub.m_dispatch(cols);
+    }
 
-        // Symbol-filtered subscribers only see rows where their symbol matches.
-        if( m_symbolColumn.has_value() )
-        {
-            std::string sym( cols[ *m_symbolColumn ] );
-            auto it = m_subscribersBySymbol.find( sym );
-            if( it != m_subscribersBySymbol.end() )
-                for( auto & sub : it -> second ) {
-                    if(sub.m_structSubscription)
-                        sub.m_structSubscription->dispatchValue(cols);
-                    
-                    else if(sub.m_dispatch)
-                        sub.m_dispatch(cols);
-                }
+    // Symbol-filtered subscribers only see rows where their symbol matches.
+    if (m_symbolColumn.has_value()) {
+      std::string sym(cols[*m_symbolColumn]);
+      auto it = m_subscribersBySymbol.find(sym);
+      if (it != m_subscribersBySymbol.end())
+        for (auto &sub : it->second) {
+          if (sub.m_structSubscription)
+            sub.m_structSubscription->dispatchValue(cols);
+
+          else if (sub.m_dispatch)
+            sub.m_dispatch(cols);
         }
+    }
 
-        if( !std::getline( m_file, m_row ) )
-        {
-            m_row.clear();
-            return DateTime::NONE();
-        }
-        cols = splitRow();
-        rowTime = dateParser( cols[ m_timeColumnIndex ] );
-    } while( rowTime == time );
+    if (!std::getline(m_file, m_row)) {
+      m_row.clear();
+      return DateTime::NONE();
+    }
+    cols = splitRow();
+    rowTime = dateParser(cols[m_timeColumnIndex]);
+  } while (rowTime == time);
 
-    return rowTime;
-}
+  return rowTime;
 }
+} // namespace csp::adapters::csv
diff --git a/cpp/csp/adapters/csv/CsvInputAdapterManager.h b/cpp/csp/adapters/csv/CsvInputAdapterManager.h
index 0e0669d73..840d12d47 100644
--- a/cpp/csp/adapters/csv/CsvInputAdapterManager.h
+++ b/cpp/csp/adapters/csv/CsvInputAdapterManager.h
@@ -7,108 +7,107 @@
 #include <csp/engine/Struct.h>
 #include <fstream>
 #include <functional>
-#include <unordered_map>
+#include <optional>
 #include <set>
 #include <string>
-#include <optional>
+#include <unordered_map>
 #include <variant>
 #include <vector>
 
-
-namespace csp::adapters::csv
-{
-
+namespace csp::adapters::csv {
 
 // Manages all csv input adapters for a single engine run.
 //
 // Lifecycle:
-//   1. Registration:  getInputAdapter() called per subscription (before engine starts)
+//   1. Registration:  getInputAdapter() called per subscription (before engine
+//   starts)
 //   2. start():       create processors → wire adapters → read first row
 //   3. processNextSimTimeSlice():  skip/dispatch loop per engine tick
 //   4. stop():        tear down all state
-class CsvInputAdapterManager final : public csp::AdapterManager
-{
+class CsvInputAdapterManager final : public csp::AdapterManager {
 public:
+  CsvInputAdapterManager(csp::Engine *engine, const Dictionary &properties);
 
-    CsvInputAdapterManager( csp::Engine *engine, const Dictionary &properties );
+  ~CsvInputAdapterManager();
 
-    ~CsvInputAdapterManager();
+  const char *name() const override { return "CsvInputAdapterManager"; }
 
-    const char *name() const override{ return "CsvInputAdapterManager"; }
+  void start(DateTime starttime, DateTime endtime) override;
+  void stop() override;
+  DateTime processNextSimTimeSlice(DateTime time) override;
 
-    void start( DateTime starttime, DateTime endtime ) override;
-    void stop() override;
-    DateTime processNextSimTimeSlice( DateTime time ) override;
-
-    ManagedSimInputAdapter * getInputAdapter( CspTypePtr & type, const Dictionary & properties, PushMode pushMode );
+  ManagedSimInputAdapter *getInputAdapter(CspTypePtr &type,
+                                          const Dictionary &properties,
+                                          PushMode pushMode);
 
 private:
-
-    void setupProcessor( const std::vector<std::string> & schema,
-                         const std::set<std::string> & neededColumns,
-                         std::optional<std::string> symbolColumn,
-                         bool subscribeAllOnEmptySymbol );
-
-    bool readNextRow();
-
-    struct StructSubscription
-    {
-        using FieldSetter = std::function<void(StructPtr &, const std::vector<std::string_view> &)>;
-
-        ManagedSimInputAdapter *m_adapter;
-        std::shared_ptr<const StructMeta> m_structMeta;
-
-        std::vector<FieldSetter> m_fieldSetters;
-
-        void dispatchValue(const std::vector<std::string_view> &cols)
-        {
-            StructPtr value = m_structMeta->create();
-
-            for(auto &setter : m_fieldSetters)
-                setter(value, cols);
-
-            m_adapter->pushTick(value);
-        }
-    };
-
-    struct Subscriber {
-        ManagedSimInputAdapter *                                          m_adapter;
-        std::variant<std::monostate, std::string, DictionaryPtr>          m_fieldMap;
-
-        std::function<void(const std::vector<std::string_view> &)>        m_dispatch;
-        std::optional<StructSubscription>                                 m_structSubscription;
-    };
-
-    // Walk registered subscribers and bind each one's m_dispatch based on its
-    // field_map + the parsed header. Must be called after m_columnNames is populated.
-    void bindSubscriberDispatchers();
-
-    using dateTimeParserfn = DateTime(*)(std::string_view);
-
-    // Registration-phase state (populated by getInputAdapter before start)
-    std::vector<Subscriber> m_subscribers;
-    std::unordered_map<std::string, std::vector<Subscriber>> m_subscribersBySymbol;
-
-    // Configuration (from properties dict)
-    csp::DateTime                       m_startTime;
-    csp::DateTime                       m_endTime;
-    std::string                         m_timeColumn;
-    std::string                         m_timeFormat;
-    std::string                         m_filename;
-    std::string                         m_delimiter;
-    std::string                         m_symbolColumnName; // configured symbol column name ("" = no symbol column)
-    bool                                m_hasHeader;
-    dateTimeParserfn                    dateParser;
-
-    // Runtime state (initialized in start, used in processNextSimTimeSlice)
-    std::vector<std::string>                    m_columnNames; // full header, in order
-    std::optional<int>                          m_symbolColumn; // Index of symbol column
-    std::vector<int>                            m_schema; // Indices to be used
-    int                                         m_timeColumnIndex;
-    std::ifstream                               m_file;
-    std::string                                 m_row; // Current cached row
+  void setupProcessor(const std::vector<std::string> &schema,
+                      const std::set<std::string> &neededColumns,
+                      std::optional<std::string> symbolColumn,
+                      bool subscribeAllOnEmptySymbol);
+
+  bool readNextRow();
+
+  struct StructSubscription {
+    using FieldSetter =
+        std::function<void(StructPtr &, const std::vector<std::string_view> &)>;
+
+    ManagedSimInputAdapter *m_adapter;
+    std::shared_ptr<const StructMeta> m_structMeta;
+
+    std::vector<FieldSetter> m_fieldSetters;
+
+    void dispatchValue(const std::vector<std::string_view> &cols) {
+      StructPtr value = m_structMeta->create();
+
+      for (auto &setter : m_fieldSetters)
+        setter(value, cols);
+
+      m_adapter->pushTick(value);
+    }
+  };
+
+  struct Subscriber {
+    ManagedSimInputAdapter *m_adapter;
+    std::variant<std::monostate, std::string, DictionaryPtr> m_fieldMap;
+
+    std::function<void(const std::vector<std::string_view> &)> m_dispatch;
+    std::optional<StructSubscription> m_structSubscription;
+  };
+
+  // Walk registered subscribers and bind each one's m_dispatch based on its
+  // field_map + the parsed header. Must be called after m_columnNames is
+  // populated.
+  void bindSubscriberDispatchers();
+
+  using dateTimeParserfn = DateTime (*)(std::string_view);
+
+  // Registration-phase state (populated by getInputAdapter before start)
+  std::vector<Subscriber> m_subscribers;
+  std::unordered_map<std::string, std::vector<Subscriber>>
+      m_subscribersBySymbol;
+
+  // Configuration (from properties dict)
+  csp::DateTime m_startTime;
+  csp::DateTime m_endTime;
+  std::string m_timeColumn;
+  std::string m_timeFormat;
+  std::string m_filename;
+  std::string m_delimiter;
+  std::string m_symbolColumnName; // configured symbol column name ("" = no
+                                  // symbol column)
+  bool m_hasHeader;
+  dateTimeParserfn dateParser;
+
+  // Runtime state (initialized in start, used in processNextSimTimeSlice)
+  std::vector<std::string> m_columnNames; // full header, in order
+  std::optional<int> m_symbolColumn;      // Index of symbol column
+  std::vector<int> m_schema;              // Indices to be used
+  int m_timeColumnIndex;
+  std::ifstream m_file;
+  std::string m_row; // Current cached row
 };
 
-}
+} // namespace csp::adapters::csv
 
 #endif // _IN_CSP_ADAPTERS_CSV_CsvInputAdapterManager_H
diff --git a/cpp/csp/python/adapters/csvadapterimpl.cpp b/cpp/csp/python/adapters/csvadapterimpl.cpp
index 4005989f5..8ac5cbecb 100644
--- a/cpp/csp/python/adapters/csvadapterimpl.cpp
+++ b/cpp/csp/python/adapters/csvadapterimpl.cpp
@@ -10,57 +10,59 @@
 
 using namespace csp::adapters::csv;
 
-namespace csp::python
-{
+namespace csp::python {
 
-//AdapterManager
-csp::AdapterManager * create_csv_adapter_manager( PyEngine * engine, const Dictionary & properties )
-{
-    return engine -> engine() -> createOwnedObject<CsvInputAdapterManager>( properties );
+// AdapterManager
+csp::AdapterManager *create_csv_adapter_manager(PyEngine *engine,
+                                                const Dictionary &properties) {
+  return engine->engine()->createOwnedObject<CsvInputAdapterManager>(
+      properties);
 }
 
-static InputAdapter * create_csv_input_adapter( csp::AdapterManager * manager, PyEngine * pyengine, PyObject * pyType, PushMode pushMode, PyObject * args )
-{
-    auto & cspType = pyTypeAsCspType( pyType );
+static InputAdapter *
+create_csv_input_adapter(csp::AdapterManager *manager, PyEngine *pyengine,
+                         PyObject *pyType, PushMode pushMode, PyObject *args) {
+  auto &cspType = pyTypeAsCspType(pyType);
 
-    PyObject * pyProperties;
-    PyObject * type;
+  PyObject *pyProperties;
+  PyObject *type;
 
-    auto * csvManager = dynamic_cast<CsvInputAdapterManager*>( manager );
-    if( !csvManager )
-        CSP_THROW( TypeError, "Expected CsvInputAdapterManager" );
+  auto *csvManager = dynamic_cast<CsvInputAdapterManager *>(manager);
+  if (!csvManager)
+    CSP_THROW(TypeError, "Expected CsvInputAdapterManager");
 
-    if( !PyArg_ParseTuple( args, "O!O!",
-                           &PyType_Type, &type,
-                           &PyDict_Type, &pyProperties ) )
-        CSP_THROW( PythonPassthrough, "" );
+  if (!PyArg_ParseTuple(args, "O!O!", &PyType_Type, &type, &PyDict_Type,
+                        &pyProperties))
+    CSP_THROW(PythonPassthrough, "");
 
-    return csvManager -> getInputAdapter( cspType, fromPython<Dictionary>( pyProperties ), pushMode );
+  return csvManager->getInputAdapter(
+      cspType, fromPython<Dictionary>(pyProperties), pushMode);
 }
 
-REGISTER_ADAPTER_MANAGER( _csv_adapter_manager, create_csv_adapter_manager );
-REGISTER_INPUT_ADAPTER(   _csv_input_adapter,   create_csv_input_adapter );
+REGISTER_ADAPTER_MANAGER(_csv_adapter_manager, create_csv_adapter_manager);
+REGISTER_INPUT_ADAPTER(_csv_input_adapter, create_csv_input_adapter);
 
-static PyModuleDef _csvadapterimpl_module = {
-        PyModuleDef_HEAD_INIT,
-        "_csvadapterimpl",
-        "_csvadapterimpl c++ module",
-        -1,
-        NULL, NULL, NULL, NULL, NULL
-};
+static PyModuleDef _csvadapterimpl_module = {PyModuleDef_HEAD_INIT,
+                                             "_csvadapterimpl",
+                                             "_csvadapterimpl c++ module",
+                                             -1,
+                                             NULL,
+                                             NULL,
+                                             NULL,
+                                             NULL,
+                                             NULL};
 
-PyMODINIT_FUNC PyInit__csvadapterimpl(void)
-{
-    PyObject* m;
+PyMODINIT_FUNC PyInit__csvadapterimpl(void) {
+  PyObject *m;
 
-    m = PyModule_Create( &_csvadapterimpl_module);
-    if( m == NULL )
-        return NULL;
+  m = PyModule_Create(&_csvadapterimpl_module);
+  if (m == NULL)
+    return NULL;
 
-    if( !InitHelper::instance().execute( m ) )
-        return NULL;
+  if (!InitHelper::instance().execute(m))
+    return NULL;
 
-    return m;
+  return m;
 }
 
-}
+} // namespace csp::python
diff --git a/csp/adapters/csv.py b/csp/adapters/csv.py
index b5fa67daa..4e446a8ab 100644
--- a/csp/adapters/csv.py
+++ b/csp/adapters/csv.py
@@ -45,10 +45,7 @@ def subscribe(
         )
 
     def _create(self, engine, memo):
-        return _csvadapterimpl._csv_adapter_manager(
-            engine,
-            self._properties
-        )
+        return _csvadapterimpl._csv_adapter_manager(engine, self._properties)
 
 
 _csv_input_adapter_def = input_adapter_def(
diff --git a/csp/tests/adapters/test_csv.py b/csp/tests/adapters/test_csv.py
index 5fe7d08b6..a3d4e7ed3 100644
--- a/csp/tests/adapters/test_csv.py
+++ b/csp/tests/adapters/test_csv.py
@@ -17,12 +17,8 @@ class PriceQuantity(csp.Struct):
 
 
 class TestCSVReader(unittest.TestCase):
-
     def setUp(self):
-        self._filename = os.path.join(
-            os.path.dirname(__file__),
-            "csv_test_data.csv"
-        )
+        self._filename = os.path.join(os.path.dirname(__file__), "csv_test_data.csv")
 
         self.reader = CsvAdapterManager(
             self._filename,
@@ -31,61 +27,31 @@ def setUp(self):
             delimiter="|",
         )
 
-
     def test_basic(self):
-
         def graph():
-
             # Subscribe AAPL
-            aapl = self.reader.subscribe(
-                PriceQuantity,
-                symbol="AAPL"
-            )
+            aapl = self.reader.subscribe(PriceQuantity, symbol="AAPL")
 
             # Subscribe IBM
-            ibm = self.reader.subscribe(
-                PriceQuantity,
-                symbol="IBM"
-            )
+            ibm = self.reader.subscribe(PriceQuantity, symbol="IBM")
 
             # Specific field (string only)
-            aapl_price = self.reader.subscribe(
-                str,
-                symbol="AAPL",
-                field_map="PRICE"
-            )
+            aapl_price = self.reader.subscribe(str, symbol="AAPL", field_map="PRICE")
 
             # Subscribe all symbols
-            all_data = self.reader.subscribe(
-                PriceQuantity
-            )
-
+            all_data = self.reader.subscribe(PriceQuantity)
 
             csp.add_graph_output("aapl", aapl)
             csp.add_graph_output("ibm", ibm)
             csp.add_graph_output("aapl_price", aapl_price)
             csp.add_graph_output("all", all_data)
 
-
-        result = csp.run(
-            graph,
-            starttime=datetime(2020, 3, 3, 9, 30)
-        )
-
+        result = csp.run(graph, starttime=datetime(2020, 3, 3, 9, 30))
 
         # AAPL
-        self.assertEqual(
-            len(result["aapl"]),
-            4
-        )
-
-        self.assertTrue(
-            all(
-                v[1].SYMBOL == "AAPL"
-                for v in result["aapl"]
-            )
-        )
+        self.assertEqual(len(result["aapl"]), 4)
 
+        self.assertTrue(all(v[1].SYMBOL == "AAPL" for v in result["aapl"]))
 
         self.assertEqual(
             [v[1] for v in result["aapl"]],
@@ -117,20 +83,10 @@ def graph():
             ],
         )
 
-
         # IBM
-        self.assertEqual(
-            len(result["ibm"]),
-            2
-        )
-
-        self.assertTrue(
-            all(
-                v[1].SYMBOL == "IBM"
-                for v in result["ibm"]
-            )
-        )
+        self.assertEqual(len(result["ibm"]), 2)
 
+        self.assertTrue(all(v[1].SYMBOL == "IBM" for v in result["ibm"]))
 
         # Single field
         self.assertEqual(
@@ -143,60 +99,27 @@ def graph():
             ],
         )
 
-
         # Subscribe all
-        self.assertEqual(
-            len(result["all"]),
-            7
-        )
-
-
+        self.assertEqual(len(result["all"]), 7)
 
     def test_starttime(self):
-
-        aapl = self.reader.subscribe(
-            str,
-            symbol="AAPL",
-            field_map="PRICE"
-        )
-
+        aapl = self.reader.subscribe(str, symbol="AAPL", field_map="PRICE")
 
         # Exact hit
-        res = csp.run(
-            aapl,
-            starttime=datetime(2020, 3, 3, 9, 30, 4)
-        )[0]
-
-
-        self.assertEqual(
-            len(res),
-            2
-        )
+        res = csp.run(aapl, starttime=datetime(2020, 3, 3, 9, 30, 4))[0]
 
-        self.assertEqual(
-            res[0][0],
-            datetime(2020, 3, 3, 9, 30, 4)
-        )
+        self.assertEqual(len(res), 2)
 
+        self.assertEqual(res[0][0], datetime(2020, 3, 3, 9, 30, 4))
 
         # Missed timestamp:
         # should start from first available tick
-        res = csp.run(
-            aapl,
-            starttime=datetime(2020, 3, 3, 9, 30, 3, 2)
-        )[0]
+        res = csp.run(aapl, starttime=datetime(2020, 3, 3, 9, 30, 3, 2))[0]
 
+        self.assertEqual(len(res), 2)
 
-        self.assertEqual(
-            len(res),
-            2
-        )
-
-        self.assertEqual(
-            res[0][0],
-            datetime(2020, 3, 3, 9, 30, 4)
-        )
+        self.assertEqual(res[0][0], datetime(2020, 3, 3, 9, 30, 4))
 
 
 if __name__ == "__main__":
-    unittest.main()
\ No newline at end of file
+    unittest.main()

From 909f6b62afb4e105eac2ca84c53bc02ada2c1d56 Mon Sep 17 00:00:00 2001
From: Andrew Sasmito <asasmito1920@gmail.com>
Date: Thu, 13 Aug 2026 01:03:30 -0400
Subject: [PATCH 5/7] Remove CmakeFiles

Signed-off-by: Andrew Sasmito <asasmito1920@gmail.com>
---
 CMakeFiles/CMakeSystem.cmake | 15 ---------------
 1 file changed, 15 deletions(-)
 delete mode 100644 CMakeFiles/CMakeSystem.cmake

diff --git a/CMakeFiles/CMakeSystem.cmake b/CMakeFiles/CMakeSystem.cmake
deleted file mode 100644
index 0473afc46..000000000
--- a/CMakeFiles/CMakeSystem.cmake
+++ /dev/null
@@ -1,15 +0,0 @@
-set(CMAKE_HOST_SYSTEM "Darwin-24.6.0")
-set(CMAKE_HOST_SYSTEM_NAME "Darwin")
-set(CMAKE_HOST_SYSTEM_VERSION "24.6.0")
-set(CMAKE_HOST_SYSTEM_PROCESSOR "arm64")
-
-
-
-set(CMAKE_SYSTEM "Darwin-24.6.0")
-set(CMAKE_SYSTEM_NAME "Darwin")
-set(CMAKE_SYSTEM_VERSION "24.6.0")
-set(CMAKE_SYSTEM_PROCESSOR "arm64")
-
-set(CMAKE_CROSSCOMPILING "FALSE")
-
-set(CMAKE_SYSTEM_LOADED 1)

From ca746a476674117859226153467274e273849350 Mon Sep 17 00:00:00 2001
From: Andrew Sasmito <asasmito1920@gmail.com>
Date: Thu, 13 Aug 2026 01:08:48 -0400
Subject: [PATCH 6/7] Remove compile_commands

Signed-off-by: Andrew Sasmito <asasmito1920@gmail.com>
---
 compile_commands.json | 1 -
 1 file changed, 1 deletion(-)
 delete mode 120000 compile_commands.json

diff --git a/compile_commands.json b/compile_commands.json
deleted file mode 120000
index 25eb4b2b4..000000000
--- a/compile_commands.json
+++ /dev/null
@@ -1 +0,0 @@
-build/compile_commands.json
\ No newline at end of file

From 02be585859725b779922d947f47faedf3be73dd9 Mon Sep 17 00:00:00 2001
From: Andrew Sasmito <asasmito1920@gmail.com>
Date: Thu, 13 Aug 2026 20:14:00 -0400
Subject: [PATCH 7/7] Address comments

Signed-off-by: Andrew Sasmito <asasmito1920@gmail.com>
---
 cpp/cmake/modules/FindDepsCsvAdapter.cmake    |   1 +
 .../adapters/csv/CsvInputAdapterManager.cpp   |  49 ++++++-
 csp/adapters/csv.py                           |   2 +-
 csp/tests/adapters/test_csv.py                | 127 ++++++++----------
 4 files changed, 106 insertions(+), 73 deletions(-)

diff --git a/cpp/cmake/modules/FindDepsCsvAdapter.cmake b/cpp/cmake/modules/FindDepsCsvAdapter.cmake
index 512d3d4f0..fa235a998 100644
--- a/cpp/cmake/modules/FindDepsCsvAdapter.cmake
+++ b/cpp/cmake/modules/FindDepsCsvAdapter.cmake
@@ -1,3 +1,4 @@
+# No external dependencies are needed at this time
 cmake_minimum_required(VERSION 3.7.2)
 
 set(DepsCsvAdapter_FOUND TRUE)
diff --git a/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp b/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp
index be7307abd..765eb6595 100644
--- a/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp
+++ b/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp
@@ -145,7 +145,7 @@ void CsvInputAdapterManager::start(DateTime starttime, DateTime endtime) {
     return parts;
   };
 
-  // --- Parse header ---
+  // Parse header
   m_columnNames.clear();
   if (m_hasHeader) {
     std::string headerLine;
@@ -154,7 +154,7 @@ void CsvInputAdapterManager::start(DateTime starttime, DateTime endtime) {
     m_columnNames = splitLine(headerLine, m_delimiter);
   }
 
-  // --- Collect the set of columns any subscriber cares about ---
+  // Collect the set of columns any subscriber cares about
   std::set<std::string> neededColumns;
   bool needAllColumns = false;
 
@@ -185,7 +185,7 @@ void CsvInputAdapterManager::start(DateTime starttime, DateTime endtime) {
     for (const auto &name : m_columnNames)
       neededColumns.insert(name);
 
-  // --- Do column-name -> index mappings through setupProcessor ---
+  // Do column-name -> index mappings through setupProcessor
   std::optional<std::string> symbolColumnOpt;
   if (!m_symbolColumnName.empty())
     symbolColumnOpt = m_symbolColumnName;
@@ -281,7 +281,48 @@ void CsvInputAdapterManager::bindSubscriberDispatchers() {
       return;
     }
 
-    // Struct field_map: needs csp::Struct construction with per-field type
+    if (std::holds_alternative<DictionaryPtr>(sub.m_fieldMap)) {
+      auto *structType =
+          static_cast<const CspStructType *>(sub.m_adapter->dataType());
+
+      auto meta = structType->meta();
+
+      StructSubscription subscription;
+      subscription.m_adapter = sub.m_adapter;
+      subscription.m_structMeta = meta;
+
+      auto &fm = std::get<DictionaryPtr>(sub.m_fieldMap);
+
+      for (auto it = fm->begin(); it != fm->end(); ++it) {
+        const std::string csvColumn = it.key();
+        const std::string structField = it.value<std::string>();
+
+        auto col = colIndex.find(csvColumn);
+
+        CSP_TRUE_OR_THROW_RUNTIME(
+            col != colIndex.end(),
+            "Column '" << csvColumn << "' not found in CSV header");
+
+        auto field = meta->field(structField);
+
+        CSP_TRUE_OR_THROW_RUNTIME(
+            field,
+            "Field '" << structField << "' not found in struct");
+
+        size_t idx = col->second;
+
+        subscription.m_fieldSetters.push_back(
+            [idx, field](StructPtr &s,
+                         const std::vector<std::string_view> &cols) {
+              field->setValue<std::string>(
+                  s.get(),
+                  std::string(cols[idx]));
+            });
+      }
+
+      sub.m_structSubscription = std::move(subscription);
+      return;
+    }
   };
 
   for (auto &sub : m_subscribers)
diff --git a/csp/adapters/csv.py b/csp/adapters/csv.py
index 4e446a8ab..3ecbe78b9 100644
--- a/csp/adapters/csv.py
+++ b/csp/adapters/csv.py
@@ -8,8 +8,8 @@ def __init__(
         self,
         filename,
         time_column,
-        symbol_column="",
         delimiter=",",
+        symbol_column="",
         has_header=True,
         time_format=None,
     ):
diff --git a/csp/tests/adapters/test_csv.py b/csp/tests/adapters/test_csv.py
index a3d4e7ed3..ae7182d81 100644
--- a/csp/tests/adapters/test_csv.py
+++ b/csp/tests/adapters/test_csv.py
@@ -1,14 +1,11 @@
 import os
 import unittest
-
 from datetime import datetime
 
 import csp
-
 from csp.adapters.csv import CsvAdapterManager
 
 
-# Current adapter only supports string fields
 class PriceQuantity(csp.Struct):
     PRICE: str
     SIZE: str
@@ -16,110 +13,104 @@ class PriceQuantity(csp.Struct):
     SYMBOL: str
 
 
+class PriceQuantity2(csp.Struct):
+    price: str
+    quantity: str
+    side: str
+
+
 class TestCSVReader(unittest.TestCase):
     def setUp(self):
         self._filename = os.path.join(os.path.dirname(__file__), "csv_test_data.csv")
 
-        self.reader = CsvAdapterManager(
-            self._filename,
-            time_column="TIME",
-            symbol_column="SYMBOL",
-            delimiter="|",
-        )
-
     def test_basic(self):
         def graph():
-            # Subscribe AAPL
-            aapl = self.reader.subscribe(PriceQuantity, symbol="AAPL")
-
-            # Subscribe IBM
-            ibm = self.reader.subscribe(PriceQuantity, symbol="IBM")
-
-            # Specific field (string only)
-            aapl_price = self.reader.subscribe(str, symbol="AAPL", field_map="PRICE")
-
-            # Subscribe all symbols
-            all_data = self.reader.subscribe(PriceQuantity)
+            reader = CsvAdapterManager(
+                self._filename,
+                time_column="TIME",
+                symbol_column="SYMBOL",
+                delimiter="|",
+            )
+
+            # Struct
+            aapl = reader.subscribe(PriceQuantity, symbol="AAPL")
+            ibm = reader.subscribe(PriceQuantity, symbol="IBM")
+
+            # Struct with fieldMapping
+            aapl2 = reader.subscribe(
+                PriceQuantity2,
+                symbol="AAPL",
+                field_map={"PRICE": "price", "SIZE": "quantity", "SIDE": "side"},
+            )
+
+            # specific field
+            aapl_price = reader.subscribe(str, symbol="AAPL", field_map="PRICE")
+
+            # all data
+            all = reader.subscribe(PriceQuantity)
 
             csp.add_graph_output("aapl", aapl)
             csp.add_graph_output("ibm", ibm)
+            csp.add_graph_output("aapl2", aapl2)
             csp.add_graph_output("aapl_price", aapl_price)
-            csp.add_graph_output("all", all_data)
+            csp.add_graph_output("all", all)
 
         result = csp.run(graph, starttime=datetime(2020, 3, 3, 9, 30))
 
-        # AAPL
         self.assertEqual(len(result["aapl"]), 4)
-
         self.assertTrue(all(v[1].SYMBOL == "AAPL" for v in result["aapl"]))
 
+        self.assertEqual(len(result["ibm"]), 2)
+        self.assertTrue(all(v[1].SYMBOL == "IBM" for v in result["ibm"]))
+
         self.assertEqual(
             [v[1] for v in result["aapl"]],
             [
-                PriceQuantity(
-                    PRICE="500.00",
-                    SIZE="100",
-                    SIDE="BUY",
-                    SYMBOL="AAPL",
-                ),
-                PriceQuantity(
-                    PRICE="400.00",
-                    SIZE="100",
-                    SIDE="BUY",
-                    SYMBOL="AAPL",
-                ),
-                PriceQuantity(
-                    PRICE="300.00",
-                    SIZE="200",
-                    SIDE="SELL",
-                    SYMBOL="AAPL",
-                ),
-                PriceQuantity(
-                    PRICE="200.00",
-                    SIZE="400",
-                    SIDE="BUY",
-                    SYMBOL="AAPL",
-                ),
+                PriceQuantity(PRICE="500.00", SIZE="100", SIDE="BUY", SYMBOL="AAPL"),
+                PriceQuantity(PRICE="400.00", SIZE="100", SIDE="BUY", SYMBOL="AAPL"),
+                PriceQuantity(PRICE="300.00", SIZE="200", SIDE="SELL", SYMBOL="AAPL"),
+                PriceQuantity(PRICE="200.00", SIZE="400", SIDE="BUY", SYMBOL="AAPL"),
             ],
         )
 
-        # IBM
-        self.assertEqual(len(result["ibm"]), 2)
-
-        self.assertTrue(all(v[1].SYMBOL == "IBM" for v in result["ibm"]))
-
-        # Single field
         self.assertEqual(
-            [v[1] for v in result["aapl_price"]],
+            [v[1] for v in result["aapl2"]],
             [
-                "500.00",
-                "400.00",
-                "300.00",
-                "200.00",
+                PriceQuantity2(price="500.00", quantity="100", side="BUY"),
+                PriceQuantity2(price="400.00", quantity="100", side="BUY"),
+                PriceQuantity2(price="300.00", quantity="200", side="SELL"),
+                PriceQuantity2(price="200.00", quantity="400", side="BUY"),
             ],
         )
 
-        # Subscribe all
+        self.assertEqual(
+            [v[1] for v in result["aapl_price"]],
+            ["500.00", "400.00", "300.00", "200.00"],
+        )
+
         self.assertEqual(len(result["all"]), 7)
 
     def test_starttime(self):
-        aapl = self.reader.subscribe(str, symbol="AAPL", field_map="PRICE")
+        reader = CsvAdapterManager(
+            self._filename,
+            time_column="TIME",
+            symbol_column="SYMBOL",
+            delimiter="|",
+        )
+        aapl = reader.subscribe(str, symbol="AAPL", field_map="PRICE")
 
         # Exact hit
         res = csp.run(aapl, starttime=datetime(2020, 3, 3, 9, 30, 4))[0]
-
         self.assertEqual(len(res), 2)
-
         self.assertEqual(res[0][0], datetime(2020, 3, 3, 9, 30, 4))
 
-        # Missed timestamp:
-        # should start from first available tick
+        # Missed, should start with first found tick
         res = csp.run(aapl, starttime=datetime(2020, 3, 3, 9, 30, 3, 2))[0]
-
         self.assertEqual(len(res), 2)
-
         self.assertEqual(res[0][0], datetime(2020, 3, 3, 9, 30, 4))
 
+        # TBD snapshoting
+
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()
\ No newline at end of file