diff --git a/CMakeLists.txt b/CMakeLists.txt index e12cb2fde..b95654bc1 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..fa235a998 --- /dev/null +++ b/cpp/cmake/modules/FindDepsCsvAdapter.cmake @@ -0,0 +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/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 new file mode 100644 index 000000000..fa5f4fd4d --- /dev/null +++ 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 new file mode 100644 index 000000000..765eb6595 --- /dev/null +++ b/cpp/csp/adapters/csv/CsvInputAdapterManager.cpp @@ -0,0 +1,417 @@ +#include "csp/core/Exception.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +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) + : AdapterManager(engine) { + m_filename = properties.get("filename", ""); + CSP_TRUE_OR_THROW_RUNTIME(!m_filename.empty(), "Filename must be provided"); + + auto tz = properties.get("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("delimiter", ","); + m_hasHeader = properties.get("hasHeader", true); + m_timeColumn = properties.get("time_column", ""); + m_symbolColumnName = properties.get("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 &schema, + const std::set &neededColumns, + std::optional 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("symbol", ""); + + auto *adapter = + engine()->createOwnedObject(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(fm)) + sub.m_fieldMap = std::get(fm); + else if (std::holds_alternative(fm)) + sub.m_fieldMap = std::get(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 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 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(sub.m_fieldMap)) { + needAllColumns = true; + } else if (std::holds_alternative(sub.m_fieldMap)) { + neededColumns.insert(std::get(sub.m_fieldMap)); + } else if (std::holds_alternative(sub.m_fieldMap)) { + auto &fm = std::get(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 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 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(sub.m_fieldMap)) { + auto *structType = + static_cast(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 &cols) { + field->setValue(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(sub.m_fieldMap)) { + const auto &colName = std::get(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 &cols) { + adapter->pushTick(std::string(cols[idx])); + }; + } + // else: leave m_dispatch null; + return; + } + + if (std::holds_alternative(sub.m_fieldMap)) { + auto *structType = + static_cast(sub.m_adapter->dataType()); + + auto meta = structType->meta(); + + StructSubscription subscription; + subscription.m_adapter = sub.m_adapter; + subscription.m_structMeta = meta; + + auto &fm = std::get(sub.m_fieldMap); + + for (auto it = fm->begin(); it != fm->end(); ++it) { + const std::string csvColumn = it.key(); + const std::string structField = it.value(); + + 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 &cols) { + field->setValue( + s.get(), + std::string(cols[idx])); + }); + } + + sub.m_structSubscription = std::move(subscription); + return; + } + }; + + 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 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 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_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); + + return rowTime; +} +} // namespace csp::adapters::csv diff --git a/cpp/csp/adapters/csv/CsvInputAdapterManager.h b/cpp/csp/adapters/csv/CsvInputAdapterManager.h new file mode 100644 index 000000000..840d12d47 --- /dev/null +++ b/cpp/csp/adapters/csv/CsvInputAdapterManager.h @@ -0,0 +1,113 @@ +#ifndef _IN_CSP_ADAPTERS_CSV_CsvInputAdapterManager_H +#define _IN_CSP_ADAPTERS_CSV_CsvInputAdapterManager_H + +#include "csp/core/Time.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +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); + + ~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 &schema, + const std::set &neededColumns, + std::optional symbolColumn, + bool subscribeAllOnEmptySymbol); + + bool readNextRow(); + + struct StructSubscription { + using FieldSetter = + std::function &)>; + + ManagedSimInputAdapter *m_adapter; + std::shared_ptr m_structMeta; + + std::vector m_fieldSetters; + + void dispatchValue(const std::vector &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 m_fieldMap; + + std::function &)> m_dispatch; + std::optional 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 m_subscribers; + std::unordered_map> + 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 m_columnNames; // full header, in order + std::optional m_symbolColumn; // Index of symbol column + std::vector 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/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 new file mode 100644 index 000000000..8ac5cbecb --- /dev/null +++ b/cpp/csp/python/adapters/csvadapterimpl.cpp @@ -0,0 +1,68 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace csp::adapters::csv; + +namespace csp::python { + +// AdapterManager +csp::AdapterManager *create_csv_adapter_manager(PyEngine *engine, + const Dictionary &properties) { + return engine->engine()->createOwnedObject( + 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(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(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; +} + +} // namespace csp::python diff --git a/csp/adapters/csv.py b/csp/adapters/csv.py index 82dc3d347..3ecbe78b9 100644 --- a/csp/adapters/csv.py +++ b/csp/adapters/csv.py @@ -1,113 +1,58 @@ -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, + delimiter=",", + symbol_column="", + 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) + return _csvadapterimpl._csv_adapter_manager(engine, self._properties) -CSVReadAdapter = py_managed_adapter_def( - "csvadapter", CSVReadAdapterImpl, ts["T"], CSVReader, symbol=str, typ="T", fieldMap=(object, None) +_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..ae7182d81 100644 --- a/csp/tests/adapters/test_csv.py +++ b/csp/tests/adapters/test_csv.py @@ -1,48 +1,53 @@ 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 class PriceQuantity(csp.Struct): - PRICE: float - SIZE: int + PRICE: str + SIZE: str SIDE: str SYMBOL: str class PriceQuantity2(csp.Struct): - price: float - quantity: int + 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._time_formatter = YYYYMMDD_TIME_formatter("TIME") def test_basic(self): def graph(): - reader = CSVReader(self._filename, self._time_formatter, symbol_column="SYMBOL", delimiter="|") + reader = CsvAdapterManager( + self._filename, + time_column="TIME", + symbol_column="SYMBOL", + delimiter="|", + ) # Struct - aapl = reader.subscribe("AAPL", PriceQuantity) - ibm = reader.subscribe("IBM", PriceQuantity) + aapl = reader.subscribe(PriceQuantity, symbol="AAPL") + ibm = reader.subscribe(PriceQuantity, symbol="IBM") # Struct with fieldMapping aapl2 = reader.subscribe( - "AAPL", PriceQuantity2, field_map={"PRICE": "price", "SIZE": "quantity", "SIDE": "side"} + PriceQuantity2, + symbol="AAPL", + field_map={"PRICE": "price", "SIZE": "quantity", "SIDE": "side"}, ) # specific field - aapl_price = reader.subscribe("AAPL", float, field_map="PRICE") + aapl_price = reader.subscribe(str, symbol="AAPL", field_map="PRICE") # all data - all = reader.subscribe_all(PriceQuantity) + all = reader.subscribe(PriceQuantity) csp.add_graph_output("aapl", aapl) csp.add_graph_output("ibm", ibm) @@ -51,6 +56,7 @@ def graph(): csp.add_graph_output("all", all) 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"])) @@ -60,33 +66,38 @@ def graph(): 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"), ], ) self.assertEqual( [v[1] for v in result["aapl2"]], [ - 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"), + 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"), ], ) - self.assertEqual([v[1] for v in result["aapl_price"]], [500.0, 400.0, 300.0, 200.0]) + 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): - reader = CSVReader(self._filename, self._time_formatter, symbol_column="SYMBOL", delimiter="|") - aapl = reader.subscribe("AAPL", float, "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] @@ -102,4 +113,4 @@ def test_starttime(self): 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. diff --git a/setup.py b/setup.py index fc2616fb1..86e1216e6 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"),