Bugfix/repeated historical market data gathering on weekends (#7710)

* Fix repeated historical market data gathering on weekends

* Update changelog

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

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 }) => {
+        data: marketDataItems.map(
-          return {
+          ({ date, isCarriedForward, marketPrice, state }) => {
-            date,
+            return {
-            marketPrice,
+              date,
-            state,
+              isCarriedForward,
-            dataSource: targetAssetProfileIdentifier.dataSource,
+              marketPrice,
-            symbol: targetAssetProfileIdentifier.symbol
+              state,
-          };
+              dataSource: targetAssetProfileIdentifier.dataSource,
-        }),
+              symbol: targetAssetProfileIdentifier.symbol
+            };
+          }
+        ),
         skipDuplicates: true
       }),
       // The market data has no relation to the asset profile and is therefore

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

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 })
+      ]);
+    });
+  });
+});

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 }) => ({
+            data: data.map(
-              dataSource,
+              ({ date, isCarriedForward, marketPrice, state }) => ({
-              symbol,
+                dataSource,
-              date: date as Date,
+                symbol,
-              marketPrice: marketPrice as number,
+                date: date as Date,
-              state: state as MarketDataState
+                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<MarketData> {
-    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<MarketData[]> {
     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
           },

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<DataGatheringItem>;
+  };
+
+  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
+      })
+    );
+  });
+});

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,

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 }
+        })
+      );
+    });
+  });
+});

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<Job[]> {
     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 }) => {
+      .filter(({ _max }) => {
-        return _count >= 6;
+        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<DataGatheringItem[]> {
-    const assetProfileIdentifiersWithCompleteMarketData =
+    const assetProfileIdentifiersWithRecentMarketData =
-      await this.getAssetProfileIdentifiersWithCompleteMarketData();
+      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 =
+    const assetProfileIdentifiersWithRecentMarketData =
-      await this.getAssetProfileIdentifiersWithCompleteMarketData();
+      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)

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 = {

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;

prisma/schema.prisma

@@ -157,13 +157,14 @@ model AuthDevice {
 }
 
 model MarketData {
-  createdAt   DateTime        @default(now())
+  createdAt        DateTime        @default(now())
-  dataSource  DataSource
+  dataSource       DataSource
-  date        DateTime
+  date             DateTime
-  id          String          @id @default(uuid())
+  id               String          @id @default(uuid())
-  marketPrice Float
+  isCarriedForward Boolean         @default(false)
-  state       MarketDataState @default(CLOSE)
+  marketPrice      Float
-  symbol      String
+  state            MarketDataState @default(CLOSE)
+  symbol           String
 
   @@unique([dataSource, date, symbol])
   @@index([dataSource])