Refactoring to put a strategy into redis, and then a queue of runs to sweep

mccaffers · mccaffers · commit 3a01f41b3419 · 2026-06-07T09:23:06.000+01:00
diff --git a/.gitignore b/.gitignore
@@ -42,4 +42,4 @@ TestResult.xcresult/
 sonarqube-generic-coverage.xml
 .cache/clangd/
 dump.rdb
-.infisical.json
+.infisical.json
diff --git a/backtesting-engine-cpp.xcodeproj/project.pbxproj b/backtesting-engine-cpp.xcodeproj/project.pbxproj
@@ -119,6 +119,8 @@
 		944D0DCF2C8C3704004DD0FC /* sonar-project.properties */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = text; path = "sonar-project.properties"; sourceTree = "<group>"; };
 		944D0DD02C8C3704004DD0FC /* README.md */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = net.daringfireball.markdown; path = README.md; sourceTree = "<group>"; };
 		944D0DD32C8C3704004DD0FC /* CMakeLists.txt */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = text; path = CMakeLists.txt; sourceTree = "<group>"; };
+		945F475C2FD5607E00D19164 /* queueKeys.hpp */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.h; path = queueKeys.hpp; sourceTree = "<group>"; };
+		945F475D2FD5614600D19164 /* run_configuration.hpp */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.h; path = run_configuration.hpp; sourceTree = "<group>"; };
 		9464E5EF2FA7466900D82BAD /* symbolScale.hpp */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.h; path = symbolScale.hpp; sourceTree = "<group>"; };
 		9464E5F02FA7467200D82BAD /* symbolScale.mm */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.objcpp; path = symbolScale.mm; sourceTree = "<group>"; };
 		94674B822D533B1D00973137 /* trade.hpp */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.h; path = trade.hpp; sourceTree = "<group>"; };
@@ -1333,6 +1335,7 @@
 		941B548F2D3BBA3B00E3BF64 /* trading_definitions */ = {
 			isa = PBXGroup;
 			children = (
+				945F475D2FD5614600D19164 /* run_configuration.hpp */,
 				941B54972D3BBAA200E3BF64 /* configuration.hpp */,
 				941B54942D3BBA8300E3BF64 /* strategy.hpp */,
 				941B54932D3BBA7300E3BF64 /* strategy_variables.hpp */,
@@ -1502,6 +1505,7 @@
 		94B8C7932D3D770800E17EB6 /* utilities */ = {
 			isa = PBXGroup;
 			children = (
+				945F475C2FD5607E00D19164 /* queueKeys.hpp */,
 				943770432FD4351100317424 /* parameterSweep.hpp */,
 				94829FC32FCC1D1200710E6E /* env.hpp */,
 				942EC55D2FBEF92F00CCBB5D /* redisConnection.hpp */,
diff --git a/include/backtestRunner.hpp b/include/backtestRunner.hpp
@@ -6,7 +6,20 @@
 
 #pragma once
 #include <string>
+#include <vector>
+#include "models/priceData.hpp"
 #include "trading_definitions/configuration.hpp"
 
+// Pulls all tick data for the run's symbols/window out of QuestDB. Expensive —
+// call once per run and reuse the result across that run's strategies.
+std::vector<PriceData> loadTicks(const std::string& questdbHost,
+                                 const std::string& symbolsCsv,
+                                 int lastMonths);
+
+// Runs one backtest against already-loaded ticks (no QuestDB access).
+void runBacktestOnTicks(const std::vector<PriceData>& ticks,
+                        const trading_definitions::Configuration& config);
+
+// Convenience for the direct path: loads ticks then runs a single backtest.
 int runBacktest(const std::string& questdbHost,
                 const trading_definitions::Configuration& config);
diff --git a/include/commands/loadCommand.hpp b/include/commands/loadCommand.hpp
@@ -6,8 +6,9 @@
 
 #pragma once
 
-// Backs the `load` subcommand: LPUSHes a strategy JSON (defined in
-// source/commands/loadCommand.cpp) onto the Redis `strategy_queue`.
+// Backs the `load` subcommand: for one RUN_ID, LPUSHes every swept strategy onto
+// BACKTESTING_QUEUE_STRATEGY:<RUN_ID>, then LPUSHes the run descriptor onto
+// BACKTESTING_QUEUE_RUN (see source/commands/loadCommand.cpp).
 class LoadCommand {
 public:
     static int run();
diff --git a/include/commands/runCommand.hpp b/include/commands/runCommand.hpp
@@ -6,8 +6,9 @@
 
 #pragma once
 
-// Backs the `run` subcommand: executes one strategy popped from the Redis
-// `strategy_queue`, or a Base64 config supplied directly on the command line.
+// Backs the `run` subcommand: drains BACKTESTING_QUEUE_RUN (loading each run's
+// QuestDB ticks once, then running its strategies), or runs a single Base64
+// Configuration supplied directly on the command line.
 class RunCommand {
 public:
     static int run(int argc, const char* argv[]);
diff --git a/include/jsonParser.hpp b/include/jsonParser.hpp
@@ -13,4 +13,6 @@
 class JsonParser {
 public:
     static trading_definitions::Configuration parseConfigurationFromBase64(const std::string& input);
+    static trading_definitions::RunConfiguration parseRunConfigurationFromBase64(const std::string& input);
+    static trading_definitions::Strategy parseStrategyFromBase64(const std::string& input);
 };
diff --git a/include/redisLoader.hpp b/include/redisLoader.hpp
@@ -7,17 +7,23 @@
 #pragma once
 
 #include <string>
+#include <vector>
 
 // LPUSH pairs with RedisRunner's RPOP so consumers observe FIFO ordering.
 class RedisLoader {
 public:
-    static int load(const std::string& rawJson,
-                    const std::string& redisHost = "127.0.0.1",
-                    int redisPort = 6379,
-                    const std::string& queueKey = "strategy_queue");
-
+    // LPUSHes a single Base64-encoded payload onto queueKey without assuming a
+    // payload type (run descriptor or strategy).
     static int loadPayload(const std::string& redisHost,
                            int redisPort,
                            const std::string& queueKey,
                            const std::string& rawJson);
+
+    // LPUSHes many payloads onto queueKey over one connection. Each payload is
+    // Base64-encoded; order is preserved (RedisRunner's RPOP then yields them
+    // in insertion order).
+    static int loadPayloadBatch(const std::string& redisHost,
+                                int redisPort,
+                                const std::string& queueKey,
+                                const std::vector<std::string>& rawJsonPayloads);
 };
diff --git a/include/redisRunner.hpp b/include/redisRunner.hpp
@@ -7,10 +7,14 @@
 #pragma once
 #include <string>
 
+// Worker entry point. Drains BACKTESTING_QUEUE_RUN: for each run it loads the
+// QuestDB tick data once, then drains that run's per-RUN_ID strategy list,
+// running every strategy against the cached ticks. Safe to launch many workers
+// concurrently — they peek the same run, load ticks once each, and compete on
+// RPOP of the shared strategy list.
 class RedisRunner {
 public:
     static int run(const std::string& questdbHost,
                    const std::string& redisHost = "127.0.0.1",
-                   int redisPort = 6379,
-                   const std::string& queueKey = "strategy_queue");
+                   int redisPort = 6379);
 };
diff --git a/include/trading_definitions.hpp b/include/trading_definitions.hpp
@@ -11,3 +11,4 @@
 #include "trading_definitions/strategy_variables.hpp"
 #include "trading_definitions/strategy.hpp"
 #include "trading_definitions/configuration.hpp"
+#include "trading_definitions/run_configuration.hpp"
diff --git a/include/trading_definitions/run_configuration.hpp b/include/trading_definitions/run_configuration.hpp
@@ -0,0 +1,27 @@
+// Backtesting Engine in C++
+//
+// (c) 2026 Ryan McCaffery | https://mccaffers.com
+// This code is licensed under MIT license (see LICENSE.txt for details)
+// ---------------------------------------
+
+#pragma once
+#include <string>
+#include <nlohmann/json.hpp>
+
+namespace trading_definitions {
+// Run-level descriptor: the unit of QuestDB tick data shared by every strategy
+// in a sweep. Carried on BACKTESTING_QUEUE_RUN and linked to its strategies via
+// RUN_ID (see queueKeys.hpp). The runner reassembles a full Configuration from
+// this plus each popped Strategy.
+struct RunConfiguration {
+  std::string RUN_ID;
+  std::string SYMBOLS;
+  int LAST_MONTHS;
+};
+NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE(RunConfiguration,
+                                   RUN_ID,
+                                   SYMBOLS,
+                                   LAST_MONTHS
+                                   );
+
+};
diff --git a/include/utilities/queueKeys.hpp b/include/utilities/queueKeys.hpp
@@ -0,0 +1,22 @@
+// Backtesting Engine in C++
+//
+// (c) 2026 Ryan McCaffery | https://mccaffers.com
+// This code is licensed under MIT license (see LICENSE.txt for details)
+// ---------------------------------------
+
+#pragma once
+
+#include <string>
+
+// Redis keys for the two-tier work queue. A sweep produces one run descriptor on
+// RUN and N strategy payloads on STRATEGY_PREFIX + RUN_ID, all linked by RUN_ID.
+namespace queue_keys {
+
+inline constexpr const char* RUN = "BACKTESTING_QUEUE_RUN";
+inline constexpr const char* STRATEGY_PREFIX = "BACKTESTING_QUEUE_STRATEGY:";
+
+inline std::string strategyKey(const std::string& runId) {
+    return std::string(STRATEGY_PREFIX) + runId;
+}
+
+}  // namespace queue_keys
diff --git a/source/backtestRunner.cpp b/source/backtestRunner.cpp
@@ -15,24 +15,37 @@
 #include "operations.hpp"
 #include "sqlManager.hpp"
 
-int runBacktest(const std::string& questdbHost,
-                const trading_definitions::Configuration& config) {
+std::vector<PriceData> loadTicks(const std::string& questdbHost,
+                                 const std::string& symbolsCsv,
+                                 int lastMonths) {
 
   DatabaseConnection db(questdbHost, 8812, "qdb", "admin", "quest");
 
   // Get a list of symbols
   std::vector<std::string> symbols;
-  std::istringstream ss(config.SYMBOLS);
+  std::istringstream ss(symbolsCsv);
   for (std::string token; std::getline(ss, token, ',');) {
     symbols.push_back(token);
   }
 
   // Get all the tick data out of QuestDB for these symbols
   std::vector<PriceData> ticks =
-      SqlManager::streamPriceData(db, symbols, config.LAST_MONTHS);
+      SqlManager::streamPriceData(db, symbols, lastMonths);
 
   printf("Total ticks streamed: %zu\n", ticks.size());
 
+  return ticks;
+}
+
+void runBacktestOnTicks(const std::vector<PriceData>& ticks,
+                        const trading_definitions::Configuration& config) {
   Operations::run(ticks, config);
+}
+
+int runBacktest(const std::string& questdbHost,
+                const trading_definitions::Configuration& config) {
+  const std::vector<PriceData> ticks =
+      loadTicks(questdbHost, config.SYMBOLS, config.LAST_MONTHS);
+  runBacktestOnTicks(ticks, config);
   return 0;
 }
diff --git a/source/commands/loadCommand.cpp b/source/commands/loadCommand.cpp
@@ -24,12 +24,14 @@
 
 #include "env.hpp"
 #include "parameterSweep.hpp"
+#include "queueKeys.hpp"
 #include "redisLoader.hpp"
 #include "trading_definitions.hpp"
 
 namespace {
 
-using trading_definitions::Configuration;
+using trading_definitions::RunConfiguration;
+using trading_definitions::Strategy;
 
 // Converts a swept double into a decimal64_t via its shortest round-trip decimal
 // string. Routing through text (rather than constructing from the binary double)
@@ -56,42 +58,48 @@ boost::decimal::decimal64_t toDecimal(double value) {
 // edit: register it here, then read it back in makeConfiguration().
 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.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});
     return generator;
 }
 
-// Maps one point in the parameter grid onto a fully-formed Configuration. Fields
-// not being swept keep their fixed defaults; swept fields are pulled from `combo`.
-Configuration makeConfiguration(const std::string& runId,
-                                const sweep::Combination& combo) {
-    using namespace boost::decimal::literals;
-    using namespace trading_definitions;
-
-    return Configuration{
+// The run-level descriptor: what tick data to pull from QuestDB. Shared by every
+// strategy in this sweep and linked to them by RUN_ID.
+RunConfiguration makeRunConfiguration(const std::string& runId) {
+    return RunConfiguration{
         .RUN_ID = runId,
         .SYMBOLS = "EURUSD,AUDUSD",
         .LAST_MONTHS = 2,
-        .STRATEGY = Strategy{
-            .UUID = "",
-            .TRADING_VARIABLES = TradingVariables{
-                .STRATEGY = "RandomStrategy",
-                .STOP_DISTANCE_IN_PIPS = toDecimal(combo.get("STOP_DISTANCE_IN_PIPS")),
-                .LIMIT_DISTANCE_IN_PIPS = toDecimal(combo.get("LIMIT_DISTANCE_IN_PIPS")),
-                .TRADING_SIZE = 1_DD,
-            },
-            .OHLC_VARIABLES = {
-                OHLCVariables{
-                    .OHLC_COUNT = combo.getInt("OHLC_COUNT"),
-                    .OHLC_MINUTES = combo.getInt("OHLC_MINUTES"),
-                },
-            },
-            .STRATEGY_VARIABLES = StrategyVariables{
-                .OHLC_RSI_VARIABLES = OHLCRSIVariables{.RSI_LONG = 60, .RSI_SHORT = 40},
+    };
+}
+
+// Maps one point in the parameter grid onto a Strategy. Fields not being swept
+// keep their fixed defaults; swept fields are pulled from `combo`.
+Strategy makeStrategy(const sweep::Combination& combo) {
+    using namespace boost::decimal::literals;
+    using namespace trading_definitions;
+
+    return Strategy{
+        .UUID = "",
+        .TRADING_VARIABLES = TradingVariables{
+            .STRATEGY = "RandomStrategy",
+            .STOP_DISTANCE_IN_PIPS = toDecimal(combo.get("STOP_DISTANCE_IN_PIPS")),
+            .LIMIT_DISTANCE_IN_PIPS = toDecimal(combo.get("LIMIT_DISTANCE_IN_PIPS")),
+            .TRADING_SIZE = 1_DD,
+        },
+        .OHLC_VARIABLES = {
+            OHLCVariables{
+                // Only read OHLC params when the sweep actually registers them;
+                // they default to 0 otherwise (see buildRandomStrategySweep).
+                .OHLC_COUNT = combo.has("OHLC_COUNT") ? combo.getInt("OHLC_COUNT") : 0,
+                .OHLC_MINUTES = combo.has("OHLC_MINUTES") ? combo.getInt("OHLC_MINUTES") : 0,
             },
         },
+        .STRATEGY_VARIABLES = StrategyVariables{
+            .OHLC_RSI_VARIABLES = OHLCRSIVariables{.RSI_LONG = 60, .RSI_SHORT = 40},
+        },
     };
 }
 
@@ -103,20 +111,34 @@ int LoadCommand::run() {
     const std::string runId = boost::uuids::to_string(boost::uuids::random_generator()());
     const std::string redisHost = env::getOr("REDIS_HOST", "127.0.0.1");
 
+    // Build random here
     const sweep::ParameterGenerator generator = buildRandomStrategySweep();
+
     const std::vector<sweep::Combination> combinations = generator.generateAllCombinations();
 
     std::cout << "LoadCommand: sweeping " << combinations.size()
               << " parameter combination(s) for RUN_ID=" << runId << std::endl;
 
-    int status = 0;
+    // Serialise every swept strategy first.
+    std::vector<std::string> strategyPayloads;
+    strategyPayloads.reserve(combinations.size());
     for (const sweep::Combination& combo : combinations) {
-        const Configuration config = makeConfiguration(runId, combo);
-        const nlohmann::json j = config;
-        const int result = RedisLoader::load(j.dump(), redisHost);
-        if (result != 0) {
-            status = result;  // remember a failure but keep loading the rest
-        }
+        const nlohmann::json j = makeStrategy(combo);
+        strategyPayloads.push_back(j.dump());
+    }
+
+    // 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(
+        redisHost, 6379, strategyKey, strategyPayloads);
+    if (strategyStatus != 0) {
+        return strategyStatus;
     }
-    return status;
+
+    // Now advertise the run so workers can pick it up.
+    const nlohmann::json runJson = makeRunConfiguration(runId);
+    return RedisLoader::loadPayload(redisHost, 6379, queue_keys::RUN,
+                                    runJson.dump());
 }
diff --git a/source/redisLoader.cpp b/source/redisLoader.cpp
diff --git a/source/redisRunner.cpp b/source/redisRunner.cpp
diff --git a/source/utilities/jsonParser.cpp b/source/utilities/jsonParser.cpp
diff --git a/source/utilities/redisConnection.cpp b/source/utilities/redisConnection.cpp
