Merge pull request #179 from CleanEngine/feat/trade-core

Feat/trade core

Author: caniro

build.gradle

@@ -77,6 +77,13 @@ dependencies {
 	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
 
 	testImplementation 'org.junit.platform:junit-platform-suite:1.10.0'
+
+	// QueryDSL
+	implementation 'com.querydsl:querydsl-jpa:5.1.0:jakarta'
+	annotationProcessor 'com.querydsl:querydsl-apt:5.1.0:jakarta'
+	annotationProcessor 'jakarta.persistence:jakarta.persistence-api'
+	annotationProcessor 'jakarta.annotation:jakarta.annotation-api'
+
 	// Spring Security + OAuth2
 	implementation 'org.springframework.boot:spring-boot-starter-security'
 	implementation 'org.springframework.boot:spring-boot-starter-oauth2-client'

src/main/java/com/cleanengine/coin/trade/application/TradeExecutedNotificationHandler.java

@@ -1,6 +1,6 @@
 package com.cleanengine.coin.trade.application;
 
-import com.cleanengine.coin.trade.entity.Trade;
+import com.cleanengine.coin.order.domain.Order;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.messaging.simp.SimpMessagingTemplate;
 import org.springframework.stereotype.Component;
@@ -23,30 +23,23 @@ public TradeExecutedNotificationHandler(SimpMessagingTemplate messagingTemplate)
     }
 
     @TransactionalEventListener
-    public void notifyAfterTradeExecuted(TradeExecutedEvent tradeExecutedEvent) {
-        Trade trade = tradeExecutedEvent.getTrade();
-        if (trade == null) {
-            log.error("체결 알림 실패! trade == null");
+    public void notifyAfterTradeExecuted(TradeOrderCompletedEvent tradeOrderCompletedEvent) {
+        // TODO : 평균단가는 별도 계산해야 함
+        Order order = tradeOrderCompletedEvent.getOrder();
+        if (order == null) {
+            log.error("체결 알림 실패! order == null");
             return ;
         }
 
-        Integer sellUserId = trade.getSellUserId();
-        Integer buyUserId = trade.getBuyUserId();
-        if (sellUserId == null || buyUserId == null) {
-            log.error("체결 알림 실패! sellUserId: {}, buyUserId: {}", sellUserId, buyUserId);
+        Integer userId = order.getUserId();
+        if (userId == null) {
+            log.error("체결 알림 실패! userId: {}", userId);
             return ;
         }
 
-        if (sellUserId != SELL_ORDER_BOT_ID) {
-            TradeExecutedNotifyDto soldDto = TradeExecutedNotifyDto.of(trade, ASK);
-            messagingTemplate.convertAndSend("/topic/tradeNotification/" + sellUserId, soldDto);
-        }
-        if (buyUserId != BUY_ORDER_BOT_ID) {
-            TradeExecutedNotifyDto boughtDto = TradeExecutedNotifyDto.of(trade, BID);
-            messagingTemplate.convertAndSend("/topic/tradeNotification/" + buyUserId, boughtDto);
-        }
-        if (sellUserId != SELL_ORDER_BOT_ID || buyUserId != BUY_ORDER_BOT_ID) {
-            log.debug("{} 체결 이벤트 구독 : {}원에 {}개, 매수인: {}, 매도인: {}", trade.getTicker(), trade.getPrice(), trade.getSize(), buyUserId, sellUserId );
+        if (userId != SELL_ORDER_BOT_ID && userId != BUY_ORDER_BOT_ID) {
+            TradeOrderCompletedNotifyDto notifyDto = TradeOrderCompletedNotifyDto.of(order);
+            messagingTemplate.convertAndSend("/topic/tradeNotification/" + userId, notifyDto);
         }
     }
 

src/main/java/com/cleanengine/coin/trade/application/TradeExecutedNotifyDto.java

@@ -1,15 +1,11 @@
 package com.cleanengine.coin.trade.application;
 
-import com.cleanengine.coin.common.error.BusinessException;
-import com.cleanengine.coin.common.response.ErrorStatus;
 import com.cleanengine.coin.order.domain.BuyOrder;
 import com.cleanengine.coin.order.domain.Order;
 import com.cleanengine.coin.order.domain.OrderStatus;
 import com.cleanengine.coin.order.domain.SellOrder;
 import com.cleanengine.coin.order.domain.spi.WaitingOrders;
 import com.cleanengine.coin.trade.entity.Trade;
-import com.cleanengine.coin.user.domain.Account;
-import com.cleanengine.coin.user.domain.Wallet;
 import com.cleanengine.coin.user.info.application.AccountService;
 import com.cleanengine.coin.user.info.application.WalletService;
 import lombok.Getter;
@@ -33,10 +29,11 @@ public class TradeExecutor {
     private final AccountService accountService;
     @Getter
     private final TradeExecutedEventPublisher tradeExecutedEventPublisher;
+    private final TradeOrderCompletedEventPublisher tradeOrderCompletedEventPublisher;
     private final TradeService tradeService;
 
     @Transactional(propagation = Propagation.REQUIRES_NEW, isolation = Isolation.READ_COMMITTED)
-    public void executeTrade(WaitingOrders waitingOrders, TradePair<Order, Order> tradePair, String ticker) {
+    public Trade executeTrade(WaitingOrders waitingOrders, TradePair<Order, Order> tradePair, String ticker) {
         BuyOrder buyOrder = tradePair.getBuyOrder();
         SellOrder sellOrder = tradePair.getSellOrder();
         log.trace("{} - 체결 시작: 매수[{} {}원 {}개] / 매도[{} {}원 {}개]", ticker, buyOrder.getId(), buyOrder.getPrice(), buyOrder.getRemainingSize(),
@@ -64,8 +61,7 @@ public void executeTrade(WaitingOrders waitingOrders, TradePair<Order, Order> tr
         sellOrder.decreaseRemainingSize(tradedSize);
 
         // 주문 완전체결 처리(잔여금액 or 잔여수량이 0)
-        removeCompletedBuyOrder(waitingOrders, buyOrder);
-        removeCompletedSellOrder(waitingOrders, sellOrder);
+        removeCompletedOrders(waitingOrders, buyOrder, sellOrder);
 
         tradeService.updateOrder(buyOrder);
         tradeService.updateOrder(sellOrder);
@@ -82,14 +78,14 @@ public void executeTrade(WaitingOrders waitingOrders, TradePair<Order, Order> tr
         }
 
         // 지갑 누적계산
-        this.updateWalletAfterTrade(buyOrder, ticker, tradedSize, totalTradedPrice);
-        this.updateWalletAfterTrade(sellOrder, ticker, tradedSize, totalTradedPrice);
+        walletService.updateWalletAfterTrade(buyOrder, ticker, tradedSize, totalTradedPrice);
 
         // 체결내역 저장
-        Trade trade = this.insertNewTrade(ticker, buyOrder, sellOrder, tradedSize, tradedPrice);
+        Trade trade = Trade.of(ticker, LocalDateTime.now(), buyOrder.getUserId(), sellOrder.getUserId(), tradedPrice, tradedSize);
 
         TradeExecutedEvent tradeExecutedEvent = TradeExecutedEvent.of(trade, buyOrder.getId(), sellOrder.getId());
         tradeExecutedEventPublisher.publish(tradeExecutedEvent);
+        return trade;
     }
 
     private static void checkZeroOrderAndThrowException(BuyOrder buyOrder, SellOrder sellOrder) {
@@ -105,35 +101,13 @@ else if (approxEquals(sellOrder.getRemainingSize(), 0.0))
     }
 
     private void increaseAccountCash(Order order, Double amount) {
-        Account account = accountService.findAccountByUserId(order.getUserId()).orElseThrow();
-        accountService.save(account.increaseCash(amount));
-    }
+        int updatedRows = accountService.increaseAccountCash(order.getUserId(), amount);
 
-    private void updateWalletAfterTrade(Order order, String ticker, double tradedSize, double totalTradedPrice) {
-        if (order instanceof BuyOrder) {
-            Wallet buyerWallet = walletService.findWalletByUserIdAndTicker(order.getUserId(), ticker);
-            double updatedBuySize = buyerWallet.getSize() + tradedSize;
-            double currentBuyPrice = buyerWallet.getBuyPrice() == null ? 0.0 : buyerWallet.getBuyPrice();
-            double updatedBuyPrice = ((currentBuyPrice * buyerWallet.getSize()) + totalTradedPrice) / updatedBuySize;
-            buyerWallet.setSize(updatedBuySize);
-            buyerWallet.setBuyPrice(updatedBuyPrice);
-            // TODO : ROI 계산
-            walletService.save(buyerWallet);
-        } else if (order instanceof SellOrder) {
-            // 매도 시에는 평단가 변동 없음
-            Wallet sellerWallet = walletService.findWalletByUserIdAndTicker(order.getUserId(), ticker);
-            walletService.save(sellerWallet);
-        } else {
-            throw new BusinessException("Unsupported order type: " + order.getClass().getName(), ErrorStatus.INTERNAL_SERVER_ERROR);
+        if (updatedRows == 0) {
+            throw new RuntimeException("account updatedRows == 0");
         }
     }
 
-    private Trade insertNewTrade(String ticker, BuyOrder buyOrder, SellOrder sellOrder, double tradeSize, Double tradePrice) {
-        Trade newTrade = Trade.of(ticker, LocalDateTime.now(), buyOrder.getUserId(), sellOrder.getUserId(), tradePrice, tradeSize);
-
-        return tradeService.save(newTrade);
-    }
-
     private static TradeUnitPriceAndSize getTradeUnitPriceAndSize(BuyOrder buyOrder, SellOrder sellOrder) {
         double tradedPrice;
         double tradedSize;
@@ -180,25 +154,32 @@ private static void writeTradingLog(BuyOrder buyOrder, SellOrder sellOrder) {
                 sellOrder.getRemainingSize());
     }
 
-    private static void removeCompletedBuyOrder(WaitingOrders waitingOrders, BuyOrder order) {
-        boolean isOrderCompleted = (isMarketOrder(order) && approxEquals(order.getRemainingDeposit(), 0.0)) ||
-                (isLimitOrder(order) && approxEquals(order.getRemainingSize(), 0.0));
-
-        if (isOrderCompleted) {
-            waitingOrders.removeOrder(order);
-            updateCompletedOrderStatus(order);
-        }
+    private void removeCompletedOrders(WaitingOrders waitingOrders, BuyOrder buyOrder, SellOrder sellOrder) {
+        removeCompletedOrder(waitingOrders, buyOrder);
+        removeCompletedOrder(waitingOrders, sellOrder);
     }
 
-    private static void removeCompletedSellOrder(WaitingOrders waitingOrders, SellOrder order) {
-        boolean isOrderCompleted = approxEquals(order.getRemainingSize(), 0.0);
+    private void removeCompletedOrder(WaitingOrders waitingOrders, Order order) {
+        boolean isOrderCompleted = false;
+
+        if (order instanceof BuyOrder buyOrder) {
+            isOrderCompleted = (isMarketOrder(buyOrder) && approxEquals(buyOrder.getRemainingDeposit(), 0.0)) ||
+                (isLimitOrder(buyOrder) && approxEquals(buyOrder.getRemainingSize(), 0.0));
+        } else if (order instanceof SellOrder sellOrder) {
+            isOrderCompleted = approxEquals(sellOrder.getRemainingSize(), 0.0);
+        }
 
         if (isOrderCompleted) {
             waitingOrders.removeOrder(order);
             updateCompletedOrderStatus(order);
+            publishOrderCompletionEvent(order);
         }
     }
 
+    private void publishOrderCompletionEvent(Order order) {
+        tradeOrderCompletedEventPublisher.publish(TradeOrderCompletedEventImpl.of(order));
+    }
+
     private static void updateCompletedOrderStatus(Order order) {
         order.setState(OrderStatus.DONE);
     }

src/main/java/com/cleanengine/coin/trade/application/TradeExecutor.java

@@ -4,11 +4,17 @@
 import com.cleanengine.coin.order.domain.Order;
 import com.cleanengine.coin.order.domain.spi.WaitingOrders;
 import com.cleanengine.coin.order.domain.spi.WaitingOrdersManager;
+import com.cleanengine.coin.trade.entity.Trade;
+import com.cleanengine.coin.trade.repository.TradeRepository;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Profile;
 import org.springframework.stereotype.Component;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
 
 @Slf4j
 @RequiredArgsConstructor
@@ -18,16 +24,31 @@ public class TradeFlowService {
     private final TradeMatcher tradeMatcher;
     private final TradeExecutor tradeExecutor;
     private final WaitingOrdersManager waitingOrdersManager;
+    private final TradeRepository tradeRepository;
+
+    private CountDownLatch testLatch; // 테스트용 후크
+
+    @Profile("trade-load-test")
+    public void setTestLatch(CountDownLatch latch) {
+        this.testLatch = latch;
+    }
 
     public void execMatchAndTrade(String ticker) {
         WaitingOrders waitingOrders = waitingOrdersManager.getWaitingOrders(ticker);
         // TODO : peek() 해온 Order 객체들을 lock -> 체결 도중 취소 방지
         Optional<TradePair<Order, Order>> tradePair = tradeMatcher.matchOrders(waitingOrders);
         boolean continueProcessing = tradePair.isPresent();
+        List<Trade> tradesToSave = new ArrayList<>();
 
         while (continueProcessing) {
             try {
-                tradeExecutor.executeTrade(waitingOrders, tradePair.get(), ticker);
+                Trade trade = tradeExecutor.executeTrade(waitingOrders, tradePair.get(), ticker);
+                tradesToSave.add(trade);
+                if (tradesToSave.size() > 10000) {
+                    tradeRepository.saveAll(tradesToSave);
+                    tradesToSave.clear();
+                }
+
                 tradePair = tradeMatcher.matchOrders(waitingOrders);
                 continueProcessing = tradePair.isPresent();
             } catch (TradeZeroOrderException e) {
@@ -41,6 +62,15 @@ public void execMatchAndTrade(String ticker) {
                 continueProcessing = false;
             }
         }
+
+        if (!tradesToSave.isEmpty()) {
+            tradeRepository.saveAll(tradesToSave);
+            tradesToSave.clear();
+        }
+
+        if (testLatch != null) {
+            testLatch.countDown();
+        }
     }
 
 }

src/main/java/com/cleanengine/coin/trade/application/TradeFlowService.java

@@ -0,0 +1,9 @@
+package com.cleanengine.coin.trade.application;
+
+import com.cleanengine.coin.order.domain.Order;
+
+public interface TradeOrderCompletedEvent {
+
+    Order getOrder();
+
+}

src/main/java/com/cleanengine/coin/trade/application/TradeOrderCompletedEvent.java

@@ -0,0 +1,22 @@
+package com.cleanengine.coin.trade.application;
+
+import com.cleanengine.coin.order.domain.Order;
+import com.cleanengine.coin.trade.entity.Trade;
+import lombok.Builder;
+import lombok.Getter;
+
+@Getter
+@Builder
+public class TradeOrderCompletedEventImpl implements TradeOrderCompletedEvent {
+
+    Order order;
+
+    private TradeOrderCompletedEventImpl(Order order) {
+        this.order = order;
+    }
+
+    public static TradeOrderCompletedEventImpl of(Order order) {
+        return new TradeOrderCompletedEventImpl(order);
+    }
+
+}

src/main/java/com/cleanengine/coin/trade/application/TradeOrderCompletedEventImpl.java

@@ -0,0 +1,19 @@
+package com.cleanengine.coin.trade.application;
+
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.stereotype.Service;
+
+@Service
+public class TradeOrderCompletedEventPublisher {
+
+    private final ApplicationEventPublisher publisher;
+
+    public TradeOrderCompletedEventPublisher(ApplicationEventPublisher publisher) {
+        this.publisher = publisher;
+    }
+
+    public void publish(TradeOrderCompletedEvent event) {
+        publisher.publishEvent(event);
+    }
+
+}