Real-Time Cryptocurrency Trade Correlation Engine: A High-Performance C++ Implementation

// Simple but inadequate approach
while (true) {
    for (auto& symbol : symbols) {
        trades = fetch_trades(symbol);              // ~50ms network call
        correlations = find_correlations(trades);   // ~1ms CPU work
        for (auto& corr : correlations) {
            result = compute_metric(corr);          // ~5ms mandatory delay
            write_to_file(result);                  // ~1ms I/O
        }
    }
}
// Total latency per cycle: 50ms + 1ms + (N × 6ms) - unacceptably high

// O(n²) comparison of all trades
for (auto& trade1 : all_trades) {
    for (auto& trade2 : all_trades) {
        if (abs(trade1.timestamp - trade2.timestamp) <= 5ms) {
            // Found correlation
        }
    }
}

std::unordered_set<std::string> already_emitted;
std::string trade_key = symbol + ":" + std::to_string(trade_id);

struct Trade {
    uint64_t trade_id{};                    // Coinbase trade ID
    std::string symbol, time_str, side, size, price;
    int64_t timestamp_us{};                 // Microsecond precision
    std::chrono::steady_clock::time_point received_at{
        std::chrono::steady_clock::now()};  // For latency tracking
};

struct TradeOptimized {
    uint64_t trade_id;
    uint32_t symbol_id;     // Enum: BTC=1, ETH=2, SOL=3
    bool is_buy;            // Instead of "buy"/"sell" string
    double size, price;     // Native numbers
    int64_t timestamp_us;
};

struct Config {
    size_t trade_queue_size = 25000;        // High-throughput buffer
    size_t filter_queue_size = 15000;       // Correlated trades only
    size_t write_queue_size = 15000;        // Output buffering
    int poll_interval_ms = 100;             // Coinbase rate limit friendly
    int64_t correlation_window_us = 5000;   // 5ms specification
    std::vector<std::string> symbols = {"BTC-USD", "ETH-USD", "SOL-USD"};
    std::string output_file = "trade_metrics.txt";
    size_t mmap_initial_size_mb = 50;       // Memory-mapped file size
    int max_consecutive_failures = 10;      // Circuit breaker threshold
    
    void load_from_env();  // Runtime configuration override
};

template <typename T>
class SmartBoundedQueue {
public:
    enum DropStrategy {
        DROP_OLDEST,    // Trade queue: preserve recent data
        DROP_NEWEST,    // Filter queue: preserve early correlations  
        BACKPRESSURE,   // Write queue: maintain ordering
        DROP_BY_PRIORITY
    };

// Scenario: API burst delivers 1000 trades in 1 second, but queue only holds 100
// Option 1: DROP_OLDEST - keep trades 901-1000, lose 1-900
// Option 2: DROP_NEWEST - keep trades 1-100, lose 901-1000

// Why DROP_OLDEST wins:
// - Recent trades more likely to correlate with incoming trades
// - Old trades already had their correlation opportunity
// - Temporal locality principle applies to market data

// Scenario: Correlation engine overwhelms worker pool
// Option 1: DROP_NEWEST - preserve older correlations, lose newer ones
// Option 2: DROP_OLDEST - preserve newer correlations, lose older ones

// Why DROP_NEWEST wins:
// - Maintaining temporal order is critical for output correctness
// - Older correlations were detected first, should be processed first
// - Worker shortage is temporary; we can detect new correlations later

// Scenario: Writer can't keep up with worker output
// Option 1: Drop computed results (data loss)
// Option 2: Apply backpressure (slow down workers)

// Why BACKPRESSURE wins:
// - Computed results represent expensive work (5ms per trade)
// - Losing computed metrics defeats the system's purpose
// - Better to slow the pipeline than lose valuable output

bool push(const T &value, int priority = 0) {
    std::unique_lock<std::mutex> lk(m_);  // Critical: one mutex per queue
    ++total_pushed_;

    if (q_.size() >= max_size_) {
        switch (strategy_) {
        case DROP_OLDEST:
            q_.pop();                    // O(1) operation
            ++dropped_count_;
            g_metrics.increment_queue_overflows();
            break;
        case BACKPRESSURE:
            if (!cv_space_.wait_for(lk, std::chrono::milliseconds(100), 
                [&] { return q_.size() < max_size_ || shutdown_; })) {
                return false;  // Timeout prevents deadlock
            }
            break;
        }
    }
    
    q_.push(value);
    cv_.notify_one();  // Wake one waiting consumer
    return true;
}

// Check every trade against every other trade
for (auto& trade1 : all_trades) {
    for (auto& trade2 : all_trades) {
        if (abs(trade1.timestamp - trade2.timestamp) <= 5000) {
            if (trade1.symbol != trade2.symbol) {
                found_correlation(trade1, trade2);
            }
        }
    }
}
// Complexity: O(n²) - unacceptable for high-frequency data

// Sort all trades by timestamp, use binary search
std::sort(all_trades.begin(), all_trades.end(), 
          [](const Trade& a, const Trade& b) { 
              return a.timestamp_us < b.timestamp_us; 
          });

for (auto& trade : all_trades) {
    auto range_start = std::lower_bound(/* trade.timestamp - 5000 */);
    auto range_end = std::upper_bound(/* trade.timestamp + 5000 */);
    // Check only trades in time window
}
// Complexity: O(n log n) for sort + O(log n) per search
// Problem: Expensive to maintain sorted order with continuous insertions

void merger() {
    std::unordered_map<std::string, std::deque<Trade>> windows;
    const int64_t W = g_config.correlation_window_us;  // 5000 μs
    
    while (!stop_all.load()) {
        Trade new_trade;
        if (!trade_q->pop(new_trade)) break;

        // 1. Clean expired trades - O(k) where k = expired trades
        int64_t cutoff = new_trade.timestamp_us - W;
        for (auto &[symbol, deque] : windows) {
            while (!deque.empty() && deque.front().timestamp_us < cutoff) {
                deque.pop_front();  // O(1) amortized
            }
        }

        // 2. Check correlations - O(m) where m = trades in window
        for (const auto &other_symbol : symbols) {
            if (other_symbol == new_trade.symbol) continue;
            
            for (const auto &existing_trade : windows[other_symbol]) {
                int64_t time_diff = std::abs(existing_trade.timestamp_us - 
                                           new_trade.timestamp_us);
                if (time_diff <= W) {
                    process_correlation(new_trade, existing_trade);
                }
            }
        }

        // 3. Add new trade to its window
        windows[new_trade.symbol].push_back(new_trade);
    }
}

static std::unordered_set<std::string> already_emitted;
static std::mutex emitted_mu;

// Generate unique identifier for each trade
std::string trade_key = trade.symbol + ":" + std::to_string(trade.trade_id);

{
    std::lock_guard lk(emitted_mu);
    if (already_emitted.count(trade_key)) {
        continue;  // Skip already processed trades
    }
    already_emitted.insert(trade_key);
}

// Traditional approach
void write_result(const std::string& data) {
    std::ofstream file("output.txt", std::ios::app);  // Open file
    file << data;                                     // Write data  
    file.flush();                                     // Force to disk
}  // Close file

// Performance characteristics:
// - open(): ~500μs (file system lookup, permission check)
// - write(): ~100μs (copy to kernel buffer)
// - flush(): ~5000μs (actual disk I/O, fsync)
// - close(): ~100μs (cleanup)
// Total: ~5700μs per write operation

class HighPerformanceWriter {
    int fd_{-1};
    char *mem_{nullptr};     // Memory-mapped region
    size_t sz_{0};           // Current file size
    std::atomic<size_t> off_{0};  // Current write offset
    
    void write_direct(const std::string& data) {
        size_t offset = off_.fetch_add(data.size());  // Atomic allocation
        if (offset + data.size() > sz_) {
            grow_file();  // Expand mapping if needed
        }
        std::memcpy(mem_ + offset, data.data(), data.size());  // Direct memory copy
        // No flush needed - OS handles dirty page writeback
    }
};

struct Entry {
    std::string data;
    uint64_t seq;  // Sequence number from correlation detection
};

static constexpr size_t RING = 4096;  // Power of 2 for efficient modulo
std::array<Entry, RING> ring_;
std::atomic<size_t> head_{0}, tail_{0};  // Lock-free head/tail pointers

// Producer side (workers)
bool write_async(std::string data, uint64_t seq) {
    size_t tail = tail_.load();
    size_t next_tail = (tail + 1) % RING;
    
    // Check if ring is full
    if (next_tail == head_.load(std::memory_order_acquire)) {
        return false;  // Backpressure signal
    }
    
    // Store data and sequence
    ring_[tail] = {std::move(data), seq};
    
    // Publish atomically
    tail_.store(next_tail, std::memory_order_release);
    return true;
}

// Consumer side (background thread)
void background_writer() {
    std::map<uint64_t, std::string> reorder_buffer;
    uint64_t next_expected_seq = 0;
    
    while (!stop_.load()) {
        // Collect from ring buffer
        size_t head = head_.load();
        size_t tail = tail_.load(std::memory_order_acquire);
        
        while (head != tail) {
            reorder_buffer.emplace(ring_[head].seq, std::move(ring_[head].data));
            head = (head + 1) % RING;
        }
        head_.store(head, std::memory_order_release);
        
        // Write in sequence order
        auto it = reorder_buffer.find(next_expected_seq);
        while (it != reorder_buffer.end()) {
            write_to_mmap(it->second);  // Actual memory-mapped write
            reorder_buffer.erase(it);
            ++next_expected_seq;
            it = reorder_buffer.find(next_expected_seq);
        }
        
        std::this_thread::sleep_for(std::chrono::microseconds(100));
    }
}

// Producer: memory_order_release on tail update
tail_.store(next_tail, std::memory_order_release);
// Ensures: All writes to ring_[tail] happen-before tail update becomes visible

// Consumer: memory_order_acquire on tail read  
size_t tail = tail_.load(std::memory_order_acquire);
// Ensures: Consumer sees all writes to ring_ that happened-before tail update

class AdaptiveSymbolManager {
    struct Metrics {
        std::atomic<uint64_t> trade_cnt{0};
        std::atomic<uint64_t> volume_usd{0};
        std::chrono::steady_clock::time_point last_trade{
            std::chrono::steady_clock::now()};
        double corr_freq{0.0};  // Exponential moving average
    };
    
    std::unordered_map<std::string, Metrics> metrics_;
    
public:
    void update_activity(const std::string& symbol, double volume) {
        auto& m = metrics_[symbol];
        ++m.trade_cnt;
        m.volume_usd += static_cast<uint64_t>(volume);
        m.last_trade = std::chrono::steady_clock::now();
    }
    
    void update_correlation(const std::string& symbol, bool found_correlation) {
        auto& m = metrics_[symbol];
        // Exponential moving average: recent events weighted more heavily
        m.corr_freq = 0.95 * m.corr_freq + 0.05 * (found_correlation ? 1.0 : 0.0);
    }
    
    std::vector<std::string> get_priority_order() {
        std::vector<std::pair<std::string, double>> scored_symbols;
        
        for (auto& [symbol, m] : metrics_) {
            // Recency factor: exponential decay based on time since last trade
            auto seconds_since_last = std::chrono::duration_cast<std::chrono::seconds>(
                std::chrono::steady_clock::now() - m.last_trade).count();
            double recency = std::exp(-seconds_since_last / 300.0);  // 5-minute half-life
            
            // Composite score: volume × correlation frequency × recency
            double score = (m.volume_usd / 1'000'000.0) * (1.0 + 10 * m.corr_freq) * recency;
            scored_symbols.emplace_back(symbol, score);
        }
        
        // Sort by score (highest first)
        std::sort(scored_symbols.begin(), scored_symbols.end(),
                  [](const auto& a, const auto& b) { return a.second > b.second; });
        
        std::vector<std::string> result;
        for (const auto& [symbol, _] : scored_symbols) {
            result.push_back(symbol);
        }
        return result;
    }
};

// In correlation engine, check symbols in priority order
auto symbol_priority = sym_mgr.get_priority_order();
for (const auto& other_symbol : symbol_priority) {
    if (other_symbol == new_trade.symbol) continue;
    
    // Check correlations with highest-priority symbols first
    for (const auto& existing_trade : windows[other_symbol]) {
        // ... correlation logic
    }
}

// Traditional file I/O approach (rejected)
void traditional_write(const std::string& data) {
    std::ofstream file("output.txt", std::ios::app);  // ~1ms syscall
    file << data;                                     // ~1ms write
    file.flush();                                     // ~5ms fsync
}  // Total: ~7ms per write

// Memory-mapped approach (chosen)
void mmap_write(const std::string& data) {
    std::memcpy(mem_ + offset_, data.data(), data.size());  // ~10μs
    offset_ += data.size();
    // Background thread handles periodic msync()
}  // Total: ~10μs per write

// Problem: Workers complete out-of-order, but file must maintain sequence
// Solution: Lock-free ring buffer with sequence numbering

struct Entry {
    std::string data;
    uint64_t seq;  // Global sequence number from merger
};

// Producer (workers) - lock-free insertion
bool write_async(std::string data, uint64_t seq) {
    size_t tail = tail_.load();
    size_t next = (tail + 1) % RING_SIZE;
    
    if (next == head_.load(std::memory_order_acquire)) {
        return false;  // Ring full
    }
    
    ring_[tail] = {std::move(data), seq};
    tail_.store(next, std::memory_order_release);  // Atomic publication
    return true;
}

// End-to-end latency tracking embedded in Trade structure
struct Trade {
    // ... trade data
    std::chrono::steady_clock::time_point received_at{
        std::chrono::steady_clock::now()};  // Timestamp on API reception
};

// Latency calculation at write completion
void worker() {
    auto end_processing = std::chrono::steady_clock::now();
    uint64_t e2e_latency = std::chrono::duration_cast<std::chrono::microseconds>(
        end_processing - trade.received_at).count();
    g_metrics.add_latency(e2e_latency);
}

// Design decision: Match worker threads to hardware concurrency
unsigned worker_count = std::max(4u, std::thread::hardware_concurrency());

// Rationale: compute_trade_metric() is CPU-bound despite sleep
// More threads than cores would increase context switching overhead
// Fewer threads would underutilize available parallelism

// Naive approach (O(n²) for each new trade)
for (auto& new_trade : incoming_trades) {
    for (auto& symbol_window : all_windows) {
        for (auto& existing_trade : symbol_window) {
            if (within_correlation_window(new_trade, existing_trade)) {
                // Process correlation
            }
        }
    }
}

// Optimized approach (O(n) amortized)
void merger_optimized() {
    // 1. Expire old trades in O(k) where k = expired trades
    int64_t cutoff = new_trade.timestamp_us - CORRELATION_WINDOW;
    for (auto& [symbol, deque] : windows) {
        while (!deque.empty() && deque.front().timestamp_us < cutoff) {
            deque.pop_front();  // O(1)
        }
    }
    
    // 2. Check correlations only in active window O(m) where m = active trades
    for (auto& [other_symbol, other_deque] : windows) {
        if (other_symbol == new_trade.symbol) continue;
        
        for (auto& existing_trade : other_deque) {
            // All trades in deque are within window by construction
            process_correlation(new_trade, existing_trade);
        }
    }
}

// Design Decision: Structure of Arrays vs Array of Structures
// Chosen: Hybrid approach optimized for access patterns

// Hot path: Sequential access for timestamp comparisons
struct TimeWindow {
    std::deque<int64_t> timestamps;     // Cache-friendly sequential access
    std::deque<Trade*> trade_ptrs;      // Parallel array for full trade data
};

// Cold path: Random access for full trade processing
struct Trade {
    // Fields ordered by access frequency (hot fields first)
    int64_t timestamp_us;        // Most frequent: correlation detection
    uint64_t trade_id;           // Frequent: deduplication
    std::string symbol;          // Frequent: routing
    std::string time_str;        // Infrequent: output formatting
    std::string side, size, price; // Infrequent: output only
};

// Problem: Fixed-size queues either waste memory or drop data
// Solution: Intelligent dropping strategies based on queue purpose

enum DropStrategy {
    DROP_OLDEST,     // Trade queue: Keep recent data, drop stale
    DROP_NEWEST,     // Filter queue: Process older correlations first  
    BACKPRESSURE,    // Write queue: Never drop, apply backpressure
    DROP_BY_PRIORITY // Future: Priority-based dropping
};

template<typename T>
class AdaptiveBoundedQueue {
    // Queue sizes calibrated from burst analysis:
    // - Trade queue: 25,000 (handles 10s of maximum observed burst rate)
    // - Filter queue: 15,000 (handles correlation amplification) 
    // - Write queue: 15,000 (handles worker output buffering)
};

class ExponentialBackoff {
    // Design parameters calibrated for Coinbase API characteristics
    std::chrono::milliseconds base_delay_{100};    // Normal poll interval
    std::chrono::milliseconds max_delay_{30000};   // Max backoff (30s)
    int consecutive_failures_{0};
    
public:
    void on_failure() {
        ++consecutive_failures_;
        current_delay_ = std::min(max_delay_, 
            std::chrono::milliseconds(current_delay_.count() * 2));
    }
    
    bool should_circuit_break(int threshold = 10) const {
        return consecutive_failures_ >= threshold;
    }
};

// Usage in API polling loop
while (!shutdown) {
    if (backoff.should_circuit_break(10)) {
        std::cerr << "Circuit breaker triggered, stopping poller\n";
        break;  // Fail-stop to prevent API abuse
    }
    
    if (api_call_successful()) {
        backoff.on_success();  // Reset to base interval
    } else {
        backoff.on_failure();  // Exponential increase
        std::this_thread::sleep_for(backoff.get_delay());
    }
}

// Problem: Sliding windows can grow unbounded during bursts
// Solution: LRU eviction with memory pressure monitoring

class BoundedSlidingWindow {
    static constexpr size_t MAX_WINDOW_SIZE = 10000;  // Safety limit
    std::unordered_map<std::string, std::deque<Trade>> windows_;
    
    void add_trade(const Trade& trade) {
        auto& window = windows_[trade.symbol];
        
        // Add new trade
        window.push_back(trade);
        
        // Enforce memory bounds (LRU eviction)
        while (window.size() > MAX_WINDOW_SIZE) {
            window.pop_front();  // Remove oldest
        }
        
        // Normal time-based expiration
        int64_t cutoff = trade.timestamp_us - CORRELATION_WINDOW_US;
        while (!window.empty() && window.front().timestamp_us < cutoff) {
            window.pop_front();
        }
    }
};

// Design Decision: Prioritized processing during overload
class LoadSheddingManager {
    enum Priority {
        HIGH = 0,    // Large correlations (>10 trades)
        MEDIUM = 1,  // Medium correlations (3-10 trades)  
        LOW = 2      // Small correlations (2 trades)
    };
    
    bool should_process(const CorrelationEvent& event) {
        if (system_load() < 0.8) {
            return true;  // Normal operation
        }
        
        // Under load: prioritize by correlation size
        if (event.correlated_trades.size() >= 10) {
            return true;  // Always process large correlations
        } else if (event.correlated_trades.size() >= 3) {
            return (rand() % 100) < 50;  // 50% sampling for medium
        } else {
            return (rand() % 100) < 10;  // 10% sampling for small
        }
    }
};

template<typename QueueType>
void drain_queue(QueueType& queue, const std::string& name) {
    auto start = std::chrono::steady_clock::now();
    const auto timeout = std::chrono::seconds(30);
    
    while (queue.size() > 0) {
        auto elapsed = std::chrono::steady_clock::now() - start;
        if (elapsed > timeout) {
            std::cerr << "Timeout draining " << name << " queue, " 
                     << queue.size() << " items remaining\n";
            break;  // Controlled data loss rather than hang
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}

// Adjust polling frequency based on response latency and queue depth
void adaptive_polling() {
    auto base_interval = std::chrono::milliseconds(100);
    auto current_interval = base_interval;
    
    while (!shutdown) {
        auto start = std::chrono::steady_clock::now();
        
        bool success = poll_coinbase_api();
        size_t queue_depth = trade_queue.size();
        
        auto latency = std::chrono::steady_clock::now() - start;
        
        if (success && queue_depth < 1000 && latency < std::chrono::milliseconds(50)) {
            // System healthy: increase polling rate
            current_interval = std::max(std::chrono::milliseconds(50), 
                                      current_interval * 9 / 10);
        } else {
            // System stressed: decrease polling rate
            current_interval = std::min(std::chrono::milliseconds(500),
                                      current_interval * 11 / 10);
        }
        
        std::this_thread::sleep_for(current_interval);
    }
}

// HTTP connection reuse and tuning
curl_easy_setopt(c, CURLOPT_TIMEOUT, 10L);
curl_easy_setopt(c, CURLOPT_CONNECTTIMEOUT, 5L);
curl_easy_setopt(c, CURLOPT_TCP_KEEPALIVE, 1L);
curl_easy_setopt(c, CURLOPT_TCP_KEEPIDLE, 120L);
curl_easy_setopt(c, CURLOPT_TCP_KEEPINTVL, 60L);

// Efficient timestamp parsing for Coinbase ISO8601 format
int64_t parse_iso8601_to_us(const std::string &s) {
    int y, M, d, h, m, sec, frac = 0, len = 0;
    if (std::sscanf(s.c_str(), "%4d-%2d-%2dT%2d:%2d:%2d.%6dZ", 
                    &y, &M, &d, &h, &m, &sec, &frac) < 6)
        return 0;
        
    // Optimize fractional second handling
    auto dot = s.find('.');
    if (dot != std::string::npos) {
        auto z = s.find('Z', dot);
        len = z - dot - 1;
        if (len > 6) len = 6;
    }
    
    int mult = 1;
    for (int i = 0; i < 6 - len; ++i) mult *= 10;
    
    std::tm tm{};
    tm.tm_year = y - 1900; tm.tm_mon = M - 1; tm.tm_mday = d;
    tm.tm_hour = h; tm.tm_min = m; tm.tm_sec = sec;
    
    time_t epoch = timegm(&tm);
    return int64_t(epoch) * 1'000'000 + int64_t(frac) * mult;
}

class ExponentialBackoff {
    std::chrono::milliseconds base_delay_;
    std::chrono::milliseconds max_delay_;
    std::chrono::milliseconds current_delay_;
    int consecutive_failures_;

public:
    ExponentialBackoff(int base_ms = 100, int max_ms = 30000)
        : base_delay_(base_ms), max_delay_(max_ms), 
          current_delay_(base_ms), consecutive_failures_(0) {}

    void on_failure() {
        ++consecutive_failures_;
        current_delay_ = std::min(max_delay_, 
            std::chrono::milliseconds(current_delay_.count() * 2));
    }
    
    bool should_circuit_break(int threshold) const {
        return consecutive_failures_ >= threshold;
    }
};

class GracefulShutdown {
    std::atomic<bool> shutdown_requested_{false};
    std::chrono::seconds drain_timeout_{30};

public:
    template <typename QueueType>
    void drain_queue(QueueType &queue, const std::string &name) {
        std::cerr << "[SHUTDOWN] Draining " << name << " queue...\n";
        auto start = std::chrono::steady_clock::now();

        while (queue.size() > 0) {
            auto elapsed = std::chrono::steady_clock::now() - start;
            if (elapsed > drain_timeout_) {
                std::cerr << "[SHUTDOWN] Timeout draining " << name 
                         << " queue, " << queue.size() << " items remaining\n";
                break;
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        }
    }
};

// Handle Coinbase rate limiting (429 responses)
if (r.code == 429) {
    auto rate_limit_remaining = r.hdrs.find("cb-ratelimit-remaining");
    if (rate_limit_remaining != r.hdrs.end()) {
        std::cerr << "[WARN] Rate limited for " << sym
                  << ", remaining: " << rate_limit_remaining->second << std::endl;
    }
    backoff.on_failure();
}

// Circuit breaker activation
if (backoff.should_circuit_break(g_config.max_consecutive_failures)) {
    std::cerr << "[ERROR] Circuit breaker triggered for " << sym 
              << " after " << backoff.failure_count() << " failures\n";
    break;
}

class SystemMetrics {
    std::atomic<uint64_t> trades_received_{0};
    std::atomic<uint64_t> trades_filtered_{0};
    std::atomic<uint64_t> trades_processed_{0};
    std::atomic<uint64_t> trades_written_{0};
    std::atomic<uint64_t> api_calls_{0};
    std::atomic<uint64_t> api_errors_{0};
    std::atomic<uint64_t> queue_overflows_{0};
    std::atomic<uint64_t> total_latency_us_{0};
    std::atomic<uint64_t> latency_samples_{0};

public:
    std::string get_json_stats() const {
        auto uptime_s = std::chrono::duration_cast<std::chrono::seconds>(
            std::chrono::steady_clock::now() - start_time_).count();

        double avg_latency_ms = latency_samples_ > 0 ? 
            double(total_latency_us_) / latency_samples_ / 1000.0 : 0.0;

        std::ostringstream oss;
        oss << "{\n"
            << "  \"uptime_seconds\": " << uptime_s << ",\n"
            << "  \"trades_received\": " << trades_received_ << ",\n"
            << "  \"trades_filtered\": " << trades_filtered_ << ",\n"
            << "  \"trades_processed\": " << trades_processed_ << ",\n"
            << "  \"trades_written\": " << trades_written_ << ",\n"
            << "  \"api_calls\": " << api_calls_ << ",\n"
            << "  \"api_errors\": " << api_errors_ << ",\n"
            << "  \"queue_overflows\": " << queue_overflows_ << ",\n"
            << "  \"avg_latency_ms\": " << avg_latency_ms << ",\n"
            << "  \"processing_rate\": " << 
                (trades_received_ > 0 ? double(trades_processed_) / trades_received_ : 0.0) << ",\n"
            << "  \"error_rate\": " << 
                (api_calls_ > 0 ? double(api_errors_) / api_calls_ : 0.0) << "\n"
            << "}";
        return oss.str();
    }
};

// Input validation for API responses
bool validate_trade_data(const json& trade) {
    return trade.contains("trade_id") && 
           trade.contains("time") &&
           trade.contains("size") &&
           trade.contains("price") &&
           trade["trade_id"].is_number() &&
           !trade["time"].get<std::string>().empty();
}

// Rate limiting compliance
struct RateLimiter {
    std::chrono::steady_clock::time_point last_request;
    std::chrono::milliseconds min_interval{100};
    
    void enforce_limit() {
        auto now = std::chrono::steady_clock::now();
        auto elapsed = now - last_request;
        if (elapsed < min_interval) {
            std::this_thread::sleep_for(min_interval - elapsed);
        }
        last_request = std::chrono::steady_clock::now();
    }
};

/*******************************************************************
 *  Spec-Compliant Coinbase Trade Aggregator
 *  Build: g++ -std=c++20 -O3 -Wall -Wextra -pedantic -pthread \
 *               main_production.cpp -o coinbase_aggregator
 * -lcurl
 *******************************************************************/
#include <algorithm>
#include <array>
#include <atomic>
#include <chrono>
#include <cmath>
#include <condition_variable>
#include <csignal>
#include <cstdint>
#include <cstring>
#include <ctime>
#include <curl/curl.h>
#include <deque>
#include <fcntl.h>
#include <fstream>
#include <iostream>
#include <map>
#include <mutex>
#include <nlohmann/json.hpp>
#include <queue>
#include <sstream>
#include <string>
#include <string_view>
#include <sys/mman.h>
#include <thread>
#include <time.h>
#include <tuple>
#include <unistd.h>
#include <unordered_map>
#include <unordered_set>
#include <vector>

using json = nlohmann::json;

/* ──────────────────────────────────────────────────────────
 *   CONFIGURATION STRUCTURE
 * ────────────────────────────────────────────────────────── */
struct Config {
  size_t trade_queue_size = 25000;
  size_t filter_queue_size = 15000;
  size_t write_queue_size = 15000;
  int poll_interval_ms = 100;           // Back to original fast polling
  int64_t correlation_window_us = 5000; // 5ms
  std::vector<std::string> symbols = {"BTC-USD", "ETH-USD",
                                      "SOL-USD"}; // Back to Coinbase symbols
  std::string output_file = "trade_metrics.txt";
  size_t mmap_initial_size_mb = 50;
  int max_consecutive_failures = 10;

  void load_from_env() {
    if (auto env_val = std::getenv("POLL_INTERVAL_MS")) {
      poll_interval_ms = std::atoi(env_val);
    }
    if (auto env_val = std::getenv("CORRELATION_WINDOW_US")) {
      correlation_window_us = std::atoll(env_val);
    }
    if (auto env_val = std::getenv("OUTPUT_FILE")) {
      output_file = env_val;
    }
  }
};

/* ──────────────────────────────────────────────────────────
 *   METRICS AND OBSERVABILITY
 * ────────────────────────────────────────────────────────── */
class SystemMetrics {
  std::atomic<uint64_t> trades_received_{0};
  std::atomic<uint64_t> trades_filtered_{0};
  std::atomic<uint64_t> trades_processed_{0};
  std::atomic<uint64_t> trades_written_{0};
  std::atomic<uint64_t> api_calls_{0};
  std::atomic<uint64_t> api_errors_{0};
  std::atomic<uint64_t> queue_overflows_{0};
  std::atomic<uint64_t> total_latency_us_{0};
  std::atomic<uint64_t> latency_samples_{0};
  mutable std::mutex stats_mutex_;
  std::chrono::steady_clock::time_point start_time_;

public:
  SystemMetrics() : start_time_(std::chrono::steady_clock::now()) {}

  void increment_received() { ++trades_received_; }
  void increment_filtered() { ++trades_filtered_; }
  void increment_processed() { ++trades_processed_; }
  void increment_written() { ++trades_written_; }
  void increment_api_calls() { ++api_calls_; }
  void increment_api_errors() { ++api_errors_; }
  void increment_queue_overflows() { ++queue_overflows_; }

  void add_latency(uint64_t latency_us) {
    total_latency_us_ += latency_us;
    ++latency_samples_;
  }

  uint64_t get_trades_received() const { return trades_received_.load(); }
  uint64_t get_trades_filtered() const { return trades_filtered_.load(); }
  uint64_t get_trades_processed() const { return trades_processed_.load(); }
  uint64_t get_trades_written() const { return trades_written_.load(); }
  uint64_t get_api_calls() const { return api_calls_.load(); }
  uint64_t get_api_errors() const { return api_errors_.load(); }
  uint64_t get_queue_overflows() const { return queue_overflows_.load(); }
  uint64_t get_latency_samples() const { return latency_samples_.load(); }
  uint64_t get_total_latency_us() const { return total_latency_us_.load(); }

  std::string get_json_stats() const {
    std::lock_guard<std::mutex> lock(stats_mutex_);
    auto uptime_s = std::chrono::duration_cast<std::chrono::seconds>(
                        std::chrono::steady_clock::now() - start_time_)
                        .count();

    uint64_t received = trades_received_;
    uint64_t filtered = trades_filtered_;
    uint64_t processed = trades_processed_;
    uint64_t written = trades_written_;
    uint64_t api_calls = api_calls_;
    uint64_t api_errors = api_errors_;
    uint64_t overflows = queue_overflows_;
    uint64_t lat_samples = latency_samples_;
    double avg_latency_ms =
        lat_samples > 0 ? double(total_latency_us_) / lat_samples / 1000.0
                        : 0.0;

    std::ostringstream oss;
    oss << "{\n"
        << "  \"uptime_seconds\": " << uptime_s << ",\n"
        << "  \"trades_received\": " << received << ",\n"
        << "  \"trades_filtered\": " << filtered << ",\n"
        << "  \"trades_processed\": " << processed << ",\n"
        << "  \"trades_written\": " << written << ",\n"
        << "  \"api_calls\": " << api_calls << ",\n"
        << "  \"api_errors\": " << api_errors << ",\n"
        << "  \"queue_overflows\": " << overflows << ",\n"
        << "  \"avg_latency_ms\": " << avg_latency_ms << ",\n"
        << "  \"processing_rate\": "
        << (received > 0 ? double(processed) / received : 0.0) << ",\n"
        << "  \"error_rate\": "
        << (api_calls > 0 ? double(api_errors) / api_calls : 0.0) << "\n"
        << "}";
    return oss.str();
  }
};

/* ──────────────────────────────────────────────────────────
 *   EXPONENTIAL BACKOFF FOR RESILIENCE
 * ────────────────────────────────────────────────────────── */
class ExponentialBackoff {
  std::chrono::milliseconds base_delay_;
  std::chrono::milliseconds max_delay_;
  std::chrono::milliseconds current_delay_;
  int consecutive_failures_;

public:
  ExponentialBackoff(int base_ms = 100, int max_ms = 30000)
      : base_delay_(base_ms), max_delay_(max_ms), current_delay_(base_ms),
        consecutive_failures_(0) {}

  void on_success() {
    consecutive_failures_ = 0;
    current_delay_ = base_delay_;
  }

  void on_failure() {
    ++consecutive_failures_;
    current_delay_ = std::min(
        max_delay_, std::chrono::milliseconds(current_delay_.count() * 2));
  }

  std::chrono::milliseconds get_delay() const { return current_delay_; }
  int failure_count() const { return consecutive_failures_; }
  bool should_circuit_break(int threshold) const {
    return consecutive_failures_ >= threshold;
  }
};

/* ──────────────────────────────────────────────────────────
 *   GRACEFUL SHUTDOWN MANAGER
 * ────────────────────────────────────────────────────────── */
class GracefulShutdown {
  std::atomic<bool> shutdown_requested_{false};
  std::atomic<bool> drain_complete_{false};
  std::chrono::seconds drain_timeout_{30};

public:
  void request_shutdown() {
    std::cerr << "[SHUTDOWN] Graceful shutdown requested...\n";
    shutdown_requested_.store(true);
  }

  bool should_shutdown() const { return shutdown_requested_.load(); }

  template <typename QueueType>
  void drain_queue(QueueType &queue, const std::string &name) {
    std::cerr << "[SHUTDOWN] Draining " << name << " queue...\n";
    auto start = std::chrono::steady_clock::now();

    while (queue.size() > 0) {
      auto elapsed = std::chrono::steady_clock::now() - start;
      if (elapsed > drain_timeout_) {
        std::cerr << "[SHUTDOWN] Timeout draining " << name << " queue, "
                  << queue.size() << " items remaining\n";
        break;
      }
      std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    std::cerr << "[SHUTDOWN] " << name << " queue drained\n";
  }

  void mark_drain_complete() { drain_complete_.store(true); }
  bool is_drain_complete() const { return drain_complete_.load(); }
};

// Global instances
static Config g_config;
static SystemMetrics g_metrics;
static GracefulShutdown g_shutdown;

std::atomic<uint64_t> total_trades_filtered{0};

/* ──────────────────────────────────────────────────────────
 *   GLOBAL DEBUG MACRO (OFF BY DEFAULT)
 * ────────────────────────────────────────────────────────── */
#ifndef ENABLE_DEBUG_LOG
#define ENABLE_DEBUG_LOG 0
#endif
#define DLOG                                                                   \
  if constexpr (ENABLE_DEBUG_LOG)                                              \
  std::cerr

/* ──────────────────────────────────────────────────────────
 *   1) THREAD-SAFE BOUNDED QUEUE TEMPLATE
 * ────────────────────────────────────────────────────────── */
template <typename T> class SmartBoundedQueue {
public:
  enum DropStrategy {
    DROP_OLDEST,
    DROP_NEWEST,
    BACKPRESSURE,
    DROP_BY_PRIORITY
  };

  SmartBoundedQueue(size_t max_size = 10000,
                    DropStrategy strategy = DROP_OLDEST)
      : max_size_(max_size), dropped_count_(0), total_pushed_(0),
        strategy_(strategy) {}

  bool push(const T &value, int /*priority*/ = 0) {
    std::unique_lock<std::mutex> lk(m_);
    ++total_pushed_;

    if (q_.size() >= max_size_) {
      switch (strategy_) {
      case DROP_OLDEST:
        q_.pop();
        ++dropped_count_;
        g_metrics.increment_queue_overflows();
        break;
      case DROP_NEWEST:
        ++dropped_count_;
        g_metrics.increment_queue_overflows();
        return false;
      case BACKPRESSURE:
        if (!cv_space_.wait_for(lk, std::chrono::milliseconds(100), [&] {
              return q_.size() < max_size_ || shutdown_;
            })) {
          ++dropped_count_;
          g_metrics.increment_queue_overflows();
          return false;
        }
        break;
      case DROP_BY_PRIORITY:
        q_.pop();
        ++dropped_count_;
        g_metrics.increment_queue_overflows();
        break;
      }
    }
    q_.push(value);
    cv_.notify_one();
    return true;
  }

  bool pop(T &out) {
    std::unique_lock<std::mutex> lk(m_);
    cv_.wait(lk, [&] { return !q_.empty() || shutdown_; });
    if (shutdown_ && q_.empty())
      return false;
    out = std::move(q_.front());
    q_.pop();
    cv_space_.notify_one();
    return true;
  }

  void shutdown() {
    {
      std::lock_guard<std::mutex> lk(m_);
      shutdown_ = true;
    }
    cv_.notify_all();
    cv_space_.notify_all();
  }

  size_t size() const {
    std::lock_guard<std::mutex> lk(m_);
    return q_.size();
  }
  uint64_t dropped() const {
    std::lock_guard<std::mutex> lk(m_);
    return dropped_count_;
  }
  uint64_t total_push() const {
    std::lock_guard<std::mutex> lk(m_);
    return total_pushed_;
  }
  double drop_rate() const {
    std::lock_guard<std::mutex> lk(m_);
    return total_pushed_ ? double(dropped_count_) / total_pushed_ : 0.0;
  }

private:
  std::queue<T> q_;
  mutable std::mutex m_;
  std::condition_variable cv_, cv_space_;
  bool shutdown_{false};
  size_t max_size_;
  uint64_t dropped_count_;
  uint64_t total_pushed_;
  DropStrategy strategy_;
};

/* ──────────────────────────────────────────────────────────
 *   2) SYMBOL-PRIORITY MANAGER
 * ────────────────────────────────────────────────────────── */
class AdaptiveSymbolManager {
  struct Metrics {
    std::atomic<uint64_t> trade_cnt{0};
    std::atomic<uint64_t> volume_usd{0};
    std::chrono::steady_clock::time_point last_trade{
        std::chrono::steady_clock::now()};
    double corr_freq{0.0};
  };
  std::unordered_map<std::string, Metrics> m_;
  std::mutex mu_;

public:
  void update_activity(const std::string &sym, double vol) {
    std::lock_guard lk(mu_);
    auto &s = m_[sym];
    ++s.trade_cnt;
    s.volume_usd += static_cast<uint64_t>(vol);
    s.last_trade = std::chrono::steady_clock::now();
  }
  void update_corr(const std::string &sym, bool hit) {
    std::lock_guard lk(mu_);
    auto &s = m_[sym];
    s.corr_freq = 0.95 * s.corr_freq + 0.05 * (hit ? 1.0 : 0.0);
  }
  std::vector<std::string> ordered() {
    std::lock_guard lk(mu_);
    std::vector<std::pair<std::string, double>> v;
    for (auto &[sym, s] : m_) {
      auto age = std::chrono::duration_cast<std::chrono::seconds>(
                     std::chrono::steady_clock::now() - s.last_trade)
                     .count();
      double rec = std::exp(-age / 300.0);
      double score =
          (s.volume_usd / 1'000'000.0) * (1.0 + 10 * s.corr_freq) * rec;
      v.emplace_back(sym, score);
    }
    std::sort(v.begin(), v.end(),
              [](auto &a, auto &b) { return a.second > b.second; });
    std::vector<std::string> out;
    for (auto &[sym, _] : v)
      out.push_back(sym);
    return out;
  }
  void print() {
    auto o = ordered();
    std::cerr << "[SymbolMgr] ";
    for (size_t i = 0; i < o.size(); ++i) {
      std::cerr << o[i] << (i + 1 == o.size() ? "\n" : " > ");
    }
  }
};

/* ──────────────────────────────────────────────────────────
 *   3) HIGH-PERFORMANCE MMAP WRITER
 * ────────────────────────────────────────────────────────── */
class HighPerformanceWriter {
  int fd_{-1};
  char *mem_{nullptr};
  size_t sz_{0};
  size_t init_sz_;
  std::atomic<size_t> off_{0};
  std::mutex mmap_mu_;

  struct Entry {
    std::string data;
    uint64_t seq;
  };
  static constexpr size_t RING = 4096;
  std::array<Entry, RING> ring_;
  std::atomic<size_t> head_{0}, tail_{0};
  std::thread io_;
  std::atomic<bool> stop_{false};

  void grow() {
    std::lock_guard lk(mmap_mu_);
    size_t ns = sz_ * 2;
    munmap(mem_, sz_);
    if (ftruncate(fd_, static_cast<off_t>(ns)) == -1) {
      std::cerr << "[Writer] ftruncate failed\n";
      return;
    }
    mem_ = static_cast<char *>(
        mmap(nullptr, ns, PROT_WRITE | PROT_READ, MAP_SHARED, fd_, 0));
    sz_ = ns;
    std::cerr << "[Writer] Expanded mmap to " << sz_ / 1024 / 1024 << "MB\n";
  }

  void copy_to_mmap(const std::string &d) {
    std::lock_guard lk(mmap_mu_);
    size_t o = off_.fetch_add(d.size());
    if (o + d.size() > sz_)
      grow();
    std::memcpy(mem_ + o, d.data(), d.size());
  }

  void background() {
    std::map<uint64_t, std::string> buf;
    uint64_t next = 0;
    while (!stop_.load()) {
      size_t h = head_.load(), t = tail_.load(std::memory_order_acquire);
      while (h != t) {
        buf.emplace(ring_[h].seq, std::move(ring_[h].data));
        h = (h + 1) % RING;
      }
      head_.store(h, std::memory_order_release);
      auto it = buf.find(next);
      while (it != buf.end()) {
        copy_to_mmap(it->second);
        buf.erase(it);
        ++next;
        it = buf.find(next);
      }
      std::this_thread::sleep_for(std::chrono::microseconds(100));
    }
    msync(mem_, off_.load(), MS_SYNC);
  }

public:
  explicit HighPerformanceWriter(const std::string &f, size_t initial_mb = 50)
      : init_sz_(initial_mb * 1024 * 1024) {
    fd_ = open(f.c_str(), O_RDWR | O_CREAT | O_TRUNC, 0644);
    if (ftruncate(fd_, static_cast<off_t>(init_sz_)) == -1) {
      std::cerr << "[Writer] initial ftruncate failed\n";
    }
    mem_ = static_cast<char *>(
        mmap(nullptr, init_sz_, PROT_READ | PROT_WRITE, MAP_SHARED, fd_, 0));
    sz_ = init_sz_;
    std::cerr << "[Writer] mmap " << sz_ / 1024 / 1024 << "MB\n";
    io_ = std::thread(&HighPerformanceWriter::background, this);
  }

  bool write_async(std::string d, uint64_t seq) {
    size_t t = tail_.load(), n = (t + 1) % RING;
    if (n == head_.load(std::memory_order_acquire))
      return false;
    ring_[t].data = std::move(d);
    ring_[t].seq = seq;
    tail_.store(n, std::memory_order_release);
    return true;
  }

  void stop() {
    stop_.store(true);
    if (io_.joinable())
      io_.join();
  }

  ~HighPerformanceWriter() {
    stop();
    if (mem_)
      munmap(mem_, sz_);
    if (fd_ != -1)
      close(fd_);
  }
};

/* ──────────────────────────────────────────────────────────
 *   4) TRADE STRUCTURES - RESTORED REAL COINBASE FORMAT
 * ────────────────────────────────────────────────────────── */
struct Trade {
  uint64_t trade_id{};
  std::string symbol, time_str, side, size, price;
  int64_t timestamp_us{};
  std::chrono::steady_clock::time_point received_at{
      std::chrono::steady_clock::now()};
};

struct RawTrade {
  uint64_t trade_id;
  std::string side, size, price, time;
};

static void from_json(const json &j, RawTrade &r) {
  r.trade_id = j.at("trade_id").get<uint64_t>();
  r.side = j.at("side");
  r.size = j.at("size");
  r.price = j.at("price");
  r.time = j.at("time");
}

/* ──────────────────────────────────────────────────────────
 *   5) ISO-8601 → microseconds
 * ────────────────────────────────────────────────────────── */
int64_t parse_iso8601_to_us(const std::string &s) {
  int y, M, d, h, m, sec, frac = 0, len = 0;
  if (std::sscanf(s.c_str(), "%4d-%2d-%2dT%2d:%2d:%2d.%6dZ", &y, &M, &d, &h, &m,
                  &sec, &frac) < 6)
    return 0;
  auto dot = s.find('.');
  if (dot != std::string::npos) {
    auto z = s.find('Z', dot);
    len = z - dot - 1;
    if (len > 6)
      len = 6;
  }
  int mult = 1;
  for (int i = 0; i < 6 - len; ++i)
    mult *= 10;
  std::tm tm{};
  tm.tm_year = y - 1900;
  tm.tm_mon = M - 1;
  tm.tm_mday = d;
  tm.tm_hour = h;
  tm.tm_min = m;
  tm.tm_sec = sec;
  time_t epoch = timegm(&tm);
  return int64_t(epoch) * 1'000'000 + int64_t(frac) * mult;
}

/* ──────────────────────────────────────────────────────────
 *   6) COMPUTE METRIC
 * ────────────────────────────────────────────────────────── */
static std::string
compute_trade_metric(const std::string &symbol, uint64_t /*trade_id*/,
                     const std::string &time_str, const std::string &side,
                     const std::string &size, const std::string &price) {
  std::this_thread::sleep_for(std::chrono::milliseconds(5));
  std::ostringstream o;
  o << symbol << '\t' << time_str << '\t' << side << '\t' << size << '\t'
    << price << '\n';
  return o.str();
}

/* ──────────────────────────────────────────────────────────
 *   7) PIPELINE QUEUES + GLOBALS
 * ────────────────────────────────────────────────────────── */
std::unique_ptr<SmartBoundedQueue<Trade>> trade_q;
std::unique_ptr<SmartBoundedQueue<std::pair<uint64_t, Trade>>> filt_q;
std::unique_ptr<SmartBoundedQueue<std::pair<uint64_t, std::string>>> write_q;

std::atomic<uint64_t> next_seq{0};
std::atomic<bool> stop_all{false};

AdaptiveSymbolManager sym_mgr;
std::unique_ptr<HighPerformanceWriter> hpw;

/* Use (symbol, trade_id) pairs for proper deduplication */
static std::unordered_set<std::string> already_emitted;
static std::mutex emitted_mu;

/* ──────────────────────────────────────────────────────────
 *   8) CURL HELPER
 * ────────────────────────────────────────────────────────── */
struct HttpResp {
  std::string data;
  long code{0};
  std::map<std::string, std::string> hdrs;
};

static size_t wr_cb(void *c, size_t s, size_t n, void *u) {
  static_cast<HttpResp *>(u)->data.append((char *)c, s * n);
  return s * n;
}

static size_t hd_cb(char *buf, size_t s, size_t n, void *u) {
  std::string h(buf, s * n);
  auto c = h.find(':');
  if (c != std::string::npos) {
    std::string k = h.substr(0, c);
    std::string v = h.substr(c + 1);
    k.erase(std::remove_if(k.begin(), k.end(), ::isspace), k.end());
    v.erase(std::remove_if(v.begin(), v.end(), ::isspace), v.end());
    std::transform(k.begin(), k.end(), k.begin(), ::tolower);
    static_cast<HttpResp *>(u)->hdrs[k] = v;
  }
  return s * n;
}

/* ──────────────────────────────────────────────────────────
 *   9) SPEC-COMPLIANT COINBASE POLLER
 * ────────────────────────────────────────────────────────── */
void poller(const std::string &sym) {
  CURL *c = curl_easy_init();
  if (!c) {
    std::cerr << "[ERROR] curl init failed for " << sym << "\n";
    return;
  }

  // FIXED: Back to real Coinbase trade endpoint
  std::string url =
      "https://api.exchange.coinbase.com/products/" + sym + "/trades";

  struct curl_slist *hdr = nullptr;
  hdr = curl_slist_append(hdr, "Accept: application/json");
  hdr = curl_slist_append(hdr, "User-Agent: TradeAggregator/1.0");

  curl_easy_setopt(c, CURLOPT_HTTPHEADER, hdr);
  curl_easy_setopt(c, CURLOPT_WRITEFUNCTION, wr_cb);
  curl_easy_setopt(c, CURLOPT_HEADERFUNCTION, hd_cb);
  curl_easy_setopt(c, CURLOPT_TIMEOUT, 10L);
  curl_easy_setopt(c, CURLOPT_CONNECTTIMEOUT, 5L);

  std::unordered_set<uint64_t> dup;
  ExponentialBackoff backoff(g_config.poll_interval_ms);

  std::cerr << "[POLLER] Starting poller for " << sym
            << " (Coinbase Exchange API)" << std::endl;

  while (!stop_all.load() && !g_shutdown.should_shutdown()) {
    HttpResp r;
    curl_easy_setopt(c, CURLOPT_URL, url.c_str());
    curl_easy_setopt(c, CURLOPT_WRITEDATA, &r);
    curl_easy_setopt(c, CURLOPT_HEADERDATA, &r);

    g_metrics.increment_api_calls();
    CURLcode rc = curl_easy_perform(c);
    curl_easy_getinfo(c, CURLINFO_RESPONSE_CODE, &r.code);

    if (rc == CURLE_OK && r.code == 200) {
      backoff.on_success();
      try {
        auto j = json::parse(r.data);

        // FIXED: Parse real Coinbase trade JSON array
        for (auto &it : j) {
          RawTrade rt = it.get<RawTrade>();

          // Skip duplicates using real trade_id
          if (!dup.insert(rt.trade_id).second)
            continue;
          if (dup.size() > 1000) {
            dup.clear();
            dup.insert(rt.trade_id);
          }

          // FIXED: Parse real ISO8601 timestamp to microseconds
          int64_t ts = parse_iso8601_to_us(rt.time);
          if (!ts)
            continue;

          // Create real trade with Coinbase data
          Trade t;
          t.trade_id = rt.trade_id; // Real Coinbase trade_id
          t.symbol = sym;
          t.time_str = rt.time; // Real ISO8601 time
          t.side = rt.side;     // Real side (buy/sell)
          t.size = rt.size;     // Real size
          t.price = rt.price;   // Real price
          t.timestamp_us = ts;  // Real timestamp in microseconds

          if (!trade_q->push(t))
            continue;

          double vol = std::stod(rt.size) * std::stod(rt.price);
          sym_mgr.update_activity(sym, vol);
          g_metrics.increment_received();

          DLOG << "[TRADE] " << sym << " id=" << rt.trade_id
               << " price=" << rt.price << " size=" << rt.size
               << " side=" << rt.side << std::endl;
        }
      } catch (const std::exception &e) {
        std::cerr << "[ERROR] JSON parse failed for " << sym << ": " << e.what()
                  << std::endl;
        g_metrics.increment_api_errors();
        backoff.on_failure();
      }
    } else {
      std::cerr << "[ERROR] API call failed for " << sym << ": curl=" << rc
                << " http=" << r.code << std::endl;
      g_metrics.increment_api_errors();
      backoff.on_failure();

      // FIXED: Handle Coinbase rate limiting properly
      if (r.code == 429) {
        auto rate_limit_remaining = r.hdrs.find("cb-ratelimit-remaining");
        if (rate_limit_remaining != r.hdrs.end()) {
          std::cerr << "[WARN] Rate limited for " << sym
                    << ", remaining: " << rate_limit_remaining->second
                    << std::endl;
        }
      }
    }

    // Circuit breaker
    if (backoff.should_circuit_break(g_config.max_consecutive_failures)) {
      std::cerr << "[ERROR] Circuit breaker triggered for " << sym << " after "
                << backoff.failure_count() << " failures\n";
      break;
    }

    std::this_thread::sleep_for(backoff.get_delay());
  }

  std::cerr << "[POLLER] Stopping poller for " << sym << std::endl;
  curl_slist_free_all(hdr);
  curl_easy_cleanup(c);
}

/* ──────────────────────────────────────────────────────────
 *   10) MERGER THREAD - FIXED FOR PROPER DEDUPLICATION
 * ────────────────────────────────────────────────────────── */
void merger() {
  std::unordered_map<std::string, std::deque<Trade>> win;
  const int64_t W =
      g_config.correlation_window_us; // Now properly 5000 µs = 5ms

  std::cerr << "[MERGER] Starting merger thread (correlation window: " << W
            << " µs = " << W / 1000.0 << " ms)\n";

  while (!stop_all.load() && !g_shutdown.should_shutdown()) {
    Trade nt;
    if (!trade_q->pop(nt))
      break;

    // FIRST: Check if this trade was already emitted
    std::string nt_key = nt.symbol + ":" + std::to_string(nt.trade_id);
    {
      std::lock_guard lk(emitted_mu);
      if (already_emitted.count(nt_key)) {
        // Skip trades we've already processed
        continue;
      }
    }

    // Clean old trades outside correlation window
    int64_t cut = nt.timestamp_us - W;
    for (auto &[_, dq] : win) {
      while (!dq.empty() && dq.front().timestamp_us < cut) {
        dq.pop_front();
      }
    }

    bool found_correlation = false;
    std::vector<Trade> correlated_trades; // Collect all correlated trades

    // Look for correlations with other symbols
    auto order = sym_mgr.ordered();
    for (const auto &os : order) {
      if (os == nt.symbol)
        continue;
      auto &dq = win[os];
      for (const auto &ex : dq) {
        int64_t time_diff = std::llabs(ex.timestamp_us - nt.timestamp_us);
        if (time_diff <= W) {
          std::string ex_key = ex.symbol + ":" + std::to_string(ex.trade_id);

          // Check if this existing trade hasn't been emitted yet
          {
            std::lock_guard lk(emitted_mu);
            if (!already_emitted.count(ex_key)) {
              correlated_trades.push_back(ex);
              found_correlation = true;
            }
          }
        }
      }
    }

    // If we found correlations, emit ALL correlated trades (including the new
    // one) ONCE
    if (found_correlation) {
      std::lock_guard lk(emitted_mu);

      // Add the new trade to correlated set
      correlated_trades.push_back(nt);

      // Emit all correlated trades and mark them as emitted
      for (const auto &trade : correlated_trades) {
        std::string trade_key =
            trade.symbol + ":" + std::to_string(trade.trade_id);
        if (already_emitted.insert(trade_key).second) {
          uint64_t s = next_seq++;
          filt_q->push({s, trade});
          g_metrics.increment_filtered();
          total_trades_filtered.fetch_add(1, std::memory_order_relaxed);
        }
      }

      // Log the correlation
      std::cerr << "[CORRELATION] Found " << correlated_trades.size()
                << " correlated trades within " << W / 1000.0 << "ms: ";
      for (size_t i = 0; i < correlated_trades.size(); ++i) {
        std::cerr << correlated_trades[i].symbol << ":"
                  << correlated_trades[i].trade_id;
        if (i < correlated_trades.size() - 1)
          std::cerr << ", ";
      }
      std::cerr << std::endl;

      sym_mgr.update_corr(nt.symbol, true);
    } else {
      sym_mgr.update_corr(nt.symbol, false);
    }

    // Always add the new trade to the window for future correlations
    win[nt.symbol].push_back(nt);
  }

  std::cerr << "[MERGER] Stopping merger thread\n";
}

/* ──────────────────────────────────────────────────────────
 *   11) WORKER THREAD
 * ────────────────────────────────────────────────────────── */
void worker() {
  while (!stop_all.load() && !g_shutdown.should_shutdown()) {
    std::pair<uint64_t, Trade> it;
    if (!filt_q->pop(it))
      break;

    std::string m = compute_trade_metric(it.second.symbol, it.second.trade_id,
                                         it.second.time_str, it.second.side,
                                         it.second.size, it.second.price);
    auto end_processing = std::chrono::steady_clock::now();

    // Calculate end-to-end latency
    uint64_t e2e_latency =
        std::chrono::duration_cast<std::chrono::microseconds>(
            end_processing - it.second.received_at)
            .count();
    g_metrics.add_latency(e2e_latency);

    write_q->push({it.first, std::move(m)});
    g_metrics.increment_processed();
  }
}

/* ──────────────────────────────────────────────────────────
 *   12) WRITER THREAD
 * ────────────────────────────────────────────────────────── */
void writer() {
  uint64_t next = 0;
  std::map<uint64_t, std::string> buf;
  bool use_mmap = bool(hpw);
  std::ofstream ofs;

  if (!use_mmap) {
    ofs.open(g_config.output_file, std::ios::app);
    if (!ofs.is_open()) {
      std::cerr << "[ERROR] Failed to open output file: "
                << g_config.output_file << std::endl;
      return;
    }
  }

  std::cerr << "[WRITER] Starting writer thread, output: "
            << g_config.output_file << " (mmap=" << (use_mmap ? "yes" : "no")
            << ")\n";

  while (!stop_all.load() && !g_shutdown.should_shutdown()) {
    std::pair<uint64_t, std::string> p;
    if (!write_q->pop(p))
      break;
    buf.emplace(p.first, std::move(p.second));
    auto it = buf.find(next);
    while (it != buf.end()) {
      if (use_mmap) {
        if (!hpw->write_async(it->second, it->first)) {
          std::cerr << "[WARN] Writer ring buffer full, dropping write\n";
        }
      } else {
        ofs << it->second;
        ofs.flush();
      }
      g_metrics.increment_written();
      buf.erase(it);
      ++next;
      it = buf.find(next);
    }
  }

  if (ofs.is_open()) {
    ofs.close();
  }

  std::cerr << "[WRITER] Stopping writer thread\n";
}

/* ──────────────────────────────────────────────────────────
 *   13) ENHANCED MONITOR WITH JSON OUTPUT
 * ────────────────────────────────────────────────────────── */
void monitor() {
  std::cerr << "[MONITOR] Starting monitoring thread\n";

  while (!stop_all.load() && !g_shutdown.should_shutdown()) {
    std::this_thread::sleep_for(std::chrono::seconds(30));

    // Print human-readable stats using public getters
    uint64_t r = g_metrics.get_trades_received();
    uint64_t f = total_trades_filtered.load();
    uint64_t c = g_metrics.get_trades_processed();
    uint64_t w = g_metrics.get_trades_written();
    uint64_t lat_samples = g_metrics.get_latency_samples();
    double avg = lat_samples > 0 ? double(g_metrics.get_total_latency_us()) /
                                       lat_samples / 1000.0
                                 : 0.0;

    std::cerr << "[MONITOR] recv=" << r << " filt=" << f << " proc=" << c
              << " write=" << w << " avg_lat=" << avg << "ms"
              << " queues[trade=" << trade_q->size()
              << " filt=" << filt_q->size() << " write=" << write_q->size()
              << "]\n";

    // Print queue health
    std::cerr << "[QUEUES] drop_rates[trade=" << trade_q->drop_rate()
              << " filt=" << filt_q->drop_rate()
              << " write=" << write_q->drop_rate() << "]\n";

    // Print correlation window info
    std::cerr << "[CONFIG] correlation_window="
              << g_config.correlation_window_us << "µs ("
              << g_config.correlation_window_us / 1000.0 << "ms)\n";

    // Print symbol priorities
    sym_mgr.print();
  }

  std::cerr << "[MONITOR] Stopping monitoring thread\n";
}

/* ──────────────────────────────────────────────────────────
 *   14) SIGNAL HANDLER WITH GRACEFUL SHUTDOWN
 * ────────────────────────────────────────────────────────── */
void sig(int signal_num) {
  std::cerr << "[SIGNAL] Received signal " << signal_num << std::endl;
  g_shutdown.request_shutdown();
  stop_all = true;
}

/* ──────────────────────────────────────────────────────────
 *   15) MAIN WITH ENHANCED STARTUP AND SHUTDOWN
 * ────────────────────────────────────────────────────────── */
int main(int argc, char *argv[]) {
  std::cerr
      << "[STARTUP] Spec-Compliant Coinbase Trade Aggregator starting...\n";

  // Load configuration
  g_config.load_from_env();

  // Process command line arguments
  for (int i = 1; i < argc; ++i) {
    std::string arg = argv[i];
    if (arg == "--help" || arg == "-h") {
      std::cout
          << "Usage: " << argv[0] << " [options]\n"
          << "Options:\n"
          << "  --help, -h          Show this help\n"
          << "  --stats             Print final statistics in JSON format\n"
          << "Environment variables:\n"
          << "  POLL_INTERVAL_MS    Polling interval (default: 100)\n"
          << "  CORRELATION_WINDOW_US Correlation window (default: 5000 = "
             "5ms)\n"
          << "  OUTPUT_FILE         Output filename (default: "
             "trade_metrics.txt)\n"
          << "Note: Using real Coinbase Exchange API for trade data\n";
      return 0;
    }
  }

  // Initialize curl
  curl_global_init(CURL_GLOBAL_ALL);
  std::signal(SIGINT, sig);
  std::signal(SIGTERM, sig);

  // Initialize queues with configured sizes using unique_ptr
  trade_q = std::make_unique<SmartBoundedQueue<Trade>>(
      g_config.trade_queue_size, SmartBoundedQueue<Trade>::BACKPRESSURE);

  filt_q = std::make_unique<SmartBoundedQueue<std::pair<uint64_t, Trade>>>(
      g_config.filter_queue_size,
      SmartBoundedQueue<std::pair<uint64_t, Trade>>::DROP_NEWEST);

  write_q =
      std::make_unique<SmartBoundedQueue<std::pair<uint64_t, std::string>>>(
          g_config.write_queue_size,
          SmartBoundedQueue<std::pair<uint64_t, std::string>>::DROP_OLDEST);

  std::cerr << "[STARTUP] Queues initialized with sizes: trade="
            << g_config.trade_queue_size
            << " filter=" << g_config.filter_queue_size
            << " write=" << g_config.write_queue_size << "\n";

  // Initialize high-performance writer
  try {
    hpw = std::make_unique<HighPerformanceWriter>(
        g_config.output_file, g_config.mmap_initial_size_mb);
    std::cerr << "[STARTUP] High-performance mmap writer initialized\n";
  } catch (const std::exception &e) {
    std::cerr << "[WARN] Failed to initialize mmap writer: " << e.what()
              << ", falling back to file I/O\n";
    hpw.reset();
  }

  // Display configuration
  std::cerr << "[CONFIG] Correlation window: " << g_config.correlation_window_us
            << " µs (" << g_config.correlation_window_us / 1000.0 << " ms)\n";
  std::cerr << "[CONFIG] Poll interval: " << g_config.poll_interval_ms
            << " ms\n";
  std::cerr << "[CONFIG] Symbols: ";
  for (const auto &s : g_config.symbols) {
    std::cerr << s << " ";
  }
  std::cerr << "\n";

  // Start poller threads
  std::vector<std::thread> poll;
  for (const auto &s : g_config.symbols) {
    poll.emplace_back(poller, s);
    std::cerr << "[STARTUP] Started poller for " << s << std::endl;
  }

  // Start processing threads
  std::thread mrg(merger);
  unsigned wc = std::max(4u, std::thread::hardware_concurrency());
  std::vector<std::thread> w;
  for (unsigned i = 0; i < wc; ++i) {
    w.emplace_back(worker);
  }
  std::cerr << "[STARTUP] Started " << wc << " worker threads\n";

  std::thread wr(writer);
  std::thread mon(monitor);

  std::cerr << "[STARTUP] All threads started successfully\n";
  std::cerr << "[INFO] Using Coinbase Exchange API for real trade data\n";
  std::cerr << "[INFO] Looking for trades within "
            << g_config.correlation_window_us / 1000.0 << "ms of each other\n";

  // Wait for pollers to complete
  for (auto &t : poll) {
    t.join();
  }
  std::cerr << "[SHUTDOWN] All pollers stopped\n";

  // Graceful shutdown sequence
  g_shutdown.drain_queue(*trade_q, "trade");
  trade_q->shutdown();
  mrg.join();
  std::cerr << "[SHUTDOWN] Merger stopped\n";

  g_shutdown.drain_queue(*filt_q, "filter");
  filt_q->shutdown();
  for (auto &t : w) {
    t.join();
  }
  std::cerr << "[SHUTDOWN] All workers stopped\n";

  g_shutdown.drain_queue(*write_q, "write");
  write_q->shutdown();
  wr.join();
  std::cerr << "[SHUTDOWN] Writer stopped\n";

  stop_all = true;
  mon.join();
  std::cerr << "[SHUTDOWN] Monitor stopped\n";

  // Stop high-performance writer
  if (hpw) {
    hpw->stop();
    std::cerr << "[SHUTDOWN] High-performance writer stopped\n";
  }

  // Print final statistics
  std::cerr << "\n[FINAL STATS]\n" << g_metrics.get_json_stats() << std::endl;

  // Check for stats output request
  for (int i = 1; i < argc; ++i) {
    if (std::string(argv[i]) == "--stats") {
      std::cout << g_metrics.get_json_stats() << std::endl;
      break;
    }
  }

  curl_global_cleanup();
  std::cerr << "[SHUTDOWN] Clean shutdown completed\n";
  return 0;
}