Permissionless Listing with Custom Oracle

# Requires Python 3.11+
pip install websockets cryptography base58

export ORDERLY_ACCOUNT_ID="0x..."
export ORDERLY_KEY="ed25519:..."
export ORDERLY_SECRET="..."
export BASE_CCY="AAPL"

python provider_ws_example.py

import asyncio
import base64
import json
import time
import uuid
from dataclasses import dataclass
from urllib.parse import urlencode

import base58
import websockets
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey


PUSH_TOPIC = "indexpricefeed"
WS_PATH = "/v1/ws/oracle/push"
DEFAULT_DOMAIN = "wss://ws-oracle.orderly.org"


@dataclass
class OracleProviderConfig:
    account_id: str
    orderly_key: str
    orderly_secret: str
    base_ccy: str
    domain: str = DEFAULT_DOMAIN
    reconnect_delay_s: float = 5.0


class OracleProviderClient:
    def __init__(self, config: OracleProviderConfig):
        self.config = config
        self._private_key = Ed25519PrivateKey.from_private_bytes(
            base58.b58decode(config.orderly_secret)
        )
        self._connected = asyncio.Event()
        self._running = False
        self._task = None
        self._ws = None

    def _sign(self, message: str) -> str:
        signature = self._private_key.sign(message.encode())
        # Orderly WebSocket auth uses unpadded base64url signatures.
        return base64.urlsafe_b64encode(signature).decode().rstrip("=")

    def _build_url(self) -> str:
        timestamp = str(int(time.time() * 1000))
        query = urlencode({
            "orderly_key": self.config.orderly_key,
            "timestamp": timestamp,
            "sign": self._sign(timestamp),
        })
        return f"{self.config.domain}{WS_PATH}/{self.config.account_id}?{query}"

    async def _receive_loop(self):
        async for raw in self._ws:
            message = json.loads(raw)
            if message.get("event") == "ping":
                await self._ws.send(json.dumps({"event": "pong"}))
            elif message.get("event") == "publish" and not message.get("success"):
                print("publish rejected", message)

    async def _connect_once(self):
        async with websockets.connect(self._build_url()) as ws:
            self._ws = ws
            auth = json.loads(await ws.recv())
            if not auth.get("success"):
                raise RuntimeError(f"auth failed: {auth.get('errorMsg')}")
            print(f"authenticated; publishing {self.config.base_ccy}")
            self._connected.set()
            await self._receive_loop()

    async def _run_forever(self):
        while self._running:
            try:
                await self._connect_once()
            except asyncio.CancelledError:
                raise
            except Exception as error:
                print(f"oracle websocket error: {error}; reconnecting...")
            finally:
                self._connected.clear()
                self._ws = None

            if self._running:
                await asyncio.sleep(self.config.reconnect_delay_s)

    async def start(self):
        if self._task is not None:
            return
        self._running = True
        self._task = asyncio.create_task(self._run_forever())

    async def stop(self):
        self._running = False
        if self._ws is not None:
            await self._ws.close()
        if self._task is not None:
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass
            self._task = None

    async def push_price(self, price: float, wait_timeout: float = 10.0) -> bool:
        try:
            await asyncio.wait_for(self._connected.wait(), timeout=wait_timeout)
        except asyncio.TimeoutError:
            return False

        payload = {
            "id": uuid.uuid4().hex[:8],
            "event": "publish",
            "topic": PUSH_TOPIC,
            "ts": int(time.time() * 1000),
            "data": {
                "base_ccy": self.config.base_ccy,
                "price": round(price, 8),
            },
        }
        ws = self._ws
        if ws is None:
            return False
        await ws.send(json.dumps(payload))
        return True

    async def __aenter__(self):
        await self.start()
        return self

    async def __aexit__(self, *exc):
        await self.stop()

import asyncio
import os

from orderly_oracle_sdk import OracleProviderClient, OracleProviderConfig


PUSH_INTERVAL_S = 1


async def get_latest_price_from_your_source() -> float:
    # Replace this with your own price source, such as your index service,
    # broker backend, RWA data provider, or exchange aggregation job.
    raise NotImplementedError("connect your production price source here")


async def run():
    config = OracleProviderConfig(
        account_id=os.environ["ORDERLY_ACCOUNT_ID"],
        orderly_key=os.environ["ORDERLY_KEY"],
        orderly_secret=os.environ["ORDERLY_SECRET"],
        base_ccy=os.getenv("BASE_CCY", "AAPL"),
        domain=os.getenv("ORACLE_WS_DOMAIN", "wss://ws-oracle.orderly.org"),
    )

    async with OracleProviderClient(config) as client:
        while True:
            price = await get_latest_price_from_your_source()
            if await client.push_price(price):
                print(f"pushed {config.base_ccy} @ {price}")
            await asyncio.sleep(PUSH_INTERVAL_S)


asyncio.run(run())