diff --git a/src/ws/PriceSubscriptionManager.js b/src/ws/PriceSubscriptionManager.js
index af5f7fd..b8cd6e8 100644
--- a/src/ws/PriceSubscriptionManager.js
+++ b/src/ws/PriceSubscriptionManager.js
@@ -1,340 +1,458 @@
-'use strict';
-
-const config = require('../config');
-const logger = require('../logger');
-const { WebSocket } = require('ws');
-
-const MAX_ASSETS_PER_CLIENT = 5;
-const MAX_CONNECTIONS = config.ws.maxConnections;
-const MAX_CONNECTIONS_PER_IP = config.ws.maxConnectionsPerIp;
-const PING_INTERVAL_MS = 30_000;
-const MAX_MISSED_PINGS = 3;
-const PRICE_CHANGE_THRESHOLD_PCT = 0.1;
-
-// Prometheus gauge — updated whenever a socket connects or disconnects.
-let wsConnectionsGauge = null;
-try {
-  const prom = require('prom-client');
-  wsConnectionsGauge = new prom.Gauge({
-    name: 'ws_connections_current',
-    help: 'Number of currently active WebSocket connections',
-  });
-} catch {
-  // prom-client not installed; gauge is a no-op.
-}
-
-function updateGauge(delta) {
-  if (wsConnectionsGauge) wsConnectionsGauge.inc(delta);
-}
+const WebSocket = require("ws");
+const config = require("../config");
+const logger = require("../logger");
+const priceOracle = require("../services/priceOracle");
+
+const MAX_ASSETS_PER_CLIENT = 20;
+const MAX_MESSAGE_BYTES = 4096;
+const VALID_ASSET_RE = /^[A-Z0-9]{1,12}$/;
+const HEARTBEAT_INTERVAL_MS = 30_000;
+const MAX_MISSED_PINGS = 2;
+const DEFAULT_MAX_CONNECTIONS_PER_IP = 10;
 
-/**
- * Tracks WebSocket subscriptions and delivers price-change pushes.
- *
- * Each socket entry:
- *   { ws, assets: Set<string>, missedPings: number }
- */
 class PriceSubscriptionManager {
-  constructor() {
-    this._clients = new Map(); // ws → { assets, missedPings }
-    this._clientIpBySocket = new Map(); // ws → string
-    this._connectionsByIp = new Map(); // ip → number
-    this._previousPrices = new Map(); // assetKey → number
-    this._assetSubscribers = new Map(); // assetKey → Set<ws>
-    this._pingTimer = null;
+  constructor(options = {}) {
+    this._clients = new Map();
+    this._assetSubscribers = new Map();
+    this._previousPrices = new Map();
+    this._clientIpBySocket = new Map();
+    this._connectionsByIp = new Map();
+    this._pollTimer = null;
+    this._heartbeatTimer = null;
+    this._draining = false;
+    this._oracle = options.oracle || priceOracle;
+    this._pollIntervalMs = options.pollIntervalMs || config.wsPricePollIntervalMs || 15_000;
+    this._heartbeatIntervalMs = options.heartbeatIntervalMs || HEARTBEAT_INTERVAL_MS;
+    this._maxConnectionsPerIp =
+      options.maxConnectionsPerIp ?? config.ws?.maxConnectionsPerIp ?? DEFAULT_MAX_CONNECTIONS_PER_IP;
+  }
+
+  get clientCount() {
+    return this._clients.size;
+  }
+
+  get trackedAssets() {
+    return Array.from(this._assetSubscribers.keys());
+  }
+
+  start() {
+    if (this._pollTimer) return;
     this._draining = false;
-    this._drainStats = { warned: 0, closed: 0, forceClosed: 0 };
+
+    this._pollTimer = setInterval(() => {
+      this._pollAndBroadcast().catch((err) => {
+        logger.error("WS price poll failed", { error: err.message });
+      });
+    }, this._pollIntervalMs);
+    this._pollTimer.unref?.();
+
+    this._heartbeatTimer = setInterval(() => {
+      this._heartbeat();
+    }, this._heartbeatIntervalMs);
+    this._heartbeatTimer.unref?.();
+
+    logger.info("PriceSubscriptionManager started", {
+      pollIntervalMs: this._pollIntervalMs,
+    });
   }
 
-  _getClientIp(req) {
-    const forwardedFor = req?.headers?.['x-forwarded-for'];
-    if (Array.isArray(forwardedFor) && forwardedFor.length > 0) {
-      return forwardedFor[0].split(',')[0].trim();
+  stop() {
+    if (this._pollTimer) {
+      clearInterval(this._pollTimer);
+      this._pollTimer = null;
     }
-    if (typeof forwardedFor === 'string') {
-      return forwardedFor.split(',')[0].trim();
+    if (this._heartbeatTimer) {
+      clearInterval(this._heartbeatTimer);
+      this._heartbeatTimer = null;
     }
-    const socket = req?.socket;
-    return socket?.remoteAddress || 'unknown';
+    logger.info("PriceSubscriptionManager stopped");
   }
 
-  /** Register a new WebSocket connection. Returns false when at capacity or draining. */
-  add(ws, req = {}) {
+  async shutdown() {
+    this._draining = true;
+    this.stop();
+
+    const closePromises = [];
+    for (const [ws] of Array.from(this._clients.entries())) {
+      closePromises.push(
+        new Promise((resolve) => {
+          const timer = setTimeout(() => {
+            try {
+              ws.terminate?.();
+            } catch {
+              // ignore termination errors
+            }
+            this._remove(ws);
+            resolve();
+          }, 3000);
+          timer.unref?.();
+
+          ws.once?.("close", () => {
+            clearTimeout(timer);
+            this._remove(ws);
+            resolve();
+          });
+
+          if (ws.readyState === WebSocket.OPEN) {
+            ws.close(1001, "Server shutting down");
+          } else {
+            clearTimeout(timer);
+            try {
+              ws.terminate?.();
+            } catch {
+              // ignore termination errors
+            }
+            this._remove(ws);
+            resolve();
+          }
+        })
+      );
+    }
+
+    await Promise.all(closePromises);
+    this._clients.clear();
+    this._assetSubscribers.clear();
+    this._previousPrices.clear();
+    this._clientIpBySocket.clear();
+    this._connectionsByIp.clear();
+    logger.info("All WebSocket connections closed");
+  }
+
+  add(ws, req = null) {
     if (this._draining) {
-      ws.close(1013, 'Server shutting down');
+      ws.close(1001, "Server is shutting down");
       return false;
     }
 
-    const clientIp = this._getClientIp(req).replace(/^::ffff:/, '');
-    const currentByIp = this._connectionsByIp.get(clientIp) || 0;
-
-    if (this._clients.size >= MAX_CONNECTIONS || currentByIp >= MAX_CONNECTIONS_PER_IP) {
-      ws.close(1013, 'Max connections reached');
+    const clientIp = this._extractClientIp(ws, req);
+    const activeForIp = this._connectionsByIp.get(clientIp) || 0;
+    if (activeForIp >= this._maxConnectionsPerIp) {
+      logger.warn("WS connection rejected: per-IP limit exceeded", {
+        ip: clientIp,
+        activeConnections: activeForIp,
+        maxConnectionsPerIp: this._maxConnectionsPerIp,
+      });
+      this._send(ws, {
+        type: "error",
+        code: "TOO_MANY_CONNECTIONS",
+        message: `Maximum ${this._maxConnectionsPerIp} concurrent WebSocket connections per IP exceeded`,
+      });
+      ws.close(1008, "Too many connections from this IP");
       return false;
     }
 
-    this._clients.set(ws, { assets: new Set(), missedPings: 0 });
-    this._clientIpBySocket.set(ws, clientIp);
-    this._connectionsByIp.set(clientIp, currentByIp + 1);
-    updateGauge(1);
-    logger.info('WS client connected', { ip: clientIp, total: this._clients.size });
-
-    ws.on('message', (raw) => this._handleMessage(ws, raw));
-    ws.on('close', () => this._remove(ws));
-    ws.on('error', (err) => {
-      logger.warn('WS client error', { error: err.message });
+    const onMessage = (raw) => this._onMessage(ws, raw);
+    const onClose = () => this._remove(ws);
+    const onError = (err) => {
+      logger.warn("WS client error", { error: err.message });
       this._remove(ws);
+    };
+    const onPong = () => {
+      const entry = this._clients.get(ws);
+      if (entry) entry.missedPings = 0;
+    };
+    const rawSocket = req?.socket || ws?._socket || null;
+    const onSocketClose = () => this._remove(ws);
+
+    this._clients.set(ws, {
+      assets: new Set(),
+      missedPings: 0,
+      listeners: { onMessage, onClose, onError, onPong, rawSocket, onSocketClose },
+    });
+    this._clientIpBySocket.set(ws, clientIp);
+    this._connectionsByIp.set(clientIp, activeForIp + 1);
+
+    ws.on("message", onMessage);
+    ws.on("close", onClose);
+    ws.on("error", onError);
+    ws.on?.("pong", onPong);
+    rawSocket?.once?.("close", onSocketClose);
+
+    this._send(ws, {
+      type: "welcome",
+      message: "Connected to SmartDrop price stream",
+      max_assets_per_client: MAX_ASSETS_PER_CLIENT,
+      poll_interval_ms: this._pollIntervalMs,
     });
-
     return true;
   }
 
-  _removeFromAssetIndex(ws, assets) {
-    for (const asset of assets) {
-      const subs = this._assetSubscribers.get(asset);
-      if (subs) {
-        subs.delete(ws);
-        if (subs.size === 0) this._assetSubscribers.delete(asset);
-      }
-    }
-  }
-
-  _addToAssetIndex(ws, assets) {
-    for (const asset of assets) {
-      let subs = this._assetSubscribers.get(asset);
-      if (!subs) {
-        subs = new Set();
-        this._assetSubscribers.set(asset, subs);
-      }
-      subs.add(ws);
-    }
-  }
-
-  _remove(ws) {
-    if (!this._clients.has(ws)) return;
-    const clientIp = this._clientIpBySocket.get(ws) || 'unknown';
-    const client = this._clients.get(ws);
-    this._removeFromAssetIndex(ws, client.assets);
-    this._clients.delete(ws);
-    this._clientIpBySocket.delete(ws);
-    const nextCount = (this._connectionsByIp.get(clientIp) || 1) - 1;
-    if (nextCount <= 0) {
-      this._connectionsByIp.delete(clientIp);
-    } else {
-      this._connectionsByIp.set(clientIp, nextCount);
+  _onMessage(ws, raw) {
+    if (typeof raw === "string" ? raw.length > MAX_MESSAGE_BYTES : raw.byteLength > MAX_MESSAGE_BYTES) {
+      this._sendError(ws, "MESSAGE_TOO_LARGE", `Message exceeds ${MAX_MESSAGE_BYTES} bytes`);
+      return;
     }
-    updateGauge(-1);
-    logger.info('WS client disconnected', { ip: clientIp, total: this._clients.size });
-  }
 
-  _handleMessage(ws, raw) {
     let msg;
     try {
-      msg = JSON.parse(raw.toString());
+      msg = JSON.parse(typeof raw === "string" ? raw : raw.toString("utf8"));
     } catch {
-      this._send(ws, { type: 'error', message: 'Invalid JSON' });
+      this._sendError(ws, "INVALID_JSON", "Message must be valid JSON");
       return;
     }
 
+    if (!msg || typeof msg.type !== "string") {
+      this._sendError(ws, "MISSING_TYPE", "Message must contain a string \"type\" field");
+      return;
+    }
+
+    switch (msg.type) {
+      case "subscribe":
+        this._handleSubscribe(ws, msg);
+        break;
+      case "unsubscribe":
+        this._handleUnsubscribe(ws, msg);
+        break;
+      case "pong":
+        this._handlePong(ws);
+        break;
+      default:
+        this._sendError(ws, "UNKNOWN_TYPE", `Unknown message type: ${msg.type}`);
+    }
+  }
+
+  _handleSubscribe(ws, msg) {
     const client = this._clients.get(ws);
     if (!client) return;
 
-    if (msg.action === 'subscribe') {
-      const requested = Array.isArray(msg.assets) ? msg.assets : [];
-      // Enforce cumulative cap: only add assets while under the limit (#124).
-      const added = [];
-      for (const a of requested) {
-        if (client.assets.size >= MAX_ASSETS_PER_CLIENT) break;
-        const key = String(a);
-        if (!client.assets.has(key)) {
-          client.assets.add(key);
-          added.push(key);
-        }
+    if (!Array.isArray(msg.assets) || msg.assets.length === 0) {
+      this._sendError(ws, "INVALID_ASSETS", "\"assets\" must be a non-empty array of asset codes");
+      return;
+    }
+
+    const normalized = [];
+    for (const raw of msg.assets) {
+      if (typeof raw !== "string") {
+        this._sendError(ws, "INVALID_ASSET_CODE", "Each asset code must be a string");
+        return;
       }
-      // Update reverse index for newly added assets.
-      if (added.length > 0) this._addToAssetIndex(ws, added);
-      if (added.length === 0 && client.assets.size >= MAX_ASSETS_PER_CLIENT && requested.length > 0) {
-        this._send(ws, { type: 'error', message: `Subscription cap reached (${MAX_ASSETS_PER_CLIENT} max)` });
-      } else {
-        this._send(ws, { type: 'subscribed', assets: [...client.assets] });
+      const code = raw.trim().toUpperCase();
+      if (!VALID_ASSET_RE.test(code)) {
+        this._sendError(ws, "INVALID_ASSET_CODE", `Invalid asset code: "${raw}". Must be 1-12 uppercase alphanumeric characters`);
+        return;
       }
+      normalized.push(code);
+    }
 
-    } else if (msg.action === 'unsubscribe') {
-      const toRemove = Array.isArray(msg.assets) ? msg.assets : [];
-      // Update reverse index before removing from client's asset set.
-      this._removeFromAssetIndex(ws, toRemove.map(String));
-      for (const a of toRemove) client.assets.delete(String(a));
-      this._send(ws, { type: 'unsubscribed', assets: [...client.assets] });
-
-    } else if (msg.action === 'pong') {
-      client.missedPings = 0;
+    const combined = new Set([...client.assets, ...normalized]);
+    if (combined.size > MAX_ASSETS_PER_CLIENT) {
+      this._sendError(
+        ws,
+        "TOO_MANY_ASSETS",
+        `Cannot subscribe to more than ${MAX_ASSETS_PER_CLIENT} assets (currently ${client.assets.size}, requested +${normalized.filter((a) => !client.assets.has(a)).length})`
+      );
+      return;
+    }
 
-    } else {
-      this._send(ws, { type: 'error', message: `Unknown action: ${msg.action}` });
+    for (const code of normalized) {
+      client.assets.add(code);
+      if (!this._assetSubscribers.has(code)) {
+        this._assetSubscribers.set(code, new Set());
+      }
+      this._assetSubscribers.get(code).add(ws);
     }
+
+    this._send(ws, {
+      type: "subscribed",
+      assets: Array.from(client.assets),
+    });
+
+    this._sendInitialSnapshot(ws, normalized).catch((err) => {
+      logger.warn("Failed to send initial price snapshot", { error: err.message });
+    });
   }
 
-  _send(ws, payload) {
-    if (ws.readyState !== WebSocket.OPEN) return;
-    try {
-      ws.send(JSON.stringify(payload));
-    } catch (err) {
-      logger.warn('WS send failed', { error: err.message });
+  _handleUnsubscribe(ws, msg) {
+    const client = this._clients.get(ws);
+    if (!client) return;
+
+    if (!Array.isArray(msg.assets) || msg.assets.length === 0) {
+      this._sendError(ws, "INVALID_ASSETS", "\"assets\" must be a non-empty array of asset codes");
+      return;
     }
-  }
 
-  /**
-   * Called after each price refresh cycle with a map of assetKey → newPrice.
-   * Pushes updates to subscribers whose watched asset changed by > 0.1%.
-   */
-  notifyPriceUpdates(freshPrices) {
-    for (const [assetKey, { price, source }] of Object.entries(freshPrices)) {
-      const prev = this._previousPrices.get(assetKey);
-
-      if (prev !== undefined && prev > 0) {
-        const changePct = ((price - prev) / prev) * 100;
-        if (Math.abs(changePct) > PRICE_CHANGE_THRESHOLD_PCT) {
-          const update = {
-            type: 'price_update',
-            asset: assetKey,
-            price_usd: price,
-            previous_price_usd: prev,
-            change_pct: parseFloat(changePct.toFixed(4)),
-            source,
-            timestamp: new Date().toISOString(),
-          };
-          this._broadcast(assetKey, update);
+    for (const raw of msg.assets) {
+      if (typeof raw !== "string") continue;
+      const code = raw.trim().toUpperCase();
+      client.assets.delete(code);
+      const subs = this._assetSubscribers.get(code);
+      if (subs) {
+        subs.delete(ws);
+        if (subs.size === 0) {
+          this._assetSubscribers.delete(code);
+          this._previousPrices.delete(code);
         }
       }
-
-      this._previousPrices.set(assetKey, price);
     }
+
+    this._send(ws, {
+      type: "unsubscribed",
+      assets: Array.from(client.assets),
+    });
   }
 
-  _broadcast(assetKey, payload) {
-    const subscribers = this._assetSubscribers.get(assetKey);
-    if (!subscribers) return;
-    for (const ws of subscribers) {
-      this._send(ws, payload);
+  _handlePong(ws) {
+    const client = this._clients.get(ws);
+    if (client) {
+      client.missedPings = 0;
     }
   }
 
-  /** Start sending heartbeat pings every 30 s; disconnect idle sockets. */
-  startHeartbeat() {
-    if (this._pingTimer) return;
-    this._pingTimer = setInterval(() => {
-      // #363 — Collect timed-out clients first, then disconnect after
-      // iteration. Modifying a Map during for-of iteration is undefined
-      // behavior in JS; snapshotting the keys avoids the mutation.
-      const timedOut = [];
-      for (const [ws, client] of this._clients) {
-        if (client.missedPings >= MAX_MISSED_PINGS) {
-          timedOut.push(ws);
-          continue;
+  _remove(ws) {
+    const client = this._clients.get(ws);
+    if (!client) return false;
+
+    if (client.listeners) {
+      const { onMessage, onClose, onError, onPong, rawSocket, onSocketClose } = client.listeners;
+      ws.removeListener?.("message", onMessage);
+      ws.removeListener?.("close", onClose);
+      ws.removeListener?.("error", onError);
+      ws.removeListener?.("pong", onPong);
+      rawSocket?.removeListener?.("close", onSocketClose);
+    }
+
+    for (const code of client.assets) {
+      const subs = this._assetSubscribers.get(code);
+      if (subs) {
+        subs.delete(ws);
+        if (subs.size === 0) {
+          this._assetSubscribers.delete(code);
+          this._previousPrices.delete(code);
         }
-        client.missedPings += 1;
-        this._send(ws, { type: 'ping' });
       }
-      for (const ws of timedOut) {
-        logger.info('WS client timed out, disconnecting');
-        ws.terminate();
-        this._remove(ws);
+    }
+    client.assets.clear();
+
+    this._clients.delete(ws);
+    const clientIp = this._clientIpBySocket.get(ws);
+    if (clientIp) {
+      this._clientIpBySocket.delete(ws);
+      const nextCount = (this._connectionsByIp.get(clientIp) || 1) - 1;
+      if (nextCount > 0) {
+        this._connectionsByIp.set(clientIp, nextCount);
+      } else {
+        this._connectionsByIp.delete(clientIp);
       }
-    }, PING_INTERVAL_MS);
+    }
+    return true;
   }
 
-  stopHeartbeat() {
-    if (this._pingTimer) {
-      clearInterval(this._pingTimer);
-      this._pingTimer = null;
+  _extractClientIp(ws, req) {
+    const forwardedFor = req?.headers?.["x-forwarded-for"];
+    if (typeof forwardedFor === "string" && forwardedFor.trim().length > 0) {
+      return forwardedFor.split(",")[0].trim();
     }
-  }
 
-  /**
-   * Gracefully drain all connected clients during server shutdown.
-   * Broadcasts a shutdown warning, then sends close frames, and force-closes
-   * any connections still open after `drainTimeoutMs` (issue #248).
-   */
-  drain(drainTimeoutMs = 5000) {
-    this.stopHeartbeat();
-    this._draining = true;
+    const candidates = [
+      req?.ip,
+      req?.socket?.remoteAddress,
+      req?.connection?.remoteAddress,
+      ws?._socket?.remoteAddress,
+    ];
+    const found = candidates.find((ip) => typeof ip === "string" && ip.trim().length > 0);
+    return found ? found.trim() : "unknown";
+  }
 
-    // Issue #364: freeze the exact set of sockets being drained right here,
-    // in the same synchronous tick as the flag flip above — add() checks
-    // this._draining as its very first statement, so nothing can be
-    // inserted into this._clients after this line runs. Every later phase
-    // below (which resumes asynchronously via setTimeout, real yield
-    // points where a lot can happen in the live map) iterates this frozen
-    // list rather than the live this._clients, so a client connecting
-    // (and being rejected) mid-drain, or one that legitimately disconnects
-    // and is removed via _remove() between phases, can never change what
-    // this specific drain run touches.
-    const drainedSockets = [...this._clients.keys()];
-    const clientCount = drainedSockets.length;
-    if (clientCount === 0) return Promise.resolve();
-
-    this._drainStats = { warned: clientCount, closed: 0, forceClosed: 0 };
-    logger.info('Draining WebSocket connections', { count: clientCount, drain_timeout_ms: drainTimeoutMs });
-
-    // Phase 1: Broadcast shutdown warning so clients can prepare
-    for (const ws of drainedSockets) {
+  async _sendInitialSnapshot(ws, assetCodes) {
+    for (const code of assetCodes) {
+      if (!this._clients.has(ws)) return;
       try {
-        this._send(ws, { type: 'server_shutdown', message: 'Server is shutting down', drain_timeout_ms: drainTimeoutMs });
+        const data = await this._oracle.getPrice(code);
+        if (data && data.price_usd !== null && this._clients.has(ws)) {
+          this._send(ws, {
+            type: "price_update",
+            asset_code: data.asset_code,
+            price_usd: data.price_usd,
+            change_24h: data.change_24h ?? null,
+            source: data.source,
+            timestamp: data.last_updated || new Date().toISOString(),
+          });
+        }
       } catch {
-        // already closed or errored — ignore
+        // Individual asset failure on snapshot is non-fatal
       }
     }
+  }
 
-    // Phase 2: After a brief grace period for clients to finish in-flight work,
-    // send close frames to initiate orderly disconnection
-    const closeDelayMs = Math.min(1000, drainTimeoutMs / 2);
-    return new Promise((resolve) => {
-      const closeTimer = setTimeout(() => {
-        for (const ws of drainedSockets) {
-          if (!this._clients.has(ws)) continue; // already disconnected on its own
-          try {
-            ws.close(1001, 'Server shutting down');
-            this._drainStats.closed++;
-          } catch {
-            // already closed or errored — ignore
-          }
-        }
-      }, closeDelayMs);
-
-      closeTimer.unref();
-
-      const deadline = setTimeout(() => {
-        let remaining = 0;
-        for (const ws of drainedSockets) {
-          if (!this._clients.has(ws)) continue; // already disconnected on its own
-          remaining++;
-          try { ws.terminate(); } catch { /* ignore */ }
-          this._remove(ws);
-          this._drainStats.forceClosed++;
-        }
-        logger.info('WebSocket drain complete', {
-          total: clientCount,
-          gracefully_closed: this._drainStats.closed,
-          force_closed: remaining,
-        });
-        resolve();
-      }, drainTimeoutMs);
-
-      deadline.unref();
-    });
+  async _pollAndBroadcast() {
+    const assets = this.trackedAssets;
+    if (assets.length === 0) return;
+
+    const results = await Promise.allSettled(
+      assets.map((code) => this._oracle.getPrice(code))
+    );
+
+    for (let i = 0; i < assets.length; i++) {
+      const code = assets[i];
+      const outcome = results[i];
+      if (outcome.status !== "fulfilled" || !outcome.value) continue;
+
+      const data = outcome.value;
+      if (data.price_usd === null) continue;
+
+      const prev = this._previousPrices.get(code);
+      if (prev !== undefined && prev === data.price_usd) continue;
+      this._previousPrices.set(code, data.price_usd);
+
+      const subs = this._assetSubscribers.get(code);
+      if (!subs || subs.size === 0) continue;
+
+      const payload = {
+        type: "price_update",
+        asset_code: data.asset_code,
+        price_usd: data.price_usd,
+        change_24h: data.change_24h ?? null,
+        source: data.source,
+        timestamp: data.last_updated || new Date().toISOString(),
+      };
+
+      for (const ws of Array.from(subs)) {
+        this._send(ws, payload);
+      }
+    }
   }
 
-  get connectionCount() {
-    return this._clients.size;
+  _heartbeat() {
+    for (const [ws, client] of Array.from(this._clients.entries())) {
+      if (ws.readyState === WebSocket.CLOSING || ws.readyState === WebSocket.CLOSED) {
+        this._remove(ws);
+        continue;
+      }
+      if (client.missedPings >= MAX_MISSED_PINGS) {
+        logger.info("Terminating unresponsive WS client");
+        this._remove(ws);
+        ws.terminate?.();
+        continue;
+      }
+      client.missedPings += 1;
+      this._send(ws, { type: "ping", timestamp: new Date().toISOString() });
+    }
   }
 
-  get isDraining() {
-    return this._draining;
+  _send(ws, data) {
+    if (ws.readyState === WebSocket.CLOSING || ws.readyState === WebSocket.CLOSED) {
+      this._remove(ws);
+      return;
+    }
+    if (ws.readyState !== WebSocket.OPEN) return;
+    try {
+      ws.send(JSON.stringify(data));
+    } catch (err) {
+      logger.warn("WS send failed", { error: err.message });
+      this._remove(ws);
+    }
   }
 
-  get drainStats() {
-    return { ...this._drainStats };
+  _sendError(ws, code, message) {
+    this._send(ws, { type: "error", code, message });
   }
 }
 
-module.exports = new PriceSubscriptionManager();
-module.exports.PriceSubscriptionManager = PriceSubscriptionManager;
+module.exports = {
+  PriceSubscriptionManager,
+  MAX_ASSETS_PER_CLIENT,
+  MAX_MESSAGE_BYTES,
+  HEARTBEAT_INTERVAL_MS,
+  MAX_MISSED_PINGS,
+  DEFAULT_MAX_CONNECTIONS_PER_IP,
+};
\ No newline at end of file
diff --git a/test/wsCleanup.test.js b/test/wsCleanup.test.js
new file mode 100644
index 0000000..26c4f38
--- /dev/null
+++ b/test/wsCleanup.test.js
@@ -0,0 +1,64 @@
+const EventEmitter = require("events");
+const WebSocket = require("ws");
+const { PriceSubscriptionManager } = require("../src/ws/PriceSubscriptionManager");
+
+function createMockSocket(readyState = WebSocket.OPEN) {
+  const ws = new EventEmitter();
+  ws.readyState = readyState;
+  ws.sent = [];
+  ws.send = jest.fn((payload) => ws.sent.push(JSON.parse(payload)));
+  ws.close = jest.fn(() => {
+    ws.readyState = WebSocket.CLOSED;
+    ws.emit("close");
+  });
+  ws.terminate = jest.fn(() => {
+    ws.readyState = WebSocket.CLOSED;
+  });
+  return ws;
+}
+
+describe("WebSocket connection cleanup (#528)", () => {
+  test("detaches event listeners and frees per-IP and asset subscriptions on disconnect", () => {
+    const mgr = new PriceSubscriptionManager({ oracle: { getPrice: jest.fn().mockResolvedValue(null) } });
+    const rawSocket = new EventEmitter();
+    const ws = createMockSocket();
+
+    expect(mgr.add(ws, { ip: "10.0.0.5", socket: rawSocket })).toBe(true);
+    ws.emit("message", JSON.stringify({ type: "subscribe", assets: ["XLM", "USDC"] }));
+    expect(mgr.clientCount).toBe(1);
+    expect(mgr.trackedAssets).toEqual(["XLM", "USDC"]);
+
+    rawSocket.emit("close");
+    expect(mgr.clientCount).toBe(0);
+    expect(mgr.trackedAssets).toEqual([]);
+    expect(ws.listenerCount("message")).toBe(0);
+    expect(ws.listenerCount("close")).toBe(0);
+    expect(ws.listenerCount("error")).toBe(0);
+  });
+
+  test("removes dead/closed sockets automatically when send fails or socket is CLOSED", () => {
+    const mgr = new PriceSubscriptionManager({ oracle: { getPrice: jest.fn().mockResolvedValue(null) } });
+    const ws = createMockSocket();
+    mgr.add(ws, { ip: "10.0.0.6" });
+    ws.emit("message", JSON.stringify({ type: "subscribe", assets: ["XLM"] }));
+
+    ws.send.mockImplementationOnce(() => {
+      throw new Error("Broken pipe");
+    });
+    mgr._send(ws, { type: "ping" });
+
+    expect(mgr.clientCount).toBe(0);
+    expect(mgr.trackedAssets).toEqual([]);
+  });
+
+  test("cleans up non-open sockets during shutdown without leaking state", async () => {
+    const mgr = new PriceSubscriptionManager({ oracle: { getPrice: jest.fn().mockResolvedValue(null) } });
+    const ws = createMockSocket();
+    mgr.add(ws, { ip: "10.0.0.7" });
+    ws.readyState = WebSocket.CLOSING;
+
+    await mgr.shutdown();
+    expect(mgr.clientCount).toBe(0);
+    expect(ws.terminate).toHaveBeenCalled();
+  });
+});
\ No newline at end of file