julia> ymat = randn(10, 5) ;

julia> @which fit!(CovMatrix(), eachrow(ymat))
fit!(o::OnlineStat{T}, yi::T) where T
     @ OnlineStatsBase /OnlineStatsBase.jl/src/OnlineStatsBase.jl:111

julia> @which fit!(CovMatrix(), eachrow(ymat))
fit!(o::OnlineStat{I}, y::T) where {I, T}
    in OnlineStatsBase at OnlineStatsBase.jl/src/OnlineStatsBase.jl:133

struct OHLCV{Ttime,Tprice,Tvol}
    open::Tprice
    high::Tprice
    low::Tprice
    close::Tprice
    volume::Tvol
    time::Ttime

    function OHLCV(
        open::Tprice,
        high::Tprice,
        low::Tprice,
        close::Tprice;
        volume::Tvol = missing,
        time::Ttime = missing,
    ) where {Ttime,Tprice,Tvol}
        new{Ttime,Tprice,Tvol}(open, high, low, close, volume, time)
    end

end

"""
    AccuDist{T}()

The AccuDist type implements an Accumulation and Distribution indicator.
"""
mutable struct AccuDist{T} <: OnlineStat{T}
    value::Union{Missing,Float64}
    n::Int

    function AccuDist{T}() where {T}
        new{T}(missing, 0)
    end
end

mutable struct AccuDist{T, S} <: OnlineStat{T, S}
    value::Union{Missing, S}
    n::Int

    function AccuDist{T, S}() where {T, S}
        new{T, S}(missing, 0)
    end
end
=#

function OnlineStatsBase._fit!(ind::AccuDist, candle::OHLCV)
    ind.n += 1
    if candle.high != candle.low
        # Calculate MFI and MFV
        mfi =
            ((candle.close - candle.low) - (candle.high - candle.close)) /
            (candle.high - candle.low)
        mfv = mfi * candle.volume
    else
        # In case high and low are equal (division by zero), return previous value if exists, otherwise return missing
        ind.value = value(ind)
        return
    end

    if !has_output_value(ind)
        ind.value = mfv
    else
        ind.value = value(ind) + mfv
    end
end

using OnlineStatsBase

mutable struct MyMean <: OnlineStat{Number}
    value::Float64
    n::Int
    MyMean() = new(0.0, 0)
end
function OnlineStatsBase._fit!(o::MyMean, y)
    o.n += 1
    o.value += (1 / o.n) * (y - o.value)
end

julia> @time reduce(Mean(), table, select=:C14)
  4.406728 seconds (40.45 M allocations: 618.137 MiB, 1.26% gc time)
Mean: n=40428967 | value=18841.8

julia> @time reduce(Variance(), table, select=:C14)
  0.152765 seconds (24.03 k allocations: 1.174 MiB)
Variance: n=40428967 | value=2.45962e7

julia> using OnlineStats

# Mockup data.
julia> vals = [1 2 3 4; 5 6 7 8; 9 10 11 12] #Note: would come from an iterator.
3×4 Array{Int64,2}:
 1   2   3   4
 5   6   7   8
 9  10  11  12

julia> (nrows, ncols) = size(vals)
(3, 4)

julia> attr1 = repeat(["a", "s", "a"], outer=ncols) #Note: would come from an iterator.
12-element Array{String,1}:
 "a"
 "s"
 "a"
 "a"
 "s"
 "a"
 "a"
 "s"
 "a"
 "a"
 "s"
 "a"

julia> attr2 = repeat(["q", "w", "e", "r"], inner=nrows) #Note: would come from an iterator.
12-element Array{String,1}:
 "q"
 "q"
 "q"
 "w"
 "w"
 "w"
 "e"
 "e"
 "e"
 "r"
 "r"
 "r"

# Define group keys.
julia> group_key = zip(attr1, attr2)
Base.Iterators.Zip{Tuple{Array{String,1},Array{String,1}}}((["a", "s", "a", "a", "s", "a", "a", "s", "a", "a", "s", "a"], ["q", "q", "q", "w", "w", "w", "e", "e", "e", "r", "r", "r"]))

julia> eltype(group_key)
Tuple{String,String}

# Mockup iterator.
julia> iter = zip(group_key, vals)
Base.Iterators.Zip{Tuple{Base.Iterators.Zip{Tuple{Array{String,1},Array{String,1}}},Array{Int64,2}}}((Base.Iterators.Zip{Tuple{Array{String,1},Array{String,1}}}((["a", "s", "a", "a", "s", "a", "a", "s", "a", "a", "s", "a"], ["q", "q", "q", "w", "w", "w", "e", "e", "e", "r", "r", "r"])), [1 2 3 4; 5 6 7 8; 9 10 11 12]))

julia> eltype(iter)
Tuple{Tuple{String,String},Int64}

julia> (i1, state) = iterate(iter)
((("a", "q"), 1), ((2, 2), 2))

julia> (i2, state) = iterate(iter, state)
((("s", "q"), 5), ((3, 3), 3))

julia> (i3, state) = iterate(iter, state)
((("a", "q"), 9), ((4, 4), 4))

# Of the 12 observations, there are 8 unique groups.
julia> first.(collect(iter)) |> unique
8-element Array{Tuple{String,String},1}:
 ("a", "q")
 ("s", "q")
 ("a", "w")
 ("s", "w")
 ("a", "e")
 ("s", "e")
 ("a", "r")
 ("s", "r")

# Setup stat.
julia> o = fit!(GroupBy(eltype(group_key), Mean()), iter)
GroupBy: Tuple{String,String} => Mean{Float64,EqualWeight}
  ├── ("a", "q"): Mean: n=2 | value=5.0
  ├── ("s", "q"): Mean: n=1 | value=5.0
  ├── ("a", "w"): Mean: n=2 | value=6.0
  ├── ("s", "w"): Mean: n=1 | value=6.0
  ├── ("a", "e"): Mean: n=2 | value=7.0
  ├── ("s", "e"): Mean: n=1 | value=7.0
  ├── ("a", "r"): Mean: n=2 | value=8.0
  └── ("s", "r"): Mean: n=1 | value=8.0

julia> o_desired = fit!(GroupBy(eltype(group_key), Group(Mean(), Variance(), Extrema())), iter)
ERROR: The input for GroupBy is a Union{Pair{Tuple{String,String},Union{Tuple{Number,Number,Number}, AbstractArray{#s28,1} where #s28<:Number, NamedTuple{names,R} where R<:Tuple{Number,Number,Number}} where names}, NamedTuple{names,Tuple{Tuple{String,String},Union{Tuple{Number,Number,Number}, AbstractArray{#s28,1} where #s28<:Number, NamedTuple{names,R} where R<:Tuple{Number,Number,Number}} where names}}, Tuple{Tuple{String,String},Union{Tuple{Number,Number,Number}, AbstractArray{#s28,1} where #s28<:Number, NamedTuple{names,R} where R<:Tuple{Number,Number,Number}} where names}} where names.  Found Char.
Stacktrace:
 [1] error(::String) at ./error.jl:33
 [2] fit!(::GroupBy{Tuple{String,String},Union{Tuple{Number,Number,Number}, AbstractArray{#s28,1} where #s28<:Number, NamedTuple{names,R} where R<:Tuple{Number,Number,Number}} where names,Group{Tuple{Mean{Float64,EqualWeight},Variance{Float64,EqualWeight},Extrema{Float64,Number}},Union{Tuple{Number,Number,Number}, AbstractArray{#s28,1} where #s28<:Number, NamedTuple{names,R} where R<:Tuple{Number,Number,Number}} where names}}, ::Char) at /Users/comara/.julia/packages/OnlineStatsBase/L6i9N/src/OnlineStatsBase.jl:108
 [3] fit! at /Users/comara/.julia/packages/OnlineStatsBase/L6i9N/src/OnlineStatsBase.jl:110 [inlined] (repeats 4 times)
 [4] top-level scope at REPL[29]:1

mutable struct Mean{T,W} <: OnlineStat{Number}
    μ::T
    weight::W
    n::Int
end
Mean(T::Type{<:Number} = Float64; weight = EqualWeight()) = Mean(zero(T), weight, 0)

using BenchmarkTools
using OnlineStatsBase

mutable struct MeanStdFilter{T}
    nu::Int
    tracker::OnlineStat
end

function MeanStdFilter(nu::Int; T::DataType=Float32)
    s = [Series(Mean(T), Variance(T)) for _ in 1:nu]
    return MeanStdFilter{T}(nu, Group(s...))
end

function _get_mean_var(m::MeanStdFilter{T}) where T
    vals = value.(value(m.tracker))
    return reinterpret(reshape, T, collect(vals))
end

function (m::MeanStdFilter)(x::AbstractVector)
    fit!(m.tracker, x)
    μσ2 = _get_mean_var(m)
    return (x .- μσ2[1,:]) ./ sqrt.(μσ2[2,:])
end

# Test runtime
nu = 4
T = Float32
m = MeanStdFilter(nu; T)

# @btime m(randn(T,nu));
@btime _get_mean_var(m);

julia> using OnlineStatsBase, OnlineStats

julia> value
WARNING: both OnlineStats and OnlineStatsBase export "value"; uses of it in module Main must be qualified
ERROR: UndefVarError: value not defined

> fit!(FTSeries(String, Mean(); transform=x->parse(Int, x)), string.(1:5))
FTSeries
└─ Mean: n=5 | value=3.0

> fit!(FTSeries(String, Mean(); transform=x->parse(Int, x)), [string.(1:5); "6.0"])
ERROR: ArgumentError: invalid base 10 digit '.' in "6.0"

mutable struct FTSeries{IN, OS, F, T} <: StatCollection{IN}
    stats::OS
    filter::F
    transform::T
    nfiltered::Int
    nfailed::Int    <--- added to keep track of transform failures 
end
function FTSeries(stats::OnlineStat...; filter=x->true, transform=identity)
    IN, OS = Union{map(input, stats)...}, typeof(stats)
    FTSeries{IN, OS, typeof(filter), typeof(transform)}(stats, filter, transform, 0, 0)
end
function FTSeries(T::Type, stats::OnlineStat...; filter=x->true, transform=identity)
    FTSeries{T, typeof(stats), typeof(filter), typeof(transform)}(stats, filter, transform, 0, 0)
end
nobs(o::FTSeries) = nobs(o.stats[1]) + o.nfailed   <--- counting failed transformations as observations, although I'm not convinced this is the best approach
@generated function _fit!(o::FTSeries{N, OS}, y) where {N, OS}
    n = length(fieldnames(OS))
    quote
        if o.filter(y)
            try
                yt = o.transform(y)
                Base.Cartesian.@nexprs $n i -> @inbounds begin
                    _fit!(o.stats[i], yt)
                end
            catch e
                o.nfailed += 1
            end
        else
            o.nfiltered += 1
        end
    end
end
function _merge!(o::FTSeries, o2::FTSeries)
    o.nfiltered += o2.nfiltered
    o.nfailed += o2.nfailed
    _merge!.(o.stats, o2.stats)
end