add ADL streaming

DaveSkender · Jul 1, 2024 · 5637575 · 5637575
1 parent f83ab96
commit 5637575
Show file tree

Hide file tree

Showing 12 changed files with 356 additions and 143 deletions.
diff --git a/src/_common/Quotes/Quote.Converters.cs b/src/_common/Quotes/Quote.Converters.cs
@@ -14,6 +14,7 @@ public static partial class QuoteUtility
     public static IReadOnlyList<Quote> ToQuoteList<TQuote>(
         this IEnumerable<TQuote> quotes)
         where TQuote : IQuote
+
         => quotes
             .OrderBy(x => x.Timestamp)
             .Select(x => x.ToQuote())
@@ -23,6 +24,7 @@ public static IReadOnlyList<Quote> ToQuoteList<TQuote>(
     internal static List<QuoteD> ToQuoteD<TQuote>(
         this IEnumerable<TQuote> quotes)
         where TQuote : IQuote
+
           => quotes
             .OrderBy(x => x.Timestamp)
             .Select(x => x.ToQuoteD())
@@ -33,6 +35,7 @@ public static IReadOnlyList<Reusable> ToReusableList<TQuote>(
         this IEnumerable<TQuote> quotes,
         CandlePart candlePart)
         where TQuote : IQuote
+
         => quotes
             .OrderBy(x => x.Timestamp)
             .Select(x => x.ToReusable(candlePart))
@@ -42,6 +45,7 @@ public static IReadOnlyList<Reusable> ToReusableList<TQuote>(
     internal static List<Reusable> ToReusableList(
         this List<QuoteD> qdList,
         CandlePart candlePart)
+
           => qdList
             .OrderBy(x => x.Timestamp)
             .Select(x => x.ToReusable(candlePart))
@@ -51,7 +55,9 @@ internal static List<Reusable> ToReusableList(
 
     // convert any IQuote type to native Quote type
     public static Quote ToQuote<TQuote>(this TQuote quote)
-        where TQuote : IQuote => new(
+        where TQuote : IQuote
+
+        => new(
             Timestamp: quote.Timestamp,
             Open: quote.Open,
             High: quote.High,
@@ -60,8 +66,9 @@ public static Quote ToQuote<TQuote>(this TQuote quote)
             Volume: quote.Volume);
 
     // convert to quote in double precision
-    private static QuoteD ToQuoteD(this IQuote quote)
-          => new(
+    internal static QuoteD ToQuoteD(this IQuote quote)
+
+        => new(
             Timestamp: quote.Timestamp,
             Open: (double)quote.Open,
             High: (double)quote.High,
@@ -70,9 +77,8 @@ private static QuoteD ToQuoteD(this IQuote quote)
             Volume: (double)quote.Volume);
 
     // convert TQuote element to a basic chainable class
-    internal static Reusable ToReusable(
-        this IQuote q,
-        CandlePart candlePart)
+    internal static Reusable ToReusable(this IQuote q, CandlePart candlePart)
+
         => candlePart switch {
 
             CandlePart.Open => new(q.Timestamp, (double)q.Open),
@@ -91,9 +97,8 @@ internal static Reusable ToReusable(
         };
 
     // convert quoteD element to reusable type
-    internal static Reusable ToReusable(
-        this QuoteD q,
-        CandlePart candlePart)
+    internal static Reusable ToReusable(this QuoteD q, CandlePart candlePart)
+
         => candlePart switch {
 
             CandlePart.Open => new(q.Timestamp, q.Open),

diff --git a/src/a-d/Adl/Adl.Series.cs b/src/a-d/Adl/Adl.Series.cs
@@ -17,7 +17,8 @@ internal static List<AdlResult> CalcAdl(
             QuoteD q = qdList[i];
 
             AdlResult r = Increment(
-                q.Timestamp, prevAdl, q.High, q.Low, q.Close, q.Volume);
+                q.Timestamp, prevAdl,
+                q.High, q.Low, q.Close, q.Volume);
 
             results.Add(r);
 

diff --git a/src/a-d/Adl/Adl.Stream.cs b/src/a-d/Adl/Adl.Stream.cs
@@ -22,6 +22,63 @@ public override string ToString()
         => Cache.Count == 0 ? "ADL" : $"ADL({Cache[0].Timestamp:d})";
 
     protected override void OnNextArrival(Act act, TQuote inbound)
-        => throw new NotImplementedException();
+    {
+        int i;
+        AdlResult r;
+
+        // handle deletes
+        if (act is Act.Delete)
+        {
+            i = Cache.FindIndex(inbound.Timestamp);
+
+            // cache entry unexpectedly not found
+            if (i == -1)
+            {
+                throw new InvalidOperationException(
+                    "Matching cache entry not found.");
+            }
+
+            r = Cache[i];
+        }
+
+        // calculate incremental value
+        else
+        {
+            i = Provider.FindIndex(inbound.Timestamp);
+
+            // source unexpectedly not found
+            if (i == -1)
+            {
+                throw new InvalidOperationException(
+                    "Matching source history not found.");
+            }
+
+            QuoteD q = inbound.ToQuoteD();
+
+            double prevAdl
+                = i == 0
+                ? 0
+                : Cache[i - 1].Adl;
+
+            // calculate ADL
+            r = Adl.Increment(
+                q.Timestamp, prevAdl,
+                q.High, q.Low, q.Close, q.Volume);
+        }
+
+        // save to cache
+        act = ModifyCache(act, r);
+
+        // send to observers
+        NotifyObservers(act, r);
+
+        // cascade update forward values (recursively)
+        if (act != Act.AddNew && i < ProviderCache.Count - 1)
+        {
+            int next = act == Act.Delete ? i : i + 1;
+            TQuote value = ProviderCache[next];
+            OnNextArrival(Act.Update, value);
+        }
+    }
     #endregion
 }
diff --git a/src/a-d/Adl/Adl.Utilities.cs b/src/a-d/Adl/Adl.Utilities.cs
@@ -14,12 +14,14 @@ public static partial class Adl
     /// for more information.</para>
     /// </summary>
     /// <param name="timestamp">Timestamp</param>
-    /// <param name="prevAdl">Last ADL value, from prior period</param>
+    /// <param name="prevAdl">
+    /// Last ADL value, from prior period
+    /// </param>
     /// <param name="high">High price, current period</param>
     /// <param name="low">Low price, current period</param>
     /// <param name="close">Close price, current period</param>
     /// <param name="volume">Volume, current period</param>
-    /// <returns>New ADL result value.</returns>
+    /// <returns>New ADL result value</returns>
     public static AdlResult Increment(
         DateTime timestamp,
         double prevAdl,

diff --git a/src/a-d/Alligator/Alligator.Stream.cs b/src/a-d/Alligator/Alligator.Stream.cs
@@ -58,34 +58,39 @@ public override string ToString()
     protected override void OnNextArrival(Act act, TIn inbound)
     {
         int i;
-        double jaw = double.NaN;
-        double lips = double.NaN;
-        double teeth = double.NaN;
+        AlligatorResult r;
 
         // handle deletes
         if (act == Act.Delete)
         {
             i = Cache.FindIndex(inbound.Timestamp);
-            AlligatorResult alligator = Cache[i];
 
-            jaw = alligator.Jaw.Null2NaN();
-            lips = alligator.Lips.Null2NaN();
-            teeth = alligator.Teeth.Null2NaN();
+            // cache entry unexpectedly not found
+            if (i == -1)
+            {
+                throw new InvalidOperationException(
+                    "Matching cache entry not found.");
+            }
+
+            r = Cache[i];
         }
 
-        // handle new values
+        // calculate incremental value
         else
         {
-
             i = Provider.FindIndex(inbound.Timestamp);
 
             // source unexpectedly not found
             if (i == -1)
             {
                 throw new InvalidOperationException(
-                    "Matching source history not found on arrival.");
+                    "Matching source history not found.");
             }
 
+            double jaw = double.NaN;
+            double lips = double.NaN;
+            double teeth = double.NaN;
+
             // calculate alligator's jaw, when in range
             if (i >= JawPeriods + JawOffset - 1)
             {
@@ -160,15 +165,15 @@ protected override void OnNextArrival(Act act, TIn inbound)
                     lips = (prevLips * (LipsPeriods - 1) + newVal) / LipsPeriods;
                 }
             }
-        }
 
-        // candidate result
-        AlligatorResult r = new() {
-            Timestamp = inbound.Timestamp,
-            Jaw = jaw.NaN2Null(),
-            Lips = lips.NaN2Null(),
-            Teeth = teeth.NaN2Null()
-        };
+            // candidate result
+            r = new() {
+                Timestamp = inbound.Timestamp,
+                Jaw = jaw.NaN2Null(),
+                Lips = lips.NaN2Null(),
+                Teeth = teeth.NaN2Null()
+            };
+        }
 
         // save to cache
         act = ModifyCache(act, r);

diff --git a/src/e-k/Ema/Ema.Stream.cs b/src/e-k/Ema/Ema.Stream.cs
@@ -36,16 +36,24 @@ public override string ToString()
     protected override void OnNextArrival(Act act, TIn inbound)
     {
         int i;
-        double ema;
+        EmaResult r;
 
         // handle deletes
         if (act is Act.Delete)
         {
             i = Cache.FindIndex(inbound.Timestamp);
-            ema = Cache[i].Ema.Null2NaN();
+
+            // cache entry unexpectedly not found
+            if (i == -1)
+            {
+                throw new InvalidOperationException(
+                    "Matching cache entry not found.");
+            }
+
+            r = Cache[i];
         }
 
-        // handle new values
+        // calculate incremental value
         else
         {
             i = Provider.FindIndex(inbound.Timestamp);
@@ -54,10 +62,11 @@ protected override void OnNextArrival(Act act, TIn inbound)
             if (i == -1)
             {
                 throw new InvalidOperationException(
-                    "Matching source history not found on arrival.");
+                    "Matching source history not found.");
             }
 
             // normal
+            double ema;
 
             if (i >= LookbackPeriods - 1)
             {
@@ -88,12 +97,12 @@ protected override void OnNextArrival(Act act, TIn inbound)
             {
                 ema = double.NaN;
             }
-        }
 
-        // candidate result
-        EmaResult r = new(
-            Timestamp: inbound.Timestamp,
-            Ema: ema.NaN2Null());
+            // candidate result
+            r = new(
+                Timestamp: inbound.Timestamp,
+                Ema: ema.NaN2Null());
+        }
 
         // save to cache
         act = ModifyCache(act, r);

diff --git a/src/e-k/Ema/Ema.Utilities.cs b/src/e-k/Ema/Ema.Utilities.cs
@@ -35,7 +35,8 @@ internal static void Validate(
         // check parameter arguments
         if (lookbackPeriods <= 0)
         {
-            throw new ArgumentOutOfRangeException(nameof(lookbackPeriods), lookbackPeriods,
+            throw new ArgumentOutOfRangeException(
+                nameof(lookbackPeriods), lookbackPeriods,
                 "Lookback periods must be greater than 0 for EMA.");
         }
     }

diff --git a/src/s-z/Sma/Sma.Stream.cs b/src/s-z/Sma/Sma.Stream.cs
@@ -38,29 +38,37 @@ public override string ToString()
     protected override void OnNextArrival(Act act, TIn inbound)
     {
         int i;
-        double sma;
+        SmaResult r;
 
         // handle deletes
         if (act == Act.Delete)
         {
             i = Cache.FindIndex(inbound.Timestamp);
-            sma = Cache[i].Sma.Null2NaN();
+
+            // cache entry unexpectedly not found
+            if (i == -1)
+            {
+                throw new InvalidOperationException(
+                    "Matching cache entry not found.");
+            }
+
+            r = Cache[i];
         }
 
-        // handle new values
+        // calculate incremental value
         else
         {
-            // calculate incremental value
             i = Provider.FindIndex(inbound.Timestamp);
 
             // source unexpectedly not found
             if (i == -1)
             {
                 throw new InvalidOperationException(
-                    "Matching source history not found on arrival.");
+                    "Matching source history not found.");
             }
 
             // normal
+            double sma;
 
             if (i >= LookbackPeriods - 1)
             {
@@ -78,12 +86,12 @@ protected override void OnNextArrival(Act act, TIn inbound)
             {
                 sma = double.NaN;
             }
-        }
 
-        // candidate result
-        SmaResult r = new(
-            Timestamp: inbound.Timestamp,
-            Sma: sma.NaN2Null());
+            // candidate result
+            r = new(
+                Timestamp: inbound.Timestamp,
+                Sma: sma.NaN2Null());
+        }
 
         // save to cache
         act = ModifyCache(act, r);