From 712a7a7e11b0fec53ba22236b909d5b1b6a92e9c Mon Sep 17 00:00:00 2001 From: Thomas Kaul <4159106+dtslvr@users.noreply.github.com> Date: Mon, 24 Aug 2026 15:51:28 +0200 Subject: [PATCH] Bugfix/repeated historical market data gathering on weekends (#7710) * Fix repeated historical market data gathering on weekends * Update changelog --- CHANGELOG.md | 15 + apps/api/src/app/admin/admin.service.ts | 28 +- .../portfolio/current-rate.service.spec.ts | 3 + .../market-data/market-data.service.spec.ts | 118 ++++++++ .../market-data/market-data.service.ts | 43 +-- .../data-gathering.processor.spec.ts | 266 ++++++++++++++++++ .../data-gathering.processor.ts | 12 +- .../data-gathering.service.spec.ts | 223 +++++++++++++++ .../data-gathering/data-gathering.service.ts | 53 +++- libs/common/src/lib/config.ts | 1 + .../migration.sql | 2 + prisma/schema.prisma | 15 +- 12 files changed, 707 insertions(+), 72 deletions(-) create mode 100644 apps/api/src/services/market-data/market-data.service.spec.ts create mode 100644 apps/api/src/services/queues/data-gathering/data-gathering.processor.spec.ts create mode 100644 apps/api/src/services/queues/data-gathering/data-gathering.service.spec.ts create mode 100644 prisma/migrations/20260824000000_added_is_carried_forward_to_market_data/migration.sql diff --git a/CHANGELOG.md b/CHANGELOG.md index 20706ee36..ef1fd7207 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,21 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## Unreleased + +### Added + +- Added `isCarriedForward` to the `MarketData` database schema + +### Changed + +- Improved the historical market data gathering by storing the market prices carried forward for the most recent dates without data from the data provider, distinguished by `isCarriedForward` +- Introduced a cooldown of 12 hours for the historical market data gathering of a symbol by retaining the completed jobs + +### Fixed + +- Fixed the repeated historical market data gathering for symbols without weekend market data on Sundays and Mondays + ## 3.59.1 - 2026-08-23 ### Added diff --git a/apps/api/src/app/admin/admin.service.ts b/apps/api/src/app/admin/admin.service.ts index ec90cfe24..915c2068e 100644 --- a/apps/api/src/app/admin/admin.service.ts +++ b/apps/api/src/app/admin/admin.service.ts @@ -347,7 +347,12 @@ export class AdminService { } const marketDataItems = await this.prismaService.marketData.findMany({ - select: { date: true, marketPrice: true, state: true }, + select: { + date: true, + isCarriedForward: true, + marketPrice: true, + state: true + }, where: { dataSource: sourceAssetProfileIdentifier.dataSource, symbol: sourceAssetProfileIdentifier.symbol @@ -370,15 +375,18 @@ export class AdminService { where: { id: targetAssetProfile.id } }), this.prismaService.marketData.createMany({ - data: marketDataItems.map(({ date, marketPrice, state }) => { - return { - date, - marketPrice, - state, - dataSource: targetAssetProfileIdentifier.dataSource, - symbol: targetAssetProfileIdentifier.symbol - }; - }), + data: marketDataItems.map( + ({ date, isCarriedForward, marketPrice, state }) => { + return { + date, + isCarriedForward, + marketPrice, + state, + dataSource: targetAssetProfileIdentifier.dataSource, + symbol: targetAssetProfileIdentifier.symbol + }; + } + ), skipDuplicates: true }), // The market data has no relation to the asset profile and is therefore diff --git a/apps/api/src/app/portfolio/current-rate.service.spec.ts b/apps/api/src/app/portfolio/current-rate.service.spec.ts index 5f2358679..ea93e8b4f 100644 --- a/apps/api/src/app/portfolio/current-rate.service.spec.ts +++ b/apps/api/src/app/portfolio/current-rate.service.spec.ts @@ -20,6 +20,7 @@ jest.mock('@ghostfolio/api/services/market-data/market-data.service', () => { createdAt: date, dataSource: DataSource.YAHOO, id: 'aefcbe3a-ee10-4c4f-9f2d-8ffad7b05584', + isCarriedForward: false, marketPrice: 1847.839966, state: 'CLOSE' }); @@ -39,6 +40,7 @@ jest.mock('@ghostfolio/api/services/market-data/market-data.service', () => { dataSource: assetProfileIdentifiers[0].dataSource, date: dateQuery.gte, id: '8fa48fde-f397-4b0d-adbc-fb940e830e6d', + isCarriedForward: false, marketPrice: 1841.823902, state: 'CLOSE', symbol: assetProfileIdentifiers[0].symbol @@ -48,6 +50,7 @@ jest.mock('@ghostfolio/api/services/market-data/market-data.service', () => { dataSource: assetProfileIdentifiers[0].dataSource, date: dateQuery.lt, id: '082d6893-df27-4c91-8a5d-092e84315b56', + isCarriedForward: false, marketPrice: 1847.839966, state: 'CLOSE', symbol: assetProfileIdentifiers[0].symbol diff --git a/apps/api/src/services/market-data/market-data.service.spec.ts b/apps/api/src/services/market-data/market-data.service.spec.ts new file mode 100644 index 000000000..73a7cd929 --- /dev/null +++ b/apps/api/src/services/market-data/market-data.service.spec.ts @@ -0,0 +1,118 @@ +import { parseDate } from '@ghostfolio/common/helper'; + +import { MarketDataService } from './market-data.service'; + +describe('MarketDataService', () => { + let marketDataService: MarketDataService; + let prismaService: { + $transaction: jest.Mock; + marketData: { + createMany: jest.Mock; + deleteMany: jest.Mock; + upsert: jest.Mock; + }; + }; + + beforeEach(() => { + prismaService = { + $transaction: jest.fn(), + marketData: { + createMany: jest.fn(), + deleteMany: jest.fn(), + upsert: jest.fn().mockResolvedValue({}) + } + }; + + marketDataService = new MarketDataService(prismaService as any); + }); + + describe('updateMany', () => { + it('does not drop isCarriedForward', async () => { + prismaService.$transaction.mockImplementation((promises) => { + return Promise.all(promises); + }); + + await marketDataService.updateMany({ + data: [ + { + dataSource: 'YAHOO', + date: parseDate('2026-08-22'), + isCarriedForward: true, + marketPrice: 100, + state: 'CLOSE', + symbol: 'AAPL' + } + ] + }); + + expect(prismaService.marketData.upsert).toHaveBeenCalledWith( + expect.objectContaining({ + create: expect.objectContaining({ isCarriedForward: true }), + update: expect.objectContaining({ isCarriedForward: true }) + }) + ); + }); + + it('resets isCarriedForward if it is omitted by the caller', async () => { + prismaService.$transaction.mockImplementation((promises) => { + return Promise.all(promises); + }); + + await marketDataService.updateMany({ + data: [ + { + dataSource: 'YAHOO', + date: parseDate('2026-08-22'), + marketPrice: 100, + state: 'CLOSE', + symbol: 'AAPL' + } + ] + }); + + expect(prismaService.marketData.upsert).toHaveBeenCalledWith( + expect.objectContaining({ + update: expect.objectContaining({ isCarriedForward: false }) + }) + ); + }); + }); + + describe('replaceForSymbol', () => { + it('does not drop isCarriedForward', async () => { + prismaService.$transaction.mockImplementation((callback) => { + return callback(prismaService); + }); + + await marketDataService.replaceForSymbol({ + data: [ + { + dataSource: 'YAHOO', + date: parseDate('2026-08-21'), + isCarriedForward: false, + marketPrice: 100, + state: 'CLOSE', + symbol: 'AAPL' + }, + { + dataSource: 'YAHOO', + date: parseDate('2026-08-22'), + isCarriedForward: true, + marketPrice: 100, + state: 'CLOSE', + symbol: 'AAPL' + } + ], + dataSource: 'YAHOO', + symbol: 'AAPL' + }); + + const { data } = prismaService.marketData.createMany.mock.calls[0][0]; + + expect(data).toEqual([ + expect.objectContaining({ isCarriedForward: false }), + expect.objectContaining({ isCarriedForward: true }) + ]); + }); + }); +}); diff --git a/apps/api/src/services/market-data/market-data.service.ts b/apps/api/src/services/market-data/market-data.service.ts index ad388ce5c..e608decca 100644 --- a/apps/api/src/services/market-data/market-data.service.ts +++ b/apps/api/src/services/market-data/market-data.service.ts @@ -2,7 +2,6 @@ import { DateQuery } from '@ghostfolio/api/app/portfolio/interfaces/date-query.i import { DataGatheringItem } from '@ghostfolio/api/services/interfaces/interfaces'; import { PrismaService } from '@ghostfolio/api/services/prisma/prisma.service'; import { DEFAULT_PROCESSOR_GATHER_HISTORICAL_MARKET_DATA_TIMEOUT } from '@ghostfolio/common/config'; -import { UpdateMarketDataDto } from '@ghostfolio/common/dtos'; import { resetHours } from '@ghostfolio/common/helper'; import { AssetProfileIdentifier } from '@ghostfolio/common/interfaces'; @@ -189,13 +188,16 @@ export class MarketDataService { }); await prisma.marketData.createMany({ - data: data.map(({ date, marketPrice, state }) => ({ - dataSource, - symbol, - date: date as Date, - marketPrice: marketPrice as number, - state: state as MarketDataState - })), + data: data.map( + ({ date, isCarriedForward, marketPrice, state }) => ({ + dataSource, + symbol, + date: date as Date, + isCarriedForward: isCarriedForward as boolean, + marketPrice: marketPrice as number, + state: state as MarketDataState + }) + ), skipDuplicates: true }); } @@ -220,27 +222,6 @@ export class MarketDataService { }); } - public async updateMarketData(params: { - data: { - state: MarketDataState; - } & UpdateMarketDataDto; - where: Prisma.MarketDataWhereUniqueInput; - }): Promise { - const { data, where } = params; - - return this.prismaService.marketData.upsert({ - where, - create: { - dataSource: where.dataSource_date_symbol.dataSource, - date: where.dataSource_date_symbol.date, - marketPrice: data.marketPrice, - state: data.state, - symbol: where.dataSource_date_symbol.symbol - }, - update: { marketPrice: data.marketPrice, state: data.state } - }); - } - /** * Upsert market data by imitating missing upsertMany functionality * with $transaction @@ -251,16 +232,18 @@ export class MarketDataService { data: Prisma.MarketDataUpdateInput[]; }): Promise { const upsertPromises = data.map( - ({ dataSource, date, marketPrice, symbol, state }) => { + ({ dataSource, date, isCarriedForward, marketPrice, symbol, state }) => { return this.prismaService.marketData.upsert({ create: { dataSource: dataSource as DataSource, date: date as Date, + isCarriedForward: isCarriedForward as boolean, marketPrice: marketPrice as number, state: state as MarketDataState, symbol: symbol as string }, update: { + isCarriedForward: (isCarriedForward ?? false) as boolean, marketPrice: marketPrice as number, state: state as MarketDataState }, diff --git a/apps/api/src/services/queues/data-gathering/data-gathering.processor.spec.ts b/apps/api/src/services/queues/data-gathering/data-gathering.processor.spec.ts new file mode 100644 index 000000000..3faeb4d6b --- /dev/null +++ b/apps/api/src/services/queues/data-gathering/data-gathering.processor.spec.ts @@ -0,0 +1,266 @@ +import { DataGatheringItem } from '@ghostfolio/api/services/interfaces/interfaces'; +import { + getAssetProfileIdentifier, + parseDate +} from '@ghostfolio/common/helper'; + +import { DataSource } from '@prisma/client'; +import { Job } from 'bull'; + +import { DataGatheringProcessor } from './data-gathering.processor'; + +describe('DataGatheringProcessor', () => { + let dataGatheringProcessor: DataGatheringProcessor; + let dataProviderService: { getHistoricalRaw: jest.Mock }; + let marketDataService: { replaceForSymbol: jest.Mock; updateMany: jest.Mock }; + + const createJob = ({ + dataSource, + date, + symbol + }: { + dataSource: DataSource; + date: string; + symbol: string; + }) => { + return { + data: { + dataSource, + symbol, + date: parseDate(date).toISOString() + } + } as unknown as Job; + }; + + const mockHistoricalData = ({ + dataSource, + prices, + symbol + }: { + dataSource: DataSource; + prices: { [date: string]: number }; + symbol: string; + }) => { + const assetProfileIdentifier = getAssetProfileIdentifier({ + dataSource, + symbol + }); + + const historicalData: { + [symbol: string]: { [date: string]: { marketPrice: number } }; + } = { [assetProfileIdentifier]: {} }; + + for (const [date, marketPrice] of Object.entries(prices)) { + historicalData[assetProfileIdentifier][date] = { marketPrice }; + } + + dataProviderService.getHistoricalRaw.mockResolvedValue(historicalData); + }; + + beforeAll(() => { + jest.useFakeTimers().setSystemTime(parseDate('2026-08-24').getTime()); + }); + + beforeEach(() => { + dataProviderService = { getHistoricalRaw: jest.fn() }; + marketDataService = { + replaceForSymbol: jest.fn(), + updateMany: jest.fn() + }; + + dataGatheringProcessor = new DataGatheringProcessor( + null, + dataProviderService as any, + marketDataService as any, + null + ); + }); + + afterAll(() => { + jest.useRealTimers(); + }); + + it('writes an all-real series without carried forward market prices', async () => { + mockHistoricalData({ + dataSource: 'COINGECKO', + symbol: 'bitcoin', + prices: { + '2026-08-17': 1, + '2026-08-18': 2, + '2026-08-19': 3, + '2026-08-20': 4, + '2026-08-21': 5, + '2026-08-22': 6, + '2026-08-23': 7 + } + }); + + await dataGatheringProcessor.gatherHistoricalMarketData( + createJob({ + dataSource: 'COINGECKO', + date: '2026-08-17', + symbol: 'bitcoin' + }) + ); + + const { data } = marketDataService.updateMany.mock.calls[0][0]; + + expect(data).toHaveLength(7); + expect( + data.every(({ isCarriedForward }) => { + return isCarriedForward === false; + }) + ).toBe(true); + }); + + it('fills an interior gap with carried forward market prices', async () => { + mockHistoricalData({ + dataSource: 'YAHOO', + symbol: 'AAPL', + prices: { + '2026-08-17': 1, + '2026-08-18': 2, + '2026-08-21': 5, + '2026-08-22': 6, + '2026-08-23': 7 + } + }); + + await dataGatheringProcessor.gatherHistoricalMarketData( + createJob({ dataSource: 'YAHOO', date: '2026-08-17', symbol: 'AAPL' }) + ); + + const { data } = marketDataService.updateMany.mock.calls[0][0]; + + expect(data).toEqual([ + expect.objectContaining({ + date: parseDate('2026-08-17'), + isCarriedForward: false, + marketPrice: 1 + }), + expect.objectContaining({ + date: parseDate('2026-08-18'), + isCarriedForward: false, + marketPrice: 2 + }), + expect.objectContaining({ + date: parseDate('2026-08-19'), + isCarriedForward: true, + marketPrice: 2 + }), + expect.objectContaining({ + date: parseDate('2026-08-20'), + isCarriedForward: true, + marketPrice: 2 + }), + expect.objectContaining({ + date: parseDate('2026-08-21'), + isCarriedForward: false, + marketPrice: 5 + }), + expect.objectContaining({ + date: parseDate('2026-08-22'), + isCarriedForward: false, + marketPrice: 6 + }), + expect.objectContaining({ + date: parseDate('2026-08-23'), + isCarriedForward: false, + marketPrice: 7 + }) + ]); + }); + + it('fills a trailing gap with carried forward market prices', async () => { + mockHistoricalData({ + dataSource: 'YAHOO', + symbol: 'AAPL', + prices: { + '2026-08-17': 1, + '2026-08-18': 2, + '2026-08-19': 3, + '2026-08-20': 4, + '2026-08-21': 5 + } + }); + + await dataGatheringProcessor.gatherHistoricalMarketData( + createJob({ dataSource: 'YAHOO', date: '2026-08-17', symbol: 'AAPL' }) + ); + + const { data } = marketDataService.updateMany.mock.calls[0][0]; + + expect(data).toHaveLength(7); + expect(data.slice(5)).toEqual([ + expect.objectContaining({ + date: parseDate('2026-08-22'), + isCarriedForward: true, + marketPrice: 5 + }), + expect.objectContaining({ + date: parseDate('2026-08-23'), + isCarriedForward: true, + marketPrice: 5 + }) + ]); + }); + + it('does not fill a leading gap', async () => { + mockHistoricalData({ + dataSource: 'YAHOO', + symbol: 'AAPL', + prices: { + '2026-08-19': 3, + '2026-08-20': 4, + '2026-08-21': 5, + '2026-08-22': 6, + '2026-08-23': 7 + } + }); + + await dataGatheringProcessor.gatherHistoricalMarketData( + createJob({ dataSource: 'YAHOO', date: '2026-08-17', symbol: 'AAPL' }) + ); + + const { data } = marketDataService.updateMany.mock.calls[0][0]; + + expect(data).toHaveLength(5); + expect(data[0]).toEqual( + expect.objectContaining({ + date: parseDate('2026-08-19'), + isCarriedForward: false, + marketPrice: 3 + }) + ); + }); + + it('labels a market price of 0 from the data provider as carried forward', async () => { + mockHistoricalData({ + dataSource: 'YAHOO', + symbol: 'AAPL', + prices: { + '2026-08-17': 1, + '2026-08-18': 2, + '2026-08-19': 0, + '2026-08-20': 4, + '2026-08-21': 5, + '2026-08-22': 6, + '2026-08-23': 7 + } + }); + + await dataGatheringProcessor.gatherHistoricalMarketData( + createJob({ dataSource: 'YAHOO', date: '2026-08-17', symbol: 'AAPL' }) + ); + + const { data } = marketDataService.updateMany.mock.calls[0][0]; + + expect(data[2]).toEqual( + expect.objectContaining({ + date: parseDate('2026-08-19'), + isCarriedForward: true, + marketPrice: 2 + }) + ); + }); +}); diff --git a/apps/api/src/services/queues/data-gathering/data-gathering.processor.ts b/apps/api/src/services/queues/data-gathering/data-gathering.processor.ts index b49b30cc8..f1e8d01cd 100644 --- a/apps/api/src/services/queues/data-gathering/data-gathering.processor.ts +++ b/apps/api/src/services/queues/data-gathering/data-gathering.processor.ts @@ -125,7 +125,6 @@ export class DataGatheringProcessor { const data: Prisma.MarketDataUpdateInput[] = []; let lastMarketPrice: number; - let numberOfMarketDataItemsToKeep = 0; while ( isBefore( @@ -154,24 +153,15 @@ export class DataGatheringProcessor { dataSource, symbol, date: getStartOfUtcDate(currentDate), + isCarriedForward: lastMarketPrice !== marketPriceOfDataProvider, marketPrice: lastMarketPrice, state: 'CLOSE' }); - - if (marketPriceOfDataProvider) { - numberOfMarketDataItemsToKeep = data.length; - } } currentDate = addDays(currentDate, 1); } - // A gap at the end means that the market data is not available yet, in - // contrast to a gap in between, which means that the market was closed. - // Therefore, the market prices which are carried forward after the last - // market price of the data provider are discarded. - data.splice(numberOfMarketDataItemsToKeep); - if (force) { await this.marketDataService.replaceForSymbol({ data, diff --git a/apps/api/src/services/queues/data-gathering/data-gathering.service.spec.ts b/apps/api/src/services/queues/data-gathering/data-gathering.service.spec.ts new file mode 100644 index 000000000..51bf9ef6a --- /dev/null +++ b/apps/api/src/services/queues/data-gathering/data-gathering.service.spec.ts @@ -0,0 +1,223 @@ +import { + DATA_GATHERING_QUEUE_PRIORITY_HIGH, + GATHER_HISTORICAL_MARKET_DATA_COOLDOWN_IN_MS +} from '@ghostfolio/common/config'; +import { parseDate } from '@ghostfolio/common/helper'; + +import { DataGatheringService } from './data-gathering.service'; + +describe('DataGatheringService', () => { + let dataGatheringQueue: { addBulk: jest.Mock; clean: jest.Mock }; + let dataGatheringService: DataGatheringService; + let dataProviderService: { getHistoricalRaw: jest.Mock }; + let prismaService: { marketData: { groupBy: jest.Mock; upsert: jest.Mock } }; + + beforeEach(() => { + dataGatheringQueue = { + addBulk: jest.fn().mockResolvedValue([]), + clean: jest.fn().mockResolvedValue([]) + }; + dataProviderService = { getHistoricalRaw: jest.fn() }; + prismaService = { + marketData: { + groupBy: jest.fn().mockResolvedValue([]), + upsert: jest.fn().mockResolvedValue({}) + } + }; + + dataGatheringService = new DataGatheringService( + null, + dataGatheringQueue as any, + dataProviderService as any, + null, + null, + prismaService as any, + null, + null + ); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + describe('getAssetProfileIdentifiersWithRecentMarketData', () => { + it('excludes carried forward market prices from the query', async () => { + jest.useFakeTimers().setSystemTime(parseDate('2026-08-23').getTime()); + + await dataGatheringService[ + 'getAssetProfileIdentifiersWithRecentMarketData' + ](); + + expect(prismaService.marketData.groupBy).toHaveBeenCalledWith( + expect.objectContaining({ + where: expect.objectContaining({ + isCarriedForward: false, + state: 'CLOSE' + }) + }) + ); + }); + + it('keeps a cryptocurrency with a real market price of yesterday on Sunday', async () => { + jest.useFakeTimers().setSystemTime(parseDate('2026-08-23').getTime()); + + prismaService.marketData.groupBy.mockResolvedValue([ + { + _max: { date: parseDate('2026-08-22') }, + dataSource: 'COINGECKO', + symbol: 'bitcoin' + }, + { + _max: { date: parseDate('2026-08-21') }, + dataSource: 'YAHOO', + symbol: 'AAPL' + } + ]); + + const assetProfileIdentifiers = + await dataGatheringService[ + 'getAssetProfileIdentifiersWithRecentMarketData' + ](); + + expect(assetProfileIdentifiers).toEqual([ + { dataSource: 'COINGECKO', symbol: 'bitcoin' } + ]); + }); + + it('drops a stock with a real market price of Friday on Monday', async () => { + jest.useFakeTimers().setSystemTime(parseDate('2026-08-24').getTime()); + + prismaService.marketData.groupBy.mockResolvedValue([ + { + _max: { date: parseDate('2026-08-23') }, + dataSource: 'COINGECKO', + symbol: 'bitcoin' + }, + { + _max: { date: parseDate('2026-08-21') }, + dataSource: 'YAHOO', + symbol: 'AAPL' + } + ]); + + const assetProfileIdentifiers = + await dataGatheringService[ + 'getAssetProfileIdentifiersWithRecentMarketData' + ](); + + expect(assetProfileIdentifiers).toEqual([ + { dataSource: 'COINGECKO', symbol: 'bitcoin' } + ]); + }); + + it('drops a stock with a late Friday close on Saturday', async () => { + jest.useFakeTimers().setSystemTime(parseDate('2026-08-22').getTime()); + + prismaService.marketData.groupBy.mockResolvedValue([ + { + _max: { date: parseDate('2026-08-20') }, + dataSource: 'YAHOO', + symbol: 'AAPL' + }, + { + _max: { date: parseDate('2026-08-21') }, + dataSource: 'YAHOO', + symbol: 'MSFT' + } + ]); + + const assetProfileIdentifiers = + await dataGatheringService[ + 'getAssetProfileIdentifiersWithRecentMarketData' + ](); + + expect(assetProfileIdentifiers).toEqual([ + { dataSource: 'YAHOO', symbol: 'MSFT' } + ]); + }); + }); + + describe('gatherRecentMarketData', () => { + it('expires completed jobs which are older than the cooldown', async () => { + jest + .spyOn(dataGatheringService as any, 'getCurrencies7D') + .mockResolvedValue([]); + jest + .spyOn(dataGatheringService as any, 'getSymbols7D') + .mockResolvedValue([]); + + await dataGatheringService.gatherRecentMarketData(); + + expect(dataGatheringQueue.clean).toHaveBeenCalledWith( + GATHER_HISTORICAL_MARKET_DATA_COOLDOWN_IN_MS, + 'completed' + ); + }); + + it('retains its completed jobs for the duration of the cooldown', async () => { + jest + .spyOn(dataGatheringService as any, 'getCurrencies7D') + .mockResolvedValue([ + { + dataSource: 'YAHOO', + date: parseDate('2026-08-01'), + symbol: 'AAPL' + } + ]); + jest + .spyOn(dataGatheringService as any, 'getSymbols7D') + .mockResolvedValue([]); + + await dataGatheringService.gatherRecentMarketData(); + + const [jobs] = dataGatheringQueue.addBulk.mock.calls[0]; + + expect(jobs[0].opts.removeOnComplete).toEqual({ + age: GATHER_HISTORICAL_MARKET_DATA_COOLDOWN_IN_MS / 1000 + }); + }); + }); + + describe('gatherSymbols', () => { + it('does not apply the cooldown to a manually triggered gathering', async () => { + await dataGatheringService.gatherSymbols({ + dataGatheringItems: [ + { + dataSource: 'YAHOO', + date: parseDate('2026-08-01'), + symbol: 'AAPL' + } + ], + priority: DATA_GATHERING_QUEUE_PRIORITY_HIGH + }); + + const [jobs] = dataGatheringQueue.addBulk.mock.calls[0]; + + expect(jobs[0].opts.removeOnComplete).toBe(true); + }); + }); + + describe('gatherSymbolForDate', () => { + it('resets isCarriedForward on a previously carried forward market price', async () => { + dataProviderService.getHistoricalRaw.mockResolvedValue({ + 'YAHOO-AAPL': { + '2026-08-22': { marketPrice: 100 } + } + }); + + await dataGatheringService.gatherSymbolForDate({ + dataSource: 'YAHOO', + date: parseDate('2026-08-22'), + symbol: 'AAPL' + }); + + expect(prismaService.marketData.upsert).toHaveBeenCalledWith( + expect.objectContaining({ + create: expect.objectContaining({ isCarriedForward: false }), + update: { marketPrice: 100, isCarriedForward: false } + }) + ); + }); + }); +}); 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 9596b7614..531f36f0d 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 @@ -11,6 +11,7 @@ import { DATA_GATHERING_QUEUE_PRIORITY_HIGH, DATA_GATHERING_QUEUE_PRIORITY_LOW, DATA_GATHERING_QUEUE_PRIORITY_MEDIUM, + GATHER_HISTORICAL_MARKET_DATA_COOLDOWN_IN_MS, GATHER_HISTORICAL_MARKET_DATA_PROCESS_JOB_NAME, GATHER_HISTORICAL_MARKET_DATA_PROCESS_JOB_OPTIONS, PROPERTY_BENCHMARKS @@ -30,7 +31,14 @@ import { InjectQueue } from '@nestjs/bull'; import { Inject, Injectable, Logger } from '@nestjs/common'; import { Prisma } from '@prisma/client'; import { Job, JobOptions, Queue } from 'bull'; -import { format, min, subDays, subMilliseconds, subYears } from 'date-fns'; +import { + format, + isBefore, + min, + subDays, + subMilliseconds, + subYears +} from 'date-fns'; import { isEmpty } from 'lodash'; import ms, { StringValue } from 'ms'; @@ -252,12 +260,23 @@ export class DataGatheringService { } public async gatherRecentMarketData() { + await this.dataGatheringQueue.clean( + GATHER_HISTORICAL_MARKET_DATA_COOLDOWN_IN_MS, + 'completed' + ); + + const removeOnComplete = { + age: GATHER_HISTORICAL_MARKET_DATA_COOLDOWN_IN_MS / 1000 + }; + await this.gatherSymbols({ + removeOnComplete, dataGatheringItems: await this.getCurrencies7D(), priority: DATA_GATHERING_QUEUE_PRIORITY_HIGH }); await this.gatherSymbols({ + removeOnComplete, dataGatheringItems: await this.getSymbols7D({ withUserSubscription: true }), @@ -265,6 +284,7 @@ export class DataGatheringService { }); await this.gatherSymbols({ + removeOnComplete, dataGatheringItems: await this.getSymbols7D({ withUserSubscription: false }), @@ -320,9 +340,10 @@ export class DataGatheringService { dataSource, date, marketPrice, - symbol + symbol, + isCarriedForward: false }, - update: { marketPrice }, + update: { marketPrice, isCarriedForward: false }, where: { dataSource_date_symbol: { dataSource, date, symbol } } }); } @@ -336,11 +357,13 @@ export class DataGatheringService { public async gatherSymbols({ dataGatheringItems, force = false, - priority + priority, + removeOnComplete = GATHER_HISTORICAL_MARKET_DATA_PROCESS_JOB_OPTIONS.removeOnComplete }: { dataGatheringItems: DataGatheringItem[]; force?: boolean; priority: number; + removeOnComplete?: JobOptions['removeOnComplete']; }): Promise { return this.addJobsToQueue( dataGatheringItems.map(({ dataSource, date, symbol }) => { @@ -355,6 +378,7 @@ export class DataGatheringService { opts: { ...GATHER_HISTORICAL_MARKET_DATA_PROCESS_JOB_OPTIONS, priority, + removeOnComplete, jobId: `${getAssetProfileIdentifier({ dataSource, symbol @@ -397,22 +421,23 @@ export class DataGatheringService { }); } - private async getAssetProfileIdentifiersWithCompleteMarketData(): Promise< + private async getAssetProfileIdentifiersWithRecentMarketData(): Promise< AssetProfileIdentifier[] > { return ( await this.prismaService.marketData.groupBy({ - _count: true, + _max: { date: true }, by: ['dataSource', 'symbol'], orderBy: [{ symbol: 'asc' }], where: { date: { gt: subDays(resetHours(new Date()), 7) }, + isCarriedForward: false, state: 'CLOSE' } }) ) - .filter(({ _count }) => { - return _count >= 6; + .filter(({ _max }) => { + return !isBefore(_max.date, getStartOfUtcDate(subDays(new Date(), 1))); }) .map(({ dataSource, symbol }) => { return { dataSource, symbol }; @@ -420,13 +445,13 @@ export class DataGatheringService { } private async getCurrencies7D(): Promise { - const assetProfileIdentifiersWithCompleteMarketData = - await this.getAssetProfileIdentifiersWithCompleteMarketData(); + const assetProfileIdentifiersWithRecentMarketData = + await this.getAssetProfileIdentifiersWithRecentMarketData(); return this.exchangeRateDataService .getCurrencyPairs() .filter(({ dataSource, symbol }) => { - return !assetProfileIdentifiersWithCompleteMarketData.some((item) => { + return !assetProfileIdentifiersWithRecentMarketData.some((item) => { return item.dataSource === dataSource && item.symbol === symbol; }); }) @@ -485,8 +510,8 @@ export class DataGatheringService { } ); - const assetProfileIdentifiersWithCompleteMarketData = - await this.getAssetProfileIdentifiersWithCompleteMarketData(); + const assetProfileIdentifiersWithRecentMarketData = + await this.getAssetProfileIdentifiersWithRecentMarketData(); return symbolProfiles .filter(({ dataSource, scraperConfiguration, symbol }) => { @@ -494,7 +519,7 @@ export class DataGatheringService { dataSource === 'MANUAL' && !isEmpty(scraperConfiguration); return ( - !assetProfileIdentifiersWithCompleteMarketData.some((item) => { + !assetProfileIdentifiersWithRecentMarketData.some((item) => { return item.dataSource === dataSource && item.symbol === symbol; }) && (dataSource !== 'MANUAL' || manualDataSourceWithScraperConfiguration) diff --git a/libs/common/src/lib/config.ts b/libs/common/src/lib/config.ts index 2f914eebf..04186cfa9 100644 --- a/libs/common/src/lib/config.ts +++ b/libs/common/src/lib/config.ts @@ -203,6 +203,7 @@ export const GATHER_ASSET_PROFILE_PROCESS_JOB_OPTIONS: JobOptions = { removeOnComplete: true }; +export const GATHER_HISTORICAL_MARKET_DATA_COOLDOWN_IN_MS = ms('12 hours'); export const GATHER_HISTORICAL_MARKET_DATA_PROCESS_JOB_NAME = 'GATHER_HISTORICAL_MARKET_DATA'; export const GATHER_HISTORICAL_MARKET_DATA_PROCESS_JOB_OPTIONS: JobOptions = { diff --git a/prisma/migrations/20260824000000_added_is_carried_forward_to_market_data/migration.sql b/prisma/migrations/20260824000000_added_is_carried_forward_to_market_data/migration.sql new file mode 100644 index 000000000..1de434fe1 --- /dev/null +++ b/prisma/migrations/20260824000000_added_is_carried_forward_to_market_data/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "MarketData" ADD COLUMN "isCarriedForward" BOOLEAN NOT NULL DEFAULT false; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index c0dd25a60..600160124 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -157,13 +157,14 @@ model AuthDevice { } model MarketData { - createdAt DateTime @default(now()) - dataSource DataSource - date DateTime - id String @id @default(uuid()) - marketPrice Float - state MarketDataState @default(CLOSE) - symbol String + createdAt DateTime @default(now()) + dataSource DataSource + date DateTime + id String @id @default(uuid()) + isCarriedForward Boolean @default(false) + marketPrice Float + state MarketDataState @default(CLOSE) + symbol String @@unique([dataSource, date, symbol]) @@index([dataSource])