From eacae27c917e64de01680a6f1022f3ee000031d2 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Thu, 27 Aug 2026 14:17:39 +0300 Subject: [PATCH] fix(intrade-bar): finalize elapsed live bars Finalize initialized live bars from the platform process cycle when their timeframe bucket has elapsed, even when no next-bucket tick arrives. Preserve source precision metadata, ignore late ticks for finalized buckets, and document the callback lifecycle contract. --- guides/api-and-header-contracts.md | 8 +- guides/platform-api-guide.md | 7 +- .../MarketDataSubscriptionManager.hpp | 86 ++++++++++-- .../intrade_bar_api_response_test.cpp | 126 +++++++++++++++--- 4 files changed, 196 insertions(+), 31 deletions(-) diff --git a/guides/api-and-header-contracts.md b/guides/api-and-header-contracts.md index bace461c..c1b84885 100644 --- a/guides/api-and-header-contracts.md +++ b/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 diff --git a/guides/platform-api-guide.md b/guides/platform-api-guide.md index ad8b7572..42ea041e 100644 --- a/guides/platform-api-guide.md +++ b/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 diff --git a/include/optionx_cpp/platforms/IntradeBarPlatform/MarketDataSubscriptionManager.hpp b/include/optionx_cpp/platforms/IntradeBarPlatform/MarketDataSubscriptionManager.hpp index 0f6933dc..7737754e 100644 --- a/include/optionx_cpp/platforms/IntradeBarPlatform/MarketDataSubscriptionManager.hpp +++ b/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 bar_batches; { std::lock_guard lock(m_mutex); + const auto now_ms = static_cast(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(subscription.timeframe) * + time_shield::MS_PER_SEC; + if (timeframe_ms == 0 || + state.current.time_ms > + std::numeric_limits::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); diff --git a/tests/intrade_bar_api/intrade_bar_api_response_test.cpp b/tests/intrade_bar_api/intrade_bar_api_response_test.cpp index 6741bfde..4b111bd5 100644 --- a/tests/intrade_bar_api/intrade_bar_api_response_test.cpp +++ b/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(timeframe_sec) * time_shield::MS_PER_SEC; + const auto now_ms = static_cast(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 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,7 +1069,7 @@ 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); @@ -1069,13 +1077,13 @@ TEST(IntradeBarApiResponses, IntradeBarBarSubscriptionAggregatesTickUpdates) { 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 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 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 delivered_batches; + market_data::MarketDataSubscriptionResult result; + + platform.on_bar_data() = + [&delivered_batches](std::unique_ptr 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 first_event_ticks; + first_event_ticks.push_back(std::move(first_batch)); + platform.event_bus().notify_async( + std::make_unique( + 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 late_event_ticks; + late_event_ticks.push_back(std::move(late_batch)); + platform.event_bus().notify_async( + std::make_unique( + 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 later_event_ticks; + later_event_ticks.push_back(std::move(later_batch)); + platform.event_bus().notify_async( + std::make_unique( + 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);