feat: Aggregator hubs for quote/tick quantization with duplicate handling #1875
Conversation
|
Important Review skippedBot user detected. To trigger a single review, invoke the You can disable this status message by setting the
📝 WalkthroughWalkthroughThis PR introduces a streaming aggregation feature that converts incoming market data streams (quotes and ticks) into larger time-based OHLCV bars. The implementation includes QuoteAggregatorHub for aggregating quote streams, TickAggregatorHub for converting tick streams to quote bars, and TickHub for managing raw tick data. Both aggregators support configuration by PeriodSize or TimeSpan, with optional gap-filling that carries forward the last known price. The feature includes comprehensive test coverage for various aggregation scenarios, late arrivals, and gap handling. Supporting infrastructure includes a new ITick interface, Tick record, and a protected visibility change to NotifyObserversOnAdd for derived classes. Assessment against linked issues
Out-of-scope changes
Possibly related PRs
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Co-authored-by: DaveSkender <8432125+DaveSkender@users.noreply.github.com>
Co-authored-by: DaveSkender <8432125+DaveSkender@users.noreply.github.com>
Signed-off-by: GitHub <noreply@github.com>
DaveSkender
changed the title
DaveSkender
changed the title
|
@copilot Determine if:
|
Analysis of both options: Option 1: Integrate into QuoteHub
Option 2: Separate TickAggregatorHub
Recommendation: Keep This maintains clean separation: Should I proceed with the current approach or implement one of these alternatives? |
|
@copilot Proceed with your recommendation |
Confirmed. The current |
DaveSkender
changed the title
|
@copilot I assumed your recommendation including the addition of the TickAggregatorHub ... please add it. |
Co-authored-by: DaveSkender <8432125+DaveSkender@users.noreply.github.com>
Added TickAggregatorHub in commit 5f1186b. Implementation includes:
Architecture maintains clean separation:
|
|
@copilot Ensure aggregator hubs handle duplicates, out of sequence and prior bar late arrivals. For quotes, dups have identical timestamp and replace prior. For ticks, dups are only assessed if unique execution id is included. |
|
Nice!
This feat would allow me to use hubs end-to-end! TBH, I could live without really-real-time updates to Close, but I do:
|
DaveSkender
temporarily deployed
to
stockindicators.dev
GitHub Actions
Inactive
|
Flow was: Flow became: Allowing: This wasn't working last night, but I think I just forgot to subscribe my higher-timeframe quotehubs to the aggregators. |
@JGronholz - Yeah, still just poking Copilot to do chunks of this. If you wanted to work on this one to get it production-ready, please do (and let me know).
|
DaveSkender
temporarily deployed
to
stockindicators.dev
GitHub Actions
Inactive
|
Under "catalog" conditions: this works perfectly. Typical real-world conditions: A typical quoteprovider (quotehub fed by kline feed) should not encounter these issues (updates reliably every 1-10 seconds). My issue:
The 5-minute klines only update about once every 10 seconds, so it's bizarre that I'm experiencing what appears to be concurrency issues... That's what prompted the test PR, which shows the aggregator hubs function perfectly. It is a threading issue in my code. The The summary document is supposed to be copy-and-pasteable for an agent to implement without being able to build. |
|
Continued in: |
|
PR #1904 fixes QuoteAggregatorHub. TickAggregatorHub test suite fails. |
|
|
@JGronholz I take it you gave up on this one? I'm trying to figure out if I need to pick it up to finish. |
Signed-off-by: Dave Skender <8432125+DaveSkender@users.noreply.github.com>
Signed-off-by: Dave Skender <8432125+DaveSkender@users.noreply.github.com>
|
@coderabbitai full review |
✅ Actions performedFull review triggered. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Fix all issues with AI agents
In `@src/_common/Quotes/Quote.AggregatorHub.cs`:
- Around line 86-113: The OnAdd implementation mutates shared instance state and
the non-thread-safe Cache without synchronization; add a private readonly lock
object (e.g., _addLock) to the class and wrap the critical sections in OnAdd —
including reads/writes of _currentBar and _currentBarTimestamp, the
Cache.IndexGte lookup + bounds check + Cache[existingIndex] = updatedBar
assignment, and the NotifyObserversOnRebuild call — inside a lock to make these
operations atomic; also apply equivalent locking around modifications to
_processedExecutionIds (Tick.AggregatorHub.cs) such as the
_processedExecutionIds.Add() call to prevent concurrent races.
In `@src/_common/Quotes/Tick.AggregatorHub.cs`:
- Around line 9-11: The _processedExecutionIds HashSet will grow unbounded;
replace it with a structure that records insertion time (e.g.,
Dictionary<string,DateTime> or ConcurrentDictionary<string,DateTime>) and
implement pruning logic in the aggregator (for example inside the method
handling new ticks, e.g., OnAdd or AddTick) to remove entries older than a
configurable time window or when a max size threshold is exceeded; ensure
thread-safety by using the existing lock/monitor or a concurrent collection, add
a configurable TimeSpan/MaxEntries field, and prune either lazily on inserts or
via a background timer to prevent memory exhaustion while still preventing
reprocessing of recent ExecutionIds.
In `@src/_common/Quotes/Tick.StreamHub.cs`:
- Around line 6-162: Add missing unit tests for TickHub to cover standalone vs
provider-backed initialization (constructors), the OnAdd behavior for
same-timestamp updates including exact duplicates, ticks without ExecutionId
(replace in cache and NotifyObserversOnAdd), ticks with same ExecutionId
(replace and NotifyObserversOnRebuild), ticks with different ExecutionId
(treated as late arrival/rebuild), and Rebuild when _isStandalone (ensure
RollbackState(fromTimestamp) and NotifyObserversOnRebuild(fromTimestamp) are
invoked); target the TickHub class and its public members OnAdd, Rebuild,
ToIndicator and constructors to validate cache state changes and observer
notifications in each scenario.
🧹 Nitpick comments (7)
src/_common/Quotes/Tick.AggregatorHub.cs (1)
94-96: Remove unused local variableisCurrentBar.The variable is computed but never referenced. The logic at line 166 implicitly relies on the
elsebranch afterisFutureBar, makingisCurrentBarredundant.♻️ Proposed fix
// Determine if this is for current bar, future bar, or past bar - bool isCurrentBar = _currentBar != null && barTimestamp == _currentBarTimestamp; bool isFutureBar = _currentBar == null || barTimestamp > _currentBarTimestamp; bool isPastBar = _currentBar != null && barTimestamp < _currentBarTimestamp;src/_common/Quotes/Tick.StreamHub.cs (1)
44-46: Redundant null check.The null-coalescing throw on line 44 already handles the null case. The subsequent
ArgumentNullException.ThrowIfNullon line 46 is unreachable.♻️ Proposed fix
public TickHub( IStreamObservable<ITick> provider) : base(provider ?? throw new ArgumentNullException(nameof(provider))) { - ArgumentNullException.ThrowIfNull(provider); - _isStandalone = false; Name = "TICK-HUB"; Reinitialize(); }src/_common/Quotes/Quote.AggregatorHub.cs (3)
82-84: Remove unused local variableisCurrentBar.The variable
isCurrentBaris assigned but never read. It's implicitly covered by theelsebranch at line 156 when neitherisPastBarnorisFutureBaris true.♻️ Proposed fix
// Determine if this is for current bar, future bar, or past bar - bool isCurrentBar = _currentBar != null && barTimestamp == _currentBarTimestamp; bool isFutureBar = _currentBar == null || barTimestamp > _currentBarTimestamp; bool isPastBar = _currentBar != null && barTimestamp < _currentBarTimestamp;
115-143: Gap filling implementation is correct but has an unused variable.The gap-filling logic correctly iterates from the last bar to the new bar timestamp, creating zero-volume bars with carried-forward close prices. However,
gapIndexon line 134 is unused.♻️ Proposed fix to remove unused variable
// Add gap bar using base class logic - (IQuote gapResult, int gapIndex) = ToIndicator(gapBar, null); + (IQuote gapResult, _) = ToIndicator(gapBar, null); AppendCache(gapResult, notify);
145-173: Future bar handling has an unused variable; current bar update logic is correct.The logic for starting a new bar and updating the current bar in-place is correct: it preserves the original open, updates high/low/close/volume, and notifies observers via rebuild. However,
indexon line 153 is unused.♻️ Proposed fix to remove unused variable
// Start a new bar _currentBar = CreateOrUpdateBar(null, barTimestamp, item); _currentBarTimestamp = barTimestamp; // Use base class to add the new bar - (IQuote result, int index) = ToIndicator(_currentBar, indexHint); + (IQuote result, _) = ToIndicator(_currentBar, indexHint); AppendCache(result, notify);tests/indicators/_common/Quotes/Tick.AggregatorHub.Tests.cs (2)
278-312: Good test for duplicate detection, with noted TODO for correction semantics.The test validates that duplicate execution IDs are skipped. The TODO comment on line 303 is a valid concern—if the second tick with the same
ExecutionIdis a correction, the current "skip duplicates" behavior may not be appropriate. This aligns with PR discussion about duplicate handling semantics.Would you like me to open an issue to track the correction-vs-duplicate semantics for execution IDs, as noted in the TODO?
3-5: Consider adding negative test cases for invalid configurations.The
QuoteAggregatorHubTestsinclude tests for invalidPeriodSize.Month,TimeSpan.Zero, and negativeTimeSpan. For test parity and comprehensive edge case coverage, consider adding similar tests forTickAggregatorHub. Based on learnings: "Cover edge cases: empty input, minimum/maximum values, boundary conditions in tests."Example tests to add:
[TestMethod] public void InvalidPeriodSize_Month_ThrowsException() { TickHub provider = new(); FluentActions .Invoking(() => provider.ToTickAggregatorHub(PeriodSize.Month)) .Should() .ThrowExactly<ArgumentException>() .WithMessage("*Month aggregation is not supported*"); provider.EndTransmission(); } [TestMethod] public void InvalidTimeSpan_Zero_ThrowsException() { TickHub provider = new(); FluentActions .Invoking(() => provider.ToTickAggregatorHub(TimeSpan.Zero)) .Should() .ThrowExactly<ArgumentOutOfRangeException>() .WithMessage("*must be greater than zero*"); provider.EndTransmission(); }
| // Handle late arrival for past bar | ||
| if (isPastBar) | ||
| { | ||
| // Find the existing bar in cache | ||
| int existingIndex = Cache.IndexGte(barTimestamp); | ||
| if (existingIndex >= 0 && existingIndex < Cache.Count && Cache[existingIndex].Timestamp == barTimestamp) | ||
| { | ||
| // Update existing past bar | ||
| IQuote existingBar = Cache[existingIndex]; | ||
| Quote updatedBar = new( | ||
| Timestamp: barTimestamp, | ||
| Open: existingBar.Open, // Keep original open | ||
| High: Math.Max(existingBar.High, item.High), | ||
| Low: Math.Min(existingBar.Low, item.Low), | ||
| Close: item.Close, // Update close | ||
| Volume: existingBar.Volume + item.Volume); | ||
|
|
||
| Cache[existingIndex] = updatedBar; | ||
|
|
||
| // Trigger rebuild from this timestamp | ||
| if (notify) | ||
| { | ||
| NotifyObserversOnRebuild(barTimestamp); | ||
| } | ||
| } | ||
|
|
||
| return; | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# First, find and examine the StreamHub base class and related implementations
fd -t f "StreamHub\.cs|QuoteProvider\.cs" src/_common/
# Also check the Quote.AggregatorHub.cs file structure
fd -t f "Quote\.AggregatorHub\.cs" src/Repository: DaveSkender/Stock.Indicators
Length of output: 393
🏁 Script executed:
#!/bin/bash
# Search for lock statements in StreamHub and Quotes directories
rg -n "lock\s*\(" src/_common/StreamHub/ src/_common/Quotes/ || echo "No locks found"
# Also search for Monitor, Interlocked, and other synchronization patterns
rg -n "Monitor\.|Interlocked\.|ReaderWriterLock" src/_common/StreamHub/ src/_common/Quotes/ || echo "No explicit sync patterns found"Repository: DaveSkender/Stock.Indicators
Length of output: 182
🏁 Script executed:
#!/bin/bash
# Check the Quote.AggregatorHub.cs file to understand the full context
wc -l src/_common/Quotes/Quote.AggregatorHub.cs
cat -n src/_common/Quotes/Quote.AggregatorHub.cs | head -120Repository: DaveSkender/Stock.Indicators
Length of output: 5353
🏁 Script executed:
#!/bin/bash
# Check what the Cache object is and how it's defined
rg -n "Cache\s*=" src/_common/Quotes/Quote.AggregatorHub.cs -B 3 -A 1
rg -n "class.*Cache" src/_common/ -t csRepository: DaveSkender/Stock.Indicators
Length of output: 54
🏁 Script executed:
#!/bin/bash
# Check the QuoteProvider base class implementation
cat -n src/_common/StreamHub/Providers/QuoteProvider.cs | head -100Repository: DaveSkender/Stock.Indicators
Length of output: 865
🏁 Script executed:
#!/bin/bash
# Check the StreamHub base class and look for synchronization
cat -n src/_common/StreamHub/StreamHub.cs | head -150Repository: DaveSkender/Stock.Indicators
Length of output: 5268
🏁 Script executed:
#!/bin/bash
# Search for Cache definition and type
rg -n "Cache\s*:" src/_common/StreamHub/ -A 2 -B 2Repository: DaveSkender/Stock.Indicators
Length of output: 54
🏁 Script executed:
#!/bin/bash
# Check if Cache is a thread-safe collection or has locks
rg -n "class.*Cache|List<|Dictionary<" src/_common/StreamHub/StreamHub.cs -B 2 -A 2 | head -50Repository: DaveSkender/Stock.Indicators
Length of output: 481
🏁 Script executed:
#!/bin/bash
# Check if there are any tests or documentation about concurrent OnAdd calls
rg -n "concurrent|thread|lock|Thread|Task\.Run" src/_common/StreamHub/ -i | head -30
# Also check for any synchronization patterns in OnAdd override methods
rg -n "void OnAdd" src/_common/ -A 20 -B 2 | head -100Repository: DaveSkender/Stock.Indicators
Length of output: 8345
🏁 Script executed:
#!/bin/bash
# Look for Observer pattern and Notify implementations to understand concurrency expectations
cat -n src/_common/StreamHub/StreamHub.Observer.csRepository: DaveSkender/Stock.Indicators
Length of output: 3092
🏁 Script executed:
#!/bin/bash
# Check for any PR comments or TODO comments about concurrency/thread safety
rg -n "TODO|FIXME|concurrent|thread.safe|race|atomic|lock" src/_common/Quotes/Quote.AggregatorHub.cs -iRepository: DaveSkender/Stock.Indicators
Length of output: 54
Add synchronization for concurrent OnAdd calls.
The late arrival handling mutates Cache[existingIndex] and instance state (_currentBar, _currentBarTimestamp) without synchronization. The base class provides no locking for OnAdd—only the Unsubscribe() method uses _unsubscribeLock, and that is specifically for preventing race conditions between EndTransmission and OnCompleted.
Cache is a plain List<TOut> (not thread-safe), and concurrent calls to OnAdd from multiple threads will cause data races:
- Lines 82–84: Reading
_currentBarand_currentBarTimestampwithout locks - Line 90:
Cache.IndexGte()lookup is not atomic with the bounds check (line 91) and mutation (line 103) - Line 103:
Cache[existingIndex] = updatedBaris not protected
Per coding guidelines: "streaming hubs must isolate instance state." Add a synchronization lock (e.g., private readonly object _addLock = new()) around the entire OnAdd method or use lock statements to protect shared state mutations. The same issue exists in Tick.AggregatorHub.cs line 88 where _processedExecutionIds.Add() is called without protection.
🤖 Prompt for AI Agents
In `@src/_common/Quotes/Quote.AggregatorHub.cs` around lines 86 - 113, The OnAdd
implementation mutates shared instance state and the non-thread-safe Cache
without synchronization; add a private readonly lock object (e.g., _addLock) to
the class and wrap the critical sections in OnAdd — including reads/writes of
_currentBar and _currentBarTimestamp, the Cache.IndexGte lookup + bounds check +
Cache[existingIndex] = updatedBar assignment, and the NotifyObserversOnRebuild
call — inside a lock to make these operations atomic; also apply equivalent
locking around modifications to _processedExecutionIds (Tick.AggregatorHub.cs)
such as the _processedExecutionIds.Add() call to prevent concurrent races.
| private Quote? _currentBar; | ||
| private DateTime _currentBarTimestamp; | ||
| private readonly HashSet<string> _processedExecutionIds = []; |
There was a problem hiding this comment.
Unbounded memory growth in _processedExecutionIds.
The HashSet<string> grows indefinitely as ticks are processed but is never pruned. For long-running streams, this will cause memory exhaustion.
Consider implementing a pruning strategy, such as:
- Clearing entries older than a configurable time window
- Using an LRU cache with a maximum size
- Pruning when the set exceeds a threshold
💡 Example: Simple time-based pruning approach
Track the timestamp when each ExecutionId was added, and periodically remove entries older than the aggregation window:
- private readonly HashSet<string> _processedExecutionIds = [];
+ private readonly Dictionary<string, DateTime> _processedExecutionIds = [];Then in OnAdd, prune old entries when adding new ones, or implement a separate cleanup mechanism.
🤖 Prompt for AI Agents
In `@src/_common/Quotes/Tick.AggregatorHub.cs` around lines 9 - 11, The
_processedExecutionIds HashSet will grow unbounded; replace it with a structure
that records insertion time (e.g., Dictionary<string,DateTime> or
ConcurrentDictionary<string,DateTime>) and implement pruning logic in the
aggregator (for example inside the method handling new ticks, e.g., OnAdd or
AddTick) to remove entries older than a configurable time window or when a max
size threshold is exceeded; ensure thread-safety by using the existing
lock/monitor or a concurrent collection, add a configurable TimeSpan/MaxEntries
field, and prune either lazily on inserts or via a background timer to prevent
memory exhaustion while still preventing reprocessing of recent ExecutionIds.
| public class TickHub | ||
| : StreamHub<ITick, ITick>, IStreamObservable<ITick> | ||
| { | ||
| /// <summary> | ||
| /// Indicates whether this TickHub is standalone (no external provider). | ||
| /// </summary> | ||
| private readonly bool _isStandalone; | ||
|
|
||
| /// <summary> | ||
| /// Initializes a new instance of the <see cref="TickHub"/> class without its own provider. | ||
| /// </summary> | ||
| /// <param name="maxCacheSize">Maximum in-memory cache size.</param> | ||
| public TickHub(int? maxCacheSize = null) | ||
| : base(new BaseProvider<ITick>()) | ||
| { | ||
| _isStandalone = true; | ||
|
|
||
| const int maxCacheSizeDefault = (int)(0.9 * int.MaxValue); | ||
|
|
||
| if (maxCacheSize is not null and > maxCacheSizeDefault) | ||
| { | ||
| string message | ||
| = $"'{nameof(maxCacheSize)}' must be less than {maxCacheSizeDefault}."; | ||
|
|
||
| throw new ArgumentOutOfRangeException( | ||
| nameof(maxCacheSize), maxCacheSize, message); | ||
| } | ||
|
|
||
| MaxCacheSize = maxCacheSize ?? maxCacheSizeDefault; | ||
| Name = "TICK-HUB"; | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Initializes a new instance of the <see cref="TickHub"/> class with a specified provider. | ||
| /// </summary> | ||
| /// <param name="provider">The tick provider.</param> | ||
| public TickHub( | ||
| IStreamObservable<ITick> provider) | ||
| : base(provider ?? throw new ArgumentNullException(nameof(provider))) | ||
| { | ||
| ArgumentNullException.ThrowIfNull(provider); | ||
|
|
||
| _isStandalone = false; | ||
| Name = "TICK-HUB"; | ||
| Reinitialize(); | ||
| } | ||
|
|
||
| /// <inheritdoc/> | ||
| protected override (ITick result, int index) | ||
| ToIndicator(ITick item, int? indexHint) | ||
| { | ||
| ArgumentNullException.ThrowIfNull(item); | ||
|
|
||
| int index = indexHint | ||
| ?? Cache.IndexGte(item.Timestamp); | ||
|
|
||
| return (item, index == -1 ? Cache.Count : index); | ||
| } | ||
|
|
||
| /// <inheritdoc/> | ||
| public override string ToString() | ||
| => $"TICKS: {Cache.Count} items"; | ||
|
|
||
| /// <summary> | ||
| /// Handles adding a new tick with special handling for same-timestamp updates | ||
| /// when TickHub is standalone (no external provider). | ||
| /// </summary> | ||
| /// <inheritdoc/> | ||
| public override void OnAdd(ITick item, bool notify, int? indexHint) | ||
| { | ||
| // for non-standalone TickHub, use standard behavior | ||
| if (!_isStandalone) | ||
| { | ||
| base.OnAdd(item, notify, indexHint); | ||
| return; | ||
| } | ||
|
|
||
| // get result and position | ||
| (ITick result, int index) = ToIndicator(item, indexHint); | ||
|
|
||
| // check if this is a same-timestamp update (not a new item at the end) | ||
| if (Cache.Count > 0 && index < Cache.Count && Cache[index].Timestamp == result.Timestamp) | ||
| { | ||
| // check if this is an exact duplicate (same values) | ||
| // if so, defer to AppendCache for overflow tracking | ||
| if (Cache[index].Equals(result)) | ||
| { | ||
| AppendCache(result, notify); | ||
| return; | ||
| } | ||
|
|
||
| // For ticks without execution IDs, replace in cache but notify as addition | ||
| // This allows aggregators to process multiple ticks at the same timestamp | ||
| bool hasExecutionId = !string.IsNullOrEmpty(result.ExecutionId); | ||
|
|
||
| if (!hasExecutionId) | ||
| { | ||
| // Replace existing tick in cache (keep only latest) | ||
| Cache[index] = result; | ||
|
|
||
| // Notify observers as if it's a new addition | ||
| // This allows aggregators to incorporate the tick's data | ||
| if (notify) | ||
| { | ||
| NotifyObserversOnAdd(result, index); | ||
| } | ||
|
|
||
| return; | ||
| } | ||
|
|
||
| // For ticks with execution IDs, replace and trigger rebuild | ||
| if (Cache[index].ExecutionId == result.ExecutionId) | ||
| { | ||
| // replace existing item at this position (same execution ID) | ||
| Cache[index] = result; | ||
|
|
||
| // notify observers to rebuild from this timestamp | ||
| if (notify) | ||
| { | ||
| NotifyObserversOnRebuild(result.Timestamp); | ||
| } | ||
|
|
||
| return; | ||
| } | ||
|
|
||
| // Different execution IDs - this shouldn't happen often, | ||
| // but treat as late arrival and trigger rebuild | ||
| } | ||
|
|
||
| // standard add behavior for new items | ||
| AppendCache(result, notify); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Rebuilds the cache from a specific timestamp. | ||
| /// For standalone TickHub, preserves cache and notifies observers. | ||
| /// </summary> | ||
| /// <inheritdoc/> | ||
| public override void Rebuild(DateTime fromTimestamp) | ||
| { | ||
| // for standalone TickHub (no external provider), | ||
| // we cannot rebuild from an empty provider cache | ||
| // instead, just notify observers to rebuild from this hub's cache | ||
| if (_isStandalone) | ||
| { | ||
| // rollback internal state | ||
| RollbackState(fromTimestamp); | ||
|
|
||
| // notify observers to rebuild from this hub | ||
| NotifyObserversOnRebuild(fromTimestamp); | ||
| return; | ||
| } | ||
|
|
||
| // standard rebuild for TickHub with external provider | ||
| base.Rebuild(fromTimestamp); | ||
| } | ||
| } |
There was a problem hiding this comment.
Missing test coverage for TickHub.
Per the pipeline failure, tests/indicators/_common/Quotes/Tick.StreamHub.Tests.cs is missing. The TickHub class has complex logic for standalone mode, same-timestamp updates, and rebuild behavior that warrants dedicated test coverage.
Do you want me to help generate test cases covering:
- Standalone vs provider-backed initialization
- Same-timestamp update scenarios (with/without ExecutionId)
- Rebuild behavior for standalone mode
🤖 Prompt for AI Agents
In `@src/_common/Quotes/Tick.StreamHub.cs` around lines 6 - 162, Add missing unit
tests for TickHub to cover standalone vs provider-backed initialization
(constructors), the OnAdd behavior for same-timestamp updates including exact
duplicates, ticks without ExecutionId (replace in cache and
NotifyObserversOnAdd), ticks with same ExecutionId (replace and
NotifyObserversOnRebuild), ticks with different ExecutionId (treated as late
arrival/rebuild), and Rebuild when _isStandalone (ensure
RollbackState(fromTimestamp) and NotifyObserversOnRebuild(fromTimestamp) are
invoked); target the TickHub class and its public members OnAdd, Rebuild,
ToIndicator and constructors to validate cache state changes and observer
notifications in each scenario.
|
@codex review |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 8e7ab8c9e0
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| // but treat as late arrival and trigger rebuild | ||
| } | ||
|
|
||
| // standard add behavior for new items | ||
| AppendCache(result, notify); |
There was a problem hiding this comment.
In standalone TickHub, when a tick arrives with the same timestamp but a different ExecutionId, this branch falls through to AppendCache (lines 132-136). AppendCache triggers Rebuild when the timestamp is not strictly greater than the last cached tick (see StreamHub.AppendCache), but TickHub.Rebuild for standalone only notifies observers and never inserts the new tick into the cache. That means valid trades sharing a timestamp are silently dropped, so downstream aggregations under-count volume and miss price updates whenever multiple executions occur in the same time slice. Consider explicitly appending/storing these ticks before notifying rebuild or supporting multiple ticks per timestamp.
Useful? React with 👍 / 👎.
| Open: existingBar.Open, // Keep original open | ||
| High: Math.Max(existingBar.High, quote.High), | ||
| Low: Math.Min(existingBar.Low, quote.Low), | ||
| Close: quote.Close, // Always use latest close | ||
| Volume: existingBar.Volume + quote.Volume); |
There was a problem hiding this comment.
CreateOrUpdateBar always aggregates by taking max/min and adding volume for every quote in the same aggregate bucket, with no handling for updated input bars that share the same timestamp. If an upstream feed resends an in-progress 1‑minute bar (common for streaming candles), the aggregate bar will add the volume again and can only move highs/lows outward, producing inflated OHLCV for the larger period. You likely need to detect duplicate input timestamps and rebuild/replace that sub-bar instead of accumulating it again.
Useful? React with 👍 / 👎.
Implements streaming aggregation of tick/trade quotes into time-period bars with optional gap filling and duplicate handling. Addresses scenarios where quote feeds skip periods (e.g., Alpaca BTC minutely data with no trade volume) and handles duplicate data intelligently.
Implementation
QuoteAggregatorHub
Extends
QuoteProvider<IQuote, IQuote>with incremental OHLC bar building for bar-to-bar aggregation:PeriodSizeenum values or customTimeSpanOnAddto detect period boundaries and update/create bars in-placeTickAggregatorHub
Extends
QuoteProvider<ITick, IQuote>with incremental OHLC bar building for tick-to-bar aggregation:ITickinterface for raw market tick data (Price, Volume, Timestamp, ExecutionId)Tickrecord implementation with optionalExecutionIdfor duplicate detectionTickHubfor managing tick streamsPeriodSizeenum values or customTimeSpanExecutionId(when provided) to filter duplicate ticksDuplicate Handling
ExecutionIdwhen provided; tracks processed IDs to skip duplicatesGap Filling
Optional carry-forward mechanism available in both aggregators:
StreamHub Integration
Both aggregators are fully chainable with downstream indicators:
Architecture
Clean separation of concerns enables flexible aggregation pipelines:
Usage
Tick-to-Bar Aggregation with Duplicate Detection
Bar-to-Bar Aggregation
Note: Month aggregation unsupported in streaming mode (use batch
Aggregate()utility). Streaming builds the current bar incrementally; final bar is always in-progress until next period starts.Original prompt
💡 You can make Copilot smarter by setting up custom instructions, customizing its development environment and configuring Model Context Protocol (MCP) servers. Learn more Copilot coding agent tips in the docs.