Investment/next improvement data set handler (#570)

* Implement the first part of price ingestion; add config for price creation & fix price generation and valiations
* split sagas for asset univers & portfolios wealth
* generate allocation for portfolios; define allocation by each day as it's required; add config for portfolio activation period by months; refactor asset universe; add checks for prices ingestion finish;
* add implementation to create a deposit of portfolio cash and allocations orders
* Adding Asset Categories and Types
* add CHANGELOG.md

---------

Co-authored-by: pawana_backbase <pawana@backbase.com>

Author: Aboruhen

CHANGELOG.md

@@ -1,6 +1,14 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## [8.9.0](https://github.com/Backbase/stream-services/compare/8.8.0...8.9.0)
+### Changed
+- Investment service
+  - add support for asset categories
+  - improve allocations setup
+  - add deposit and orders
+  - price ingest async tasks complete check 
+
 ## [8.8.0](https://github.com/Backbase/stream-services/compare/8.7.0...8.8.0)
 ### Changed
 - Investment service

stream-dbs-clients/src/main/java/com/backbase/stream/clients/config/InvestmentClientConfig.java

@@ -1,10 +1,14 @@
 package com.backbase.stream.clients.config;
 
 import com.backbase.investment.api.service.ApiClient;
+import com.backbase.investment.api.service.v1.AllocationsApi;
 import com.backbase.investment.api.service.v1.AssetUniverseApi;
+import com.backbase.investment.api.service.v1.AsyncBulkGroupsApi;
 import com.backbase.investment.api.service.v1.ClientApi;
 import com.backbase.investment.api.service.v1.FinancialAdviceApi;
+import com.backbase.investment.api.service.v1.InvestmentApi;
 import com.backbase.investment.api.service.v1.InvestmentProductsApi;
+import com.backbase.investment.api.service.v1.PaymentsApi;
 import com.backbase.investment.api.service.v1.PortfolioApi;
 import com.fasterxml.jackson.annotation.JsonInclude.Include;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -38,7 +42,8 @@ public InvestmentClientConfig() {
      */
     @Bean
     @ConditionalOnMissingBean
-    public ApiClient investmentApiClient(WebClient interServiceWebClient, ObjectMapper objectMapper, DateFormat dateFormat) {
+    public ApiClient investmentApiClient(WebClient interServiceWebClient, ObjectMapper objectMapper,
+        DateFormat dateFormat) {
         ObjectMapper mapper = objectMapper.copy();
         mapper.setSerializationInclusion(Include.NON_EMPTY);
         mapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
@@ -83,4 +88,28 @@ public AssetUniverseApi assetUniverseApi(ApiClient investmentApiClient) {
         return new AssetUniverseApi(investmentApiClient);
     }
 
+    @Bean
+    @ConditionalOnMissingBean
+    public AllocationsApi allocationsApi(ApiClient investmentApiClient) {
+        return new AllocationsApi(investmentApiClient);
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public InvestmentApi investmentApi(ApiClient investmentApiClient) {
+        return new InvestmentApi(investmentApiClient);
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public PaymentsApi paymentsApi(ApiClient investmentApiClient) {
+        return new PaymentsApi(investmentApiClient);
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public AsyncBulkGroupsApi asyncBulkGroupsApi(ApiClient investmentApiClient) {
+        return new AsyncBulkGroupsApi(investmentApiClient);
+    }
+
 }

stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentIngestionConfigurationProperties.java

@@ -0,0 +1,20 @@
+
+package com.backbase.stream.configuration;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Configuration properties governing investment client ingestion behavior.
+ */
+@Data
+@ConfigurationProperties(prefix = "backbase.bootstrap.ingestions.investment")
+public class InvestmentIngestionConfigurationProperties {
+
+    private boolean assetUniversEnabled = true;
+    private boolean wealthEnabled = true;
+    private int portfolioActivationPastMonths = 1;
+
+
+}
+

stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentSagaConfigurationProperties.java

@@ -1,17 +1,24 @@
 package com.backbase.stream.configuration;
 
 import com.backbase.investment.api.service.ApiClient;
+import com.backbase.investment.api.service.v1.AllocationsApi;
 import com.backbase.investment.api.service.v1.AssetUniverseApi;
 import com.backbase.investment.api.service.v1.ClientApi;
 import com.backbase.investment.api.service.v1.FinancialAdviceApi;
+import com.backbase.investment.api.service.v1.InvestmentApi;
 import com.backbase.investment.api.service.v1.InvestmentProductsApi;
+import com.backbase.investment.api.service.v1.PaymentsApi;
 import com.backbase.investment.api.service.v1.PortfolioApi;
 import com.backbase.stream.clients.autoconfigure.DbsApiClientsAutoConfiguration;
+import com.backbase.stream.investment.saga.InvestmentAssetUniversSaga;
 import com.backbase.stream.investment.saga.InvestmentSaga;
+import com.backbase.stream.investment.service.AsyncTaskService;
 import com.backbase.stream.investment.service.CustomIntegrationApiService;
+import com.backbase.stream.investment.service.InvestmentAssetPriceService;
 import com.backbase.stream.investment.service.InvestmentAssetUniverseService;
 import com.backbase.stream.investment.service.InvestmentClientService;
 import com.backbase.stream.investment.service.InvestmentModelPortfolioService;
+import com.backbase.stream.investment.service.InvestmentPortfolioAllocationService;
 import com.backbase.stream.investment.service.InvestmentPortfolioService;
 import lombok.RequiredArgsConstructor;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -24,7 +31,7 @@
     DbsApiClientsAutoConfiguration.class,
 })
 @EnableConfigurationProperties({
-    InvestmentSagaConfigurationProperties.class
+    InvestmentIngestionConfigurationProperties.class
 })
 @RequiredArgsConstructor
 @Configuration
@@ -43,8 +50,10 @@ public CustomIntegrationApiService customIntegrationApiService(ApiClient apiClie
 
     @Bean
     public InvestmentPortfolioService investmentPortfolioService(PortfolioApi portfolioApi,
-        InvestmentProductsApi investmentProductsApi) {
-        return new InvestmentPortfolioService(investmentProductsApi, portfolioApi);
+        InvestmentProductsApi investmentProductsApi, PaymentsApi paymentsApi,
+        InvestmentIngestionConfigurationProperties configurationProperties) {
+        return new InvestmentPortfolioService(investmentProductsApi, portfolioApi, paymentsApi,
+            configurationProperties);
     }
 
     @Bean
@@ -59,13 +68,37 @@ public InvestmentModelPortfolioService investmentModelPortfolioService(Financial
         return new InvestmentModelPortfolioService(financialAdviceApi, customIntegrationApiService);
     }
 
+    @Bean
+    public InvestmentAssetPriceService investmentAssetPriceService(AssetUniverseApi assetUniverseApi) {
+        return new InvestmentAssetPriceService(assetUniverseApi);
+    }
+
+    @Bean
+    public InvestmentPortfolioAllocationService investmentPortfolioAllocationService(AllocationsApi allocationsApi,
+        AssetUniverseApi assetUniverseApi, InvestmentApi investmentApi,
+        CustomIntegrationApiService customIntegrationApiService) {
+        return new InvestmentPortfolioAllocationService(allocationsApi, assetUniverseApi, investmentApi,
+            customIntegrationApiService);
+    }
+
     @Bean
     public InvestmentSaga investmentSaga(InvestmentClientService investmentClientService,
         InvestmentPortfolioService investmentPortfolioService,
         InvestmentModelPortfolioService investmentModelPortfolioService,
-        InvestmentAssetUniverseService investmentAssetUniverseService) {
-        return new InvestmentSaga(investmentClientService, investmentPortfolioService, investmentModelPortfolioService,
-            investmentAssetUniverseService);
+        InvestmentPortfolioAllocationService investmentPortfolioAllocationService, AsyncTaskService asyncTaskService,
+        InvestmentIngestionConfigurationProperties coreConfigurationProperties) {
+        return new InvestmentSaga(investmentClientService, investmentPortfolioService,
+            investmentPortfolioAllocationService, investmentModelPortfolioService, asyncTaskService,
+            coreConfigurationProperties);
+    }
+
+    @Bean
+    public InvestmentAssetUniversSaga investmentStaticDataSaga(
+        InvestmentAssetUniverseService investmentAssetUniverseService,
+        InvestmentAssetPriceService investmentAssetPriceService,
+        InvestmentIngestionConfigurationProperties coreConfigurationProperties) {
+        return new InvestmentAssetUniversSaga(investmentAssetUniverseService, investmentAssetPriceService,
+            coreConfigurationProperties);
     }
 
 }

stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentServiceConfiguration.java

@@ -5,7 +5,6 @@
 import com.backbase.investment.api.service.v1.model.StatusA10Enum;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import java.net.URI;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -29,8 +28,24 @@ public record Asset(
     List<String> categories,
     URI logo,
     String externalId,
-    String description
-) {
+    String description,
+    Double defaultPrice
+) implements AssetKey {
+
+    @Override
+    public String getIsin() {
+        return isin;
+    }
+
+    @Override
+    public String getMarket() {
+        return market;
+    }
+
+    @Override
+    public String getCurrency() {
+        return currency;
+    }
 
     /**
      * Creates a record from the generated API model.
@@ -54,7 +69,8 @@ public static Asset fromModel(com.backbase.investment.api.service.v1.model.Asset
             categories == null ? List.of() : categories.stream().map(AssetCategory::getCode).toList(),
             asset.getLogo(),
             asset.getExternalId(),
-            asset.getDescription()
+            asset.getDescription(),
+            100d
         );
     }
 

stream-investment/investment-core/src/main/java/com/backbase/stream/investment/Asset.java

@@ -0,0 +1,21 @@
+package com.backbase.stream.investment;
+
+import java.util.Map;
+
+public interface AssetKey {
+
+    String getIsin();
+
+    String getMarket();
+
+    String getCurrency();
+
+    default Map<String, Object> getAssetMap() {
+        return Map.of("isin", getIsin(), "market", getMarket(), "currency", getCurrency());
+    }
+
+    default String getKeyString() {
+        return getIsin() + "_" + getMarket() + "_" + getCurrency();
+    }
+
+}

stream-investment/investment-core/src/main/java/com/backbase/stream/investment/AssetKey.java

@@ -0,0 +1,21 @@
+package com.backbase.stream.investment;
+
+public record AssetPrice(String isin, String market, String currency, double price,
+                         RandomParam randomParam) implements AssetKey {
+
+    @Override
+    public String getIsin() {
+        return isin;
+    }
+
+    @Override
+    public String getMarket() {
+        return market;
+    }
+
+    @Override
+    public String getCurrency() {
+        return currency;
+    }
+
+}

stream-investment/investment-core/src/main/java/com/backbase/stream/investment/AssetPrice.java

@@ -0,0 +1,42 @@
+package com.backbase.stream.investment;
+
+import com.backbase.investment.api.service.v1.model.AssetCategory;
+import com.backbase.investment.api.service.v1.model.AssetCategoryType;
+import com.backbase.investment.api.service.v1.model.GroupResult;
+import com.backbase.investment.api.service.v1.model.Market;
+import com.backbase.investment.api.service.v1.model.MarketSpecialDay;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@EqualsAndHashCode
+@Data
+@Builder
+public class InvestmentAssetData {
+
+    private List<Market> markets;
+    private List<MarketSpecialDay> marketSpecialDays;
+    private List<AssetCategoryType> assetCategoryTypes;
+    private List<AssetCategory> assetCategories;
+    private List<Asset> assets;
+    private List<AssetPrice> assetPrices;
+    private List<GroupResult> priceAsyncTasks;
+
+    public Map<String, AssetPrice> getPriceByAsset() {
+        return Objects.requireNonNullElse(assetPrices, List.<AssetPrice>of()).stream()
+            .collect(Collectors.toMap(AssetKey::getKeyString, Function.identity()));
+    }
+
+    public Map<UUID, Asset> getAssetByUuid() {
+        return Objects.requireNonNullElse(assets, List.<Asset>of()).stream()
+            .collect(Collectors.toMap(Asset::uuid, Function.identity()));
+
+    }
+
+}

stream-investment/investment-core/src/main/java/com/backbase/stream/investment/InvestmentAssetData.java

@@ -0,0 +1,54 @@
+package com.backbase.stream.investment;
+
+import com.backbase.investment.api.service.v1.model.AssetCategory;
+import com.backbase.investment.api.service.v1.model.AssetCategoryType;
+import com.backbase.investment.api.service.v1.model.GroupResult;
+import com.backbase.investment.api.service.v1.model.Market;
+import com.backbase.investment.api.service.v1.model.MarketSpecialDay;
+import com.backbase.stream.worker.model.StreamTask;
+import java.util.List;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@EqualsAndHashCode(callSuper = true)
+@Data
+public class InvestmentAssetsTask extends StreamTask {
+
+    private final InvestmentAssetData data;
+
+    public InvestmentAssetsTask(String unitOfWorkId, InvestmentAssetData data) {
+        super(unitOfWorkId);
+        this.data = data;
+    }
+
+    @Override
+    public String getName() {
+        return "investment";
+    }
+
+    public void setMarkets(List<Market> markets) {
+        data.setMarkets(markets);
+    }
+
+    public void setMarketSpecialDays(List<MarketSpecialDay> marketSpecialDays) {
+        data.setMarketSpecialDays(marketSpecialDays);
+    }
+
+    public void setAssetCategoryTypes(List<AssetCategoryType> assetCategoryTypes) {
+        data.setAssetCategoryTypes(assetCategoryTypes);
+    }
+
+    public void setAssetCategories(List<AssetCategory> assetCategories) {
+        data.setAssetCategories(assetCategories);
+    }
+
+    public void setAssets(List<Asset> assets) {
+        data.setAssets(assets);
+    }
+
+    public InvestmentAssetsTask setPriceTasks(List<GroupResult> tasks) {
+        data.setPriceAsyncTasks(tasks);
+        return this;
+    }
+
+}