diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecUploadCashInfo.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecUploadCashInfo.java new file mode 100644 index 000000000..134c7d791 --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecUploadCashInfo.java @@ -0,0 +1,15 @@ +package ru.spcex.clearing.service.execution; + +import java.time.Instant; +import ru.clearing.classes.statics.data.execution.ExecutionCommon; + +public record ExecUploadCashInfo(Long id, Instant createDt) { + + public static ExecUploadCashInfo cash(Long id, Instant createDt) { + return new ExecUploadCashInfo(id, createDt); + } + + public static ExecUploadCashInfo cash(ExecutionCommon exec) { + return new ExecUploadCashInfo(exec.getId(), exec.getCreated()); + } +} diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionCurrencyComponent.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionCurrencyComponent.java index 5382dffac..d49a78a83 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionCurrencyComponent.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionCurrencyComponent.java @@ -6,10 +6,8 @@ import java.time.LocalDate; import java.time.format.DateTimeFormatter; import java.util.Collection; import java.util.HashMap; -import java.util.HashSet; import java.util.Map; import java.util.Optional; -import java.util.Set; import java.util.function.Function; import org.apache.kafka.clients.producer.Producer; import org.slf4j.Logger; @@ -66,7 +64,7 @@ public class ExecutionCurrencyComponent implements IExecutionUploadComponent { private final Function stradesValidator; private final KafkaSender kafkaSender; private static final DateTimeFormatter contractFormatter = DateTimeFormatter.ofPattern("ddMMyy"); - private final Set cash = new HashSet<>(); + private final Map cash = new HashMap<>(); @@ -106,14 +104,16 @@ public class ExecutionCurrencyComponent implements IExecutionUploadComponent { // } public void processNewTS(Long fromId) { - if (fromId == null) { + boolean cleanLoad = fromId == null; + if (cleanLoad) { cash.clear(); + initExecCash(); } LocalDate today = LocalDate.now(); log.debug("Start check new S_TRADE at {}, fromId={}", today, fromId); //выбираем STrades на сегодня с правильным section Collection sTrades; - if (fromId == null) { + if (cleanLoad) { sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of( "tradeDate", LocalDate.now(), "section", Section.CURR.getKey() @@ -135,30 +135,20 @@ public class ExecutionCurrencyComponent implements IExecutionUploadComponent { } //убираем уже добавленные в ExecutionCurrency - sTrades.removeIf(sTrd -> { - Side sTrdSide = IEnumKey.getEnumByKey(Side.class, sTrd.getOperation()); - if (sTrdSide == null || !(sTrdSide.equals(Side.BUY) || sTrdSide.equals(Side.SELL))) { - log.warn("cannot parse strade.tradeNum={} strade.operation {}", sTrd.getTradeNum(), sTrd.getOperation()); - return true; - } -// MoneyFlowSide excDepSide = sTrdSide == Side.BUY ? MoneyFlowSide.BUY : MoneyFlowSide.SELL; - ExecUploadKey excKey = ExecUploadKey.cash(sTrd.getTradeDate(), - sTrd.getTradeNum(), - sTrdSide.getKey()); - if (fromId != null) { - return cash.contains(excKey); - } else { - ExecutionCurrency exec = executionCurrencyImdg.getFirstObjectByFieldValues( - Map.of("tradingDate", sTrd.getTradeDate(), - "exchangeExecutionId", sTrd.getTradeNum(), - "side", sTrdSide.getKey())); - if (exec != null) { - cash.add(excKey); + if (!cleanLoad) { + sTrades.removeIf(sTrd -> { + Side sTrdSide = IEnumKey.getEnumByKey(Side.class, sTrd.getOperation()); + if (sTrdSide == null || !(sTrdSide.equals(Side.BUY) || sTrdSide.equals(Side.SELL))) { + log.warn("cannot parse strade.tradeNum={} strade.operation {}", sTrd.getTradeNum(), sTrd.getOperation()); return true; } - return false; - } - }); +// MoneyFlowSide excDepSide = sTrdSide == Side.BUY ? MoneyFlowSide.BUY : MoneyFlowSide.SELL; + ExecUploadKey excKey = ExecUploadKey.cash(sTrd.getTradeDate(), + sTrd.getTradeNum(), + sTrdSide.getKey()); + return cash.containsKey(excKey); + }); + } log.info("{} strades left after already-added filtering", sTrades.size()); Map execsToInsert = new HashMap<>(); @@ -175,9 +165,17 @@ public class ExecutionCurrencyComponent implements IExecutionUploadComponent { ExecutionCurrency newEC; try { newEC = createExecutionCurrency(sTrd, validator); - newEC.setId(idGen.nextId()); + ExecUploadKey execKey = ExecUploadKey.cash(newEC); + ExecUploadCashInfo storedId; + if (!cleanLoad || (storedId = cash.get(execKey)) == null) { + newEC.setId(idGen.nextId()); + cash.put(execKey, ExecUploadCashInfo.cash(newEC)); + } else { + newEC.setId(storedId.id()); + newEC.setUpdated(newEC.getCreated()); + newEC.setCreated(storedId.createDt()); + } execsToInsert.put(newEC.getId(), newEC); - cash.add(ExecUploadKey.cash(newEC)); //executionCurrencyImdg.insert(newEC); sendNotification(newEC); log.debug("New executionCurrency.id={} was created.", newEC.getId()); diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionDepositComponent.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionDepositComponent.java index 8ff718bb3..8f6ec8641 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionDepositComponent.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionDepositComponent.java @@ -6,10 +6,8 @@ import java.time.LocalDate; import java.time.format.DateTimeFormatter; import java.util.Collection; import java.util.HashMap; -import java.util.HashSet; import java.util.Map; import java.util.Optional; -import java.util.Set; import java.util.function.Function; import org.apache.kafka.clients.producer.Producer; import org.slf4j.Logger; @@ -69,7 +67,7 @@ public class ExecutionDepositComponent implements IExecutionUploadComponent { private final IMessageResolver msgResolver; private final Function stradesValidator; private final KafkaSender kafkaSender; - private final Set cash = new HashSet<>(); + private final Map cash = new HashMap<>(); private static final DateTimeFormatter contractFormatter = DateTimeFormatter.ofPattern("ddMMyy"); @@ -105,13 +103,15 @@ public class ExecutionDepositComponent implements IExecutionUploadComponent { } public void processNewTS(Long fromId) { - if (fromId == null) { + boolean cleanLoad = fromId == null; + if (cleanLoad) { cash.clear(); + initExecCash(); } log.debug("Start check new S_TRADE after {}, fromId={}", tradingDay, fromId); //выбираем STrades на сегодня с правильным section Collection sTrades; - if (fromId == null) { + if (cleanLoad) { sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of( "tradeDate", LocalDate.now(), "section", Section.MKR.getKey() @@ -133,30 +133,20 @@ public class ExecutionDepositComponent implements IExecutionUploadComponent { } //убираем уже добавленные в ExecutionDeposit - sTrades.removeIf(sTrd -> { - Side sTrdSide = IEnumKey.getEnumByKey(Side.class, sTrd.getOperation()); - if (sTrdSide == null || !(sTrdSide.equals(Side.BUY) || sTrdSide.equals(Side.SELL))) { - log.warn("cannot parse strade.tradeNum={} strade.operation {}", sTrd.getTradeNum(), sTrd.getOperation()); - return true; - } - MoneyFlowSide excDepSide = sTrdSide.equals(Side.BUY) ? MoneyFlowSide.BUY : MoneyFlowSide.SELL; - ExecUploadKey excKey = ExecUploadKey.cash(sTrd.getTradeDate(), - sTrd.getTradeNum(), - excDepSide.getKey()); - if (fromId != null) { - return cash.contains(excKey); - } else { - ExecutionDeposit exec = executionDepositImdg.getFirstObjectByFieldValues( - Map.of("tradingDate", sTrd.getTradeDate(), - "exchangeExecutionId", sTrd.getTradeNum(), - "side", excDepSide.getKey())); - if (exec != null) { - cash.add(excKey); + if (!cleanLoad) { + sTrades.removeIf(sTrd -> { + Side sTrdSide = IEnumKey.getEnumByKey(Side.class, sTrd.getOperation()); + if (sTrdSide == null || !(sTrdSide.equals(Side.BUY) || sTrdSide.equals(Side.SELL))) { + log.warn("cannot parse strade.tradeNum={} strade.operation {}", sTrd.getTradeNum(), sTrd.getOperation()); return true; } - return false; - } - }); + MoneyFlowSide excDepSide = sTrdSide.equals(Side.BUY) ? MoneyFlowSide.BUY : MoneyFlowSide.SELL; + ExecUploadKey excKey = ExecUploadKey.cash(sTrd.getTradeDate(), + sTrd.getTradeNum(), + excDepSide.getKey()); + return cash.containsKey(excKey); + }); + } log.info("{} strades left after already-added filtering", sTrades.size()); Map execsToInsert = new HashMap<>(); @@ -173,8 +163,16 @@ public class ExecutionDepositComponent implements IExecutionUploadComponent { ExecutionDeposit newED; try { newED = createExecutionDeposit(sTrd, validator); - newED.setId(idGen.nextId()); - cash.add(ExecUploadKey.cash(newED)); + ExecUploadKey execKey = ExecUploadKey.cash(newED); + ExecUploadCashInfo storedId; + if (!cleanLoad || (storedId = cash.get(execKey)) == null) { + newED.setId(idGen.nextId()); + cash.put(execKey, ExecUploadCashInfo.cash(newED)); + } else { + newED.setId(storedId.id()); + newED.setUpdated(newED.getCreated()); + newED.setCreated(storedId.createDt()); + } execsToInsert.put(newED.getId(), newED); //executionDepositImdg.insert(newED); sendNotification(newED); diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionFondComponent.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionFondComponent.java index 6b7ab1527..69af95c4c 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionFondComponent.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionFondComponent.java @@ -6,10 +6,8 @@ import java.time.Instant; import java.time.LocalDate; import java.util.Collection; import java.util.HashMap; -import java.util.HashSet; import java.util.Map; import java.util.Optional; -import java.util.Set; import java.util.function.Function; import org.apache.kafka.clients.producer.Producer; import org.slf4j.Logger; @@ -84,7 +82,7 @@ public class ExecutionFondComponent implements IExecutionUploadComponent { private final KafkaSender kafkaSender; private final NotificationSender notifications; private final boolean valuation; - private final Set cash = new HashSet<>(); + private final Map cash = new HashMap<>(); @Autowired @@ -124,13 +122,15 @@ public class ExecutionFondComponent implements IExecutionUploadComponent { } public void processNewTS(Long fromId) { - if (fromId == null) { + boolean cleanLoad = fromId == null; + if (cleanLoad) { cash.clear(); + initExecCash(); } log.debug("Start check new S_TRADE after {}, fromId={}", tradingDay, fromId); //выбираем STrades на сегодня с правильным section Collection sTrades; - if (fromId == null) { + if (cleanLoad) { sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of( "tradeDate", LocalDate.now(), "section", Section.FOND.getKey() @@ -152,28 +152,20 @@ public class ExecutionFondComponent implements IExecutionUploadComponent { } //убираем уже добавленные в ExecutionDeposit - sTrades.removeIf(sTrd -> { - Side sTrdSide = IEnumKey.getEnumByKey(Side.class, sTrd.getOperation()); - if (sTrdSide == null || !(sTrdSide.equals(Side.BUY) || sTrdSide.equals(Side.SELL))) { - log.warn("cannot parse strade.tradeNum={} strade.operation {}", sTrd.getTradeNum(), sTrd.getOperation()); - return true; - } - Side excDepSide = sTrdSide.equals(Side.BUY) ? Side.BUY : Side.SELL; - ExecUploadKey excKey = ExecUploadKey.cash(sTrd.getTradeDate(), sTrd.getTradeNum(), excDepSide.getKey()); - if (fromId != null) { - return cash.contains(excKey); - } else { - ExecutionFond exec = executionFondImdg.getFirstObjectByFieldValues( - Map.of("tradingDate", sTrd.getTradeDate(), - "exchangeExecutionId", sTrd.getTradeNum(), - "side", excDepSide.getKey())); - if (exec != null) { - cash.add(excKey); + if (!cleanLoad) { + sTrades.removeIf(sTrd -> { + Side sTrdSide = IEnumKey.getEnumByKey(Side.class, sTrd.getOperation()); + if (sTrdSide == null || !(sTrdSide.equals(Side.BUY) || sTrdSide.equals(Side.SELL))) { + log.warn("cannot parse strade.tradeNum={} strade.operation {}", sTrd.getTradeNum(), sTrd.getOperation()); return true; } - return false; - } - }); + Side excDepSide = sTrdSide.equals(Side.BUY) ? Side.BUY : Side.SELL; + ExecUploadKey excKey = ExecUploadKey.cash(sTrd.getTradeDate(), + sTrd.getTradeNum(), + excDepSide.getKey()); + return cash.containsKey(excKey); + }); + } log.info("{} strades left after already-added filtering", sTrades.size()); Map execsToInsert = new HashMap<>(); @@ -190,8 +182,16 @@ public class ExecutionFondComponent implements IExecutionUploadComponent { ExecutionFond newED; try { newED = createExecutionFond(sTrd, validator); - newED.setId(idGen.nextId()); - cash.add(ExecUploadKey.cash(newED)); + ExecUploadKey execKey = ExecUploadKey.cash(newED); + ExecUploadCashInfo storedId; + if (!cleanLoad || (storedId = cash.get(execKey)) == null) { + newED.setId(idGen.nextId()); + cash.put(execKey, ExecUploadCashInfo.cash(newED)); + } else { + newED.setId(storedId.id()); + newED.setUpdated(newED.getCreated()); + newED.setCreated(storedId.createDt()); + } execsToInsert.put(newED.getId(), newED); //executionFondImdg.insert(newED); sendNotification(newED); diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionUploadCashUtil.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionUploadCashUtil.java index 08e20a377..06b7b3073 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionUploadCashUtil.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionUploadCashUtil.java @@ -2,20 +2,19 @@ package ru.spcex.clearing.service.execution; import java.time.LocalDate; import java.util.Collection; -import java.util.Set; +import java.util.Map; import ru.clearing.classes.statics.data.execution.ExecutionCommon; import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder; public class ExecutionUploadCashUtil { public static - void loadExecutions(Imdg executionImdg, Set cash) { + void loadExecutions(Imdg executionImdg, Map cash) { LocalDate today = LocalDate.now(); ImdgPredicateBuilder pb = executionImdg.predicateBuilder(); Collection execs = executionImdg.getCollectionObjectsByPredicate( pb.equals("tradingDate", today) ); - execs.forEach(e -> cash.add(ExecUploadKey.cash(e))); + execs.forEach(e -> cash.put(ExecUploadKey.cash(e), ExecUploadCashInfo.cash(e))); } - }