treshenkov/clearing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

execution uploading optimization

Browse source
This commit is contained in:
ialbert 2024-08-07 15:17:29 +03:00
parent fc8d235fce
commit 4bf2c45c69
5 changed files with 99 additions and 89 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecUploadCashInfo.java Normal file
View file

@ -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());
}
}

56
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionCurrencyComponent.java
View file

@ -6,10 +6,8 @@ import java.time.LocalDate;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
import java.util.Collection; import java.util.Collection;
import java.util.HashMap; import java.util.HashMap;
import java.util.HashSet;
import java.util.Map; import java.util.Map;
import java.util.Optional; import java.util.Optional;
import java.util.Set;
import java.util.function.Function; import java.util.function.Function;
import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.Producer;
import org.slf4j.Logger; import org.slf4j.Logger;
@ -66,7 +64,7 @@ public class ExecutionCurrencyComponent implements IExecutionUploadComponent {
private final Function<STrades, IValidator> stradesValidator; private final Function<STrades, IValidator> stradesValidator;
private final KafkaSender kafkaSender; private final KafkaSender kafkaSender;
private static final DateTimeFormatter contractFormatter = DateTimeFormatter.ofPattern("ddMMyy"); private static final DateTimeFormatter contractFormatter = DateTimeFormatter.ofPattern("ddMMyy");
private final Set<ExecUploadKey> cash = new HashSet<>(); private final Map<ExecUploadKey, ExecUploadCashInfo> cash = new HashMap<>();
@ -106,14 +104,16 @@ public class ExecutionCurrencyComponent implements IExecutionUploadComponent {
// } // }
public void processNewTS(Long fromId) { public void processNewTS(Long fromId) {
if (fromId == null) { boolean cleanLoad = fromId == null;
if (cleanLoad) {
cash.clear(); cash.clear();
initExecCash();
} }
LocalDate today = LocalDate.now(); LocalDate today = LocalDate.now();
log.debug("Start check new S_TRADE at {}, fromId={}", today, fromId); log.debug("Start check new S_TRADE at {}, fromId={}", today, fromId);
//выбираем STrades на сегодня с правильным section //выбираем STrades на сегодня с правильным section
Collection<STrades> sTrades; Collection<STrades> sTrades;
if (fromId == null) { if (cleanLoad) {
sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of( sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of(
"tradeDate", LocalDate.now(), "tradeDate", LocalDate.now(),
"section", Section.CURR.getKey() "section", Section.CURR.getKey()
@ -135,30 +135,20 @@ public class ExecutionCurrencyComponent implements IExecutionUploadComponent {
} }
//убираем уже добавленные в ExecutionCurrency //убираем уже добавленные в ExecutionCurrency
sTrades.removeIf(sTrd -> { if (!cleanLoad) {
Side sTrdSide = IEnumKey.getEnumByKey(Side.class, sTrd.getOperation()); sTrades.removeIf(sTrd -> {
if (sTrdSide == null || !(sTrdSide.equals(Side.BUY) || sTrdSide.equals(Side.SELL))) { Side sTrdSide = IEnumKey.getEnumByKey(Side.class, sTrd.getOperation());
log.warn("cannot parse strade.tradeNum={} strade.operation {}", sTrd.getTradeNum(), sTrd.getOperation()); if (sTrdSide == null || !(sTrdSide.equals(Side.BUY) || sTrdSide.equals(Side.SELL))) {
return true; log.warn("cannot parse strade.tradeNum={} strade.operation {}", sTrd.getTradeNum(), sTrd.getOperation());
}
// 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);
return true; 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()); log.info("{} strades left after already-added filtering", sTrades.size());
Map<Long, ExecutionCurrency> execsToInsert = new HashMap<>(); Map<Long, ExecutionCurrency> execsToInsert = new HashMap<>();
@ -175,9 +165,17 @@ public class ExecutionCurrencyComponent implements IExecutionUploadComponent {
ExecutionCurrency newEC; ExecutionCurrency newEC;
try { try {
newEC = createExecutionCurrency(sTrd, validator); 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); execsToInsert.put(newEC.getId(), newEC);
cash.add(ExecUploadKey.cash(newEC));
//executionCurrencyImdg.insert(newEC); //executionCurrencyImdg.insert(newEC);
sendNotification(newEC); sendNotification(newEC);
log.debug("New executionCurrency.id={} was created.", newEC.getId()); log.debug("New executionCurrency.id={} was created.", newEC.getId());

56
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionDepositComponent.java
View file

@ -6,10 +6,8 @@ import java.time.LocalDate;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
import java.util.Collection; import java.util.Collection;
import java.util.HashMap; import java.util.HashMap;
import java.util.HashSet;
import java.util.Map; import java.util.Map;
import java.util.Optional; import java.util.Optional;
import java.util.Set;
import java.util.function.Function; import java.util.function.Function;
import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.Producer;
import org.slf4j.Logger; import org.slf4j.Logger;
@ -69,7 +67,7 @@ public class ExecutionDepositComponent implements IExecutionUploadComponent {
private final IMessageResolver msgResolver; private final IMessageResolver msgResolver;
private final Function<STrades, IValidator> stradesValidator; private final Function<STrades, IValidator> stradesValidator;
private final KafkaSender kafkaSender; private final KafkaSender kafkaSender;
private final Set<ExecUploadKey> cash = new HashSet<>(); private final Map<ExecUploadKey, ExecUploadCashInfo> cash = new HashMap<>();
private static final DateTimeFormatter contractFormatter = DateTimeFormatter.ofPattern("ddMMyy"); private static final DateTimeFormatter contractFormatter = DateTimeFormatter.ofPattern("ddMMyy");
@ -105,13 +103,15 @@ public class ExecutionDepositComponent implements IExecutionUploadComponent {
} }
public void processNewTS(Long fromId) { public void processNewTS(Long fromId) {
if (fromId == null) { boolean cleanLoad = fromId == null;
if (cleanLoad) {
cash.clear(); cash.clear();
initExecCash();
} }
log.debug("Start check new S_TRADE after {}, fromId={}", tradingDay, fromId); log.debug("Start check new S_TRADE after {}, fromId={}", tradingDay, fromId);
//выбираем STrades на сегодня с правильным section //выбираем STrades на сегодня с правильным section
Collection<STrades> sTrades; Collection<STrades> sTrades;
if (fromId == null) { if (cleanLoad) {
sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of( sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of(
"tradeDate", LocalDate.now(), "tradeDate", LocalDate.now(),
"section", Section.MKR.getKey() "section", Section.MKR.getKey()
@ -133,30 +133,20 @@ public class ExecutionDepositComponent implements IExecutionUploadComponent {
} }
//убираем уже добавленные в ExecutionDeposit //убираем уже добавленные в ExecutionDeposit
sTrades.removeIf(sTrd -> { if (!cleanLoad) {
Side sTrdSide = IEnumKey.getEnumByKey(Side.class, sTrd.getOperation()); sTrades.removeIf(sTrd -> {
if (sTrdSide == null || !(sTrdSide.equals(Side.BUY) || sTrdSide.equals(Side.SELL))) { Side sTrdSide = IEnumKey.getEnumByKey(Side.class, sTrd.getOperation());
log.warn("cannot parse strade.tradeNum={} strade.operation {}", sTrd.getTradeNum(), sTrd.getOperation()); if (sTrdSide == null || !(sTrdSide.equals(Side.BUY) || sTrdSide.equals(Side.SELL))) {
return true; log.warn("cannot parse strade.tradeNum={} strade.operation {}", sTrd.getTradeNum(), sTrd.getOperation());
}
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);
return true; 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()); log.info("{} strades left after already-added filtering", sTrades.size());
Map<Long, ExecutionDeposit> execsToInsert = new HashMap<>(); Map<Long, ExecutionDeposit> execsToInsert = new HashMap<>();
@ -173,8 +163,16 @@ public class ExecutionDepositComponent implements IExecutionUploadComponent {
ExecutionDeposit newED; ExecutionDeposit newED;
try { try {
newED = createExecutionDeposit(sTrd, validator); newED = createExecutionDeposit(sTrd, validator);
newED.setId(idGen.nextId()); ExecUploadKey execKey = ExecUploadKey.cash(newED);
cash.add(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); execsToInsert.put(newED.getId(), newED);
//executionDepositImdg.insert(newED); //executionDepositImdg.insert(newED);
sendNotification(newED); sendNotification(newED);

54
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionFondComponent.java
View file

@ -6,10 +6,8 @@ import java.time.Instant;
import java.time.LocalDate; import java.time.LocalDate;
import java.util.Collection; import java.util.Collection;
import java.util.HashMap; import java.util.HashMap;
import java.util.HashSet;
import java.util.Map; import java.util.Map;
import java.util.Optional; import java.util.Optional;
import java.util.Set;
import java.util.function.Function; import java.util.function.Function;
import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.Producer;
import org.slf4j.Logger; import org.slf4j.Logger;
@ -84,7 +82,7 @@ public class ExecutionFondComponent implements IExecutionUploadComponent {
private final KafkaSender kafkaSender; private final KafkaSender kafkaSender;
private final NotificationSender notifications; private final NotificationSender notifications;
private final boolean valuation; private final boolean valuation;
private final Set<ExecUploadKey> cash = new HashSet<>(); private final Map<ExecUploadKey, ExecUploadCashInfo> cash = new HashMap<>();
@Autowired @Autowired
@ -124,13 +122,15 @@ public class ExecutionFondComponent implements IExecutionUploadComponent {
} }
public void processNewTS(Long fromId) { public void processNewTS(Long fromId) {
if (fromId == null) { boolean cleanLoad = fromId == null;
if (cleanLoad) {
cash.clear(); cash.clear();
initExecCash();
} }
log.debug("Start check new S_TRADE after {}, fromId={}", tradingDay, fromId); log.debug("Start check new S_TRADE after {}, fromId={}", tradingDay, fromId);
//выбираем STrades на сегодня с правильным section //выбираем STrades на сегодня с правильным section
Collection<STrades> sTrades; Collection<STrades> sTrades;
if (fromId == null) { if (cleanLoad) {
sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of( sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of(
"tradeDate", LocalDate.now(), "tradeDate", LocalDate.now(),
"section", Section.FOND.getKey() "section", Section.FOND.getKey()
@ -152,28 +152,20 @@ public class ExecutionFondComponent implements IExecutionUploadComponent {
} }
//убираем уже добавленные в ExecutionDeposit //убираем уже добавленные в ExecutionDeposit
sTrades.removeIf(sTrd -> { if (!cleanLoad) {
Side sTrdSide = IEnumKey.getEnumByKey(Side.class, sTrd.getOperation()); sTrades.removeIf(sTrd -> {
if (sTrdSide == null || !(sTrdSide.equals(Side.BUY) || sTrdSide.equals(Side.SELL))) { Side sTrdSide = IEnumKey.getEnumByKey(Side.class, sTrd.getOperation());
log.warn("cannot parse strade.tradeNum={} strade.operation {}", sTrd.getTradeNum(), sTrd.getOperation()); if (sTrdSide == null || !(sTrdSide.equals(Side.BUY) || sTrdSide.equals(Side.SELL))) {
return true; log.warn("cannot parse strade.tradeNum={} strade.operation {}", sTrd.getTradeNum(), sTrd.getOperation());
}
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);
return true; 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()); log.info("{} strades left after already-added filtering", sTrades.size());
Map<Long, ExecutionFond> execsToInsert = new HashMap<>(); Map<Long, ExecutionFond> execsToInsert = new HashMap<>();
@ -190,8 +182,16 @@ public class ExecutionFondComponent implements IExecutionUploadComponent {
ExecutionFond newED; ExecutionFond newED;
try { try {
newED = createExecutionFond(sTrd, validator); newED = createExecutionFond(sTrd, validator);
newED.setId(idGen.nextId()); ExecUploadKey execKey = ExecUploadKey.cash(newED);
cash.add(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); execsToInsert.put(newED.getId(), newED);
//executionFondImdg.insert(newED); //executionFondImdg.insert(newED);
sendNotification(newED); sendNotification(newED);

7
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionUploadCashUtil.java
View file

@ -2,20 +2,19 @@ package ru.spcex.clearing.service.execution;
import java.time.LocalDate; import java.time.LocalDate;
import java.util.Collection; import java.util.Collection;
import java.util.Set; import java.util.Map;
import ru.clearing.classes.statics.data.execution.ExecutionCommon; import ru.clearing.classes.statics.data.execution.ExecutionCommon;
import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder; import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder;
public class ExecutionUploadCashUtil { public class ExecutionUploadCashUtil {
public static <T extends ExecutionCommon> public static <T extends ExecutionCommon>
void loadExecutions(Imdg<T> executionImdg, Set<ExecUploadKey> cash) { void loadExecutions(Imdg<T> executionImdg, Map<ExecUploadKey, ExecUploadCashInfo> cash) {
LocalDate today = LocalDate.now(); LocalDate today = LocalDate.now();
ImdgPredicateBuilder pb = executionImdg.predicateBuilder(); ImdgPredicateBuilder pb = executionImdg.predicateBuilder();
Collection<T> execs = executionImdg.getCollectionObjectsByPredicate( Collection<T> execs = executionImdg.getCollectionObjectsByPredicate(
pb.equals("tradingDate", today) pb.equals("tradingDate", today)
); );
execs.forEach(e -> cash.add(ExecUploadKey.cash(e))); execs.forEach(e -> cash.put(ExecUploadKey.cash(e), ExecUploadCashInfo.cash(e)));
} }
} }