Generating Intraday Prices for the Assets

pawana_backbase · pawana_backbase · commit b91f0b19cecf · 2026-01-20T22:00:03.000+05:30
diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentServiceConfiguration.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentServiceConfiguration.java
@@ -17,6 +17,7 @@
 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.InvestmentIntradayAssetPriceService;
 import com.backbase.stream.investment.service.InvestmentModelPortfolioService;
 import com.backbase.stream.investment.service.InvestmentPortfolioAllocationService;
 import com.backbase.stream.investment.service.InvestmentPortfolioService;
@@ -73,6 +74,11 @@ public InvestmentAssetPriceService investmentAssetPriceService(AssetUniverseApi
         return new InvestmentAssetPriceService(assetUniverseApi);
     }
 
+    @Bean
+    public InvestmentIntradayAssetPriceService investmentIntradayAssetPriceService(AssetUniverseApi assetUniverseApi) {
+        return new InvestmentIntradayAssetPriceService(assetUniverseApi);
+    }
+
     @Bean
     public InvestmentPortfolioAllocationService investmentPortfolioAllocationService(AllocationsApi allocationsApi,
         AssetUniverseApi assetUniverseApi, InvestmentApi investmentApi,
@@ -96,9 +102,10 @@ public InvestmentSaga investmentSaga(InvestmentClientService investmentClientSer
     public InvestmentAssetUniversSaga investmentStaticDataSaga(
         InvestmentAssetUniverseService investmentAssetUniverseService,
         InvestmentAssetPriceService investmentAssetPriceService,
+        InvestmentIntradayAssetPriceService investmentIntradayAssetPriceService,
         InvestmentIngestionConfigurationProperties coreConfigurationProperties) {
         return new InvestmentAssetUniversSaga(investmentAssetUniverseService, investmentAssetPriceService,
-            coreConfigurationProperties);
+            investmentIntradayAssetPriceService, coreConfigurationProperties);
     }
 
 }
diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/AssetLatestPrice.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/AssetLatestPrice.java
@@ -0,0 +1,7 @@
+package com.backbase.stream.investment;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.UUID;
+
+public record AssetLatestPrice(@JsonProperty("uuid") UUID uuid,
+                               @JsonProperty("latest_price") LatestPrice latestPrice) {}
diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/InvestmentAssetData.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/InvestmentAssetData.java
@@ -27,6 +27,7 @@ public class InvestmentAssetData {
     private List<Asset> assets;
     private List<AssetPrice> assetPrices;
     private List<GroupResult> priceAsyncTasks;
+    private List<GroupResult> intradayPriceAsyncTasks;
 
     public Map<String, AssetPrice> getPriceByAsset() {
         return Objects.requireNonNullElse(assetPrices, List.<AssetPrice>of()).stream()
diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/InvestmentAssetsTask.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/InvestmentAssetsTask.java
@@ -51,4 +51,8 @@ public InvestmentAssetsTask setPriceTasks(List<GroupResult> tasks) {
         return this;
     }
 
+    public InvestmentAssetsTask setIntradayPriceTasks(List<GroupResult> tasks) {
+        data.setIntradayPriceAsyncTasks(tasks);
+        return this;
+    }
 }
diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/LatestPrice.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/LatestPrice.java
@@ -0,0 +1,15 @@
+package com.backbase.stream.investment;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.time.OffsetDateTime;
+
+public record LatestPrice(
+    @JsonProperty("amount") Double amount,
+    @JsonProperty("datetime") OffsetDateTime datetime,
+    @JsonProperty("open_price") Double openPrice,
+    @JsonProperty("high_price") Double highPrice,
+    @JsonProperty("low_price") Double lowPrice,
+    @JsonProperty("previous_close_price") Double previousClosePrice
+) {
+
+}
diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/PaginatedExpandedAssetList.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/PaginatedExpandedAssetList.java
@@ -0,0 +1,27 @@
+package com.backbase.stream.investment;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@EqualsAndHashCode
+@Data
+@Builder
+public class PaginatedExpandedAssetList {
+
+    public static final String JSON_PROPERTY_COUNT = "count";
+    private Integer count;
+
+    public static final String JSON_PROPERTY_NEXT = "next";
+    private URI next;
+
+    public static final String JSON_PROPERTY_PREVIOUS = "previous";
+    private URI previous;
+
+    public static final String JSON_PROPERTY_RESULTS = "results";
+    private List<AssetLatestPrice> results = new ArrayList<>();
+
+}
diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/saga/InvestmentAssetUniversSaga.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/saga/InvestmentAssetUniversSaga.java
@@ -10,6 +10,7 @@
 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.InvestmentIntradayAssetPriceService;
 import com.backbase.stream.investment.service.InvestmentPortfolioService;
 import com.backbase.stream.worker.StreamTaskExecutor;
 import com.backbase.stream.worker.model.StreamTask;
@@ -59,32 +60,34 @@ public class InvestmentAssetUniversSaga implements StreamTaskExecutor<Investment
 
     private final InvestmentAssetUniverseService assetUniverseService;
     private final InvestmentAssetPriceService investmentAssetPriceService;
+    private final InvestmentIntradayAssetPriceService investmentIntradayAssetPriceService;
     private final InvestmentIngestionConfigurationProperties coreConfigurationProperties;
 
     @Override
     public Mono<InvestmentAssetsTask> executeTask(InvestmentAssetsTask streamTask) {
         if (!coreConfigurationProperties.isAssetUniversEnabled()) {
-            log.warn("Skip investment asset univers saga execution: taskId={}, taskName={}",
+            log.warn("Skip investment asset universe saga execution: taskId={}, taskName={}",
                 streamTask.getId(), streamTask.getName());
             return Mono.just(streamTask);
         }
-        log.info("Starting investment saga execution: taskId={}, taskName={}",
+        log.info("Starting investment asset universe saga execution: taskId={}, taskName={}",
             streamTask.getId(), streamTask.getName());
-        return createMarkets(streamTask)
-            .flatMap(this::createMarketSpecialDays)
-            .flatMap(this::createAssetCategoryTypes)
-            .flatMap(this::createAssetCategories)
+        return upsertMarkets(streamTask)
+            .flatMap(this::upsertMarketSpecialDays)
+            .flatMap(this::upsertAssetCategoryTypes)
+            .flatMap(this::upsertAssetCategories)
             .flatMap(this::createAssets)
             .flatMap(this::upsertPrices)
+            .flatMap(this::upsertIntradayPrices)
             .doOnSuccess(completedTask -> log.info(
-                "Successfully completed investment saga: taskId={}, taskName={}, state={}",
+                "Successfully completed investment asset universe saga: taskId={}, taskName={}, state={}",
                 completedTask.getId(), completedTask.getName(), completedTask.getState()))
             .doOnError(throwable -> {
-                log.error("Failed to execute investment saga: taskId={}, taskName={}",
+                log.error("Failed to execute investment asset universe saga: taskId={}, taskName={}",
                     streamTask.getId(), streamTask.getName(), throwable);
                 streamTask.error(INVESTMENT, OP_UPSERT, RESULT_FAILED,
                     streamTask.getName(), streamTask.getId(),
-                    "Investment saga failed: " + throwable.getMessage());
+                    "Investment asset universe saga failed: " + throwable.getMessage());
                 streamTask.setState(State.FAILED);
             })
             .onErrorResume(throwable -> Mono.just(streamTask));
@@ -96,6 +99,11 @@ private Mono<InvestmentAssetsTask> upsertPrices(InvestmentAssetsTask investmentT
             .map(investmentTask::setPriceTasks);
     }
 
+    private Mono<InvestmentAssetsTask> upsertIntradayPrices(InvestmentAssetsTask investmentTask) {
+        return investmentIntradayAssetPriceService.ingestIntradayPrices()
+            .map(investmentTask::setIntradayPriceTasks);
+    }
+
     /**
      * Rollback is not implemented for investment saga.
      *
@@ -112,7 +120,7 @@ public Mono<InvestmentAssetsTask> rollBack(InvestmentAssetsTask streamTask) {
         return Mono.empty();
     }
 
-    public Mono<InvestmentAssetsTask> createMarkets(InvestmentAssetsTask investmentTask) {
+    public Mono<InvestmentAssetsTask> upsertMarkets(InvestmentAssetsTask investmentTask) {
         InvestmentAssetData investmentData = investmentTask.getData();
         int marketCount = investmentData.getMarkets() != null ? investmentData.getMarkets().size() : 0;
         log.info("Starting investment market creation: taskId={}, marketCount={}",
@@ -161,15 +169,15 @@ public Mono<InvestmentAssetsTask> createMarkets(InvestmentAssetsTask investmentT
     }
 
     /**
-     * Creates or upserts market special days for the investment task.
+     * Upserts market special days for the investment task.
      *
      * <p>This method processes each market special day in the task data by invoking
      * the asset universe service. It updates the task state and logs progress for observability.
      *
      * @param investmentTask the investment task containing market special day data
      * @return Mono emitting the updated investment task with market special days set
      */
-    public Mono<InvestmentAssetsTask> createMarketSpecialDays(InvestmentAssetsTask investmentTask) {
+    public Mono<InvestmentAssetsTask> upsertMarketSpecialDays(InvestmentAssetsTask investmentTask) {
         InvestmentAssetData investmentData = investmentTask.getData();
         int marketSpecialDayCount =
             investmentData.getMarketSpecialDays() != null ? investmentData.getMarketSpecialDays().size() : 0;
@@ -219,15 +227,15 @@ public Mono<InvestmentAssetsTask> createMarketSpecialDays(InvestmentAssetsTask i
     }
 
     /**
-     * Creates or upserts asset categories for the investment task.
+     * Upserts asset categories for the investment task.
      *
      * <p>This method processes each asset category in the task data by invoking
      * the asset universe service. It updates the task state and logs progress for observability.
      *
      * @param investmentTask the investment task containing asset category data
      * @return Mono emitting the updated investment task with asset categories set
      */
-    public Mono<InvestmentAssetsTask> createAssetCategories(InvestmentAssetsTask investmentTask) {
+    public Mono<InvestmentAssetsTask> upsertAssetCategories(InvestmentAssetsTask investmentTask) {
         InvestmentAssetData investmentData = investmentTask.getData();
         int categoryCount =
             investmentData.getAssetCategories() != null ? investmentData.getAssetCategories().size() : 0;
@@ -274,7 +282,7 @@ public Mono<InvestmentAssetsTask> createAssetCategories(InvestmentAssetsTask inv
             });
     }
 
-    public Mono<InvestmentAssetsTask> createAssetCategoryTypes(InvestmentAssetsTask investmentTask) {
+    public Mono<InvestmentAssetsTask> upsertAssetCategoryTypes(InvestmentAssetsTask investmentTask) {
         InvestmentAssetData investmentData = investmentTask.getData();
         int typeCount =
             investmentData.getAssetCategoryTypes() != null ? investmentData.getAssetCategoryTypes().size() : 0;
diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentAssetUniverseService.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentAssetUniverseService.java
@@ -59,7 +59,17 @@ public Mono<Market> getOrCreateMarket(MarketRequest marketRequest) {
             .flatMap(existingMarket -> {
                 log.info("Market already exists: {}", existingMarket.getCode());
                 log.debug("Market already exists: {}", existingMarket);
-                return Mono.just(existingMarket);
+                return assetUniverseApi.updateMarket(existingMarket.getCode(), marketRequest)
+                    .doOnSuccess(updatedMarket -> log.info("Updated market: {}", updatedMarket))
+                    .doOnError(error -> {
+                        if (error instanceof WebClientResponseException w) {
+                            log.error("Error updating market: {} : HTTP {} -> {}", marketRequest.getCode(),
+                                w.getStatusCode(), w.getResponseBodyAsString());
+                        } else {
+                            log.error("Error updating market: {} : {}", marketRequest.getCode(),
+                                error.getMessage(), error);
+                        }
+                    });
             })
             // If Mono is empty (market not found), create the market
             .switchIfEmpty(assetUniverseApi.createMarket(marketRequest)
@@ -141,7 +151,20 @@ public Mono<MarketSpecialDay> getOrCreateMarketSpecialDay(MarketSpecialDayReques
                         .findFirst();
                     if (matchingSpecialDay.isPresent()) {
                         log.info("Market special day already exists for day: {}", marketSpecialDayRequest);
-                        return Mono.just(matchingSpecialDay.get());
+                        return assetUniverseApi.updateMarketSpecialDay(matchingSpecialDay.get().getUuid().toString(),
+                                marketSpecialDayRequest)
+                            .doOnSuccess(updatedMarketSpecialDay ->
+                                log.info("Updated market special day: {}", updatedMarketSpecialDay))
+                            .doOnError(error -> {
+                                if (error instanceof WebClientResponseException w) {
+                                    log.error("Error updating market special day : {} : HTTP {} -> {}",
+                                        marketSpecialDayRequest,
+                                        w.getStatusCode(), w.getResponseBodyAsString());
+                                } else {
+                                    log.error("Error updating market special day {} : {}", marketSpecialDayRequest,
+                                        error.getMessage(), error);
+                                }
+                            });
                     } else {
                         log.debug("No market special day exists for day: {}", marketSpecialDayRequest);
                         return Mono.empty();
@@ -212,7 +235,18 @@ public Mono<AssetCategory> createAssetCategory(AssetCategoryRequest assetCategor
                         .findFirst();
                     if (matchingCategory.isPresent()) {
                         log.info("Asset category already exists for code: {}", assetCategoryRequest.getCode());
-                        return Mono.just(matchingCategory.get());
+                        return assetUniverseApi.updateAssetCategory(matchingCategory.get().getUuid().toString(), assetCategoryRequest)
+                            .doOnSuccess(updatedCategory -> log.info("Updated asset category: {}", updatedCategory))
+                            .doOnError(error -> {
+                                if (error instanceof WebClientResponseException w) {
+                                    log.error("Error updating asset category: {} : HTTP {} -> {}", assetCategoryRequest.getCode(),
+                                        w.getStatusCode(), w.getResponseBodyAsString());
+                                } else {
+                                    log.error("Error updating asset category: {} : {}", assetCategoryRequest.getCode(),
+                                        error.getMessage(), error);
+                                }
+                            })
+                            .onErrorResume(e -> Mono.empty());
                     } else {
                         log.debug("No asset category exists for code: {}", assetCategoryRequest.getCode());
                         return Mono.empty();
@@ -232,6 +266,7 @@ public Mono<AssetCategory> createAssetCategory(AssetCategoryRequest assetCategor
                                 error.getMessage(), error);
                         }
                     })
+                    .onErrorResume(e -> Mono.empty())
             );
     }
 
@@ -259,7 +294,18 @@ public Mono<AssetCategoryType> createAssetCategoryType(AssetCategoryTypeRequest
                         .findFirst();
                     if (matchingType.isPresent()) {
                         log.info("Asset category type already exists for code: {}", assetCategoryTypeRequest.getCode());
-                        return Mono.just(matchingType.get());
+                        return assetUniverseApi.updateAssetCategoryType(matchingType.get().getUuid().toString(), assetCategoryTypeRequest)
+                            .doOnSuccess(updatedType -> log.info("Updated asset category type: {}", updatedType))
+                            .doOnError(error -> {
+                                if (error instanceof WebClientResponseException w) {
+                                    log.error("Error updating asset category type: {} : HTTP {} -> {}",
+                                        assetCategoryTypeRequest.getCode(), w.getStatusCode(), w.getResponseBodyAsString());
+                                } else {
+                                    log.error("Error updating asset category type: {} : {}",
+                                        assetCategoryTypeRequest.getCode(), error.getMessage(), error);
+                                }
+                            })
+                            .onErrorResume(e -> Mono.empty());
                     } else {
                         log.debug("No asset category type exists for code: {}", assetCategoryTypeRequest.getCode());
                         return Mono.empty();
diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentIntradayAssetPriceService.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentIntradayAssetPriceService.java
diff --git a/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/saga/InvestmentAssetUniversSagaTest.java b/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/saga/InvestmentAssetUniversSagaTest.java
diff --git a/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentIntradayAssetPriceServiceTest.java b/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentIntradayAssetPriceServiceTest.java

Original file line number	Diff line number	Diff line change
`@@ -51,4 +51,8 @@ public InvestmentAssetsTask setPriceTasks(List<GroupResult> tasks) {`
`51`	`51`	`return this;`
`52`	`52`	`}`
`53`	`53`
	`54`	`+ public InvestmentAssetsTask setIntradayPriceTasks(List<GroupResult> tasks) {`
	`55`	`+ data.setIntradayPriceAsyncTasks(tasks);`
	`56`	`+ return this;`
	`57`	`+ }`
`54`	`58`	`}`