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

state machine actions

Browse source
This commit is contained in:
ialbert 2025-06-27 17:56:09 +03:00
parent c88ce7f56f
commit dff000b496
5 changed files with 392 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/DataEnum.java
View file

@ -6,9 +6,13 @@ public enum DataEnum {
section,
session, //Session
dealsPrepared, //List<ExecutionCommon>
dealsPreparedFond,
dealsPreparedDeposit,
dealsPreparedCurrency,
counterPartyId, //Long
obligationAdmissionStashedRgs, //Map<Long, Registry>
paymentInstructionReturnMkr, //List<PaymentInstruction>
paymentInstructionSecurity, //List<PaymentInstruction>
paymentsWereCreatedDepositReturn, //Collection<PaymentInstruction>
paymentInfo,
}

87
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/AgainReviseStage3Action.java Normal file
View file

@ -0,0 +1,87 @@
package ru.spcex.clearing.session.state.action;
import java.util.Collection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.statemachine.StateContext;
import org.springframework.stereotype.Service;
import ru.clearing.classes.statics.data.registry.Registry;
import ru.spcex.clearing.error.ClearingError;
import ru.spcex.clearing.error.ClearingRuntimeException;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.utilities.NotificationNewRequest;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.clearing.session.stage.TaskType;
import ru.spcex.clearing.session.state.SsnEvent;
import ru.spcex.platform.enumeration.AccountType;
import ru.spcex.platform.enumeration.ObjectType;
import ru.spcex.platform.enumeration.Priority;
import ru.spcex.platform.enumeration.RegistryTradingParams;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
import ru.spcex.platform.imdg.api.predicate.ImdgPredicate;
import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder;
import ru.spcex.platform.imdg.api.predicate.specific.RegistryCodeSqlBuilder;
import ru.spcex.platform.utils.enumeration.EnumMessage;
import static ru.spcex.platform.utils.number.BigDecimalUtil.safeBD;
@Service
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class AgainReviseStage3Action extends AbstractSessionActionForOkErrorHandling {
private final Logger log = LoggerFactory.getLogger(getClass());
private final Imdg<Registry> registryImdg;
private final KafkaSender kafkaSender;
@Autowired
public AgainReviseStage3Action(ImdgProvider imdgProvider, KafkaSender kafkaSender) {
this.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class);
this.kafkaSender = kafkaSender;
}
@Override
public void actualExecute(StateContext<TaskType, SsnEvent> ctx) {
//clnr info depo
ImdgPredicateBuilder pb = registryImdg.predicateBuilder();
ImdgPredicate prdct = pb.and(
pb.sql(RegistryCodeSqlBuilder.getInstance(RegistryTradingParams.A__T).build()),
pb.or(
pb.equals("accountType", AccountType.Clrn.getKey()),
pb.equals("accountType", AccountType.Info.getKey()),
pb.equals("accountType", AccountType.Depo.getKey())
)
);
Collection<Registry> regsAT = registryImdg.getCollectionObjectsByPredicate(prdct);
log.trace("Select {} registry's by query \"{}\" for revision step 3", regsAT.size(), prdct);
int errorRegs = 0;
for (Registry reg : regsAT) {
if (reg.getBalance() == null && reg.getPlanBalance() == null) {
log.debug("Can not verify registry[{}] with null balance and planBalance", reg.getId());
} else {
if (reg.getBalance() == null || reg.getPlanBalance() == null) {
log.warn("Can not verify registry[{}] with null balance xor planBalance", reg.getId());
}
if (safeBD(reg.getBalance()).compareTo(safeBD(reg.getPlanBalance())) != 0) {
log.info("Revision: registry id={}, companyId={}, balance={}, plannedBalance={}",
reg.getId(), reg.getCompanyId(), reg.getBalance(), reg.getPlanBalance());
errorRegs++;
}
}
}
if (errorRegs > 0) {
log.warn("После сверки обнаружена разница между плановым и фактическим балансом. Всего {} регистров не совпали.", errorRegs);
NotificationNewRequest nRequest = new NotificationNewRequest();
nRequest.setObjectType(ObjectType.rgst.getKey());
nRequest.setPriority(Priority.HIGH.getKey());
nRequest.setComment("Сверка по результатам клиринговой сессии завершена с ошибками.");
kafkaSender.sendRequestToQueue(Consts.NOTIFICATION_NEW, nRequest);
throw new ClearingRuntimeException(new EnumMessage(ClearingError.PlanBalanceReviseError));
}
}
}

40
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/CompoundDealsPrepareAction.java Normal file
View file

@ -0,0 +1,40 @@
package ru.spcex.clearing.session.state.action;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.statemachine.StateContext;
import ru.clearing.classes.statics.data.execution.ExecutionCommon;
import ru.spcex.clearing.session.stage.TaskType;
import ru.spcex.clearing.session.state.DataEnum;
import ru.spcex.clearing.session.state.SsnEvent;
import ru.spcex.platform.classes.base.interfaces.ExecutionType;
public class CompoundDealsPrepareAction extends AbstractSessionActionForOkErrorHandling {
private final Logger log = LoggerFactory.getLogger(getClass());
private ExecutionType type;
public CompoundDealsPrepareAction(ExecutionType type) {
this.type = type;
}
public void setType(ExecutionType type) {
this.type = type;
}
@Override
public void actualExecute(StateContext<TaskType, SsnEvent> ctx) {
Map<Object, Object> ctxVars = ctx.getExtendedState().getVariables();
@SuppressWarnings("unchecked")
List<ExecutionCommon> execs = (List<ExecutionCommon>) ctx.getExtendedState().get(DataEnum.dealsPrepared, List.class);
if (execs == null) {execs = Collections.emptyList();}
log.info("{} executions extracted: {}", type, execs.size());
switch (type) {
case ExecutionFond -> ctxVars.put(DataEnum.dealsPreparedFond, execs);
case ExecutionDeposit -> ctxVars.put(DataEnum.dealsPreparedDeposit, execs);
case ExecutionCurrency -> ctxVars.put(DataEnum.dealsPreparedCurrency, execs);
}
}
}

18
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/FormingPaymentInstructionAssetsAction.java
View file

@ -7,7 +7,6 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.function.Function;
@ -100,6 +99,7 @@ public class FormingPaymentInstructionAssetsAction extends AbstractSessionAction
private SessionType sessionType;
private final PaymentStateMarkService dmvSrv;
private final AssetTBFProcessing assets;
private boolean disableCompanyCondition = false;
@Autowired
public FormingPaymentInstructionAssetsAction(ImdgProvider imdgProvider,
@ -123,6 +123,10 @@ public class FormingPaymentInstructionAssetsAction extends AbstractSessionAction
this.assets = assets;
}
public void setDisableCompanyCondition(boolean disableCompanyCondition) {
this.disableCompanyCondition = disableCompanyCondition;
}
@Override
protected void actualExecute(StateContext<TaskType, SsnEvent> ctx) {
Long sessionId = ctx.getExtendedState().get(DataEnum.sessionId, Long.class);
@ -169,10 +173,14 @@ public class FormingPaymentInstructionAssetsAction extends AbstractSessionAction
if (AccountType.Clrn.equalsByKey(registry.getAccountType())) {
counterAcc = accountImdg.getSingleObjectByID(registry.getAccountId());
} else if (AccountType.Info.equalsByKey(registry.getAccountType())) {
counterAcc = accountImdg.getFirstObjectByFieldValues(
Map.of("companyId", 1L,
"currency", registry.getSecuritySymbol(),
"accountType", AccountType.Anlt.getKey())
ImdgPredicateBuilder pb = accountImdg.predicateBuilder();
ImdgPredicate prdct = pb.and(
disableCompanyCondition ? pb.alwaysTrue() : pb.equals("companyId", 1L),
pb.equals("currency", registry.getSecuritySymbol()),
pb.equals("accountType", AccountType.Anlt.getKey())
);
counterAcc = accountImdg.getFirstObjectByPredicate(
prdct
);
}
Long senderId;

248
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/state/action/FormingPaymentInstructionDepositReturnAction.java Normal file
View file

@ -0,0 +1,248 @@
package ru.spcex.clearing.session.state.action;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.statemachine.StateContext;
import org.springframework.stereotype.Service;
import ru.clearing.classes.statics.data.account.Account;
import ru.clearing.classes.statics.data.payment.PaymentInstruction;
import ru.clearing.classes.statics.data.registry.Registry;
import ru.clearing.classes.statics.data.sdf.SDf03;
import ru.spcex.clearing.component.GroupRgsKey;
import ru.spcex.clearing.error.ClearingError;
import ru.spcex.clearing.error.ClearingRuntimeException;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.balance.ExportToFileRequest;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.clearing.service.Sdf03Creator;
import ru.spcex.clearing.service.builder.PaymentInstructionBuilderFinalMkrDeals;
import ru.spcex.clearing.service.registry.RegistryManager;
import ru.spcex.clearing.session.stage.TaskType;
import ru.spcex.clearing.session.state.DataEnum;
import ru.spcex.clearing.session.state.SsnEvent;
import ru.spcex.platform.enumeration.AccountStatus;
import ru.spcex.platform.enumeration.AccountType;
import ru.spcex.platform.enumeration.Allowed;
import ru.spcex.platform.enumeration.RegistryStatus;
import ru.spcex.platform.enumeration.RegistryTradingParams;
import static ru.spcex.platform.enumeration.RegistryTradingParams.*;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
import ru.spcex.platform.imdg.api.predicate.ImdgPredicate;
import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder;
import ru.spcex.platform.imdg.api.predicate.specific.RegistryCodeSqlBuilder;
import ru.spcex.platform.utils.collection.Pair;
import ru.spcex.platform.utils.enumeration.EnumMessage;
import ru.spcex.platform.utils.enumeration.IEnumKey;
import static ru.spcex.platform.utils.number.BigDecimalUtil.safeBD;
@Service
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class FormingPaymentInstructionDepositReturnAction extends AbstractSessionActionForOkErrorHandling {
private final Logger log = LoggerFactory.getLogger(getClass());
private final ImdgProvider imdgProvider;
private final Imdg<Registry> registryImdg;
private final Imdg<PaymentInstruction> paymentInstructionImdg;
private final Imdg<Account> accountImdg;
private final Imdg<SDf03> sDf03Imdg;
private final KafkaSender kafkaSender;
private final Sdf03Creator sdf03Creator;
private final RegistryManager rgsMng;
@Autowired
public FormingPaymentInstructionDepositReturnAction(ImdgProvider imdgProvider,
KafkaSender kafkaSender, Sdf03Creator sdf03Creator, RegistryManager rgsMng) {
this.kafkaSender = kafkaSender;
this.imdgProvider = imdgProvider;
this.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class);
this.accountImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Account, Account.class);
this.sDf03Imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SDf03, SDf03.class);
this.paymentInstructionImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_PaymentInstruction, PaymentInstruction.class);
this.sdf03Creator = sdf03Creator;
this.rgsMng = rgsMng;
}
@Override
protected void actualExecute(StateContext<TaskType, SsnEvent> ctx) {
Long sessionId = ctx.getExtendedState().get(DataEnum.sessionId, Long.class);
Instant now = Instant.now();
RegistryCodeSqlBuilder registryCodeSqlBuilder = RegistryCodeSqlBuilder.getInstance(RegistryTradingParams.OM_T, RegistryTradingParams.TM_T, RegistryTradingParams.OS_T, RegistryTradingParams.TS_T);
String registryCodeCondition = registryCodeSqlBuilder.build();
ImdgPredicateBuilder pb = registryImdg.predicateBuilder();
ImdgPredicate prdct = pb.and(
pb.sql(registryCodeCondition),
pb.equals("registryStatus", RegistryStatus.OK.getKey()),
pb.equals("sessionId", sessionId)
);
Collection<Registry> obligations = registryImdg.getCollectionObjectsByPredicate(prdct)
.stream()
.filter(rgs -> rgs.getValueDate() != null)
.filter(rgs -> rgs.getSettlementDate() != null)
.filter(rgs -> rgs.getSettlementDate().isAfter(rgs.getValueDate()))
.toList();
Map<GroupRgsKey, List<Registry>> registriesByGroup = obligations
.stream().
collect(Collectors.groupingBy(registry -> new GroupRgsKey(registry.getGroupId(), registry.getMarket())));
log.info("registry groups found {}", registriesByGroup.size());
ArrayList<PaymentInstruction> pmtCreated = new ArrayList<>();
for (Map.Entry<GroupRgsKey, List<Registry>> entry : registriesByGroup.entrySet()) {
List<Registry> groupRgs = entry.getValue();
Optional<Registry> lmtO = groupRgs.stream().filter(rgs -> RegistryManager.equalsByCode(RegistryTradingParams.OM_T, rgs)).findFirst();
Optional<Registry> cmtO = groupRgs.stream().filter(rgs -> RegistryManager.equalsByCode(RegistryTradingParams.TM_T, rgs)).findFirst();
if (lmtO.isEmpty() || cmtO.isEmpty()) {
log.error("groupId/market {}/{} LM*T or CM*T not found", entry.getKey().groupId(), entry.getKey().market());
continue;
}
Registry lm_t = lmtO.get(); //obligation by money
Registry cm_t = cmtO.get();
Account tranAcc = accountImdg.getFirstObjectBySQL(("accountType = '%s'" +
" and status = '%s'" +
" and processingSign = '%s'" +
" and currency = '%s'")
.formatted(AccountType.Tran.getKey(),
AccountStatus.ACTIVE.getKey(),
Allowed.ALLOWED.getKey(),
lm_t.getSecuritySymbol()));
if (tranAcc == null) {
throw new ClearingRuntimeException(new EnumMessage(ClearingError.AccountNotPresent, "accountType = " + AccountType.Tran.getKey()));
}
Optional<Registry> dmx = rgsMng.searchDmxByCounterPartyNotOk(lm_t);
if (dmx.isPresent()) {
log.debug("groupId {} LM*T#id={}, DM*X#id={} found, no action needed, skipping liability",
lm_t.getGroupId(), lm_t.getId(), dmx.get().getId());
dmx.get().setRegistryStatus(RegistryStatus.OK.getKey());
setUpdatedStoreInImdg(dmx.get(), now);
continue;
}
Optional<Registry> dmtInfo = rgsMng.searchDmtInfoNotOk(lm_t);
if (dmtInfo.isPresent()) {
log.debug("groupId {} LM*T#id={}, DM*T(INFO)#id={} found",
lm_t.getGroupId(), lm_t.getId(), dmtInfo.get().getId());
Pair<PaymentInstruction, PaymentInstruction> pmts = PaymentInstructionBuilderFinalMkrDeals.builder(imdgProvider)
.lm_t(lm_t)
.cm_t(cm_t)
.tranAcc(tranAcc)
.sessionId(sessionId)
.paymentPurposeLmt("По итогу клиринга. Возврат депозита " + lm_t.getContract() + " по ТКР " + lm_t.getTradingClearingRegistry() + " по итогу клиринга")
.paymentPurpose("По итогу клиринга. Возврат депозита " + cm_t.getContract() + " по ТКР " + cm_t.getTradingClearingRegistry() + " по итогу клиринга")
.currency(lm_t.getSecuritySymbol())
.build();
Pair.forEach(pmts, paymentInstructionImdg::insert);
Pair.forEach(pmts, pmtCreated::add);
lm_t.setPaymentId(pmts.getFirst().getId());
cm_t.setPaymentId(pmts.getSecond().getId());
lm_t.setUpdated(Instant.now());
cm_t.setUpdated(Instant.now());
registryImdg.update(lm_t);
registryImdg.update(cm_t);
dmtInfo.get().setRegistryStatus(RegistryStatus.OK.getKey());
setUpdatedStoreInImdg(dmtInfo.get(), now);
log.debug("groupId {} LM*T#id={} CM*T#id={} created PmtInst.id={} and PmtInst.id={}",
lm_t.getGroupId(), lm_t.getId(), cm_t.getId(), pmts.getFirst().getId(), pmts.getSecond().getId());
continue;
}
Optional<Registry> dmtClnr = rgsMng.searchDmtClrnNotOk(lm_t);
if (dmtClnr.isPresent()) {
log.debug("groupId {} {}#id={} found",
lm_t.getGroupId(), dmtClnr.get().getRegistryCode(), dmtClnr.get().getId());
{
//изменение активов - блокируем средства беред отправкой sdf'ов
Optional<Registry> amfO = rgsMng.findRelatedAsset(lm_t.getTradingClearingRegistryId(), lm_t.getCompanyId(), lm_t.getSecuritySymbol(), AM_F);
Optional<Registry> payerAmtO = rgsMng.findRelatedAsset(lm_t.getTradingClearingRegistryId(), lm_t.getCompanyId(), lm_t.getSecuritySymbol(), AM_T);
Optional<Registry> ambO = rgsMng.findRelatedAsset(lm_t.getTradingClearingRegistryId(), lm_t.getCompanyId(), lm_t.getSecuritySymbol(), AM_B);
Optional<Registry> receiverAmtO = rgsMng.findRelatedAsset(cm_t.getTradingClearingRegistryId(), cm_t.getCompanyId(), lm_t.getSecuritySymbol(), AM_B);
log.debug("changing A* registers based on LM_T.id={} and CM_T.id={} found AM*F.id={}, AM*T.id={}, AM*B.id={}, AM*B.id={}",
lm_t.getId(),
cm_t.getId(),
amfO.map(Registry::getId).orElse(null),
payerAmtO.map(Registry::getId).orElse(null),
ambO.map(Registry::getId).orElse(null),
receiverAmtO.map(Registry::getId).orElse(null)
);
//amfO.ifPresent(amf -> {
// amf.setBalance(safeBD(amf.getBalance()).subtract(safeBD(lm_t.getBalance())));
// setUpdatedStoreInImdg(amf, now);
//});
payerAmtO.ifPresent(amt -> {
amt.setSettledDebit(safeBD(amt.getSettledDebit()).add(safeBD(lm_t.getBalance())));
setUpdatedStoreInImdg(amt, now);
});
//ambO.ifPresent(amb -> {
// amb.setBalance(safeBD(amb.getBalance()).add(safeBD(lm_t.getBalance())));
// setUpdatedStoreInImdg(amb, now);
//});
receiverAmtO.ifPresent(amt -> {
// у отправителя и получателя одинаково, см. в FormingPaymentInstruction
amt.setSettledCredit(safeBD(amt.getSettledDebit()).add(safeBD(lm_t.getBalance())));
setUpdatedStoreInImdg(amt, now);
});
}
Pair<PaymentInstruction, PaymentInstruction> pmts = PaymentInstructionBuilderFinalMkrDeals.builder(imdgProvider)
.lm_t(lm_t)
.cm_t(cm_t)
.tranAcc(tranAcc)
.sessionId(sessionId)
.paymentPurpose("Возврат депозита " + cm_t.getContract() + " по ТКР " + cm_t.getTradingClearingRegistry() + " по итогу клиринга")
.paymentPurposeLmt("Возврат депозита " + lm_t.getContract() + " по ТКР " + lm_t.getTradingClearingRegistry() + " по итогу клиринга")
.currency(lm_t.getSecuritySymbol())
.build();
Pair.forEach(pmts, paymentInstructionImdg::insert);
Pair.forEach(pmts, pmtCreated::add);
log.debug("groupId {} LM*T#id={}, CM*T#id={} found, PaymentInstruction id={} and id={} created",
lm_t.getGroupId(), lm_t.getId(), cm_t.getId(), pmts.getFirst().getId(), pmts.getSecond().getId());
dmtClnr.get().setRegistryStatus(RegistryStatus.OK.getKey());
setUpdatedStoreInImdg(dmtClnr.get(), now);
}
}
sendSdfs(pmtCreated);
ctx.getExtendedState().getVariables().put(DataEnum.paymentsWereCreatedDepositReturn, pmtCreated);
}
private void setUpdatedStoreInImdg(Registry rgs, Instant now) {
rgs.setUpdated(now);
registryImdg.update(rgs);
}
private void sendSdfs(List<PaymentInstruction> formedPaymentInstructions) {
List<SDf03> sDf03Created = new ArrayList<>();
for (PaymentInstruction paymentInstruction : formedPaymentInstructions) {
Account account = accountImdg.getSingleObjectByID(paymentInstruction.getCreditLeg_accountId());
if (List.of(AccountType.Corr, AccountType.Clrn, AccountType.Tran, AccountType.Anlt, AccountType.Info)
.contains(IEnumKey.getEnumByKey(AccountType.class, account.getAccountType()))) {
//если info подставить anlt (единственный счет в системе)
sDf03Created.add(sdf03Creator.create(paymentInstruction));
}
}
Long sdf03GroupId = !sDf03Created.isEmpty() ? imdgProvider.getImdgIdGenerator().nextId() : null;
for (SDf03 sDf03 : sDf03Created) {
sDf03.setGenerationId(sdf03GroupId);
sDf03Imdg.insert(sDf03);
}
if (sdf03GroupId != null) {
ExportToFileRequest exportToFileRequest = new ExportToFileRequest();
exportToFileRequest.setNameOfTable("DF-03");
exportToFileRequest.setSdfGroupId(sdf03GroupId);
kafkaSender.sendRequestToQueue(Consts.EXPORT_PROCESS, exportToFileRequest);
}
}
}