[Spring] Batch : STO 토큰 상세 페이지 캔들 차트

프로젝트 진행 중 STO 토큰 상세 페이지에서 보여지는 '캔들 차트' 를 구현하는 역할을 담당하게 되었다. (주식과 유사하게 동작)

 

사용 기술 : Spring Boot 3, Spring Batch 5

 

 

아래는 상세 페이지의 목업이다.

 

화면 좌측 상단에 보이는 차트가 '캔들 차트' 라고 한다.

캔들 차트에는

- 단위 기간 동안 체결된 거래 가격 중 가장 높은 가격을 의미하는 '고가',

- 단위 기간 동안 체결된 거래 가격 중 가장 낮은 가격을 의미하는 '저가',

- 단위 기간 중 처음으로 체결된 가격을 의미하는 '시가'

- 단위 기간 중 마지막으로 체결된 가격을 의미하는 '종가' 

가 보여진다.

 

 

캔들차트 개념

https://kbthink.com/stock/chart.html

캔들차트 보는 법｜주식 초보를 위한 캔들차트 종류부터 패턴까지 기초 총정리

 

 

 

요구사항은 아래와 같다.

 

- 사용자는 상세 페이지 접속 시 기본적으로 '분' 에 캔들 차트를 조회한다 :
- 사용자가 상세 페이지 접속 시 '분' 에 대한 캔들 차트 데이터를 DB에서 조회해서 화면에 전달한다.

- 사용자가 접속한 시점 이전의 분에 대해서는 1분 동안의 고가와 저가를 Http Rest로 화면에 전달한다. 
- 사용자가 접속한 시점의 분에 대한 고가와 저가는 실시간 웹소켓으로 구현한다.

 

예를 들어 사용자가 4월 7일 오후 7시 2분 53초에 상세 페이지에 접근했다면, 사용자에게는 Http Rest로

4월 7일 오후 7시 1 ~ 2분 사이의 고가와 저가,

4월 7일 오후 7시 0 ~ 1분 사이의 고가와 저가,

4월 7일 오후 6시 59분 ~  7시 0분 사이의 고가와 저가,

4월 7일 오후 6시 58분 ~  59분 사이의 고가와 저가

와 같이 리스트 형태로 화면에 전달해야 한다.

 

아래는 관련 테이블의 ERD이며 1분을 예시로 들었지만 마찬가지로 1시간, 1일, 1달, 1년 단위를 함께 테이블에서 조회해서 나타낸다.

 

 

구현

사용자가 상세 페이지로 접속할 때 기본값이 1분 단위이기 때문에 CANDLE_MINUTES의 데이터를 가져와서 DTO로 화면에 전달한다.

사용자가 시간, 일, 달, 년을 누를 경우 각 테이블에 맞게 스프링 부트 서버를 거쳐 DB에서 데이터를 조회하여 화면에 반환하는 로직이다.

사용자가 접속한 뒤 실시간으로 갱신되는 고가와 저가에 있어서만 Stomp로 구현하며 이에 대해서는 추후 작성할 예정이다.

 

CANDLE_MINUTES, HOURS, DAYS, MONTH, YEAR에 대해 값을 채워주는 작업을 스프링 배치 서버에서 담당한다.

스프링 배치는 간단한 작업을 구현할 수 있는 Tasklet 방식과 무거운 작업을 구현할 수 있는 Chunk 방식이 있으며 나는 각 토큰(주식) 별로 매 분마다 CANDLE_MINUTES 테이블에 insert 작업이 일어나는 것이 가벼운 작업이 아니라고 판단하여 Chunk 방식으로 구현했다.

 

 

아래는 배치에 대한 간략한 개념들이다.

https://nippyclouding.tistory.com/64

[Spring] Batch 1

 

 

배치로 구현할 목록

 

- 1분 간격으로 TRADES 테이블 (거래 체결 완료)에서 단위 기간(1분) 내에 체결된 거래들을 모두 모아 
체결 거래 리스트 중 가장 큰 가격을 고가로, 가장 작은 가격을 저가로 CANDLE_MINUTES 테이블에 insert 한다.

- 1시간 간격으로 CANDLE_MINUTES 테이블의 고가, 저가 데이터 중 가장 큰 가격을 고가, 가장 작은 가격을 저가로
CANDLE_HOURS 테이블에 insert 한다.

- 1일 간격으로 CANDLE_HOURS 테이블의 고가, 저가 데이터 중 가장 큰 가격을 고가, 가장 작은 가격을 저가로
CANDLE_DAYS 테이블에 insert 한다.

- 1달 간격으로 CANDLE_DAYS 테이블의 고가, 저가 데이터 중 가장 큰 가격을 고가, 가장 작은 가격을 저가로
CANDLE_MONTHS 테이블에 insert 한다.

- 1년 간격으로 CANDLE_MONTH 테이블의 고가, 저가 데이터 중 가장 큰 가격을 고가, 가장 작은 가격을 저가로
CANDLE_YEARS 테이블에 insert 한다.

 

 

배치 작업을 하기 전 해당 스프링 부트 서버가 배치 작업을 수행할 수 있도록 Annotation을 설정해야 한다.

@SpringBootApplication
@EnableScheduling
public class BatchApplication {
    public static void main(String[] args) {
       SpringApplication.run(BatchApplication.class, args);
    }
}

 

 

 

아래는 1분마다 배치가 동작하는 CANDLE_MINUTES 구현 코드를 가져온 예시이다.

@Configuration
@RequiredArgsConstructor
public class CandleMinuteJobConfig {

    private final JobRepository jobRepository;
    private final PlatformTransactionManager txManager;
    private final TokenRepository tokenRepository;
    private final CandleMinuteProcessor processor;
    private final CandleMinuteWriter writer;

    @Bean
    public Job candleMinuteJob() {
        return new JobBuilder("candleMinuteJob", jobRepository)
                .start(candleMinuteStep())
                .build();
    }

    @Bean
    public Step candleMinuteStep() {
        return new StepBuilder("candleMinuteStep", jobRepository)
                .<Token, CandleMinute>chunk(50, txManager)
                .reader(candleMinuteReader())
                .processor(processor)
                .writer(writer)
                .build();
    }

    @Bean
    public ListItemReader<Token> candleMinuteReader() {
        List<Token> tokens = tokenRepository.findAllByTokenStatus(TokenStatus.TRADING);
        return new ListItemReader<>(tokens);
    }
}

 

 

1. Job을 Bean으로 등록

 

Job은 배치가 수행할 작업의 단위이다. 하나의 Job은 여러 Step으로 이루어져있다.

@Bean
public Job candleMinuteJob() {
    return new JobBuilder("candleMinuteJob", jobRepository)
            .start(candleMinuteStep())
            .build();
}

 

 

 

2. Step을 Bean으로 등록

 

Step은 Job을 구성할 대상들이다. 해당 배치 작업에서는 하나의 Step만으로 이루어져있다.

<Token, CandleMinute>chunk(50, txManager)

- Token : 배치 작업의 입력 값으로 Token이 들어오고

- CandleMinute : 배치 작업의 출력 값으로 CandleMinute이 나간다. (DB에 저장)

50, txManager : 50개의 Token 단위로 배치 작업이 동작하며 중간에 실패 시 트랜잭션이 모두 롤백된다.

@Bean
public Step candleMinuteStep() {
    return new StepBuilder("candleMinuteStep", jobRepository)
            .<Token, CandleMinute>chunk(50, txManager)
            .reader(candleMinuteReader())
            .processor(processor)
            .writer(writer)
            .build();
}

 

reader -> processor -> writer

 

3 - 1. Reader을 등록 : Reader은 별도 클래스로 구성하지 않게 하며 단순히 @Bean으로 Reader를 등록한다.

 

TokenStatus가 Trading (거래 가능) 인 상태의 Token들을 데이터베이스에서 조회하여 List로 Processor에 전달한다. 

@Bean
public ListItemReader<Token> candleMinuteReader() {
    List<Token> tokens = tokenRepository.findAllByTokenStatus(TokenStatus.TRADING);
    return new ListItemReader<>(tokens);
}

 

 

 

3 - 2. Processor : 배치 작업의 실질적인 동작이 일어나는 곳

 

@Bean으로 등록하지 않고 @Component로 수동 등록하며, implements ItemProcessor을 구현한다.

<Token, CandleMinute>의 Token은 입력으로 Reader에 들어올 데이터, CandleMinute은 출력으로 Writer에 나갈 데이터 타입이다.

Reader에서 List으로 값이 들어오며 Processor은 List 내부 값을 하나씩 처리한다.

Writer로 내보낼 때도 List로 전달하는 것이 아닌 단일 데이터 (CandleMinute)로 전달한다.

 

1분 동안 거래가 일어난 Trade를 리스트로 조회하여 해당 거래 내역들 중 가장 큰 값을 고가, 가장 작은 값을 저가로 다룬다.

@Component
@RequiredArgsConstructor
public class CandleMinuteProcessor implements ItemProcessor<Token, CandleMinute> { 
    // 들어오는 데이터 : Token, 나가는 데이터 : CandleMinute

    private final TradeRepository tradeRepository;

    @Override
    public CandleMinute process(Token token) throws Exception {
        // 배치가 오후 2시 4분 52초에 시작했을 경우 : truncatedTo로 52초를 버린다 
        // => to : 4분 00초, from : to 에서 1분 지난 시간까지
        LocalDateTime to = LocalDateTime.now().truncatedTo(ChronoUnit.MINUTES);
        LocalDateTime from = to.minusMinutes(1);

        List<Trade> trades = tradeRepository.findByTokenIdAndExecutedAtBetween(token.getTokenId(), from, to);

        if (trades.isEmpty()) return null; // 거래가 없으면 null 반환

        // list 에서 가장 큰 값, 가장 작은 값 찾기
        double highPrice = trades.stream().mapToDouble(Trade::getTradePrice).max().getAsDouble();
        double lowPrice  = trades.stream().mapToDouble(Trade::getTradePrice).min().getAsDouble();

        return CandleMinute.builder()
                .tokenId(token.getTokenId())
                .highPrice(highPrice)
                .lowPrice(lowPrice)
                .candleTime(from)
                .build();
    }
}

 

 

 

3 - 3. Writer : 배치 작업을 한 뒤 저장하거나 출력하는 동작이 이루어지는 곳

 

마찬가지로 @Bean으로 등록하지 않고 @Component로 수동 등록하며, implements ItemWriter를 구현한다.

saveAll 메서드를 통해 벌크 insert로 하나의 insert 쿼리를 이용해 수많은 데이터들을 데이터베이스에 insert한다.

@Component
@RequiredArgsConstructor
public class CandleMinuteWriter implements ItemWriter<CandleMinute> { // 쓰는 데이터 : CandleMinute
    // reader -> processor -> writer 흐름

    private final CandleMinuteRepository candleMinuteRepository;

    @Override
    public void write(Chunk<? extends CandleMinute> chunk) throws Exception {
        candleMinuteRepository.saveAll(chunk.getItems()); // 벌크 insert : 하나의 쿼리로 전부 insert
    }
}

 

 

 

 

배치 작업을 모두 구현하면 배치 서버에서 작업이 돌아가도록 구현해야 한다.

먼저 properties 또는 yaml 파일에 서버 실행 시 배치가 무분별하게 동작하지 않도록 하기 위해 아래 코드를 추가한다.

// application.properties 기준
spring.batch.job.enabled=false

 

 

이후 배치 서버에서 어떤 배치 작업을 언제 수행할지를 설정하는 'Scheduler'를 Bean으로 등록한다.

@Component
@RequiredArgsConstructor
@Slf4j
public class CandleScheduler {

    private final JobLauncher jobLauncher;

    @Qualifier("candleMinuteJob")
    private final Job candleMinuteJob;


    @Scheduled(cron = "0 * * * * *")  // 매 분 0초
    public void runCandleMinuteJob() throws Exception {
        JobParameters params = new JobParametersBuilder()
                .addLong("currentTime", System.currentTimeMillis())
                .toJobParameters();

        log.info("캔들 차트 1분 배치 고가 저가 작업 시작");
        jobLauncher.run(candleMinuteJob, params);
    }
}

 

 

배치 서버를 실행한 뒤 매 분마다 아래와 같은 로그와 함께 배치 작업이 수행되는 것을 확인할 수 있다.