Bosatsu packages

Yichus/Series

Builtin package (resource series.bosatsu). This is the complete package source used by this build. The export list names its public API; definitions below give the types and behavior.

New to the API? Start with Make a calculator or Build an API, then use this page to look up a definition.

package Yichus/Series

from Bosatsu/Predef import (
  Int, List, Bool, True, False, LT, EQ, GT,
  cmp_Int, foldl_List, concat, reverse, range, map_List,
  Option, Some, None
)
from Yichus/Num/Float64 import (
  Float64,
  addf, subf, mulf, divf,
  int_to_Float64,
  eq_Float64
)
from Yichus/Geometry import Line, Point, y_at, horizontal

export (
  moving_average,
  linear_regression,
  project_linear,
  overlay
)

exposes Yichus/Geometry

# Series algebra over List[Point]. Every helper is ordinary Bosatsu, so
# Why/Math can walk into a smoothing or regression the same way it walks
# into Geometry. project_linear is one way to build a forecast series;
# a graph Projection layer draws whatever points the compute program
# produced and does not assume this form.

def zero() -> Float64:
  int_to_Float64(0)

def drop_head(pts: List[Point]) -> List[Point]:
  match pts:
    case []: []
    case [_, *rest]: rest

def add_y(acc: Float64, p: Point) -> Float64:
  Point(_, y) = p
  addf(acc, y)

def sum_y(pts: List[Point]) -> Float64:
  pts.foldl_List(zero(), add_y)

struct MaAcc(recent: List[Point], n: Int, out: List[Point])

def ma_step(window: Int, acc: MaAcc, p: Point) -> MaAcc:
  MaAcc(recent, n, out) = acc
  grown = concat(recent, [p])
  overflow = match cmp_Int(n.add(1), window):
    case GT: True
    case _: False
  sized = match overflow:
    case True: drop_head(grown)
    case False: grown
  new_n = match overflow:
    case True: window
    case False: n.add(1)
  Point(x, _) = p
  avg = divf(sum_y(sized), int_to_Float64(new_n))
  MaAcc(sized, new_n, [Point(x, avg), *out])

# Trailing mean of y, keeping each point's x. window <= 1 is identity.
def moving_average(points: List[Point], window: Int) -> List[Point]:
  match cmp_Int(window, 1):
    case LT | EQ: points
    case GT:
      MaAcc(_, _, out) = points.foldl_List(
        MaAcc([], 0, []),
        (acc, p) -> ma_step(window, acc, p)
      )
      reverse(out)

struct Stats(n: Int, sx: Float64, sy: Float64, sxy: Float64, sxx: Float64)

def stats_zero() -> Stats:
  Stats(0, zero(), zero(), zero(), zero())

def stats_step(acc: Stats, p: Point) -> Stats:
  Stats(n, sx, sy, sxy, sxx) = acc
  Point(x, y) = p
  Stats(
    n.add(1),
    addf(sx, x),
    addf(sy, y),
    addf(sxy, mulf(x, y)),
    addf(sxx, mulf(x, x))
  )

# A horizontal through the mean y. An empty series divides 0 by 0: the
# IEEE NaN line, which the graph runtime surfaces as a problem note —
# never a fabricated flat line at some invented level.
def mean_y_line(sy: Float64, n: Int) -> Line:
  horizontal(divf(sy, int_to_Float64(n)))

# Ordinary least squares. n < 2 or zero x-variance: a horizontal line
# through the mean y; an empty series has no line (NaN, surfaced).
def linear_regression(points: List[Point]) -> Line:
  Stats(n, sx, sy, sxy, sxx) = points.foldl_List(stats_zero(), stats_step)
  match cmp_Int(n, 2):
    case LT: mean_y_line(sy, n)
    case _:
      nf = int_to_Float64(n)
      denom = subf(mulf(nf, sxx), mulf(sx, sx))
      match eq_Float64(denom, zero()):
        case True: mean_y_line(sy, n)
        case False:
          slope = divf(subf(mulf(nf, sxy), mulf(sx, sy)), denom)
          intercept = divf(subf(sy, mulf(slope, sx)), nf)
          Line(slope, intercept)

def last_point(pts: List[Point]) -> Option[Point]:
  pts.foldl_List(None, (_, p) -> Some(p))

def project_one(line: Line, last_x: Float64, k: Int) -> Point:
  x = addf(last_x, int_to_Float64(k.add(1)))
  Point(x, y_at(line, x))

# Fit a line, then sample it at last_x+1 .. last_x+horizon. Empty series
# or non-positive horizon yields []. One way to build a forecast series;
# a Projection layer does not require this form.
def project_linear(points: List[Point], horizon: Int) -> List[Point]:
  match last_point(points):
    case None: []
    case Some(Point(last_x, _)):
      match cmp_Int(horizon, 0):
        case LT | EQ: []
        case GT:
          line = linear_regression(points)
          map_List(range(horizon), k -> project_one(line, last_x, k))

# The series as given. A graph Projection layer draws this overlay —
# vintage paths, compound growth, OLS samples, whatever the compute
# program built.
def overlay(points: List[Point]) -> List[Point]:
  points