Rocket

Project.toml

@@ -5,6 +5,7 @@ version = "1.0.0-DEV"
 
 [deps]
 OnlineStatsBase = "925886fa-5bf2-5e8e-b522-a9147a512338"
+Rocket = "df971d30-c9d6-4b37-b8ff-e965b2cb3a40"
 
 [compat]
 julia = "1.6.7"

src/OnlinePortfolioAnalysis.jl

@@ -5,12 +5,14 @@ using OnlineStatsBase
 export SimpleAssetReturn
 export LogAssetReturn
 export StdDev
+export CumulativeReturn
 
 export fit!, value
 
 abstract type PortfolioAnalysis{T} <: OnlineStat{T} end
 
 include("asset_return.jl")
+include("cumulative_return.jl")
 include("std_dev.jl")
 
 include("value_at_risk.jl")

src/cumulative_return.jl

@@ -0,0 +1,18 @@
+mutable struct CumulativeReturn{T} <: PortfolioAnalysis{T}
+    value::T
+    n::Int
+
+    sum::Sum
+
+    function CumulativeReturn{T}() where {T}
+        val = zero(T)
+        s = Sum()
+        new{T}(val, 0, s)
+    end
+end
+
+function OnlineStatsBase._fit!(stat::CumulativeReturn, data)
+    fit!(stat.sum, data)
+    stat.n += 1
+    stat.value = 1 + value(stat.sum)
+end

test/runtests.jl

@@ -1,5 +1,6 @@
 using OnlinePortfolioAnalysis
 using OnlineStatsBase
+using Rocket
 using Test
 
 const ATOL = 0.0001
@@ -48,5 +49,27 @@ const weights = [0.4, 0.4, 0.2]
 
     end
 
+    #@testset "CumulativeReturn" begin
+    #    cum_ret = CumulativeReturn{Float64}()
+    #    fit!(cum_ret, TSLA)
+    #    @test isapprox(value(cum_ret), 265.9177, atol=ATOL)
+    #end
+
+    @testset "CumulativeReturn from prices" begin
+        source = from(TSLA)
+        ret = SimpleAssetReturn{Float64}()
+        cum_ret = CumulativeReturn{Float64}()
+
+        mapped_source = source |> 
+            map(Union{Missing,Float64}, price -> (fit!(ret, price); value(ret))) |>
+            filter(!ismissing) |>
+            map(Float64, r -> (fit!(cum_ret, r); value(cum_ret)))
+
+        observer(value) = println("Received value: ", value)
+        subscribe!(mapped_source, observer)
+        #fit!(ret, TSLA)
+        #@test isapprox(value(cum_ret), 265.9177, atol=ATOL)
+    end
+
 
 end