socket.io로 Upbit거래소의 가상화폐 가격 수집하기 - InfluxDB로 비트코인 분석하기 #2

본 시리즈는 각 프로그램의 개념을 다루는 글이 아닌, 프로젝트 관점에서 툴을 활용하는 내용을 담고있다.

따라서 개념 설명보다는 프로젝트에서 툴을 어떻게 활용하는지에 대한 기록이다.

 

Series

• docker-compose로 InfluxDB 설치하기 - InfluxDB로 비트코인 분석하기 #1
• websocket으로 Upbit거래소의 가상화폐 가격 수집하기 - InfluxDB로 비트코인 분석하기 #2
• InfluxDB에서 제공하는 Query로 비트코인 보조지표 만들기  - InfluxDB로 비트코인 분석하기 #3 [작성중]
• Grafana를 활용하여 InfluxDB 시계열데이터 시각화 하기 - InfluxDB로 비트코인 분석하기 #4 [작성중]

데이터 수집

프로젝트 생성

$ npm init

 

websocket, influxdb-client 설치

$ npm install --save ws @influxdata/influxdb-client

 

nodejs와 socket을 활용한 데이터 수신하여 InfluxDB에 적재

const WebSocket = require('ws');
const ws = new WebSocket('wss://api.upbit.com/websocket/v1');

const {InfluxDB, Point} = require('@influxdata/influxdb-client');

const org = process.env.INFLUXDB_ORG;

const client = new InfluxDB({
  url: process.env.INFLUXDB_URL,
  token: process.env.INFLUXDB_TOKEN,
});

ws.on('open', function open(){
  const requestData = [
    {
      'ticket': 'UNIQUE_TICKET',
    }, {
      'type': 'trade',
      'codes': [
        process.env.INCOME_COIN_CODE,
      ],
      isOnlyRealtime: true,
    },
  ];
  ws.send(
    JSON.stringify(requestData),
  );
});

ws.on('message', function incoming(message){
  try{
    const incomingData = JSON.parse(message.toString());

    let connection;

    if(incomingData.code === process.env.INCOME_COIN_CODE)
      connection = client.getWriteApi(org, process.env.INFLUXDB_BUCKET);

    if(connection){
      connection.writePoint(
        new Point('trade')
          .timestamp(new Date(incomingData.timestamp))
          .intField('trade_volume', incomingData.trade_volume)
          .intField('trade_price', incomingData.trade_price)
          .intField('change_price', incomingData.change_price)
          .intField('change_price', incomingData.change_price)
          .intField('prev_closing_price', incomingData.prev_closing_price)
          .floatField('sequential_id', incomingData.sequential_id)
          .stringField('code', incomingData.code)
          .stringField('type', incomingData.type)
          .stringField('stream_type', incomingData.stream_type)
          .stringField('trade_date', incomingData.trade_date)
          .stringField('trade_time', incomingData.trade_time)
          .stringField('ask_bid', incomingData.ask_bid)
          .stringField('change', incomingData.change),
      );

      connection.close().then().catch(e => process.exit(0));
    }

  }catch(e){
    console.log(e);
    process.exit(0);
  }
});

ws.onclose = function(e){
  process.exit(0);
};

본 시리즈에서는 모듈을 컨테이너화 하여 코인별로 독립된 컨테이너로 실행하기 위해 필요한 변수들을 환경변수로 빼 두었다.

환경변수(process.env.*)를 직접 입력하여 테스트를 해볼 수 있다.

 

환경변수에 대한 설명은 아래 참조.

INCOME_COIN_CODE: 코인 종류 (KRW-BTC, KRW-ETH, 등)

INFLUXDB_ORG: InfluxDB ORG

INFLUXDB_URL: InfluxDB URL

INFLUXDB_TOKEN: InfluxDB Token

INFLUXDB_BUCKET: 데이터를 적재할 InfluxDB Bucket

 

배포

Dockerizing

폴더구조

docker-compose.yml
income
  |  Dockerfile
  |  index.js
  |  package.json

 

Dockerfile

FROM node:14
WORKDIR /sources
COPY . .
RUN npm install
CMD [ "node", "index.js" ]

 

docker-compose.yml

# docker-compose.yml
version: '3'
services:
  influxdb:
    image: influxdb:2.1.1
    volumes:
      - ./influxdb:/var/lib/influxdb2
    ports:
      - 28086:8086
      - 28083:8083
    restart: always
    
  income-upbit-krw-btc:
    build:
      context: ./income/.
    environment:
      - INCOME_COIN_CODE=KRW-BTC
      - INFLUXDB_ORG=
      - INFLUXDB_URL=
      - INFLUXDB_TOKEN=
      - INFLUXDB_BUCKET=UPBIT-KRW-BTC
    restart: always

다른 코인도 데이터를 받아보고 싶다면 InfluxDB Bucket을 생성하고 service를 추가해 주면 된다.

 

운영

서비스 실행

$ sudo docker-compose up -d income-upbit-krw-btc

influxDB는 이전 글에서 이미 실행시켜 두었으니 income-upbit-krw-btc 서비스만 실행.

 

데이터 적재 확인

 

참조

WebSocket을 이용한 업비트 시세 수신

업비트 개발자 센터

socket.io-client - npm

socket.io-client