Venue feeds: presets and the resolver

A WebSocketFeed worker knows no venue and no consumer. It serves topics: it is configured with the topics it fills (topics, changed at runtime by its serve command) and maps each one to a venue item with its topicitems rules. This package keeps two things for it:

the presets: per venue, its connection protocol as plain feed config and which of its market families it serves on which channel (feeds/<preset>.json, embedded in Virtufin.MarketData.DevKit and packed under feeds/, shipped in virtufin-marketdata-devkit as package data and embedded in the npm package @virtufin/marketdata-devkit);
the resolver and orchestrator: the orchestrator-side library that turns the topics a set of workers listens to into the feeds that serve them, and opens, updates and deletes those feeds (FeedResolver, FeedOrchestrator; Python feeds.resolve, feeds.apply, feeds.delete; TypeScript the same names in feeds of @virtufin/marketdata-devkit).

Nothing in the platform matches topics to feeds: the WorkManager runs workers, the feed serves what it is told, the worker subscribes to its topics. Who opens which feed is decided by the orchestrator (a notebook, a script, the TUI), with this library.

Presets

Preset	Venue	URL	Login	Serves (family: channel)	Limits	Routing key
`binance`	`binance`	`wss://stream.binance.com:9443/stream`	none	`orderbook.update`: `depth@100ms`, `trade.update`: `trade`, `candle.update`: `kline_<interval>`	1024 streams	`stream` = `{symbol:lower}@{channel}`
`alpaca`	`alpaca`	`wss://stream.data.alpaca.markets/v1beta3/crypto/us`	`auth` with `{secret:alpacakey}`/`{secret:alpacasecret}`, expect `"msg":"authenticated"`, fail `"T":"error"`	`orderbook.update`: `orderbooks`, `trade.update`: `trades`	10 order books, 1 connection	`T,S` = `o\\|BTC/USD`, `t\\|BTC/USD`, arrays split
`fakevenue-binance`	`binance`	`wss://fake-venue.virtufin.com/ws`	none	`orderbook.update`: `depth@100ms`	none	as `binance`
`fakevenue-alpaca`	`alpaca`	`wss://fake-venue.virtufin.com/alpaca`	`auth` with a fixed key (the fake venue takes any)	`orderbook.update`: `orderbooks`	none	as `alpaca`

A preset file:

{
  "preset": "alpaca",
  "venue": "alpaca",
  "maxconnections": 1,
  "secrets": {"alpacakey": "the Alpaca API key id", "alpacasecret": "the Alpaca API secret"},
  "config": {"url": "wss://...", "handshake": [...], "subscribeframe": "...", "frameper": "channel",
             "maxitemsperchannel": {"orderbooks": 10}, "routing": {"key_path": "T,S", "split_arrays": true}},
  "topics": {
    "orderbook.update": {"channel": "orderbooks", "routingkey": "o|{symbol}"},
    "trade.update": {"channel": "trades", "routingkey": "t|{symbol}"}
  }
}

config is the feed's config, decoded (see the config schema). topics maps a family of the venue's naming fragment (<entity>.<event>) to its channel template and, where one connection carries several kinds of one instrument, its own routing key. maxconnections is how many connections the venue allows; the resolver never plans more.

The rules come from the naming fragments

VenueFeeds.Config(preset, overrides, world) (C#), feeds.config(preset, overrides, world=...) (Python) and feeds.config(preset, overrides, world) (TypeScript) render the feed config: the preset's config, symbolformat/symbolcase from the venue (virtufin.venue.<venue>), and topicitems from the venue's bound market entities. For binance in world act:

[
  {"topic": "act.exchange.binance.orderbook.update.{pair}", "channel": "depth@100ms"},
  {"topic": "act.exchange.binance.trade.update.{pair}", "channel": "trade"},
  {"topic": "act.exchange.binance.candle.update.{pair}.{interval}", "channel": "kline_{interval}", "values": {"interval": {"1mo": "1M"}}}
]

Each topic template is the entity's topic pattern with its identity parameters as placeholders; the venue's interval names (1M for the canonical 1mo) are the rule's values. No topic is written by hand, and a new venue is a new fragment binding plus a preset file here, never code in the feed.

Since 0.19.0 every family a feed serves is per instrument: trade.update is act.exchange.<venue>.trade.update.<PAIR> and candle.update is act.exchange.<venue>.candle.update.<PAIR>.<interval> (they were one topic for every pair), like orderbook.update. The topic is the whole dependency: a worker that reads BTC_USDT's trades subscribes to that topic, and the resolver knows which item to subscribe at the venue.

The resolver

var plan = FeedResolver.Resolve(topics, new FeedResolverOptions
{
    Owner = runGroup,                                                       // name prefix and owner tag
    Presets = new Dictionary<string, string> { ["binance"] = "fakevenue-binance" }, // optional, per venue
    Current = await orchestrator.CurrentAsync(runGroup),                    // optional: keep topics where they are
});

plan = feeds.resolve(topics, owner=run_group, presets={"binance": "fakevenue-binance"},
                     current=await feeds.current(run_group, api.gateway.workmanager))

import { feeds } from "@virtufin/marketdata-devkit";

const workManager = feeds.gatewayWorkManager(apiClient); // @virtufin/api ApiClient: invokeJson("workmanager", ...)
const plan = feeds.resolve(topics, { owner: runGroup, presets: { binance: "fakevenue-binance" },
                                     current: await feeds.current(runGroup, workManager) });

Step	What happens
input	topics (or the workers' catalog topics: `FeedResolver.TopicsOf(ListWorkers)`, `feeds.topics_of`, TypeScript `feeds.topicsOf`); templated ones are skipped
match	each topic is parsed by the naming library: a venue-bound market entity's event. Its venue's preset (default: the preset named like the venue) must serve its family; anything else is listed in `Unserved` with the reason (`orderbook.snapshot` is an HttpGet worker's, not a feed's)
group	per (preset, world): a feed's rules are rendered for one world
split	the group's topics, sorted, fill as few feeds as `maxitems` and `maxitemsperchannel` allow; a topic stays on the feed that serves it now (`Current`) while that feed has room
limits	at most `maxconnections` feeds per preset, over all its worlds (one at Alpaca: the account's limit); topics that do not fit are listed in `OverLimit` with the limit, and the plan is incomplete: `ThrowIfIncomplete()` / `raise_if_incomplete()` / `throwIfIncomplete()`, never a silent drop
names	`<owner>-<preset>-<world>-<n>` (`run-7-binance-act-1`, `run-7-binance-hyp-replay-1`): the same topics give the same plan in any order

The orchestrator

var feeds = new FeedOrchestrator(workManager, feedPackageUrl, SourceCredential.Virtufin, new FeedOrchestratorOptions
{
    Secrets = new Dictionary<string, IReadOnlyDictionary<string, string>>
    {
        ["alpaca"] = new Dictionary<string, string> { ["alpacakey"] = "alpaca-paper-key", ["alpacasecret"] = "alpaca-paper-secret" },
    },
    Config = new Dictionary<string, string> { ["publishtopic"] = unrouted, ["releasewhenidleminutes"] = "10" },
    Tags = new Dictionary<string, string> { ["source"] = "my-notebook" },
});
await feeds.ApplyAsync(plan);     // create or update by name; prune the owner's feeds the plan no longer names
await feeds.DeleteAsync(runGroup); // delete every feed the owner opened

await feeds.apply(plan, api.gateway.workmanager, feeds.package_source(url),
                  secrets={"alpaca": {"alpacakey": "alpaca-paper-key", "alpacasecret": "alpaca-paper-secret"}},
                  config={"publishtopic": unrouted}, tags={"source": "my-script"})
await feeds.delete(run_group, api.gateway.workmanager)

await feeds.apply(plan, workManager, feeds.packageSource(url), {
  secrets: { alpaca: { alpacakey: "alpaca-paper-key", alpacasecret: "alpaca-paper-secret" } },
  config: { publishtopic: unrouted }, tags: { source: "my-script" },
});
await feeds.delete(runGroup, workManager);

Create or update by name. For each feed of the plan, the running feed with that name tag (and the owner's feedowner tag) is sent serve {"topics": [...]}; if none runs, one is created (config autoreload=false, the preset for the plan's world, its topics), started and awaited. Running it again changes nothing; running it with other topics sends the feeds only the difference to the venue.
Order-independent. Workers subscribe to their topics whenever they start; a feed serves its topics whenever it runs. Apply before or after the workers, and again whenever their topics change.
Self-release. With releasewhenidleminutes a feed that served nothing for that long deletes itself. Its release and a serve are decided under one lock in the feed: a serve that lands first cancels the release, one that lands after it is refused (the feed reports releasing), and the apply then creates the feed again under its name. A release racing an apply always ends with exactly one feed serving the topics, never two and never none (tested on the devkit's LocalWorkManager with the real feed).
Prune and delete. The owner's feeds the plan no longer names are deleted; DeleteAsync(owner) / delete(owner, ...) / feeds.delete(owner, ...) removes every feed the owner opened, and only those.
Secrets are references into the worker Secret, never values; a preset whose login needs secrets that are not given is refused before anything is created.

examples/dotnet/platform.cs wraps both for notebooks (ServeTopicsAsync), examples/python/websocket_feed.py for the Python scripts (serve_topics), and examples/typescript/examples/websocketfeed.ts uses them directly.

TypeScript

@virtufin/marketdata-devkit (npm, the @virtufin scope on the Gitea registry, version = this repo's LIBRARY_VERSION) is the TypeScript side:

Export	What
`NAMING_FRAGMENTS`, `fragments()`, `marketRegistry()`	the `virtufin.market` and venue fragments, embedded from `naming/` at build; `fragments()` adds `virtufin.worlds` from `@virtufin/worker-devkit` (TypeScript has no fragment discovery: the caller composes, `NameRegistry.compose`)
`marketEntity(entity, venue)`	`MarketNames.At`: the market entity bound to a venue (`marketEntity("orderbook", "binance").topic("update", ACT, CurrencyPair.parse("BTC_USDT"))`)
`feeds.presets()`, `feeds.config()`, `feeds.topicItems()`, `feeds.channel()`, ...	`VenueFeeds`: the presets, embedded from `feeds/` at build
`feeds.resolve()`, `FeedPlan`, `FeedSpec`	`FeedResolver`
`feeds.apply()`, `feeds.current()`, `feeds.delete()`	`FeedOrchestrator`, through a `GatewayWorkManager` (`feeds.gatewayWorkManager(apiClient)`, or any object with the five WorkManager methods in the gateway's JSON mapping)

The package embeds the repo's files when it is built (scripts/embed.mjs), so it carries no copy that can drift; its tests check the embedded files against the repo's.

Conformance

conformance/feed-resolver.v1.json holds the vectors every language runs: the topicitems each preset renders per world, the channels, the feed names and the plans resolve makes (groups, sorting, unserved reasons, maxitems and maxitemsperchannel splits, Alpaca's one connection over all worlds, sticky current placement), each also resolved in reverse input order. C# FeedConformanceTests, Python test_feed_conformance.py and TypeScript test/conformance.test.ts read it; a behaviour change is a vector change first.

The schema

Every preset's config and every rendered config is the decoded form of the feed's config schema. feeds/schema/ holds a copy; the tests check it is identical to the one packed in the Virtufin.Worker.WebSocketFeed package they run the presets on, and validate every preset and its rendered config against it (C# and Python).