streaming preview: initial baseline

Stock.Indicators.sln

@@ -21,9 +21,15 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tests.Performance", "tests\
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Build", "Build", "{3A4158F9-4165-4823-9526-0CFAACCF1ACC}"
 	ProjectSection(SolutionItems) = preProject
-		.lgtm.yml = .lgtm.yml
 		.github\build.main.yml = .github\build.main.yml
 		.github\dependabot.yml = .github\dependabot.yml
+		gitversion.yml = gitversion.yml
+	EndProjectSection
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Observe.Streaming", "tests\observe\Observe.Streaming.csproj", "{14DEC3AF-9AF2-4A66-8BEE-C342C6CC4307}"
+	ProjectSection(ProjectDependencies) = postProject
+		{11CD6C7E-871F-4903-AEAD-58E034C6521D} = {11CD6C7E-871F-4903-AEAD-58E034C6521D}
+		{8D0F1781-EDA3-4C51-B05D-D33FF1156E49} = {8D0F1781-EDA3-4C51-B05D-D33FF1156E49}
 	EndProjectSection
 EndProject
 Global
@@ -48,6 +54,10 @@ Global
 		{3BD4837B-D197-41FD-A286-A3256D0770E1}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{3BD4837B-D197-41FD-A286-A3256D0770E1}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{3BD4837B-D197-41FD-A286-A3256D0770E1}.Release|Any CPU.Build.0 = Release|Any CPU
+		{14DEC3AF-9AF2-4A66-8BEE-C342C6CC4307}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{14DEC3AF-9AF2-4A66-8BEE-C342C6CC4307}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{14DEC3AF-9AF2-4A66-8BEE-C342C6CC4307}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{14DEC3AF-9AF2-4A66-8BEE-C342C6CC4307}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE

src/GlobalSuppressions.cs

@@ -39,3 +39,23 @@
     Justification = "Not compatible with `or` statement (Microsoft bug)",
     Scope = "member",
     Target = "~M:Skender.Stock.Indicators.ResultUtility.Condense``1(System.Collections.Generic.IEnumerable{``0})~System.Collections.Generic.IEnumerable{``0}")]
+
+[assembly: SuppressMessage(
+    "Naming",
+    "CA1725:Parameter names should match base declaration",
+    Justification = "The microsoft OnError implementation uses reserved word Error",
+    Scope = "member",
+    Target = "~M:Skender.Stock.Indicators.QuoteObserver.OnError(System.Exception)")]
+
+[assembly: SuppressMessage(
+    "Naming",
+    "CA1716:Identifiers should not match keywords",
+    Justification = "The microsoft OnError implementation uses reserved word Error",
+    Scope = "member",
+    Target = "~M:Skender.Stock.Indicators.QuoteObserver.OnError(System.Exception)")]
+
+[assembly: SuppressMessage(
+    "Naming",
+    "CA1716:Identifiers should not match keywords",
+    Justification = "The microsoft OnError implementation uses reserved word Error",
+    Scope = "member", Target = "~M:Skender.Stock.Indicators.TupleObserver.OnError(System.Exception)")]

src/_common/Generics/Seek.cs

@@ -4,12 +4,22 @@ namespace Skender.Stock.Indicators;
 
 public static class Seeking
 {
-    // FIND by DATE
-    /// <include file='./info.xml' path='info/type[@name="Find"]/*' />
+    // FIND SERIES by DATE
+    /// <include file='./info.xml' path='info/type[@name="FindSeries"]/*' />
     ///
     public static TSeries? Find<TSeries>(
         this IEnumerable<TSeries> series,
         DateTime lookupDate)
         where TSeries : ISeries => series
             .FirstOrDefault(x => x.Date == lookupDate);
+
+    // FIND INDEX by DATE
+    /// <include file='./info.xml' path='info/type[@name="FindIndex"]/*' />
+    ///
+    public static int FindIndex<TSeries>(
+        this List<TSeries> series,
+        DateTime lookupDate)
+        where TSeries : ISeries => series == null
+            ? -1
+            : series.FindIndex(x => x.Date == lookupDate);
 }

src/_common/Generics/info.xml

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="utf-8"?>
 
 <info>
-  <type name="Find">
+  <type name="FindSeries">
     <summary> Finds time series values on a specific date.
       <para>
         See <see href="https://dotnet.StockIndicators.dev/utilities/#find-indicator-result-by-date?utm_source=library&amp;utm_medium=inline-help&amp;utm_campaign=embedded">documentation</see> for more information.
@@ -10,8 +10,21 @@
     <typeparam name="TSeries">Any series type.</typeparam>
     <param name="series">Time series to evaluate.</param>
     <param name="lookupDate">Exact date to lookup.</param>
-    <returns>First
-      record in the series on the date specified.</returns>
+    <returns>First record in the series on the date specified.</returns>
+  </type>
+  <type name="FindIndex">
+    <summary>
+      Finds time series index on a specific date.
+      <para>
+        See <see href="https://dotnet.StockIndicators.dev/utilities/#find-indicator-result-by-date?utm_source=library">documentation</see> for more information.
+      </para>
+    </summary>
+    <typeparam name="TSeries">Any series type.</typeparam>
+    <param name="series">Time series to evaluate.</param>
+    <param name="lookupDate">Exact date to lookup.</param>
+    <returns>
+      First index in the series of the date specified or -1 if not found.
+    </returns>
   </type>
   <type name="Prune">
     <summary> Removes a specific quantity from the beginning of the time series list.

src/_common/Observables/ChainProvider.cs

@@ -0,0 +1,194 @@
+namespace Skender.Stock.Indicators;
+
+// TUPLE OBSERVER and TUPLE PROVIDER (CHAIN STREAM)
+
+public abstract class ChainProvider
+    : TupleObserver, IObservable<(DateTime Date, double Value)>
+{
+    // fields
+    private readonly List<IObserver<(DateTime Date, double Value)>> observers;
+
+    // initialize
+    protected ChainProvider()
+    {
+        observers = new();
+        ProtectedChain = new();
+        Warmup = true;
+    }
+
+    // properties
+    internal IEnumerable<(DateTime Date, double Value)> Output => ProtectedChain;
+
+    internal List<(DateTime Date, double Value)> ProtectedChain { get; set; }
+
+    private int OverflowCount { get; set; }
+
+    private bool Warmup { get; set; }
+
+    // METHODS
+
+    // subscribe observer
+    public IDisposable Subscribe(IObserver<(DateTime Date, double Value)> observer)
+    {
+        if (!observers.Contains(observer))
+        {
+            observers.Add(observer);
+        }
+
+        return new Unsubscriber(observers, observer);
+    }
+
+    // close all observations
+    public void EndTransmission()
+    {
+        foreach (IObserver<(DateTime Date, double Value)> observer in observers.ToArray())
+        {
+            if (observers.Contains(observer))
+            {
+                observer.OnCompleted();
+            }
+        }
+
+        observers.Clear();
+    }
+
+    // add one
+    internal void SendToChain<TResult>(TResult result)
+        where TResult : IReusableResult
+    {
+        // candidate result
+        (DateTime Date, double Value) r = new(result.Date, result.Value.Null2NaN());
+
+        int length = ProtectedChain.Count;
+
+        // initialize
+        if (length == 0 && result.Value != null)
+        {
+            // add new tuple
+            ProtectedChain.Add(r);
+            Warmup = false;
+
+            // notify observers
+            NotifyObservers(r);
+            return;
+        }
+
+        // do not proceed until first non-null Value recieved
+        if (Warmup && result.Value == null)
+        {
+            return;
+        }
+        else
+        {
+            Warmup = false;
+        }
+
+        (DateTime lastDate, _) = ProtectedChain[length - 1];
+
+        // add tuple
+        if (r.Date > lastDate)
+        {
+            // add new tuple
+            ProtectedChain.Add(r);
+
+            // notify observers
+            NotifyObservers(r);
+        }
+
+        // same date or tuple recieved
+        else if (r.Date <= lastDate)
+        {
+            // check for overflow condition
+            // where same tuple continues (possible circular condition)
+            if (r.Date == lastDate)
+            {
+                OverflowCount++;
+
+                if (OverflowCount > 100)
+                {
+                    string msg = "A repeated Chain update exceeded the 100 attempt threshold. "
+                      + "Check and remove circular chains or check your Chain provider.";
+
+                    EndTransmission();
+
+                    throw new OverflowException(msg);
+                }
+            }
+            else
+            {
+                OverflowCount = 0;
+            }
+
+            // seek old tuple
+            int foundIndex = ProtectedChain
+                .FindIndex(x => x.Date == r.Date);
+
+            // found
+            if (foundIndex >= 0)
+            {
+                ProtectedChain[foundIndex] = r;
+            }
+
+            // add missing tuple
+            else
+            {
+                ProtectedChain.Add(r);
+
+                // re-sort cache
+                ProtectedChain = ProtectedChain
+                    .ToSortedList();
+            }
+
+            // let observer handle old + duplicates
+            NotifyObservers(r);
+        }
+    }
+
+    // add many
+    internal void SendToChain<TResult>(IEnumerable<TResult> results)
+        where TResult : IReusableResult
+    {
+        List<TResult> added = results
+            .ToSortedList();
+
+        for (int i = 0; i < added.Count; i++)
+        {
+            SendToChain(added[i]);
+        }
+    }
+
+    // notify observers
+    private void NotifyObservers((DateTime Date, double Value) tuple)
+    {
+        List<IObserver<(DateTime Date, double Value)>> obsList = observers.ToList();
+
+        for (int i = 0; i < obsList.Count; i++)
+        {
+            IObserver<(DateTime Date, double Value)> obs = obsList[i];
+            obs.OnNext(tuple);
+        }
+    }
+
+    // unsubscriber
+    private class Unsubscriber : IDisposable
+    {
+        private readonly List<IObserver<(DateTime Date, double Value)>> observers;
+        private readonly IObserver<(DateTime Date, double Value)> observer;
+
+        // identify and save observer
+        public Unsubscriber(List<IObserver<(DateTime Date, double Value)>> observers, IObserver<(DateTime Date, double Value)> observer)
+        {
+            this.observers = observers;
+            this.observer = observer;
+        }
+
+        // remove single observer
+        public void Dispose()
+        {
+            if (observer != null && observers.Contains(observer))
+            {
+                observers.Remove(observer);
+            }
+        }
+    }
+}

src/_common/Observables/QuoteObserver.cs

@@ -0,0 +1,29 @@
+namespace Skender.Stock.Indicators;
+
+// OBSERVER of QUOTES (BOILERPLATE)
+
+public abstract class QuoteObserver : IObserver<Quote>
+{
+    // fields
+    private IDisposable? unsubscriber;
+
+    // properites
+    internal QuoteProvider? Supplier { get; set; }
+
+    // methods
+    public virtual void Subscribe()
+        => unsubscriber = Supplier != null
+            ? Supplier.Subscribe(this)
+            : throw new ArgumentNullException(nameof(Supplier));
+
+    public virtual void OnCompleted() => Unsubscribe();
+
+    public virtual void OnError(Exception error) => throw error;
+
+    public virtual void OnNext(Quote value)
+    {
+        // » handle new quote with override in observer
+    }
+
+    public virtual void Unsubscribe() => unsubscriber?.Dispose();
+}