Levandor

pmv2 Position-Manager NATS Consumer — multi-source signal ingestion

10 min read

Generalizes the autotrade engine into a position manager that ingests multiple signal sources over NATS JetStream, keyed on source + action. First consumer = @weather (buy the LOCKED winning temperature bucket). The reviewed money-path (reserve → quantize → assert → submit → fills → reconcile → exits → settle → caps → breaker) is preserved byte-for-byte; only signal ingestion and the order-derivation front-half generalize. Built + reviewed + gate-green on autotrade-foundation; NOT deployed.

For Agents — state at a glance

Branch autotrade-foundation, 6 commits in order: 4c0b7d4 (T1 generalize TradeSignal) → 1bc8989 (T2 wire.rs) → 9abc27b (T3 plan_order seam) → eb6ca9a (T4 per-source mode) → b7f8437 (T5 consumer seam + NATS adapter) → 74faf7d (T6 Pmv2Consumer subcommand). Spec: docs/superpowers/specs/2026-06-16-pmv2-position-manager-nats-design.md. Plan: docs/superpowers/plans/2026-06-17-pmv2-position-manager-nats-consumer.md. Wire schema source-of-truth: weather_bet docs/signal-schema.md @ 3187585 (producer wx-signal, crates/wx-signal/src/nats.rs::SignalPayload). Module is still autotrade — the spec proposed an internal rename autotrade → position_manager but it is deferred (paths/wikilinks unchanged). weather mode defaults off, so the consumer is a complete no-op until armed. Going live requires TWO operator actions (global mode live AND weather.mode = live).

Ingestion pipeline (NATS → execution) 

graph TD
    WX["@weather wx-signal<br/><i>publishes flat JSON</i>"] -->|"weather.signals"| JS["NATS JetStream<br/>stream WEATHER_SIGNALS<br/>durable <b>pmv2</b>"]
    JS -->|"pull batch, ack_wait=30s"| NATS["nats.rs<br/><i>connect / fetch / ack</i>"]
    NATS --> WIRE["wire.rs::parse_signal<br/><b>trust boundary</b><br/>Accept / Skip / Reject"]
    WIRE -->|"Accept → TradeSignal"| ICON["runner.rs::ingest_for_consumer<br/><i>staleness-first</i>"]
    ICON --> IE["engine.rs::ingest_entry<br/><i>dispatch on Execution</i>"]
    IE -->|"Directive"| PD["plan_directive<br/><i>weather</i>"]
    IE -->|"Context"| PC["plan_context<br/><i>cohort, unchanged</i>"]
    PD --> BACK["shared back-half<br/>try_reserve → OrderPlan → mode"]
    PC --> BACK
    BACK -->|"dry"| REH["dry_rehearsal<br/>build+sign, NO POST"]
    BACK -->|"live"| SUB["submit_live<br/>real CLOB POST"]
    NATS -. "ack_action(ConsumerOutcome)" .-> NATS
    style WIRE fill:#3d2020,stroke:#a55,color:#fff
    style BACK fill:#264653,stroke:#2a9d8f,color:#fff
    style JS fill:#2d2d2d,stroke:#888,color:#fff

T1 — Generalized TradeSignal (4c0b7d4)

crates/polymarket-fetch/src/pmv2/autotrade/signal.rs. TradeSignal was reshaped so any source can drive the engine. The discriminator that used to be SignalAction::{Enter,Exit} is now an Execution enum carried on the signal:

pub enum Execution {
    Directive { max_price: f64 },                                   // weather: explicit price cap
    Context   { reference_price: f64, conviction: f64, source_score: Option<f64> }, // cohort
}
  • Directive (weather) — the producer sends an explicit max_price limit cap plus market constraints; the consumer sizes and the limit is ≤ max_price. No cohort price gates.
  • Context (cohort) — the producer sends reference_price (frozen at first detection, feeds the chase gate) + conviction (sizing) + source_score; the consumer computes the limit via gates and sizes via conviction.

TradeSignal now carries the market identifiers directly: signal_id (the dedup key), action: Side, condition_id, outcome_index, outcome_label, token_id, neg_risk: Option<bool>, tick_size/min_size: Option<f64>, title, event_slug, slug, origin_ref, execution: Execution, valid_until_ts: Option<i64>.

Load-bearing — dedup is byte-identical mid-migration

TradeSignal::dedup_key() now returns signal_id directly (no recompute). The cohort emitter sets signal_id == legacy orders::dedup_key(.., "enter", ..) — the exact same 6-discriminator string the old path produced. Result: the dedup key is byte-for-byte unchanged for the cohort lane, so running the old in-process trigger and the new NATS path side-by-side cannot double-order. The ON CONFLICT(dedup_key) DO NOTHING partial index dedups across both. (action is currently #[allow(dead_code)] on the struct — it’s wired but the dispatch keys on source + Execution, not the bare Side.)

This supersedes the original SignalAction shape documented in autotrade-signal-type (that note describes the pre-generalization Enter{reference_price}/Exit enum).

T2 — wire.rs: the trust boundary (1bc8989)

crates/polymarket-fetch/src/pmv2/autotrade/wire.rs. Parses @weather’s flat, self-describing JSON envelope (SignalPayload, mirrored from wx-signal::nats::SignalPayload) into a verdict:

pub enum ParseOutcome { Accept(TradeSignal), Skip(String), Reject(String) }
pub fn parse_signal(bytes: &[u8]) -> ParseOutcome   // never panics

The fail policy is asymmetric and deliberate:

ConditionVerdictRationale
Malformed JSON / missing-or-invalid required common field (id, source, schema_version, action, condition_id, yes_token_id, max_price, valid_until_ts)Reject (fail-CLOSED)A broken envelope must never execute
Unknown source (or a source/action we don’t yet handle)Skip (fail-OPEN)Future sources must not crash or hard-reject the consumer
action = SELLSkipSELL execution not yet implemented (exits stay internally triggered); valid on the wire today
schema_version newer than knownSkipForward-compat: don’t choke on a future envelope

For Agents

wire.rs is the trust boundary between an external producer and our money-path. It never panics and never executes on a malformed or unknown input. fail-CLOSED (Reject) is reserved for known-bad / missing-required; fail-OPEN (Skip) is the default for unknown / not-yet-supported so an unforeseen source can’t take the consumer down.

T3 — plan_order seam (9abc27b)

crates/polymarket-fetch/src/pmv2/autotrade/engine.rs. ingest_entry (engine.rs:322) dispatches on Execution:

  • plan_context (engine.rs:151) — the cohort path, byte-for-byte unchanged. Same gates (liveness/category/price_bounds/ev_floor/min_conviction/chase/exposure) including the ask-band guard from autotrade-resolution-proposed-entry-guard.
  • plan_directive (engine.rs:247) — the weather path. token_id-direct (no token_for label lookup); outcome_index derived from the market’s clob_token_ids; neg_risk verified against the live CLOB (mismatch ⇒ refuse + alert — a wrong neg_risk selects the wrong EIP-712 contract); directive_limit_price = floor_to_tick(min(max_price, ask)); directive_size_usd is the no-conviction base size; NO cohort price gates (weather legitimately buys up to max_price = 0.95).

Both feed the shared, untouched back-half: try_reserve (dedup + caps under the config-row lock) → OrderPlan → mode branch. A new stale_gate (gates.rs:63, stale_gate(now_unix, valid_until_ts)) is added here and wired into the consumer in T5.

Directive path skips cohort price gates by design

The ask-band / chase / EV gates are Cohort/Context-only. plan_directive must NOT run them — the weather strategy’s whole thesis is buying a near-certain locked bucket at up to 0.95, which the cohort price-bounds gate would reject. The safety net for the Directive path is the producer’s max_price + the neg_risk verify + the same caps/reserve back-half.

T4 — Per-source mode (eb6ca9a)

crates/polymarket-fetch/src/pmv2/autotrade/config.rs + migration 20260617000000_pmv2_source_mode.sql.

  • New column: pmv2_autotrade_sources.mode TEXT NOT NULL DEFAULT 'off' CHECK (mode IN ('off','dry_run','live')).
  • effective_mode(global, source) -> Mode (config.rs:43) = min on the ordering off < dry_run < live. A source can only ever be as conservative as, or more conservative than, the global mode — never less. Tested: (live, dry) → dry, (live, off) → off, (dry, live) → dry, (off, live) → off, (live, live) → live.
  • Cohort backfilled to live in the same migration (UPDATE ... SET mode='live' WHERE source_id='first_mover_core') so effective == global for the cohort — its behavior is exactly preserved.
  • Command: /autotrade source <id> mode <off|dry|live>.

Load-bearing — engine_cfg.mode is overridden to the effective mode

build_autotrade_deps sets engine_cfg.mode = effective_mode(global, source). This is what prevents a dry source under a live global from POSTing: the engine’s own mode-gate reads the already-narrowed mode, so a dry source can never reach submit_live. Without this override, the global live would leak through to a source meant to stay in rehearsal.

T5 — Consumer seam + NATS adapter (b7f8437)

Two pieces:

Consumer seam — runner.rs::ingest_for_consumer (runner.rs:142) + ConsumerOutcome (engine.rs:63):

pub enum ConsumerOutcome { Persisted, Skipped(String), Stale, Transient(String), Terminal(String) }

ingest_for_consumer is staleness-first: it runs stale_gate before any I/O and returns Stale immediately if past valid_until_ts. Otherwise it calls ingest_entry and reuses the existing dry_rehearsal (build+sign, no POST) and submit_live paths verbatim. Breaker is fail-safe: config::load(pool).map_or(true, |c| c.breaker_tripped) — a config-load error reads as tripped = true (refuse), never as “all clear”.

NATS adapter — crates/polymarket-fetch/src/pmv2/autotrade/nats.rs:

  • Ack discipline via ack_action(&ConsumerOutcome) -> AckAction (nats.rs:17, const fn):

    ConsumerOutcomeAckActionMeaning
    Persisted, Skipped(_)Ackdurably persisted or definitive skip (dedup/gate) — only ack after persist
    Transient(_)NakCLOB/RPC/DB down → redeliver later with backoff
    Stale, Terminal(_)Termpast valid_until_ts / unknown-bad / max-deliver → stop the redeliver loop, alert

  • async-nats 0.49 with the nkeys feature; ConnectOptions::new().nkey(seed) where seed = the raw NKey seed (SU…) read straight from NATS_NKEY_SEED (not a creds file), no TLS.

  • Binds the existing pmv2 durable on stream WEATHER_SIGNALS, subject weather.signals (config owned by @deploy: max_deliver=5, ack_wait=30s, DeliverNew).

For Agents — why at-least-once is safe

Delivery is at-least-once, made effectively-once by idempotency: ON CONFLICT(dedup_key) DO NOTHING on signal_id guarantees ≤1 order per signal_id, ever — a redelivery hits the dedup and no-ops. Nats-Msg-Id = id so JetStream’s server-side dedup aligns with our app dedup on a single key. Ack only ever happens after the decision is durably persisted — never ack-then-process. A persistently-transient signal exhausts max_deliver=5 → we Term + alert before the cap, so nothing is silently dropped. (Same terminal-vs-transient philosophy as the exit-cursor 404 fix in autotrade-exit-cursor-404-wedge-fix.)

T6 — Pmv2Consumer subcommand (74faf7d)

crates/polymarket-fetch/src/cli.rs (Pmv2Consumer variant) + main.rs::run_pmv2_consumer + migration 20260617010000_pmv2_weather_source.sql.

The engine is a one-shot per-tick batch (scheduled-run invokes it once per cycle), so a tokio::spawn-ed NATS loop inside it would die when the tick process exits. The consumer therefore needs its own dedicated long-running process — the pmv2-consumer subcommand.

  • Dormant-safe: if NATS_URL/NATS_NKEY_SEED are absent it logs "NATS_URL/NATS_NKEY_SEED absent; consumer dormant, exiting 0" and exits 0 (a missing env is not an error — the process is simply a no-op until configured).
  • Mode = the weather effective mode, which defaults off ⇒ skips all signals until armed.
  • Migration seeds the weather source row: INSERT ... (source_id, enabled, mode) VALUES ('weather', TRUE, 'off') ON CONFLICT DO NOTHING — present but disabled.

Key gotchas

async-nats 0.49 JetStream errors are Box<dyn StdError> (no anyhow::From)

JetStream calls (get_stream, get_consumer, fetch, …) return Box<dyn std::error::Error + Send + Sync>, which has no anyhow::From impl — a bare ? won’t compile. The adapter uses a tiny map_err helper nats_err(e) = anyhow::anyhow!("{e}") (nats.rs:36) at every JetStream call site instead of ?.

Going live requires TWO operator actions

The consumer only POSTs when effective_mode == live, i.e. both the global mode is live and weather.mode = live. effective_mode = min(global, source) makes either one being non-live keep weather in rehearsal. A single flip is never enough — by design.

NATS env

NATS_URL (tailnet nats.taild4189d.ts.net:4222) + NATS_NKEY_SEED, both in the POLYMARKET_FETCH_ENV GH secret. The seed in that env is the pmv2 consumer identity (pubkey UDYV…ZEDD) with perms to subscribe weather.signals + bind/consume the pmv2 durable. ⚠️ At spec time the container could not yet DNS-resolve the tailnet host (@deploy fix pending) — the connect layer is blocked on that, everything else builds.

Status & remaining work

Built + reviewed + gate-green on autotrade-foundation; NOT deployed

Remaining before arming weather live:

  1. Deploy the branch (still off main; the VM runs older code).
  2. @deploy launcher for the new pmv2-consumer subcommand — it’s a dedicated long-running service unit, not part of scheduled-run. Also needs the tailnet DNS-resolution fix so the container can reach nats.taild4189d.ts.net.
  3. Manual dry-validate — publish a real weather signal, confirm the consumer emits a DRY-RUN rehearsal (build+sign, no POST) — this is @weather’s “dry-run fill-validation = capital gate.”
  4. Only then flip global live + weather.mode = live.
  • cohort-firstmover-nats-migration — the follow-on that puts the cohort first-mover ENTRY onto this same lane (publishes to cohort.signals, consumed by this Pmv2Consumer); deletes the in-process runner::run_entry. Note: it also folds entry-reconcile + preflight restoration into the shared pre-arm gate below (deleting the watcher-side autotrade_ready() removed the per-tick reconcile_entries).
  • autotrade-signal-type — the pre-generalization TradeSignal / SignalAction::{Enter,Exit}; T1 here replaces SignalAction with the Execution enum
  • pmv2-autotrade-engine-design — the locked engine design; this generalizes its ingestion while preserving the reviewed money-path
  • autotrade-gates — the gate chain reused by plan_context; stale_gate added here for the consumer
  • autotrade-market-data-trait — MarketData::resolve / ResolvedMarket (clob_token_ids, neg_risk) used by plan_directive
  • autotrade-engine-impl — ingest_entry / try_reserve shared back-half the consumer reuses
  • autotrade-commands-bot-wiring — /autotrade surface; /autotrade source <id> mode … added in T4
  • autotrade-3c2-complete-handoff — the reviewed live-submit / lifecycle / safety phase this sits on top of
  • autotrade-exit-cursor-404-wedge-fix — same terminal-vs-transient ack philosophy as the consumer’s ack_action
  • weather-bet — the @weather strategy / wx-signal producer that publishes to weather.signals
  • polymarket-fetch — project overview

Graph View

Backlinks