The value source is the working half of a column: one instance per (row, column) that wires up to a data origin, pushes cell values into the grid through a sink, and cleans up on disposal. This chapter covers the source lifecycle, the services it draws on, and the threading rules that keep live data callbacks fast and correct.

IColumnValueSource: the lifecycle

A source is owned by the row's actor. Its lifecycle hooks always run on that single actor thread, and the engine never invokes two hooks on the same source concurrently. This is the same single-threaded discipline the indicator contract uses.

IColumnValueSource.csnamespace SabrTrader.Pipeline.MarketAnalyzer;

public interface IColumnValueSource : IAsyncDisposable
{
    // Wire the source up against the engine + row context. Called
    // exactly once, before any value is observed. Long-running work
    // (subscribing to live feeds, requesting backfill) happens here or
    // is dispatched via the actor's scheduler; it MUST NOT block.
    Task StartAsync(IColumnValueSink sink, CancellationToken cancellationToken);
}

• StartAsync is called exactly once, before any value is observed. Subscribe to your feed or start a backfill here. Return promptly; dispatch long-running work. Honour the CancellationToken for any awaited work.
• DisposeAsync (from IAsyncDisposable) is your teardown. Unhook every subscription, timer and event you registered. The source is disposed on row removal or panel dispose.

Publishing: IColumnValueSink & CellValue

The sink is the engine-owned callback through which a source publishes cell payloads. You call Publish with an immutable CellValue; the engine queues it and returns immediately.

IColumnValueSink.csnamespace SabrTrader.Pipeline.MarketAnalyzer;

public interface IColumnValueSink
{
    // Publish a new value for this column on this row.
    // Use CellValue.Initializing for "no data yet".
    void Publish(CellValue value);
}

public readonly record struct CellValue(CellState State, object? RawValue, string? ErrorMessage = null)
{
    public static CellValue Ready(object? value)     => new(CellState.Ready, value);
    public static CellValue Initializing             { get; } = new(CellState.Initializing, null);
    public static CellValue Error(string message)    => new(CellState.Error, null, message);
}

public enum CellState : byte
{
    Initializing = 0,   // source has not produced its first value yet (warming up)
    Ready        = 1,   // source has produced a valid value
    Error        = 2,   // source surfaced an error; cell shows an error indicator
}

The three factory members map to the three phases every cell goes through, and the grid renders each one distinctly (placeholder, value, or red badge) with no per-column branching from you:

Threading rules

The contract is generous about where Publish may be called from, and strict about what it does:

• Safe from the source's actor thread: the normal path, e.g. from inside StartAsync.
• Safe from arbitrary background threads: a live tick callback, a timer tick. The engine marshals the payload onto the row actor before any downstream effect fires.
• Non-blocking: the engine queues the payload and returns, which keeps live data callbacks latency-free. Compute the value first, then publish; keep heavy work and locks out of the callback.

IColumnContext: the service surface

The context is the engine service surface handed to every CreateSource call. Sources resolve only the services they need. The core services are first-class members; the row's feed and richer host services come through GetService, so the SDK never references the engine or trading layers.

IColumnContext.csnamespace SabrTrader.Pipeline.MarketAnalyzer;

public interface IColumnContext
{
    // Engine clock. Tests inject a fake; production uses UTC system time.
    IMarketAnalyzerClock Clock { get; }

    // The data provider that owns a row, or null when none is available yet.
    IDataProvider? ResolveDataProvider(InstrumentRef instrument);

    // User-visible time zone for daily aggregates. UTC unless the host sets one.
    TimeZoneInfo SessionTimeZone { get; }

    // Session-day boundary resolver.
    ISessionResolver SessionResolver { get; }

    // An OPTIONAL host service by type, or null when the host does not offer it.
    // Prefer the typed GetService<T>() extension.
    object? GetService(Type serviceType) => null;

    // The provider-native symbol for backfill / live subscription.
    // Default: identity. Idempotent.
    string ResolveBackfillSymbol(InstrumentRef instrument) => instrument.SymbolId;

    // Calls onAvailable each time a data provider becomes available for the
    // instrument. Returns a token that stops the observation, or null when the
    // host offers no such signal.
    IDisposable? ObserveProviderAvailability(InstrumentRef instrument, Action onAvailable) => null;
}

public static class ColumnContextExtensions
{
    // Typed GetService: the service, or null.
    public static T? GetService<T>(this IColumnContext context) where T : class;

    // The venue-native symbol for this row: the row feed's NativeSymbol, or
    // ResolveBackfillSymbol for a source built outside a row.
    public static string NativeSymbolFor(this IColumnContext context, InstrumentRef instrument);
}

// Abstraction over DateTime.UtcNow; tests inject a controllable fake.
public interface IMarketAnalyzerClock
{
    DateTime UtcNow { get; }
}

InstrumentRef & symbol translation

The row a source serves is identified by an InstrumentRef: a value-equality key of a symbol plus an optional provider key.

InstrumentRef.csnamespace SabrTrader.Pipeline.MarketAnalyzer;

public readonly record struct InstrumentRef(string SymbolId, string? ProviderKey = null)
{
    public override string ToString();   // "providerKey:SymbolId", or just SymbolId when ProviderKey is null

    // Throwing validator; trims both parts and rejects a blank symbol.
    public static InstrumentRef Validated(string symbolId, string? providerKey = null);
}

• SymbolId is the instrument symbol the row was added with ("MNQM6", "BTC-USD", or a root such as "ES"). Treat it as opaque.
• ProviderKey is the stable IDataProvider.Key of the provider serving the row; null means "resolve through the registry's default provider".

Sessions: ISessionResolver & the clock

Session-aware columns (today's open/high/low/volume, net change vs. prior close, session VWAP) anchor to the trading-session boundary the resolver reports. The resolver maps an instrument plus a moment to the UTC instant the current session began.

ISessionResolver.csnamespace SabrTrader.Pipeline.MarketAnalyzer;

public interface ISessionResolver
{
    // The UTC start of the trading session that momentUtc falls inside,
    // for instrument. Pure function: same inputs => same output, no
    // hidden state. Two timestamps in the same session return the same
    // start; the first timestamp past the next session's start returns
    // that later start.
    DateTime GetSessionStartUtc(InstrumentRef instrument, DateTime momentUtc);
}

The host ships a generic resolver configured per session (CME ETH, NYSE RTH, FX 24-hour weeks, crypto 24/7). To detect a session roll, compare the session start for the current tick's timestamp against the one you cached; when it advances, reset your running session state (sums, highs, lows) and start over.

The row's market data: IInstrumentFeed

The service a column reaches for most often is the row's feed. A row has ONE binding to market data (one provider resolution, one native-symbol translation, one venue subscription) and every column on that row shares it, so two columns on the same row always show the same contract. Resolve it with context.GetService<IInstrumentFeed>(). Its handlers run on the provider's feed thread; registering and disposing is safe from any thread.

IInstrumentFeed.csnamespace SabrTrader.Pipeline.MarketAnalyzer;

public interface IInstrumentFeed
{
    string NativeSymbol { get; }         // the ONE symbol this row subscribes under, resolved once
    IDataProvider? Provider { get; }     // the bound provider; null while none is available
    RowBinding Binding { get; }          // bound to which provider, or why not
    RowQuote Quote { get; }              // the current merged quote, always readable

    event Action? Rebound;               // provider appeared, reconnected or was swapped

    IDisposable OnQuote(Action<RowQuoteChange> handler);        // replays a known quote at once
    IDisposable OnTicks(Action<Tick> handler);
    IDisposable OnSummary(Action<MarketSummaryUpdate> handler);

    Task<IReadOnlyList<Bar>> LoadBarsAsync(BarSpecification spec, TimeSpan? lookback, CancellationToken ct);
}

LastPriceSource.cs (StartAsync)public Task StartAsync(IColumnValueSink sink, CancellationToken cancellationToken)
{
    var feed = _context.GetService<IInstrumentFeed>();
    if (feed is null)
    {
        sink.Publish(CellValue.Ready(null));    // no feed on this host: blank, not an error
        return Task.CompletedTask;
    }

    sink.Publish(CellValue.Initializing);
    _sub = feed.OnQuote(change =>
    {
        // React only to the field you render: a bid tick should not repaint a Last cell.
        if ((change.Changed & RowQuoteFields.Last) != 0 && change.Quote.HasLast)
            sink.Publish(CellValue.Ready(change.Quote.Last));
    });
    return Task.CompletedTask;
}

RowQuote is one snapshot, merged from the trade tape where the venue streams one and from the venue's level-1 statements otherwise, so a cell never mixes a fresh bid with a stale last. A trade print outranks a statement for Last, and a settlement outranks a prior close for PreviousClose. An unknown field is double.NaN; the Has* members test for it, and Last and PreviousClose also carry WHERE they came from:

Optional services via GetService

GetService<T>() exposes optional host seams without coupling your plugin to the host assembly. It returns null when the host does not offer the service; a missing service is a host-configuration fact, never an exception. IInstrumentFeed is resolved this way; the other SDK-visible seam is the instrument description source:

IInstrumentInfoSources.csnamespace SabrTrader.Pipeline.MarketAnalyzer;

// Optional host service: human-readable instrument descriptions
// (e.g. "Micro E-mini Nasdaq-100").
public interface IInstrumentDescriptionSource
{
    // Description for a NATIVE symbol (after root-to-contract translation),
    // or null when unknown.
    string? GetDescription(string nativeSymbol);
}

DescriptionSource.cs (StartAsync)public Task StartAsync(IColumnValueSink sink, CancellationToken cancellationToken)
{
    var descriptions = _context.GetService<IInstrumentDescriptionSource>();
    if (descriptions is null)
    {
        sink.Publish(CellValue.Ready(null));   // host offers no descriptions: blank cell, not an error
        return Task.CompletedTask;
    }

    string native = _context.NativeSymbolFor(_instrument);
    sink.Publish(CellValue.Ready(descriptions.GetDescription(native)));   // null -> blank
    return Task.CompletedTask;
}

A complete session-VWAP source

This source ties the chapter together: it takes the row's feed, listens to its ticks, accumulates a session-anchored VWAP, resets on each session roll via the ISessionResolver, and unhooks in DisposeAsync. Pair it with a definition whose DataType is ColumnDataType.Price (as in Defining a column).

SessionVwapSource.csusing System;
using System.Threading;
using System.Threading.Tasks;
using SabrTrader.Pipeline.MarketAnalyzer;
using SabrTrader.Pipeline.Ticks;

namespace MyPlugin.Columns;

internal sealed class SessionVwapSource : IColumnValueSource
{
    private readonly InstrumentRef _instrument;
    private readonly IColumnContext _context;

    private IDisposable? _subscription;
    private DateTime _sessionStartUtc = DateTime.MinValue;
    private double _priceVolume;   // running sum of price * size
    private long   _volume;        // running sum of size
    private volatile bool _disposed;

    public SessionVwapSource(InstrumentRef instrument, IColumnContext context)
    {
        _instrument = instrument;
        _context = context;
    }

    public Task StartAsync(IColumnValueSink sink, CancellationToken cancellationToken)
    {
        IInstrumentFeed? feed = _context.GetService<IInstrumentFeed>();
        if (feed is null)
        {
            sink.Publish(CellValue.Error("No market data for this row."));
            return Task.CompletedTask;
        }

        sink.Publish(CellValue.Initializing);                 // warming up

        // Feed handlers run on the provider's feed thread; sink.Publish marshals for us.
        _subscription = feed.OnTicks(tick =>
        {
            if (_disposed) return;
            if ((tick.Flags & TickFlags.Trade) == 0) return;  // VWAP counts trades only

            // Reset the accumulator when the trading session rolls.
            DateTime sessionStart =
                _context.SessionResolver.GetSessionStartUtc(_instrument, tick.ExchangeTimestampUtc);
            if (sessionStart != _sessionStartUtc)
            {
                _sessionStartUtc = sessionStart;
                _priceVolume = 0;
                _volume = 0;
            }

            _priceVolume += tick.Price * tick.Size;
            _volume      += tick.Size;
            if (_volume > 0)
                sink.Publish(CellValue.Ready(_priceVolume / _volume));
        });

        return Task.CompletedTask;
    }

    public ValueTask DisposeAsync()
    {
        _disposed = true;
        _subscription?.Dispose();
        return ValueTask.CompletedTask;
    }
}

The shape of a well-behaved value source: its fields are touched only from the feed callback, so it needs no locks; the sink does the thread marshalling; the resolver owns the session math; and time comes from the tick's exchange timestamp rather than DateTime.UtcNow.