RTDS (Real-Time Data Stream) — Operator Summary

Operator-facing summary of the ADR-004 RTDS integration in pm-engine. For the activation procedure, see docs/operations/rtds-cutover-runbook.md.

What it does

pm-engine can subscribe to Polymarket's public market-wide activity feed (wss://ws-live-data.polymarket.com) and route every trade / orders_matched print into two consumers:

• Actor-tagged SELL-print buffer — feeds actor-tagged SELL prints into the directional stop-loss trade-confirm gate so the engine can exclude its OWN stop-SELL fills before they self-confirm a stop (RtdsActorBuffer, own-wallet excluded at the read seam).
• Toxic-flow detector — sliding-window one-sided pressure on a condition_id yields Decision::CancelOrder for our open orders on that cid. Operator-gated behind two independent config.db knobs (directional.toxic_flow.log_only then active_cancel_enabled).

The transport client + parser are shared with pm-sdk (pm_sdk::rtds — RtdsClient::spawn(cfg) / RtdsClient::spawn_with_metrics(cfg, metrics), RtdsClientConfig).

Сквозной транспорт, допуск по токену (rtds-transport-remediation)

pm-sdk больше не фильтрует ничего сам. До этого изменения клиент держал собственный пул слагов (FilterSet) и молча ронял каждый разобранный кадр, чей slug/event_slug не совпадал с пулом, — эта проверка дублировала допуск движка и, на живом, но тихом отслеживаемом рынке, заставляла транспорт объявлять эпоху непокрытой из-за пустого совпадения, а не из-за реальной проблемы сокета. FilterSet, update_filters и DropReason::Filtered сняты из публичного API pm-sdk целиком (rg -n 'FilterSet|update_filters' crates — пусто); клиент безусловно отдаёт наверх КАЖДЫЙ успешно разобранный кадр темы activity всей площадки.

Допуск — единственная проверка на шве движка. TrackedTokens::is_tracked (порт, реализованный MarketStateStore и RtdsActorBuffer) вызывается РОВНО ОДИН РАЗ на кадр, до попытки построить TradePrint. Посторонний токен получает причину untracked_token и не идёт дальше, но подтверждает, что соединение живо и кадр реально доставлен. Пустое отслеживаемое множество (рынков пока не открыто) не влияет на само соединение: транспорт поднимается и остаётся подписанным исключительно по признаку потребителя (см. «Активация по потребителю» выше), а при пустом множестве каждый доставленный кадр просто получает untracked_token — ноль допущенных, а не разрыв или фиктивный gap.

Двухэтапная классификация. Первый этап — в pm-sdk при разборе кадра (pm_sdk::rtds::metrics::DropReason): платформенный текстовый PING/PONG классифицируется как ProtocolText (ожидаемый трафик, не потеря разбора), кадр чужой темы/типа — как ParseUnknownTopic, невалидный JSON и прочие ошибки формы (ParseMissingField / ParseInvalidTimestamp / ParseInvalidPrice / ParseInvalidSize / ParseMalformedJson / ParseInvalidNumericOther + MissingDedupeField) — как потеря разбора. RtdsMetrics::events_dropped_parser_loss_total() — единственное определение потери разбора в коде, сумма именно этих семи причин (не включает ProtocolText, Stale, ChannelFull, ParseStale/ParseUnknownTopic, которые не являются потерей формы). Второй этап — на шве движка (adapters::rtds::AdmissionDropReason, описано выше): доставленный и успешно разобранный кадр не допущен как торговое свидетельство по одной из четырёх причин (untracked_token / missing_actor / unrepresentable_print / future_timestamp).

Активация по потребителю (решение оператора 2026-08-31)

Note: this supersedes ADR-004's original Q1 "always-on, no master flag" premise, and it also supersedes the ENGINE_RTDS_ENABLED env master-flag design (introduced 2026-06-15, retired 2026-08-31), which carried a strategy-behavior switch outside config.db (ADR-015). The current design (change rtds-consumer-driven-activation) has no master flag anywhere: the composition root opens the connection iff the active strategy definition names a consumer.

The RTDS WS task spawns iff the active strategy's directional definition has at least one of:

• stop_loss.confirm_source = "rtds_actor_buffer" — the stop-confirm gate;
• a present directional.toxic_flow section — the toxic-flow detector.

Neither present ⇒ no connection is opened, the stop-loss confirm path reads the anonymous LastTradePrice slice unchanged (bit-identical to the pre-RTDS hot loop), and the startup log names the reason (rtds: transport disabled, no consumer configured). When a consumer is configured and the endpoint is unreachable, the shared client retries with backoff and the engine keeps trading on CLOB data.

There is no environment switch: ENGINE_RTDS_ENABLED is retired and a set value is rejected at startup the same way as the retired PM_ENGINE_DRY_RUN (ConfigError::RtdsEnvRetired, config/loader.rs) — it is not a compatibility alias for the new field.

Blocking incompatibility — stop_loss.flow_sell_share_min_permille + confirm_source = "rtds_actor_buffer". The pair is an invariant of the typed stop_loss section (ConfigError::StopFlowShareWithRtdsConfirmSource, config/loader.rs): the flow-imbalance gate takes its BUY/SELL totals from the anonymous market-data store, while rtds_actor_buffer reads the confirming prints from the RTDS actor buffer — different retention bounds, different coverage epochs, different end-to-end latency, so the share would judge a candidate against a DIFFERENT tape (design.md D5 of lossless-stop-confirm-trade-window). Set confirm_source = "anonymous_tape", or omit flow_sell_share_min_permille. The check lives in pm_engine::config::loader::validate_strategy_config, called by BOTH the engine's startup loader and the config-save cascade (pm_engine::config::versioning::validate_version_snapshot, invoked by the dashboard configurator's POST/PUT /api/strategies, by pm-cli strategy add, and by pm-cli strategy versions create) — the incompatible pair is now rejected AT SAVE for any path that writes a full definition, not only discovered at the next restart/Reset. Подкоманда pm-cli strategy update снята целиком 2026-08-31 (карточка remove-pm-cli-strategy-update); правка любого поля stop_loss на существующей строке идёт только через pm-cli strategy versions create либо конфигуратор — оба пути проходят ту же валидацию. Pre-flight check: scripts/stop-evidence-forward-check.sh --preflight-only; ordering: ../operations/rtds-cutover-runbook.md §1.

Toxic-flow detector

Sliding-window detector on each tracked condition_id (semantics carried over verbatim from the prior implementation):

within toxic_flow.window_secs:
  signed_volume = sum(sign(side) × size)  // BUY = +, SELL = -
  if |signed_volume| > toxic_flow.threshold_size:
    emit Decision::CancelOrder { reason = rtds_toxic_flow:<signed_volume> }

Prints with size < toxic_flow.min_trade_size are dropped before window aggregation (anti wash-trade flood).

Harvester activation status and verified limitations

Active toxic-flow cancellation is not approved for Harvester. Keep log_only=true (or omit the block) until every item below is remediated and validated on post-fix forward data. Configuring directional.toxic_flow only starts the transport (it is a consumer per §Активация по потребителю above); it does not make active cancellation safe.

The 2026-08-01 code-path audit found five defects (strategies::directional::config::ToxicFlowDefect). rtds-transport-remediation closed one (wire counters, D4 below) and implemented the exclusion mechanism for a second without closing it (actor identity, pending live evidence); the remaining three are unchanged:

• Actor identity (mechanism implemented, defect stays open). The detector now takes a mandatory own_wallet and excludes prints whose actor == own_wallet before they enter the sliding window, so "foreign trades" is a defensible description again — but the closing evidence (a live RTDS print matching one of our own fills on the full leg tuple: transactionHash, asset, proxyWallet == owner_address, side, size) has not been collected yet, so ActorIdentity stays in ToxicFlowDefect::open() until it is.
• Aggregation is keyed only by condition_id. asset_id / token and outcome are discarded, so complementary Yes/No prints share one signed window even though their economic direction differs.
• Cancellation is also filtered only by condition_id and order side. BUY pressure cancels every matching SELL, including the Harvester profit exit; SELL pressure cancels matching entry BUYs. It is not scoped to the traded token/outcome.
• A cancelled profit exit leaves the position open. The directional exit-presence invariant re-emits a missing primary exit on a later market tick, but there is no toxic-flow hold-down/hysteresis, so repeated cancel/re-emit churn is possible.
• RTDS is a live-only side input and was never represented in deterministic capture/replay; после снятия бэктеста 2026-08-21 паритет для этой защиты проверяется только на живой сессии и телеметрии.

Validation now blocks active cancel while defects are open. Saving or starting a strategy with directional.toxic_flow.log_only = false AND active_cancel_enabled = true — the exact pair that ToxicFlowGates::resolve turns into a live Decision::CancelOrder — is rejected while ToxicFlowDefect::open() is non-empty (currently actor identity, cross-outcome aggregation, cancel-token scope, exit hysteresis). The rejection runs on the same cascade as every other strategy-config check: engine startup, the dashboard configurator's save path, and pm-cli strategy add/versions create. A row with log_only = true (any active_cancel_enabled value) still saves and starts — it just never reaches Decision::CancelOrder. Toxic-flow counters constructed inside the engine (RtdsMetrics::toxic_*) are not wired to the Prometheus /metrics exporter — no describe_counter! call publishes them — so baseline calibration reads the structured tracing logs named below, not a Prometheus row.

Implementation seams: adapters/rtds/mod.rs (raw fan-out), adapters/rtds/toxic_flow.rs (condition-only window), adapters/rtds/toxic_flow_consumer.rs (condition/side cancellation), and strategies/directional/strategy.rs::ensure_primary_exit (re-emission).

Configuration — config.db ONLY (ADR-015)

Toxic-flow knobs live in the directional.toxic_flow block of the strategy definition JSON in config.db (typed ToxicFlowConfig). env is forbidden for these knobs. The block is optional; when it is absent the toxic-flow consumer loop is not wired.

Field	Unit / type	Notes
window_secs	seconds (u64, REQUIRED, > 0)	sliding-window horizon
threshold_size	at-rest micro-shares (Shares, REQUIRED, > 0)	`
min_trade_size	at-rest micro-shares (Shares, REQUIRED, ≥ 0)	dust floor, dropped pre-window
log_only	bool (default true, fail-safe)	Gate 1 — log + count, no cancels
active_cancel_enabled	bool (default false, fail-safe)	Gate 2 — cancels only when both gates aligned

Two-gate activation

Cancel emission is gated by two independent config.db flags. Both must be aligned before any cancel Decision flows.

log_only	active_cancel_enabled	Behaviour
true (default)	(any)	log-only: rtds.toxic_flow.detected (log_only) line, no Decision
false	false (default)	"would cancel": rtds.toxic_flow.would_cancel (active_cancel disabled) line, no Decision
false	true	rejected at save/startup while ToxicFlowDefect::open() is non-empty (§«Harvester activation status» above) — the row does not save and the engine does not start with it

Default-safe gating prevents a single misconfigured flag from sweeping live maker bids. Walk the activation through docs/operations/rtds-cutover-runbook.md before flipping any operator gate.

Observability

Toxic-flow decision counters (RtdsMetrics::toxic_detected_total / toxic_would_cancel_total / toxic_actual_cancels_total / toxic_dust_dropped_total) live only inside the process — they are never passed to a metrics::counter!/describe_counter! call, so they do not appear on the Prometheus /metrics exporter. Baseline calibration and gate walkthroughs read the structured tracing logs at each decision point instead: rtds.toxic_flow.detected (log_only), rtds.toxic_flow.would_cancel (active_cancel disabled), rtds.toxic_flow.actual_cancel.

RTDS transport rows that ARE on /metrics (SSOT — the describe_*! calls in crates/pm-engine/src/observability/metrics.rs):

• pm_engine_rtds_tracked_tokens (gauge) — size of the engine's admitted token set; the same handle bundle spawn_rtds_adapter returns to its caller.
• pm_engine_rtds_own_fill_observation_queue_len (gauge), pm_engine_rtds_own_tx_hash_resolver_heartbeat_age_seconds (gauge), pm_engine_rtds_own_tx_hash_resolver_stalled_total (counter), pm_engine_rtds_own_tx_hash_source_total{source=confirmed|clob_trades} (counter), pm_engine_rtds_own_tx_hash_pending_unresolved_total{reason= retries_exhausted|shutdown|channel_closed} (counter), pm_engine_rtds_own_tx_hash_orphan_confirmed_total (counter), pm_engine_rtds_own_fill_observation_dropped_total{reason=closed} (counter) — the own-fill tx_hash resolver that backs the stop-confirm self-exclusion path. All of this group exist ONLY while the RTDS transport is up (rtds_telemetry.is_some() gates the writer).
• pm_engine_stop_trade_source_gaps_total{source=rtds, reason=parser_loss| reconnect|overflow|source_silent} — the transport-level parse-loss sum (RtdsMetrics::events_dropped_parser_loss_total(), seven closed reasons — §«Сквозной транспорт» above) surfaces here as reason=parser_loss on the shared stop-evidence coverage-gap row, not as its own metric name.

None of these old, non-prefixed names exist any more: rtds_subscription_active_filters (replaced by pm_engine_rtds_tracked_tokens with a different meaning — set size, not filter count) and rtds_events_dropped_filtered (no replacement; nothing reads a per-filter drop count once the client stopped filtering).

The dashboard exposes RTDS-specific surfaces only when the running engine explicitly advertises rtds_enabled=true. Disabled and legacy/unknown runtimes hide the RTDS cards, alert, and toxic-flow form step; enabled-but-unhealthy runtimes remain visible. Hiding the form is presentation-only: existing directional.toxic_flow JSON is preserved on unrelated edits.

See also

• docs/operations/rtds-cutover-runbook.md — operator activation walkthrough.
• crates/pm-engine/src/adapters/rtds/ — implementation (toxic_flow.rs detector, toxic_flow_consumer.rs gating).
• pm_sdk::rtds — shared transport client + parser.