A stateful process from a to b.

The existential state type s is hidden; the observable interface is the triple inject step extract. Keeping the triple as the primitive (rather than fusing extract into the step) preserves the streaming-statistics invariant that the first output is extract (inject x), before any step.

Create a Process from a (pure) binary operation.

Precompose a pure function before a process.

Postcompose a pure function after a process.

List specialization of scanStream.

List specialization of foldStream.

Most common statistics are averages, which are some sort of aggregation of values (sum) and some sort of sample size (count).

Pattern for an Averager.

A sum count

extract the average from an Averager

av gives NaN on zero divide

substitute a default value on zero-divide

av_ (Averager (0,0)) x == x

online f g is a Process where f is a transformation of the data and g is a decay function (usually convergent to zero) applied at each step.

online id id == av

online is best understood by examining usage to produce a moving average and standard deviation:

An exponentially-weighted moving average with a decay rate of 0.9

ma r == online id (*r)

An exponentially-weighted moving average of the square.

sqma r = online (\x -> x * x) (* r)

Parallel-style exponentially-weighted standard deviation computation:

std r = parWith (\s ss -> sqrt (ss - s ** 2)) (ma r) (sqma r)

A moving average using a decay rate of r. r=1 represents the simple average, and r=0 represents the latest value.

>>> fold' (ma 0) ([1..100])
100.0

>>> fold' (ma 1) ([1..100])
50.5

>>> fold' (ma 0.99) xs0
9.713356299018187e-2

absolute average

>>> fold' (absma 1) xs0
0.8075705557429647

average square

fold' (ma r) . fmap (**2) == fold' (sqma r)

standard deviation

The construction of standard deviation, combining ma and sqma in parallel:

std r = parWith (\s ss -> sqrt (ss - s ** (one+one))) (ma r) (sqma r)

The average deviation of the numbers 1..1000 is about 1 / sqrt 12 * 1000 https://en.wikipedia.org/wiki/Uniform_distribution_(continuous)#Standard_uniform

>>> fold' (std 1) [0..1000]
288.9636655359978

The average deviation with a decay of 0.99

>>> fold' (std 0.99) [0..1000]
99.28328803163829

>>> fold' (std 1) xs0
1.0126438036262801

The covariance of a tuple given an underlying central tendency fold.

>>> fold' (cov (ma 1)) xsp
0.7818936662586868

correlation of a tuple, specialised to Guassian

>>> fold' (corrGauss 1) xsp
0.7978347126677433

a generalised version of correlation of a tuple

>>> fold' (corr (ma 1) (std 1)) xsp
0.7978347126677433

corr (ma r) (std r) == corrGauss r

The beta in a simple linear regression of an (independent variable, single dependent variable) tuple given an underlying central tendency fold.

This is a generalisation of the classical regression formula, where averages are replaced by Process statistics.

\[ \begin{align} \beta & = \frac{n\sum xy - \sum x \sum y}{n\sum x^2 - (\sum x)^2} \\ & = \frac{n^2 \overline{xy} - n^2 \bar{x} \bar{y}}{n^2 \overline{x^2} - n^2 \bar{x}^2} \\ & = \frac{\overline{xy} - \bar{x} \bar{y}}{\overline{x^2} - \bar{x}^2} \\ \end{align} \]

>>> fold' (beta1 (ma 1)) $ zipWith (\x y -> (y, x + y)) xs0 xs1
0.999747321294513

The alpha in a simple linear regression of an (independent variable, single dependent variable) tuple given an underlying central tendency fold.

\[ \begin{align} \alpha & = \frac{\sum y \sum x^2 - \sum x \sum xy}{n\sum x^2 - (\sum x)^2} \\ & = \frac{n^2 \bar{y} \overline{x^2} - n^2 \bar{x} \overline{xy}}{n^2 \overline{x^2} - n^2 \bar{x}^2} \\ & = \frac{\bar{y} \overline{x^2} - \bar{x} \overline{xy}}{\overline{x^2} - \bar{x}^2} \\ \end{align} \]

>>> fold' (alpha1 (ma 1)) $ zipWith (\x y -> ((3+y), x + 0.5 * (3 + y))) xs0 xs1
1.3680496627365146e-2

The (alpha, beta) tuple in a simple linear regression of an (independent variable, single dependent variable) tuple given an underlying central tendency fold.

>>> fold' (reg1 (ma 1)) $ zipWith (\x y -> ((3+y), x + 0.5 * (3 + y))) xs0 xs1
(1.3680496627365146e-2,0.4997473212944953)

multiple regression

\[ \begin{align} {\hat {{\mathbf {B}}}}=({\mathbf {X}}^{{{\rm {T}}}}{\mathbf {X}})^{{ -1}}{\mathbf {X}}^{{{\rm {T}}}}{\mathbf {Y}} \end{align} \]

\[ \begin{align} {\mathbf {X}}={\begin{bmatrix}{\mathbf {x}}_{1}^{{{\rm {T}}}}\\{\mathbf {x}}_{2}^{{{\rm {T}}}}\\\vdots \\{\mathbf {x}}_{n}^{{{\rm {T}}}}\end{bmatrix}}={\begin{bmatrix}x_{{1,1}}&\cdots &x_{{1,k}}\\x_{{2,1}}&\cdots &x_{{2,k}}\\\vdots &\ddots &\vdots \\x_{{n,1}}&\cdots &x_{{n,k}}\end{bmatrix}} \end{align} \]

>>> let ys = zipWith3 (\x y z -> 0.1 * x + 0.5 * y + 1 * z) xs0 xs1 xs2
>>> let zs = zip (zipWith (\x y -> F.array @Vector @'[2] [x,y]) xs1 xs2) ys
>>> fold' (beta 0.99) zs
[0.6228820021456606,0.8461936860075405]

alpha in a multiple regression

multiple regression

>>> let ys = zipWith3 (\x y z -> 0.1 * x + 0.5 * y + 1 * z) xs0 xs1 xs2
>>> let zs = zip (zipWith (\x y -> F.array @Vector @'[2] [x,y]) xs1 xs2) ys
>>> fold' (reg 0.99) zs
([0.6228820021456606,0.8461936860075405],2.536775201287266e-2)

accumulated sum

constant Process

most recent value

most recent value if it exists, previous value otherwise.

delay input values by 1

delays values by n steps

delay [0] == delay1 0

delay [] == id

delay [1,2] = delay1 2 . delay1 1

>>> scan (delay [-2,-1]) [0..3]
[-2,-1,0,1]

Autocorrelation example:

scan (parWith (,) id (delay [0]) >>> beta (ma 0.99)) xs0

a moving window of a's, most recent at the front of the sequence

binomial operator applied to last and this value

generalised diff function.

Unchanged since last time.

Count observed values

Sum values of a key-value pair.

Convert a Process to a Process operating on lists.

A rough Median. The average absolute value of the stat is used to callibrate estimate drift towards the median

onlineL1' takes a function and turns it into a Process where the step is an incremental update of an (isomorphic) median statistic.

moving median