v3.0.0 preview | Streaming indicators and quotes

[DaveSkender]

We’ll be releasing a series of progressive pre-release preview versions to evaluate and stabilize features for streaming indicators from live WebSockets and other incremental quote sources. As we evolve our approach, your feedback is appreciated. Runnable examples are available.

Warning
We expect possibly 5-10 preview pre-release versions before we have a sufficiently stable core to make an official v3 package. These preview versions are all experimental and will be quite volatile with breaking changes and conventions.

[mo3in]

What you think about using Rx.net for streaming data and processing indicators?

[DaveSkender]

I did look at Rx.NET when working on #824. I left some notes there. Seems like a solid library overall.

It’s probably better for people using this library than for us to use it internal to the library. The native C# .NET observables are easy enough to use and seem to work well for what we’re doing, so it’s not needed. For us and the features we need, Rx.NET appeared to only add a layer of complexity and performance costs.

With that said, I don’t know it extremely well. What are your thoughts on it?

[DaveSkender]

I was just checking out the new v6 for Rx.NET (aka reactive extensions). Much of the core reactive observable framework is part of .NET core. From what I can tell, you’d really only need these extensions for advanced use cases. It’s an excellent overall framework for handling modern data streams, worth exploring, and will be fully compatible with v3 of our library.

[mo3in]

Yes, but some of extensions likes Buffer, Window, Groupby[Until] can useful.

And What your Opponent about Subjects?
They have different use cases and is it possible to use them?

[DaveSkender]

I don’t have any opposition to any of those specific functions and may take another look. My only hesitation for use inside this indicator library isn’t about utility, it’s more about need and performance costs that often come with “helper” libraries like these.

Many of the basic elements in the Rx.NET library are similar to LINQ in it has useful, simpler developer syntax that’s excellent for 99% of its intended users. If small performance costs aren't going to make a difference in your external code, I'd recommend using it.

Given our general guiding principles for this library, I’d consider adding some of these if there are valuable pass-through usability benefits to end users to offset these small performance costs.

Generally speaking, for a base math-style library like ours, it’s often better for end users when we internally use the ugly native C# syntax than the simpler developer-friendly ones from third-party libraries. For example, using a for loop to iteratively sum an array is faster than the LINQ myarray.Sum() extension, which has no residual end user functionality benefits, so we use the uglier, faster looping technique.

[DaveSkender]

I’ll probably upgrade to use Rx.NET in my own personal trading system for processing and interacting with this indicators library. ML.NET is another that users seem to like as well, in the .NET ecosystem.

[elpht]

Hi, I'm right now trying out v3.0.0 preview and wanted to ask whether there is or will be an option to limit the size of the data collections held by both, observable and observer, as I didn't find any yet. Thanks!

[DaveSkender]

The idea of these collections growing in an uncontrolled manner has crossed my mind. Since these are instantiated classes, I think we can add an optional setting so you can supply a max size that'll trigger some auto-pruning after new data arrives. Makes sense.

• auto-prune in-memory stream data #1095

[elAndyG]

Just wanted to share a note, for "determine best approach for handling out of sequence quotes", or the "auto-aggregator", you may need a parameter to determine what to do with missing candles. I found that in premarket or sometimes just after open, some candles will be missing because of extremely limited volume.

I believe the correct way to do it is to use the last available value as that candle's value. For instance, if you have candles for 10:01, 10:02, 10:04, 10:05, the missing 10:03 could potentially mess up any calculations. So 10:03 can just have the same value as 10:02. Or, the avg between 10:02 and 10:04 is another approach ive seen.

great library! and great work!

[DaveSkender]

Yes, I think this'll be interesting to figure out. And yes, probably part of an optional and configurable auto-aggregator utility tool. I think most users will want visibility and control over anything that creates magic numbers through extrapolation or interpolations.

Inside an aggregate window / current period

Inside a current aggregate window is relatively easy since you're just doing min/max/first/last, resend. We handle this "update of last" well already, but do need to insert the new aggregator into that process.

• quote-stream auto-aggregator #1093

Outside a current window (historical)

The challenging part here is when, outside of a current period, there's a late historical change, once that has increasingly larger impact the further back it goes. This can happen with an upstream correction or similarly for splits / dividend adjustments, an even more insidious impact.

For example, say you're streaming all day long with a single quote streams and 100 indicators and 30 with 2 layers of chaining. If an old or updated quote comes in for 500 periods ago you now have a rather big situation where there's potentially large and complex chain of cleanup work, $500(100+30\times2)=80,000$ in this case. It's possible to do, but if this happens frequently in large datasets, you can imagine the spiking loads that can occur.

I say potential, because at some point it leans more towards moot than useful, it's ancient history a user may not want to address at all. The option to do this may fall into the same settings category as @elpht recommends.

• handle out-of-sequence late-arrival quotes #1094

[MhAllan]

Let me share some modeling with you, I did for series of data I had before

This design should help you write indicators for real time series very easy and quickly, and should handle series size, and improve performance by reducing arrays allocation

Expand for detail and code samples

Cache methods results, this interface for caching methods using their names and an object that represent the method parameters

public interface IMethodCacher
{
    bool TryGetFromCache<TRet>(string key, object parameters, out TRet result);
    void SetCache<TRet>(string key, object parameters, TRet value);
    TRet ExecuteAndCache<TRet>(object obj, Func<TRet> func, [CallerMemberName] string methodName = "");
    Task<TRet> ExecuteAndCacheAsync<TRet>(object parameters, Func<Task<TRet>> func, [CallerMemberName] string methodName = "");
}

Implement it, I chose concurrent dictionary here

Two dictionaries one for the method name, the other inside it is for that method parameters

public abstract class MethodCacher : IMethodCacher
{
    readonly ConcurrentDictionary<string, ConcurrentDictionary<object, object>> _c =
        new ConcurrentDictionary<string, ConcurrentDictionary<object, object>>();

    public bool TryGetFromCache<TRet>(string key, object parameters, out TRet result)
    {
        result = default;
        if (!_c.TryGetValue(key, out var dic))
            return false;
        if (!dic.TryGetValue(parameters, out var obj))
            return false;

        result = (TRet)obj;
        return true;
    }

    public void SetCache<TRet>(string key, object parameters, TRet value)
    {
        if (!_c.TryGetValue(key, out var dic))
            dic = _c.AddOrUpdate(key, new ConcurrentDictionary<object, object>(), (k, v) => v);

        dic.AddOrUpdate(parameters, value, (k, v) => v);
    }

    public async Task<TRet> ExecuteAndCacheAsync<TRet>(object parameters, Func<Task<TRet>> func,
        [CallerMemberName] string caller = "")
    {
        if (!_c.TryGetValue(caller, out var dic))
        {
            dic = _c.AddOrUpdate(caller, new ConcurrentDictionary<object, object>(), (k, v) => v);
        }

        if (!dic.TryGetValue(parameters, out var result))
        {
            result = await func();
            dic.AddOrUpdate(parameters, result, (k, v) => v);
        }

        return (TRet)result;
    }

    public TRet ExecuteAndCache<TRet>(object obj, Func<TRet> func, [CallerMemberName] string caller = "")
    {
        if(!_c.TryGetValue(caller, out var dic))
        {
            dic = _c.AddOrUpdate(caller, new ConcurrentDictionary<object, object>(), (k, v) => v);
        }

        if(!dic.TryGetValue(obj, out var result))
        {
            result = func();
            dic.AddOrUpdate(obj, result, (k, v) => v);
        }

        return (TRet)result;
    }

    public bool TryGetFromCache<TRet>(string key, out TRet result) => 
        TryGetFromCache(key, new MethodCacheParams(), out  result);

    public void SetCache<TRet>(string key, TRet value) =>
        SetCache(key, new MethodCacheParams());
}

Now extend it, so it looks like method call

Define IEquatable to compare parameters and work as dictionary key for the method parameters, go for up to 6 parameters

record struct MethodCacheParams();
record struct MethodCacheParams<T>(T a) where T : struct, INumber<T>;
record struct MethodCacheParams<T1, T2>(T1 a1, T2 a2)
   where T1 : struct, INumber<T1>
   where T2 : struct, INumber<T2>;

record struct MethodCacheParams<T1, T2, T3>(T1 a1, T2 a2, T3 a3)
   where T1 : struct, INumber<T1>
   where T2 : struct, INumber<T2>
   where T3 : struct, INumber<T3>;

// and so on until T6

Write the extensions

public static class MethodCacherExtensions
{
public static TRet ExecuteAndCache<TRet>(this IMethodCacher mc, Func<TRet> func, [CallerMemberName] string caller = "") 
    => mc.ExecuteAndCache(new MethodCacheParams(), func, caller);

public static TRet ExecuteAndCache<TRet, T1>(this IMethodCacher mc, T1 a1, Func<TRet> func, [CallerMemberName] string caller = "") where T1 : struct, INumber<T1>
    => mc.ExecuteAndCache(new MethodCacheParams<T1>(a1), func, caller);

public static TRet ExecuteAndCache<TRet, T1, T2>(this IMethodCacher mc, T1 a1, T2 a2, Func<TRet> func, [CallerMemberName] string caller = "")
    where T1 : struct, INumber<T1>
    where T2 : struct, INumber<T2>
    => mc.ExecuteAndCache(new MethodCacheParams<T1, T2>(a1, a2), func, caller);

// and so on until T6
. . . 

Usage example, this is how the results holder will look like

static void Main(string[] args)
{
    var obj = new MyObj();
    obj.X = "Hello world!";

    var x = obj.GetX(1, 2, 3);
    Console.WriteLine(x);

    obj.X = "Good bye";

    x = obj.GetX(2, 3, 4);
    Console.WriteLine(x);

    x = obj.GetX(1, 2, 3);
    Console.WriteLine(x);

    Console.ReadLine();
}

public class MyObj : MethodCacher
{
    public string X { get; set; }

    public MyResult GetX(int a, int b, int c)
    {
        var func = () => new MyResult(X);

        return this.ExecuteAndCache(a, b, c, func);
    }
}

Implement series navigation

Define series

This model will have

• _isCapped: should be set to a positive number in real time series, it will cap the size the of the collection, once above the cap, old items will be deleted
• Prev: give the previous item of a given item
• EnumeratePrev(): enumerate previous items of a given item

public interface IReadOnlySeries<T> : IReadOnlyList<T>
{
    IEnumerable<T> EnumeratePrevious(T item, bool includeItem);
    IEnumerable<T> EnumerateNext(T item, bool includeItem);
    T Prev(T item);
    T Next(T item);
    int IndexOf(T item);
}

public interface ISeries<T> : IReadOnlySeries<T> 
{
    void Add(T item);
}

public class Series<T> : ISeries<T>
{
    readonly int _maxItems;
    readonly List<T> _items;
    readonly bool _isCapped;
    readonly object _lock;

    public Series(int max = -1)
    {
        _maxItems = max;
        _isCapped = max > 0;
        if(_isCapped)
        {
            _items = new List<T>(max);
            _lock = new object();
        }
        else
        {
            _items = new List<T>();
        }
    }

    public T this[int index] => _items[index];
    public int Count => _items.Count;

    public void Add(T item)
    {
        if(_lock != null)
            Monitor.Enter(_lock);
        try
        {
            if (_isCapped && _items.Count == _maxItems)
            {
                var first = _items[0];
                _items.RemoveAt(0);
                OnItemOutdated(first);
            }
            
            _items.Add(item);
            OnItemAdded(item);
        }
        finally
        {
            if(_lock != null)
                Monitor.Exit(_lock);
        }
    }

    public int IndexOf(T item) => _items.IndexOf(item);

    public IEnumerator<T> GetEnumerator() => _items.GetEnumerator();

    IEnumerator IEnumerable.GetEnumerator() => _items.GetEnumerator();

    public IEnumerable<T> EnumeratePrevious(T item, bool includeItem)
    {
        var index = _items.IndexOf(item);
        index += includeItem ? 1 : 0;
        return _items.Take(index);
    }

    public IEnumerable<T> EnumerateNext(T item, bool includeItem)
    {
        var index = _items.IndexOf(item);
        index += includeItem ? 0 : 1;
        return _items.Skip(index);
    }

    public T Prev(T item)
    {
        var index = _items.IndexOf(item);
        if (index <= 0)
            return default;
        return _items[index - 1];
    }

    public T Next(T item)
    {
        var index = _items.IndexOf(item);
        if (index < 0 || index >= _items.Count - 1)
            return default;
        return _items[index + 1];
    }

    protected virtual void OnItemAdded(T item) { }
    protected virtual void OnItemOutdated(T item) { }
}

Define the candle model

This model will have

• Index: gives an idea where are you in the series, Index + 1 is the count of item an indicator is interested in when calculating for this candle
• Prev: go to previous candle and access it's indicator results using the MethodCacher mechanism above
• EnumeratePrev(): allows you to get the enumeration of all candles in the series that are not outdated (if the series is capped) and prior to this candle

public interface ICandle
{
    public int Index { get; } // => _series.IndexOf(this);
    public ICandle Prev(); // => _series.Prev(this);
    public ICandle Next(); // => _series.Next(this);
    public IEnumerable<ICandle> EnumeratePrev(bool includeSelf = true); // => _series.EnumeratePrevious(this, includeSelf);
    public IEnumerable<ICandle> EnumerateNext(bool includeSelf = true); // => _series.EnumerateNext(this, includeSelf);
    public void SetSeries(ISeriece<ICandle> series);

Create the candles series, this will hold candles, set a cap so old ones will be removed, and allow them to navigate by Next() and Prev() methods

public interface ICandleSeries : ISeries<ICandle>
{
}

public class CandleSeries : Series<ICandle>, ICandleSeries
{
    public CandleSeries(int max) : base(max)
    {
    }

    protected override void OnItemAdded(ICandle c)
    {
        c.SetSeries(this); // so in the Indicators we can do c.Prev() and c.Next()
 . . .

Implement the indicators using cachable methods that return indicators results, and implement the navigation in the candle as well. All in the Candle model.

public class Candle : MethodCacher, ICandle
{
     ICandleSeries _series;
    //cached method
    public EmaResult GetEma(int period)
    {
        var func = () => EmaCalculation(period);
        
        //use method cacher
        return ExecuteAndCache(period, func);
    }

    private EmaResult EmaCalculation(int perdiod)
    {
         //get previous item
        var prev = this.Prev();
        if(prev == null) 
                  . . . 
         var prevEma = prev.GetEma(period) // This will come from memory, not calculated

        //Calculate
    }

   //series methods
    public void SetSeries(ICandleSeries series) => _series = series;
    public ICandle Prev() => _series.Prev(this);
    public ICandle Next() => _series.Next(this);
    public IEnumerable<ICandle> EnumeratePrev(bool includeSelf = true) => _series.EnumeratePrevious(this, includeSelf);
}

Or write an Indicator

public static class SomeIndicator
{
        public static SomeResult Calculate(ICandle c, int p1, double p2, ...)
       {
              //prev candles
              var prevCandles = c.EnumeratePrev().TakeLast(p1) ...

              //prev
              var prev = c.Prev();

              //prev of prev
              var prev_prev = c.Prev().Prev();

               //cached result
               var r = prev.GetResult(p1, p2, ...) //cached according to method name and parameters.

              // and so on.
       }
}

Hope it helps, I might have copied inconsistent pieces, let me know if it is not clear.

[DaveSkender]

Nice. Thanks for the contributions. I'll have to digest this a bit when I get back into my open PR for the next preview version. This is timely and potentially helpful, but I'll have to review and understand it better first.

[MhAllan]

You welcome, I edited it with more explanation. if you have questions reach me.

[DaveSkender]

This was helpful food for thought, to help me consider some different approaches. Thank you.

collapsing / hiding due to length and age

[kagudimov]

Hi there! This might be a bit late, but I really liked your library and wanted to share an idea.

A small refactoring of the indicators using yield return could open up the possibility of using the same indicator implementation for both batch and incremental data processing.

Here’s an example of how SMA and EMA could look when implemented with yield return (for simplicity, I’m omitting the initialization period):

public static IEnumerable<SmaResult> CalcSma(this IEnumerable<double> data, int period)
{
    if (period <= 0)
        throw new ArgumentException("Period must be greater than zero.", nameof(period));

    double sum = 0;
    Queue<double> window = new Queue<double>(period);

    foreach (var item in data)
    {
        if (window.Count == period)
            sum -= window.Dequeue();
        window.Enqueue(item);

        sum += item;

        double sma = sum / window.Count;

        yield return new SmaResult(sma);
    }
}

public record SmaResult(double Sma);

public static IEnumerable<EmaResult> CalcEma(this IEnumerable<double> data, int period)
{
    if (period <= 0)
        throw new ArgumentException("Period must be greater than zero.", nameof(period));

    double multiplier = 2.0 / (period + 1);

    var sma = new Incrementor<double, SmaResult>(s => CalcSma(s, period));

    double ema = 0;
    int i = 0;
    foreach (var item in data)
    {
        if (i++ < period)
        {
            ema = sma.CalcNext(item).Sma;
        }
        else
        {
            ema += (item - ema) * multiplier;
        }

        yield return new EmaResult(ema);
    }
}

public record EmaResult(double Ema);

Example of batch processing:

List<double> BatchExample(List<double> data, int period)
{
    var result = data.CalcSma(period)
        .Select(x => x.Sma)
        .ToList();
    return result;
}

Example of incremental processing:

void IncrementalExample(int period)
{
    var ema = new Incrementor<double, EmaResult>(s => s.CalcEma(period));

    while(true)
    {
        double x = GetNextValue();

        double currentEma = ema.CalcNext(x).Ema;
        // Do something with currentEma
    }
}

The transformation of a method that accepts an IEnumerable into an incremental one is made possible by a helper class called Incrementor, which works with a special "collection" that always contains just two elements: the current and the next.

Such a class could be implemented like this, for example:

internal class Incrementor<TSource, TValue> where TSource : notnull
{
    private readonly TASingleItemDataSource<TSource> _source = new TASingleItemDataSource<TSource>();
    private readonly IEnumerator<TValue> _result;

    public TValue Current => _result.Current;

    public Incrementor(Func<IEnumerable<TSource>, IEnumerable<TValue>> calc)
    {
        _result = calc(_source).GetEnumerator();
    }

    public TValue CalcNext(TSource value)
    {
        _source.SetNext(value);
        _result.MoveNext();
        return _result.Current;
    }
}

internal class TASingleItemDataSource<TValue> : IEnumerable<TValue> where TValue : notnull
{
    private readonly Enumerator _enumerator = new();

    public void SetNext(TValue item)
    {
        _enumerator.SetNext(item);
    }

    public IEnumerator<TValue> GetEnumerator()
    {
        return _enumerator;
    }

    IEnumerator IEnumerable.GetEnumerator()
    {
        return GetEnumerator();
    }

    private class Enumerator : IEnumerator<TValue>
    {
        private bool _hasNext;
        private TValue? _next;
        private bool _hasCurrent;
        private TValue? _current;

        public TValue Current => _hasCurrent ? _current! : throw new InvalidOperationException("No value is set");

        object IEnumerator.Current => Current;

        public void SetNext(TValue item)
        {
            if (_hasNext)
                throw new InvalidOperationException("Next item is already set.");
            _next = item;
            _hasNext = true;
        }

        public bool MoveNext()
        {
            if (_hasNext)
            {
                _current = _next;
                _hasCurrent = true;
                _next = default;
                _hasNext = false;
                return true;
            }
            return false;
        }

        public void Dispose()
        {
        }

        public void Reset()
        {
            throw new NotImplementedException();
        }
    }
}

I believe that the performance of both batch and incremental processing would remain acceptable. At the same time, the code would stay readable and easy to maintain.

[DaveSkender]

Thank you. I've def been thinking about this yield return approach. I've added it to our backlog:

• yield return for IEnumerable indicator results #1376
• similar to conversation in heap allocation optimalisation #1323 (comment)

[El-Chacal]

Hello @DaveSkender , I've been testing the streaming features lately and I've been struggling with a particular scenario where a quote being added to a QuoteHub is still live and will be updated as new ticks arrive (example if you are aggregating data for the current day into a 1-day quote). I couldn't find any straightforward scenario to access the quote and override it with the new values except removing it and inserting a new updated instance of it in the QuoteHub instance. but this causes any indicator subscribed to the QuoteHub to generate an index out of range exception immediately. are you familiar with the issue or am I doing things wrong ?

[DaveSkender]

I know what you're saying. There's two things I'm designing in with the streaming hubs in mind. And it's good to understand that what I'm calling a "quote" is really an aggregate "bar" that doesn't need to be re-aggregated. Taking in ticks or trade data and doing your own aggregation is really an entirely separate processing step.

1. Hubs will innately self-correct if a repeat (corrections) or out-of-order quote bars arrives -- you don't need to remove this yourself before replacing it, just Add(quote). Note: this does not aggregate quotes it just takes them as there are.
2. We also have an aggregator utility in mind (not yet implemented), which could be something like a QuoteAggregatorHub that you'd put in front of the QuoteHub, or just a configurable extension of it, to essentially wait and collect ticks or trades until it fully forms a time bar before releasing the aggregate to the hub. This could also release as its building that bar without waiting -- supplying the same bar date, which is handled by the hubs normally as a "replace last" kind of update.

[DaveSkender]

With that said, I do need to look into your scenario more closely -- I do recall there was some limitation I ran into when quotes aren't linear. It's possible I haven't quite handled all those scenarios yet. You've probably found a good test scenario that we need to cover better.

• Is there's a limitation on self-healing of QuoteHub inbound streams? #1585

[JGronholz]

@DaveSkender, since you provided backward-compatibility like a pro, my code wasn't broken when I came back around to this, and I didn't go investigating to discover the streaming features until a couple of days ago. Gives me a pine-vibe, but is a major leap. Thank you!

I have not pondered every edge-case that could complicate this, but I bet you'll appreciate the risk/reward for this:

Currently, RollbackState() must duplicate the indicator logic to reset the state.

Memory is cheaper than compute, so let's cache the state, instead:

StreamHub<TIn, TState, TOut> where TState : IIndicatorState
internal List<TState> StateCache { get; } = [];
//internal Dictionary<DateTime, TState> StateCache { get; } = []; //even faster

(XXXResult result, XXXState state, int index) ToIndicator(IReusable item, int? indexHint)

This makes RollbackState(rollbackDate) an O(1) op, for every indicator!

State = StateCache[rollbackDate];
return;

That, allows incremental updates to the current/last quote very quickly:

QuoteHub.Add(quote)
{
    if (IsLastAndDuplicate(quote))
    {
         allSubscribers.RollbackState(quote.Timestamp) //"instant" now
    }
    Add(quote)
}

....ish.

Rsi, for example:

RsiState rs = new(avgGain, avgLoss, sumGain, sumLoss);
return (r, rs, i);

I volunteer.

[DaveSkender]

Thanks @JGronholz this is good timing as I'm doing some performance tuning right now before publishing the next v3 preview release. I like some of your ideas here and will give them a closer inspection.

I volunteer.

If you want to do a PR to demo some of this, I'm perfectly okay with that ... would love the help.

[DaveSkender]

I see you started one, much appreciated!

• feat: Add StreamHubState<TIn, TState, TOut> for stateful indicators with cached rollback #1837

[eschgi]

Hi @DaveSkender,

can you give me a rough estimate of when you'll release the next preview of version V3? I'd like to test it.
Thanks in the meantime.

[DaveSkender]

Days away, or sooner! Doing some quick checks now!

As an alternate, we do auto-publish latest dev daily packages here too if you are okay with our wildly unstable pre-NuGet packages.

[DaveSkender]

I just published version 3.0.0-preview.2 with some quick release notes.

[JGronholz]

@DaveSkender, using the static series, I noticed there were just a few indicators whose results implement ISeries, instead of IReusable, so they lacked the all-powerful Value property. Working on what I consider a powerful upgrade to StreamHub, those same indicators are throwing a fit.

Is there a reason for some results implementing ISeries instead of IReusable? Some are less obvious than SMA, but each result has a definite winner for where to point Value. If this was an oversight, fix incoming... 😁

[DaveSkender]

This is intentional for indicators where there was no single standard data element to pass on for chaining. Did you see any ISeries types that were missed, but have a value to pass?

[eschgi]

Hi @DaveSkender,

thanks first for the new preview release, i will test it soon. :)

One thing is not clear to me if the following scenario is supported. I plan to (re)calculate my indicators every X seconds, my quotes/candles are per minute. I would like to update always the latest quote/candle until it is closed. Is that scenario already supported? When yes, is this supported by buffer lists or just by quote hubs? Or do i need to use only closed quotes/candles?

Thanks in the meantime.

[DaveSkender]

Currently, the StreamHub style will automatically handle both revision and healing for late arrival (prior timestamp arriving later than current one). In other words, if you're doing a quoteHub.Add(quote); where inserte quote has the same IQuote.Timestamp as the prior one, it will replace and cascade similar replacements in subscribed indicators ... at least that's how it's meant to work.

With that said, for doing aggregation we still have some work to implement for that utility.

• quote-stream auto-aggregator #1093

[DaveSkender]

I'll also look into whether we can do some kind of limited replace/heal in BufferList based indicators; however, that style is really meant to be a simpler faster incremental "middle-ground" without as many state management features.

• Can BufferList have limited healing? #1874
• feat: BufferList duplicate quote handling #1876 (experiment)

[JGronholz]

With the static types, my flow was:

while(true)
{
Update candles > Compute indicators > in-memory persist > best-effort durable persist
}

With the streamhubs, I'm at a fork and need to decide:

1. Hub = result factory/extruder
   Make hubs just bigger than required (lookback/etc.).
   After each quote is added to hub, persist LastItem from all hubs to external canonical write-through cache.
   Hub caches remain small, and internal-ish.

2. Hub = canonical in-memory cache
   Write-through caching to durable...
   Concurrency/racing...
   Charting, PropertyChanged...
   Unclear, so far.

3. Back to static
   Worked. Fast (enough).
   Wasn't broke; shouldn't have fixed.

[DaveSkender]

Yeah, there’s definitely more we need to do to explain the underlying utility of the hub observability aspect and extensibility, but you’re def starting to see the architectural differences and uses of it.

• docs: Explain external utility/extensibility of hub Subscribe() #1895

The library’s built-in hubs are a way to take the complexity out of the raw indicator handling aspects; but also gives users an ability to do much more advanced customizable wrapping for their unique needs via infinite possibilities with Subscribe() options.

// your custom instantiation
var observer = new ReusableObserver(
  ..

The library built-in hubs already implements this with simpler syntax like var emaHub = quoteHub.ToEmaHub(n); however, these built-in ones already have the needed onNext() kinds of interfaces defined, except with notably more functionality, including both IObserver and IObservable (if chainable), etc.

In other words, you really only need to make a customized IObserver<T> when you want to do something uniquely in your own code outside of the library.

[DaveSkender]

Side note: the reason the built in kinds are called “hubs” instead of “observers” is because they implement much more: IObserver<IReusable> receivers, IObservable<T> senders, “memory” caches, and other utilities to support ongoing streams, auto-healing, etc.

[DaveSkender]

With that said, I’ll take a look at your PR to see if it helps users with creating these external customizations.

• feat: Add extensible ReusableObserver implementing IStreamObserver<IReusable> #1894

[JGronholz]

This reflects the problem I had with invariance in the interfaces.
I was hoping to cast all hubs to StreamHub<ISeries/IReusable, IReusable> and be done, but it wasn't that simple.
As I mentioned in that PR, copilot found only one interface capable of storing them all (IChainProvider<IReusable>).

Kind of a hack, but I'm not actually looking to manage the full StreamHub.
I just wanted to observe the Values at the end of the chains (internal TOut LastItem).
Although this didn't end up as pretty as I'd hoped, it is extremely functional.

Every StreamHub Does() stuff, internally, but none appear explicitly designed to interact with the outside world (i.e. UI).
I assume standard .NET events were deliberately omitted for hot-path performance.
ReusableObserver provides an opportunity to Do() something at/with the tail-end of the chain.

So:

1. I can box any ChainHub to an IChainProvider
2. I can subscribe a ReusableObserver to any IChainProvider, evaluate Value, and automatically Do() stuff
3. ReusableObserver can live outside the library
   😎

[DaveSkender]

I'm still going to take a look at this when I'm working up some better examples that demonstrate capabilities in docs, so I'm not giving up on it -- there's certainly going to be areas where utils like ReusableObserver can help.

StreamHub Does() stuff, internally, but none appear explicitly designed to interact with the outside world

Yes. This is true. There's a lot going on inside a hub that can go very wrong if there's hooks to mess with the internals. This is normal for public libraries and part of intentional design.

With that said, I do want users to be able to make custom hubs, so there's a balancing act to figure out what is exposed and how to encapsulate things for safety.

assume standard .NET events were deliberately omitted

Yes. I compared the built-in Events and it, like LINQ, is designed as an easy way to implement some standard aspects of observer design patter, but they come at a much higher overhead cost. Under the hood, they implement IObserver<T> and IObservable<T> just like we do but with a narrower set of actions. Ours is setup to have only what's needed and nothing more, for better performance.

[JGronholz]

@DaveSkender
When I switched to the hubs, I did not RTFM, and discovered they weren't thread-safe the hard way.
When I replaced the persistence service with a hub manager, I didn't carry over the synchronization.
Repeatedly thinking I'd fixed the problem (I shouldn't even be having), it kept popping back up. Old news.
I did not actually RTFM for the exchange library, either...

Neither library contains synchronization primitives, allowing/forcing me to choose my own.

In the end, I realized that synchronizing the upstream-est quoteprovider synchronizes everything downstream. If QuoteHub is the primary upstream source (for not-just-me), you might consider adding a boolean to QuoteHub ctor enabling your favorite flavor of internal sync. Aggregators would want this, too. This would make all downstream chains optionally thread-safe, out-of-the-box.
Making syncprovider injectable sounds like nightmare.
Maybe as side-car with simple overrides:

QuoteHubSynced : QuoteHub
override void XXX() { sync ( base.XXX() ); }
override void YYY() { sync ( base.YYY() ); }

[DaveSkender]

I’m working up a few ideas to make the thread locking aspect a bit less annoying, though there will still be some feed-unique aspects only the caller will be able to handle.

• bug: Stream(async)Hubs are not thread-safe and produce ArgumentOutOfBoundsExceptions in the real world #1925