Spaces:

Ypeng12
/

quant-ai

Running

File size: 11,257 Bytes

f86fb3b

#include <iostream>
#include <vector>
#include <algorithm>
#include <numeric>
#include <chrono>
#include <iomanip>
#include <thread>
#include "../include/core/types.hpp"
#include "../include/core/time_utils.hpp"
#include "../include/core/spsc_queue.hpp"
#include "../include/core/mpsc_queue.hpp"
#include "../include/core/object_pool.hpp"
#include "../include/protocol/binary_protocol.hpp"
#include "../include/order_book/matching_engine.hpp"
#include "../include/risk/risk_engine.hpp"
#include "../include/engine/trading_engine.hpp"

struct LatencyStats {
    double min_ns{0};
    double p50_ns{0};
    double p90_ns{0};
    double p95_ns{0};
    double p99_ns{0};
    double p99_9_ns{0};
    double max_ns{0};
    double avg_ns{0};
    double throughput_mps{0}; // Million packets per second
};

LatencyStats compute_stats(std::vector<uint64_t>& latencies, double total_time_sec) {
    LatencyStats stats;
    if (latencies.empty()) return stats;

    std::sort(latencies.begin(), latencies.end());
    size_t n = latencies.size();

    stats.min_ns = latencies.front();
    stats.p50_ns = latencies[n * 50 / 100];
    stats.p90_ns = latencies[n * 90 / 100];
    stats.p95_ns = latencies[n * 95 / 100];
    stats.p99_ns = latencies[n * 99 / 100];
    stats.p99_9_ns = latencies[n * 999 / 1000];
    stats.max_ns = latencies.back();

    uint64_t sum = std::accumulate(latencies.begin(), latencies.end(), 0ULL);
    stats.avg_ns = static_cast<double>(sum) / n;
    stats.throughput_mps = (static_cast<double>(n) / 1'000'000.0) / total_time_sec;
    return stats;
}

void print_stats_row(const std::string& name, const LatencyStats& s) {
    std::cout << "| " << std::left << std::setw(32) << name
              << " | " << std::right << std::setw(8) << std::fixed << std::setprecision(2) << s.throughput_mps << " M/s"
              << " | " << std::setw(7) << std::setprecision(1) << s.p50_ns << " ns"
              << " | " << std::setw(7) << std::setprecision(1) << s.p95_ns << " ns"
              << " | " << std::setw(7) << std::setprecision(1) << s.p99_ns << " ns"
              << " | " << std::setw(7) << std::setprecision(1) << s.p99_9_ns << " ns"
              << " |\n";
}

// 1. SPSC Ring Buffer Queue Benchmark across depths
template <size_t Depth>
LatencyStats benchmark_spsc_queue(const std::string& label, size_t iterations) {
    (void)label;
    quant::core::LockFreeSPSCQueue<uint64_t, Depth> queue;
    std::vector<uint64_t> latencies;
    latencies.reserve(iterations);

    std::atomic<bool> start_flag{false};
    std::atomic<bool> done_flag{false};

    std::thread consumer([&]() {
        while (!start_flag.load(std::memory_order_acquire)) {}
        size_t received = 0;
        while (received < iterations) {
            uint64_t val;
            if (queue.pop(val)) {
                uint64_t now = quant::core::TimeUtils::now_ns();
                if (now >= val) {
                    latencies.push_back(now - val);
                }
                ++received;
            } else {
                quant::core::ThreadUtils::cpu_pause();
            }
        }
        done_flag.store(true, std::memory_order_release);
    });

    start_flag.store(true, std::memory_order_release);
    auto t0 = std::chrono::high_resolution_clock::now();

    for (size_t i = 0; i < iterations; ++i) {
        uint64_t ts = quant::core::TimeUtils::now_ns();
        while (!queue.push(ts)) {
            quant::core::ThreadUtils::cpu_pause();
        }
    }

    consumer.join();
    auto t1 = std::chrono::high_resolution_clock::now();
    double sec = std::chrono::duration<double>(t1 - t0).count();
    return compute_stats(latencies, sec);
}

// 2. MPSC Queue Concurrent Producer Benchmark (1, 2, 4 threads)
LatencyStats benchmark_mpsc_queue(size_t num_producers, size_t total_iterations) {
    quant::core::LockFreeMPSCQueue<uint64_t, 65536> queue;
    std::vector<uint64_t> latencies;
    latencies.reserve(total_iterations);

    size_t iters_per_prod = total_iterations / num_producers;
    std::atomic<bool> start_flag{false};

    std::thread consumer([&]() {
        while (!start_flag.load(std::memory_order_acquire)) {}
        size_t received = 0;
        while (received < total_iterations) {
            uint64_t val;
            if (queue.pop(val)) {
                uint64_t now = quant::core::TimeUtils::now_ns();
                if (now >= val) {
                    latencies.push_back(now - val);
                }
                ++received;
            } else {
                quant::core::ThreadUtils::cpu_pause();
            }
        }
    });

    std::vector<std::thread> producers;
    producers.reserve(num_producers);
    for (size_t p = 0; p < num_producers; ++p) {
        producers.emplace_back([&]() {
            while (!start_flag.load(std::memory_order_acquire)) {}
            for (size_t i = 0; i < iters_per_prod; ++i) {
                uint64_t ts = quant::core::TimeUtils::now_ns();
                while (!queue.push(ts)) {
                    quant::core::ThreadUtils::cpu_pause();
                }
            }
        });
    }

    auto t0 = std::chrono::high_resolution_clock::now();
    start_flag.store(true, std::memory_order_release);

    for (auto& th : producers) th.join();
    consumer.join();

    auto t1 = std::chrono::high_resolution_clock::now();
    double sec = std::chrono::duration<double>(t1 - t0).count();
    return compute_stats(latencies, sec);
}

// 3. Binary Protocol Serialization/Deserialization Benchmark
LatencyStats benchmark_binary_protocol(size_t iterations) {
    std::vector<uint64_t> latencies;
    latencies.reserve(iterations);

    std::array<uint8_t, quant::protocol::MAX_FRAME_SIZE> buffer;
    quant::protocol::NewOrderSinglePayload order{
        1, 1001, quant::core::Symbol("TSLA"),
        static_cast<uint8_t>(quant::core::Side::BUY),
        static_cast<uint8_t>(quant::core::OrderType::LIMIT),
        2500000, 100
    };

    auto t0 = std::chrono::high_resolution_clock::now();
    for (size_t i = 0; i < iterations; ++i) {
        uint64_t s = quant::core::TimeUtils::now_ns();
        size_t len = quant::protocol::BinaryProtocol::serialize_message(
            std::span<uint8_t>(buffer),
            quant::protocol::MessageType::NEW_ORDER_SINGLE,
            i + 1,
            s,
            order
        );
        auto* parsed = quant::protocol::BinaryProtocol::parse_payload<quant::protocol::NewOrderSinglePayload>(
            std::span<const uint8_t>(buffer.data(), len)
        );
        (void)parsed;
        uint64_t e = quant::core::TimeUtils::now_ns();
        latencies.push_back(e - s);
    }
    auto t1 = std::chrono::high_resolution_clock::now();
    double sec = std::chrono::duration<double>(t1 - t0).count();
    return compute_stats(latencies, sec);
}

// 4. Order Book & Matching Engine Core Benchmark
LatencyStats benchmark_matching_engine(size_t iterations) {
    quant::order_book::MatchingEngine engine(quant::core::Symbol("AAPL"));
    std::vector<uint64_t> latencies;
    latencies.reserve(iterations);

    // Pre-populate resting asks
    for (size_t i = 0; i < 50; ++i) {
        engine.process_new_order(
            1, i + 1, quant::core::Side::SELL, quant::core::OrderType::LIMIT,
            quant::core::Price::from_raw(1500000 + i * 100), 1000,
            quant::core::TimeUtils::now_ns()
        );
    }

    auto t0 = std::chrono::high_resolution_clock::now();
    for (size_t i = 0; i < iterations; ++i) {
        uint64_t s = quant::core::TimeUtils::now_ns();
        // Alternating matching buy orders
        engine.process_new_order(
            2, 100000 + i, quant::core::Side::BUY, quant::core::OrderType::LIMIT,
            quant::core::Price::from_raw(1500000 + (i % 50) * 100), 10, s
        );
        uint64_t e = quant::core::TimeUtils::now_ns();
        latencies.push_back(e - s);
    }
    auto t1 = std::chrono::high_resolution_clock::now();
    double sec = std::chrono::duration<double>(t1 - t0).count();
    return compute_stats(latencies, sec);
}

// 5. Burst Order Traffic Benchmark (100,000 orders in tight loop)
LatencyStats benchmark_burst_traffic(size_t iterations) {
    quant::order_book::MatchingEngine engine(quant::core::Symbol("NVDA"));
    std::vector<uint64_t> latencies;
    latencies.reserve(iterations);

    auto t0 = std::chrono::high_resolution_clock::now();
    for (size_t i = 0; i < iterations; ++i) {
        uint64_t s = quant::core::TimeUtils::now_ns();
        engine.process_new_order(
            1, i + 1,
            (i % 2 == 0) ? quant::core::Side::BUY : quant::core::Side::SELL,
            quant::core::OrderType::LIMIT,
            quant::core::Price::from_raw(1000000 + (i % 20) * 100),
            100,
            s
        );
        uint64_t e = quant::core::TimeUtils::now_ns();
        latencies.push_back(e - s);
    }
    auto t1 = std::chrono::high_resolution_clock::now();
    double sec = std::chrono::duration<double>(t1 - t0).count();
    return compute_stats(latencies, sec);
}

int main() {
    std::cout << "========================================================================================================\n";
    std::cout << "⚡ QUANT.AI MODERN C++20 PRODUCTION TRADING ENGINE BENCHMARK SUITE\n";
    std::cout << "========================================================================================================\n";
    std::cout << "| Benchmark Scenario               | Throughput |    p50    |    p95    |    p99    |   p99.9   |\n";
    std::cout << "|----------------------------------|------------|-----------|-----------|-----------|-----------|\n";

    constexpr size_t OPS = 500000;

    // SPSC Queue across depths
    auto spsc_256 = benchmark_spsc_queue<256>("Lock-Free SPSC (Depth 256)", OPS);
    print_stats_row("Lock-Free SPSC (Depth 256)", spsc_256);

    auto spsc_1024 = benchmark_spsc_queue<1024>("Lock-Free SPSC (Depth 1024)", OPS);
    print_stats_row("Lock-Free SPSC (Depth 1024)", spsc_1024);

    auto spsc_64k = benchmark_spsc_queue<65536>("Lock-Free SPSC (Depth 65536)", OPS);
    print_stats_row("Lock-Free SPSC (Depth 65536)", spsc_64k);

    // MPSC Queue concurrency: 1, 2, 4 threads
    auto mpsc_1 = benchmark_mpsc_queue(1, OPS);
    print_stats_row("Lock-Free MPSC (1 Thread)", mpsc_1);

    auto mpsc_2 = benchmark_mpsc_queue(2, OPS);
    print_stats_row("Lock-Free MPSC (2 Threads)", mpsc_2);

    auto mpsc_4 = benchmark_mpsc_queue(4, OPS);
    print_stats_row("Lock-Free MPSC (4 Threads)", mpsc_4);

    // Binary protocol zero-copy framing
    auto bin_stat = benchmark_binary_protocol(OPS);
    print_stats_row("Binary Protocol (Zero-Copy)", bin_stat);

    // Matching Engine price-time priority matching
    auto match_stat = benchmark_matching_engine(OPS);
    print_stats_row("Matching Engine (Order Match)", match_stat);

    // Burst traffic (100k+ orders)
    auto burst_stat = benchmark_burst_traffic(100000);
    print_stats_row("Burst Traffic (100k Orders)", burst_stat);

    std::cout << "========================================================================================================\n";
    std::cout << "✅ Benchmark completed. Sustained throughput & low latency confirmed across all concurrency pipelines.\n";
    return 0;
}