Pub/Sub Topics

Purpose¶

Canonical naming scheme for Dapr pub/sub topics and state keys across all Virtufin services. Defines the routing taxonomy, CloudEvents envelope contract, state-key taxonomy, and scenario-id axis. Subsumes the prior cross-cutting "Per-service Pubsub Topics" requirement.

Related specs: - Scenarios — scenario registry, triplet definition, LIVE bootstrap. - Cross-cutting — API-mediated pub/sub (envelope contract travels through the API gRPC). - Worker Management — worker + connection lifecycle events (workmanager.lifecycle, websocketmanager.lifecycle). - Proto-to-Client Mapping — code-generation conventions for the gRPC types referenced here.

Architecture¶

Two orthogonal routing axes plus a scenario-agnostic infrastructure axis:

Axis	Pattern	Scope	Example
Market data world	`act.exchange.<venue>.<entity>.<event>[.<id>]` or `hyp.<name>.exchange.<venue>.<entity>.<event>[.<id>]`	data-universe-bound; shared across scenarios	`act.exchange.binance.orderbook.update.BTCUSDT`
Scenario world	`sc.<scenarioId>.<domain>.<entity>.<event>[.<id>]`	triplet-bound (market + portfolio + strategy)	`sc.LIVE.trading.order.filled.<order_id>`
Tier 0 / infrastructure	`<service>.<category>`	scenario-agnostic	`websocketmanager.lifecycle`, `state.change`

Rationale: - Market data is world-scoped because it depends on the data universe (live vs hypothetical replay) and is shareable across scenarios. Two paper runs sharing the same Binance feed do not double-publish ticks. - Trading / position / risk / pnl are scenario-scoped because each depends on the full triplet. P&L in particular is meaningless without knowing all three legs. - Tier 0 is scenario-agnostic because infra lifecycle (workers up, connections reconnecting) is a deployment concern, not a trading concern.

act and hyp.<name> are both top-level world markers. act is the single actual reality (act.exchange.*); a hypothetical world carries a <name> segment identifying which hypothetical reality is in play (hyp.BACKTEST.exchange.*). Bare hyp (no name) is never a valid topic segment — it denotes the category of all hypothetical worlds, not a specific one (see Scenarios spec §Named Hypothetical Worlds). The data universe (ALL, EQ_EUROPE, EQ_2008_2009, …) is not encoded in the topic — it travels in the marketuniverse CloudEvent envelope extension (see below) and on the producing WebSocketManager connection's universe tag. The scenario triplet (market + portfolio + strategy legs) lives in the scenario registry.

Requirements¶

Requirement: Topic Patterns¶

Services SHALL publish to and subscribe from topics matching one of the following patterns, to keep the taxonomy consistent and routable across the platform. This is a naming convention, not an enforced contract: no service (publisher, subscriber, or virtufin-api itself) validates or rejects topics against these patterns before publish -- see cross-cutting spec for why no enforcement point owns this check.

Scenario: Tier 0 / infrastructure topic¶

WHEN a service publishes a lifecycle / infra event
THEN the topic SHALL match <service>.<category> (e.g. websocketmanager.lifecycle, workmanager.lifecycle)
AND the <category> SHALL be a domain noun (lifecycle, infra, audit) — not a state name

Scenario: Market data topic (real)¶

WHEN a service publishes real (live) market data
THEN the topic SHALL match act.exchange.<venue>.<entity>.<event>[.<entity_id>]
AND the data universe SHALL be carried in the marketuniverse envelope extension (ALL for unrestricted, EQ_EUROPE for a live subset) — it is not a topic segment

Scenario: Market data topic (hypothetical)¶

WHEN a service publishes hypothetical (replay / backtest) market data
THEN the topic SHALL match hyp.<name>.exchange.<venue>.<entity>.<event>[.<entity_id>]
AND <name> SHALL identify the specific hypothetical world (see Scenarios spec §Named Hypothetical Worlds) — bare hyp (no name) SHALL NOT appear in a topic
AND the data universe SHALL be carried in the marketuniverse envelope extension, independent of <name> — the same universe MAY appear under different hypothetical worlds and vice versa

Scenario: Trading topic¶

WHEN a service publishes a trading-domain event (order lifecycle, trade lifecycle)
THEN the topic SHALL match sc.<scenarioId>.trading.<order|trade>.<event>.<id>
AND <scenarioId> SHALL be either the reserved LIVE or a registered opaque scenario ID

Scenario: Position topic¶

WHEN a service publishes a position-domain event
THEN the topic SHALL match sc.<scenarioId>.position.<symbol>.<event> or sc.<scenarioId>.position.portfolio.<event>

Scenario: Risk topic¶

WHEN a service publishes a risk-domain event
THEN the topic SHALL match sc.<scenarioId>.risk.<metric>.<event> (consolidated; per-symbol routing via subject)

Scenario: P&L topic¶

WHEN a service publishes a P&L-domain event
THEN the topic SHALL match sc.<scenarioId>.pnl.<metric>.<event> (consolidated; per-symbol routing via subject)

Requirement: No Topic or Scenario-State Validation Anywhere¶

No service — publisher, subscriber, virtufin-api's Pubsub.PublishEvent handler, or any shared helper — SHALL validate a topic's shape against the patterns above, or a scenario ID's registration/lifecycle status, before publish. This was previously a per-service inline requirement (added by the 2026-07-09 pubsub-pattern-fixes change); reversed because it duplicated Virtufin domain vocabulary (universes, scenario IDs, lifecycle status) into every publishing service's own code, including services whose charter is explicitly "pure infrastructure, no domain logic" (WebSocketManager, WorkManager). virtufin-websocketmanager's TopicValidator was the one place this was actually implemented; it has been removed in full.

Scenario: Malformed or unregistered topic is published anyway¶

WHEN a service publishes to a topic that matches no pattern in this spec, or to an sc.<scenarioId>.* topic whose scenarioId is not registered (or is paused/archived/deleted)
THEN the publish SHALL proceed; Dapr/NATS is the only enforcement point, accepting whatever it considers a legal topic string
AND detecting and correcting a bad topic or stale scenario reference is a monitoring/observability concern (dead-letter, consumer-side logging), not a publish-time rejection

Requirement: CloudEvents Envelope¶

Every event SHALL be a CloudEvents v1.0 envelope. The CE base attributes (id, source, type, time, subject when applicable, specversion, datacontenttype) SHALL be carried as ce- prefixed metadata; data SHALL be the JSON payload.

Scenario: ce-type format¶

WHEN a service publishes an event
THEN ce-type SHALL match one of:
Tier 0: com.virtufin.<service>.<category>.<event> (e.g. com.virtufin.websocketmanager.lifecycle.connected)
Market data: com.virtufin.exchange.<venue>.<entity>.<event> (e.g. com.virtufin.exchange.binance.orderbook.updated)
Trading: com.virtufin.trading.<order|trade>.<event> (e.g. com.virtufin.trading.order.filled)
Position: com.virtufin.position.<event> (e.g. com.virtufin.position.flipped)
Risk: com.virtufin.risk.<event> (e.g. com.virtufin.risk.exposure.updated)
P&L: com.virtufin.pnl.<event> (e.g. com.virtufin.pnl.realized.updated)

Scenario: ce-source format¶

WHEN a service publishes an event
THEN ce-source SHALL be the publishing service's URN (e.g. urn:com.virtufin.websocketmanager, urn:com.virtufin.trading-engine)
AND the URN SHALL be stable per service (not per instance)

Scenario: subject attribute (per-entity events)¶

WHEN a service publishes an event referring to a specific entity (order, trade, position, symbol)
THEN ce-subject SHALL be set to <entity_type>/<entity_id> (e.g. order/<order_id>, position/BTCUSDT, currency_pair/BTCUSDT)
AND consolidated topics (risk, pnl, portfolio) SHALL also use subject to enable per-entity routing without topic proliferation

Requirement: Common Envelope Extensions¶

Every Tier 1 event SHALL carry the following CloudEvents extensions. Tier 0 events MAY carry correlationid only.

Extension	Type	Required	Notes
`correlationid`	UUID	Tier 1 yes; Tier 0 no	Ties fills back to originating order across services
`scenarioid`	string	Tier 1 yes	Opaque registry key; `LIVE` for production
`runid`	UUID	Tier 1 yes when parallel backtests	Disambiguates concurrent runs sharing a scenario
`exchange`	string	Tier 1 yes (when applicable)	Exchange name (`binance`, `coinbase`, …); distinct from `ce-source`
`clocktype`	`wall` \| `historical`	Tier 1 yes	`wall` for prod + paper; `historical` for backtest replay
`eventtime`	RFC3339	Tier 1 yes (= `ce-time`)	Logical event time matching `clocktype`
`walltime`	RFC3339	Tier 1 yes	Always real clock at publish — for tracing / logging
`marketworld`	`act` \| `hyp.<name>`	Tier 1 yes	Triplet leg (world) for market
`marketuniverse`	string	Tier 1 yes	Triplet leg (universe) for market; present for both `act` and `hyp.<name>` (use `"ALL"` for unrestricted)
`portfolioworld`	`act` \| `hyp.<name>`	Tier 1 yes	Triplet leg for portfolio
`strategyworld`	`act` \| `hyp.<name>`	Tier 1 yes	Triplet leg for strategy

CloudEvents extension attribute names are restricted to lowercase alphanumerics — no dots, no underscores. These four are deliberately concatenated (marketworld, not market.world or market_world) to satisfy that constraint, matching every other extension in this table.

Scenario: Envelope on a production fill event¶

WHEN a service publishes sc.LIVE.trading.order.filled.<order_id>
THEN the CloudEvent extensions SHALL include scenarioid=LIVE, runid=<uuid>, correlationid=<uuid>, exchange=binance, clocktype=wall, eventtime=<rfc3339>, walltime=<rfc3339>, marketworld=act, marketuniverse=ALL, portfolioworld=act, strategyworld=act

Scenario: Envelope on a backtest fill event¶

WHEN a backtest engine publishes sc.S7F3A9B.trading.order.filled.<order_id>
THEN the envelope SHALL include scenarioid=S7F3A9B, clocktype=historical, eventtime=<historical rfc3339> (the replayed tick's timestamp), walltime=<real rfc3339> (now), marketworld=hyp.BACKTEST, marketuniverse=EQ_2008_2009, portfolioworld=hyp.BACKTEST, strategyworld=hyp.BACKTEST
AND runid SHALL differ for concurrent backtests against the same scenario

Scenario: clocktype rationale¶

WHEN a strategy computes a time-series indicator (e.g. 20-period moving average) from a backtest stream
THEN it SHALL use eventtime (which is historical for backtests) to maintain correct indicator semantics
AND it SHALL use walltime only for tracing, logging, and alerting

Requirement: Trade Lifecycle Events¶

The trading domain SHALL model the lifecycle: submitted → accepted → (partially_filled →)* filled → cleared → settled. For Binance CEX, cleared and settled collapse to near-instantaneous but the events remain distinct for future DeFi / TradFi integration.

Scenario: Order submitted¶

WHEN a service submits an order to an exchange
THEN it SHALL publish sc.<scenarioId>.trading.order.submitted.<order_id> with ce-type = com.virtufin.trading.order.submitted
AND the data payload SHALL include order_id, side (buy|sell), type (market|limit), qty, price?, venue, submitted_at

Scenario: Order accepted¶

WHEN the venue acknowledges the order
THEN the service SHALL publish sc.<scenarioId>.trading.order.accepted.<order_id> with venue_order_id, accepted_at

Scenario: Order rejected¶

WHEN an order is rejected by pre-trade risk or the venue
THEN the service SHALL publish sc.<scenarioId>.trading.order.rejected.<order_id> with reject_reason, reject_source (risk|venue)

Scenario: Order amended (mutation, not transition)¶

WHEN an order's price or quantity is changed in-place
THEN the service SHALL publish sc.<scenarioId>.trading.order.amended.<order_id> with new_price?, new_qty?, amended_at
AND the order's status SHALL remain unchanged (pending or partially_filled)

Scenario: Order cancelled¶

WHEN an order is cancelled by the user or system
THEN the service SHALL publish sc.<scenarioId>.trading.order.cancelled.<order_id> with cancelled_at, cancelled_by (user|system|venue)

Scenario: Order expired¶

WHEN an order's time-in-force is reached without a fill
THEN the service SHALL publish sc.<scenarioId>.trading.order.expired.<order_id> with expired_at

Scenario: Order filled (workhorse event)¶

WHEN a partial or full fill is received from the venue (Binance WebSocket fill callback)
THEN the service SHALL publish sc.<scenarioId>.trading.order.filled.<order_id> with fill_id, qty_filled, price, fee, liquidity (maker|taker), remaining_qty, filled_at, realized_pnl? (optional, included if computed at fill time)
AND for partial fills, remaining_qty SHALL be the post-fill remaining quantity (decreasing across events)
AND when remaining_qty == 0, the order is implicitly fully filled (no separate order.completed event)

Scenario: Trade cleared¶

WHEN an order is fully filled and counterparty obligations are netted
THEN the service SHALL publish sc.<scenarioId>.trading.trade.cleared.<trade_id> with trade_id, order_id, cleared_at
AND this event SHALL fire exactly once per trade (not per fill)

Scenario: Trade settled¶

WHEN asset transfer for a trade is final
THEN the service SHALL publish sc.<scenarioId>.trading.trade.settled.<trade_id> with trade_id, order_id, settled_at, tx_hash? (null for CEX, populated for on-chain settlement)
AND for Binance CEX, cleared and settled SHALL be emitted back-to-back by the publisher
AND for DeFi / TradFi integration, the gap between cleared and settled may span minutes to days

Requirement: Position Lifecycle Events¶

Position state SHALL be derived from signed size plus a liquidation sub-state. position.flipped SHALL be a first-class atomic event because the risk engine needs direction-change awareness without a flat interval.

Scenario: Position opened¶

WHEN a new position is established (size transitions from 0 to non-zero)
THEN the service SHALL publish sc.<scenarioId>.position.<symbol>.opened with side (long|short), qty, avg_price, opened_at

Scenario: Position increased¶

WHEN an existing position grows in size (same side)
THEN the service SHALL publish sc.<scenarioId>.position.<symbol>.increased with new_qty, new_avg_price, delta_qty

Scenario: Position decreased¶

WHEN an existing position shrinks in size (same side, partial close)
THEN the service SHALL publish sc.<scenarioId>.position.<symbol>.decreased with new_qty, realized_pnl, delta_qty

Scenario: Position flipped (atomic direction change)¶

WHEN a position changes from long to short or vice versa in a single fill sequence
THEN the service SHALL publish sc.<scenarioId>.position.<symbol>.flipped with from_side, to_side, qty, price, order_id, realized_pnl
AND this event SHALL be atomic — no intermediate position.closed + position.opened sequence

Scenario: Position closed¶

WHEN a position returns to flat (size = 0) via user close
THEN the service SHALL publish sc.<scenarioId>.position.<symbol>.closed with realized_pnl, closed_at

Scenario: Position liquidated (forced)¶

WHEN a position is force-closed by the venue
THEN the service SHALL publish sc.<scenarioId>.position.<symbol>.liquidated with liquidation_price, realized_pnl, liquidated_at

Scenario: Liquidation warning (signal, not transition)¶

WHEN margin ratio crosses a warning threshold
THEN the service SHALL publish sc.<scenarioId>.position.<symbol>.liquidation.warning with margin_ratio, threshold
AND the position's primary state SHALL NOT change; only the liquidation_status sub-state becomes warning

Scenario: Position adjustment (corporate action)¶

WHEN a corporate action (split, dividend, symbol change) mutates a position's size or avg_price
THEN the service SHALL publish sc.<scenarioId>.position.<symbol>.adjustment with adjustment_type (split|dividend|symbol_change), new_qty, new_avg_price, corporate_action_id

Scenario: Portfolio snapshot¶

WHEN a full portfolio snapshot is published (consolidated)
THEN the service SHALL publish sc.<scenarioId>.position.portfolio.snapshot with total_value, total_pnl, position_count

Requirement: Market Data Events¶

Market data events SHALL live under the world-scoped pattern (act.exchange.<venue>.* or hyp.<name>.exchange.<venue>.*) — hyp carries one more segment than act (the <name> slot); neither carries the data universe, which travels in the marketuniverse envelope extension. Per-currency-pair cardinality SHALL be used for orderbook.*, trade.tape, ticker.update, funding.rate, mark.price, ohlcv.candle.*. Consolidated (no per-pair segment) for instrument.info.

Scenario: Order book snapshot¶

WHEN the venue publishes a full order book snapshot at depth N
THEN the service SHALL publish act\|hyp.<name>.exchange.<venue>.orderbook.snapshot.<currency_pair> (act omits <name>) with bids[], asks[], last_update_id, snapshot_at

Scenario: Order book update (delta)¶

WHEN the venue publishes an L2/L3 order book delta
THEN the service SHALL publish act\|hyp.<name>.exchange.<venue>.orderbook.update.<currency_pair> (act omits <name>) with bids_delta[], asks_delta[], first_update_id, last_update_id

Scenario: Public trade tape¶

WHEN a public trade occurs on the venue
THEN the service SHALL publish act\|hyp.<name>.exchange.<venue>.trade.tape.<currency_pair> (act omits <name>) with trade_id, price, qty, side, trade_time

Scenario: Ticker update¶

WHEN the venue publishes 24h rolling stats
THEN the service SHALL publish act\|hyp.<name>.exchange.<venue>.ticker.update.<currency_pair> (act omits <name>) with last_price, volume_24h, change_24h_pct

Scenario: Funding rate¶

WHEN the venue publishes a perpetual funding rate tick
THEN the service SHALL publish act\|hyp.<name>.exchange.<venue>.funding.rate.<currency_pair> (act omits <name>) with funding_rate, next_funding_time

Scenario: Mark price¶

WHEN the venue publishes a mark price update (derivatives)
THEN the service SHALL publish act\|hyp.<name>.exchange.<venue>.mark.price.<currency_pair> (act omits <name>) with mark_price, index_price

Scenario: OHLCV candle¶

WHEN a candle interval closes
THEN the service SHALL publish act\|hyp.<name>.exchange.<venue>.ohlcv.candle.<interval>.<currency_pair> (act omits <name>) with open, high, low, close, volume, candle_close_time

Scenario: Instrument info¶

WHEN instrument metadata changes (lot size, tick size, status)
THEN the service SHALL publish act\|hyp.<name>.exchange.<venue>.instrument.info (act omits <name>) with symbol, status, base_asset, quote_asset, lot_size, tick_size

Requirement: Risk Events¶

Risk events SHALL be consolidated (one topic per metric). Per-symbol routing SHALL use ce-subject.

Scenario: Exposure update¶

WHEN risk greeks (delta / gamma / vega / theta) are recomputed
THEN the service SHALL publish sc.<scenarioId>.risk.exposure.update with greeks: { delta, gamma, vega, theta }, underlyings[]

Scenario: Margin update¶

WHEN margin used or available changes
THEN the service SHALL publish sc.<scenarioId>.risk.margin.update with margin_used, margin_available, margin_ratio

Scenario: Drawdown alert¶

WHEN drawdown crosses a configured threshold
THEN the service SHALL publish sc.<scenarioId>.risk.drawdown.alert with drawdown_pct, threshold_pct, peak_value, current_value

Scenario: Limit breach¶

WHEN a configured risk limit is breached
THEN the service SHALL publish sc.<scenarioId>.risk.limit.breach with limit_name, limit_value, actual_value, severity

Scenario: Scenario P&L (stress test)¶

WHEN a stress / scenario P&L computation completes
THEN the service SHALL publish sc.<scenarioId>.risk.scenario.pnl with scenario_name, pnl, scenario_inputs

Requirement: P&L Events¶

P&L events SHALL be consolidated. The full triplet context is carried in the envelope (not the topic) because P&L is meaningless without it.

Scenario: Realized P&L update¶

WHEN realized P&L is recomputed (e.g. on fill or position close)
THEN the service SHALL publish sc.<scenarioId>.pnl.realized.update with realized_pnl, by_symbol: { <symbol>: <pnl> }, period (daily|session|cumulative)

Scenario: Unrealized P&L update¶

WHEN unrealized P&L is recomputed (mark-to-market)
THEN the service SHALL publish sc.<scenarioId>.pnl.unrealized.update with unrealized_pnl, by_symbol: { <symbol>: <pnl> }, mark_prices: { <symbol>: <price> }

Requirement: Tier 0 / Infrastructure Events¶

Tier 0 events SHALL be scenario-agnostic and SHALL follow the <service>.<category> pattern. Access control SHALL move to NATS account / subject permissions (no public / internal segment in the topic).

Scenario: Worker lifecycle event¶

WHEN a worker changes state (created, started, online, degraded, stopping, stopped, failed)
THEN the WorkManager SHALL publish to workmanager.lifecycle with ce-type = com.virtufin.workmanager.lifecycle.<state>
AND the event SHALL carry worker_id, group?, topic, and state-specific fields (error_type?, error_message?)

Scenario: Connection lifecycle event¶

WHEN a WebSocket connection changes state (registered, connecting, connected, degraded, reconnecting, closed, failed)
THEN the WebSocketManager SHALL publish to websocketmanager.lifecycle with ce-type = com.virtufin.websocketmanager.lifecycle.<state>
AND the event SHALL carry connection_id, url, instance_id, and state-specific fields

Scenario: Infra events (connection / worker detail)¶

WHEN finer-grained infra events are needed (heartbeats, grouplock changes)
THEN the service SHALL publish to <service>.infra.<sub_category> (e.g. websocketmanager.infra.connection.heartbeat, workmanager.infra.worker.heartbeat, workmanager.infra.grouplock)

Requirement: Cardinality Rule¶

Services SHALL prefer consolidated topics (one topic, per-entity routing via ce-subject) over per-entity topics (.<entity_id> segment). Per-entity topics SHALL be used only when subscribers need broker-level per-entity subscription or routing.

Scenario: Per-entity topic justified¶

WHEN subscribers need to subscribe at the broker level to a single entity (e.g., fills for one specific order, order book for one specific pair)
THEN the topic SHALL carry the entity ID segment (trading.order.filled.<order_id>, exchange.binance.orderbook.update.<currency_pair>)

Scenario: Consolidated topic preferred¶

WHEN subscribers typically want aggregate views (P&L totals, risk limits, portfolio snapshots)
THEN the topic SHALL be consolidated and ce-subject SHALL carry the entity reference (risk.exposure.update with subject=position/BTCUSDT)

Scenario: act and hyp carry no universe segment¶

WHEN a topic uses the act.exchange.<venue>.* or hyp.<name>.exchange.<venue>.* pattern
THEN the hyp form SHALL have exactly one more segment than the act form (the <name> slot)
AND the data universe SHALL NOT appear in either topic — it travels in the marketuniverse envelope extension, not as a topic segment
AND brokers and wildcard subscriptions SHALL NOT assume segment-count parity between the two forms — subscribe to act.exchange.* and hyp.*.exchange.* as two separate patterns rather than one shared positional pattern

Requirement: NATS Permissions (replacing pub/prv)¶

Access control SHALL be enforced via NATS account / subject permissions, not via topic naming. pub / prv segments are NOT part of any topic pattern.

Scenario: Market data permissions¶

WHEN NATS permissions are configured
THEN act.exchange.* SHALL allow publishers = exchange connectors, subscribers = all trading / risk / backtest services
AND hyp.*.exchange.* SHALL allow publishers = backtest engines, subscribers = strategy services (no production services) — note the extra * for the <name> segment versus act

Scenario: Scenario event permissions¶

WHEN NATS permissions are configured
THEN sc.*.trading.* SHALL allow publishers = trading / strategy services, subscribers = risk / position / pnl / dashboards
AND sc.*.position.* SHALL allow publishers = trading services, subscribers = risk / pnl
AND sc.*.risk.* and sc.*.pnl.* SHALL allow publishers = risk / pnl services, subscribers = dashboards / alerting
AND sc.LIVE.* writes SHALL require elevated permissions (production role)

Scenario: Tier 0 permissions¶

WHEN NATS permissions are configured
THEN <service>.lifecycle and <service>.infra.* SHALL allow publishers = owning service, subscribers = all services

Requirement: State Keys¶

State keys SHALL share the topic taxonomy. After the <app-name>|| prefix:

Namespace	Pattern	Example
Tier 0 / infra	`<service>.<entity>[.<id>]`	`workmanager.worker.<worker_id>`
Market data cache	`act\\|hyp.<name>.exchange.<venue>.<entity>[.<id>]`	`act.exchange.binance.orderbook.BTCUSDT` (act omits `<name>`)
Scenario state	`sc.<scenarioId>.<domain>.<entity>[.<id>]`	`sc.LIVE.trading.order.<order_id>`
Scenario registry	`scenario.<scenarioId>`	`scenario.LIVE`

Scenario: Order state key¶

WHEN a service writes order state
THEN the key SHALL be sc.<scenarioId>.trading.order.<order_id>
AND the value SHALL be JSON { status, side, type, qty, remaining_qty, avg_price, venue, created_at, updated_at }

Scenario: Trade state key¶

WHEN a service writes trade state
THEN the key SHALL be sc.<scenarioId>.trading.trade.<trade_id>
AND the value SHALL be JSON { status, order_id, cleared_at?, settled_at?, tx_hash? }

Scenario: Position state key¶

WHEN a service writes position state
THEN the key SHALL be sc.<scenarioId>.position.<symbol>
AND the value SHALL be JSON { size, avg_price, realized_pnl, liquidation_status, updated_at }

Scenario: Portfolio state key¶

WHEN a service writes portfolio state
THEN the key SHALL be sc.<scenarioId>.position.portfolio
AND the value SHALL be JSON { total_value, total_pnl, position_count, updated_at }

Scenario: Indicator state key¶

WHEN a service persists a deployed indicator's current value (e.g. IndicatorWorkerBase in virtufin-indicator-devkit)
THEN the key SHALL be sc.<scenarioId>.indicator.<name>.<symbol> (e.g. sc.LIVE.indicator.sma10.BTCUSDT)
AND the value SHALL be JSON { value, updated_at } (the indicator's own encoding of its current value, plus a write timestamp)
AND this key is scenario-scoped rather than market-data-cache-scoped (contrast the "Order book state key" scenario below): an indicator fed by scenario-specific replayed/synthetic data under hyp.* needs run disambiguation two concurrent backtests of the same symbol would otherwise collide on — an order book has no such need, since it's a genuinely singular venue-level fact regardless of observer

Scenario: Worker state key¶

WHEN the WorkManager writes worker state
THEN the key SHALL be workmanager.worker.<worker_id>
AND the value SHALL be JSON { status, last_heartbeat, started_at, ... }

Scenario: Connection state key¶

WHEN the WebSocketManager writes connection state
THEN the key SHALL be websocketmanager.connection.<connection_id>
AND the value SHALL be JSON { status, last_heartbeat, connected_at, ... }

Scenario: Order book state key¶

WHEN the WebSocketManager writes per-pair order book cache
THEN the key SHALL be act\|hyp.<name>.exchange.<venue>.orderbook.<currency_pair> (act omits <name>)
AND the value SHALL be JSON { bids, asks, last_update_id } (consolidated, single source of truth per venue + pair)

Scenario: Scenario registry key¶

WHEN a service writes scenario metadata
THEN the key SHALL be scenario.<scenarioId>
AND the value SHALL be the scenario triplet definition (see Scenarios spec)

Requirement: Source-of-Truth Rule¶

Each piece of derived or owned state SHALL have exactly one writer.

Scenario: Single publisher per status field¶

WHEN two publishers attempt to write the same status field on the same entity (e.g. both writing order.status)
THEN this SHALL be treated as a design bug — the entity has exactly one owner for status transitions
AND consumers SHALL treat the status field as monotonically advancing through the state machine

Scenario: Single owner per derived metric¶

WHEN multiple services could compute the same derived metric (e.g. portfolio total value)
THEN exactly one aggregator SHALL own the write; other services SHALL subscribe to the topic for read access

Requirement: Delete-on-Deploy (No Deprecation)¶

Topic and state-key renames SHALL NOT carry a deprecation window. Old names SHALL be deleted on the same commit that introduces new names.

Scenario: Rename lifecycle¶

WHEN a topic or state-key name is changed
THEN the change SHALL be a single commit per affected repo
AND old names SHALL be removed in the same commit (no parallel publishing under both names)
AND the migration document SHALL list old → new mappings

Requirement: Multi-Publisher Tier 2¶

Subscribers of Tier 2 (scenario-scoped) topics SHALL NOT assume single-source. A scenario topic MAY have multiple publishers (e.g., aggregator + per-leg publisher).

Scenario: Multiple publishers on same topic¶

WHEN two services publish to the same scenario topic (e.g., a strategy publishes order.filled and a replayer also publishes the same event during replay)
THEN consumers SHALL handle events from either source
AND ce-source SHALL identify the publishing service for filtering / dedup

Requirement: Run-ID in Envelope, Not Topic¶

Backtest replay and parallel scenarios SHALL NOT introduce run-id segments into topics. Run isolation SHALL be achieved via the runid envelope extension.

WHEN two backtests run concurrently against the same scenario (sc.S7F3A9B)
THEN both SHALL publish to sc.S7F3A9B.trading.*
AND each event SHALL carry a distinct runid extension
AND consumers that want a single backtest's events SHALL filter by runid

Scenario: Run-id dropped from topic¶

WHEN a service migrates from binance-depth.<run_id> to the new scheme
THEN the topic SHALL become act.exchange.binance.orderbook.update.<currency_pair> (no run-id segment)
AND the run-id SHALL travel in the runid envelope extension (or be omitted for production traffic)

Requirement: Subscription Implementation Pattern¶

Dapr-based subscribe handlers in any Virtufin service SHALL open one Dapr subscription per topic per service process and fan out the resulting event stream in-process to every interested consumer within that process. Per-call Dapr subscriptions (one per gRPC call or one per caller) are PROHIBITED because they cause the same event to be silently load-balanced across callers instead of broadcast to each.

Scenario: Multiple callers on one topic¶

WHEN two gRPC clients subscribe to the same topic in the same API process
THEN each caller SHALL receive every message on that topic (true 1:many broadcast within the process)
AND the Dapr sidecar SHALL see exactly one subscription per topic per process (competing consumers between API replicas is the only competing-consumer layer)

Scenario: New caller arrives¶

WHEN a third gRPC client subscribes to a topic that already has two subscribers in the same process
THEN the existing Dapr subscription SHALL be reused; the new caller's in-process handler SHALL be added to the fan-out list; no new Dapr subscription SHALL be opened

Scenario: Last caller leaves¶

WHEN the last gRPC client unsubscribes from a topic
THEN the in-process fan-out registration SHALL be removed; the Dapr subscription SHALL be disposed; subsequent callers SHALL open a new Dapr subscription on demand

Scenario: In-process exception on a subscriber¶

WHEN an in-process subscriber's handler throws an exception while processing a fanned-out message
THEN the broker SHALL continue delivering to the remaining subscribers (per-subscriber fault isolation)
AND the failed subscriber SHALL be logged but SHALL NOT cause the message to be re-delivered to the surviving subscribers

Services that consume pub/sub messages SHALL guarantee at-least-once processing with a bounded retry count, followed by dead-letter publication on exhaustion — messages SHALL NOT be silently lost on an uncaught processing exception. Where this guarantee is enforced depends on the consumption model:

A service holding its own direct Dapr subscription (the classic HTTP subscribe model) SHALL return TopicResponseAction.Retry on uncaught exceptions so Dapr's maxRedeliveryCount and dead-letter topic configuration take effect. Drop SHALL be reserved for explicit business-logic skips (e.g., "no local worker registered for this topic"); it SHALL NOT be used to silently discard a message on an uncaught exception.
A service consuming pub/sub via the streaming Pubsub.Subscribe gRPC API (see the "Subscription Implementation Pattern" requirement above) is one of potentially several in-process fan-out consumers per Dapr subscription; the Gateway acks Dapr's broker before any consumer processes the message, so broker-level redelivery is not available to it — retrying at the broker would double-process the fan-out's other, already-succeeded consumers. Such a consumer SHALL implement its own bounded-retry-then-dead-letter mechanism at the application level: retry the failed operation up to a configured attempt count with backoff, and on exhaustion publish the original message to the configured dead-letter topic (default <topic>-dead) rather than discarding it.

Scenario: Worker throws on every message¶

WHEN a service holding its own direct Dapr subscription throws on every message it receives
THEN Dapr SHALL re-deliver the message up to maxRedeliveryCount (default 3) times
AND after the redelivery cap is reached the message SHALL be published to the configured dead-letter topic (default <topic>-dead)
AND the handler's process SHALL NOT terminate

Scenario: Explicit business-logic skip¶

WHEN a handler determines the message is not relevant (e.g., no local worker on the topic) and explicitly chooses to skip it
THEN the handler MAY return TopicResponseAction.Drop and the message SHALL be acked with no redelivery

Scenario: Handler times out¶

WHEN a handler does not return within MessageHandlingPolicy timeout (default 30 s)
THEN Dapr SHALL re-deliver the message (Retry policy)
AND the timed-out handler's process SHALL remain responsive

Scenario: Streaming-API consumer throws on every message¶

WHEN a consumer of the streaming Pubsub.Subscribe API throws processing every message on a topic
THEN the consumer SHALL retry the operation up to its configured attempt count with backoff
AND after attempts are exhausted the consumer SHALL publish the original message to the configured dead-letter topic (default <topic>-dead)
AND the consumer's process SHALL NOT terminate and SHALL continue processing subsequent messages