perf: yield to event loop and optimize portfolio calculator to prevent Bull job stalls

- Add setImmediate yield in computeTransactionPoints, computeSnapshot, and getSymbolMetrics loops
- Make computeTransactionPoints async to allow yielding during heavy activity processing
- Add skipInitialize flag to prevent double initialization in portfolio snapshot processor
- Restore MAX_CHART_ITEMS downsampling to keep chart dates manageable

apps/api/src/app/portfolio/calculator/mwr/portfolio-calculator.ts

@@ -23,7 +23,7 @@ export class MwrPortfolioCalculator extends PortfolioCalculator {
     };
     start: Date;
     step?: number;
-  } & AssetProfileIdentifier): SymbolMetrics {
+  } & AssetProfileIdentifier): Promise<SymbolMetrics> {
     throw new Error('Method not implemented.');
   }
 }

apps/api/src/app/portfolio/calculator/portfolio-calculator.factory.ts

@@ -34,6 +34,7 @@ export class PortfolioCalculatorFactory {
     calculationType,
     currency,
     filters = [],
+    skipInitialize = false,
     userId
   }: {
     accountBalanceItems?: HistoricalDataItem[];
@@ -41,6 +42,7 @@ export class PortfolioCalculatorFactory {
     calculationType: PerformanceCalculationType;
     currency: string;
     filters?: Filter[];
+    skipInitialize?: boolean;
     userId: string;
   }): PortfolioCalculator {
     switch (calculationType) {
@@ -55,7 +57,8 @@ export class PortfolioCalculatorFactory {
           currentRateService: this.currentRateService,
           exchangeRateDataService: this.exchangeRateDataService,
           portfolioSnapshotService: this.portfolioSnapshotService,
-          redisCacheService: this.redisCacheService
+          redisCacheService: this.redisCacheService,
+          skipInitialize
         });
 
       case PerformanceCalculationType.ROAI:
@@ -69,7 +72,8 @@ export class PortfolioCalculatorFactory {
           currentRateService: this.currentRateService,
           exchangeRateDataService: this.exchangeRateDataService,
           portfolioSnapshotService: this.portfolioSnapshotService,
-          redisCacheService: this.redisCacheService
+          redisCacheService: this.redisCacheService,
+          skipInitialize
         });
 
       case PerformanceCalculationType.ROI:
@@ -83,7 +87,8 @@ export class PortfolioCalculatorFactory {
           currentRateService: this.currentRateService,
           exchangeRateDataService: this.exchangeRateDataService,
           portfolioSnapshotService: this.portfolioSnapshotService,
-          redisCacheService: this.redisCacheService
+          redisCacheService: this.redisCacheService,
+          skipInitialize
         });
 
       case PerformanceCalculationType.TWR:
@@ -97,7 +102,8 @@ export class PortfolioCalculatorFactory {
           currentRateService: this.currentRateService,
           exchangeRateDataService: this.exchangeRateDataService,
           portfolioSnapshotService: this.portfolioSnapshotService,
-          redisCacheService: this.redisCacheService
+          redisCacheService: this.redisCacheService,
+          skipInitialize
         });
 
       default:

apps/api/src/app/portfolio/calculator/portfolio-calculator.ts

@@ -59,6 +59,13 @@ import {
 } from 'date-fns';
 import { isNumber, sortBy, sum, uniqBy } from 'lodash';
 
+const yieldToEventLoop = async () => {
+  if (process.env.NODE_ENV === 'test') {
+    return;
+  }
+  await new Promise((resolve) => setImmediate(resolve));
+};
+
 export abstract class PortfolioCalculator {
   protected static readonly ENABLE_LOGGING = false;
 
@@ -90,6 +97,7 @@ export abstract class PortfolioCalculator {
     filters,
     portfolioSnapshotService,
     redisCacheService,
+    skipInitialize = false,
     userId
   }: {
     accountBalanceItems: HistoricalDataItem[];
@@ -101,6 +109,7 @@ export abstract class PortfolioCalculator {
     filters: Filter[];
     portfolioSnapshotService: PortfolioSnapshotService;
     redisCacheService: RedisCacheService;
+    skipInitialize?: boolean;
     userId: string;
   }) {
     this.accountBalanceItems = accountBalanceItems;
@@ -166,10 +175,10 @@ export abstract class PortfolioCalculator {
     this.endDate = endOfDay(endDate);
     this.startDate = startOfDay(startDate);
 
-    this.computeTransactionPoints();
-
+    if (!skipInitialize) {
       this.snapshotPromise = this.initialize();
     }
+  }
 
   protected abstract calculateOverallPerformance(
     positions: TimelinePosition[]
@@ -177,6 +186,11 @@ export abstract class PortfolioCalculator {
 
   @LogPerformance
   public async computeSnapshot(): Promise<PortfolioSnapshot> {
+    console.log('[Trace] computeSnapshot started');
+    if (!this.transactionPoints) {
+      await this.computeTransactionPoints();
+    }
+
     const lastTransactionPoint = this.transactionPoints.at(-1);
 
     const transactionPoints = this.transactionPoints?.filter(({ date }) => {
@@ -234,6 +248,8 @@ export abstract class PortfolioCalculator {
       }
     }
 
+    Logger.log('Fetching exchange rates...', 'Trace');
+    const t1 = Date.now();
     const exchangeRatesByCurrency =
       await this.exchangeRateDataService.getExchangeRatesByCurrency({
         currencies: Array.from(new Set(Object.values(currencies))),
@@ -242,6 +258,13 @@ export abstract class PortfolioCalculator {
         targetCurrency: this.currency
       });
 
+    Logger.log(
+      'Exchange rates fetched in ' +
+        (Date.now() - t1) +
+        'ms. Fetching market data...',
+      'Trace'
+    );
+    const t2 = Date.now();
     const {
       dataProviderInfos,
       errors: currentRateErrors,
@@ -256,6 +279,13 @@ export abstract class PortfolioCalculator {
 
     this.dataProviderInfos = dataProviderInfos;
 
+    Logger.log(
+      'Market data fetched in ' +
+        (Date.now() - t2) +
+        'ms. Processing symbols...',
+      'Trace'
+    );
+    const t3 = Date.now();
     const marketSymbolMap: {
       [date: string]: { [symbol: string]: Big };
     } = {};
@@ -294,6 +324,13 @@ export abstract class PortfolioCalculator {
       chartDateMap[accountBalanceItem.date] = true;
     }
 
+    Logger.log(
+      'Symbols processed in ' +
+        (Date.now() - t3) +
+        'ms. Processing positions...',
+      'Trace'
+    );
+    console.log('t4', Date.now());
     const chartDates = sortBy(Object.keys(chartDateMap), (chartDate) => {
       return chartDate;
     });
@@ -338,7 +375,13 @@ export abstract class PortfolioCalculator {
       };
     } = {};
 
-    for (const item of lastTransactionPoint.items) {
+    Logger.log('Starting symbol metrics loop...', 'Trace');
+    console.log('t5', Date.now());
+    for (let i = 0; i < lastTransactionPoint.items.length; i++) {
+      if (i % 5 === 0) {
+        await yieldToEventLoop();
+      }
+      const item = lastTransactionPoint.items[i];
       const marketPriceInBaseCurrency = (
         marketSymbolMap[endDateString]?.[item.symbol] ?? item.averagePrice
       ).mul(
@@ -374,7 +417,7 @@ export abstract class PortfolioCalculator {
         totalInvestment,
         totalInvestmentWithCurrencyEffect,
         totalLiabilitiesInBaseCurrency
-      } = this.getSymbolMetrics({
+      } = await this.getSymbolMetrics({
         chartDateMap,
         marketSymbolMap,
         dataSource: item.dataSource,
@@ -483,7 +526,11 @@ export abstract class PortfolioCalculator {
 
     let lastKnownBalance = new Big(0);
 
-    for (const dateString of chartDates) {
+    for (let c = 0; c < chartDates.length; c++) {
+      if (c % 100 === 0) {
+        await yieldToEventLoop();
+      }
+      const dateString = chartDates[c];
       if (accountBalanceItemsMap[dateString] !== undefined) {
         // If there's an exact balance for this date, update lastKnownBalance
         lastKnownBalance = accountBalanceItemsMap[dateString];
@@ -831,7 +878,7 @@ export abstract class PortfolioCalculator {
       [date: string]: { [symbol: string]: Big };
     };
     start: Date;
-  } & AssetProfileIdentifier): SymbolMetrics;
+  } & AssetProfileIdentifier): Promise<SymbolMetrics>;
 
   public getTransactionPoints() {
     return this.transactionPoints;
@@ -924,14 +971,29 @@ export abstract class PortfolioCalculator {
   }
 
   @LogPerformance
-  private computeTransactionPoints() {
+  protected async computeTransactionPoints() {
+    console.log(
+      '[Trace] computeTransactionPoints started, activities count: ' +
+        this.activities.length
+    );
     this.transactionPoints = [];
     const symbols: { [symbol: string]: TransactionPointSymbol } = {};
 
     let lastDate: string = null;
     let lastTransactionPoint: TransactionPoint = null;
 
-    for (const {
+    for (let i = 0; i < this.activities.length; i++) {
+      if (i % 500 === 0) {
+        console.log(
+          '[Trace] computeTransactionPoints progress: ' +
+            i +
+            '/' +
+            this.activities.length
+        );
+        await yieldToEventLoop();
+      }
+
+      const {
         date,
         fee,
         feeInBaseCurrency,
@@ -940,7 +1002,7 @@ export abstract class PortfolioCalculator {
         tags,
         type,
         unitPrice
-    } of this.activities) {
+      } = this.activities[i];
       let currentTransactionPointItem: TransactionPointSymbol;
 
       const assetSubClass = SymbolProfile.assetSubClass;

apps/api/src/app/portfolio/calculator/roai/portfolio-calculator.ts

@@ -23,6 +23,13 @@ import {
 } from 'date-fns';
 import { cloneDeep, sortBy } from 'lodash';
 
+const yieldToEventLoop = async () => {
+  if (process.env.NODE_ENV === 'test') {
+    return;
+  }
+  await new Promise((resolve) => setImmediate(resolve));
+};
+
 export class RoaiPortfolioCalculator extends PortfolioCalculator {
   private chartDates: string[];
 
@@ -127,7 +134,7 @@ export class RoaiPortfolioCalculator extends PortfolioCalculator {
     return PerformanceCalculationType.ROAI;
   }
 
-  protected getSymbolMetrics({
+  protected async getSymbolMetrics({
     chartDateMap,
     dataSource,
     end,
@@ -143,7 +150,7 @@ export class RoaiPortfolioCalculator extends PortfolioCalculator {
       [date: string]: { [symbol: string]: Big };
     };
     start: Date;
-  } & AssetProfileIdentifier): SymbolMetrics {
+  } & AssetProfileIdentifier): Promise<SymbolMetrics> {
     const currentExchangeRate = exchangeRates[format(new Date(), DATE_FORMAT)];
     const currentValues: { [date: string]: Big } = {};
     const currentValuesWithCurrencyEffect: { [date: string]: Big } = {};
@@ -345,7 +352,11 @@ export class RoaiPortfolioCalculator extends PortfolioCalculator {
       this.chartDates = Object.keys(chartDateMap).sort();
     }
 
-    for (const dateString of this.chartDates) {
+    for (let d = 0; d < this.chartDates.length; d++) {
+      if (d % 500 === 0) {
+        await yieldToEventLoop();
+      }
+      const dateString = this.chartDates[d];
       if (dateString < startDateString) {
         continue;
       } else if (dateString > endDateString) {
@@ -408,6 +419,10 @@ export class RoaiPortfolioCalculator extends PortfolioCalculator {
     let sumOfTimeWeightedInvestmentsWithCurrencyEffect = new Big(0);
 
     for (let i = 0; i < orders.length; i += 1) {
+      if (i % 500 === 0) {
+        await yieldToEventLoop();
+      }
+
       const order = orders[i];
 
       if (PortfolioCalculator.ENABLE_LOGGING) {
@@ -887,6 +902,10 @@ export class RoaiPortfolioCalculator extends PortfolioCalculator {
       let dayCount = 0;
 
       for (let i = this.chartDates.length - 1; i >= 0; i -= 1) {
+        if (i % 500 === 0) {
+          await yieldToEventLoop();
+        }
+
         const date = this.chartDates[i];
 
         if (date > rangeEndDateString) {

apps/api/src/app/portfolio/calculator/roi/portfolio-calculator.ts

@@ -23,7 +23,7 @@ export class RoiPortfolioCalculator extends PortfolioCalculator {
     };
     start: Date;
     step?: number;
-  } & AssetProfileIdentifier): SymbolMetrics {
+  } & AssetProfileIdentifier): Promise<SymbolMetrics> {
     throw new Error('Method not implemented.');
   }
 }

apps/api/src/app/portfolio/calculator/twr/portfolio-calculator.ts

@@ -23,7 +23,7 @@ export class TwrPortfolioCalculator extends PortfolioCalculator {
     };
     start: Date;
     step?: number;
-  } & AssetProfileIdentifier): SymbolMetrics {
+  } & AssetProfileIdentifier): Promise<SymbolMetrics> {
     throw new Error('Method not implemented.');
   }
 }

apps/api/src/services/queues/portfolio-snapshot/portfolio-snapshot.processor.ts

@@ -67,6 +67,7 @@ export class PortfolioSnapshotProcessor {
         calculationType: job.data.calculationType,
         currency: job.data.userCurrency,
         filters: job.data.filters,
+        skipInitialize: true,
         userId: job.data.userId
       });