TakerStream

import time
import uuid
from rfq_test.clients.websocket import TakerStreamClient

taker_ws = TakerStreamClient(
    endpoint=config.indexer.ws_endpoint,
    request_address=taker.inj_address,
    timeout=10.0,
)
await taker_ws.connect()

expiry_ms = int(time.time() * 1000) + 5 * 60 * 1000

request_data = {
    "client_id": str(uuid.uuid4()),
    "market_id": config.default_market.id,
    "direction": "long",
    "margin": "200",
    "quantity": "100",
    "worst_price": "5",
    "expiry": expiry_ms,
}

ack = await taker_ws.send_request(
    request_data,
    wait_for_response=True,
    response_timeout=5.0,
)
rfq_id = int(ack["rfq_id"])

const clientId = crypto.randomUUID();
const expiry = Date.now() + 5 * 60 * 1000;

const request = {
  type: "rfq_request",
  client_id: clientId,
  market_id: INJUSDC_MARKET_ID,
  direction: "long",                   // lowercase string – NOT an integer
  margin: "200",
  quantity: "100",
  worst_price: "5",
  expiry,
};

// request_address is sent as TakerStream metadata when opening the stream.
ws.send(JSON.stringify(request));

const ack = await waitForRequestAck(ws, clientId);
const rfqId = Number(ack.rfq_id);

quotes = await taker_ws.collect_quotes(
    rfq_id=rfq_id,
    timeout=0.5,         # TrueCurrent uses 500ms; API takers may tune this
    min_quotes=1,
)

if not quotes:
    raise RuntimeError("No quotes received")

# For a long: lowest price wins. For a short: highest price wins.
best = min(quotes, key=lambda q: float(q["price"]))

const quotes: Quote[] = [];

ws.on("message", (raw: Buffer) => {
  const msg = JSON.parse(raw.toString());
  if (msg.type === "rfq_quote" && msg.rfq_id === rfqId) {
    quotes.push(msg);
  }
});

await new Promise((r) => setTimeout(r, 500));   // TrueCurrent default; tune for API use

if (quotes.length === 0) {
  throw new Error("No quotes received");
}

const best = quotes
  .slice()
  .sort((a, b) => Number(a.price) - Number(b.price))[0];

request_data = {
    "client_id": str(uuid.uuid4()),
    "market_id": market.id,
    "direction": "long",
    "margin": "200",
    "quantity": "100",
    "worst_price": "5",
    "expiry": int(time.time() * 1000) + 300_000,
}

ack = await taker_ws.send_request(request_data, wait_for_response=True, response_timeout=5.0)
real_rfq_id = int(ack["rfq_id"])
quotes = await taker_ws.collect_quotes(rfq_id=real_rfq_id, timeout=0.5, min_quotes=1)

class QuoteRouter:
    def __init__(self):
        self.pending: dict[tuple[str, int], asyncio.Queue] = {}

    async def submit(self, taker: str, rfq_id: int, request: dict) -> list[dict]:
        queue: asyncio.Queue = asyncio.Queue()
        key = (taker, rfq_id)
        self.pending[key] = queue
        await taker_ws.send_request(request)

        quotes = []
        try:
            while True:
                quote = await asyncio.wait_for(queue.get(), timeout=5.0)
                quotes.append(quote)
        except asyncio.TimeoutError:
            pass
        finally:
            del self.pending[key]
        return quotes

    def on_message(self, msg: dict):
        taker = msg.get("taker") or msg.get("request_address")
        rfq_id = msg.get("rfq_id")
        if taker is None or rfq_id is None:
            return
        key = (taker, int(rfq_id))
        if key in self.pending:
            self.pending[key].put_nowait(msg)

import asyncio

MAX_RETRIES = 10
BASE_DELAY = 1.0

for attempt in range(MAX_RETRIES):
    try:
        await taker_ws.connect()
        await taker_ws.run()
    except ConnectionError as e:
        delay = min(BASE_DELAY * (2 ** attempt), 30.0)
        print(f"Disconnected: {e}. Retry in {delay}s")
        await asyncio.sleep(delay)

taker_ws = TakerStreamClient(endpoint, request_address, timeout=15.0)
taker_ws.on_message = lambda m: print("<-", m)
await taker_ws.connect()