[codex] Finalize elapsed Intrade live bars

guides/api-and-header-contracts.md

@@ -194,10 +194,10 @@ Contract rules:
   a local time series should upsert by `(provider_id, subscription_id, symbol,
   timeframe, time_ms)` until a `FINALIZED` payload for the same key arrives.
   Appending every incomplete snapshot as a new candle will create duplicate bars.
-- Tick-driven live bar aggregation finalizes a bar when the first tick from the
-  next timeframe bucket arrives. If the stream becomes silent, the latest bar can
-  remain `INCOMPLETE`. Future work: add timer/process-based finalization as a
-  separate change.
+- Live bar aggregation finalizes a bar when the first tick from the next
+  timeframe bucket arrives or when platform `process()` observes that the
+  current bucket has elapsed. The process-time path does not require a later
+  tick and emits the final snapshot through the normal `on_bar_data()` callback.
 - `on_market_data_status()` is a separate stream-status callback. Data callbacks
   should carry data batches, not connection lifecycle sentinel payloads.
 - `on_market_data_status()` is a stream-level event bus, not a per-subscription

guides/platform-api-guide.md

@@ -115,9 +115,10 @@ Subscription rules:
 - Live bar streams can deliver several `INCOMPLETE` snapshots with the same
   `(provider_id, subscription_id, symbol, timeframe, time_ms)` key before the
   final `FINALIZED` snapshot. Treat them as upserts, not append-only candles.
-- Tick-driven bar streams finalize the current bar only when a tick from the next
-  timeframe bucket arrives. Timer/process-based finalization is tracked as
-  future work.
+- Live bar streams finalize the current bar when a tick from the next timeframe
+  bucket arrives or when a platform `process()` cycle observes that the bucket
+  has elapsed. With `run(false)`, callers must keep pumping `process()` for this
+  timer-based final snapshot to be delivered.
 - `MarketDataContinuityService` routes recovered historical bars into the same
   `BarDataBatch` pipeline and marks them as `HISTORICAL`/`BACKFILL`.
 - `BaseMarketDataProvider` is non-copyable and non-movable so provider identity

include/optionx_cpp/platforms/IntradeBarPlatform/MarketDataSubscriptionManager.hpp

@@ -110,6 +110,8 @@ namespace optionx::platforms::intrade_bar {
             Bar current; ///< Current in-progress bar.
             std::uint64_t first_tick_time_ms = 0; ///< Earliest tick timestamp in the current bar bucket.
             std::uint64_t last_tick_time_ms = 0; ///< Latest tick timestamp in the current bar bucket.
+            std::uint32_t price_digits = 0; ///< Price precision inherited from the source stream.
+            std::uint32_t volume_digits = 0; ///< Volume precision inherited from the source stream.
             bool initialized = false; ///< True after the first valid tick was applied.
         };
 
@@ -170,7 +172,8 @@ namespace optionx::platforms::intrade_bar {
         /// \brief Finds or creates a pending bar delivery batch.
         market_data::BarDataBatch& pending_bar_batch_for(
                 const market_data::MarketDataSubscriptionHandle& subscription,
-                const events::TickUpdateBatch& source_batch);
+                std::uint32_t price_digits,
+                std::uint32_t volume_digits);
 
         /// \brief Removes queued tick data for an inactive subscription.
         /// \pre The caller holds m_mutex.
@@ -191,6 +194,13 @@ namespace optionx::platforms::intrade_bar {
         /// \brief Returns the event timestamp used for live bar bucketing.
         static std::uint64_t tick_time_ms(const Tick& tick) noexcept;
 
+        /// \brief Finalizes an in-progress bar after its wall-clock bucket elapsed.
+        static bool finalize_elapsed_bar(
+                const market_data::MarketDataSubscriptionHandle& subscription,
+                std::uint64_t now_ms,
+                BarAggregationState& state,
+                Bar& finalized);
+
         /// \brief Applies one tick to a live bar accumulator.
         static void append_bar_updates(
                 const market_data::MarketDataSubscriptionHandle& subscription,
@@ -614,6 +624,26 @@ namespace optionx::platforms::intrade_bar {
         std::vector<market_data::BarDataBatch> bar_batches;
         {
             std::lock_guard<std::mutex> lock(m_mutex);
+            const auto now_ms = static_cast<std::uint64_t>(OPTIONX_TIMESTAMP_MS);
+            for (const auto& [id, subscription] : m_bar_subscriptions) {
+                auto state_it = m_bar_states.find(id);
+                if (state_it == m_bar_states.end()) continue;
+
+                Bar finalized;
+                if (!finalize_elapsed_bar(
+                        subscription,
+                        now_ms,
+                        state_it->second,
+                        finalized)) {
+                    continue;
+                }
+
+                auto& batch = pending_bar_batch_for(
+                    subscription,
+                    state_it->second.price_digits,
+                    state_it->second.volume_digits);
+                batch.items.push_back(std::move(finalized));
+            }
             tick_batches.swap(m_pending_tick_batches);
             bar_batches.swap(m_pending_bar_batches);
         }
@@ -819,8 +849,13 @@ namespace optionx::platforms::intrade_bar {
                 if (subscription.symbol != normalized_symbol) continue;
                 if (!source_matches_subscription(subscription, event.source())) continue;
 
-                auto& batch = pending_bar_batch_for(subscription, source_batch);
                 auto& state = m_bar_states[subscription.id];
+                state.price_digits = source_batch.price_digits;
+                state.volume_digits = source_batch.volume_digits;
+                auto& batch = pending_bar_batch_for(
+                    subscription,
+                    state.price_digits,
+                    state.volume_digits);
                 for (const auto& tick : source_batch.items) {
                     append_bar_updates(subscription, tick, state, batch.items);
                 }
@@ -852,7 +887,8 @@ namespace optionx::platforms::intrade_bar {
     inline market_data::BarDataBatch&
     MarketDataSubscriptionManager::pending_bar_batch_for(
             const market_data::MarketDataSubscriptionHandle& subscription,
-            const events::TickUpdateBatch& source_batch) {
+            std::uint32_t price_digits,
+            std::uint32_t volume_digits) {
         for (auto& batch : m_pending_bar_batches) {
             if (batch.subscription.id == subscription.id) {
                 return batch;
@@ -864,8 +900,8 @@ namespace optionx::platforms::intrade_bar {
         created.type = market_data::MarketDataType::BARS;
         created.symbol = subscription.symbol;
         created.timeframe = subscription.timeframe;
-        created.price_digits = source_batch.price_digits;
-        created.volume_digits = source_batch.volume_digits;
+        created.price_digits = price_digits;
+        created.volume_digits = volume_digits;
         m_pending_bar_batches.push_back(std::move(created));
         return m_pending_bar_batches.back();
     }
@@ -950,6 +986,35 @@ namespace optionx::platforms::intrade_bar {
         return tick.time_ms != 0 ? tick.time_ms : tick.received_ms;
     }
 
+    inline bool MarketDataSubscriptionManager::finalize_elapsed_bar(
+            const market_data::MarketDataSubscriptionHandle& subscription,
+            std::uint64_t now_ms,
+            BarAggregationState& state,
+            Bar& finalized) {
+        if (!state.initialized ||
+            state.current.has_flag(MarketDataFlags::FINALIZED) ||
+            subscription.timeframe <= 0) {
+            return false;
+        }
+
+        const auto timeframe_ms =
+            static_cast<std::uint64_t>(subscription.timeframe) *
+            time_shield::MS_PER_SEC;
+        if (timeframe_ms == 0 ||
+            state.current.time_ms >
+                std::numeric_limits<std::uint64_t>::max() - timeframe_ms) {
+            return false;
+        }
+
+        const auto bucket_end_ms = state.current.time_ms + timeframe_ms;
+        if (now_ms < bucket_end_ms) return false;
+
+        state.current.set_flag(MarketDataFlags::INCOMPLETE, false);
+        state.current.set_flag(MarketDataFlags::FINALIZED);
+        finalized = state.current;
+        return true;
+    }
+
     inline void MarketDataSubscriptionManager::append_bar_updates(
             const market_data::MarketDataSubscriptionHandle& subscription,
             const Tick& tick,
@@ -970,12 +1035,17 @@ namespace optionx::platforms::intrade_bar {
         const auto bucket_ms = (timestamp_ms / timeframe_ms) * timeframe_ms;
         const auto price_type = market_price_type_from_bar_price_source(subscription.price_source);
 
-        if (state.initialized && bucket_ms < state.current.time_ms) {
-            return;
+        if (state.initialized) {
+            if (bucket_ms < state.current.time_ms) return;
+            if (bucket_ms == state.current.time_ms &&
+                state.current.has_flag(MarketDataFlags::FINALIZED)) {
+                return;
+            }
         }
 
         if (!state.initialized || state.current.time_ms != bucket_ms) {
-            if (state.initialized) {
+            if (state.initialized &&
+                !state.current.has_flag(MarketDataFlags::FINALIZED)) {
                 state.current.set_flag(MarketDataFlags::INCOMPLETE, false);
                 state.current.set_flag(MarketDataFlags::FINALIZED);
                 updates.push_back(state.current);

tests/intrade_bar_api/intrade_bar_api_response_test.cpp

@@ -91,6 +91,13 @@ events::TickUpdateBatch make_market_data_batch(
         tick.volume_digits);
 }
 
+std::uint64_t current_bar_bucket_ms(std::int64_t timeframe_sec = 60) {
+    const auto timeframe_ms =
+        static_cast<std::uint64_t>(timeframe_sec) * time_shield::MS_PER_SEC;
+    const auto now_ms = static_cast<std::uint64_t>(OPTIONX_TIMESTAMP_MS);
+    return (now_ms / timeframe_ms) * timeframe_ms;
+}
+
 void publish_account_status(
         IntradeBarPlatform& platform,
         AccountUpdateStatus status) {
@@ -1027,12 +1034,13 @@ TEST(IntradeBarApiResponses, IntradeBarBarSubscriptionAggregatesTickUpdates) {
     EXPECT_EQ(result.subscription.stream_type, market_data::MarketDataType::BARS);
     EXPECT_EQ(result.subscription.timeframe, 60);
 
+    const auto bucket_ms = current_bar_bucket_ms() + time_shield::MS_PER_MIN;
     auto first_batch = make_market_data_batch("EURUSD", 1.10000, 1.10020);
-    first_batch.items[0].time_ms = 120000;
+    first_batch.items[0].time_ms = bucket_ms;
     auto second_batch = make_market_data_batch("EURUSD", 1.10040, 1.10060);
-    second_batch.items[0].time_ms = 121000;
+    second_batch.items[0].time_ms = bucket_ms + 1000;
     auto third_batch = make_market_data_batch("EURUSD", 1.09980, 1.10000);
-    third_batch.items[0].time_ms = 180000;
+    third_batch.items[0].time_ms = bucket_ms + time_shield::MS_PER_MIN;
 
     std::vector<events::TickUpdateBatch> event_ticks;
     event_ticks.push_back(std::move(first_batch));
@@ -1050,7 +1058,7 @@ TEST(IntradeBarApiResponses, IntradeBarBarSubscriptionAggregatesTickUpdates) {
     EXPECT_EQ(delivered_batch.subscription.id, result.subscription.id);
 
     const auto& first_update = delivered_batch.items[0];
-    EXPECT_EQ(first_update.time_ms, 120000u);
+    EXPECT_EQ(first_update.time_ms, bucket_ms);
     EXPECT_DOUBLE_EQ(first_update.open, 1.10010);
     EXPECT_DOUBLE_EQ(first_update.high, 1.10010);
     EXPECT_DOUBLE_EQ(first_update.low, 1.10010);
@@ -1061,21 +1069,21 @@ TEST(IntradeBarApiResponses, IntradeBarBarSubscriptionAggregatesTickUpdates) {
     EXPECT_EQ(first_update.price_type(), MarketPriceType::MID);
 
     const auto& second_update = delivered_batch.items[1];
-    EXPECT_EQ(second_update.time_ms, 120000u);
+    EXPECT_EQ(second_update.time_ms, bucket_ms);
     EXPECT_DOUBLE_EQ(second_update.open, 1.10010);
     EXPECT_DOUBLE_EQ(second_update.high, 1.10050);
     EXPECT_DOUBLE_EQ(second_update.low, 1.10010);
     EXPECT_DOUBLE_EQ(second_update.close, 1.10050);
     EXPECT_TRUE(second_update.has_flag(MarketDataFlags::INCOMPLETE));
 
     const auto& finalized = delivered_batch.items[2];
-    EXPECT_EQ(finalized.time_ms, 120000u);
+    EXPECT_EQ(finalized.time_ms, bucket_ms);
     EXPECT_DOUBLE_EQ(finalized.close, 1.10050);
     EXPECT_TRUE(finalized.has_flag(MarketDataFlags::FINALIZED));
     EXPECT_FALSE(finalized.has_flag(MarketDataFlags::INCOMPLETE));
 
     const auto& next_bar = delivered_batch.items[3];
-    EXPECT_EQ(next_bar.time_ms, 180000u);
+    EXPECT_EQ(next_bar.time_ms, bucket_ms + time_shield::MS_PER_MIN);
     EXPECT_DOUBLE_EQ(next_bar.open, 1.09990);
     EXPECT_DOUBLE_EQ(next_bar.close, 1.09990);
     EXPECT_TRUE(next_bar.has_flag(MarketDataFlags::INCOMPLETE));
@@ -1110,12 +1118,13 @@ TEST(IntradeBarApiResponses, IntradeBarBarSubscriptionIgnoresPreviousBucketTicks
 
     ASSERT_TRUE(result);
 
+    const auto bucket_ms = current_bar_bucket_ms() + time_shield::MS_PER_MIN;
     auto first_batch = make_market_data_batch("EURUSD", 1.10000, 1.10020);
-    first_batch.items[0].time_ms = 120000;
+    first_batch.items[0].time_ms = bucket_ms;
     auto next_batch = make_market_data_batch("EURUSD", 1.10040, 1.10060);
-    next_batch.items[0].time_ms = 180000;
+    next_batch.items[0].time_ms = bucket_ms + time_shield::MS_PER_MIN;
     auto late_batch = make_market_data_batch("EURUSD", 1.20000, 1.20020);
-    late_batch.items[0].time_ms = 120500;
+    late_batch.items[0].time_ms = bucket_ms + 500;
 
     std::vector<events::TickUpdateBatch> event_ticks;
     event_ticks.push_back(std::move(first_batch));
@@ -1128,10 +1137,12 @@ TEST(IntradeBarApiResponses, IntradeBarBarSubscriptionIgnoresPreviousBucketTicks
 
     ASSERT_EQ(bar_callback_count, 1);
     ASSERT_EQ(delivered_batch.items.size(), 3u);
-    EXPECT_EQ(delivered_batch.items[0].time_ms, 120000u);
-    EXPECT_EQ(delivered_batch.items[1].time_ms, 120000u);
+    EXPECT_EQ(delivered_batch.items[0].time_ms, bucket_ms);
+    EXPECT_EQ(delivered_batch.items[1].time_ms, bucket_ms);
     EXPECT_TRUE(delivered_batch.items[1].has_flag(MarketDataFlags::FINALIZED));
-    EXPECT_EQ(delivered_batch.items[2].time_ms, 180000u);
+    EXPECT_EQ(
+        delivered_batch.items[2].time_ms,
+        bucket_ms + time_shield::MS_PER_MIN);
     EXPECT_DOUBLE_EQ(delivered_batch.items[2].open, 1.10050);
     EXPECT_DOUBLE_EQ(delivered_batch.items[2].close, 1.10050);
     EXPECT_DOUBLE_EQ(delivered_batch.items[2].high, 1.10050);
@@ -1167,10 +1178,11 @@ TEST(IntradeBarApiResponses, IntradeBarBarSubscriptionKeepsCloseFromLatestTickIn
 
     ASSERT_TRUE(result);
 
+    const auto bucket_ms = current_bar_bucket_ms() + time_shield::MS_PER_MIN;
     auto later_batch = make_market_data_batch("EURUSD", 1.10040, 1.10060);
-    later_batch.items[0].time_ms = 121000;
+    later_batch.items[0].time_ms = bucket_ms + 1000;
     auto earlier_batch = make_market_data_batch("EURUSD", 1.10000, 1.10020);
-    earlier_batch.items[0].time_ms = 120000;
+    earlier_batch.items[0].time_ms = bucket_ms;
 
     std::vector<events::TickUpdateBatch> event_ticks;
     event_ticks.push_back(std::move(later_batch));
@@ -1183,7 +1195,7 @@ TEST(IntradeBarApiResponses, IntradeBarBarSubscriptionKeepsCloseFromLatestTickIn
     ASSERT_EQ(bar_callback_count, 1);
     ASSERT_EQ(delivered_batch.items.size(), 2u);
     const auto& updated = delivered_batch.items[1];
-    EXPECT_EQ(updated.time_ms, 120000u);
+    EXPECT_EQ(updated.time_ms, bucket_ms);
     EXPECT_DOUBLE_EQ(updated.open, 1.10010);
     EXPECT_DOUBLE_EQ(updated.close, 1.10050);
     EXPECT_DOUBLE_EQ(updated.high, 1.10050);
@@ -1193,6 +1205,88 @@ TEST(IntradeBarApiResponses, IntradeBarBarSubscriptionKeepsCloseFromLatestTickIn
     platform.shutdown();
 }
 
+TEST(IntradeBarApiResponses, IntradeBarBarSubscriptionFinalizesElapsedBucketWithoutNextTick) {
+    IntradeBarPlatform platform;
+    platform.run(false);
+    std::vector<market_data::BarDataBatch> delivered_batches;
+    market_data::MarketDataSubscriptionResult result;
+
+    platform.on_bar_data() =
+        [&delivered_batches](std::unique_ptr<market_data::BarDataBatch> batch) {
+            if (batch) {
+                delivered_batches.push_back(std::move(*batch));
+            }
+        };
+
+    ASSERT_TRUE(platform.subscribe_bars(
+        market_data::BarSubscriptionRequest(
+            "EUR/USD",
+            60,
+            BarPriceSource::MID,
+            market_data::MarketDataTransport::POLLING),
+        [&result](market_data::MarketDataSubscriptionResult subscription_result) {
+            result = std::move(subscription_result);
+        }));
+    ASSERT_TRUE(result);
+
+    const auto elapsed_bucket_ms =
+        current_bar_bucket_ms() - time_shield::MS_PER_MIN;
+    auto first_batch = make_market_data_batch("EURUSD", 1.10000, 1.10020);
+    first_batch.items[0].time_ms = elapsed_bucket_ms;
+    first_batch.price_digits = 5;
+    first_batch.volume_digits = 2;
+    std::vector<events::TickUpdateBatch> first_event_ticks;
+    first_event_ticks.push_back(std::move(first_batch));
+    platform.event_bus().notify_async(
+        std::make_unique<events::PriceUpdateEvent>(
+            std::move(first_event_ticks)));
+    pump_platform(platform);
+
+    ASSERT_EQ(delivered_batches.size(), 1u);
+    ASSERT_EQ(delivered_batches[0].items.size(), 2u);
+    EXPECT_EQ(delivered_batches[0].price_digits, 5u);
+    EXPECT_EQ(delivered_batches[0].volume_digits, 2u);
+    const auto& incomplete = delivered_batches[0].items[0];
+    const auto& finalized = delivered_batches[0].items[1];
+    EXPECT_TRUE(incomplete.has_flag(MarketDataFlags::INCOMPLETE));
+    EXPECT_FALSE(incomplete.has_flag(MarketDataFlags::FINALIZED));
+    EXPECT_FALSE(finalized.has_flag(MarketDataFlags::INCOMPLETE));
+    EXPECT_TRUE(finalized.has_flag(MarketDataFlags::FINALIZED));
+
+    auto late_batch = make_market_data_batch("EURUSD", 1.20000, 1.20020);
+    late_batch.items[0].time_ms = elapsed_bucket_ms + 500;
+    std::vector<events::TickUpdateBatch> late_event_ticks;
+    late_event_ticks.push_back(std::move(late_batch));
+    platform.event_bus().notify_async(
+        std::make_unique<events::PriceUpdateEvent>(
+            std::move(late_event_ticks)));
+    pump_platform(platform);
+
+    EXPECT_EQ(delivered_batches.size(), 1u);
+
+    pump_platform(platform);
+    EXPECT_EQ(delivered_batches.size(), 1u);
+
+    const auto later_bucket_ms =
+        elapsed_bucket_ms + (2 * time_shield::MS_PER_MIN);
+    auto later_batch = make_market_data_batch("EURUSD", 1.10100, 1.10120);
+    later_batch.items[0].time_ms = later_bucket_ms + 500;
+    std::vector<events::TickUpdateBatch> later_event_ticks;
+    later_event_ticks.push_back(std::move(later_batch));
+    platform.event_bus().notify_async(
+        std::make_unique<events::PriceUpdateEvent>(
+            std::move(later_event_ticks)));
+    pump_platform(platform);
+
+    ASSERT_EQ(delivered_batches.size(), 2u);
+    ASSERT_EQ(delivered_batches[1].items.size(), 1u);
+    const auto& later_bar = delivered_batches[1].items[0];
+    EXPECT_EQ(later_bar.time_ms, later_bucket_ms);
+    EXPECT_TRUE(later_bar.has_flag(MarketDataFlags::INCOMPLETE));
+    EXPECT_FALSE(later_bar.has_flag(MarketDataFlags::FINALIZED));
+    platform.shutdown();
+}
+
 TEST(IntradeBarApiResponses, IntradeBarBarSubscriptionClearsPendingUpdatesOnUnsubscribe) {
     IntradeBarPlatform platform;
     platform.run(false);