Browse Source

Task/simplify recent market data gathering (#7714)

* Simplify recent market data gathering

* Update changelog
pull/7720/head^2
Thomas Kaul 3 days ago
committed by GitHub
parent
commit
b4e8a4090d
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
  1. 3
      CHANGELOG.md
  2. 35
      apps/api/jest-environment-tz.js
  3. 3
      apps/api/src/app/admin/admin.controller.ts
  4. 5
      apps/api/src/services/data-provider/coingecko/coingecko.service.ts
  5. 8
      apps/api/src/services/data-provider/data-provider.service.ts
  6. 9
      apps/api/src/services/data-provider/manual/manual.service.ts
  7. 10
      apps/api/src/services/data-provider/rapid-api/rapid-api.service.ts
  8. 5
      apps/api/src/services/data-provider/yahoo-finance/yahoo-finance.service.ts
  9. 70
      apps/api/src/services/queues/data-gathering/data-gathering.processor.time-zone.spec.ts
  10. 36
      apps/api/src/services/queues/data-gathering/data-gathering.processor.ts
  11. 133
      apps/api/src/services/queues/data-gathering/data-gathering.service.spec.ts
  12. 75
      apps/api/src/services/queues/data-gathering/data-gathering.service.ts
  13. 5
      libs/common/src/lib/helper.ts
  14. 1
      package-lock.json
  15. 1
      package.json

3
CHANGELOG.md

@ -12,10 +12,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Harmonized the icons and labels in the access table to share the portfolio - Harmonized the icons and labels in the access table to share the portfolio
- Migrated the create and edit access dialogs to dedicated routes - Migrated the create and edit access dialogs to dedicated routes
- Improved the response of the historical market data gathering endpoint for a specific date - Improved the response of the historical market data gathering endpoint for a specific date
- Improved the historical market data gathering by loading the asset profiles with recent market data in a single database query per run
### Fixed ### Fixed
- Fixed the country mapping of Macau in the _Financial Modeling Prep_ service - Fixed the country mapping of Macau in the _Financial Modeling Prep_ service
- Fixed the date of the gathered historical market data for instances running in a time zone other than UTC
- Fixed the repeated historical market data gathering for instances running in a time zone other than UTC
## 3.60.0 - 2026-08-24 ## 3.60.0 - 2026-08-24

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

@ -0,0 +1,35 @@
const NodeEnvironment = require('jest-environment-node').TestEnvironment;
// Jest gives each test file a copy of `process`, so a test cannot change the
// time zone at run time. This environment sets `TZ` on the real process of the
// worker, which resets the internal date cache of Node.
class TimeZoneEnvironment extends NodeEnvironment {
constructor(config, context) {
super(config, context);
this.previousTimeZone = process.env.TZ;
this.timeZone = config.projectConfig.testEnvironmentOptions?.timeZone;
}
async setup() {
if (this.timeZone) {
process.env.TZ = this.timeZone;
}
await super.setup();
}
async teardown() {
try {
await super.teardown();
} finally {
if (this.previousTimeZone === undefined) {
delete process.env.TZ;
} else {
process.env.TZ = this.previousTimeZone;
}
}
}
}
module.exports = TimeZoneEnvironment;

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

@ -37,6 +37,7 @@ import type {
RequestWithUser RequestWithUser
} from '@ghostfolio/common/types'; } from '@ghostfolio/common/types';
import { utc } from '@date-fns/utc';
import { import {
Body, Body,
Controller, Controller,
@ -204,7 +205,7 @@ export class AdminController {
@Param('dateString') dateString: string, @Param('dateString') dateString: string,
@Param('symbol') symbol: string @Param('symbol') symbol: string
): Promise<MarketData> { ): Promise<MarketData> {
const date = parseISO(dateString); const date = parseISO(dateString, { in: utc });
if (!isDate(date)) { if (!isDate(date)) {
throw new HttpException( throw new HttpException(

5
apps/api/src/services/data-provider/coingecko/coingecko.service.ts

@ -18,6 +18,7 @@ import {
LookupResponse LookupResponse
} from '@ghostfolio/common/interfaces'; } from '@ghostfolio/common/interfaces';
import { utc } from '@date-fns/utc';
import { Injectable, Logger, OnModuleInit } from '@nestjs/common'; import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
import { import {
AssetClass, AssetClass,
@ -147,7 +148,9 @@ export class CoinGeckoService implements DataProviderInterface, OnModuleInit {
} = {}; } = {};
for (const [timestamp, marketPrice] of prices) { for (const [timestamp, marketPrice] of prices) {
result[format(fromUnixTime(timestamp / 1000), DATE_FORMAT)] = { result[
format(fromUnixTime(timestamp / 1000), DATE_FORMAT, { in: utc })
] = {
marketPrice marketPrice
}; };
} }

8
apps/api/src/services/data-provider/data-provider.service.ts

@ -35,6 +35,7 @@ import {
} from '@ghostfolio/common/interfaces'; } from '@ghostfolio/common/interfaces';
import type { Granularity, UserWithSettings } from '@ghostfolio/common/types'; import type { Granularity, UserWithSettings } from '@ghostfolio/common/types';
import { utc } from '@date-fns/utc';
import { Inject, Injectable, Logger, OnModuleInit } from '@nestjs/common'; import { Inject, Injectable, Logger, OnModuleInit } from '@nestjs/common';
import { DataSource, MarketData, Prisma, SymbolProfile } from '@prisma/client'; import { DataSource, MarketData, Prisma, SymbolProfile } from '@prisma/client';
import { Big } from 'big.js'; import { Big } from 'big.js';
@ -494,8 +495,11 @@ export class DataProviderService implements OnModuleInit {
[date: string]: DataProviderHistoricalResponse; [date: string]: DataProviderHistoricalResponse;
} = {}; } = {};
for (const date of eachDayOfInterval({ end: to, start: from })) { for (const date of eachDayOfInterval(
data[format(date, DATE_FORMAT)] = { marketPrice: 100 }; { end: to, start: from },
{ in: utc }
)) {
data[format(date, DATE_FORMAT, { in: utc })] = { marketPrice: 100 };
} }
promises.push( promises.push(

9
apps/api/src/services/data-provider/manual/manual.service.ts

@ -14,7 +14,7 @@ import { SymbolProfileService } from '@ghostfolio/api/services/symbol-profile/sy
import { import {
DATE_FORMAT, DATE_FORMAT,
extractNumberFromString, extractNumberFromString,
getYesterday getStartOfUtcDateOfYesterday
} from '@ghostfolio/common/helper'; } from '@ghostfolio/common/helper';
import { import {
DataProviderHistoricalResponse, DataProviderHistoricalResponse,
@ -24,6 +24,7 @@ import {
ScraperConfiguration ScraperConfiguration
} from '@ghostfolio/common/interfaces'; } from '@ghostfolio/common/interfaces';
import { utc } from '@date-fns/utc';
import { Injectable, Logger } from '@nestjs/common'; import { Injectable, Logger } from '@nestjs/common';
import { DataSource, SymbolProfile } from '@prisma/client'; import { DataSource, SymbolProfile } from '@prisma/client';
import * as cheerio from 'cheerio'; import * as cheerio from 'cheerio';
@ -96,11 +97,11 @@ export class ManualService implements DataProviderInterface {
let date = from; let date = from;
while (isBefore(date, to)) { while (isBefore(date, to)) {
historical[format(date, DATE_FORMAT)] = { historical[format(date, DATE_FORMAT, { in: utc })] = {
marketPrice: defaultMarketPrice marketPrice: defaultMarketPrice
}; };
date = addDays(date, 1); date = addDays(date, 1, { in: utc });
} }
return historical; return historical;
@ -114,7 +115,7 @@ export class ManualService implements DataProviderInterface {
}); });
return { return {
[format(getYesterday(), DATE_FORMAT)]: { [format(getStartOfUtcDateOfYesterday(), DATE_FORMAT, { in: utc })]: {
marketPrice: value marketPrice: value
} }
}; };

10
apps/api/src/services/data-provider/rapid-api/rapid-api.service.ts

@ -9,7 +9,10 @@ import {
} from '@ghostfolio/api/services/data-provider/interfaces/data-provider.interface'; } from '@ghostfolio/api/services/data-provider/interfaces/data-provider.interface';
import { FetchService } from '@ghostfolio/api/services/fetch/fetch.service'; import { FetchService } from '@ghostfolio/api/services/fetch/fetch.service';
import { ghostfolioFearAndGreedIndexSymbolStocks } from '@ghostfolio/common/config'; import { ghostfolioFearAndGreedIndexSymbolStocks } from '@ghostfolio/common/config';
import { DATE_FORMAT, getYesterday } from '@ghostfolio/common/helper'; import {
DATE_FORMAT,
getStartOfUtcDateOfYesterday
} from '@ghostfolio/common/helper';
import { import {
DataProviderHistoricalResponse, DataProviderHistoricalResponse,
DataProviderInfo, DataProviderInfo,
@ -17,6 +20,7 @@ import {
LookupResponse LookupResponse
} from '@ghostfolio/common/interfaces'; } from '@ghostfolio/common/interfaces';
import { utc } from '@date-fns/utc';
import { Injectable, Logger } from '@nestjs/common'; import { Injectable, Logger } from '@nestjs/common';
import { DataSource, SymbolProfile } from '@prisma/client'; import { DataSource, SymbolProfile } from '@prisma/client';
import { format } from 'date-fns'; import { format } from 'date-fns';
@ -66,7 +70,9 @@ export class RapidApiService implements DataProviderInterface {
if (fgi) { if (fgi) {
return { return {
[format(getYesterday(), DATE_FORMAT)]: { [format(getStartOfUtcDateOfYesterday(), DATE_FORMAT, {
in: utc
})]: {
marketPrice: fgi.previousClose.value marketPrice: fgi.previousClose.value
} }
}; };

5
apps/api/src/services/data-provider/yahoo-finance/yahoo-finance.service.ts

@ -19,6 +19,7 @@ import {
LookupResponse LookupResponse
} from '@ghostfolio/common/interfaces'; } from '@ghostfolio/common/interfaces';
import { utc } from '@date-fns/utc';
import { Injectable, Logger } from '@nestjs/common'; import { Injectable, Logger } from '@nestjs/common';
import { DataSource, SymbolProfile } from '@prisma/client'; import { DataSource, SymbolProfile } from '@prisma/client';
import { addDays, format, isSameDay } from 'date-fns'; import { addDays, format, isSameDay } from 'date-fns';
@ -103,7 +104,7 @@ export class YahooFinanceService implements DataProviderInterface {
} = {}; } = {};
for (const historicalItem of historicalResult) { for (const historicalItem of historicalResult) {
response[format(historicalItem.date, DATE_FORMAT)] = { response[format(historicalItem.date, DATE_FORMAT, { in: utc })] = {
marketPrice: historicalItem.dividends marketPrice: historicalItem.dividends
}; };
} }
@ -174,7 +175,7 @@ export class YahooFinanceService implements DataProviderInterface {
: undefined); : undefined);
if (marketPrice) { if (marketPrice) {
response[format(date, DATE_FORMAT)] = { marketPrice }; response[format(date, DATE_FORMAT, { in: utc })] = { marketPrice };
} }
} }

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

@ -0,0 +1,70 @@
/**
* @jest-environment <rootDir>/jest-environment-tz.js
* @jest-environment-options {"timeZone": "America/New_York"}
*/
import { DataGatheringItem } from '@ghostfolio/api/services/interfaces/interfaces';
import { Job } from 'bull';
import { DataGatheringProcessor } from './data-gathering.processor';
describe('DataGatheringProcessor in a time zone behind UTC', () => {
let dataGatheringProcessor: DataGatheringProcessor;
let dataProviderService: { getHistoricalRaw: jest.Mock };
let marketDataService: { replaceForSymbol: jest.Mock; updateMany: jest.Mock };
beforeAll(() => {
// 2026-08-23 21:30 in New York, but already 2026-08-24 in UTC
jest.useFakeTimers().setSystemTime(new Date('2026-08-24T01:30:00.000Z'));
});
beforeEach(() => {
dataProviderService = {
getHistoricalRaw: jest.fn().mockResolvedValue({
'COINGECKO-bitcoin': {
'2026-08-21': { marketPrice: 5 },
'2026-08-22': { marketPrice: 6 },
'2026-08-23': { marketPrice: 7 }
}
})
};
marketDataService = {
replaceForSymbol: jest.fn(),
updateMany: jest.fn()
};
dataGatheringProcessor = new DataGatheringProcessor(
null,
dataProviderService as any,
marketDataService as any,
null
);
});
afterAll(() => {
jest.useRealTimers();
});
it('gathers up to the last complete UTC day and does not shift the dates', async () => {
await dataGatheringProcessor.gatherHistoricalMarketData({
data: {
dataSource: 'COINGECKO',
// 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, marketPrice }) => {
return { date, marketPrice };
})
).toEqual([
{ 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 }
]);
});
});

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

@ -17,19 +17,12 @@ import {
} from '@ghostfolio/common/helper'; } from '@ghostfolio/common/helper';
import { AssetProfileIdentifier } from '@ghostfolio/common/interfaces'; import { AssetProfileIdentifier } from '@ghostfolio/common/interfaces';
import { utc } from '@date-fns/utc';
import { Process, Processor } from '@nestjs/bull'; import { Process, Processor } from '@nestjs/bull';
import { Injectable, Logger } from '@nestjs/common'; import { Injectable, Logger } from '@nestjs/common';
import { Prisma } from '@prisma/client'; import { Prisma } from '@prisma/client';
import { Job } from 'bull'; import { Job } from 'bull';
import { import { addDays, format, isBefore, parseISO } from 'date-fns';
addDays,
format,
getDate,
getMonth,
getYear,
isBefore,
parseISO
} from 'date-fns';
import { DataGatheringService } from './data-gathering.service'; import { DataGatheringService } from './data-gathering.service';
@ -108,7 +101,8 @@ export class DataGatheringProcessor {
this.logger.log( this.logger.log(
`Historical market data gathering has been started for ${symbol} (${dataSource}) at ${format( `Historical market data gathering has been started for ${symbol} (${dataSource}) at ${format(
currentDate, currentDate,
DATE_FORMAT DATE_FORMAT,
{ in: utc }
)}${force ? ' (forced update)' : ''}` )}${force ? ' (forced update)' : ''}`
); );
@ -124,24 +118,13 @@ export class DataGatheringProcessor {
}); });
const data: Prisma.MarketDataUpdateInput[] = []; const data: Prisma.MarketDataUpdateInput[] = [];
const startOfUtcDateOfToday = getStartOfUtcDate(new Date());
let lastMarketPrice: number; let lastMarketPrice: number;
while ( while (isBefore(currentDate, startOfUtcDateOfToday)) {
isBefore(
currentDate,
new Date(
Date.UTC(
getYear(new Date()),
getMonth(new Date()),
getDate(new Date()),
0
)
)
)
) {
const marketPriceOfDataProvider = const marketPriceOfDataProvider =
historicalData[assetProfileIdentifier]?.[ historicalData[assetProfileIdentifier]?.[
format(currentDate, DATE_FORMAT) format(currentDate, DATE_FORMAT, { in: utc })
]?.marketPrice; ]?.marketPrice;
if (marketPriceOfDataProvider) { if (marketPriceOfDataProvider) {
@ -159,7 +142,7 @@ export class DataGatheringProcessor {
}); });
} }
currentDate = addDays(currentDate, 1); currentDate = addDays(currentDate, 1, { in: utc });
} }
if (force) { if (force) {
@ -175,7 +158,8 @@ export class DataGatheringProcessor {
this.logger.log( this.logger.log(
`Historical market data gathering has been completed for ${symbol} (${dataSource}) at ${format( `Historical market data gathering has been completed for ${symbol} (${dataSource}) at ${format(
currentDate, currentDate,
DATE_FORMAT DATE_FORMAT,
{ in: utc }
)}` )}`
); );
} catch (error) { } catch (error) {

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

@ -42,8 +42,8 @@ describe('DataGatheringService', () => {
}); });
describe('getAssetProfileIdentifiersWithRecentMarketData', () => { describe('getAssetProfileIdentifiersWithRecentMarketData', () => {
it('excludes carried forward market prices from the query', async () => { it('queries real market prices since the start of yesterday (UTC)', async () => {
jest.useFakeTimers().setSystemTime(parseDate('2026-08-23').getTime()); jest.useFakeTimers().setSystemTime(new Date('2026-08-24T14:00:00.000Z'));
await dataGatheringService[ await dataGatheringService[
'getAssetProfileIdentifiersWithRecentMarketData' 'getAssetProfileIdentifiersWithRecentMarketData'
@ -51,28 +51,51 @@ describe('DataGatheringService', () => {
expect(prismaService.marketData.groupBy).toHaveBeenCalledWith( expect(prismaService.marketData.groupBy).toHaveBeenCalledWith(
expect.objectContaining({ expect.objectContaining({
where: expect.objectContaining({ where: {
date: { gte: new Date('2026-08-23T00:00:00.000Z') },
isCarriedForward: false, isCarriedForward: false,
state: 'CLOSE' state: 'CLOSE'
}
})
);
});
it('includes the Friday close when it runs on Saturday', async () => {
jest.useFakeTimers().setSystemTime(new Date('2026-08-22T14:00:00.000Z'));
await dataGatheringService[
'getAssetProfileIdentifiersWithRecentMarketData'
]();
expect(prismaService.marketData.groupBy).toHaveBeenCalledWith(
expect.objectContaining({
where: expect.objectContaining({
date: { gte: new Date('2026-08-21T00:00:00.000Z') }
}) })
}) })
); );
}); });
it('keeps a cryptocurrency with a real market price of yesterday on Sunday', async () => { it('excludes the Friday close when it runs on Sunday', async () => {
jest.useFakeTimers().setSystemTime(parseDate('2026-08-23').getTime()); jest.useFakeTimers().setSystemTime(new Date('2026-08-23T14:00:00.000Z'));
await dataGatheringService[
'getAssetProfileIdentifiersWithRecentMarketData'
]();
expect(prismaService.marketData.groupBy).toHaveBeenCalledWith(
expect.objectContaining({
where: expect.objectContaining({
date: { gte: new Date('2026-08-22T00:00:00.000Z') }
})
})
);
});
it('maps the query result to asset profile identifiers', async () => {
prismaService.marketData.groupBy.mockResolvedValue([ prismaService.marketData.groupBy.mockResolvedValue([
{ { dataSource: 'COINGECKO', symbol: 'bitcoin' },
_max: { date: parseDate('2026-08-22') }, { dataSource: 'YAHOO', symbol: 'AAPL' }
dataSource: 'COINGECKO',
symbol: 'bitcoin'
},
{
_max: { date: parseDate('2026-08-21') },
dataSource: 'YAHOO',
symbol: 'AAPL'
}
]); ]);
const assetProfileIdentifiers = const assetProfileIdentifiers =
@ -81,68 +104,52 @@ describe('DataGatheringService', () => {
](); ]();
expect(assetProfileIdentifiers).toEqual([ expect(assetProfileIdentifiers).toEqual([
{ dataSource: 'COINGECKO', symbol: 'bitcoin' } { dataSource: 'COINGECKO', symbol: 'bitcoin' },
{ dataSource: 'YAHOO', symbol: 'AAPL' }
]); ]);
}); });
});
it('drops a stock with a real market price of Friday on Monday', async () => { describe('gatherRecentMarketData', () => {
jest.useFakeTimers().setSystemTime(parseDate('2026-08-24').getTime()); it('queries the asset profiles with recent market data once and reuses them', async () => {
const assetProfileIdentifiersWithRecentMarketData = [
{ dataSource: 'COINGECKO', symbol: 'bitcoin' }
];
prismaService.marketData.groupBy.mockResolvedValue([ prismaService.marketData.groupBy.mockResolvedValue(
{ assetProfileIdentifiersWithRecentMarketData
_max: { date: parseDate('2026-08-23') }, );
dataSource: 'COINGECKO',
symbol: 'bitcoin'
},
{
_max: { date: parseDate('2026-08-21') },
dataSource: 'YAHOO',
symbol: 'AAPL'
}
]);
const assetProfileIdentifiers = const getCurrencies7D = jest
await dataGatheringService[ .spyOn(dataGatheringService as any, 'getCurrencies7D')
'getAssetProfileIdentifiersWithRecentMarketData' .mockReturnValue([]);
](); const getSymbols7D = jest
.spyOn(dataGatheringService as any, 'getSymbols7D')
.mockResolvedValue([]);
expect(assetProfileIdentifiers).toEqual([ await dataGatheringService.gatherRecentMarketData();
{ dataSource: 'COINGECKO', symbol: 'bitcoin' }
]);
});
it('drops a stock with a late Friday close on Saturday', async () => { expect(prismaService.marketData.groupBy).toHaveBeenCalledTimes(1);
jest.useFakeTimers().setSystemTime(parseDate('2026-08-22').getTime());
prismaService.marketData.groupBy.mockResolvedValue([ expect(getCurrencies7D).toHaveBeenCalledWith({
{ assetProfileIdentifiersWithRecentMarketData
_max: { date: parseDate('2026-08-20') }, });
dataSource: 'YAHOO',
symbol: 'AAPL'
},
{
_max: { date: parseDate('2026-08-21') },
dataSource: 'YAHOO',
symbol: 'MSFT'
}
]);
const assetProfileIdentifiers = expect(getSymbols7D).toHaveBeenCalledWith({
await dataGatheringService[ assetProfileIdentifiersWithRecentMarketData,
'getAssetProfileIdentifiersWithRecentMarketData' withUserSubscription: true
](); });
expect(assetProfileIdentifiers).toEqual([ expect(getSymbols7D).toHaveBeenCalledWith({
{ dataSource: 'YAHOO', symbol: 'MSFT' } assetProfileIdentifiersWithRecentMarketData,
]); withUserSubscription: false
});
}); });
});
describe('gatherRecentMarketData', () => {
it('expires completed jobs which are older than the cooldown', async () => { it('expires completed jobs which are older than the cooldown', async () => {
jest jest
.spyOn(dataGatheringService as any, 'getCurrencies7D') .spyOn(dataGatheringService as any, 'getCurrencies7D')
.mockResolvedValue([]); .mockReturnValue([]);
jest jest
.spyOn(dataGatheringService as any, 'getSymbols7D') .spyOn(dataGatheringService as any, 'getSymbols7D')
.mockResolvedValue([]); .mockResolvedValue([]);
@ -158,7 +165,7 @@ describe('DataGatheringService', () => {
it('retains its completed jobs for the duration of the cooldown', async () => { it('retains its completed jobs for the duration of the cooldown', async () => {
jest jest
.spyOn(dataGatheringService as any, 'getCurrencies7D') .spyOn(dataGatheringService as any, 'getCurrencies7D')
.mockResolvedValue([ .mockReturnValue([
{ {
dataSource: 'YAHOO', dataSource: 'YAHOO',
date: parseDate('2026-08-01'), date: parseDate('2026-08-01'),

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

@ -20,25 +20,19 @@ import {
DATE_FORMAT, DATE_FORMAT,
getAssetProfileIdentifier, getAssetProfileIdentifier,
getStartOfUtcDate, getStartOfUtcDate,
resetHours getStartOfUtcDateOfYesterday
} from '@ghostfolio/common/helper'; } from '@ghostfolio/common/helper';
import { import {
AssetProfileIdentifier, AssetProfileIdentifier,
BenchmarkProperty BenchmarkProperty
} from '@ghostfolio/common/interfaces'; } from '@ghostfolio/common/interfaces';
import { utc } from '@date-fns/utc';
import { InjectQueue } from '@nestjs/bull'; import { InjectQueue } from '@nestjs/bull';
import { Inject, Injectable, Logger } from '@nestjs/common'; import { Inject, Injectable, Logger } from '@nestjs/common';
import { Prisma } from '@prisma/client'; import { Prisma } from '@prisma/client';
import { Job, JobOptions, Queue } from 'bull'; import { Job, JobOptions, Queue } from 'bull';
import { import { format, min, subDays, subMilliseconds, subYears } from 'date-fns';
format,
isBefore,
min,
subDays,
subMilliseconds,
subYears
} from 'date-fns';
import { isEmpty } from 'lodash'; import { isEmpty } from 'lodash';
import ms, { StringValue } from 'ms'; import ms, { StringValue } from 'ms';
@ -269,15 +263,21 @@ export class DataGatheringService {
age: GATHER_HISTORICAL_MARKET_DATA_COOLDOWN_IN_MS / 1000 age: GATHER_HISTORICAL_MARKET_DATA_COOLDOWN_IN_MS / 1000
}; };
const assetProfileIdentifiersWithRecentMarketData =
await this.getAssetProfileIdentifiersWithRecentMarketData();
await this.gatherSymbols({ await this.gatherSymbols({
removeOnComplete, removeOnComplete,
dataGatheringItems: await this.getCurrencies7D(), dataGatheringItems: this.getCurrencies7D({
assetProfileIdentifiersWithRecentMarketData
}),
priority: DATA_GATHERING_QUEUE_PRIORITY_HIGH priority: DATA_GATHERING_QUEUE_PRIORITY_HIGH
}); });
await this.gatherSymbols({ await this.gatherSymbols({
removeOnComplete, removeOnComplete,
dataGatheringItems: await this.getSymbols7D({ dataGatheringItems: await this.getSymbols7D({
assetProfileIdentifiersWithRecentMarketData,
withUserSubscription: true withUserSubscription: true
}), }),
priority: DATA_GATHERING_QUEUE_PRIORITY_MEDIUM priority: DATA_GATHERING_QUEUE_PRIORITY_MEDIUM
@ -286,6 +286,7 @@ export class DataGatheringService {
await this.gatherSymbols({ await this.gatherSymbols({
removeOnComplete, removeOnComplete,
dataGatheringItems: await this.getSymbols7D({ dataGatheringItems: await this.getSymbols7D({
assetProfileIdentifiersWithRecentMarketData,
withUserSubscription: false withUserSubscription: false
}), }),
priority: DATA_GATHERING_QUEUE_PRIORITY_LOW priority: DATA_GATHERING_QUEUE_PRIORITY_LOW
@ -322,29 +323,37 @@ export class DataGatheringService {
date, date,
symbol symbol
}: { date: Date } & AssetProfileIdentifier) { }: { date: Date } & AssetProfileIdentifier) {
const startOfUtcDate = getStartOfUtcDate(date);
try { try {
const historicalData = await this.dataProviderService.getHistoricalRaw({ const historicalData = await this.dataProviderService.getHistoricalRaw({
assetProfileIdentifiers: [{ dataSource, symbol }], assetProfileIdentifiers: [{ dataSource, symbol }],
from: date, from: startOfUtcDate,
to: date to: startOfUtcDate
}); });
const marketPrice = const marketPrice =
historicalData[getAssetProfileIdentifier({ dataSource, symbol })]?.[ historicalData[getAssetProfileIdentifier({ dataSource, symbol })]?.[
format(date, DATE_FORMAT) format(startOfUtcDate, DATE_FORMAT, { in: utc })
]?.marketPrice; ]?.marketPrice;
if (marketPrice) { if (marketPrice) {
return await this.prismaService.marketData.upsert({ return await this.prismaService.marketData.upsert({
create: { create: {
dataSource, dataSource,
date,
marketPrice, marketPrice,
symbol, symbol,
date: startOfUtcDate,
isCarriedForward: false isCarriedForward: false
}, },
update: { marketPrice, isCarriedForward: false }, update: { marketPrice, isCarriedForward: false },
where: { dataSource_date_symbol: { dataSource, date, symbol } } where: {
dataSource_date_symbol: {
dataSource,
symbol,
date: startOfUtcDate
}
}
}); });
} }
} catch (error) { } catch (error) {
@ -382,7 +391,7 @@ export class DataGatheringService {
jobId: `${getAssetProfileIdentifier({ jobId: `${getAssetProfileIdentifier({
dataSource, dataSource,
symbol symbol
})}-${format(date, DATE_FORMAT)}` })}-${format(date, DATE_FORMAT, { in: utc })}`
} }
}; };
}) })
@ -426,28 +435,25 @@ export class DataGatheringService {
> { > {
return ( return (
await this.prismaService.marketData.groupBy({ await this.prismaService.marketData.groupBy({
_max: { date: true },
by: ['dataSource', 'symbol'], by: ['dataSource', 'symbol'],
orderBy: [{ symbol: 'asc' }],
where: { where: {
date: { gt: subDays(resetHours(new Date()), 7) }, date: {
gte: getStartOfUtcDateOfYesterday()
},
isCarriedForward: false, isCarriedForward: false,
state: 'CLOSE' state: 'CLOSE'
} }
}) })
) ).map(({ dataSource, symbol }) => {
.filter(({ _max }) => { return { dataSource, symbol };
return !isBefore(_max.date, getStartOfUtcDate(subDays(new Date(), 1))); });
})
.map(({ dataSource, symbol }) => {
return { dataSource, symbol };
});
} }
private async getCurrencies7D(): Promise<DataGatheringItem[]> { private getCurrencies7D({
const assetProfileIdentifiersWithRecentMarketData = assetProfileIdentifiersWithRecentMarketData
await this.getAssetProfileIdentifiersWithRecentMarketData(); }: {
assetProfileIdentifiersWithRecentMarketData: AssetProfileIdentifier[];
}): DataGatheringItem[] {
return this.exchangeRateDataService return this.exchangeRateDataService
.getCurrencyPairs() .getCurrencyPairs()
.filter(({ dataSource, symbol }) => { .filter(({ dataSource, symbol }) => {
@ -459,7 +465,7 @@ export class DataGatheringService {
return { return {
dataSource, dataSource,
symbol, symbol,
date: subDays(resetHours(new Date()), 7) date: subDays(getStartOfUtcDate(new Date()), 7, { in: utc })
}; };
}); });
} }
@ -499,8 +505,10 @@ export class DataGatheringService {
} }
private async getSymbols7D({ private async getSymbols7D({
assetProfileIdentifiersWithRecentMarketData,
withUserSubscription = false withUserSubscription = false
}: { }: {
assetProfileIdentifiersWithRecentMarketData: AssetProfileIdentifier[];
withUserSubscription?: boolean; withUserSubscription?: boolean;
}): Promise<DataGatheringItem[]> { }): Promise<DataGatheringItem[]> {
const symbolProfiles = const symbolProfiles =
@ -510,9 +518,6 @@ export class DataGatheringService {
} }
); );
const assetProfileIdentifiersWithRecentMarketData =
await this.getAssetProfileIdentifiersWithRecentMarketData();
return symbolProfiles return symbolProfiles
.filter(({ dataSource, scraperConfiguration, symbol }) => { .filter(({ dataSource, scraperConfiguration, symbol }) => {
const manualDataSourceWithScraperConfiguration = const manualDataSourceWithScraperConfiguration =
@ -528,7 +533,7 @@ export class DataGatheringService {
.map((symbolProfile) => { .map((symbolProfile) => {
return { return {
...symbolProfile, ...symbolProfile,
date: subDays(resetHours(new Date()), 7) date: subDays(getStartOfUtcDate(new Date()), 7, { in: utc })
}; };
}); });
} }

5
libs/common/src/lib/helper.ts

@ -1,3 +1,4 @@
import { utc } from '@date-fns/utc';
import { NumberParser } from '@internationalized/number'; import { NumberParser } from '@internationalized/number';
import { import {
AccessType, AccessType,
@ -465,6 +466,10 @@ export function getStartOfUtcDate(aDate: Date) {
return date; return date;
} }
export function getStartOfUtcDateOfYesterday() {
return subDays(getStartOfUtcDate(new Date()), 1, { in: utc });
}
export function getStringOrNull(aString: string | null | undefined) { export function getStringOrNull(aString: string | null | undefined) {
const trimmedString = aString?.trim(); const trimmedString = aString?.trim();

1
package-lock.json

@ -160,6 +160,7 @@
"husky": "9.1.7", "husky": "9.1.7",
"jest": "30.3.0", "jest": "30.3.0",
"jest-environment-jsdom": "30.2.0", "jest-environment-jsdom": "30.2.0",
"jest-environment-node": "30.3.0",
"jest-preset-angular": "17.0.0", "jest-preset-angular": "17.0.0",
"nx": "23.1.1", "nx": "23.1.1",
"prettier": "3.9.6", "prettier": "3.9.6",

1
package.json

@ -204,6 +204,7 @@
"husky": "9.1.7", "husky": "9.1.7",
"jest": "30.3.0", "jest": "30.3.0",
"jest-environment-jsdom": "30.2.0", "jest-environment-jsdom": "30.2.0",
"jest-environment-node": "30.3.0",
"jest-preset-angular": "17.0.0", "jest-preset-angular": "17.0.0",
"nx": "23.1.1", "nx": "23.1.1",
"prettier": "3.9.6", "prettier": "3.9.6",

Loading…
Cancel
Save