Feature/optimize 7d data gathering by prioritization (#3575)

* Optimize 7d data gathering by prioritization

* Update changelog

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Changed
+
+- Optimized the 7d data gathering by prioritizing the currencies
+
 ### Fixed
 
 - Fixed the table sorting of the holdings tab on the home page

apps/api/src/app/admin/admin.controller.ts

@@ -81,10 +81,11 @@ export class AdminController {
   @Post('gather/max')
   @UseGuards(AuthGuard('jwt'), HasPermissionGuard)
   public async gatherMax(): Promise<void> {
-    const uniqueAssets = await this.dataGatheringService.getUniqueAssets();
+    const assetProfileIdentifiers =
+      await this.dataGatheringService.getAllAssetProfileIdentifiers();
 
     await this.dataGatheringService.addJobsToQueue(
-      uniqueAssets.map(({ dataSource, symbol }) => {
+      assetProfileIdentifiers.map(({ dataSource, symbol }) => {
         return {
           data: {
             dataSource,
@@ -107,10 +108,11 @@ export class AdminController {
   @Post('gather/profile-data')
   @UseGuards(AuthGuard('jwt'), HasPermissionGuard)
   public async gatherProfileData(): Promise<void> {
-    const uniqueAssets = await this.dataGatheringService.getUniqueAssets();
+    const assetProfileIdentifiers =
+      await this.dataGatheringService.getAllAssetProfileIdentifiers();
 
     await this.dataGatheringService.addJobsToQueue(
-      uniqueAssets.map(({ dataSource, symbol }) => {
+      assetProfileIdentifiers.map(({ dataSource, symbol }) => {
         return {
           data: {
             dataSource,

apps/api/src/services/cron.service.ts

@@ -45,10 +45,11 @@ export class CronService {
   @Cron(CronService.EVERY_SUNDAY_AT_LUNCH_TIME)
   public async runEverySundayAtTwelvePm() {
     if (await this.isDataGatheringEnabled()) {
-      const uniqueAssets = await this.dataGatheringService.getUniqueAssets();
+      const assetProfileIdentifiers =
+        await this.dataGatheringService.getAllAssetProfileIdentifiers();
 
       await this.dataGatheringService.addJobsToQueue(
-        uniqueAssets.map(({ dataSource, symbol }) => {
+        assetProfileIdentifiers.map(({ dataSource, symbol }) => {
           return {
             data: {
               dataSource,

apps/api/src/services/data-gathering/data-gathering.service.ts

@@ -10,6 +10,7 @@ import {
   DATA_GATHERING_QUEUE,
   DATA_GATHERING_QUEUE_PRIORITY_HIGH,
   DATA_GATHERING_QUEUE_PRIORITY_LOW,
+  DATA_GATHERING_QUEUE_PRIORITY_MEDIUM,
   GATHER_HISTORICAL_MARKET_DATA_PROCESS,
   GATHER_HISTORICAL_MARKET_DATA_PROCESS_OPTIONS,
   PROPERTY_BENCHMARKS
@@ -62,9 +63,22 @@ export class DataGatheringService {
   }
 
   public async gather7Days() {
-    const dataGatheringItems = await this.getSymbols7D();
     await this.gatherSymbols({
-      dataGatheringItems,
+      dataGatheringItems: await this.getCurrencies7D(),
+      priority: DATA_GATHERING_QUEUE_PRIORITY_HIGH
+    });
+
+    await this.gatherSymbols({
+      dataGatheringItems: await this.getSymbols7D({
+        withUserSubscription: true
+      }),
+      priority: DATA_GATHERING_QUEUE_PRIORITY_MEDIUM
+    });
+
+    await this.gatherSymbols({
+      dataGatheringItems: await this.getSymbols7D({
+        withUserSubscription: false
+      }),
       priority: DATA_GATHERING_QUEUE_PRIORITY_LOW
     });
   }
@@ -138,7 +152,7 @@ export class DataGatheringService {
     });
 
     if (!uniqueAssets) {
-      uniqueAssets = await this.getUniqueAssets();
+      uniqueAssets = await this.getAllAssetProfileIdentifiers();
     }
 
     if (uniqueAssets.length <= 0) {
@@ -270,7 +284,7 @@ export class DataGatheringService {
     );
   }
 
-  public async getUniqueAssets(): Promise<UniqueAsset[]> {
+  public async getAllAssetProfileIdentifiers(): Promise<UniqueAsset[]> {
     const symbolProfiles = await this.prismaService.symbolProfile.findMany({
       orderBy: [{ symbol: 'asc' }]
     });
@@ -290,73 +304,83 @@ export class DataGatheringService {
       });
   }
 
-  private getEarliestDate(aStartDate: Date) {
-    return min([aStartDate, subYears(new Date(), 10)]);
-  }
-
-  private async getSymbols7D(): Promise<IDataGatheringItem[]> {
-    const startDate = subDays(resetHours(new Date()), 7);
-
-    const symbolProfiles = await this.prismaService.symbolProfile.findMany({
-      orderBy: [{ symbol: 'asc' }],
-      select: {
-        dataSource: true,
-        scraperConfiguration: true,
-        symbol: true
-      }
-    });
-
-    // Only consider symbols with incomplete market data for the last
-    // 7 days
-    const symbolsWithCompleteMarketData = (
+  private async getAssetProfileIdentifiersWithCompleteMarketData(): Promise<
+    UniqueAsset[]
+  > {
+    return (
       await this.prismaService.marketData.groupBy({
         _count: true,
-        by: ['symbol'],
+        by: ['dataSource', 'symbol'],
         orderBy: [{ symbol: 'asc' }],
         where: {
-          date: { gt: startDate },
+          date: { gt: subDays(resetHours(new Date()), 7) },
           state: 'CLOSE'
         }
       })
     )
-      .filter((group) => {
-        return group._count >= 6;
+      .filter(({ _count }) => {
+        return _count >= 6;
       })
-      .map((group) => {
-        return group.symbol;
+      .map(({ dataSource, symbol }) => {
+        return { dataSource, symbol };
       });
+  }
+
+  private async getCurrencies7D(): Promise<IDataGatheringItem[]> {
+    const assetProfileIdentifiersWithCompleteMarketData =
+      await this.getAssetProfileIdentifiersWithCompleteMarketData();
 
-    const symbolProfilesToGather = symbolProfiles
+    return this.exchangeRateDataService
+      .getCurrencyPairs()
+      .filter(({ dataSource, symbol }) => {
+        return !assetProfileIdentifiersWithCompleteMarketData.some((item) => {
+          return item.dataSource === dataSource && item.symbol === symbol;
+        });
+      })
+      .map(({ dataSource, symbol }) => {
+        return {
+          dataSource,
+          symbol,
+          date: subDays(resetHours(new Date()), 7)
+        };
+      });
+  }
+
+  private getEarliestDate(aStartDate: Date) {
+    return min([aStartDate, subYears(new Date(), 10)]);
+  }
+
+  private async getSymbols7D({
+    withUserSubscription = false
+  }: {
+    withUserSubscription?: boolean;
+  }): Promise<IDataGatheringItem[]> {
+    const symbolProfiles =
+      await this.symbolProfileService.getSymbolProfilesByUserSubscription({
+        withUserSubscription
+      });
+
+    const assetProfileIdentifiersWithCompleteMarketData =
+      await this.getAssetProfileIdentifiersWithCompleteMarketData();
+
+    return symbolProfiles
       .filter(({ dataSource, scraperConfiguration, symbol }) => {
         const manualDataSourceWithScraperConfiguration =
           dataSource === 'MANUAL' && !isEmpty(scraperConfiguration);
 
         return (
-          !symbolsWithCompleteMarketData.includes(symbol) &&
+          !assetProfileIdentifiersWithCompleteMarketData.some((item) => {
+            return item.dataSource === dataSource && item.symbol === symbol;
+          }) &&
           (dataSource !== 'MANUAL' || manualDataSourceWithScraperConfiguration)
         );
       })
       .map((symbolProfile) => {
         return {
           ...symbolProfile,
-          date: startDate
+          date: subDays(resetHours(new Date()), 7)
         };
       });
-
-    const currencyPairsToGather = this.exchangeRateDataService
-      .getCurrencyPairs()
-      .filter(({ symbol }) => {
-        return !symbolsWithCompleteMarketData.includes(symbol);
-      })
-      .map(({ dataSource, symbol }) => {
-        return {
-          dataSource,
-          symbol,
-          date: startDate
-        };
-      });
-
-    return [...currencyPairsToGather, ...symbolProfilesToGather];
   }
 
   private async getSymbolsMax(): Promise<IDataGatheringItem[]> {

apps/api/src/services/symbol-profile/symbol-profile.service.ts

@@ -91,6 +91,40 @@ export class SymbolProfileService {
       });
   }
 
+  public async getSymbolProfilesByUserSubscription({
+    withUserSubscription = false
+  }: {
+    withUserSubscription?: boolean;
+  }) {
+    return this.prismaService.symbolProfile.findMany({
+      include: {
+        Order: {
+          include: {
+            User: true
+          }
+        }
+      },
+      orderBy: [{ symbol: 'asc' }],
+      where: {
+        Order: withUserSubscription
+          ? {
+              some: {
+                User: {
+                  Subscription: { some: { expiresAt: { gt: new Date() } } }
+                }
+              }
+            }
+          : {
+              every: {
+                User: {
+                  Subscription: { none: { expiresAt: { gt: new Date() } } }
+                }
+              }
+            }
+      }
+    });
+  }
+
   public updateSymbolProfile({
     assetClass,
     assetSubClass,