Browse Source

Fix repeated historical market data gathering on weekends

pull/7710/head
Thomas Kaul 5 days ago
parent
commit
981648e230
  1. 28
      apps/api/src/app/admin/admin.service.ts
  2. 46
      apps/api/src/services/market-data/market-data.service.spec.ts
  3. 2
      apps/api/src/services/market-data/market-data.service.ts
  4. 30
      apps/api/src/services/queues/data-gathering/data-gathering.processor.spec.ts
  5. 2
      apps/api/src/services/queues/data-gathering/data-gathering.processor.ts
  6. 46
      apps/api/src/services/queues/data-gathering/data-gathering.service.spec.ts
  7. 12
      apps/api/src/services/queues/data-gathering/data-gathering.service.ts
  8. 4
      libs/common/src/lib/config.ts

28
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

46
apps/api/src/services/market-data/market-data.service.spec.ts

@ -52,6 +52,30 @@ describe('MarketDataService', () => {
})
);
});
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', () => {
@ -91,26 +115,4 @@ describe('MarketDataService', () => {
]);
});
});
describe('updateMarketData', () => {
it('resets isCarriedForward for a manually edited market price', async () => {
await marketDataService.updateMarketData({
data: { marketPrice: 100, state: 'CLOSE' },
where: {
dataSource_date_symbol: {
dataSource: 'YAHOO',
date: parseDate('2026-08-22'),
symbol: 'AAPL'
}
}
});
expect(prismaService.marketData.upsert).toHaveBeenCalledWith(
expect.objectContaining({
create: expect.objectContaining({ isCarriedForward: false }),
update: expect.objectContaining({ isCarriedForward: false })
})
);
});
});
});

2
apps/api/src/services/market-data/market-data.service.ts

@ -243,7 +243,7 @@ export class MarketDataService {
symbol: symbol as string
},
update: {
isCarriedForward: isCarriedForward as boolean,
isCarriedForward: (isCarriedForward ?? false) as boolean,
marketPrice: marketPrice as number,
state: state as MarketDataState
},

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

@ -233,4 +233,34 @@ describe('DataGatheringProcessor', () => {
})
);
});
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
})
);
});
});

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

@ -153,7 +153,7 @@ export class DataGatheringProcessor {
dataSource,
symbol,
date: getStartOfUtcDate(currentDate),
isCarriedForward: !marketPriceOfDataProvider,
isCarriedForward: lastMarketPrice !== marketPriceOfDataProvider,
marketPrice: lastMarketPrice,
state: 'CLOSE'
});

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

@ -1,6 +1,6 @@
import {
GATHER_HISTORICAL_MARKET_DATA_COOLDOWN_IN_MS,
GATHER_HISTORICAL_MARKET_DATA_PROCESS_JOB_OPTIONS
DATA_GATHERING_QUEUE_PRIORITY_HIGH,
GATHER_HISTORICAL_MARKET_DATA_COOLDOWN_IN_MS
} from '@ghostfolio/common/config';
import { parseDate } from '@ghostfolio/common/helper';
@ -155,15 +155,49 @@ describe('DataGatheringService', () => {
);
});
it('retains completed jobs for the duration of the cooldown', () => {
expect(
GATHER_HISTORICAL_MARKET_DATA_PROCESS_JOB_OPTIONS.removeOnComplete
).toEqual({
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({

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

@ -265,12 +265,18 @@ export class DataGatheringService {
'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
}),
@ -278,6 +284,7 @@ export class DataGatheringService {
});
await this.gatherSymbols({
removeOnComplete,
dataGatheringItems: await this.getSymbols7D({
withUserSubscription: false
}),
@ -350,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<Job[]> {
return this.addJobsToQueue(
dataGatheringItems.map(({ dataSource, date, symbol }) => {
@ -369,6 +378,7 @@ export class DataGatheringService {
opts: {
...GATHER_HISTORICAL_MARKET_DATA_PROCESS_JOB_OPTIONS,
priority,
removeOnComplete,
jobId: `${getAssetProfileIdentifier({
dataSource,
symbol

4
libs/common/src/lib/config.ts

@ -212,9 +212,7 @@ export const GATHER_HISTORICAL_MARKET_DATA_PROCESS_JOB_OPTIONS: JobOptions = {
delay: ms('1 minute'),
type: 'exponential'
},
removeOnComplete: {
age: GATHER_HISTORICAL_MARKET_DATA_COOLDOWN_IN_MS / 1000
}
removeOnComplete: true
};
export const GATHER_STATISTICS_PROCESS_JOB_OPTIONS: JobOptions = {

Loading…
Cancel
Save