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

Merge remote-tracking branch 'origin/CLS-731' into dev

Browse source
This commit is contained in:
AKurakin 2024-08-22 18:48:16 +03:00
commit 3ed5d8eab2
1 changed files with 63 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

106
clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/services/TradeImporterService.java
View file

@ -1,26 +1,7 @@
package ru.spcex.clearing.trade.importer.services; package ru.spcex.clearing.trade.importer.services;
import org.slf4j.Logger; import static ru.spcex.clearing.platform.messaging.domain.Consts.S_TRADES_IMPORTED;
import org.slf4j.LoggerFactory; import static ru.spcex.clearing.trade.importer.error.TradeImporterError.sTradesNotValid;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import ru.clearing.classes.statics.data.misc.BlacklistMarket;
import ru.clearing.classes.statics.data.misc.Market;
import ru.clearing.classes.statics.data.misc.STrades;
import ru.clearing.classes.statics.data.scheduler.ClearingCalendar;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.platform.messaging.domain.cud.utilities.STradesImportedRequest;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.platform.enumeration.Side;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
import ru.spcex.platform.utils.enumeration.EnumMessage;
import ru.spcex.platform.utils.enumeration.IMessageResolver;
import ru.spcex.platform.utils.log.ExceptionUtils;
import java.math.BigDecimal; import java.math.BigDecimal;
import java.sql.Date; import java.sql.Date;
@ -36,10 +17,28 @@ import java.util.Map;
import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer; import java.util.function.Consumer;
import java.util.function.Supplier; import java.util.function.Supplier;
import org.slf4j.Logger;
import static ru.spcex.clearing.platform.messaging.domain.Consts.S_TRADES_IMPORTED; import org.slf4j.LoggerFactory;
import static ru.spcex.clearing.scheduler.ISchedulerChecker.isWorkDayByTradingCalendar; import org.springframework.beans.factory.annotation.Value;
import static ru.spcex.clearing.trade.importer.error.TradeImporterError.sTradesNotValid; import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import ru.clearing.classes.statics.data.misc.BlacklistMarket;
import ru.clearing.classes.statics.data.misc.Market;
import ru.clearing.classes.statics.data.misc.STrades;
import ru.clearing.classes.statics.data.scheduler.ClearingCalendar;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.platform.messaging.domain.cud.utilities.STradesImportedRequest;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.platform.enumeration.DayStatus;
import ru.spcex.platform.enumeration.Side;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
import ru.spcex.platform.utils.enumeration.EnumMessage;
import ru.spcex.platform.utils.enumeration.IMessageResolver;
import ru.spcex.platform.utils.log.ExceptionUtils;
@Service @Service
@EnableScheduling @EnableScheduling
@ -74,23 +73,15 @@ public class TradeImporterService {
} }
public synchronized void process(boolean byCommand) { public synchronized void process(boolean byCommand) {
if (!byCommand) { if (!byCommand && !checkTradingDay()) {
LocalDate today = LocalDate.now(); return;
log.debug("Verify trading day {} by calendar", today);
ClearingCalendar calendar = clearingCalendarImdg.getFirstObjectByFieldValues(Map.of("clearingDate", today));
boolean isWeekend = Arrays.asList(DayOfWeek.SATURDAY, DayOfWeek.SUNDAY).contains(today.getDayOfWeek());
boolean isTradingDay = calendar == null ? !isWeekend : isWorkDayByTradingCalendar(calendar, isWeekend);
if (!isTradingDay) {
log.trace("Ignore by scheduler no trading day {}", today);
return;
}
} }
log.info("Start process import STrades from DB. byCommand={}", byCommand); log.info("Start process import STrades from DB. byCommand={}", byCommand);
if (log.isDebugEnabled()) { if (log.isDebugEnabled()) {
String countQuery = String.format("SELECT count(*) FROM %s.Trades", schema); String countQuery = String.format("SELECT count(*) FROM %s.Trades", schema);
log.trace("Query: {}", countQuery); log.trace("Query: {}", countQuery);
Collection<Long> tradesCountFromDB = jdbcTemplate.query(countQuery, Collection<Long> tradesCountFromDB = jdbcTemplate.query(countQuery,
(resultSet, i) -> resultSet.getObject(1, Long.class)); (resultSet, i) -> resultSet.getObject(1, Long.class));
log.debug("Count of Trades in db: {}. Will be loading.", tradesCountFromDB); log.debug("Count of Trades in db: {}. Will be loading.", tradesCountFromDB);
} }
@ -135,7 +126,7 @@ public class TradeImporterService {
}); });
log.debug("From DB {} STrades has loaded. Created {} new STrades, {} updated.", log.debug("From DB {} STrades has loaded. Created {} new STrades, {} updated.",
rows.get(), created.get(), updated.get()); rows.get(), created.get(), updated.get());
if (created.get() > 0 || (byCommand && updated.get() > 0)) { if (created.get() > 0 || (byCommand && updated.get() > 0)) {
if (byCommand) { if (byCommand) {
log.debug("Successfully import STrades from DB by command. Send to kafka command, topic={}", S_TRADES_IMPORTED); log.debug("Successfully import STrades from DB by command. Send to kafka command, topic={}", S_TRADES_IMPORTED);
@ -152,6 +143,35 @@ public class TradeImporterService {
} }
} }
private boolean checkTradingDay() {
LocalDate today = LocalDate.now();
log.debug("Verify trading day {} by calendar", today);
ClearingCalendar calendar = clearingCalendarImdg.getFirstObjectByFieldValues(Map.of("clearingDate", today));
boolean isTradingDate;
if (!Arrays.asList(DayOfWeek.SATURDAY, DayOfWeek.SUNDAY).contains(today.getDayOfWeek())) {
isTradingDate = true;
if (calendar != null && DayStatus.DayOff.equalsByKey(calendar.getDayStatus())) {
isTradingDate = false;
}
} else {
isTradingDate = false;
if (calendar != null && DayStatus.Workday.equalsByKey(calendar.getDayStatus())) {
isTradingDate = true;
}
}
if (!isTradingDate) {
log.trace("Ignore by scheduler no trading day {}", today);
return false;
}
return true;
}
/** /**
* returns OK - everything is kk * returns OK - everything is kk
*/ */
@ -185,16 +205,16 @@ public class TradeImporterService {
public STrades getSTradesFromImdg(STrades tradesDb, Imdg<STrades> sTradesImdg) { public STrades getSTradesFromImdg(STrades tradesDb, Imdg<STrades> sTradesImdg) {
return sTradesImdg.getFirstObjectByFieldValues(Map.of("tradeDate", tradesDb.getTradeDate(), return sTradesImdg.getFirstObjectByFieldValues(Map.of("tradeDate", tradesDb.getTradeDate(),
"tradeNum", tradesDb.getTradeNum(), "tradeNum", tradesDb.getTradeNum(),
"operation", tradesDb.getOperation(), "operation", tradesDb.getOperation(),
"classCode", tradesDb.getClassCode())); "classCode", tradesDb.getClassCode()));
} }
private boolean isValidTrades(STrades trades) { private boolean isValidTrades(STrades trades) {
return trades.getTradeDate() != null return trades.getTradeDate() != null
&& trades.getTradeNum() != null && trades.getTradeNum() != null
&& StringUtils.hasText(trades.getOperation()) && StringUtils.hasText(trades.getOperation())
&& trades.getTradeNum() != null; && trades.getTradeNum() != null;
} }
public STrades readSTrades(ResultSet resultSet) throws SQLException { public STrades readSTrades(ResultSet resultSet) throws SQLException {