Browse Source

Simplify recent market data gathering

task/simplify-recent-market-data-gathering
Thomas Kaul 5 days ago
parent
commit
be9f40d194
  1. 12
      apps/api/jest-environment-tz.js
  2. 18
      apps/api/src/services/queues/data-gathering/data-gathering.processor.time-zone.spec.ts
  3. 3
      apps/api/src/services/queues/data-gathering/data-gathering.processor.ts
  4. 55
      apps/api/src/services/queues/data-gathering/data-gathering.service.spec.ts
  5. 7
      apps/api/src/services/queues/data-gathering/data-gathering.service.ts

12
apps/api/jest-environment-tz.js

@ -21,9 +21,15 @@ class TimeZoneEnvironment extends NodeEnvironment {
}
async teardown() {
await super.teardown();
process.env.TZ = this.previousTimeZone;
try {
await super.teardown();
} finally {
if (this.previousTimeZone === undefined) {
delete process.env.TZ;
} else {
process.env.TZ = this.previousTimeZone;
}
}
}
}

18
apps/api/src/services/queues/data-gathering/data-gathering.processor.time-zone.spec.ts

@ -3,7 +3,6 @@
* @jest-environment-options {"timeZone": "America/New_York"}
*/
import { DataGatheringItem } from '@ghostfolio/api/services/interfaces/interfaces';
import { parseDate } from '@ghostfolio/common/helper';
import { Job } from 'bull';
@ -46,25 +45,26 @@ describe('DataGatheringProcessor in a time zone behind UTC', () => {
jest.useRealTimers();
});
it('gathers up to the last complete UTC day', async () => {
it('gathers up to the last complete UTC day and does not shift the dates', async () => {
await dataGatheringProcessor.gatherHistoricalMarketData({
data: {
dataSource: 'COINGECKO',
symbol: 'bitcoin',
date: parseDate('2026-08-21').toISOString()
// The queue enqueues dates at midnight (UTC), not at local midnight
date: new Date('2026-08-21T00:00:00.000Z').toISOString(),
symbol: 'bitcoin'
}
} as unknown as Job<DataGatheringItem>);
const { data } = marketDataService.updateMany.mock.calls[0][0];
expect(
data.map(({ date }) => {
return date;
data.map(({ date, marketPrice }) => {
return { date, marketPrice };
})
).toEqual([
new Date('2026-08-21T00:00:00.000Z'),
new Date('2026-08-22T00:00:00.000Z'),
new Date('2026-08-23T00:00:00.000Z')
{ date: new Date('2026-08-21T00:00:00.000Z'), marketPrice: 5 },
{ date: new Date('2026-08-22T00:00:00.000Z'), marketPrice: 6 },
{ date: new Date('2026-08-23T00:00:00.000Z'), marketPrice: 7 }
]);
});
});

3
apps/api/src/services/queues/data-gathering/data-gathering.processor.ts

@ -17,6 +17,7 @@ import {
} from '@ghostfolio/common/helper';
import { AssetProfileIdentifier } from '@ghostfolio/common/interfaces';
import { utc } from '@date-fns/utc';
import { Process, Processor } from '@nestjs/bull';
import { Injectable, Logger } from '@nestjs/common';
import { Prisma } from '@prisma/client';
@ -121,7 +122,7 @@ export class DataGatheringProcessor {
while (isBefore(currentDate, getStartOfUtcDate(new Date()))) {
const marketPriceOfDataProvider =
historicalData[assetProfileIdentifier]?.[
format(currentDate, DATE_FORMAT)
format(currentDate, DATE_FORMAT, { in: utc })
]?.marketPrice;
if (marketPriceOfDataProvider) {

55
apps/api/src/services/queues/data-gathering/data-gathering.service.spec.ts

@ -92,22 +92,6 @@ describe('DataGatheringService', () => {
);
});
it('excludes the Friday close when it runs on Monday', async () => {
jest.useFakeTimers().setSystemTime(new Date('2026-08-24T14:00:00.000Z'));
await dataGatheringService[
'getAssetProfileIdentifiersWithRecentMarketData'
]();
expect(prismaService.marketData.groupBy).toHaveBeenCalledWith(
expect.objectContaining({
where: expect.objectContaining({
date: { gte: new Date('2026-08-23T00:00:00.000Z') }
})
})
);
});
it('maps the query result to asset profile identifiers', async () => {
prismaService.marketData.groupBy.mockResolvedValue([
{ dataSource: 'COINGECKO', symbol: 'bitcoin' },
@ -127,10 +111,45 @@ describe('DataGatheringService', () => {
});
describe('gatherRecentMarketData', () => {
it('queries the asset profiles with recent market data once and reuses them', async () => {
const assetProfileIdentifiersWithRecentMarketData = [
{ dataSource: 'COINGECKO', symbol: 'bitcoin' }
];
prismaService.marketData.groupBy.mockResolvedValue(
assetProfileIdentifiersWithRecentMarketData
);
const getCurrencies7D = jest
.spyOn(dataGatheringService as any, 'getCurrencies7D')
.mockReturnValue([]);
const getSymbols7D = jest
.spyOn(dataGatheringService as any, 'getSymbols7D')
.mockResolvedValue([]);
await dataGatheringService.gatherRecentMarketData();
expect(prismaService.marketData.groupBy).toHaveBeenCalledTimes(1);
expect(getCurrencies7D).toHaveBeenCalledWith({
assetProfileIdentifiersWithRecentMarketData
});
expect(getSymbols7D).toHaveBeenCalledWith({
assetProfileIdentifiersWithRecentMarketData,
withUserSubscription: true
});
expect(getSymbols7D).toHaveBeenCalledWith({
assetProfileIdentifiersWithRecentMarketData,
withUserSubscription: false
});
});
it('expires completed jobs which are older than the cooldown', async () => {
jest
.spyOn(dataGatheringService as any, 'getCurrencies7D')
.mockResolvedValue([]);
.mockReturnValue([]);
jest
.spyOn(dataGatheringService as any, 'getSymbols7D')
.mockResolvedValue([]);
@ -146,7 +165,7 @@ describe('DataGatheringService', () => {
it('retains its completed jobs for the duration of the cooldown', async () => {
jest
.spyOn(dataGatheringService as any, 'getCurrencies7D')
.mockResolvedValue([
.mockReturnValue([
{
dataSource: 'YAHOO',
date: parseDate('2026-08-01'),

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

@ -267,7 +267,7 @@ export class DataGatheringService {
await this.gatherSymbols({
removeOnComplete,
dataGatheringItems: await this.getCurrencies7D({
dataGatheringItems: this.getCurrencies7D({
assetProfileIdentifiersWithRecentMarketData
}),
priority: DATA_GATHERING_QUEUE_PRIORITY_HIGH
@ -427,7 +427,6 @@ export class DataGatheringService {
return (
await this.prismaService.marketData.groupBy({
by: ['dataSource', 'symbol'],
orderBy: [{ symbol: 'asc' }],
where: {
date: { gte: getStartOfUtcDate(subDays(new Date(), 1)) },
isCarriedForward: false,
@ -439,11 +438,11 @@ export class DataGatheringService {
});
}
private async getCurrencies7D({
private getCurrencies7D({
assetProfileIdentifiersWithRecentMarketData
}: {
assetProfileIdentifiersWithRecentMarketData: AssetProfileIdentifier[];
}): Promise<DataGatheringItem[]> {
}): DataGatheringItem[] {
return this.exchangeRateDataService
.getCurrencyPairs()
.filter(({ dataSource, symbol }) => {

Loading…
Cancel
Save