diff --git a/apps/api/src/services/cron/cron.service.ts b/apps/api/src/services/cron/cron.service.ts index a0d71e794..b3e60ae59 100644 --- a/apps/api/src/services/cron/cron.service.ts +++ b/apps/api/src/services/cron/cron.service.ts @@ -39,8 +39,8 @@ export class CronService { @Cron(CronService.EVERY_HOUR_AT_RANDOM_MINUTE) public async runEveryHourAtRandomMinute() { if (await this.isDataGatheringEnabled()) { - await this.dataGatheringService.gatherRecentMarketData(); await this.dataGatheringService.gatherHourlyMarketData(); + await this.dataGatheringService.gatherRecentMarketData(); } } diff --git a/apps/api/src/services/queues/data-gathering/data-gathering.service.ts b/apps/api/src/services/queues/data-gathering/data-gathering.service.ts index 810768ef5..1a2fa720e 100644 --- a/apps/api/src/services/queues/data-gathering/data-gathering.service.ts +++ b/apps/api/src/services/queues/data-gathering/data-gathering.service.ts @@ -69,91 +69,6 @@ export class DataGatheringService { return this.dataGatheringQueue.addBulk(jobs); } - public async gatherRecentMarketData() { - await this.gatherSymbols({ - 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 - }); - } - - public async gatherMax() { - const dataGatheringItems = await this.getSymbolsMax(); - await this.gatherSymbols({ - dataGatheringItems, - priority: DATA_GATHERING_QUEUE_PRIORITY_LOW - }); - } - - public async gatherSymbol({ dataSource, date, symbol }: DataGatheringItem) { - const dataGatheringItems = (await this.getSymbolsMax()) - .filter((dataGatheringItem) => { - return ( - dataGatheringItem.dataSource === dataSource && - dataGatheringItem.symbol === symbol - ); - }) - .map((item) => ({ - ...item, - date: date ?? item.date - })); - - await this.gatherSymbols({ - dataGatheringItems, - force: true, - priority: DATA_GATHERING_QUEUE_PRIORITY_HIGH - }); - } - - public async gatherSymbolForDate({ - dataSource, - date, - symbol - }: { date: Date } & AssetProfileIdentifier) { - try { - const historicalData = await this.dataProviderService.getHistoricalRaw({ - assetProfileIdentifiers: [{ dataSource, symbol }], - from: date, - to: date - }); - - const marketPrice = - historicalData[getAssetProfileIdentifier({ dataSource, symbol })][ - format(date, DATE_FORMAT) - ].marketPrice; - - if (marketPrice) { - return await this.prismaService.marketData.upsert({ - create: { - dataSource, - date, - marketPrice, - symbol - }, - update: { marketPrice }, - where: { dataSource_date_symbol: { dataSource, date, symbol } } - }); - } - } catch (error) { - this.logger.error(error); - } finally { - return undefined; - } - } - public async gatherAssetProfiles( aAssetProfileIdentifiers?: AssetProfileIdentifier[] ) { @@ -328,6 +243,91 @@ export class DataGatheringService { } } + public async gatherMax() { + const dataGatheringItems = await this.getSymbolsMax(); + await this.gatherSymbols({ + dataGatheringItems, + priority: DATA_GATHERING_QUEUE_PRIORITY_LOW + }); + } + + public async gatherRecentMarketData() { + await this.gatherSymbols({ + 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 + }); + } + + public async gatherSymbol({ dataSource, date, symbol }: DataGatheringItem) { + const dataGatheringItems = (await this.getSymbolsMax()) + .filter((dataGatheringItem) => { + return ( + dataGatheringItem.dataSource === dataSource && + dataGatheringItem.symbol === symbol + ); + }) + .map((item) => ({ + ...item, + date: date ?? item.date + })); + + await this.gatherSymbols({ + dataGatheringItems, + force: true, + priority: DATA_GATHERING_QUEUE_PRIORITY_HIGH + }); + } + + public async gatherSymbolForDate({ + dataSource, + date, + symbol + }: { date: Date } & AssetProfileIdentifier) { + try { + const historicalData = await this.dataProviderService.getHistoricalRaw({ + assetProfileIdentifiers: [{ dataSource, symbol }], + from: date, + to: date + }); + + const marketPrice = + historicalData[getAssetProfileIdentifier({ dataSource, symbol })][ + format(date, DATE_FORMAT) + ].marketPrice; + + if (marketPrice) { + return await this.prismaService.marketData.upsert({ + create: { + dataSource, + date, + marketPrice, + symbol + }, + update: { marketPrice }, + where: { dataSource_date_symbol: { dataSource, date, symbol } } + }); + } + } catch (error) { + this.logger.error(error); + } finally { + return undefined; + } + } + public async gatherSymbols({ dataGatheringItems, force = false,