Browse Source

Merge 777f36cad3 into 1dab25c7dc

pull/7029/merge
Thomas Kaul 3 months ago
committed by GitHub
parent
commit
745a2c3b79
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
  1. 6
      CHANGELOG.md
  2. 4
      apps/api/src/app/admin/queue/queue.module.ts
  3. 36
      apps/api/src/app/admin/queue/queue.service.ts
  4. 2
      apps/api/src/app/app.module.ts
  5. 6
      apps/api/src/app/endpoints/watchlist/interfaces/watchlist-value.interface.ts
  6. 10
      apps/api/src/app/endpoints/watchlist/watchlist.module.ts
  7. 135
      apps/api/src/app/endpoints/watchlist/watchlist.service.ts
  8. 4
      apps/api/src/app/redis-cache/redis-cache.service.ts
  9. 3
      apps/api/src/services/queues/watchlist/interfaces/watchlist-queue-job.interface.ts
  10. 31
      apps/api/src/services/queues/watchlist/watchlist-computation.service.ts
  11. 52
      apps/api/src/services/queues/watchlist/watchlist.module.ts
  12. 152
      apps/api/src/services/queues/watchlist/watchlist.processor.ts
  13. 11
      libs/common/src/lib/config.ts

6
CHANGELOG.md

@ -5,6 +5,12 @@ 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/), 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). and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## Unreleased
### Changed
- Improved the performance of the watchlist by caching its data and computing it via a queue
## 3.10.0 - 2026-06-13 ## 3.10.0 - 2026-06-13
### Changed ### Changed

4
apps/api/src/app/admin/queue/queue.module.ts

@ -1,6 +1,7 @@
import { DataGatheringQueueModule } from '@ghostfolio/api/services/queues/data-gathering/data-gathering.module'; import { DataGatheringQueueModule } from '@ghostfolio/api/services/queues/data-gathering/data-gathering.module';
import { PortfolioSnapshotQueueModule } from '@ghostfolio/api/services/queues/portfolio-snapshot/portfolio-snapshot.module'; import { PortfolioSnapshotQueueModule } from '@ghostfolio/api/services/queues/portfolio-snapshot/portfolio-snapshot.module';
import { StatisticsGatheringQueueModule } from '@ghostfolio/api/services/queues/statistics-gathering/statistics-gathering.module'; import { StatisticsGatheringQueueModule } from '@ghostfolio/api/services/queues/statistics-gathering/statistics-gathering.module';
import { WatchlistComputationQueueModule } from '@ghostfolio/api/services/queues/watchlist/watchlist.module';
import { Module } from '@nestjs/common'; import { Module } from '@nestjs/common';
@ -12,7 +13,8 @@ import { QueueService } from './queue.service';
imports: [ imports: [
DataGatheringQueueModule, DataGatheringQueueModule,
PortfolioSnapshotQueueModule, PortfolioSnapshotQueueModule,
StatisticsGatheringQueueModule StatisticsGatheringQueueModule,
WatchlistComputationQueueModule
], ],
providers: [QueueService] providers: [QueueService]
}) })

36
apps/api/src/app/admin/queue/queue.service.ts

@ -2,7 +2,8 @@ import {
DATA_GATHERING_QUEUE, DATA_GATHERING_QUEUE,
PORTFOLIO_SNAPSHOT_COMPUTATION_QUEUE, PORTFOLIO_SNAPSHOT_COMPUTATION_QUEUE,
QUEUE_JOB_STATUS_LIST, QUEUE_JOB_STATUS_LIST,
STATISTICS_GATHERING_QUEUE STATISTICS_GATHERING_QUEUE,
WATCHLIST_COMPUTATION_QUEUE
} from '@ghostfolio/common/config'; } from '@ghostfolio/common/config';
import { AdminJobs } from '@ghostfolio/common/interfaces'; import { AdminJobs } from '@ghostfolio/common/interfaces';
@ -18,7 +19,9 @@ export class QueueService {
@InjectQueue(PORTFOLIO_SNAPSHOT_COMPUTATION_QUEUE) @InjectQueue(PORTFOLIO_SNAPSHOT_COMPUTATION_QUEUE)
private readonly portfolioSnapshotQueue: Queue, private readonly portfolioSnapshotQueue: Queue,
@InjectQueue(STATISTICS_GATHERING_QUEUE) @InjectQueue(STATISTICS_GATHERING_QUEUE)
private readonly statisticsGatheringQueue: Queue private readonly statisticsGatheringQueue: Queue,
@InjectQueue(WATCHLIST_COMPUTATION_QUEUE)
private readonly watchlistComputationQueue: Queue
) {} ) {}
public async deleteJob(aId: string) { public async deleteJob(aId: string) {
@ -28,6 +31,10 @@ export class QueueService {
job = await this.portfolioSnapshotQueue.getJob(aId); job = await this.portfolioSnapshotQueue.getJob(aId);
} }
if (!job) {
job = await this.watchlistComputationQueue.getJob(aId);
}
return job?.remove(); return job?.remove();
} }
@ -42,6 +49,7 @@ export class QueueService {
await this.dataGatheringQueue.clean(300, queueStatus); await this.dataGatheringQueue.clean(300, queueStatus);
await this.portfolioSnapshotQueue.clean(300, queueStatus); await this.portfolioSnapshotQueue.clean(300, queueStatus);
await this.statisticsGatheringQueue.clean(300, queueStatus); await this.statisticsGatheringQueue.clean(300, queueStatus);
await this.watchlistComputationQueue.clean(300, queueStatus);
} }
} }
@ -52,6 +60,10 @@ export class QueueService {
job = await this.portfolioSnapshotQueue.getJob(aId); job = await this.portfolioSnapshotQueue.getJob(aId);
} }
if (!job) {
job = await this.watchlistComputationQueue.getJob(aId);
}
return job?.promote(); return job?.promote();
} }
@ -62,18 +74,24 @@ export class QueueService {
limit?: number; limit?: number;
status?: JobStatus[]; status?: JobStatus[];
}): Promise<AdminJobs> { }): Promise<AdminJobs> {
const [dataGatheringJobs, portfolioSnapshotJobs, statisticsGatheringJobs] = const [
await Promise.all([ dataGatheringJobs,
this.dataGatheringQueue.getJobs(status), portfolioSnapshotJobs,
this.portfolioSnapshotQueue.getJobs(status), statisticsGatheringJobs,
this.statisticsGatheringQueue.getJobs(status) watchlistComputationJobs
]); ] = await Promise.all([
this.dataGatheringQueue.getJobs(status),
this.portfolioSnapshotQueue.getJobs(status),
this.statisticsGatheringQueue.getJobs(status),
this.watchlistComputationQueue.getJobs(status)
]);
const jobsWithState = await Promise.all( const jobsWithState = await Promise.all(
[ [
...dataGatheringJobs, ...dataGatheringJobs,
...portfolioSnapshotJobs, ...portfolioSnapshotJobs,
...statisticsGatheringJobs ...statisticsGatheringJobs,
...watchlistComputationJobs
] ]
.filter((job) => { .filter((job) => {
return job; return job;

2
apps/api/src/app/app.module.ts

@ -10,6 +10,7 @@ import { PrismaModule } from '@ghostfolio/api/services/prisma/prisma.module';
import { PropertyModule } from '@ghostfolio/api/services/property/property.module'; import { PropertyModule } from '@ghostfolio/api/services/property/property.module';
import { DataGatheringQueueModule } from '@ghostfolio/api/services/queues/data-gathering/data-gathering.module'; import { DataGatheringQueueModule } from '@ghostfolio/api/services/queues/data-gathering/data-gathering.module';
import { PortfolioSnapshotQueueModule } from '@ghostfolio/api/services/queues/portfolio-snapshot/portfolio-snapshot.module'; import { PortfolioSnapshotQueueModule } from '@ghostfolio/api/services/queues/portfolio-snapshot/portfolio-snapshot.module';
import { WatchlistComputationQueueModule } from '@ghostfolio/api/services/queues/watchlist/watchlist.module';
import { import {
BULL_BOARD_ROUTE, BULL_BOARD_ROUTE,
DEFAULT_LANGUAGE_CODE, DEFAULT_LANGUAGE_CODE,
@ -167,6 +168,7 @@ import { UserModule } from './user/user.module';
SymbolModule, SymbolModule,
TagsModule, TagsModule,
UserModule, UserModule,
WatchlistComputationQueueModule,
WatchlistModule WatchlistModule
], ],
providers: [I18nService] providers: [I18nService]

6
apps/api/src/app/endpoints/watchlist/interfaces/watchlist-value.interface.ts

@ -0,0 +1,6 @@
import { WatchlistResponse } from '@ghostfolio/common/interfaces';
export interface WatchlistValue {
expiration: number;
watchlist: WatchlistResponse['watchlist'];
}

10
apps/api/src/app/endpoints/watchlist/watchlist.module.ts

@ -1,11 +1,11 @@
import { RedisCacheModule } from '@ghostfolio/api/app/redis-cache/redis-cache.module';
import { TransformDataSourceInRequestModule } from '@ghostfolio/api/interceptors/transform-data-source-in-request/transform-data-source-in-request.module'; import { TransformDataSourceInRequestModule } from '@ghostfolio/api/interceptors/transform-data-source-in-request/transform-data-source-in-request.module';
import { TransformDataSourceInResponseModule } from '@ghostfolio/api/interceptors/transform-data-source-in-response/transform-data-source-in-response.module'; import { TransformDataSourceInResponseModule } from '@ghostfolio/api/interceptors/transform-data-source-in-response/transform-data-source-in-response.module';
import { BenchmarkModule } from '@ghostfolio/api/services/benchmark/benchmark.module';
import { DataProviderModule } from '@ghostfolio/api/services/data-provider/data-provider.module'; import { DataProviderModule } from '@ghostfolio/api/services/data-provider/data-provider.module';
import { ImpersonationModule } from '@ghostfolio/api/services/impersonation/impersonation.module'; import { ImpersonationModule } from '@ghostfolio/api/services/impersonation/impersonation.module';
import { MarketDataModule } from '@ghostfolio/api/services/market-data/market-data.module';
import { PrismaModule } from '@ghostfolio/api/services/prisma/prisma.module'; import { PrismaModule } from '@ghostfolio/api/services/prisma/prisma.module';
import { DataGatheringQueueModule } from '@ghostfolio/api/services/queues/data-gathering/data-gathering.module'; import { DataGatheringQueueModule } from '@ghostfolio/api/services/queues/data-gathering/data-gathering.module';
import { WatchlistComputationQueueModule } from '@ghostfolio/api/services/queues/watchlist/watchlist.module';
import { SymbolProfileModule } from '@ghostfolio/api/services/symbol-profile/symbol-profile.module'; import { SymbolProfileModule } from '@ghostfolio/api/services/symbol-profile/symbol-profile.module';
import { Module } from '@nestjs/common'; import { Module } from '@nestjs/common';
@ -16,15 +16,15 @@ import { WatchlistService } from './watchlist.service';
@Module({ @Module({
controllers: [WatchlistController], controllers: [WatchlistController],
imports: [ imports: [
BenchmarkModule,
DataGatheringQueueModule, DataGatheringQueueModule,
DataProviderModule, DataProviderModule,
ImpersonationModule, ImpersonationModule,
MarketDataModule,
PrismaModule, PrismaModule,
RedisCacheModule,
SymbolProfileModule, SymbolProfileModule,
TransformDataSourceInRequestModule, TransformDataSourceInRequestModule,
TransformDataSourceInResponseModule TransformDataSourceInResponseModule,
WatchlistComputationQueueModule
], ],
providers: [WatchlistService] providers: [WatchlistService]
}) })

135
apps/api/src/app/endpoints/watchlist/watchlist.service.ts

@ -1,23 +1,34 @@
import { BenchmarkService } from '@ghostfolio/api/services/benchmark/benchmark.service'; import { RedisCacheService } from '@ghostfolio/api/app/redis-cache/redis-cache.service';
import { DataProviderService } from '@ghostfolio/api/services/data-provider/data-provider.service'; import { DataProviderService } from '@ghostfolio/api/services/data-provider/data-provider.service';
import { MarketDataService } from '@ghostfolio/api/services/market-data/market-data.service';
import { PrismaService } from '@ghostfolio/api/services/prisma/prisma.service'; import { PrismaService } from '@ghostfolio/api/services/prisma/prisma.service';
import { DataGatheringService } from '@ghostfolio/api/services/queues/data-gathering/data-gathering.service'; import { DataGatheringService } from '@ghostfolio/api/services/queues/data-gathering/data-gathering.service';
import { WatchlistComputationService } from '@ghostfolio/api/services/queues/watchlist/watchlist-computation.service';
import { SymbolProfileService } from '@ghostfolio/api/services/symbol-profile/symbol-profile.service'; import { SymbolProfileService } from '@ghostfolio/api/services/symbol-profile/symbol-profile.service';
import {
WATCHLIST_COMPUTATION_QUEUE_PRIORITY_HIGH,
WATCHLIST_COMPUTATION_QUEUE_PRIORITY_LOW,
WATCHLIST_PROCESS_JOB_NAME,
WATCHLIST_PROCESS_JOB_OPTIONS
} from '@ghostfolio/common/config';
import { WatchlistResponse } from '@ghostfolio/common/interfaces'; import { WatchlistResponse } from '@ghostfolio/common/interfaces';
import { BadRequestException, Injectable } from '@nestjs/common'; import { BadRequestException, Injectable, Logger } from '@nestjs/common';
import { DataSource, Prisma } from '@prisma/client'; import { DataSource, Prisma } from '@prisma/client';
import { isAfter } from 'date-fns';
import { WatchlistValue } from './interfaces/watchlist-value.interface';
@Injectable() @Injectable()
export class WatchlistService { export class WatchlistService {
private readonly logger = new Logger(WatchlistService.name);
public constructor( public constructor(
private readonly benchmarkService: BenchmarkService,
private readonly dataGatheringService: DataGatheringService, private readonly dataGatheringService: DataGatheringService,
private readonly dataProviderService: DataProviderService, private readonly dataProviderService: DataProviderService,
private readonly marketDataService: MarketDataService,
private readonly prismaService: PrismaService, private readonly prismaService: PrismaService,
private readonly symbolProfileService: SymbolProfileService private readonly redisCacheService: RedisCacheService,
private readonly symbolProfileService: SymbolProfileService,
private readonly watchlistComputationService: WatchlistComputationService
) {} ) {}
public async createWatchlistItem({ public async createWatchlistItem({
@ -66,6 +77,10 @@ export class WatchlistService {
}, },
where: { id: userId } where: { id: userId }
}); });
await this.redisCacheService.remove(
this.redisCacheService.getWatchlistKey({ userId })
);
} }
public async deleteWatchlistItem({ public async deleteWatchlistItem({
@ -87,69 +102,69 @@ export class WatchlistService {
}, },
where: { id: userId } where: { id: userId }
}); });
await this.redisCacheService.remove(
this.redisCacheService.getWatchlistKey({ userId })
);
} }
public async getWatchlistItems( public async getWatchlistItems(
userId: string userId: string
): Promise<WatchlistResponse['watchlist']> { ): Promise<WatchlistResponse['watchlist']> {
const user = await this.prismaService.user.findUnique({ let cachedWatchlist: WatchlistResponse['watchlist'];
select: { let isCachedWatchlistExpired = false;
watchlist: {
select: { dataSource: true, symbol: true } try {
} const cachedWatchlistValue = await this.redisCacheService.get(
}, this.redisCacheService.getWatchlistKey({ userId })
where: { id: userId } );
});
const [assetProfiles, quotes] = await Promise.all([ const { expiration, watchlist }: WatchlistValue =
this.symbolProfileService.getSymbolProfiles(user.watchlist), JSON.parse(cachedWatchlistValue);
this.dataProviderService.getQuotes({
items: user.watchlist.map(({ dataSource, symbol }) => { cachedWatchlist = watchlist;
return { dataSource, symbol };
}) if (isAfter(new Date(), new Date(expiration))) {
}) isCachedWatchlistExpired = true;
]); }
} catch {}
const watchlist = await Promise.all(
user.watchlist.map(async ({ dataSource, symbol }) => { if (cachedWatchlist) {
const assetProfile = assetProfiles.find((profile) => { this.logger.debug(`Fetched watchlist of user '${userId}' from cache`);
return profile.dataSource === dataSource && profile.symbol === symbol;
if (isCachedWatchlistExpired) {
// Compute in the background
this.watchlistComputationService.addJobToQueue({
data: { userId },
name: WATCHLIST_PROCESS_JOB_NAME,
opts: {
...WATCHLIST_PROCESS_JOB_OPTIONS,
jobId: userId,
priority: WATCHLIST_COMPUTATION_QUEUE_PRIORITY_LOW
}
}); });
}
const [allTimeHigh, trends] = await Promise.all([ return cachedWatchlist;
this.marketDataService.getMax({ }
dataSource,
symbol
}),
this.benchmarkService.getBenchmarkTrends({ dataSource, symbol })
]);
const performancePercent =
this.benchmarkService.calculateChangeInPercentage(
allTimeHigh?.marketPrice,
quotes[symbol]?.marketPrice
);
return {
dataSource,
symbol,
marketCondition:
this.benchmarkService.getMarketCondition(performancePercent),
name: assetProfile?.name,
performances: {
allTimeHigh: {
performancePercent,
date: allTimeHigh?.date
}
},
trend50d: trends.trend50d,
trend200d: trends.trend200d
};
})
);
return watchlist.sort((a, b) => { // Wait for computation
return a.name.localeCompare(b.name); await this.watchlistComputationService.addJobToQueue({
data: { userId },
name: WATCHLIST_PROCESS_JOB_NAME,
opts: {
...WATCHLIST_PROCESS_JOB_OPTIONS,
jobId: userId,
priority: WATCHLIST_COMPUTATION_QUEUE_PRIORITY_HIGH
}
}); });
const job = await this.watchlistComputationService.getJob(userId);
if (job) {
await job.finished();
}
return this.getWatchlistItems(userId);
} }
} }

4
apps/api/src/app/redis-cache/redis-cache.service.ts

@ -76,6 +76,10 @@ export class RedisCacheService {
return `quote-${getAssetProfileIdentifier({ dataSource, symbol })}`; return `quote-${getAssetProfileIdentifier({ dataSource, symbol })}`;
} }
public getWatchlistKey({ userId }: { userId: string }) {
return `watchlist-${userId}`;
}
public async isHealthy() { public async isHealthy() {
const HEALTH_CHECK_TIMEOUT = ms('5 seconds'); const HEALTH_CHECK_TIMEOUT = ms('5 seconds');

3
apps/api/src/services/queues/watchlist/interfaces/watchlist-queue-job.interface.ts

@ -0,0 +1,3 @@
export interface WatchlistQueueJob {
userId: string;
}

31
apps/api/src/services/queues/watchlist/watchlist-computation.service.ts

@ -0,0 +1,31 @@
import { WATCHLIST_COMPUTATION_QUEUE } from '@ghostfolio/common/config';
import { InjectQueue } from '@nestjs/bull';
import { Injectable } from '@nestjs/common';
import { JobOptions, Queue } from 'bull';
import { WatchlistQueueJob } from './interfaces/watchlist-queue-job.interface';
@Injectable()
export class WatchlistComputationService {
public constructor(
@InjectQueue(WATCHLIST_COMPUTATION_QUEUE)
private readonly watchlistComputationQueue: Queue
) {}
public async addJobToQueue({
data,
name,
opts
}: {
data: WatchlistQueueJob;
name: string;
opts?: JobOptions;
}) {
return this.watchlistComputationQueue.add(name, data, opts);
}
public async getJob(jobId: string) {
return this.watchlistComputationQueue.getJob(jobId);
}
}

52
apps/api/src/services/queues/watchlist/watchlist.module.ts

@ -0,0 +1,52 @@
import { RedisCacheModule } from '@ghostfolio/api/app/redis-cache/redis-cache.module';
import { BenchmarkModule } from '@ghostfolio/api/services/benchmark/benchmark.module';
import { ConfigurationModule } from '@ghostfolio/api/services/configuration/configuration.module';
import { DataProviderModule } from '@ghostfolio/api/services/data-provider/data-provider.module';
import { MarketDataModule } from '@ghostfolio/api/services/market-data/market-data.module';
import { PrismaModule } from '@ghostfolio/api/services/prisma/prisma.module';
import { WatchlistComputationService } from '@ghostfolio/api/services/queues/watchlist/watchlist-computation.service';
import { SymbolProfileModule } from '@ghostfolio/api/services/symbol-profile/symbol-profile.module';
import {
DEFAULT_PROCESSOR_WATCHLIST_COMPUTATION_TIMEOUT,
WATCHLIST_COMPUTATION_QUEUE
} from '@ghostfolio/common/config';
import { BullAdapter } from '@bull-board/api/bullAdapter';
import { BullBoardModule } from '@bull-board/nestjs';
import { BullModule } from '@nestjs/bull';
import { Module } from '@nestjs/common';
import { WatchlistProcessor } from './watchlist.processor';
@Module({
exports: [BullModule, WatchlistComputationService],
imports: [
BenchmarkModule,
BullBoardModule.forFeature({
adapter: BullAdapter,
name: WATCHLIST_COMPUTATION_QUEUE,
options: {
displayName: 'Watchlist Computation',
readOnlyMode: process.env.BULL_BOARD_IS_READ_ONLY !== 'false'
}
}),
BullModule.registerQueue({
name: WATCHLIST_COMPUTATION_QUEUE,
settings: {
lockDuration: parseInt(
process.env.PROCESSOR_WATCHLIST_COMPUTATION_TIMEOUT ??
DEFAULT_PROCESSOR_WATCHLIST_COMPUTATION_TIMEOUT.toString(),
10
)
}
}),
ConfigurationModule,
DataProviderModule,
MarketDataModule,
PrismaModule,
RedisCacheModule,
SymbolProfileModule
],
providers: [WatchlistComputationService, WatchlistProcessor]
})
export class WatchlistComputationQueueModule {}

152
apps/api/src/services/queues/watchlist/watchlist.processor.ts

@ -0,0 +1,152 @@
import { WatchlistValue } from '@ghostfolio/api/app/endpoints/watchlist/interfaces/watchlist-value.interface';
import { RedisCacheService } from '@ghostfolio/api/app/redis-cache/redis-cache.service';
import { BenchmarkService } from '@ghostfolio/api/services/benchmark/benchmark.service';
import { ConfigurationService } from '@ghostfolio/api/services/configuration/configuration.service';
import { DataProviderService } from '@ghostfolio/api/services/data-provider/data-provider.service';
import { MarketDataService } from '@ghostfolio/api/services/market-data/market-data.service';
import { PrismaService } from '@ghostfolio/api/services/prisma/prisma.service';
import { SymbolProfileService } from '@ghostfolio/api/services/symbol-profile/symbol-profile.service';
import {
CACHE_TTL_INFINITE,
DEFAULT_PROCESSOR_WATCHLIST_COMPUTATION_CONCURRENCY,
WATCHLIST_COMPUTATION_QUEUE,
WATCHLIST_PROCESS_JOB_NAME
} from '@ghostfolio/common/config';
import { WatchlistResponse } from '@ghostfolio/common/interfaces';
import { Process, Processor } from '@nestjs/bull';
import { Injectable, Logger } from '@nestjs/common';
import { Job } from 'bull';
import { addMilliseconds } from 'date-fns';
import { WatchlistQueueJob } from './interfaces/watchlist-queue-job.interface';
@Injectable()
@Processor(WATCHLIST_COMPUTATION_QUEUE)
export class WatchlistProcessor {
private readonly logger = new Logger(WatchlistProcessor.name);
public constructor(
private readonly benchmarkService: BenchmarkService,
private readonly configurationService: ConfigurationService,
private readonly dataProviderService: DataProviderService,
private readonly marketDataService: MarketDataService,
private readonly prismaService: PrismaService,
private readonly redisCacheService: RedisCacheService,
private readonly symbolProfileService: SymbolProfileService
) {}
@Process({
concurrency: parseInt(
process.env.PROCESSOR_WATCHLIST_COMPUTATION_CONCURRENCY ??
DEFAULT_PROCESSOR_WATCHLIST_COMPUTATION_CONCURRENCY.toString(),
10
),
name: WATCHLIST_PROCESS_JOB_NAME
})
public async calculateWatchlist(job: Job<WatchlistQueueJob>) {
try {
const startTime = performance.now();
const { userId } = job.data;
this.logger.log(
`Watchlist calculation of user '${userId}' has been started`
);
const user = await this.prismaService.user.findUnique({
select: {
watchlist: {
select: { dataSource: true, symbol: true }
}
},
where: { id: userId }
});
const [assetProfiles, quotes] = await Promise.all([
this.symbolProfileService.getSymbolProfiles(user.watchlist),
this.dataProviderService.getQuotes({
items: user.watchlist.map(({ dataSource, symbol }) => {
return { dataSource, symbol };
})
})
]);
let isComplete = user.watchlist.length > 0;
const watchlist: WatchlistResponse['watchlist'] = await Promise.all(
user.watchlist.map(async ({ dataSource, symbol }) => {
const assetProfile = assetProfiles.find((profile) => {
return (
profile.dataSource === dataSource && profile.symbol === symbol
);
});
const [allTimeHigh, trends] = await Promise.all([
this.marketDataService.getMax({
dataSource,
symbol
}),
this.benchmarkService.getBenchmarkTrends({ dataSource, symbol })
]);
if (!allTimeHigh?.marketPrice || !quotes[symbol]?.marketPrice) {
isComplete = false;
}
const performancePercent =
this.benchmarkService.calculateChangeInPercentage(
allTimeHigh?.marketPrice,
quotes[symbol]?.marketPrice
);
return {
dataSource,
symbol,
marketCondition:
this.benchmarkService.getMarketCondition(performancePercent),
name: assetProfile?.name,
performances: {
allTimeHigh: {
performancePercent,
date: allTimeHigh?.date
}
},
trend50d: trends.trend50d,
trend200d: trends.trend200d
};
})
);
const sortedWatchlist = watchlist.sort((a, b) => {
return a.name.localeCompare(b.name);
});
this.logger.log(
`Watchlist calculation of user '${userId}' has been completed in ${(
(performance.now() - startTime) /
1000
).toFixed(3)} seconds`
);
const expiration = addMilliseconds(
new Date(),
isComplete ? this.configurationService.get('CACHE_QUOTES_TTL') : 0
);
await this.redisCacheService.set(
this.redisCacheService.getWatchlistKey({ userId }),
JSON.stringify({
expiration: expiration.getTime(),
watchlist: sortedWatchlist
} as WatchlistValue),
CACHE_TTL_INFINITE
);
return sortedWatchlist;
} catch (error) {
this.logger.error(error);
throw new Error(error);
}
}
}

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

@ -79,6 +79,10 @@ export const PORTFOLIO_SNAPSHOT_COMPUTATION_QUEUE_PRIORITY_LOW =
export const STATISTICS_GATHERING_QUEUE = 'STATISTICS_GATHERING_QUEUE'; export const STATISTICS_GATHERING_QUEUE = 'STATISTICS_GATHERING_QUEUE';
export const WATCHLIST_COMPUTATION_QUEUE = 'WATCHLIST_COMPUTATION_QUEUE';
export const WATCHLIST_COMPUTATION_QUEUE_PRIORITY_HIGH = 1;
export const WATCHLIST_COMPUTATION_QUEUE_PRIORITY_LOW = Number.MAX_SAFE_INTEGER;
export const DEFAULT_COLOR_SCHEME: ColorScheme = 'LIGHT'; export const DEFAULT_COLOR_SCHEME: ColorScheme = 'LIGHT';
export const DEFAULT_CURRENCY = 'USD'; export const DEFAULT_CURRENCY = 'USD';
export const DEFAULT_DATE_FORMAT_MONTH_YEAR = 'MMM yyyy'; export const DEFAULT_DATE_FORMAT_MONTH_YEAR = 'MMM yyyy';
@ -91,6 +95,8 @@ export const DEFAULT_PROCESSOR_GATHER_ASSET_PROFILE_CONCURRENCY = 1;
export const DEFAULT_PROCESSOR_GATHER_HISTORICAL_MARKET_DATA_CONCURRENCY = 1; export const DEFAULT_PROCESSOR_GATHER_HISTORICAL_MARKET_DATA_CONCURRENCY = 1;
export const DEFAULT_PROCESSOR_PORTFOLIO_SNAPSHOT_COMPUTATION_CONCURRENCY = 1; export const DEFAULT_PROCESSOR_PORTFOLIO_SNAPSHOT_COMPUTATION_CONCURRENCY = 1;
export const DEFAULT_PROCESSOR_PORTFOLIO_SNAPSHOT_COMPUTATION_TIMEOUT = 30000; export const DEFAULT_PROCESSOR_PORTFOLIO_SNAPSHOT_COMPUTATION_TIMEOUT = 30000;
export const DEFAULT_PROCESSOR_WATCHLIST_COMPUTATION_CONCURRENCY = 1;
export const DEFAULT_PROCESSOR_WATCHLIST_COMPUTATION_TIMEOUT = 30000;
export const DEFAULT_REDACTED_PATHS = [ export const DEFAULT_REDACTED_PATHS = [
'accounts[*].balance', 'accounts[*].balance',
@ -221,6 +227,11 @@ export const PORTFOLIO_SNAPSHOT_PROCESS_JOB_OPTIONS: JobOptions = {
removeOnComplete: true removeOnComplete: true
}; };
export const WATCHLIST_PROCESS_JOB_NAME = 'WATCHLIST';
export const WATCHLIST_PROCESS_JOB_OPTIONS: JobOptions = {
removeOnComplete: true
};
export const HEADER_KEY_IMPERSONATION = 'Impersonation-Id'; export const HEADER_KEY_IMPERSONATION = 'Impersonation-Id';
export const HEADER_KEY_TIMEZONE = 'Timezone'; export const HEADER_KEY_TIMEZONE = 'Timezone';
export const HEADER_KEY_TOKEN = 'Authorization'; export const HEADER_KEY_TOKEN = 'Authorization';

Loading…
Cancel
Save