> ## Documentation Index
> Fetch the complete documentation index at: https://docs.chainstream.io/llms.txt
> Use this file to discover all available pages before exploring further.

# 실시간 스트리밍

> WebSocket 실시간 데이터 스트리밍 가이드

ChainStream은 강력한 실시간 데이터 스트리밍 기능을 제공하여 개발자가 온체인 이벤트, 트랜잭션, 상태 변경을 즉시 수신할 수 있습니다. 이 문서에서는 WebSocket 연결, 구독 메커니즘, 모범 사례를 다룹니다.

***

## 연결

### WebSocket 엔드포인트

```
wss://realtime-dex.chainstream.io/connection/websocket
```

### 인증

연결 수립 시 URL에 Access Token을 제공합니다:

```
wss://realtime-dex.chainstream.io/connection/websocket?token=YOUR_ACCESS_TOKEN
```

<Tabs>
  <Tab title="SDK 사용 (추천)">
    SDK가 연결과 인증을 자동으로 처리합니다. subscribe 메서드를 호출하기만 하면 됩니다:

    ```javascript theme={null}
    import { ChainStreamClient } from '@chainstream-io/sdk';
    import { Resolution } from '@chainstream-io/sdk/openapi';

    const client = new ChainStreamClient(process.env.CHAINSTREAM_ACCESS_TOKEN);

    // 직접 구독, SDK가 연결과 인증을 자동으로 처리
    client.stream.subscribeTokenCandles({
      chain: 'sol',
      tokenAddress: '6p6xgHyF7AeE6TZkSmFsko444wqoP15icUSqi2jfGiPN',
      resolution: Resolution._1m,
      callback: (data) => {
        console.log('Received data:', data);
      }
    });
    ```

    <Note>
      SDK는 연결 상태를 자동으로 감지하고 필요할 때 연결을 수립합니다. 수동으로 `connect()`를 호출할 필요가 없습니다.
    </Note>
  </Tab>

  <Tab title="네이티브 WebSocket 사용">
    네이티브 WebSocket 사용 시 연결 후 `connect` 메시지를 보내 인증을 완료합니다:

    ```javascript theme={null}
    const token = process.env.CHAINSTREAM_ACCESS_TOKEN;
    const ws = new WebSocket(
      `wss://realtime-dex.chainstream.io/connection/websocket?token=${token}`
    );

    ws.onopen = () => {
      console.log('WebSocket connection established');
      
      // connect 메시지를 보내 인증 완료
      ws.send(JSON.stringify({
        connect: {
          token: token,
          name: 'js'
        },
        id: 1
      }));
    };

    ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      
      // connect 응답 처리
      if (data.connect) {
        console.log('✅ Authentication successful, client ID:', data.connect.client);
        
        // 인증 후 구독 시작
        ws.send(JSON.stringify({
          subscribe: {
            channel: 'dex-candle:sol_6p6xgHyF7AeE6TZkSmFsko444wqoP15icUSqi2jfGiPN_1m'
          },
          id: 2
        }));
      }
      
      // 구독 데이터 처리
      if (data.push) {
        console.log('Received data:', data.push.pub.data);
      }
    };
    ```
  </Tab>

  <Tab title="커맨드 라인 테스트">
    `wscat`으로 테스트:

    ```bash theme={null}
    wscat -c "wss://realtime-dex.chainstream.io/connection/websocket?token=YOUR_ACCESS_TOKEN"
    ```

    연결 후 connect 메시지 전송:

    ```json theme={null}
    {"connect":{"token":"YOUR_ACCESS_TOKEN","name":"test"},"id":1}
    ```
  </Tab>
</Tabs>

### 연결 응답

인증 성공 시 다음과 같은 응답을 받게 됩니다:

```json theme={null}
{
  "id": 1,
  "connect": {
    "client": "0f819f5f-7d8b-4949-9433-0e91bbfe1cdb",
    "version": "0.0.0 OSS",
    "expires": true,
    "ttl": 86002,
    "ping": 25,
    "pong": true
  }
}
```

| 필드       | 설명             |
| :------- | :------------- |
| `client` | 고유 클라이언트 식별자   |
| `ttl`    | 토큰 잔여 유효시간 (초) |
| `ping`   | 하트비트 간격 (초)    |
| `pong`   | pong 응답 지원 여부  |

***

## 구독 유형

ChainStream WebSocket은 다양한 데이터 구독 유형을 지원합니다:

| 카테고리      | 채널 접두사                                                                                                                                                                  | 설명                                 |
| :-------- | :---------------------------------------------------------------------------------------------------------------------------------------------------------------------- | :--------------------------------- |
| 캔들 (USD)  | `dex-candle:`                                                                                                                                                           | 토큰 캔들 (토큰 / 풀 / 페어 변형)             |
| 캔들 (네이티브) | `dex-candle-in-native:`                                                                                                                                                 | 위와 동일하나 체인의 네이티브 자산으로 가격 표시        |
| 풀 캔들      | `dex-pool-candle:` / `dex-pair-candle:`                                                                                                                                 | 풀 기준 또는 페어 기준 캔들                   |
| 토큰 통계     | `dex-token-stats:`                                                                                                                                                      | 다중 시간 창 거래 통계 (1m, 5m, ... 1W, 1M) |
| 홀더 통계     | `dex-token-holding:`                                                                                                                                                    | 홀더 분포 및 잔고 태그                      |
| 토큰 공급     | `dex-token-supply:`                                                                                                                                                     | 공급량 및 시가총액 업데이트                    |
| 유동성       | `dex-token-liquidity:` / `dex-token-total-liquidity:`                                                                                                                   | 최대 풀 / 전체 유동성                      |
| 신규 토큰     | `dex-new-token:` / `dex-new-tokens:` / `dex-new-tokens-metadata:`                                                                                                       | 신규 상장 (개별 이벤트 / 배치 / 메타데이터)        |
| 토큰 거래     | `dex-trade:`                                                                                                                                                            | 토큰 주소로 필터링된 거래                     |
| 지갑 잔고     | `dex-wallet-balance:`                                                                                                                                                   | 지갑 토큰 잔고 변동                        |
| 지갑 거래     | `dex-wallet-trade:`                                                                                                                                                     | 지갑 주소로 필터링된 거래                     |
| 지갑 PnL    | `dex-wallet-token-pnl:` / `dex-wallet-pnl-list:`                                                                                                                        | 토큰별 / 집계 지갑 PnL                    |
| 랭킹        | `dex-ranking-list:` / `dex-ranking-token-stats-list:` / `dex-ranking-token-holding-list:` / `dex-ranking-token-supply-list:` / `dex-ranking-token-bounding-curve-list:` | 랭킹 멤버십 + 토큰별 통계                    |
| DEX 풀     | `dex-pool-balance:`                                                                                                                                                     | 풀 유동성 스냅샷                          |

<Tip>
  전체 구독 유형, 파라미터, 응답 형식은 [WebSocket API 레퍼런스](/ko/api-reference/endpoint/websocket/api)를 참고하세요. SDK(`client.stream.subscribeTokenCandles`, `subscribeTokenStats`, `subscribeTokenTrade`, ...)는 이러한 채널 문자열을 래핑하므로 직접 구성할 필요가 없습니다.
</Tip>

### 구독 형식 예시

```javascript theme={null}
// 캔들 데이터 구독
ws.send(JSON.stringify({
  subscribe: {
    channel: 'dex-candle:sol_6p6xgHyF7AeE6TZkSmFsko444wqoP15icUSqi2jfGiPN_1m'
  },
  id: 2
}));

// 토큰 통계 구독
ws.send(JSON.stringify({
  subscribe: {
    channel: 'dex-token-stats:sol_6p6xgHyF7AeE6TZkSmFsko444wqoP15icUSqi2jfGiPN'
  },
  id: 3
}));

// 새 토큰 구독
ws.send(JSON.stringify({
  subscribe: {
    channel: 'dex-new-token:sol'
  },
  id: 4
}));
```

### 구독 해제

```javascript theme={null}
ws.send(JSON.stringify({
  unsubscribe: {
    channel: 'dex-candle:sol_6p6xgHyF7AeE6TZkSmFsko444wqoP15icUSqi2jfGiPN_1m'
  },
  id: 5
}));
```

***

## 메시지 형식

### 요청 메시지

**Connect 메시지 (인증):**

```json theme={null}
{
  "connect": {
    "token": "YOUR_ACCESS_TOKEN",
    "name": "client_name"
  },
  "id": 1
}
```

**Subscribe 메시지:**

```json theme={null}
{
  "subscribe": {
    "channel": "dex-candle:sol_xxx_1m"
  },
  "id": 2
}
```

**Unsubscribe 메시지:**

```json theme={null}
{
  "unsubscribe": {
    "channel": "dex-candle:sol_xxx_1m"
  },
  "id": 3
}
```

### 응답 메시지

**구독 확인:**

```json theme={null}
{
  "id": 2,
  "subscribe": {}
}
```

**데이터 푸시:**

```json theme={null}
{
  "push": {
    "channel": "dex-candle:sol_xxx_1m",
    "pub": {
      "data": {
        "o": 0.001234,
        "c": 0.001256,
        "h": 0.001280,
        "l": 0.001200,
        "v": 1234567,
        "t": 1706745600
      }
    }
  }
}
```

**오류 메시지:**

```json theme={null}
{
  "id": 2,
  "error": {
    "code": 100,
    "message": "invalid channel"
  }
}
```

***

## 하트비트

WebSocket 연결을 유지하려면 주기적인 하트비트 메시지가 필요합니다. connect 응답의 `ping` 필드(보통 25초)를 기준으로 이 간격 내에 하트비트를 보내세요:

```javascript theme={null}
// 하트비트 메시지
ws.send(JSON.stringify({}));

// 또는 ping 전송
ws.send(JSON.stringify({ ping: {} }));
```

<Warning>
  지정된 시간 내(보통 ping 간격의 3배)에 메시지가 전송되지 않으면 서버가 연결을 끊습니다.
</Warning>

***

## 전체 예시

<CodeGroup>
  ```javascript JavaScript theme={null}
  const WebSocket = require('ws');

  class ChainStreamWebSocket {
    constructor(accessToken) {
      this.accessToken = accessToken;
      this.ws = null;
      this.messageId = 0;
      this.reconnectAttempts = 0;
      this.maxReconnectAttempts = 10;
      this.subscriptions = new Set();
      this.pingInterval = null;
    }

    connect() {
      const url = `wss://realtime-dex.chainstream.io/connection/websocket?token=${this.accessToken}`;
      this.ws = new WebSocket(url);

      this.ws.onopen = () => {
        console.log('WebSocket connection established');
        this.reconnectAttempts = 0;
        
        // connect 메시지 전송
        this.send({
          connect: {
            token: this.accessToken,
            name: 'nodejs'
          }
        });
      };

      this.ws.onmessage = (event) => {
        const data = JSON.parse(event.data);
        this.handleMessage(data);
      };

      this.ws.onclose = (event) => {
        console.log(`Connection closed: ${event.code}`);
        this.stopPing();
        
        if (event.code !== 1000) {
          this.scheduleReconnect();
        }
      };

      this.ws.onerror = (error) => {
        console.error('WebSocket error:', error.message);
      };
    }

    handleMessage(data) {
      // connect 응답 처리
      if (data.connect) {
        console.log('✅ Authentication successful');
        this.startPing(data.connect.ping || 25);
        this.resubscribe();
        return;
      }

      // 데이터 푸시 처리
      if (data.push) {
        console.log(`[${data.push.channel}]`, data.push.pub.data);
        return;
      }

      // 오류 처리
      if (data.error) {
        console.error('Error:', data.error.message);
        return;
      }
    }

    send(message) {
      message.id = ++this.messageId;
      this.ws.send(JSON.stringify(message));
    }

    subscribe(channel) {
      this.subscriptions.add(channel);
      this.send({ subscribe: { channel } });
    }

    unsubscribe(channel) {
      this.subscriptions.delete(channel);
      this.send({ unsubscribe: { channel } });
    }

    resubscribe() {
      this.subscriptions.forEach(channel => {
        this.send({ subscribe: { channel } });
      });
    }

    startPing(interval) {
      this.pingInterval = setInterval(() => {
        if (this.ws.readyState === WebSocket.OPEN) {
          this.ws.send('{}');
        }
      }, interval * 1000);
    }

    stopPing() {
      if (this.pingInterval) {
        clearInterval(this.pingInterval);
        this.pingInterval = null;
      }
    }

    scheduleReconnect() {
      if (this.reconnectAttempts >= this.maxReconnectAttempts) {
        console.error('❌ Max reconnect attempts reached');
        return;
      }

      const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000);
      console.log(`⏳ Reconnecting in ${delay}ms... (attempt ${this.reconnectAttempts + 1})`);
      this.reconnectAttempts++;

      setTimeout(() => this.connect(), delay);
    }

    close() {
      this.stopPing();
      if (this.ws) {
        this.ws.close(1000, 'Normal closure');
      }
    }
  }

  // 사용 예시
  const client = new ChainStreamWebSocket(process.env.CHAINSTREAM_ACCESS_TOKEN);
  client.connect();

  // 연결 수립 후 구독
  setTimeout(() => {
    client.subscribe('dex-candle:sol_6p6xgHyF7AeE6TZkSmFsko444wqoP15icUSqi2jfGiPN_1m');
    client.subscribe('dex-new-token:sol');
  }, 1000);
  ```

  ```python Python theme={null}
  import asyncio
  import websockets
  import json
  import os

  class ChainStreamWebSocket:
      def __init__(self, access_token):
          self.access_token = access_token
          self.ws = None
          self.message_id = 0
          self.subscriptions = set()
          self.ping_interval = 25
      
      async def connect(self):
          url = f"wss://realtime-dex.chainstream.io/connection/websocket?token={self.access_token}"
          
          async with websockets.connect(url) as ws:
              self.ws = ws
              print("WebSocket connection established")
              
              # connect 메시지 전송
              await self.send({
                  "connect": {
                      "token": self.access_token,
                      "name": "python"
                  }
              })
              
              # 하트비트와 메시지 수신 시작
              await asyncio.gather(
                  self.heartbeat(),
                  self.receive()
              )
      
      async def send(self, message):
          self.message_id += 1
          message["id"] = self.message_id
          await self.ws.send(json.dumps(message))
      
      async def receive(self):
          async for message in self.ws:
              data = json.loads(message)
              await self.handle_message(data)
      
      async def handle_message(self, data):
          if "connect" in data:
              print("✅ Authentication successful")
              self.ping_interval = data["connect"].get("ping", 25)
              await self.resubscribe()
          elif "push" in data:
              channel = data["push"]["channel"]
              pub_data = data["push"]["pub"]["data"]
              print(f"[{channel}]", pub_data)
          elif "error" in data:
              print(f"Error: {data['error']['message']}")
      
      async def subscribe(self, channel):
          self.subscriptions.add(channel)
          await self.send({"subscribe": {"channel": channel}})
      
      async def resubscribe(self):
          for channel in self.subscriptions:
              await self.send({"subscribe": {"channel": channel}})
      
      async def heartbeat(self):
          while True:
              await asyncio.sleep(self.ping_interval)
              if self.ws and self.ws.open:
                  await self.ws.send("{}")

  # 사용 예시
  async def main():
      client = ChainStreamWebSocket(os.environ["CHAINSTREAM_ACCESS_TOKEN"])
      
      # 구독 사전 추가
      client.subscriptions.add("dex-candle:sol_6p6xgHyF7AeE6TZkSmFsko444wqoP15icUSqi2jfGiPN_1m")
      client.subscriptions.add("dex-new-token:sol")
      
      await client.connect()

  asyncio.run(main())
  ```
</CodeGroup>

***

## 모범 사례

### 성능 최적화

<CardGroup cols={2}>
  <Card title="필터 사용" icon="filter">
    필요한 데이터만 구독하여 대역폭을 줄이세요. CEL 표현식으로 데이터를 필터링하세요.
  </Card>

  <Card title="배치 처리" icon="layer-group">
    고빈도 데이터는 하나씩 처리하지 말고 배치로 처리하세요. 메시지 큐를 사용하여 버퍼링하세요.
  </Card>

  <Card title="로컬 캐싱" icon="database">
    토큰 정보와 같은 정적 데이터를 캐시하여 반복 처리를 줄이세요.
  </Card>

  <Card title="연결 재사용" icon="plug">
    하나의 연결로 여러 채널을 구독할 수 있습니다. 여러 연결을 생성하지 마세요.
  </Card>
</CardGroup>

### 오류 처리

1. **오류 이벤트 리스닝** — 연결 및 데이터 오류를 즉시 처리
2. **재시도 메커니즘 구현** — 재연결 시 지수 백오프 사용
3. **로그 기록** — 문제 해결을 위한 주요 이벤트 기록
4. **그레이스풀 디그레이데이션** — WebSocket 사용 불가 시 폴링으로 전환

### 리소스 관리

```javascript theme={null}
// ✅ 불필요한 채널은 즉시 구독 해제
client.unsubscribe('dex-candle:sol_xxx_1m');

// ✅ 연결을 우아하게 종료
function gracefulClose() {
  // 1. 하트비트 중지
  client.stopPing();
  
  // 2. 연결 종료
  client.close();
}
```

***

## 관련 문서

<CardGroup cols={2}>
  <Card title="WebSocket API 레퍼런스" icon="plug" href="/ko/api-reference/endpoint/websocket/api">
    전체 구독 유형 및 파라미터
  </Card>

  <Card title="가격 알림 봇" icon="bell" href="/ko/docs/tutorials/build-price-alert-bot">
    실습: 가격 모니터링 봇 만들기
  </Card>
</CardGroup>