Every file of a working market-data venue: plugin, descriptor, manifest, session, data port, historical source, tick source, catalog and tests.

What we are building

A complete, honest venue: one REST + WebSocket vendor exposed as a market-data venue with historical bars, live trades, instrument metadata and a typed catalog. Every file below is the real shape; only the vendor's own REST and socket clients are left out.

The project

SabrTrader.Pipeline.Example.csproj<Project Sdk="Microsoft.NET.Sdk">
  <PropertyGroup>
    <TargetFramework>net10.0</TargetFramework>
    <Nullable>enable</Nullable>
    <RootNamespace>SabrTrader.Pipeline.Example</RootNamespace>
  </PropertyGroup>

  <ItemGroup>
    <!-- CONTRACTS ONLY. An engine reference breaks type identity under the plugin load context. -->
    <PackageReference Include="SabrTrader.Sdk" />
  </ItemGroup>
</Project>

Plugin & descriptor

ExampleVenuePlugin.csusing SabrTrader.Pipeline.Providers;
using SabrTrader.Pipeline.Venues;

namespace SabrTrader.Pipeline.Example;

public sealed class ExampleVenuePlugin : IVenuePlugin
{
    public IReadOnlyList<IVenueDescriptor> Venues { get; } =
        new IVenueDescriptor[] { new ExampleVenueDescriptor() };
}

internal sealed class ExampleVenueDescriptor : IVenueDescriptor
{
    internal const string TypeId       = "Example";
    internal const string FieldApiKey  = "ApiKey";
    internal const string FieldPaper   = "UsePaper";

    public VenueManifest Manifest { get; } = new(
        TypeId: TypeId,
        DisplayName: "Example Data",
        Category: VenueCategory.MarketData,
        AssetClasses: new[] { VenueAssetClass.Stock, VenueAssetClass.Future },
        Capabilities: VenueCapabilities.LiveTrades
                      | VenueCapabilities.HistoricalBars
                      | VenueCapabilities.InstrumentCatalog,
        SettingsSchema: new[]
        {
            new ProviderCredentialField(FieldApiKey, "API key",
                ProviderCredentialKind.Secret,
                HelpText: "Issued in the Example portal under Settings → API. " +
                          "Leave blank when editing to keep the saved key."),
            new ProviderCredentialField(FieldPaper, "Use sandbox",
                ProviderCredentialKind.Toggle,
                Required: false, DefaultValue: "false"),
        },
        SupportsPaper: true,
        StateStorageFolder: "example");

    public bool TryValidateSettings(IReadOnlyDictionary<string, string> values, out string? error)
    {
        if (values.TryGetValue(FieldApiKey, out var key) && key.Length is > 0 and < 20)
        {
            error = "That API key looks truncated: Example keys are 32 characters.";
            return false;
        }
        error = null;
        return true;
    }

    public IVenueSession CreateSession(VenueSessionContext context)
        => new ExampleVenueSession(context);

    public IVenueSymbolParser? CreateSymbolParser(ISharedSymbolParsers shared)
        => new ExampleSymbolParser(shared);
}

internal sealed class ExampleSymbolParser(ISharedSymbolParsers shared) : IVenueSymbolParser
{
    public VenueSymbolParse Parse(string rawSymbol)
        => rawSymbol.StartsWith("F:", StringComparison.Ordinal)
            ? shared.Futures.Parse(rawSymbol[2..])
            : shared.Equities.Parse(rawSymbol);
}

The session

ExampleVenueSession.csinternal sealed class ExampleVenueSession : IVenueSession
{
    private readonly VenueSessionContext _context;
    private ExampleDataProvider? _provider;
    private int _status;

    public ExampleVenueSession(VenueSessionContext context)
        => _context = context ?? throw new ArgumentNullException(nameof(context));

    public ProviderConnectionStatus Status => (ProviderConnectionStatus)Volatile.Read(ref _status);
    public event Action<ProviderConnectionStatus>? StatusChanged;
    public event Action<VenueStreams>? StreamsRestored;
    public VenueFailure? LastFailure { get; private set; }

    public IDataProvider? Data => _provider;
    public ITradingProvider? Trading => null;

    public async Task ConnectAsync(CancellationToken cancellationToken)
    {
        SetStatus(ProviderConnectionStatus.Connecting);
        _context.Notices.Log("Connecting…");

        bool paper = _context.Settings.TryGetValue(ExampleVenueDescriptor.FieldPaper, out var p)
                     && bool.TryParse(p, out var b) && b;

        var provider = new ExampleDataProvider(
            apiKey: _context.Settings[ExampleVenueDescriptor.FieldApiKey],
            paper: paper,
            context: _context);

        // The venue's own socket healed itself: tell the host to re-register feeds
        // and backfill the gap. No full session cycle needed.
        provider.SocketRestored += () => StreamsRestored?.Invoke(VenueStreams.MarketData);

        try
        {
            await provider.ConnectAsync(cancellationToken).ConfigureAwait(false);
            _provider = provider;
            SetStatus(ProviderConnectionStatus.Connected);
            _context.Notices.Log("Connected.");
        }
        catch (Exception ex)
        {
            try { provider.Dispose(); } catch { }
            bool permanent = ex is ExampleAuthException;
            LastFailure = new VenueFailure(
                permanent
                    ? "Example rejected the API key: verify it in the portal."
                    : $"Example connect failed: {ex.Message}",
                permanent);
            SetStatus(ProviderConnectionStatus.Failed);
            throw new InvalidOperationException(LastFailure.Message, ex);
        }
    }

    public async Task DisconnectAsync()
    {
        var provider = Interlocked.Exchange(ref _provider, null);
        if (provider is not null)
        {
            try { await provider.DisconnectAsync().ConfigureAwait(false); } catch { }
            try { provider.Dispose(); } catch { }
        }
        SetStatus(ProviderConnectionStatus.Disconnected);
    }

    public async ValueTask DisposeAsync() => await DisconnectAsync().ConfigureAwait(false);

    private void SetStatus(ProviderConnectionStatus next)
    {
        var prev = (ProviderConnectionStatus)Interlocked.Exchange(ref _status, (int)next);
        if (prev != next) StatusChanged?.Invoke(next);
    }
}

The data port

ExampleDataProvider.csinternal sealed class ExampleDataProvider : IDataProvider, IVenueInstrumentCatalog
{
    private readonly ExampleRestClient _rest;
    private readonly ExampleSocketClient _socket;
    private readonly InMemoryInstrumentMetadata _instruments = new();
    private readonly List<VenueCatalogEntry> _entries = new();
    private IBackfillProvider _backfill = null!;
    private bool _tickDerived;

    public ExampleDataProvider(string apiKey, bool paper, VenueSessionContext context)
    {
        _rest   = new ExampleRestClient(apiKey, paper);
        _socket = new ExampleSocketClient(apiKey, paper);
        LiveTicks = new ExampleTickSource(_socket);

        var composed = (context.BackfillComposer ?? DirectVenueBackfillComposer.Instance)
            .Compose(new VenueBackfillParts(
                ProviderKey:      "Example",
                VenueDisplayName: "Example Data",
                TimeBars:         new ExampleHistoricalBars(_rest),
                Ticks:            NoHistoricalTicks.Instance,    // this venue serves no tape
                Instruments:      _instruments,
                DirectBackfill:   new ExampleDirectBackfill(_rest),
                PeriodSupport:    ExamplePeriods.Instance,
                DefaultLookback:  TimeSpan.FromDays(30),
                Log:              context.Notices.Log));

        _backfill    = composed.Backfill;
        _tickDerived = composed.SupportsTickDerived;
    }

    public string Key => "Example";
    public string DisplayName => "Example Data";

    public ProviderCapabilities Capabilities =>
        ProviderCapabilities.Backfill
        | ProviderCapabilities.LiveTicks
        | ProviderCapabilities.InstrumentMetadata;

    public IBackfillProvider? Backfill => _backfill;
    public ITickSource? LiveTicks { get; }
    public IInstrumentMetadata? Instruments => _instruments;

    // BarSpecification is an abstract record with one concrete type per bar kind:
    // match on the type.
    public bool SupportsBarSpec(BarSpecification spec)
        => spec is TimeBarSpec || _tickDerived;

    public ProviderConnectionStatus Status { get; private set; }
        = ProviderConnectionStatus.Disconnected;
    public event Action<ProviderConnectionStatus>? ConnectionStatusChanged;
    public event Action? SocketRestored;

    public async Task ConnectAsync(CancellationToken cancellationToken = default)
    {
        Status = ProviderConnectionStatus.Connecting;
        ConnectionStatusChanged?.Invoke(Status);

        await _rest.AuthenticateAsync(cancellationToken).ConfigureAwait(false);
        await _socket.ConnectAsync(cancellationToken).ConfigureAwait(false);
        _socket.Restored += () => SocketRestored?.Invoke();

        await LoadCatalogAsync(cancellationToken).ConfigureAwait(false);

        Status = ProviderConnectionStatus.Connected;
        ConnectionStatusChanged?.Invoke(Status);
    }

    public async Task DisconnectAsync()
    {
        try { await _socket.DisconnectAsync().ConfigureAwait(false); } catch { }
        Status = ProviderConnectionStatus.Disconnected;
        ConnectionStatusChanged?.Invoke(Status);
    }

    public void Dispose() { _socket.Dispose(); _rest.Dispose(); }

    // ── IVenueInstrumentCatalog ────────────────────────────────────────────────
    public IReadOnlyList<VenueCatalogEntry> ListEntries()
    {
        lock (_entries) return _entries.ToArray();
    }

    public event Action? EntriesChanged;

    private async Task LoadCatalogAsync(CancellationToken ct)
    {
        var listing = await _rest.ListInstrumentsAsync(ct).ConfigureAwait(false);
        lock (_entries)
        {
            _entries.Clear();
            foreach (var i in listing)
            {
                // ONE call registers every facet; nulls stay unregistered.
                _instruments.Register(i.Symbol, i.TickSize,
                    pointValue: i.PointValue,
                    quoteCurrency: i.Currency,
                    category: i.IsFuture ? InstrumentCategory.Future : InstrumentCategory.Stock);

                _entries.Add(new VenueCatalogEntry(
                    InstrumentId: i.Symbol,
                    AssetClass:   i.IsFuture ? VenueAssetClass.Future : VenueAssetClass.Stock,
                    Exchange:     i.Mic,
                    Description:  i.Name,
                    ProductCode:  i.Root,
                    Expiration:   i.Expiry));
            }
        }
        EntriesChanged?.Invoke();
    }
}

The historical source

ExampleHistoricalBars.csinternal sealed class ExampleHistoricalBars(ExampleRestClient rest) : IHistoricalTimeBarSource
{
    // The venue keeps five years of bars and serves at most this many pages per request.
    private static readonly TimeSpan HistoryDepth = TimeSpan.FromDays(5 * 365);
    private const int MaxPages = 50;

    public async Task<HistoryServed<Bar>> LoadTimeBars(
        HistoryRequest request, TimeSpan period, CancellationToken ct = default)
    {
        if (!rest.IsAuthenticated)
            throw new BackfillSourceUnavailableException("Example Data is not connected.");

        // Clamp to what the venue actually holds; venue knowledge stays in the venue.
        var floor = DateTime.UtcNow - HistoryDepth;
        if (request.ToUtcExclusive <= floor)
            return HistoryServed<Bar>.None(request);   // older than anything the venue keeps
        var fromUtc = request.FromUtc < floor ? floor : request.FromUtc;

        var bars = new List<Bar>();
        string? cursor = null;
        for (int pages = 0; pages < MaxPages; pages++)
        {
            var page = await rest.LoadBarsAsync(request.InstrumentId, period, fromUtc,
                                                request.ToUtcExclusive, cursor, ct).ConfigureAwait(false);
            foreach (var b in page.Bars)
                bars.Add(new Bar(
                    StartUtc: b.OpenUtc,
                    EndUtc:   b.OpenUtc + period,
                    Open: b.Open, High: b.High, Low: b.Low, Close: b.Close,
                    Volume: b.Volume, TickCount: b.Trades));
            cursor = page.NextCursor;
            if (cursor is null)
                return HistoryServed<Bar>.Span(bars, request, fromUtc, request.ToUtcExclusive);
        }

        // The page budget ran out: vouch only for what arrived, so the rest is asked for again.
        var servedTo = bars.Count > 0 ? bars[^1].EndUtc : fromUtc;
        return HistoryServed<Bar>.Span(bars, request, fromUtc, servedTo);
    }
}

The tick source

ExampleTickSource.csinternal sealed class ExampleTickSource(ExampleSocketClient socket) : ITickSource
{
    private readonly ConcurrentDictionary<string, WireSubscription> _subs = new(StringComparer.Ordinal);

    public IDisposable Subscribe(string instrumentId, Action<Tick> onTick)
    {
        var sub = _subs.GetOrAdd(instrumentId, id => new WireSubscription(this, id));
        sub.Add(onTick);
        return new Handle(sub, onTick);
    }

    public void Dispose()
    {
        foreach (var sub in _subs.Values) sub.Dispose();
        _subs.Clear();
    }

    private sealed class WireSubscription : IDisposable
    {
        private readonly ExampleTickSource _owner;
        private readonly string _id;
        private readonly object _gate = new();
        private Action<Tick>[] _handlers = Array.Empty<Action<Tick>>();
        private IDisposable? _wire;

        public WireSubscription(ExampleTickSource owner, string id)
        {
            _owner = owner;
            _id = id;
        }

        public void Add(Action<Tick> handler)
        {
            lock (_gate)
            {
                _handlers = _handlers.Append(handler).ToArray();
                // First consumer opens the venue-side stream; the rest ride it.
                _wire ??= _owner.OpenWire(_id, Dispatch);
            }
        }

        public void Remove(Action<Tick> handler)
        {
            lock (_gate)
            {
                _handlers = _handlers.Where(h => h != handler).ToArray();
                if (_handlers.Length != 0) return;
                _wire?.Dispose();      // last consumer releases the venue stream
                _wire = null;
            }
        }

        // On the socket thread. No allocation, no locking across the callback.
        private void Dispatch(in Tick tick)
        {
            var handlers = Volatile.Read(ref _handlers);
            for (int i = 0; i < handlers.Length; i++) handlers[i](tick);
        }

        public void Dispose() { _wire?.Dispose(); _wire = null; }
    }

    private IDisposable OpenWire(string instrumentId, TickHandler onTick)
        => socket.SubscribeTrades(instrumentId, onTick);

    // The vendor socket client's callback shape: ticks by reference, no copy per event.
    internal delegate void TickHandler(in Tick tick);

    private sealed record Handle(WireSubscription Sub, Action<Tick> Handler) : IDisposable
    {
        public void Dispose() => Sub.Remove(Handler);
    }
}

The catalog

The catalog above is implemented directly on the data provider, so the host's AsCapability<IVenueInstrumentCatalog>() probe finds it with no extra wiring. Two details in that code carry weight:

• The entry set is rebuilt under a lock and published by copy: ListEntries returns a snapshot array, never the live list.
• EntriesChanged fires after the swap, so a consumer that re-reads on the event can never observe a half-built catalog.

The tests

A session test needs a VenueSessionContext. Its required services are two small interfaces, so a test builds the context from in-memory fakes:

ExampleSessionTests.cspublic sealed class ExampleSessionTests
{
    private static VenueSessionContext Context(string apiKey) => new(
        profile:  new VenueProfileInfo(Guid.NewGuid(), "test"),
        settings: new Dictionary<string, string> { [ExampleVenueDescriptor.FieldApiKey] = apiKey },
        notices:  new NullNotices(),
        storage:  new MemoryStorage());

    [Fact]
    public async Task Connect_Failure_On_Bad_Key_Is_Permanent()
    {
        await using var session = new ExampleVenueSession(Context("rejected-key"));

        await Assert.ThrowsAsync<InvalidOperationException>(
            () => session.ConnectAsync(CancellationToken.None));

        Assert.True(session.LastFailure!.Permanent);
        Assert.Equal(ProviderConnectionStatus.Failed, session.Status);
    }

    [Fact]
    public async Task Dispose_Before_Connect_Is_Safe()
    {
        var session = new ExampleVenueSession(Context(""));
        await session.DisposeAsync();      // must not throw: the dialog's cancel path relies on it
    }

    private sealed class NullNotices : IVenueNoticeSink
    {
        public void Post(VenueNotice notice) { }
        public void Log(string message) { }
    }

    private sealed class MemoryStorage : IVenueStateStorage
    {
        private readonly Dictionary<string, string> _files = new();
        public string? ReadText(string name) => _files.GetValueOrDefault(name);
        public void WriteText(string name, string content) => _files[name] = content;
    }
}

Where to go next