diff --git a/ranking/src/main/java/youthfi/ranking/config/TopologyConfig.java b/ranking/src/main/java/youthfi/ranking/config/TopologyConfig.java index 081bef5..46ca230 100644 --- a/ranking/src/main/java/youthfi/ranking/config/TopologyConfig.java +++ b/ranking/src/main/java/youthfi/ranking/config/TopologyConfig.java @@ -4,18 +4,15 @@ import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.utils.Bytes; -import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.kstream.*; -import org.apache.kafka.streams.state.KeyValueStore; -import org.apache.kafka.streams.state.Stores; +import org.apache.kafka.streams.state.*; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import youthfi.ranking.model.ExecutionRow; -import youthfi.ranking.model.PortfolioAgg; import youthfi.ranking.model.RankItem; -import youthfi.ranking.model.UserStockRow; +import youthfi.ranking.transformer.RealizedRateFifoTransformer; import youthfi.ranking.transformer.TopNTransformer; import youthfi.ranking.util.DebeziumParser; @@ -27,94 +24,53 @@ public class TopologyConfig { private static final ObjectMapper M = new ObjectMapper(); - @Value("${topics.userstock}") String userStockTopic; @Value("${topics.execution}") String executionTopic; @Value("${topics.out}") String outTopic; @Bean public KStream kStream(StreamsBuilder builder) { - // 1) userstock: KTable (key = userId|stockId) - KTable userStock = builder - .stream(userStockTopic, Consumed.with(Serdes.String(), Serdes.String())) - .mapValues(DebeziumParser::parseUserStock) - .filter((k,v) -> v != null) - .selectKey((k,v) -> v.getUserId() + "|" + v.getStockId()) - .toTable(Materialized.>as("userstock-table") - .withKeySerde(Serdes.String()) - .withValueSerde(userStockSerde())); - - // 2) execution: KTable (key = userId|stockId, value = ExecutionRow) - KTable executionTable = builder + // 0) execution 파싱 (키 = userId|stockId) + KStream execStream = builder .stream(executionTopic, Consumed.with(Serdes.String(), Serdes.String())) .mapValues(DebeziumParser::parseExecution) .filter((k,v) -> v != null) - .selectKey((k,v) -> v.getUserId() + "|" + v.getStockId()) - .toTable(Materialized.>as("execution-table") - .withKeySerde(Serdes.String()) - .withValueSerde(executionSerde())); - - // 3) KTable-KTable 조인: userstock과 execution을 userId|stockId로 조인 - KTable positions = userStock.leftJoin( - executionTable, - // 조인 결과: row(UserStockRow) + exec(ExecutionRow) - (row, exec) -> { - double currentPrice = (exec == null) ? row.getAvgPrice() : exec.getPrice(); - double invested = row.getAvgPrice() * row.getHoldingQuantity(); - double current = currentPrice * row.getHoldingQuantity(); - - // 디버깅 로그 - System.out.println("DEBUG: userId=" + row.getUserId() + - ", stockId=" + row.getStockId() + - ", avgPrice=" + row.getAvgPrice() + - ", currentPrice=" + currentPrice + - ", holdingQuantity=" + row.getHoldingQuantity() + - ", invested=" + invested + - ", current=" + current + - ", profitRate=" + ((current - invested) / invested * 100.0)); - - return new PortfolioAgg(invested, current); - } - ); - - // 4) userId 단위로 합산 (KTable.reduce: add / subtract 둘 다 필요) - Serde aggSerde = portfolioSerde(); - KTable userAgg = positions - .groupBy( - (userIdStockId, agg) -> KeyValue.pair(userIdStockId.split("\\|", 2)[0], agg), - Grouped.with(Serdes.String(), aggSerde) - ) - .reduce( - // add - (oldV, newV) -> new PortfolioAgg( - (oldV==null?0:oldV.getInvested()) + newV.getInvested(), - (oldV==null?0:oldV.getCurrent()) + newV.getCurrent() - ), - // subtract (해당 키의 이전값이 빠질 때 역연산) - (oldV, newV) -> new PortfolioAgg( - (oldV==null?0:oldV.getInvested()) - newV.getInvested(), - (oldV==null?0:oldV.getCurrent()) - newV.getCurrent() - ), - Materialized.>as("user-agg-store") + .selectKey((k,v) -> v.getUserId() + "|" + v.getStockId()); + + // 1) BUY 롯 상태 스토어 (키=userId|stockId, 값=Deque 직렬화 바이트) + var lotsStoreBuilder = Stores.keyValueStoreBuilder( + Stores.persistentKeyValueStore("buy-lots"), + Serdes.String(), Serdes.ByteArray()); + builder.addStateStore(lotsStoreBuilder); + + // 2) FIFO 처리 → SELL 시 실현 수익률(Double) 방출 + KStream realizedRatePerTrade = + execStream.transformValues( + () -> new RealizedRateFifoTransformer("buy-lots"), + "buy-lots" + ).filter((k,v) -> v != null); + + // 3) 유저별 최신 실현 수익률 유지 + KTable userLatestRate = realizedRatePerTrade + .selectKey((userStockKey, rate) -> userStockKey.split("\\|", 2)[0]) // userId + .groupByKey(Grouped.with(Serdes.String(), Serdes.Double())) + .reduce((oldV, newV) -> newV, + Materialized.>as("user-latest-realized-rate") .withKeySerde(Serdes.String()) - .withValueSerde(aggSerde) + .withValueSerde(Serdes.Double()) ); - // 5) 수익률 계산 스트림 - KStream userProfitRate = userAgg - .toStream() - .mapValues(PortfolioAgg::profitRatePct); - - // 6) Top10 계산 (StateStore) - var storeBuilder = Stores.keyValueStoreBuilder( + // 4) Top10 계산 (StateStore) + var topStoreBuilder = Stores.keyValueStoreBuilder( Stores.persistentKeyValueStore("top10-store"), Serdes.String(), Serdes.Double()); - builder.addStateStore(storeBuilder); + builder.addStateStore(topStoreBuilder); - KStream> top10 = userProfitRate + KStream> top10 = userLatestRate + .toStream() .transformValues(() -> new TopNTransformer("top10-store"), "top10-store"); - // 7) JSON 직렬화 후 단일 키로 발행 (예외 처리 포함) + // 5) JSON 직렬화 후 발행 top10 .mapValues(list -> { try { return M.writeValueAsString(list); } @@ -126,54 +82,13 @@ public KStream kStream(StreamsBuilder builder) { return builder.stream(outTopic, Consumed.with(Serdes.String(), Serdes.String())); } - // ---- Serde helpers (Lombok 클래스 기준: getter 사용) ---- - private Serde userStockSerde() { - var ser = new org.apache.kafka.common.serialization.Serializer() { - @Override public byte[] serialize(String topic, UserStockRow d) { - if (d == null) return null; - try { - return M.writeValueAsBytes(d); - } catch (Exception e) { - throw new RuntimeException(e); - } - } - }; - var de = new org.apache.kafka.common.serialization.Deserializer() { - @Override public UserStockRow deserialize(String topic, byte[] bytes) { - if (bytes == null) return null; - try { - return M.readValue(bytes, UserStockRow.class); - } catch (Exception e) { - throw new RuntimeException(e); - } - } - }; - return Serdes.serdeFrom(ser, de); - } - - private Serde portfolioSerde() { - var ser = new org.apache.kafka.common.serialization.Serializer() { - @Override public byte[] serialize(String topic, PortfolioAgg d) { - if (d == null) return null; - String s = d.getInvested() + "," + d.getCurrent(); - return s.getBytes(StandardCharsets.UTF_8); - } - }; - var de = new org.apache.kafka.common.serialization.Deserializer() { - @Override public PortfolioAgg deserialize(String topic, byte[] bytes) { - if (bytes == null) return null; - String[] p = new String(bytes, StandardCharsets.UTF_8).split(","); - return new PortfolioAgg(Double.parseDouble(p[0]), Double.parseDouble(p[1])); - } - }; - return Serdes.serdeFrom(ser, de); - } - + // ---- Serde: ExecutionRow ---- private Serde executionSerde() { var ser = new org.apache.kafka.common.serialization.Serializer() { @Override public byte[] serialize(String topic, ExecutionRow d) { if (d == null) return null; - String s = d.getUserId() + "," + d.getStockId() + "," + d.getPrice(); + String s = d.getUserId() + "," + d.getStockId() + "," + d.getPrice() + + "," + d.getIsBuy() + "," + d.getQuantity() + "," + d.getTsMs(); return s.getBytes(StandardCharsets.UTF_8); } }; @@ -181,9 +96,16 @@ private Serde executionSerde() { @Override public ExecutionRow deserialize(String topic, byte[] bytes) { if (bytes == null) return null; String[] p = new String(bytes, StandardCharsets.UTF_8).split(","); - return new ExecutionRow(p[0], p[1], Double.parseDouble(p[2])); + return new ExecutionRow( + p[0], // userId + p[1], // stockId + Double.parseDouble(p[2]), // price + Integer.parseInt(p[3]), // isBuy + Long.parseLong(p[4]), // quantity + (p.length >= 6 ? Long.parseLong(p[5]) : System.currentTimeMillis()) // tsMs + ); } }; return Serdes.serdeFrom(ser, de); } -} \ No newline at end of file +} diff --git a/ranking/src/main/java/youthfi/ranking/model/BuyLot.java b/ranking/src/main/java/youthfi/ranking/model/BuyLot.java new file mode 100644 index 0000000..d2a56ff --- /dev/null +++ b/ranking/src/main/java/youthfi/ranking/model/BuyLot.java @@ -0,0 +1,10 @@ +package youthfi.ranking.model; + +import lombok.*; + +@Getter @Setter @NoArgsConstructor @AllArgsConstructor +public class BuyLot { + private double price; // 매수가 + private long qty; // 남은 수량 + private long tsMs; // 매수 시각 +} diff --git a/ranking/src/main/java/youthfi/ranking/model/ExecutionRow.java b/ranking/src/main/java/youthfi/ranking/model/ExecutionRow.java index 939f137..aaf04ab 100644 --- a/ranking/src/main/java/youthfi/ranking/model/ExecutionRow.java +++ b/ranking/src/main/java/youthfi/ranking/model/ExecutionRow.java @@ -20,4 +20,14 @@ public class ExecutionRow { @JsonProperty("price") private double price; -} \ No newline at end of file + + @JsonProperty("is_buy") + private int isBuy; + + @JsonProperty("quantity") + private long quantity; + + @JsonProperty("ts_ms") + private long tsMs; + +} diff --git a/ranking/src/main/java/youthfi/ranking/model/PortfolioAgg.java b/ranking/src/main/java/youthfi/ranking/model/PortfolioAgg.java deleted file mode 100644 index f517823..0000000 --- a/ranking/src/main/java/youthfi/ranking/model/PortfolioAgg.java +++ /dev/null @@ -1,28 +0,0 @@ -package youthfi.ranking.model; - -import lombok.AllArgsConstructor; -import lombok.Getter; -import lombok.NoArgsConstructor; -import lombok.Setter; - -@Getter @Setter @NoArgsConstructor @AllArgsConstructor -public class PortfolioAgg { - private double invested; - private double current; - - public static PortfolioAgg zero() { - return new PortfolioAgg(0, 0); - } - - public PortfolioAgg add(PortfolioAgg o) { - return new PortfolioAgg(invested + o.invested, current + o.current); - } - - public PortfolioAgg sub(PortfolioAgg o) { - return new PortfolioAgg(invested - o.invested, current - o.current); - } - - public double profitRatePct() { - return invested <= 0 ? 0.0 : (current - invested) / invested * 100.0; - } -} diff --git a/ranking/src/main/java/youthfi/ranking/model/UserStockRow.java b/ranking/src/main/java/youthfi/ranking/model/UserStockRow.java deleted file mode 100644 index acea636..0000000 --- a/ranking/src/main/java/youthfi/ranking/model/UserStockRow.java +++ /dev/null @@ -1,27 +0,0 @@ -package youthfi.ranking.model; - -import com.fasterxml.jackson.annotation.JsonIgnoreProperties; -import com.fasterxml.jackson.annotation.JsonProperty; -import lombok.AllArgsConstructor; -import lombok.Getter; -import lombok.NoArgsConstructor; -import lombok.Setter; - -@Getter -@Setter -@NoArgsConstructor -@AllArgsConstructor -@JsonIgnoreProperties(ignoreUnknown = true) -public class UserStockRow { - @JsonProperty("user_id") - private String userId; - - @JsonProperty("stock_id") - private String stockId; - - @JsonProperty("holding_quantity") - private int holdingQuantity; - - @JsonProperty("avg_price") - private double avgPrice; -} \ No newline at end of file diff --git a/ranking/src/main/java/youthfi/ranking/transformer/RealizedRateFifoTransformer.java b/ranking/src/main/java/youthfi/ranking/transformer/RealizedRateFifoTransformer.java new file mode 100644 index 0000000..0b79b39 --- /dev/null +++ b/ranking/src/main/java/youthfi/ranking/transformer/RealizedRateFifoTransformer.java @@ -0,0 +1,91 @@ +package youthfi.ranking.transformer; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.kafka.streams.kstream.ValueTransformerWithKey; +import org.apache.kafka.streams.processor.ProcessorContext; +import org.apache.kafka.streams.state.KeyValueStore; +import youthfi.ranking.model.BuyLot; +import youthfi.ranking.model.ExecutionRow; + +import java.util.ArrayDeque; +import java.util.Deque; + +public class RealizedRateFifoTransformer implements ValueTransformerWithKey { + private static final ObjectMapper M = new ObjectMapper(); + + private final String storeName; + private KeyValueStore store; + + public RealizedRateFifoTransformer(String storeName) { + this.storeName = storeName; + } + + @Override @SuppressWarnings("unchecked") + public void init(ProcessorContext context) { + this.store = (KeyValueStore) context.getStateStore(storeName); + } + + @Override + public Double transform(String key, ExecutionRow e) { + if (e == null || key == null) return null; + + Deque q = load(key); + + if (e.getIsBuy() == 1) { + if (e.getQuantity() > 0 && e.getPrice() > 0) { + q.addLast(new BuyLot(e.getPrice(), e.getQuantity(), e.getTsMs())); + save(key, q); + } + return null; // 매수만 있을 땐 수익률 없음 + } + + //매도 했으면 + if (e.getIsBuy() == 0) { + long remain = e.getQuantity(); + if (remain <= 0) return null; + + double pnl = 0.0; + double cost = 0.0; + while (remain > 0 && !q.isEmpty()) { + BuyLot b = q.peekFirst(); + long used = Math.min(remain, b.getQty()); + pnl += (e.getPrice() - b.getPrice()) * used; + cost += b.getPrice() * used; + + b.setQty(b.getQty() - used); + remain -= used; + if (b.getQty() == 0) q.removeFirst(); + } + + // 초과 매도(remain>0)는 무시 + save(key, q); + + if (cost <= 0.0) return 0.0; + return (pnl / cost) * 100.0; // 가중평균 실현 수익률 + } + + return null; + } + + @Override public void close() {} + + // ---- 상태 직렬화/역직렬화(JSON) ---- + private Deque load(String key) { + try { + byte[] bytes = store.get(key); + if (bytes == null) return new ArrayDeque<>(); + BuyLot[] arr = M.readValue(bytes, BuyLot[].class); + Deque q = new ArrayDeque<>(); + for (BuyLot b : arr) q.addLast(b); + return q; + } catch (Exception ex) { + return new ArrayDeque<>(); + } + } + private void save(String key, Deque q) { + try { + BuyLot[] arr = q.toArray(new BuyLot[0]); + store.put(key, M.writeValueAsBytes(arr)); + } catch (Exception ignored) {} + } +} diff --git a/ranking/src/main/java/youthfi/ranking/util/DebeziumParser.java b/ranking/src/main/java/youthfi/ranking/util/DebeziumParser.java index 38040e0..1022256 100644 --- a/ranking/src/main/java/youthfi/ranking/util/DebeziumParser.java +++ b/ranking/src/main/java/youthfi/ranking/util/DebeziumParser.java @@ -1,32 +1,63 @@ package youthfi.ranking.util; -import youthfi.ranking.model.ExecutionRow; -import youthfi.ranking.model.UserStockRow; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import youthfi.ranking.model.ExecutionRow; + +import java.math.BigDecimal; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.time.format.DateTimeFormatter; + public final class DebeziumParser { private static final ObjectMapper M = new ObjectMapper(); + private static final DateTimeFormatter DT_MICRO = + DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss[.n]"); - public static UserStockRow parseUserStock(String v){ - try { - JsonNode after = M.readTree(v).path("payload").path("after"); - if (after.isMissingNode() || after.isNull()) return null; - String userId = after.path("user_id").asText(); - String stockId = after.path("stock_id").asText(); - int qty = after.path("holding_quantity").asInt(); - double avg = after.path("avg_price").asDouble(); - return new UserStockRow(userId, stockId, qty, avg); - } catch (Exception e) { return null; } - } + private DebeziumParser() {} public static ExecutionRow parseExecution(String v){ try { - JsonNode after = M.readTree(v).path("payload").path("after"); + JsonNode root = M.readTree(v); + JsonNode payload = root.path("payload"); + if (payload.isMissingNode()) return null; + + String op = payload.path("op").asText(""); + if ("d".equals(op)) return null; // 삭제 이벤트 무시 + + long fallbackTs = payload.has("ts_ms") ? payload.path("ts_ms").asLong() : System.currentTimeMillis(); + JsonNode after = payload.path("after"); if (after.isMissingNode() || after.isNull()) return null; - String userId = after.path("user_id").asText(); - String stockId = after.path("stock_id").asText(); - double price = after.path("price").asDouble(); - return new ExecutionRow(userId, stockId, price); - } catch (Exception e) { return null; } + + String userId = after.path("user_id").asText(null); + String stockId = after.path("stock_id").asText(null); + if (userId == null || userId.isBlank() || stockId == null || stockId.isBlank()) return null; + + // price decimal 안전 파싱 + BigDecimal priceDec = after.path("price").isNumber() + ? after.path("price").decimalValue() + : new BigDecimal(after.path("price").asText("0")); + double price = priceDec.doubleValue(); + if (price <= 0) return null; + + int isBuy = after.path("is_buy").asInt(); + + long qty = after.path("quantity").isNumber() + ? after.path("quantity").longValue() + : Long.parseLong(after.path("quantity").asText("0")); + if (qty <= 0) return null; + + long tsMs = fallbackTs; + String tradeAtStr = after.path("trade_at").asText(null); + if (tradeAtStr != null && !tradeAtStr.isBlank()) { + LocalDateTime ldt = LocalDateTime.parse(tradeAtStr, DT_MICRO); + // 필요 시 Asia/Seoul로 바꿔도 됨. 여기선 UTC 기준. + tsMs = ldt.toInstant(ZoneOffset.UTC).toEpochMilli(); + } + + return new ExecutionRow(userId, stockId, price, isBuy, qty, tsMs); + } catch (Exception e) { + return null; + } } -} \ No newline at end of file +}