Refactoring operations to be within a thread to scale up backtesting sweeps

Author: mccaffers

CMakeLists.txt

@@ -11,7 +11,7 @@ execute_process(
 project(BacktestingEngine)
 
 # Set the C++ standard
-set(CMAKE_CXX_STANDARD 20)
+set(CMAKE_CXX_STANDARD 23)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 
 # Configure libpqxx build

backtesting-engine-cpp.xcodeproj/project.pbxproj

@@ -141,6 +141,8 @@
 		94829FC32FCC1D1200710E6E /* env.hpp */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.h; path = env.hpp; sourceTree = "<group>"; };
 		94829FC42FCC1D1A00710E6E /* env.cpp */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.cpp; path = env.cpp; sourceTree = "<group>"; };
 		948A9CCD2C906A5600E23669 /* CONVENTIONS.md */ = {isa = PBXFileReference; lastKnownFileType = net.daringfireball.markdown; path = CONVENTIONS.md; sourceTree = "<group>"; };
+		94B4F02A2FD618B300B08FB4 /* backtestLog.hpp */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.h; path = backtestLog.hpp; sourceTree = "<group>"; };
+		94B4F02B2FD618B300B08FB4 /* threadPool.hpp */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.h; path = threadPool.hpp; sourceTree = "<group>"; };
 		94BBA4512D2EA2640010E04D /* build.sh */ = {isa = PBXFileReference; lastKnownFileType = text.script.sh; path = build.sh; sourceTree = "<group>"; };
 		94C331A02FA899A8006BD690 /* decimal_json.hpp */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.h; path = decimal_json.hpp; sourceTree = "<group>"; };
 		94CD832A2D2D22C900041BBA /* config.yml */ = {isa = PBXFileReference; lastKnownFileType = text.yaml; path = config.yml; sourceTree = "<group>"; };
@@ -1505,6 +1507,8 @@
 		94B8C7932D3D770800E17EB6 /* utilities */ = {
 			isa = PBXGroup;
 			children = (
+				94B4F02A2FD618B300B08FB4 /* backtestLog.hpp */,
+				94B4F02B2FD618B300B08FB4 /* threadPool.hpp */,
 				945F475C2FD5607E00D19164 /* queueKeys.hpp */,
 				943770432FD4351100317424 /* parameterSweep.hpp */,
 				94829FC32FCC1D1200710E6E /* env.hpp */,
@@ -3811,7 +3815,7 @@
 				ASSETCATALOG_COMPILER_GENERATE_SWIFT_ASSET_SYMBOL_EXTENSIONS = YES;
 				CLANG_ANALYZER_NONNULL = YES;
 				CLANG_ANALYZER_NUMBER_OBJECT_CONVERSION = YES_AGGRESSIVE;
-				CLANG_CXX_LANGUAGE_STANDARD = "gnu++20";
+				CLANG_CXX_LANGUAGE_STANDARD = "gnu++23";
 				CLANG_ENABLE_MODULES = YES;
 				CLANG_ENABLE_OBJC_ARC = YES;
 				CLANG_ENABLE_OBJC_WEAK = YES;
@@ -3872,7 +3876,7 @@
 				ASSETCATALOG_COMPILER_GENERATE_SWIFT_ASSET_SYMBOL_EXTENSIONS = YES;
 				CLANG_ANALYZER_NONNULL = YES;
 				CLANG_ANALYZER_NUMBER_OBJECT_CONVERSION = YES_AGGRESSIVE;
-				CLANG_CXX_LANGUAGE_STANDARD = "gnu++20";
+				CLANG_CXX_LANGUAGE_STANDARD = "gnu++23";
 				CLANG_ENABLE_MODULES = YES;
 				CLANG_ENABLE_OBJC_ARC = YES;
 				CLANG_ENABLE_OBJC_WEAK = YES;

include/tradingResults.hpp

@@ -36,6 +36,7 @@ struct TradingResultsStats {
 struct TradingResults {
     std::string RUN_ID;
     std::string timestamp;
+    double durationSeconds;          // wall-clock seconds for this run's backtest
     trading_definitions::Configuration config;
     TradingResultsStats results;
 

include/utilities/backtestLog.hpp

@@ -0,0 +1,38 @@
+// Backtesting Engine in C++
+//
+// (c) 2026 Ryan McCaffery | https://mccaffers.com
+// This code is licensed under MIT license (see LICENSE.txt for details)
+// ---------------------------------------
+
+#ifndef UTILITIES_BACKTEST_LOG_HPP
+#define UTILITIES_BACKTEST_LOG_HPP
+
+#include <atomic>
+#include <iostream>
+#include <mutex>
+#include <string_view>
+
+// Cross-cutting logging switches for backtest execution.
+//
+// When backtests run concurrently (see RedisRunner's thread pool) the chatty
+// per-strategy output would both interleave and race on the shared std::cout /
+// std::cerr. RedisRunner sets `quiet` so those call sites skip their output
+// entirely (no stream access -> no race), while error() serialises the rare
+// failure path behind a mutex so genuine problems still surface safely.
+namespace backtest_log {
+
+inline std::atomic<bool> quiet{false};
+
+inline std::mutex& errorMutex() {
+    static std::mutex m;
+    return m;
+}
+
+inline void error(std::string_view message) {
+    std::scoped_lock lock(errorMutex());
+    std::cerr << message << std::endl;
+}
+
+}  // namespace backtest_log
+
+#endif  // UTILITIES_BACKTEST_LOG_HPP

include/utilities/threadPool.hpp

@@ -0,0 +1,119 @@
+// Backtesting Engine in C++
+//
+// (c) 2026 Ryan McCaffery | https://mccaffers.com
+// This code is licensed under MIT license (see LICENSE.txt for details)
+// ---------------------------------------
+
+#ifndef UTILITIES_THREAD_POOL_HPP
+#define UTILITIES_THREAD_POOL_HPP
+
+#include <condition_variable>
+#include <cstddef>
+#include <exception>
+#include <functional>
+#include <mutex>
+#include <queue>
+#include <stop_token>
+#include <thread>
+#include <utility>
+#include <vector>
+
+// A small fixed-size pool for CPU-bound work, built on std::jthread.
+//
+// std::jthread (not std::thread) buys us three things here:
+//   * Constructor exception safety: if the OS throws while spawning the Nth
+//     worker, unwinding the vector auto-request_stop()s and joins the workers
+//     already started. std::thread would hit un-joined threads -> terminate().
+//   * A trivial destructor: ~std::jthread does request_stop() then join(), so
+//     there is no manual stop flag, signal, or join loop to maintain.
+//   * Native cancellation: each worker takes a std::stop_token and waits with
+//     condition_variable_any, so a stop request wakes the wait directly.
+//
+// submit() applies backpressure: it blocks once `capacity_` tasks are in flight
+// (queued + executing). This bounds memory and paces the producer to the
+// workers. wait() blocks until everything has finished and never throws, so it
+// is safe to call during stack unwinding (e.g. from a destructor). The first
+// exception thrown by any task is captured and surfaced via takeError().
+class ThreadPool {
+public:
+    explicit ThreadPool(std::size_t threads) : capacity_(threads * 2) {
+        workers_.reserve(threads);
+        for (std::size_t i = 0; i < threads; ++i) {
+            workers_.emplace_back([this](std::stop_token st) { workerLoop(st); });
+        }
+    }
+
+    ThreadPool(const ThreadPool&) = delete;
+    ThreadPool& operator=(const ThreadPool&) = delete;
+
+    // Enqueue a task, blocking while `capacity_` tasks are already in flight.
+    void submit(std::function<void()> task) {
+        std::unique_lock lock(mutex_);
+        slotFree_.wait(lock, [this] { return inFlight_ < capacity_; });
+        tasks_.push(std::move(task));
+        ++inFlight_;
+        workAvailable_.notify_one();
+    }
+
+    // Block until no task is queued or executing. Never throws.
+    void wait() {
+        std::unique_lock lock(mutex_);
+        idle_.wait(lock, [this] { return inFlight_ == 0; });
+    }
+
+    // Returns and clears the first exception captured from a task, if any.
+    std::exception_ptr takeError() {
+        std::scoped_lock lock(mutex_);
+        return std::exchange(firstError_, nullptr);
+    }
+
+private:
+    void workerLoop(std::stop_token st) {
+        for (;;) {
+            std::function<void()> task;
+            {
+                std::unique_lock lock(mutex_);
+                workAvailable_.wait(lock, st, [this] { return !tasks_.empty(); });
+                if (tasks_.empty()) {
+                    return;  // woken by a stop request with nothing left to do
+                }
+                task = std::move(tasks_.front());
+                tasks_.pop();
+            }
+
+            std::exception_ptr err;
+            try {
+                task();
+            } catch (...) {
+                err = std::current_exception();
+            }
+
+            {
+                std::scoped_lock lock(mutex_);
+                if (err && !firstError_) {
+                    firstError_ = err;
+                }
+                --inFlight_;
+                if (inFlight_ == 0) {
+                    idle_.notify_all();
+                }
+            }
+            slotFree_.notify_one();
+        }
+    }
+
+    std::mutex mutex_;
+    std::condition_variable_any workAvailable_;  // workers wait here (stop-aware)
+    std::condition_variable slotFree_;           // submit() waits here
+    std::condition_variable idle_;               // wait() waits here
+    std::queue<std::function<void()>> tasks_;
+    std::size_t inFlight_ = 0;
+    std::size_t capacity_;
+    std::exception_ptr firstError_;
+
+    // Declared last so the jthreads stop and join before the synchronisation
+    // members they touch above are destroyed.
+    std::vector<std::jthread> workers_;
+};
+
+#endif  // UTILITIES_THREAD_POOL_HPP

scripts/build.sh

@@ -27,7 +27,7 @@ fi
 
 # 1. Generate build files (Passing your CXX flags directly to CMake instead of configure)
 cmake .. \
-  -DCMAKE_CXX_STANDARD=20 \
+  -DCMAKE_CXX_STANDARD=23 \
   -DCMAKE_BUILD_TYPE=Release \
   -DSKIP_BUILD_TEST=ON \
   -DCMAKE_EXPORT_COMPILE_COMMANDS=ON

scripts/build_dep.sh

@@ -26,7 +26,7 @@ SDK_PATH="$(xcrun --show-sdk-path 2>/dev/null || true)"
 build_dep() {
     local dep_dir="$1"
     local -a cmake_args=(
-        -DCMAKE_CXX_STANDARD=20
+        -DCMAKE_CXX_STANDARD=23
         -DCMAKE_BUILD_TYPE=Release
         -DSKIP_BUILD_TEST=ON
         -DCMAKE_CXX_FLAGS="-w"

source/commands/loadCommand.cpp

@@ -8,7 +8,8 @@
 
 #include <array>
 #include <charconv>
-#include <iostream>
+#include <print>
+#include <ranges>
 #include <stdexcept>
 #include <string>
 #include <system_error>
@@ -26,7 +27,8 @@
 #include "parameterSweep.hpp"
 #include "queueKeys.hpp"
 #include "redisLoader.hpp"
-#include "trading_definitions.hpp"
+#include "run_configuration.hpp"
+#include "strategy.hpp"
 
 namespace {
 
@@ -60,8 +62,8 @@ sweep::ParameterGenerator buildRandomStrategySweep() {
     sweep::ParameterGenerator generator;
     // generator.addRange("OHLC_COUNT", 80, 20, 140);  // 80, 100, 120, 140
     // generator.addList("OHLC_MINUTES", {1, 3, 5, 8});
-    generator.addList("STOP_DISTANCE_IN_PIPS", {1.0, 1.5, 2.0});
-    generator.addList("LIMIT_DISTANCE_IN_PIPS", {1.0, 1.5, 2.0});
+    generator.addList("STOP_DISTANCE_IN_PIPS", {1.0, 1.5, 10.0});
+    generator.addList("LIMIT_DISTANCE_IN_PIPS", {1.0, 1.5, 10.0});
     return generator;
 }
 
@@ -70,8 +72,8 @@ sweep::ParameterGenerator buildRandomStrategySweep() {
 RunConfiguration makeRunConfiguration(const std::string& runId) {
     return RunConfiguration{
         .RUN_ID = runId,
-        .SYMBOLS = "EURUSD,AUDUSD",
-        .LAST_MONTHS = 2,
+        .SYMBOLS = "EURUSD",
+        .LAST_MONTHS = 6,
     };
 }
 
@@ -82,7 +84,9 @@ Strategy makeStrategy(const sweep::Combination& combo) {
     using namespace trading_definitions;
 
     return Strategy{
-        .UUID = "",
+        // Each parameter combination gets its own UUID so a single backtest
+        // result is uniquely identifiable and traceable back to its inputs.
+        .UUID = boost::uuids::to_string(boost::uuids::random_generator()()),
         .TRADING_VARIABLES = TradingVariables{
             .STRATEGY = "RandomStrategy",
             .STOP_DISTANCE_IN_PIPS = toDecimal(combo.get("STOP_DISTANCE_IN_PIPS")),
@@ -108,36 +112,40 @@ Strategy makeStrategy(const sweep::Combination& combo) {
 int LoadCommand::run() {
     // One RUN_ID identifies the whole sweep; each combination becomes its own
     // queue entry, distinguished by its parameter values.
-    const std::string runId = boost::uuids::to_string(boost::uuids::random_generator()());
-    const std::string redisHost = env::getOr("REDIS_HOST", "127.0.0.1");
+    const auto runId = boost::uuids::to_string(boost::uuids::random_generator()());
+    const auto redisHost = env::getOr("REDIS_HOST", "127.0.0.1");
 
     // Build random here
-    const sweep::ParameterGenerator generator = buildRandomStrategySweep();
+    const auto generator = buildRandomStrategySweep();
 
-    const std::vector<sweep::Combination> combinations = generator.generateAllCombinations();
+    const auto combinations = generator.generateAllCombinations();
 
-    std::cout << "LoadCommand: sweeping " << combinations.size()
-              << " parameter combination(s) for RUN_ID=" << runId << std::endl;
+    std::println("LoadCommand: sweeping {} parameter combination(s) for RUN_ID={}",
+                 combinations.size(), runId);
 
-    // Serialise every swept strategy first.
-    std::vector<std::string> strategyPayloads;
-    strategyPayloads.reserve(combinations.size());
-    for (const sweep::Combination& combo : combinations) {
-        const nlohmann::json j = makeStrategy(combo);
-        strategyPayloads.push_back(j.dump());
-    }
+    // Serialise every swept strategy first. combinations is a sized range, so
+    // std::ranges::to reserves up front (no manual reserve needed). The json type
+    // is pinned explicitly because makeStrategy returns a Strategy and relies on
+    // the implicit conversion for .dump().
+    const auto strategyPayloads =
+        combinations | std::views::transform([](const sweep::Combination& combo) {
+            const nlohmann::json j = makeStrategy(combo);
+            return j.dump();
+        }) | std::ranges::to<std::vector<std::string>>();
 
     // Push all strategies BEFORE the run descriptor. A worker that sees the run
     // immediately drains the strategy list and retires the run when empty, so the
     // full set must already be present the moment the run becomes visible.
-    const std::string strategyKey = queue_keys::strategyKey(runId);
-    const int strategyStatus = RedisLoader::loadPayloadBatch(
+    const auto strategyKey = queue_keys::strategyKey(runId);
+    const auto strategyStatus = RedisLoader::loadPayloadBatch(
         redisHost, 6379, strategyKey, strategyPayloads);
     if (strategyStatus != 0) {
         return strategyStatus;
     }
 
-    // Now advertise the run so workers can pick it up.
+    // Now advertise the run so workers can pick it up. runJson is pinned to
+    // nlohmann::json (not auto) because makeRunConfiguration returns a
+    // RunConfiguration and relies on the implicit conversion for .dump().
     const nlohmann::json runJson = makeRunConfiguration(runId);
     return RedisLoader::loadPayload(redisHost, 6379, queue_keys::RUN,
                                     runJson.dump());

source/commands/runCommand.cpp

@@ -6,33 +6,26 @@
 
 #include "runCommand.hpp"
 
-#include <iostream>
 #include <string>
 
 #include "backtestRunner.hpp"
 #include "env.hpp"
 #include "jsonParser.hpp"
 #include "redisRunner.hpp"
 
-namespace {
-
-int runBacktestFromBase64(const std::string& questdbHost,
-                          const std::string& base64Config) {
-    auto config = JsonParser::parseConfigurationFromBase64(base64Config);
-    return runBacktest(questdbHost, config);
-}
-
-}  // namespace
-
 int RunCommand::run(int argc, const char* argv[]) {
     if (argc < 3) {
-        std::cerr << "Usage: BacktestingEngine run <questdb-host>\n"
-                  << "       BacktestingEngine run <questdb-host> <base64-config>"
-                  << std::endl;
+        // C++23 std::println is faster, safer, and cleaner than iostreams
+        std::println(stderr, "Usage: BacktestingEngine run <questdb-host>\n"
+                             "       BacktestingEngine run <questdb-host> <base64-config>");
         return 1;
     }
+
     if (argc == 3) {
         return RedisRunner::run(argv[2], env::getOr("REDIS_HOST", "127.0.0.1"));
     }
-    return runBacktestFromBase64(argv[2], argv[3]);
-}
+
+    // Else we'll just parse one base64 blob
+    auto config = JsonParser::parseConfigurationFromBase64(argv[3]);
+    return runBacktest(argv[2], config);
+}