End-to-end Python walkthrough

$ uv add multiedge-relay          # or: pip install multiedge-relay
$ export MULTIEDGE_API_KEY="mesk_..."   # publisher- or admin-scoped key from the portal

import os

from multiedge_relay import Signal, SignalPublisher

STRATEGY_ID = "str_your_feed_id"          # from the portal's feed page

signal = Signal(
    strategy_id=STRATEGY_ID,
    # Deterministic idempotency key: one per strategy per signal date.
    # Re-running this script re-acks (200, duplicate=True) instead of
    # publishing twice.
    client_signal_id=f"{STRATEGY_ID}:2026-09-07",
    # Informational label echoed to subscribers; the relay enforces the
    # schema stored on your feed regardless.
    schema_version="portfolio_rebalance/1.1",
    payload={
        "kind": "portfolio_rebalance",
        "signal_date": "2026-09-07",
        "planned_execution_date": "2026-09-08",
        "positions": [
            {"ticker": "SPY", "action": "INITIALIZE", "signal_portfolio_weight": 0.55},
            {"ticker": "TLT", "action": "INITIALIZE", "signal_portfolio_weight": 0.3},
            {"ticker": "GLD", "action": "INITIALIZE", "signal_portfolio_weight": 0.15},
        ],
    },
)

with SignalPublisher(api_key=os.environ["MULTIEDGE_API_KEY"]) as publisher:
    ack = publisher.publish(signal)
    # ack.sequence: gapless per-strategy log position — the ordering truth.
    # ack.duplicate: True when this client_signal_id was already accepted.
    print(f"seq={ack.sequence} id={ack.signal_id} duplicate={ack.duplicate}")

# A daily rebalance derives actions from the weight change — but ALWAYS
# sends the COMPLETE post-trade book. A ticker absent from a non-empty
# positions list has a target weight of 0 and is LIQUIDATED, so never
# send "changes only".

def build_positions(previous: dict[str, float], target: dict[str, float]) -> list[dict]:
    """Complete positions list: every target ticker, affirmatively labeled."""
    positions = []
    for ticker in sorted(target):
        old, new = previous.get(ticker, 0.0), target[ticker]
        if ticker not in previous:
            action = "BUY"            # entering the book
        elif new > old:
            action = "BUY"            # adding to the position
        elif new < old:
            action = "SELL"           # trimming (weight 0 would mean exit)
        else:
            action = "HOLD"           # unchanged — stated, not omitted
        positions.append({
            "ticker": ticker,
            "action": action,
            "signal_portfolio_weight": new,   # ALWAYS the post-trade target
        })
    # Anything held yesterday but absent from `target` is deliberately NOT
    # appended: its absence IS the liquidation instruction.
    return positions


previous = {"SPY": 0.55, "TLT": 0.30, "GLD": 0.15}
target = {"SPY": 0.60, "TLT": 0.25, "GLD": 0.15}   # GLD unchanged -> HOLD

signal = Signal(
    strategy_id=STRATEGY_ID,
    client_signal_id=f"{STRATEGY_ID}:2026-09-14",
    schema_version="portfolio_rebalance/1.1",
    payload={
        "kind": "portfolio_rebalance",
        "signal_date": "2026-09-14",
        "planned_execution_date": "2026-09-15",
        "positions": build_positions(previous, target),
    },
)

# No trades today? Publish the day anyway. An ABSENT day is
# indistinguishable from an outage; an EMPTY one is not. The heartbeat
# never trades and never liquidates — it is an explicit "no action".

heartbeat = Signal(
    strategy_id=STRATEGY_ID,
    client_signal_id=f"{STRATEGY_ID}:2026-09-15",
    schema_version="portfolio_rebalance/1.1",
    payload={
        "kind": "portfolio_rebalance",
        "signal_date": "2026-09-15",
        "planned_execution_date": "2026-09-16",
        "positions": [],  # empty = heartbeat: no action, nothing liquidated
    },
)

with SignalPublisher(api_key=os.environ["MULTIEDGE_API_KEY"]) as publisher:
    publisher.publish(heartbeat)

from multiedge_relay import IdempotencyConflict

# The ledger is append-only: a correction is a NEW signal, not an edit.
# Same signal_date, FULL corrected book, and a NEW client_signal_id with
# a revision suffix — ":r2", then ":r3", and so on. The receiving rule:
# the HIGHEST sequence for a signal_date is authoritative.

corrected = Signal(
    strategy_id=STRATEGY_ID,
    client_signal_id=f"{STRATEGY_ID}:2026-09-14:r2",   # <- new id, same date
    schema_version="portfolio_rebalance/1.1",
    payload={
        "kind": "portfolio_rebalance",
        "signal_date": "2026-09-14",
        "planned_execution_date": "2026-09-15",
        "positions": [
            {"ticker": "SPY", "action": "BUY", "signal_portfolio_weight": 0.55},
            {"ticker": "TLT", "action": "SELL", "signal_portfolio_weight": 0.3},
            {"ticker": "GLD", "action": "HOLD", "signal_portfolio_weight": 0.15},
        ],
    },
)

with SignalPublisher(api_key=os.environ["MULTIEDGE_API_KEY"]) as publisher:
    try:
        ack = publisher.publish(corrected)
    except IdempotencyConflict as conflict:
        # You forgot the :r2 suffix: the relay refuses to silently discard
        # a changed payload under an already-used id — HTTP 409
        # client_signal_id_conflict, naming the original signal. Bump the
        # revision suffix and publish again. (A byte-identical retry is
        # NOT a conflict: it re-acks 200 with duplicate=True.)
        print(f"id already used by signal {conflict.signal_id}; bump to :r3")
        raise

from multiedge_relay import (
    AuthError,          # 401/403 — wrong key; never retried
    PublishFailed,      # retries exhausted; signal spilled to the disk DLQ
    ValidationRejected, # 422/413 — fix the payload; never retried
)

# The publisher already retries transient failures (408/429/5xx and
# transport errors) with jittered backoff for up to ~90 s — long enough
# to ride out a relay deployment. You only handle the TERMINAL outcomes.

with SignalPublisher(api_key=os.environ["MULTIEDGE_API_KEY"]) as publisher:
    try:
        ack = publisher.publish(signal)
    except ValidationRejected as exc:
        # The relay named what is wrong (schema_violation, payload_too_large,
        # ...). Retrying identical bytes cannot succeed — fix the payload.
        print(f"rejected: {exc}")
    except AuthError:
        # Rotate/fix the API key; the SDK never retries an auth failure.
        raise
    except PublishFailed as exc:
        # Retry budget exhausted (relay unreachable). The signal was
        # appended to the disk DLQ FIRST, so nothing is lost:
        #   $ multiedge dlq list
        #   $ multiedge dlq resend      # deduplicated by client_signal_id
        print(f"spilled to {exc.dlq_path} after {exc.attempts} attempts")

# NOTE on publish_many: it is N independent requests and NOT atomic —
# never use it to split one portfolio. One date = one publish().

import os

from multiedge_relay import ReceivedSignal, SignalMeta, SignalSubscriber

def handle(signal: ReceivedSignal, meta: SignalMeta) -> None:
    # Called once per signal, in sequence order — catch-up backlog first
    # (meta.source == "catchup"), then live deliveries ("live").
    # signal.payload is the COMPLETE portfolio for signal.payload["signal_date"].
    print(meta.source, signal.sequence, signal.payload["signal_date"])

subscriber = SignalSubscriber(
    api_key=os.environ["MULTIEDGE_API_KEY"],   # subscriber-scoped key
    strategy_id="str_the_feed_you_follow",
    on_signal=handle,
)
# Blocks: resumes from the durable cursor (~/.multiedge/cursor), replays
# everything missed while offline, then polls for live signals. The cursor
# commits only AFTER handle() returns — a crash mid-handle redelivers.
subscriber.run()

from multiedge_relay import SignalSubscriber, SqliteStateStore

store = SqliteStateStore()   # one local SQLite file: cursor + dedup together

subscriber = SignalSubscriber(
    api_key=os.environ["MULTIEDGE_API_KEY"],
    strategy_id="str_the_feed_you_follow",
    # Delivery is at-least-once; exactly_once() wraps your handler so a
    # crash-redelivered or replayed signal_id never runs it twice.
    on_signal=store.exactly_once(handle),
    cursor_store=store,
)
subscriber.run()

def handle(signal: ReceivedSignal, meta: SignalMeta) -> None:
    """Apply one complete portfolio signal to the trading book."""
    payload = signal.payload

    # 1. Heartbeat: an explicit no-action day. Record it (it proves the
    #    publisher was alive) and stop — it never trades, never liquidates.
    if not payload["positions"]:
        record_heartbeat(payload["signal_date"], signal.sequence)
        return

    # 2. Corrections: the HIGHEST sequence per signal_date wins. If a
    #    later signal for this date was already applied, this one is stale.
    if applied_sequence_for(payload["signal_date"]) >= signal.sequence:
        return

    # 3. The list is the COMPLETE post-trade book: build target weights,
    #    with every ticker NOT listed at an implicit weight of 0.
    targets = {
        p["ticker"]: p["signal_portfolio_weight"] for p in payload["positions"]
    }
    # HOLD lines land here too — their unchanged weight produces no order.

    # 4. Diff against current holdings. A held ticker missing from
    #    `targets` gets target 0 — that is the LIQUIDATION the contract
    #    defines for omission; never skip it.
    for ticker in set(current_holdings()) | set(targets):
        rebalance_to(ticker, targets.get(ticker, 0.0))

    mark_applied(payload["signal_date"], signal.sequence)

from multiedge_relay import SignatureVerificationError, verify_signature

@app.post("/webhooks/multiedge")
async def receive(request: Request):
    body = await request.body()   # RAW bytes — never re-parse before verifying
    try:
        # HMAC-SHA256 over "{ts}." + body with your per-endpoint secret;
        # constant-time compare; 5-minute freshness window.
        signal = verify_signature(body, dict(request.headers), WEBHOOK_SECRET)
    except SignatureVerificationError:
        return Response(status_code=401)   # untrusted — do not process
    # The relay retries non-2xx deliveries up a ladder, so dedupe:
    with store.process(signal) as fresh:   # same SqliteStateStore as above
        if fresh:
            handle(signal, meta=None)
    return {"ok": True}   # ACK duplicates too, so the retry ladder stops