Skip to content

Bridge

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

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 (virtufin-market-data-engines). One-worker-per-symbol was judged operationally heavy; nothing about an indicator's own state needs isolating any other way. The constructor takes a Func<TIndicator> factory (not a single instance) so a fresh, empty indicator can be created the first time a new symbol is seen.

flowchart TD
    E[CloudEvent] --> SYM["ResolveSymbol(input, payload)"]
    SYM --> GET["IndicatorFor(symbol)"]
    GET --> DEC["DecodeSample(input, payload)"]
    DEC --> ADD["indicator.Add(sample)"]
    ADD --> ENC["EncodeValue(indicator.Value)"]
    ENC --> SAVE["SaveState (per-symbol key)"]
    SAVE --> ENV["BuildEnvelope"]
    ENV --> RESP[response / .changed CloudEvent]

Abstract hooks

A concrete subclass implements three hooks:

  • DecodeSample(CloudEvent, JsonNode) : TSample — decode the triggering payload into a sample. No canonical wire shape is imposed here — an SMA-over-candles worker's sample shape looks nothing like an RSI-over-ticks worker's.
  • ResolveSymbol(CloudEvent, JsonNode) : string — the instrument/symbol this sample belongs to, used for both per-symbol indicator dispatch and the state-key/topic scoping below.
  • EncodeValue(TValue) : JsonObject — encode the indicator's current value as the response/changed-event payload.

State persistence

Each update is persisted via the API gateway's statestore.SaveState RPC (CreateGateway lazily builds an ApiClient/ManagedGatewayAdapter for SaveState calls, mirroring BinanceOrderBookWorker.EnsureGateway's idempotent-rebuild pattern — overridable so tests can inject a fake IGatewayClient). The state key is:

sc.<scenarioId>.indicator.<name>.<symbol>

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.DevKit.WorkerBase.PublishTopicAttribute (publishtopic), set by BuildEnvelope to sc.<scenarioId>.indicator.<name>.<symbol>.changed; 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, same as StrategyWorkerBase.State/Portfolio and ExecutorWorkerBase.State on the other two bridges in this framework.

Limitations

One sample in, one response out, matching WorkerBase.IWorker.ProcessAsync's one-response-per-trigger limit — same limitation StrategyWorkerBase/ExecutorWorkerBase document on their own sides of the trading loop.