Bridge

IndicatorWorkerBase<TIndicator, TSample, TValue> is the bridge between an IIndicator<TSelf, TSample, TValue> and the WorkManager's worker contract (IWorker.ProcessAsync(CloudEvent input, IWorkerContext context)).

Per-symbol dispatch

A worker instance tracks one indicator instance per symbol seen, not one shared instance across everything it observes — mirrors BinanceOrderBookWorker's multi-pair-per-instance precedent (market-data-workers). One-worker-per-symbol was judged operationally heavy; nothing about an indicator's own state needs isolating any other way. The constructor takes a factory (not a single instance) so a fresh, empty indicator can be created the first time a new symbol is seen:

• Func<TIndicator>: every symbol gets the same indicator.
• Func<CloudEvent, TIndicator>: the factory gets the event that first shows the symbol, so parameters can come from the worker's config, e.g. ConfigResolution.ResolvePositiveInt(input, "windowcandles", 20). One package then serves sma-7 and sma-30 deployments.

Sample source and encoder

A worker is constructed from its worker type, an ISampleSource<TSample>, an IValueEncoder<TValue> and the indicator factory. The source decides which events are samples (Accepts; an event it rejects is skipped: no state change, no output, no error), which symbol a sample belongs to (ResolveSymbol) and how it decodes (Decode); the encoder turns the value into the event's data and the saved state. No canonical wire shape is imposed: an SMA-over-candles worker's sample looks nothing like an RSI-over-ticks worker's.

flowchart TD
    E[CloudEvent] --> ACC{"source.Accepts(input)"}
    ACC -->|no| SKIP[skip: no output]
    ACC -->|yes| SYM["source.ResolveSymbol(input)"]
    SYM --> GET["IndicatorFor(symbol, input)"]
    GET --> DEC["source.Decode(input)"]
    DEC --> ADD["indicator.Add(sample)"]
    ADD --> ENC["encoder.Encode(indicator.Value, input)"]
    ENC --> SAVE["SaveState (per-symbol key)"]
    SAVE --> ENV["BuildEnvelope"]
    ENV --> RESP[response / .changed CloudEvent]

Topic and subject

A changed event is published on sc.<scenarioId>.indicator.<name>.changed.<PAIR> with ce-type com.virtufin.indicator.<name>.changed and ce-subject (all built by the naming library from this package's virtufin.indicator fragment; see Naming: <name> is lowercase kebab, <PAIR> canonical) currency_pair/<PAIR> (openspec pubsub-topics, "Indicator Events"; the subject was the bare symbol before 0.4.0).

Envelope

Changed events carry the pubsub-topics spec's Common Envelope Extensions. eventtime (= ce-time) comes from the input's eventtime, else its ce-time, else now, so indicators run on event time. walltime is the publish time. clocktype (default wall), exchange, marketworld, marketuniverse, portfolioworld, strategyworld and runid are carried forward from the input event or config when present.

State persistence

Each update is persisted via the State API of the worker's context (IWorkerContext.Api): the cluster's gateway under the Runtime, the in-memory LocalVirtufinApi under a LocalWorkManager or a BehaviorTestHost, so a worker saves state locally with no gateway configured. CreateStateStore(input, context) wraps it as an ApiClientIndicatorStateStore; a test injects a fake IIndicatorStateStore through the StateStore property. An apihost config value (host[:port], https://host[:port] for TLS) names an explicit other gateway instead, dialled through Worker.DevKit's WorkerApiEndpoint (the host's key is sent only to its own configured gateway). The state key is:

sc.<scenarioId>.indicator.<name>.<PAIR>

Scenario-scoped, not market-data-cache-scoped (act|hyp.<name>.exchange.<venue>.*, the pattern BinanceOrderBookWorker uses for order books) — a real correctness requirement: an indicator fed by scenario-specific replayed/synthetic data under hyp.* needs disambiguation two concurrent backtests of the same symbol would otherwise collide on. An order book has no such need (it's a genuinely singular venue-level fact regardless of observer), which is why that pattern differs.

Response routing

Like the trading envelope (StrategyWorkerBase/ExecutorWorkerBase), ce-type stays a stable identifier (com.virtufin.indicator.<name>.changed) independent of the per-scenario topic. Routing goes through Virtufin.Worker.Host.WorkerBase.PublishTopicAttribute (publishtopic), set by BuildEnvelope to sc.<scenarioId>.indicator.<name>.changed.<PAIR>; WorkManager reads that extension to pick the publish topic and strips it before publish, so subscribers never see it.

State

Per-symbol indicator instances live in the worker instance's own dictionary — the engine keeps one worker object alive for the deployment's lifetime, so state threads across deliveries, as in the strategy and execution bridges.

Redelivery

Inputs arrive at least once since WorkManager 0.38.0: after a hard crash the samples of the last cursor window (≤250 ms) come again (Inputs across a move). IndicatorWorkerBase folds each sample once with no code of its own, by input-id deduplication in BehaviorWorkerBase (Inputs can arrive twice): a sample whose source and id the worker already took is dropped before it is decoded, so no window counts it twice and no .changed event repeats. A crashed producer (BinanceCandles, BinanceTrades) republishes what it published under the same ids, so the running indicator drops those too. A moved indicator starts with empty windows and folds the replayed samples once. replay stays on (default).

Limitations

One sample in, one .changed event out. IndicatorWorkerBase is built on BehaviorWorkerBase (Virtufin.Behavior.DevKit): the indicator runs as an IAsyncProcess per symbol (indicator.AsProcess() from IndicatorBehaviorExtensions lifts any indicator to a process), scenarioid and indicatorname are checked before the sample is folded in, and the value is saved before the in-memory state advances -- a failed save is an error and the sample does not count.

A lighter alternative

IndicatorWorkerBase is the full bridge: per-symbol dispatch, State persistence, response routing. For cases that don't need any of that (inline in a strategy, in a test, a one-off script), lib's IndicatorExtensions.Adapt (Virtufin.Core.Behavior) builds an indicator that takes a raw input, here a CloudEvent, decoding it into a sample before folding:

var sma = SimpleMovingAverage<DateTimeOffset, DecimalAmount>.Empty(fromCount, isExpired)
    .Adapt((CloudEvent ce) => DecodeCandle(ce))
    .Add(candleEvent);

Pacing: publish and save intervals

An indicator that folds a high-rate input (trades, quotes) cannot announce or save per sample. Two optional keys, both defaulting to 0 = every sample:

Key	Effect
publishintervalms	at most one changed event per symbol per interval, latest wins. The fold runs on every sample; only the announcement is thinned, and a symbol with no due sample simply emits nothing (there is no timer)
saveintervalms	at most one state write per symbol per interval, in the background, always the newest value. 0 keeps the inline save, where the in-memory state only advances after the write succeeds

Invalid values fail the sample with an error response naming the key. The gate and writer live in IndicatorPacing/PublishGate/LatestValueWriter.

Sample sources and encoders

An indicator's input family and its wire encoding are separate from the fold:

• ISampleSource<TSample> decodes one family (candles, trades, quotes, book) into samples and owns the Accepts/ResolveSymbol decisions. Decode takes the whole CloudEvent, so JSON and binary payloads are both possible.
• IValueEncoder<TValue> produces EncodedPayload: the JSON state that is saved under the pair's key, and optionally bytes for the event (with content type and dataschema). The state stays JSON even when the published value is binary.

Every IndicatorWorkerBase is constructed with a source, an encoder and an indicator factory. Samples arrive ordered per key, but sample time is the venue's and is not guaranteed monotonic (a late trade, a re-sent full book or a replay can go backwards), so a windowed fold must not assume it is.

Binary input works through Behavior.DevKit's LenientJsonCodec: a source-driven worker turns a non-JSON payload into an empty object and the source reads the bytes from the event. Binary output is EncodedPayload.WithBinary.