feat(hft): finalize dual-threaded engine & alignment verification

Author: quantDIY

benchmark_hft_engine.cpp

@@ -0,0 +1,92 @@
+
+#include "quanux/MarketTick.hpp"
+#include "quanux/SPSCQueue.hpp"
+#include <atomic>
+#include <chrono>
+#include <iostream>
+#include <thread>
+#include <vector>
+
+using namespace quanux;
+
+// Signal simplified for benchmark
+struct Signal {
+  uint64_t tick_ts;
+  uint32_t instrument_id;
+  double z_score;
+  double volatility;
+};
+
+int main() {
+  SPSCQueue<Signal> queue(4096);
+  std::atomic<bool> running{true};
+  std::vector<uint64_t> latencies;
+  latencies.reserve(1000000);
+
+  // Consumer Thread (Execution)
+  std::thread consumer([&]() {
+    Signal sig;
+    while (running) {
+      if (queue.pop(sig)) {
+        auto now = std::chrono::steady_clock::now();
+        auto tick_time = std::chrono::steady_clock::time_point(
+            std::chrono::nanoseconds(sig.tick_ts));
+        auto latency = std::chrono::duration_cast<std::chrono::nanoseconds>(
+                           now - tick_time)
+                           .count();
+        latencies.push_back(latency);
+      }
+    }
+  });
+
+  // Producer Loop (Simulation)
+  std::cout << "Running Micro-Benchmark (1M messages)..." << std::endl;
+  auto start = std::chrono::steady_clock::now();
+
+  for (int i = 0; i < 1000000; ++i) {
+    auto now = std::chrono::steady_clock::now();
+    uint64_t ts = std::chrono::duration_cast<std::chrono::nanoseconds>(
+                      now.time_since_epoch())
+                      .count();
+    Signal sig{ts, 1, 3.0, 0.5}; // Always trigger
+    while (!queue.push(sig)) {
+      // spin
+    }
+  }
+
+  // Wait for drain
+  std::this_thread::sleep_for(std::chrono::milliseconds(200));
+  running = false;
+  consumer.join();
+  auto end = std::chrono::steady_clock::now();
+
+  std::cout << "Done." << std::endl;
+
+  // Calculate Stats
+  uint64_t sum = 0;
+  uint64_t min_lat = -1;
+  uint64_t max_lat = 0;
+  for (auto lat : latencies) {
+    sum += lat;
+    if (lat < min_lat)
+      min_lat = lat;
+    if (lat > max_lat)
+      max_lat = lat;
+  }
+
+  if (!latencies.empty()) {
+    std::cout << "Avg Latency: " << (sum / latencies.size()) << " ns"
+              << std::endl;
+    std::cout << "Min Latency: " << min_lat << " ns" << std::endl;
+    std::cout << "Max Latency: " << max_lat << " ns" << std::endl;
+    std::cout << "Throughput: "
+              << (1000000.0 /
+                  std::chrono::duration_cast<std::chrono::milliseconds>(end -
+                                                                        start)
+                      .count() *
+                  1000)
+              << " msg/sec" << std::endl;
+  }
+
+  return 0;
+}

check_struct.cpp

@@ -0,0 +1,22 @@
+#include "quanux/MarketTick.hpp"
+#include <cstddef>
+#include <iostream>
+
+int main() {
+  std::cout << "Sizeof MarketTick: " << sizeof(quanux::MarketTick) << std::endl;
+  std::cout << "Alignof MarketTick: " << alignof(quanux::MarketTick)
+            << std::endl;
+  std::cout << "Offset internal_arrival_ts: "
+            << offsetof(quanux::MarketTick, internal_arrival_ts) << std::endl;
+  std::cout << "Offset _pad: " << offsetof(quanux::MarketTick, _pad)
+            << std::endl;
+
+  if (sizeof(quanux::MarketTick) == 64) {
+    std::cout << "SUCCESS: 64 bytes" << std::endl;
+    return 0;
+  } else {
+    std::cout << "FAILURE: " << sizeof(quanux::MarketTick) << " bytes"
+              << std::endl;
+    return 1;
+  }
+}

verify_hft_engine.py

@@ -0,0 +1,66 @@
+
+import asyncio
+import nats
+import struct
+import time
+import random
+
+# MarketTick Structure (64-byte aligned)
+# uint64_t local_rec_ts;   // 8
+# uint64_t exchange_ts;    // 8
+# double price;            // 8
+# uint32_t size;           // 4
+# uint32_t flags;          // 4
+# uint32_t instrument_id;  // 4
+# uint64_t internal_arrival_ts; // 8
+# uint64_t processing_start_ts; // 8
+# uint8_t _pad[8];         // 8 (Explicit) + 4 (Implicit after instrument_id)
+# Total: 64 bytes
+
+# Q=8, d=8, I=4. 
+# Layout: Q Q d I I I [4x implicit] Q Q [8x explicit]
+STRUCT_FMT = "QQdIII4xQQ8x"
+
+async def run():
+    nc = await nats.connect("nats://localhost:4222")
+    
+    print("Connected to NATS. Sending binary ticks to MARKET.BIN...")
+
+    instrument_id = 101 # Example ID for "AAPL"
+    price = 150.0
+
+    start_time = time.time()
+    
+    for i in range(100):
+        price += random.uniform(-0.05, 0.05)
+        now_ns = int(time.time_ns())
+        
+        # Pack the struct
+        # We leave internal timestamps 0, engine will fill them
+        payload = struct.pack(STRUCT_FMT,
+                              now_ns,       # local_rec_ts
+                              now_ns,       # exchange_ts
+                              price,        # price
+                              100,          # size
+                              0,            # flags
+                              instrument_id,# instrument_id
+                              0,            # internal_arrival_ts (placeholder)
+                              0             # processing_start_ts (placeholder)
+                              )
+        
+        await nc.publish("MARKET.BIN", payload)
+        
+        if i % 10 == 0:
+            print(f"Sent tick {i}: Price {price:.2f}")
+            
+        # Burst simulation
+        # await asyncio.sleep(0.0001) 
+
+    # Determine throughput
+    elapsed = time.time() - start_time
+    print(f"Sent 100 ticks in {elapsed:.4f}s")
+    
+    await nc.close()
+
+if __name__ == '__main__':
+    asyncio.run(run())