From 21cd02b20a067251fbb0b7b959275139d9a6cdba Mon Sep 17 00:00:00 2001 From: ialbert Date: Tue, 6 Aug 2024 19:45:50 +0300 Subject: [PATCH] execution uploading optimization --- .../clearing/service/EventsReceiver.java | 14 ++++++- .../service/execution/ExecUploadKey.java | 37 +++++++++++++++++++ .../execution/ExecutionCurrencyComponent.java | 34 ++++++++++++++--- .../execution/ExecutionDepositComponent.java | 32 +++++++++++++--- .../execution/ExecutionFondComponent.java | 32 +++++++++++++--- .../execution/ExecutionUploadCashUtil.java | 21 +++++++++++ .../execution/IExecutionUploadComponent.java | 5 +++ 7 files changed, 156 insertions(+), 19 deletions(-) create mode 100644 clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecUploadKey.java create mode 100644 clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionUploadCashUtil.java create mode 100644 clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/IExecutionUploadComponent.java diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java index 048997fa2..a947ca486 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java @@ -31,6 +31,9 @@ import ru.spcex.clearing.platform.messaging.domain.cud.utilities.STradesImported import ru.spcex.clearing.platform.messaging.service.QueueConsumerV2; import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate; import ru.spcex.clearing.platform.messaging.service.Status; +import ru.spcex.clearing.service.execution.ExecutionCurrencyComponent; +import ru.spcex.clearing.service.execution.ExecutionDepositComponent; +import ru.spcex.clearing.service.execution.ExecutionFondComponent; import ru.spcex.clearing.service.executors.Sdf06Executor; import ru.spcex.clearing.service.executors.Sdf10Executor; import ru.spcex.clearing.service.payment.PaymentInstructionOutboundService; @@ -76,6 +79,9 @@ public class EventsReceiver extends QueueConsumerV2 implements InitializingBean private final StatementServiceV2 statementService; private final SessionTerminator sessionTerminator; private final PaymentInstructionOutboundService pmtOutboundService; + private final ExecutionCurrencyComponent execCurrUpload; + private final ExecutionDepositComponent execDepUpload; + private final ExecutionFondComponent execFondUpload; @Autowired public EventsReceiver(@Qualifier("kafkaConsumer") Consumer kafkaQueue, Producer kafkaResponseQueue, @@ -87,7 +93,7 @@ public class EventsReceiver extends QueueConsumerV2 implements InitializingBean CurrencySession currencySession, PrimaryAuctionB0Session primaryAuctionB0Session, PrimaryAuctionT0Session primaryAuctionT0Session, IntermediateMkrSession intermediateMkrSession, FinalMkrSession finalMkrSession, UnitedSession unitedSession, ReturnDepositSession returnDepositSession, SessionManager sessionManager, Sdf06Executor sdf06Executor, - Sdf10Executor sdf10Executor, BalanceRevise balanceRevise, Sdf05Sender sdf05Sender, StatementServiceV2 statementService, SessionTerminator sessionTerminator, PaymentInstructionOutboundService pmtOutboundService) { + Sdf10Executor sdf10Executor, BalanceRevise balanceRevise, Sdf05Sender sdf05Sender, StatementServiceV2 statementService, SessionTerminator sessionTerminator, PaymentInstructionOutboundService pmtOutboundService, ExecutionCurrencyComponent execCurrUpload, ExecutionDepositComponent execDepUpload, ExecutionFondComponent execFondUpload) { super(kafkaQueue, kafkaResponseQueue); this.environment = environment; this.errorResolver = errorResolver; @@ -110,10 +116,16 @@ public class EventsReceiver extends QueueConsumerV2 implements InitializingBean this.statementService = statementService; this.sessionTerminator = sessionTerminator; this.pmtOutboundService = pmtOutboundService; + this.execCurrUpload = execCurrUpload; + this.execDepUpload = execDepUpload; + this.execFondUpload = execFondUpload; } @Override public void afterPropertiesSet() { + execCurrUpload.initExecCash(); + execDepUpload.initExecCash(); + execFondUpload.initExecCash(); callback(Sdf04Request.class) .setConsumer(event -> { Sdf04Request requestPayload = event.getRequestPayload(); diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecUploadKey.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecUploadKey.java new file mode 100644 index 000000000..678dbb3ea --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecUploadKey.java @@ -0,0 +1,37 @@ +package ru.spcex.clearing.service.execution; + +import java.time.LocalDate; +import java.util.Objects; +import ru.clearing.classes.statics.data.execution.ExecutionCommon; + +public record ExecUploadKey(LocalDate tradeDate, Long exchangeExecutionId, String side) { + + public static ExecUploadKey cash(LocalDate tradingDate, Long exchangeExecutionId, String side) { + return new ExecUploadKey(tradingDate, exchangeExecutionId, side); + } + + public static ExecUploadKey cash(ExecutionCommon exec) { + return new ExecUploadKey(exec.getTradingDate(), exec.getExchangeExecutionId(), exec.getSide()); + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + ExecUploadKey that = (ExecUploadKey) o; + + if (!Objects.equals(tradeDate, that.tradeDate)) return false; + if (!Objects.equals(exchangeExecutionId, that.exchangeExecutionId)) + return false; + return Objects.equals(side, that.side); + } + + @Override + public int hashCode() { + int result = tradeDate != null ? tradeDate.hashCode() : 0; + result = 31 * result + (exchangeExecutionId != null ? exchangeExecutionId.hashCode() : 0); + result = 31 * result + (side != null ? side.hashCode() : 0); + return result; + } +} 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 53380373a..cdaf242b1 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 @@ -5,8 +5,11 @@ import java.time.Instant; 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; @@ -32,6 +35,7 @@ import ru.spcex.clearing.service.validation.ValidationStored; import ru.spcex.platform.enumeration.Section; import ru.spcex.platform.enumeration.Side; import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.api.ImdgId; import ru.spcex.platform.imdg.api.ImdgProvider; import ru.spcex.platform.imdg.api.predicate.ImdgPredicate; import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder; @@ -48,11 +52,12 @@ import ru.spcex.platform.utils.validation.IValidator; */ @Component @EnableScheduling -public class ExecutionCurrencyComponent { +public class ExecutionCurrencyComponent implements IExecutionUploadComponent { private final Logger log = LoggerFactory.getLogger(getClass()); private final Imdg sTradeImdg; private final Imdg executionCurrencyImdg; + private final ImdgId idGen; private final Imdg listingImdg; // protected transient Long tradeNum; //todo used? @@ -61,6 +66,7 @@ public class ExecutionCurrencyComponent { private final Function stradesValidator; private final KafkaSender kafkaSender; private static final DateTimeFormatter contractFormatter = DateTimeFormatter.ofPattern("ddMMyy"); + private final Set cash = new HashSet<>(); @@ -77,6 +83,12 @@ public class ExecutionCurrencyComponent { this.msgResolver = msgResolver; // resetTradingDay(); + this.idGen = imdgProvider.getImdgIdGenerator(); + } + + @Override + public void initExecCash() { + ExecutionUploadCashUtil.loadExecutions(executionCurrencyImdg, cash); } // /** @@ -127,13 +139,19 @@ public class ExecutionCurrencyComponent { return true; } // MoneyFlowSide excDepSide = sTrdSide == Side.BUY ? MoneyFlowSide.BUY : MoneyFlowSide.SELL; - return executionCurrencyImdg.getFirstObjectByFieldValues( - Map.of("tradingDate", sTrd.getTradeDate(), - "exchangeExecutionId", sTrd.getTradeNum(), - "side", sTrdSide.getKey())) != null; + ExecUploadKey execCash = ExecUploadKey.cash(sTrd.getTradeDate(), + sTrd.getTradeNum(), + sTrdSide.getKey()); + return cash.contains(execCash); +// return executionCurrencyImdg.getFirstObjectByFieldValues( +// Map.of("tradingDate", sTrd.getTradeDate(), +// "exchangeExecutionId", sTrd.getTradeNum(), +// "side", sTrdSide.getKey())) != null; + }); log.info("{} strades left after already-added filtering", sTrades.size()); + Map execsToInsert = new HashMap<>(); for (STrades sTrd : sTrades) { log.trace("S_TRADE[{}] new", sTrd.getId()); @@ -147,7 +165,10 @@ public class ExecutionCurrencyComponent { ExecutionCurrency newEC; try { newEC = createExecutionCurrency(sTrd, validator); - executionCurrencyImdg.insert(newEC); + newEC.setId(idGen.nextId()); + execsToInsert.put(newEC.getId(), newEC); + cash.add(ExecUploadKey.cash(newEC)); + //executionCurrencyImdg.insert(newEC); sendNotification(newEC); log.debug("New executionCurrency.id={} was created.", newEC.getId()); } catch (ClearingException ce) { @@ -156,6 +177,7 @@ public class ExecutionCurrencyComponent { log.error("When create new ExecutionCurrency by STrade[{}] error: {}", sTrd.getId(), ExceptionUtils.getStackTrace(e)); } } + executionCurrencyImdg.putAll(execsToInsert, 200); Long newMaxTradeNum = sTrades.stream().mapToLong(STrades::getTradeNum).max().orElse(0); // orElseGet(() -> tradeNum) log.info("Process completed. Next tradeNum is {}", newMaxTradeNum); 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 337b1db81..15621f243 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 @@ -5,8 +5,11 @@ import java.time.Instant; 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; @@ -34,6 +37,7 @@ import ru.spcex.platform.enumeration.MoneyFlowSide; import ru.spcex.platform.enumeration.Section; import ru.spcex.platform.enumeration.Side; import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.api.ImdgId; import ru.spcex.platform.imdg.api.ImdgProvider; import ru.spcex.platform.imdg.api.predicate.ImdgPredicate; import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder; @@ -51,12 +55,13 @@ import ru.spcex.platform.utils.validation.IValidator; */ @Component @EnableScheduling -public class ExecutionDepositComponent { +public class ExecutionDepositComponent implements IExecutionUploadComponent { private final Logger log = LoggerFactory.getLogger(getClass()); private final Imdg sTradeImdg; private final Imdg executionDepositImdg; private final Imdg listingImdg; + private final ImdgId idGen; //fixme ждать ТЗ Long tradeNum; @@ -64,6 +69,7 @@ public class ExecutionDepositComponent { private final IMessageResolver msgResolver; private final Function stradesValidator; private final KafkaSender kafkaSender; + private final Set cash = new HashSet<>(); private static final DateTimeFormatter contractFormatter = DateTimeFormatter.ofPattern("ddMMyy"); @@ -76,6 +82,7 @@ public class ExecutionDepositComponent { this.sTradeImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_STrades, STrades.class); this.executionDepositImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_ExecutionDeposit, ExecutionDeposit.class); this.listingImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Listing, Listing.class); + this.idGen = imdgProvider.getImdgIdGenerator(); this.stradesValidator = stradesValidator; this.kafkaSender = kafkaSender; this.msgResolver = msgResolver; @@ -130,13 +137,17 @@ public class ExecutionDepositComponent { return true; } MoneyFlowSide excDepSide = sTrdSide.equals(Side.BUY) ? MoneyFlowSide.BUY : MoneyFlowSide.SELL; - return executionDepositImdg.getFirstObjectByFieldValues( - Map.of("tradingDate", sTrd.getTradeDate(), - "exchangeExecutionId", sTrd.getTradeNum(), - "side", excDepSide.getKey())) != null; + return cash.contains( + ExecUploadKey.cash(sTrd.getTradeDate(), sTrd.getTradeNum(), excDepSide.getKey()) + ); + //return executionDepositImdg.getFirstObjectByFieldValues( + // Map.of("tradingDate", sTrd.getTradeDate(), + // "exchangeExecutionId", sTrd.getTradeNum(), + // "side", excDepSide.getKey())) != null; }); log.info("{} strades left after already-added filtering", sTrades.size()); + Map execsToInsert = new HashMap<>(); for (STrades sTrd : sTrades) { log.trace("S_TRADE[{}] new", sTrd.getId()); @@ -150,7 +161,10 @@ public class ExecutionDepositComponent { ExecutionDeposit newED; try { newED = createExecutionDeposit(sTrd, validator); - executionDepositImdg.insert(newED); + newED.setId(idGen.nextId()); + cash.add(ExecUploadKey.cash(newED)); + execsToInsert.put(newED.getId(), newED); + //executionDepositImdg.insert(newED); sendNotification(newED); log.debug("New executionDeposit.id={} was created.", newED.getId()); } catch (ClearingException ce) { @@ -159,6 +173,7 @@ public class ExecutionDepositComponent { log.error("When create new ExecutionDeposit by STrade[{}] error: {}", sTrd.getId(), ExceptionUtils.getStackTrace(e)); } } + executionDepositImdg.putAll(execsToInsert, 200); Long newMaxTradeNum = sTrades.stream().mapToLong(STrades::getTradeNum).max().orElseGet(() -> tradeNum); log.info("Process completed. Next tradeNum is {}", newMaxTradeNum); @@ -301,4 +316,9 @@ public class ExecutionDepositComponent { } return Optional.ofNullable(counterValidator.getStored(ValidationStored.STradesTradingClearingRegistry)); } + + @Override + public void initExecCash() { + ExecutionUploadCashUtil.loadExecutions(executionDepositImdg, cash); + } } 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 471daad2b..99d3f5b61 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 @@ -5,8 +5,11 @@ import java.math.RoundingMode; 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; @@ -44,6 +47,7 @@ import ru.spcex.platform.enumeration.Priority; import ru.spcex.platform.enumeration.Section; import ru.spcex.platform.enumeration.Side; import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.api.ImdgId; import ru.spcex.platform.imdg.api.ImdgProvider; import ru.spcex.platform.imdg.api.predicate.ImdgPredicate; import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder; @@ -60,7 +64,7 @@ import ru.spcex.platform.utils.validation.IValidator; @Component @EnableScheduling -public class ExecutionFondComponent { +public class ExecutionFondComponent implements IExecutionUploadComponent { private final Logger log = LoggerFactory.getLogger(getClass()); private final Imdg sTradeImdg; @@ -70,6 +74,7 @@ public class ExecutionFondComponent { private final Imdg fixedIncomeCashFlowImdg; private final Imdg marketImdg; private final Imdg crossRateImdg; + private final ImdgId idGen; //fixme ждать ТЗ Long tradeNum; @@ -79,6 +84,7 @@ public class ExecutionFondComponent { private final KafkaSender kafkaSender; private final NotificationSender notifications; private final boolean valuation; + private final Set cash = new HashSet<>(); @Autowired @@ -94,6 +100,7 @@ public class ExecutionFondComponent { this.marketImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Market, Market.class); this.fixedIncomeCashFlowImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_FixedIncomeCashFlow, FixedIncomeCashFlow.class); this.crossRateImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SCrossRate, SCrossRate.class); + this.idGen = imdgProvider.getImdgIdGenerator(); this.valuation = settings.getTrade().getValuation(); //fixme this.stradesValidator = stradesValidator; this.kafkaSender = kafkaSender; @@ -149,13 +156,17 @@ public class ExecutionFondComponent { return true; } Side excDepSide = sTrdSide.equals(Side.BUY) ? Side.BUY : Side.SELL; - return executionFondImdg.getFirstObjectByFieldValues( - Map.of("tradingDate", sTrd.getTradeDate(), - "exchangeExecutionId", sTrd.getTradeNum(), - "side", excDepSide.getKey())) != null; + return cash.contains( + ExecUploadKey.cash(sTrd.getTradeDate(), sTrd.getTradeNum(), excDepSide.getKey()) + ); + //return executionFondImdg.getFirstObjectByFieldValues( + // Map.of("tradingDate", sTrd.getTradeDate(), + // "exchangeExecutionId", sTrd.getTradeNum(), + // "side", excDepSide.getKey())) != null; }); log.info("{} strades left after already-added filtering", sTrades.size()); + Map execsToInsert = new HashMap<>(); for (STrades sTrd : sTrades) { log.info("new S_TRADE[{}], valuation {}", sTrd.getId(), valuation); @@ -169,7 +180,10 @@ public class ExecutionFondComponent { ExecutionFond newED; try { newED = createExecutionFond(sTrd, validator); - executionFondImdg.insert(newED); + newED.setId(idGen.nextId()); + cash.add(ExecUploadKey.cash(newED)); + execsToInsert.put(newED.getId(), newED); + //executionFondImdg.insert(newED); sendNotification(newED); log.debug("New executionDeposit.id={} was created.", newED.getId()); } catch (ClearingException ce) { @@ -178,6 +192,7 @@ public class ExecutionFondComponent { log.error("When create new ExecutionDeposit by STrade[{}] error: {}", sTrd.getId(), ExceptionUtils.getStackTrace(e)); } } + executionFondImdg.putAll(execsToInsert, 200); Long newMaxTradeNum = sTrades.stream().mapToLong(STrades::getTradeNum).max().orElseGet(() -> tradeNum); log.info("Process completed. Next tradeNum is {}", newMaxTradeNum); @@ -340,4 +355,9 @@ public class ExecutionFondComponent { private void logError(Long sTradeId, EnumMessage msg) { log.warn("sTrade id={} {}", sTradeId, msgResolver.resolve(msg)); } + + @Override + public void initExecCash() { + ExecutionUploadCashUtil.loadExecutions(executionFondImdg, cash); + } } 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 new file mode 100644 index 000000000..08e20a377 --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionUploadCashUtil.java @@ -0,0 +1,21 @@ +package ru.spcex.clearing.service.execution; + +import java.time.LocalDate; +import java.util.Collection; +import java.util.Set; +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) { + LocalDate today = LocalDate.now(); + ImdgPredicateBuilder pb = executionImdg.predicateBuilder(); + Collection execs = executionImdg.getCollectionObjectsByPredicate( + pb.equals("tradingDate", today) + ); + execs.forEach(e -> cash.add(ExecUploadKey.cash(e))); + } + +} diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/IExecutionUploadComponent.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/IExecutionUploadComponent.java new file mode 100644 index 000000000..67e0d84fd --- /dev/null +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/IExecutionUploadComponent.java @@ -0,0 +1,5 @@ +package ru.spcex.clearing.service.execution; + +public interface IExecutionUploadComponent { + void initExecCash(); +}