BSJ: AssetPrice, Activation Date, Valuation fixes

CHANGELOG.md

@@ -1,6 +1,11 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## [10.10.0]
+### Changed
+- Updated stream-investment to be able to seed intraday prices to all the assets
+- fix valuation and activation date in investment
+
 ## [10.9.0]
 ### Changed
 - Add product-document links in investment

stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentIntradayAssetPriceService.java

@@ -17,7 +17,9 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.ThreadLocalRandom;
 import javax.annotation.Nonnull;
 import lombok.RequiredArgsConstructor;
@@ -42,6 +44,10 @@
 @RequiredArgsConstructor
 public class InvestmentIntradayAssetPriceService {
 
+    static final int LIST_ASSET_PAGE_SIZE = 50;
+    private static final List<String> ASSET_EXPAND_FIELDS = List.of("market", "latest_price");
+    private static final String ASSET_LIST_FIELDS = "uuid,market,latest_price";
+
     private final AssetUniverseApi assetUniverseApi;
 
     /**
@@ -62,55 +68,31 @@ public Mono<List<GroupResult>> ingestIntradayPrices() {
     @Nonnull
     private Mono<List<List<GroupResult>>> generateIntradayPrices() {
         log.info("Generating Intraday Prices for Assets");
-        return assetUniverseApi.listAssetsWithResponseSpec(
-                null, null, null, null,
-                List.of("market", "latest_price"),
-                null, null,
-                "uuid,market,latest_price",
-                null, null, null, null, null,
-                null, null, null, null
-            ) // Above API returns custom projection with market and latest price expanded only, hence needs a custom return type.
-            .bodyToMono(PaginatedExpandedAssetList.class)
-            .flatMap(paginatedAssetList -> {
-
-                if (paginatedAssetList.getCount() == 0) {
+        AtomicInteger pageCounter = new AtomicInteger(0);
+        AtomicInteger assetCount = new AtomicInteger(0);
+
+        return listAssetsPage(LIST_ASSET_PAGE_SIZE, 0)
+            .expand(page -> {
+                if (page.getNext() == null) {
+                    return Mono.empty();
+                }
+                int nextOffset = pageCounter.incrementAndGet() * LIST_ASSET_PAGE_SIZE;
+                return listAssetsPage(LIST_ASSET_PAGE_SIZE, nextOffset);
+            })
+            .flatMap(page -> {
+                List<AssetWithMarketAndLatestPrice> results =
+                    Objects.requireNonNullElse(page.getResults(), List.of());
+                assetCount.addAndGet(results.size());
+                return Flux.fromIterable(results);
+            })
+            .flatMap(this::createIntradayPricesForAssetIfPresent)
+            .collectList()
+            .doOnSuccess(results -> {
+                if (assetCount.get() == 0) {
                     log.warn("No assets found with latest prices to generate intraday prices");
-                    return Mono.just(List.<List<GroupResult>>of());
+                } else {
+                    log.debug("Processed {} assets for intraday price generation", assetCount.get());
                 }
-
-                return Flux.fromIterable(paginatedAssetList.getResults())
-                    .flatMap(assetWithMarketAndLatestPrice -> {
-                        List<OASCreatePriceRequest> requests =
-                            generateIntradayPricesForAsset(assetWithMarketAndLatestPrice);
-
-                        log.debug("Generated intraday price requests: {}", requests.size());
-                        log.trace("Generated intraday price requests: {}", requests);
-
-                        if (requests.isEmpty()) {
-                            return Mono.empty();
-                        }
-
-                        return assetUniverseApi.bulkCreateIntradayAssetPrice(requests, null, null, null)
-                            .collectList()
-                            .doOnSuccess(created ->
-                                log.info(
-                                    "Successfully triggered creation of {} intraday prices for asset ({})",
-                                    requests.size(),
-                                    assetWithMarketAndLatestPrice.uuid()
-                                )
-                            )
-                            .doOnError(WebClientResponseException.class, ex ->
-                                log.error(
-                                    "Failed to create intraday prices for asset ({}): status={}, body={}",
-                                    assetWithMarketAndLatestPrice.uuid(),
-                                    ex.getStatusCode(),
-                                    ex.getResponseBodyAsString(),
-                                    ex
-                                )
-                            )
-                            .onErrorResume(e -> Mono.empty());
-                    })
-                    .collectList();
             })
             .doOnError(error -> {
                 if (error instanceof WebClientResponseException w) {
@@ -127,7 +109,50 @@ private Mono<List<List<GroupResult>>> generateIntradayPrices() {
                     );
                 }
             });
+    }
+
+    private Mono<PaginatedExpandedAssetList> listAssetsPage(int limit, int offset) {
+        return assetUniverseApi.listAssetsWithResponseSpec(
+                null, null, null, null,
+                ASSET_EXPAND_FIELDS,
+                null, null,
+                ASSET_LIST_FIELDS,
+                null, limit, null, null, offset,
+                null, null, null, null)
+            // Above API returns custom projection with market and latest price expanded only, hence needs a custom return type.
+            .bodyToMono(PaginatedExpandedAssetList.class);
+    }
+
+    private Mono<List<GroupResult>> createIntradayPricesForAssetIfPresent(
+        AssetWithMarketAndLatestPrice assetWithMarketAndLatestPrice) {
+        List<OASCreatePriceRequest> requests = generateIntradayPricesForAsset(assetWithMarketAndLatestPrice);
+
+        log.debug("Generated intraday price requests: {}", requests.size());
+        log.trace("Generated intraday price requests: {}", requests);
+
+        if (requests.isEmpty()) {
+            return Mono.empty();
+        }
 
+        return assetUniverseApi.bulkCreateIntradayAssetPrice(requests, null, null, null)
+            .collectList()
+            .doOnSuccess(created ->
+                log.info(
+                    "Successfully triggered creation of {} intraday prices for asset ({})",
+                    requests.size(),
+                    assetWithMarketAndLatestPrice.uuid()
+                )
+            )
+            .doOnError(WebClientResponseException.class, ex ->
+                log.error(
+                    "Failed to create intraday prices for asset ({}): status={}, body={}",
+                    assetWithMarketAndLatestPrice.uuid(),
+                    ex.getStatusCode(),
+                    ex.getResponseBodyAsString(),
+                    ex
+                )
+            )
+            .onErrorResume(e -> Mono.empty());
     }
 
     /**

stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentPortfolioAllocationService.java

@@ -298,17 +298,26 @@ private static Double calculateTrades(List<OASAllocationPositionCreateRequest> p
         return portfolioPositions.stream().map(a -> a.getPrice() * a.getShares()).reduce(0.0, Double::sum);
     }
 
+    private static OASAllocationCreateRequest buildCashOnlyAllocation(double amount, LocalDate valuationDate) {
+        double roundedAmount = roundPrice(amount);
+        return new OASAllocationCreateRequest()
+            .cashActive(roundedAmount)
+            .tradeTotal(0.0)
+            .balance(roundedAmount)
+            .invested(roundedAmount)
+            .earnings(0.0)
+            .valuationDate(valuationDate);
+    }
+
     public Mono<Deposit> createDepositAllocation(Deposit deposit) {
         String portfolioId = deposit.getPortfolio().toString();
         LocalDate valuationDate = Optional.ofNullable(deposit.getCompletedAt()).map(OffsetDateTime::toLocalDate)
             .orElse(LocalDate.now());
         return getAllocations(portfolioId, valuationDate.minusDays(4), valuationDate.plusDays(5), 10)
             .filter(Predicate.not(l -> l.stream().filter(a -> CollectionUtils.isEmpty(a.getPositions()))
                 .toList().isEmpty()))
-            .switchIfEmpty(upsertAllocations(portfolioId, List.of(
-                new OASAllocationCreateRequest()
-                    .cashActive(deposit.getAmount())
-                    .valuationDate(valuationDate)))
+            .switchIfEmpty(upsertAllocations(portfolioId, List.of(buildCashOnlyAllocation(deposit.getAmount(),
+                valuationDate)))
                 .onErrorResume(ex -> Mono.empty())
             )
             .onErrorResume(ex -> {

stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentPortfolioService.java

@@ -24,6 +24,7 @@
 import com.backbase.stream.investment.model.InvestmentPortfolio;
 import com.backbase.stream.investment.model.InvestmentPortfolioTradingAccount;
 import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
@@ -181,19 +182,20 @@ private Mono<PortfolioList> patchPortfolio(
         String uuid = existingProduct.getUuid().toString();
         List<UUID> associatedClients = getClients(investmentArrangement, clientsByLeId);
 
-        PatchedPortfolioUpdateRequest patchedPortfolioUpdateRequest = new PatchedPortfolioUpdateRequest()
-            .product(investmentArrangement.getInvestmentProductId())
-            .externalId(investmentArrangement.getExternalId())
-            .name(investmentArrangement.getName())
-            .clients(associatedClients)
-            .status(StatusA3dEnum.ACTIVE)
-            .activated(OffsetDateTime.now().minusMonths(config.getPortfolio().getActivationPastMonths()))
-            .extraData(investmentArrangement.getExtraData());
-
         log.debug("Attempting to patch existing portfolio: uuid={}, externalId={}, extraData={}",
             uuid, investmentArrangement.getExternalId(), investmentArrangement.getExtraData());
 
-        return portfolioApi.patchPortfolio(uuid, null, null, null, patchedPortfolioUpdateRequest)
+        return resolveActivationDate(existingProduct)
+            .map(activated -> new PatchedPortfolioUpdateRequest()
+                .product(investmentArrangement.getInvestmentProductId())
+                .externalId(investmentArrangement.getExternalId())
+                .name(investmentArrangement.getName())
+                .clients(associatedClients)
+                .status(StatusA3dEnum.ACTIVE)
+                .activated(activated)
+                .extraData(investmentArrangement.getExtraData()))
+            .flatMap(patchedPortfolioUpdateRequest -> portfolioApi.patchPortfolio(
+                uuid, null, null, null, patchedPortfolioUpdateRequest))
             .doOnSuccess(updated -> {
                 log.info("Successfully patched existing investment portfolio: uuid={}", updated.getUuid());
                 investmentArrangement.setInvestmentProductId(updated.getUuid());
@@ -238,7 +240,7 @@ private Mono<PortfolioList> createNewPortfolio(InvestmentArrangement investmentA
             .currency(Optional.ofNullable(investmentArrangement.getCurrency())
                 .orElse(config.getPortfolio().getDefaultCurrency()))
             .status(StatusA3dEnum.ACTIVE)
-            .activated(OffsetDateTime.now().minusMonths(config.getPortfolio().getActivationPastMonths()))
+            .activated(computePortfolioActivationDate())
             .extraData(investmentArrangement.getExtraData());
 
         log.debug("Creating investment portfolio: externalId={}, name={}, extraData={}",
@@ -275,14 +277,7 @@ private static List<UUID> getClients(InvestmentArrangement investmentArrangement
     public Mono<Deposit> upsertDeposits(InvestmentPortfolio investmentPortfolio) {
         PortfolioList portfolio = investmentPortfolio.getPortfolio();
         double initAmount = investmentPortfolio.getInitialCashOrDefault(config.getDeposit().getDefaultAmount());
-        return paymentsApi.listDeposits(null, null, null, null, null,
-                null, portfolio.getUuid(), null, null, null)
-            .filter(Objects::nonNull)
-            // Use flatMap with Mono.justOrEmpty() to safely handle null results without NPE,
-            // ensuring switchIfEmpty fallback triggers for both null and empty deposit lists.
-            .flatMap(paginatedResult ->
-                Mono.justOrEmpty(paginatedResult.getResults())
-                    .filter(list -> !list.isEmpty()))
+        return listPortfolioDeposits(portfolio.getUuid())
             .flatMap(deposits -> {
                 double deposited = deposits.stream().mapToDouble(Deposit::getAmount).sum();
                 double remaining = initAmount - deposited;
@@ -294,20 +289,29 @@ public Mono<Deposit> upsertDeposits(InvestmentPortfolio investmentPortfolio) {
             .onErrorResume(ex -> Mono.just(new Deposit()
                     .portfolio(portfolio.getUuid())
                     .amount(initAmount)
-                    .completedAt(portfolio.getActivated().plusDays(2))
+                    .completedAt(portfolio.getActivated())
                 )
             );
     }
 
+    private Mono<List<Deposit>> listPortfolioDeposits(UUID portfolioUuid) {
+        return paymentsApi.listDeposits(null, null, null, null, null,
+                null, portfolioUuid, null, null, null)
+            .filter(Objects::nonNull)
+            .flatMap(paginatedResult ->
+                Mono.justOrEmpty(paginatedResult.getResults())
+                    .filter(list -> !list.isEmpty()));
+    }
+
     @Nonnull
     private Mono<Deposit> createDeposit(PortfolioList portfolio, double defaultAmount) {
+        OffsetDateTime activated = portfolio.getActivated();
         return paymentsApi.createDeposit(new DepositRequest()
                 .portfolio(portfolio.getUuid())
                 .provider(config.getDeposit().getProvider())
-                .reason(UUID.randomUUID().toString())
                 .status(Status08fEnum.COMPLETED)
-                .transactedAt(portfolio.getActivated().plusDays(1))
-                .completedAt(portfolio.getActivated().plusDays(2))
+                .transactedAt(activated)
+                .completedAt(activated)
                 .amount(defaultAmount)
                 .depositType(DepositTypeEnum.TRANSFER)
                 .reason("Initial deposit")
@@ -718,6 +722,37 @@ private void logPortfolioTradingAccountError(String operation, String idLabel, S
         }
     }
 
+    private OffsetDateTime computePortfolioActivationDate() {
+        return OffsetDateTime.now(ZoneOffset.UTC).minusMonths(config.getPortfolio().getActivationPastMonths());
+    }
+
+    /**
+     * Resolves the activation timestamp for an existing portfolio.
+     *
+     * <p>Priority: earliest deposit date, then the portfolio's current activation date,
+     * then {@link #computePortfolioActivationDate()} for portfolios without either.
+     */
+    private Mono<OffsetDateTime> resolveActivationDate(PortfolioList existingPortfolio) {
+        UUID portfolioUuid = existingPortfolio.getUuid();
+        OffsetDateTime fallback = Optional.ofNullable(existingPortfolio.getActivated())
+            .orElseGet(this::computePortfolioActivationDate);
+
+        return listPortfolioDeposits(portfolioUuid)
+            .map(deposits -> earliestDepositDate(deposits).orElse(fallback))
+            .defaultIfEmpty(fallback)
+            .onErrorResume(ex -> {
+                log.warn("Failed to resolve activation date for portfolio: uuid={}", portfolioUuid, ex);
+                return Mono.just(fallback);
+            });
+    }
+
+    private Optional<OffsetDateTime> earliestDepositDate(List<Deposit> deposits) {
+        return deposits.stream()
+            .map(deposit -> Optional.ofNullable(deposit.getCompletedAt()).orElse(deposit.getTransactedAt()))
+            .filter(Objects::nonNull)
+            .min(OffsetDateTime::compareTo);
+    }
+
     /**
      * Logs portfolio creation errors with detailed information about the failure.
      *