calculated

package
v1.1.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 26, 2026 License: MIT Imports: 22 Imported by: 0

Documentation

Overview

Package calculated evaluates the expression language used by EVMABIEncodeUnpackedExpr channels to derive new stream values from observed ones.

It is shared by every plugin version, so anything here that changes an output changes it for already-deployed DONs.

Expressions

An expression is an expr-lang expression evaluated against an environment of stream values and functions, and must evaluate to a decimal. Builtins are disabled, so only the functions registered in defaultEnv are callable.

Streams are named by identifier: s10001 for a scalar stream, and s10001_bid / s10001_benchmark / s10001_ask for a quote stream. A quote has no bare value — s10001 alone is not bound for a quote stream. s10001_timestamp is bound for timestamped streams. Only streams the channel declares are in scope. 

Div(Add(s1, s2), s3)

Stream history

History(<stream identifier>, <depth>) reads the last <depth> agreed values of a stream: 

Avg(History(s10001, 10))
EMA(History(s10001, 50), 20)
TWAP(History(s10001, 600), {window: Duration("5m"), minSamples: 240,
                            maxHeadGap: 30, maxInteriorGap: 10, maxTailGap: 30})

The call is both the declaration of how much history to persist and the read of it. There is no separate configuration: the depth kept for a stream is the deepest any live channel asks for.

It is resolved at compile time, not called at run time — the AST is rewritten so each call becomes an identifier bound to the loaded window (history_ast.go). Two consequences: both arguments must be literal (a bare stream identifier and an integer), and the depth is known before evaluation, which is what lets the plugin know how much to keep.

History requires replicated state, so it works only on protocol versions that have it. On v30 an expression using History fails closed: no value, and the channel does not report.

Functions

Scalar functions, unchanged by history: Add, Sub, Mul, Div, Pow, Sqrt, Ln, Log, Abs, Round, Ceil, Floor, Duration, IsZero, IsNegative, IsPositive, and the comparisons EQ/Equal, GT/GreaterThan, GTE/GreaterThanOrEqual, LT/LessThan, LTE/LessThanOrEqual.

Accepting either a history window or a list of scalars: Avg, Sum, Min, Max. The two forms cannot be mixed — Avg(History(s1, 10), s2) is ambiguous about whether the scalar is another sample or a weight, so it is rejected.

Window-only: 

Count      number of values
First      oldest value
Last       newest value (the value this round agreed on)
Median     middle value; mean of the two middles for an even length
Variance   population variance
Stddev     population standard deviation
Delta      newest minus oldest
PctChange  (newest - oldest) / oldest, as a fraction: 0.05 is a 5% rise
Spread     maximum minus minimum
SMA(w, n)  simple mean of the newest n
WMA(w, n)  linearly weighted, newest weighted n and the oldest of the n weighted 1
EMA(w, n)  seeded with the mean of the oldest n, then alpha = 2/(n+1) newest-ward
TWAP(w, c) time-weighted average price over c.window, filling gaps by type

A window may only be passed directly to one of these. Add(History(s1, 10), 2) is rejected when the expression is validated, not left to fail during evaluation.

Warmup

A window is readable only once it holds at least the requested depth. Until then the expression is not evaluated, no value is written, and the channel is not reportable — so its coverage watermark does not advance and no gap is falsely claimed.

The operational consequence: adding a History call to a live channel, or raising its depth, stops that channel reporting until the window fills. Deploy the change as a NEW channel, wait for its history to be satisfied, then retire the old one. Lowering a depth takes effect the next round with no gap.

Limits

Depth per (stream, aggregator) pair, the number of such pairs, the per-round byte budget and the per-record size are all bounded by constants in llo/protocol/limits.go. They are hardcoded because they determine persisted state and so must not vary per node.

A pair denied history because a cap was reached gets none at all, and channels reading it do not report. There is no silently shortened window.

Evaluation

A round runs in three phases: prepare, evaluate, apply.

Prepare is sequential. It reads everything shared — the opts cache, the round's stream aggregates, and the HistoryReader, whose one-read-per-pair memoization is stateful and so cannot be driven from several goroutines. It hands each channel a fully materialized input: an environment and its bound windows.

Evaluate is pure and runs on a small worker pool, one channel per unit of work. Nothing it reads is shared and it writes only its own result slot, so no synchronization is needed beyond waiting for the workers. Rounds whose total work is small enough that dispatching would cost more than evaluating run inline instead.

Apply is sequential and walks channels in ascending channel ID order, writing aggregates and logging failures. The order is fixed there, not inherited from the worker pool, which is what keeps the outcome independent of scheduling.

Channel definitions are not written. Which calculated streams a channel reports is derived from its opts by protocol.EffectiveStreams, so evaluation contributes nothing to replicated state. v3.0 is the exception: it commits channel definitions in its outcome and so keeps appending to them, via ProcessCalculatedStreamsWithDefinitionAppend.

Determinism

Expression results become consensus values, so identical inputs must give bit-identical output on every node. See decimalmath.go: no float64 anywhere in the calculation, no reliance on decimal.DivisionPrecision (a mutable global), fixed rounding at every step of an iterative calculation, and a lock around shopspring/decimal's transcendental functions, which are not concurrency-safe.

Div and Avg are pinned to the precision they have always effectively used (16); functions added with stream history use 18. Changing the former would move the trailing digits of every existing calculated stream.

Validation

ValidateExpression and ValidateChannelExpressions apply every static rule without evaluating anything, and are what a report codec's VerifyForAdmission should call so an unusable definition cannot reach consensus. They belong there rather than in Verify because Verify also runs against already-committed definitions, and rejecting one of those stops an oracle from observing at all. ProcessCalculatedStreamsDryRun goes further, evaluating against synthesized inputs, and is for offline configuration tooling.

Index

Constants

View Source 
const HistoryFunctionName = "History"

HistoryFunctionName is the DSL form that declares and reads a window of a stream's past agreed values: 

History(s10001, 600)

It is a compile-time form, not a runtime function. The AST is rewritten so each call becomes a plain identifier bound to the loaded window, for two reasons:

  1. The requested depth must be known before evaluation, because it is what tells the plugin how much history to keep. A depth discovered at run time is discovered a round too late.
  2. A stream identifier passed as an argument would resolve through the environment to a scalar, so the function body would receive a single number rather than a reference to the stream.

Both arguments must therefore be literal: a bare stream identifier and an integer.

Variables

View Source 
var (
	// ErrTWAPRejected is what every rejection satisfies errors.Is against, so
	// callers can detect a rejected window without inspecting the reasons.
	ErrTWAPRejected = errors.New("TWAP window rejected by the acceptance rule")

	// ErrTWAPConfig is returned for a malformed configuration. Configuration is
	// static, so this is a deployment error rather than a data condition.
	ErrTWAPConfig = errors.New("invalid TWAP configuration")
)

TWAP is ported from the original spec (ADR 0013/0014/0015), semantics are unchanged. It is a port rather than a reuse for two reasons: the source works from decoded values and a clock, while here the input is an already-agreed history window; and the source computes in float64, which is not guaranteed bit-identical across architectures and so cannot appear in a consensus path. Every logarithm, exponential and division below is decimal at a fixed precision.

View Source 
var ErrHistoryExpression = errors.New("invalid History expression")

ErrHistoryExpression is returned for any expression that misuses History. It is a static error: the expression is rejected at configuration time and at compile time, never accepted and left to fail during evaluation.

View Source 
var ErrInsufficientHistory = protocol.ErrInsufficientStreamHistory

ErrInsufficientHistory is returned when a window holds fewer records than the expression asked for. It means "not yet evaluable" — during warmup, after a depth increase, or after corrupt state was discarded — and must never be treated as zero or silently substituted with a shorter window.

View Source 
var ErrSeriesAsScalar = errors.New("history window cannot be used as a scalar")

ErrSeriesAsScalar is returned when a window reaches a function that takes scalars. Static analysis rejects this at configuration time (history_ast.go), so reaching it means the analysis was bypassed; it exists so the failure is loud rather than a silently coerced value.

Functions

func Abs 

func Abs(x any) (decimal.Decimal, error)

Abs returns the absolute value of x

func Add 

func Add(x, y any) (decimal.Decimal, error)

Add returns the sum of x and y

func AggregatorByStream added in v1.1.1 

func AggregatorByStream(cd llotypes.ChannelDefinition) (map[llotypes.StreamID]llotypes.Aggregator, error)

AggregatorByStream indexes a channel's non-calculated streams by stream ID. It is the mapping from what a History call names (a stream) to what history is keyed by (a stream and an aggregator), and is exported so the plugin derives its persisted requirements from exactly the same mapping evaluation uses.

A stream appearing twice under different aggregators makes any History call naming it ambiguous — the DSL has no way to say which aggregation is meant — so this is rejected rather than resolved arbitrarily. Rejecting is deterministic; picking one would differ between nodes if map iteration order ever leaked in.

func Avg 

func Avg(x ...any) (decimal.Decimal, error)

Avg returns the average of x elements, which may be a single history window or a list of scalars.

NOTE: the division is pinned to legacyDivisionPrecision, which is what this function has always effectively used via decimal.DivisionPrecision. Raising it would change the trailing digits of every existing calculated stream.

func Ceil 

func Ceil(x any) (decimal.Decimal, error)

Ceil returns the ceiling of x

func Count added in v1.1.1 

func Count(x any) (decimal.Decimal, error)

Count returns the number of values in a history window.

An empty window is an error rather than a count of zero, matching every other window function: a window is only bound once it holds exactly the requested depth, so an empty one means something upstream went wrong, and a zero flowing into a stream value would hide that.

func Delta added in v1.1.1 

func Delta(x any) (decimal.Decimal, error)

Delta returns the change across a history window: newest minus oldest.

func Div 

func Div(x, y any) (decimal.Decimal, error)

Div returns the quotient of x and y

func EMA added in v1.1.1 

func EMA(x any, n any) (decimal.Decimal, error)

EMA returns the exponential moving average of a history window with smoothing period n.

The series is seeded with the simple mean of its oldest n values, then iterated newest-ward with the conventional smoothing factor: 

alpha = 2 / (n + 1)
ema   = value*alpha + ema*(1 - alpha)

Determinism note: the recursive form is path-dependent, so the seed rule, the iteration count and the rounding must all be fixed. They are: the window length is exactly the depth the expression requested, the channel does not evaluate until the window is that deep, alpha is an exact decimal fraction rather than a float, and every step is rounded to the package precision. That is what makes the result reproducible across oracles and across restarts.

func Equal 

func Equal(x, y any) (bool, error)

Equal returns true if x and y are equal

func ExpressionStreamIDs added in v1.1.1 

func ExpressionStreamIDs(optsCache *protocol.OptsCache, cd llotypes.ChannelDefinition, cid llotypes.ChannelID) ([]llotypes.StreamID, error)

ExpressionStreamIDs returns the calculated (expression) stream IDs declared by a channel's opts, in declaration order. It is the source of truth for which calculated streams a channel is expected to produce.

Returns an error if the opts cannot be resolved, declare no expressions, or declare a zero expression stream ID.

func Expressions added in v1.1.1 

func Expressions(optsCache *protocol.OptsCache, cd llotypes.ChannelDefinition, cid llotypes.ChannelID) ([]string, error)

Expressions returns a channel's expressions in declaration order.

It exists so the plugin can derive history requirements from the same source of truth evaluation uses — the channel's opts — rather than from the streams appended to the channel definition, which only appear once evaluation has got that far.

An ABI entry with no expression is an error, not a skip. Such an entry names a calculated stream that nothing can produce: evalExpression fails on it, the stream stays absent, and the channel is therefore never reportable. Reporting that when the definition is validated is the whole point of validating it.

func First added in v1.1.1 

func First(x any) (decimal.Decimal, error)

First returns the oldest value in a history window.

func Floor 

func Floor(x any) (decimal.Decimal, error)

Floor returns the floor of x

func GreaterThan 

func GreaterThan(x, y any) (bool, error)

GreaterThan returns true if x is greater than y

func GreaterThanOrEqual 

func GreaterThanOrEqual(x, y any) (bool, error)

GreaterThanOrEqual returns true if x is greater than or equal to y

func IsNegative 

func IsNegative(x any) (bool, error)

IsNegative returns true if x is negative

func IsPositive 

func IsPositive(x any) (bool, error)

IsPositive returns true if x is positive

func IsZero 

func IsZero(x any) (bool, error)

IsZero returns true if x is zero

func Last added in v1.1.1 

func Last(x any) (decimal.Decimal, error)

Last returns the newest value in a history window, which is the value the current round agreed on.

func LessThan 

func LessThan(x, y any) (bool, error)

LessThan returns true if x is less than y

func LessThanOrEqual 

func LessThanOrEqual(x, y any) (bool, error)

LessThanOrEqual returns true if x is less than or equal to y

func Ln 

func Ln(x any) (decimal.Decimal, error)

Ln returns the natural logarithm of x.

func Log 

func Log(x, y any) (decimal.Decimal, error)

Log returns the logarithms of y with base x. This is equivalent to log_x(y).

We use this formula: 

             ln(y)
log_x(y)  =  ----
             ln(x)

func Max 

func Max(x ...any) (decimal.Decimal, error)

Max returns the maximum of x elements, which may be a single history window or a list of scalars.

func Median added in v1.1.1 

func Median(x any) (decimal.Decimal, error)

Median returns the middle value of a history window: the exact middle for an odd length, the mean of the two middle values for an even one.

func Min 

func Min(x ...any) (decimal.Decimal, error)

Min returns the minimum of x elements

func Mul 

func Mul(x, y any) (decimal.Decimal, error)

Mul returns the product of x and y

func NewEnv 

func NewEnv(observationTimestampNanoseconds uint64) environment

NewEnv returns a new environment with the default functions

func ParseDuration 

func ParseDuration(x string) (time.Duration, error)

ParseDuration parses a duration string into a time.ParseDuration

func PctChange added in v1.1.1 

func PctChange(x any) (decimal.Decimal, error)

PctChange returns the fractional change across a history window, (newest - oldest) / oldest. It is a fraction, not a percentage: 0.05 is a 5% rise.

func Pow 

func Pow(x, y any) (decimal.Decimal, error)

Pow returns x, raised to the power of y

func ProcessCalculatedStreams 

func ProcessCalculatedStreams(lggr logger.Logger, channelDefinitions llotypes.ChannelDefinitions, streamAggregates protocol.StreamAggregates, observationTimestampNanoseconds uint64, optsCache *protocol.OptsCache, history HistoryReader)

ProcessCalculatedStreams evaluates expressions for each channel of the EVMABIEncodeUnpackedExpr format and writes the evaluated values into streamAggregates. It is version-agnostic: both the v3.0 and v3.1 plugins call it with their own outcome/precursor fields.

channelDefinitions is read-only. The calculated streams a channel reports are derived from its opts by protocol.EffectiveStreams, not stored on the definition, so a definition is exactly what was voted on and evaluation contributes nothing to replicated state.

Processing runs in three phases — prepare, evaluate, apply — so that only the pure part is parallelized:

  • prepare is sequential. It touches everything that is shared or not goroutine-safe: the opts cache, the stream aggregates it reads inputs from, and the HistoryReader, whose one-read-per-pair memoization is inherently stateful.
  • evaluate is pure and runs concurrently. A compiled program plus an environment is all it needs, and nothing it touches is shared with another channel.
  • apply is sequential and walks channels in ascending channel ID order, so the aggregates written and the errors logged do not depend on how the work was scheduled.

func ProcessCalculatedStreamsDryRun 

func ProcessCalculatedStreamsDryRun(expression string) error

ProcessCalculatedStreamsDryRun processes the calculated streams for the given expression against synthetic inputs and returns an error if it cannot be evaluated. Useful for validating expressions.

func ProcessCalculatedStreamsWithDefinitionAppend deprecated added in v1.1.1 

func ProcessCalculatedStreamsWithDefinitionAppend(lggr logger.Logger, channelDefinitions llotypes.ChannelDefinitions, streamAggregates protocol.StreamAggregates, observationTimestampNanoseconds uint64, optsCache *protocol.OptsCache, history HistoryReader)

ProcessCalculatedStreamsWithDefinitionAppend behaves as ProcessCalculatedStreams but additionally appends each channel's calculated streams to its channel definition.

Deprecated: for v3.0 only. v3.0 commits channelDefinitions as part of its outcome.

func Round 

func Round(x any, precision int) (decimal.Decimal, error)

Round returns the rounded value of x to the given precision

func SMA added in v1.1.1 

func SMA(x any, n any) (decimal.Decimal, error)

SMA returns the simple moving average of the newest n values of a history window.

func Spread added in v1.1.1 

func Spread(x any) (decimal.Decimal, error)

Spread returns the range of a history window: maximum minus minimum.

func Sqrt 

func Sqrt(x any) (decimal.Decimal, error)

Sqrt returns the square root of x. Returns error for negative values.

func Stddev added in v1.1.1 

func Stddev(x any) (decimal.Decimal, error)

Stddev returns the population standard deviation of a history window.

func Sub 

func Sub(x, y any) (decimal.Decimal, error)

Sub returns the difference of x and y

func Sum added in v1.1.1 

func Sum(x ...any) (decimal.Decimal, error)

Sum returns the total of a history window or a list of scalars.

Add remains the binary form; Sum is the aggregate one.

func Truncate 

func Truncate(x any, precision int) (decimal.Decimal, error)

Truncate truncates off digits from the number, without rounding.

func ValidateChannelExpressions added in v1.1.1 

func ValidateChannelExpressions(optsCache *protocol.OptsCache, cd llotypes.ChannelDefinition, cid llotypes.ChannelID) error

ValidateChannelExpressions validates every expression a channel's opts declare. Errors are joined so one pass reports all of them.

func ValidateExpression added in v1.1.1 

func ValidateExpression(expression string) error

ValidateExpression reports whether an expression is statically well formed.

It is the check to run before a channel definition reaches consensus: it parses, rewrites History calls, and applies every static rule (argument shapes, depth caps, per-expression fan-out, window positions, reserved names, TWAP configuration satisfiability). It does not evaluate, so it needs no stream values and no persisted state, and it is a pure function of the expression string.

A statically invalid expression can never produce a value, so accepting one into a channel definition means accepting a channel that will never report.

func Variance added in v1.1.1 

func Variance(x any) (decimal.Decimal, error)

Variance returns the population variance of a history window.

Population rather than sample variance: the window is the whole series being described, not a sample drawn from a larger one.

func WMA added in v1.1.1 

func WMA(x any, n any) (decimal.Decimal, error)

WMA returns the linearly weighted moving average of the newest n values of a history window, weighting the newest value n and the oldest of the n values 1. 

WMA = (n*x[newest] + (n-1)*x[newest-1] + ... + 1*x[oldest]) / (n + (n-1) + ... + 1)

Types

type Field added in v1.1.1 

type Field uint8

Field selects which part of a stored stream value a window projects. One stored window serves every field, so History(s1, 10), History(s1_bid, 10) and History(s1_ask, 10) share a single series in state and differ only here.

NOTE: Field lives in this file rather than with Series because the AST pass is what parses field suffixes and it must not depend on the evaluation types. 

const (
	FieldValue Field = iota
	FieldBid
	FieldAsk
	FieldBenchmark
)

func (Field) String added in v1.1.1 

func (f Field) String() string

type HistoryReader added in v1.1.1 

type HistoryReader interface {
	// Series returns the newest count records of a pair, projected to a field.
	// It must return ErrInsufficientHistory when fewer than count records are
	// stored.
	Series(streamID llotypes.StreamID, aggregator llotypes.Aggregator, count uint32, field Field) (Series, error)
}

HistoryReader is the read side of persisted stream history.

Implementations must be memoized: one underlying state read per (streamID, aggregator) pair per round no matter how many channels or expressions ask, since that is what keeps history within the per-round state budget. Projection to a field and to a depth is cheap and happens per call.

A nil HistoryReader means history is unavailable (the v30 plugin has no replicated key-value state), and expressions using History must then fail closed rather than evaluate against an empty window.

type HistoryRef added in v1.1.1 

type HistoryRef struct {
	StreamID llotypes.StreamID
	Field    Field
	Count    uint32
}

HistoryRef is one History call recovered from an expression: which stream and field it reads, and how deep. It is both the read and the declaration — the plugin derives the depth it must persist per stream from these.

func AnalyzeExpressionHistory added in v1.1.1 

func AnalyzeExpressionHistory(expression string) ([]HistoryRef, error)

AnalyzeExpressionHistory returns the History references an expression declares: which stream and field each reads, and how deep.

This is the declaration side of the DSL. Callers use it to derive how much history to persist per stream, so it must stay a pure function of the expression string. An expression using no History returns no references and no error; an expression misusing History returns ErrHistoryExpression and no references, and must not be evaluated.

The returned slice is deduplicated, ordered by (streamID, field, depth), and shared with the cache: callers must not modify it.

func (HistoryRef) String added in v1.1.1 

func (r HistoryRef) String() string

type Series added in v1.1.1 

type Series struct {
	// contains filtered or unexported fields
}

Series is an immutable window of a stream's past agreed values, oldest first, with the timestamp each value was observed at.

Timestamps travel with the values rather than being derived from position: rounds are not evenly spaced (stalls, leader changes, minimum report intervals), so any time-weighted function or gap check needs the real observation times. This is also why _timestamp is not a rangeable field — the timestamps are already here.

func NewSeries added in v1.1.1 

func NewSeries(values []decimal.Decimal, timestamps []uint64) (Series, error)

NewSeries builds a series from parallel value and timestamp slices. It is exported for tests and for history readers outside this package.

func SeriesFromRecords added in v1.1.1 

func SeriesFromRecords(records []protocol.StreamHistoryRecord, field Field) (Series, error)

SeriesFromRecords projects stored history records onto a single field. One stored window serves every field, so this is where History(s1, N), History(s1_bid, N) and History(s1_ask, N) diverge.

It is exported so history readers in other packages (the v31 plugin) can build a Series without duplicating the field and type handling.

func (Series) Len added in v1.1.1 

func (s Series) Len() int

Len is the number of values in the window.

func (Series) Newest added in v1.1.1 

func (s Series) Newest() (decimal.Decimal, error)

Newest returns the most recent value.

func (Series) Oldest added in v1.1.1 

func (s Series) Oldest() (decimal.Decimal, error)

Oldest returns the least recent value.

func (Series) String added in v1.1.1 

func (s Series) String() string

func (Series) Timestamps added in v1.1.1 

func (s Series) Timestamps() []uint64

Timestamps returns the observation time of each value in nanoseconds, parallel to Values and strictly increasing. The slice must be treated as read-only.

func (Series) Values added in v1.1.1 

func (s Series) Values() []decimal.Decimal

Values returns the window's values, oldest first. The slice must be treated as read-only.

type TWAPRejection added in v1.1.1 

type TWAPRejection struct {
	Reasons                                            []TWAPRejectionReason
	M, Ghead, Gint, Gtail                              int
	MinSamples, MaxHeadGap, MaxInteriorGap, MaxTailGap int
	WindowStartSeconds, WindowEndSeconds               int64
	Records                                            int
}

TWAPRejection carries the measured statistics alongside the thresholds they failed, so an operator can tell a thin window from a stalled feed without reproducing the calculation.

func (*TWAPRejection) Error added in v1.1.1 

func (e *TWAPRejection) Error() string

func (*TWAPRejection) Is added in v1.1.1 

func (e *TWAPRejection) Is(target error) bool

type TWAPRejectionReason added in v1.1.1 

type TWAPRejectionReason string

TWAPRejectionReason enumerates why a window failed the acceptance rule. A window can fail several checks at once. 

const (
	// ReasonInsufficientSamples: M < minSamples, the coverage floor.
	ReasonInsufficientSamples TWAPRejectionReason = "min_samples"
	// ReasonHeadGapTooLong: Ghead > maxHeadGap, backfilled prefix too long.
	ReasonHeadGapTooLong TWAPRejectionReason = "head_gap_too_long"
	// ReasonInteriorGapTooLong: Gint > maxInteriorGap, longest both-sides-anchored gap too long.
	ReasonInteriorGapTooLong TWAPRejectionReason = "interior_gap_too_long"
	// ReasonTailGapTooLong: Gtail > maxTailGap, carry-forward suffix too long.
	ReasonTailGapTooLong TWAPRejectionReason = "tail_gap_too_long"
)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.