Merge remote-tracking branch 'origin/dev' into dev
This commit is contained in:
ialbert
commit
9e00b96e95
5 changed files with 44 additions and 77 deletions
|
|
@ -20,13 +20,16 @@ public class ClearingService implements DisposableBean {
|
|||
private final PaymentUpdateBySdf04 paymentUpdater;
|
||||
private final VerificationResultComponent verificationResultComponent;
|
||||
private final Clearing clearing;
|
||||
private final ExecutionDepositComponent executionDepositComponent;
|
||||
|
||||
@Autowired
|
||||
public ClearingService(SdfCreatorBySTLDPayment sdfCreator, PaymentUpdateBySdf04 paymentUpdater,
|
||||
VerificationResultComponent verificationResultComponent, Clearing clearing) {
|
||||
VerificationResultComponent verificationResultComponent, Clearing clearing,
|
||||
ExecutionDepositComponent executionDepositComponent) {
|
||||
this.sdfCreator = sdfCreator;
|
||||
this.paymentUpdater = paymentUpdater;
|
||||
this.verificationResultComponent = verificationResultComponent;
|
||||
this.executionDepositComponent = executionDepositComponent;
|
||||
this.clearing = clearing;
|
||||
this.executor = Executors.newSingleThreadExecutor();
|
||||
}
|
||||
|
|
@ -81,4 +84,16 @@ public class ClearingService implements DisposableBean {
|
|||
}
|
||||
});
|
||||
}
|
||||
|
||||
public void executeSTrade() {
|
||||
log.info("start STrade check for ExecutionDeposit");
|
||||
executor.execute(() -> {
|
||||
try {
|
||||
executionDepositComponent.processNewTS();
|
||||
} catch (Throwable e) {
|
||||
log.error("{}", ExceptionUtils.getStackTrace(e));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -34,6 +34,9 @@ public class EventsReceiver extends QueueConsumer implements InitializingBean {
|
|||
callback(Object.class) //todo check Object suitable
|
||||
.setConsumer(event -> clearingService.executeClearing())
|
||||
.forDestination(Task.startOfClearing.topic(), callbacks::put);
|
||||
callback(Object.class)
|
||||
.setConsumer(event -> clearingService.executeSTrade())
|
||||
.forDestination(Task.getOfTrades.topic(), callbacks::put);
|
||||
init();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
|
|||
import ru.spcex.clearing.platform.messaging.domain.Consts;
|
||||
import ru.spcex.clearing.platform.messaging.domain.cud.registry.DealRegisterNewRequest;
|
||||
import ru.spcex.clearing.platform.messaging.domain.cud.securitites.MoneyMarketSecurityNewRequest;
|
||||
import ru.spcex.platform.classes.base.SpcexObjectBase;
|
||||
import ru.spcex.platform.enumeration.Allowed;
|
||||
import ru.spcex.platform.enumeration.Market;
|
||||
import ru.spcex.platform.imdg.api.Imdg;
|
||||
|
|
@ -96,11 +97,10 @@ public class ExecutionDepositComponent {
|
|||
log.info("Reset trading day for search STrade: tradeNum={}, tradeDat={}", tradeNum, tradingDay);
|
||||
}
|
||||
|
||||
@Scheduled(cron = "${clearing-service.scheduler.check-s-trade}")
|
||||
public void processNewTS() {
|
||||
log.debug("Start check new S_TRADE. Start tradeNum={}", tradeNum);
|
||||
log.debug("Start check new S_TRADE after {}", tradingDay);
|
||||
ImdgPredicateBuilder pb = sTradeImdg.predicateBuilder();
|
||||
ImdgPredicate sql = pb.and(pb.greater("tradeNum", tradeNum), pb.greatEqual("tradeDateTime", tradingDay));
|
||||
ImdgPredicate sql = pb.greatEqual("tradeDateTime", tradingDay);
|
||||
Collection<STrade> sTrades = sTradeImdg.getCollectionObjectsByPredicate(sql);
|
||||
log.info("Found {} new s_trade with trade_num>{}", sTrades.size(), tradeNum);
|
||||
|
||||
|
|
@ -112,12 +112,13 @@ public class ExecutionDepositComponent {
|
|||
// Выявление новых сделок необходимо выполнить следующие контрольные проверки:
|
||||
|
||||
// Проверить все инструменты.
|
||||
Set<String> secCodesOfSecurity;
|
||||
{
|
||||
Set<String> secCodesOfSTrade = sTrades.stream().map(STrade::getSecCode).filter(Objects::nonNull).collect(Collectors.toSet());
|
||||
log.debug("Verify {} instruments for {} STrade's.", secCodesOfSTrade.size(), sTrades.size());
|
||||
ImdgPredicate allIn = securityImdg.predicateBuilder().in("securitySymbol", secCodesOfSTrade.toArray(new String[0]));
|
||||
Collection<Security> foundSecurities = securityImdg.getCollectionObjectsByPredicate(allIn);
|
||||
Set<String> secCodesOfSecurity = foundSecurities.stream().map(Security::getSecuritySymbol).filter(Objects::nonNull).collect(Collectors.toSet());
|
||||
secCodesOfSecurity = foundSecurities.stream().map(Security::getSecuritySymbol).filter(Objects::nonNull).collect(Collectors.toSet());
|
||||
if (secCodesOfSecurity.containsAll(secCodesOfSTrade)) {
|
||||
log.debug("All {} Security found by {} secCodes from STrade",
|
||||
secCodesOfSecurity.size(), secCodesOfSTrade.size());
|
||||
|
|
@ -126,43 +127,41 @@ public class ExecutionDepositComponent {
|
|||
notFoundSymbol.removeAll(secCodesOfSecurity);
|
||||
log.info("Found only {} Security by {} secCodes from STrade. Not found: {}",
|
||||
secCodesOfSecurity.size(), secCodesOfSTrade.size(), notFoundSymbol);
|
||||
createNewSecurities(notFoundSymbol);
|
||||
log.info("Stop till they all will be created");
|
||||
|
||||
auditMessage("В security нет записей с securitySymbol", notFoundSymbol);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
Long generationId = idGenerator.nextId();
|
||||
log.info("generationId = {}", generationId);
|
||||
LocalDate today = LocalDate.now();
|
||||
for (STrade trade : sTrades) {
|
||||
log.trace("Check s_trade[{}].tradeNum={}", trade.getId(), trade.getTradeNum());
|
||||
log.trace("Check s_trade[{}].tradeNum={} on date {}", trade.getId(), trade.getTradeNum(), today);
|
||||
Collection<ExecutionDeposit> existsEDeposit = executionDepositImdg.getCollectionObjectsByFieldValues(Map.of(
|
||||
"exchangeExecutionId", trade.getTradeNum(),
|
||||
"exchangeExecutionTime", trade.getTradeDateTime()
|
||||
"clearingDate", today
|
||||
));
|
||||
|
||||
if (existsEDeposit.isEmpty()) {
|
||||
log.trace("S_TRADE[{}] new", trade.getId());
|
||||
ExecutionDeposit newED = null;
|
||||
|
||||
// Проверка secCode
|
||||
if (!secCodesOfSecurity.contains(trade.getSecCode())) {
|
||||
log.error("Error {}: STrade[{}].secCode={} not found",
|
||||
ClearingError.RecordNotFound.getId(), trade.getId(), trade.getSecCode());
|
||||
continue;
|
||||
}
|
||||
|
||||
ExecutionDeposit newED;
|
||||
try {
|
||||
newED = createExecutionDeposit(trade, Allowed.ALLOWED/*todo уточнить момент заполнения*/, generationId);
|
||||
verification(newED);
|
||||
newED = createExecutionDeposit(trade, null, null);
|
||||
executionDepositImdg.insert(newED);
|
||||
sendNotification(newED);
|
||||
} catch (ClearingException ce) {
|
||||
auditMessage(ce);
|
||||
} catch (Exception e) {
|
||||
if (newED != null) {
|
||||
newED.setCoverageStatus(Allowed.DENIED.getKey());
|
||||
}
|
||||
log.error("When create new ExecutionDeposit by STrade[{}]", trade.getId());
|
||||
log.error("When create new ExecutionDeposit by STrade[{}] error: {}", trade.getId(), ExceptionUtils.getStackTrace(e));
|
||||
}
|
||||
|
||||
} else {
|
||||
long[] idToLong = existsEDeposit.stream().mapToLong(ed -> ed.getId()).toArray();
|
||||
log.warn("S_TRADE[{}] already has executionDeposit: {}", trade.getId(), Arrays.toString(idToLong));
|
||||
long[] idToLong = existsEDeposit.stream().mapToLong(SpcexObjectBase::getId).toArray();
|
||||
log.trace("S_TRADE[{}] already has executionDeposit: {}", trade.getId(), Arrays.toString(idToLong));
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -171,14 +170,6 @@ public class ExecutionDepositComponent {
|
|||
log.info("Process completed. Next tradeNum is {}", newMaxTradeNum);
|
||||
}
|
||||
|
||||
protected void verification(ExecutionDeposit forED) throws ClearingException {
|
||||
/*todo Рассчитанные в КС контрольные суммы (общее количество сделок и суммарный объем заключенных сделок в денежном выражении)
|
||||
должны совпадать со значениями, рассчитанными Торговой системой:
|
||||
count(execution[tradingDay]) = count (trade_arqua)
|
||||
*/
|
||||
// использовать ли VerificationResultComponent для сверки или здесь код добавить.
|
||||
|
||||
}
|
||||
|
||||
protected void auditMessage(ClearingException ce) {
|
||||
log.error("AUDIT error code {}: {}", ce.getEnumMsg(), ce.getMessage());
|
||||
|
|
@ -192,51 +183,6 @@ public class ExecutionDepositComponent {
|
|||
log.error("audit \"clearing-service\", errorText: {}", txt);
|
||||
}
|
||||
|
||||
/**
|
||||
* в очередь kafka для модуля securities-service сообщение о добавлении инструмента с параметром securitySymbol=s_trade.sec_code
|
||||
*
|
||||
* @param newSymbolRequest
|
||||
*/
|
||||
protected void createNewSecurities(Collection<String> newSymbolRequest) {
|
||||
final String destination = Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW;
|
||||
List<String> symbolRequests = new ArrayList<>(newSymbolRequest); // чтобы в случае ошибки отобразить номер в логе
|
||||
List<Future<RecordMetadata>> sendAll = new ArrayList<>(symbolRequests.size());
|
||||
for (String newSymbol : symbolRequests) {
|
||||
if (newSymbol == null || newSymbol.isEmpty()) {
|
||||
log.warn("Empty SecuritySumbol");
|
||||
} else {
|
||||
MoneyMarketSecurityNewRequest requestPayload = new MoneyMarketSecurityNewRequest();
|
||||
requestPayload.setSecuritySymbol(newSymbol);
|
||||
|
||||
BaseRequest<Object> request = new BaseRequest<>();
|
||||
request.setId(idGenerator.nextId());
|
||||
request.setActionType(ActionType.NEW);
|
||||
request.setRequestPayload(requestPayload);
|
||||
|
||||
// saveRequestToStorage(destination, request); //сохраняет данные о запросе в хранилище
|
||||
log.trace("Send to {} new symbol \"{}\" ", destination, newSymbol);
|
||||
Future<RecordMetadata> send = kafka.send(new ProducerRecord<>(destination, request));
|
||||
sendAll.add(send);
|
||||
}
|
||||
}
|
||||
int i = 0;
|
||||
for (Future<RecordMetadata> future : sendAll) {
|
||||
try {
|
||||
future.get(); // get exception
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
String about = i < symbolRequests.size() ? symbolRequests.get(i) : "(out of range i=" + i + ")";
|
||||
log.warn("Thread interrupted! On send symbol \"{}\"", about);
|
||||
throw new RuntimeException(e);
|
||||
} catch (ExecutionException e) {
|
||||
String about = i < symbolRequests.size() ? symbolRequests.get(i) : "(out of range i=" + i + ")";
|
||||
log.error("Error send message for symbol \"{}\" to {}: {}", destination,
|
||||
about, ExceptionUtils.getStackTrace(e.getCause() == null ? e : e.getCause()));
|
||||
}
|
||||
i++;
|
||||
}
|
||||
}
|
||||
|
||||
protected void sendNotification(ExecutionDeposit forED) {
|
||||
final String destination = Consts.REGISTRY_DEAL_REGISTER_NEW;
|
||||
DealRegisterNewRequest requestPayload = new DealRegisterNewRequest();
|
||||
|
|
|
|||
|
|
@ -20,4 +20,3 @@ clearing-service.kafka-producer.linger-ms=1
|
|||
clearing-service.kafka-producer.buffer-memory=33554432
|
||||
|
||||
clearing-service.scheduler.check-payment-instruction=*/5 * * * * *
|
||||
clearing-service.scheduler.check-s-trade=0 * 0 1 * ?
|
||||
|
|
|
|||
|
|
@ -27,6 +27,10 @@ public class ExecutionDepositMapStore extends TemplateMapStore<ExecutionDeposit>
|
|||
return IMDGDistributedNames.Map_ExecutionDeposit;
|
||||
}
|
||||
|
||||
public String[] getIndexingField() {
|
||||
return new String[]{"exchangeExecutionId"};
|
||||
}
|
||||
|
||||
@Override
|
||||
public String[] getFields() {
|
||||
return new String[]{"ID", "CREATED_AT", "UPDATED_AT",
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue