feat: aggregate portfolio data across accounts

Add authenticated all-account portfolio, operations, income and event endpoints. Preserve source accounts, exact currency totals and scoped cursor pagination.
This commit is contained in:
2026-09-09 20:36:31 +03:00
parent 33f703d9a3
commit eaf2f67b34
15 changed files with 909 additions and 83 deletions
+10
View File
@@ -0,0 +1,10 @@
import { ApiProperty } from '@nestjs/swagger';
/** Source account of a row in an all-accounts response. */
export class AccountReferenceDto {
@ApiProperty()
id: string;
@ApiProperty()
name: string;
}
+342
View File
@@ -0,0 +1,342 @@
import {
BadRequestException,
ConflictException,
ServiceUnavailableException,
} from '@nestjs/common';
jest.mock('./tbank-access.service', () => ({ TbankAccessService: class {} }));
import { BrokerageOperationsService } from './brokerage-operations.service';
import { IncomeAnalyticsService } from './income-analytics.service';
import { PortfolioService } from './portfolio.service';
import { UpcomingEventsService } from './upcoming-events.service';
import type { TbankConnectionContext } from './tbank-access.service';
function setup() {
const context = {
token: 'token',
generation: 'first',
accounts: [
{ id: 'a', name: 'Основной', openedOn: '2020-01-01' },
{ id: 'b', name: 'ИИС', openedOn: '2026-01-01' },
],
};
const access = {
withSupportedAccounts: jest.fn(
(_user: string, handler: (value: typeof context) => unknown) =>
handler(context),
),
withAccountForConnection: jest.fn(
(
_connection: TbankConnectionContext,
_id: string,
handler: () => unknown,
) => handler(),
),
};
const tbank = {
getOperationsByCursor: jest.fn(),
getPortfolio: jest.fn().mockResolvedValue({ positions: [] }),
getPositions: jest.fn().mockResolvedValue({ money: [] }),
getDividends: jest.fn().mockResolvedValue([]),
getInstrumentByFigi: jest.fn().mockResolvedValue({ name: 'Облигация' }),
};
const metadata = { resolve: jest.fn().mockResolvedValue(null) };
const requestContext = { set: jest.fn() };
const operations = new BrokerageOperationsService(
access as never,
tbank as never,
requestContext as never,
metadata as never,
);
const portfolio = new PortfolioService(
access as never,
tbank as never,
requestContext as never,
metadata as never,
);
const moex = { bondEvents: jest.fn().mockResolvedValue([]) };
const events = new UpcomingEventsService(
access as never,
tbank as never,
moex as never,
requestContext as never,
{ info: jest.fn(), event: jest.fn() } as never,
);
return {
context,
access,
tbank,
operations,
portfolio,
events,
moex,
income: new IncomeAnalyticsService(operations),
};
}
const operation = (
id: string,
second: number,
type = 'OPERATION_TYPE_COUPON',
value = 1,
) => ({
id,
date: { seconds: String(second) },
type,
payment: { currency: 'RUB', units: value },
});
const query = { from: '2025-01-01', to: '2026-12-31', limit: 2 };
describe('All accounts', () => {
afterEach(() => jest.useRealTimers());
it('merges exact amounts by currency and keeps equal instruments on different accounts', async () => {
const { portfolio, tbank } = setup();
tbank.getPortfolio.mockResolvedValue({
positions: [
{
figi: 'SAME',
instrumentType: 'share',
quantity: { units: 1 },
currentPrice: { currency: 'RUB', nano: 100000000 },
},
],
});
tbank.getPositions.mockImplementation((_token: string, id: string) => ({
money: [{ currency: id === 'a' ? 'RUB' : 'USD', nano: 200000000 }],
}));
const result = await portfolio.getAll('user');
expect(result.totalsByCurrency).toEqual([
{ currency: 'RUB', value: '0.4' },
{ currency: 'USD', value: '0.2' },
]);
expect(result.cashBalances).toEqual([
{ currency: 'RUB', value: '0.2' },
{ currency: 'USD', value: '0.2' },
]);
expect(result.positions.map((position) => position.account?.id)).toEqual([
'a',
'b',
]);
});
it('rejects missing access and a partial portfolio instead of publishing incomplete totals', async () => {
const { portfolio, access, tbank } = setup();
access.withSupportedAccounts.mockImplementationOnce(() => {
throw new ConflictException({ code: 'TBANK_NOT_CONNECTED' });
});
await expect(portfolio.getAll('user')).rejects.toBeInstanceOf(
ConflictException,
);
expect(tbank.getPortfolio).not.toHaveBeenCalled();
tbank.getPortfolio.mockRejectedValueOnce(new ServiceUnavailableException());
await expect(portfolio.getAll('user')).rejects.toBeInstanceOf(
ServiceUnavailableException,
);
});
it('marks an aggregate stale and uses its oldest snapshot timestamp', async () => {
jest.useFakeTimers().setSystemTime(new Date('2026-09-01T10:00:00Z'));
const { portfolio, tbank } = setup();
const first = await portfolio.getAll('user');
jest.setSystemTime(new Date('2026-09-01T10:01:00Z'));
tbank.getPortfolio.mockRejectedValueOnce(new ServiceUnavailableException());
const result = await portfolio.getAll('user');
expect(result.isStale).toBe(true);
expect(result.lastSuccessfulAt).toBe(first.lastSuccessfulAt);
});
it('paginates interleaved streams without dropping buffered rows, preserving duplicate IDs', async () => {
const { operations, tbank } = setup();
tbank.getOperationsByCursor.mockImplementation(
(
_token: string,
{ accountId, cursor }: { accountId: string; cursor?: string },
) => {
if (accountId === 'a')
return cursor
? { items: [operation('last', 3)] }
: {
items: [operation('same', 9), operation('a2', 5)],
hasNext: true,
nextCursor: 'a-next',
};
return { items: [operation('same', 8), operation('b2', 4)] };
},
);
const first = await operations.getAll('user', query);
const second = await operations.getAll('user', {
...query,
cursor: first.nextCursor!,
});
const repeated = await operations.getAll('user', {
...query,
cursor: first.nextCursor!,
});
const third = await operations.getAll('user', {
...query,
cursor: second.nextCursor!,
});
expect(first.items.map((item) => [item.id, item.account?.id])).toEqual([
['same', 'a'],
['same', 'b'],
]);
expect(second.items.map((item) => item.id)).toEqual(['a2', 'b2']);
expect(repeated).toEqual(second);
expect(third.items.map((item) => item.id)).toEqual(['last']);
expect(third.hasNext).toBe(false);
});
it('continues through empty filtered pages and breaks ties by account', async () => {
const { operations, tbank } = setup();
tbank.getOperationsByCursor.mockImplementation(
(_token: string, { cursor }: { cursor?: string }) =>
cursor
? { items: [operation('income', 5)] }
: {
items: [operation('fee', 8, 'OPERATION_TYPE_BROKER_FEE')],
hasNext: true,
nextCursor: 'next',
},
);
const page = await operations.getAll('user', {
...query,
categories: ['income'],
});
expect(page.items.map((item) => item.account?.id)).toEqual(['a', 'b']);
expect(page.items.every((item) => item.category === 'income')).toBe(true);
});
it('rejects damaged cursor chains and partial streams', async () => {
const { operations, tbank } = setup();
tbank.getOperationsByCursor.mockResolvedValue({
items: [],
hasNext: true,
nextCursor: 'loop',
});
await expect(operations.getAll('user', query)).rejects.toBeInstanceOf(
ServiceUnavailableException,
);
tbank.getOperationsByCursor.mockRejectedValue(
new ServiceUnavailableException(),
);
await expect(
operations.getAll('another-user', query),
).rejects.toBeInstanceOf(ServiceUnavailableException);
});
it('binds continuations to user, connection, accounts and filters', async () => {
const { operations, context, tbank } = setup();
tbank.getOperationsByCursor.mockResolvedValue({
items: [operation('first', 8), operation('last', 4)],
});
const first = await operations.getAll('user', query);
const next = { ...query, cursor: first.nextCursor! };
await expect(operations.getAll('other', next)).rejects.toBeInstanceOf(
BadRequestException,
);
await expect(
operations.getAll('user', { ...next, categories: ['fee'] }),
).rejects.toBeInstanceOf(BadRequestException);
context.generation = 'replaced';
await expect(operations.getAll('user', next)).rejects.toBeInstanceOf(
BadRequestException,
);
context.generation = 'first';
context.accounts.pop();
await expect(operations.getAll('user', next)).rejects.toBeInstanceOf(
BadRequestException,
);
expect(tbank.getOperationsByCursor).toHaveBeenCalledTimes(2);
});
it('collects income across the same cohort and computes comparisons from all receipts', async () => {
const { income, tbank } = setup();
tbank.getOperationsByCursor.mockImplementation(
(
_token: string,
{ accountId, from }: { accountId: string; from: { seconds: string } },
) => {
const year = new Date(Number(from.seconds) * 1000).getUTCFullYear();
const second =
Date.parse(
year === 2025 ? '2025-08-15T12:00:00Z' : '2026-08-15T12:00:00Z',
) / 1000;
return {
items:
year === 2025 && accountId === 'b'
? []
: [
operation(
'same',
second,
'OPERATION_TYPE_COUPON',
year === 2025 ? 2 : 3,
),
],
};
},
);
const result = await income.analytics(
'user',
'all',
'2026-08-01',
'2026-08-31',
);
expect(result.totalsByCurrency[0].total.value).toBe('6');
expect(
result.yearComparison?.totalsByCurrency[0].compared.total.value,
).toBe('2');
expect(
result.yearComparison?.totalsByCurrency[0].difference.total.value,
).toBe('4');
expect(result.entries.map((entry) => entry.account?.id)).toEqual([
'a',
'b',
]);
expect(
new Set(result.entries.map((entry) => entry.instrumentKey)).size,
).toBe(2);
});
it('preserves separate future events and fails the section when one source fails', async () => {
const { events, tbank, moex } = setup();
const today = new Date().toISOString().slice(0, 10);
tbank.getPortfolio.mockResolvedValue({
positions: [
{
figi: 'BOND',
ticker: 'BOND',
instrumentType: 'bond',
quantity: { units: 2 },
},
],
});
moex.bondEvents.mockResolvedValue([
{ type: 'coupon', date: today, value: '0.1', currency: 'RUB' },
]);
const result = await events.getAll('user');
expect(result.events.map((event) => event.account?.id)).toEqual(['a', 'b']);
expect(result.incomeTotalsByCurrency).toEqual([
{ currency: 'RUB', value: '0.4' },
]);
tbank.getPortfolio
.mockResolvedValueOnce({ positions: [] })
.mockRejectedValueOnce(new ServiceUnavailableException());
await expect(events.getAll('user')).rejects.toBeInstanceOf(
ServiceUnavailableException,
);
});
it('returns valid empty results for an empty account set', async () => {
const { context, portfolio, operations, income, events, tbank } = setup();
context.accounts = [];
expect((await portfolio.getAll('user')).positions).toEqual([]);
expect((await operations.getAll('user', query)).hasNext).toBe(false);
expect((await income.summary('user', 'all')).totalsByCurrency).toEqual([]);
expect((await events.getAll('user')).events).toEqual([]);
expect(tbank.getOperationsByCursor).not.toHaveBeenCalled();
expect(tbank.getPortfolio).not.toHaveBeenCalled();
});
});
+4
View File
@@ -12,6 +12,7 @@ import {
Min,
} from 'class-validator';
import { MoneyDto } from './portfolio.dto';
import { AccountReferenceDto } from './account-reference.dto';
export const operationCategories = [
'trade',
@@ -59,6 +60,9 @@ export class BrokerageOperationsQueryDto {
}
export class BrokerageOperationDto {
@ApiPropertyOptional({ type: AccountReferenceDto })
account?: AccountReferenceDto;
@ApiProperty()
id: string;
@@ -18,6 +18,15 @@ import { BrokerageOperationsService } from './brokerage-operations.service';
export class BrokerageOperationsController {
constructor(private readonly operations: BrokerageOperationsService) {}
@Get('all/operations')
@ApiOkResponse({ type: BrokerageOperationsPageDto })
all(
@Req() request: Request & { user: AccessPayload },
@Query() query: BrokerageOperationsQueryDto,
) {
return this.operations.getAll(request.user.sub, query);
}
@Get(':accountId/operations')
@ApiOkResponse({ type: BrokerageOperationsPageDto })
get(
+229 -1
View File
@@ -1,4 +1,11 @@
import { Injectable, ServiceUnavailableException } from '@nestjs/common';
import {
BadRequestException,
Injectable,
ServiceUnavailableException,
} from '@nestjs/common';
import { randomUUID } from 'node:crypto';
import { AccountReferenceDto } from './account-reference.dto';
import { BrokerageOperationDto } from './brokerage-operation.dto';
import { LRUCache } from 'lru-cache';
import { RequestContextService } from '../common/logging/request-context.service';
import {
@@ -14,6 +21,21 @@ import { TbankAccessService } from './tbank-access.service';
import { TbankClient } from './tbank.client';
import { DateOnlyRange, dateOnlyBounds } from './income-period';
type OperationStream = {
account: AccountReferenceDto;
pending: BrokerageOperationDto[];
cursor?: string;
finished: boolean;
seen: string[];
lastOccurredAt?: number;
};
type AllOperationsContinuation = {
signature: string;
range: { from: Date; to: Date };
streams: OperationStream[];
page?: Promise<BrokerageOperationsPageDto>;
};
const freshForMs = 30_000;
const maxCachedPages = 100;
@@ -25,6 +47,15 @@ export class BrokerageOperationsService {
ttlAutopurge: true,
});
private readonly continuations = new LRUCache<
string,
AllOperationsContinuation
>({
max: 100,
ttl: 5 * 60_000,
ttlAutopurge: true,
});
constructor(
private readonly access: TbankAccessService,
private readonly tbank: TbankClient,
@@ -45,6 +76,202 @@ export class BrokerageOperationsService {
);
}
async getAll(
userId: string,
query: BrokerageOperationsQueryDto,
): Promise<BrokerageOperationsPageDto> {
return this.access.withSupportedAccounts(
userId,
async ({ token, generation, accounts }) => {
const signature = JSON.stringify({
userId,
generation,
accounts: accounts.map((account) => account.id).sort(),
from: query.from ?? null,
to: query.to ?? null,
categories: [...new Set(query.categories ?? [])].sort(),
limit: query.limit ?? 100,
});
const continuation = query.cursor
? this.continuations.get(query.cursor)
: undefined;
if (
query.cursor &&
(!continuation || continuation.signature !== signature)
)
throw new BadRequestException({ code: 'INVALID_OPERATIONS_CURSOR' });
if (continuation?.page) return continuation.page;
const range = continuation?.range ?? this.range(query);
const streams: OperationStream[] = continuation
? continuation.streams.map((stream) => ({
...stream,
pending: [...stream.pending],
seen: [...stream.seen],
}))
: accounts.map(({ id, name }) => ({
account: { id, name },
pending: [],
finished: false,
seen: [],
}));
const result = this.mergePage(
token,
generation,
userId,
query,
signature,
range,
streams,
);
if (continuation) continuation.page = result;
try {
return await result;
} catch (error) {
if (continuation) continuation.page = undefined;
throw error;
}
},
);
}
private async mergePage(
token: string,
generation: string,
userId: string,
query: BrokerageOperationsQueryDto,
signature: string,
range: { from: Date; to: Date },
streams: OperationStream[],
): Promise<BrokerageOperationsPageDto> {
const items: BrokerageOperationDto[] = [];
// Each stream retains unconsumed rows. A page boundary never discards them.
while (items.length < (query.limit ?? 100)) {
for (const stream of streams) {
while (!stream.pending.length && !stream.finished) {
const page = await this.page(
token,
generation,
userId,
stream.account.id,
{ ...query, cursor: stream.cursor },
range,
);
for (const item of page.items) {
const time = Date.parse(item.occurredAt);
if (
!Number.isFinite(time) ||
(stream.lastOccurredAt !== undefined &&
time > stream.lastOccurredAt)
)
throw new ServiceUnavailableException({
code: 'TBANK_UNAVAILABLE',
});
stream.lastOccurredAt = time;
}
if (page.hasNext) {
if (!page.nextCursor || stream.seen.includes(page.nextCursor))
throw new ServiceUnavailableException({
code: 'TBANK_UNAVAILABLE',
});
stream.seen.push(page.nextCursor);
stream.cursor = page.nextCursor;
} else stream.finished = true;
stream.pending = page.items.map((item) => ({
...item,
account: stream.account,
}));
}
}
const next = streams
.filter((stream) => stream.pending.length)
.sort(
(left, right) =>
Date.parse(right.pending[0].occurredAt) -
Date.parse(left.pending[0].occurredAt) ||
left.account.id.localeCompare(right.account.id),
)[0];
if (!next) break;
items.push(next.pending.shift()!);
}
const hasNext = streams.some(
(stream) => stream.pending.length || !stream.finished,
);
const nextCursor = hasNext ? randomUUID() : null;
if (nextCursor)
this.continuations.set(nextCursor, { signature, range, streams });
return {
accountId: 'all',
items,
hasNext,
nextCursor,
observedAt: new Date().toISOString(),
};
}
private async collectAllRanges(
userId: string,
ranges:
| DateOnlyRange[]
| ((accountOpenedAt: Date | undefined) => DateOnlyRange[]),
): Promise<{
accountOpenedAt: Date | undefined;
pages: BrokerageOperationsPageDto[][];
}> {
return this.access.withSupportedAccounts(
userId,
async ({ token, generation, accounts }) => {
// The current cohort includes newly opened accounts with no earlier receipts.
const oldest = accounts
.map((account) => account.openedOn)
.filter((date): date is string => Boolean(date))
.sort()[0];
const accountOpenedAt =
oldest && accounts.every((account) => account.openedOn)
? new Date(`${oldest}T00:00:00+03:00`)
: undefined;
const selected =
typeof ranges === 'function' ? ranges(accountOpenedAt) : ranges;
const pages: BrokerageOperationsPageDto[][] = selected.map(() => []);
for (const account of accounts) {
for (const [index, range] of selected.entries()) {
const collected = await this.collectRange(
token,
generation,
userId,
account.id,
range,
);
pages[index].push(
...collected.map((page) => ({
...page,
items: page.items.map((item) => ({
...item,
account: { id: account.id, name: account.name },
})),
})),
);
}
}
return {
accountOpenedAt,
pages: pages.map((group) => [
{
...emptyPage('all'),
items: group
.flatMap((page) => page.items)
.sort(
(left, right) =>
Date.parse(right.occurredAt) -
Date.parse(left.occurredAt) ||
left.account!.id.localeCompare(right.account!.id),
),
},
]),
};
},
);
}
async collectRanges(
userId: string,
accountId: string,
@@ -55,6 +282,7 @@ export class BrokerageOperationsService {
accountOpenedAt: Date | undefined;
pages: BrokerageOperationsPageDto[][];
}> {
if (accountId === 'all') return this.collectAllRanges(userId, ranges);
this.requestContext.set({ userId, brokerageAccountId: accountId });
return this.access.withAccount(
userId,
+15
View File
@@ -19,6 +19,21 @@ import { IncomeAnalyticsService } from './income-analytics.service';
export class IncomeAnalyticsController {
constructor(private readonly income: IncomeAnalyticsService) {}
@Get('all/income-summary')
@ApiOkResponse({ type: IncomeSummaryDto })
allSummary(@Req() request: Request & { user: AccessPayload }) {
return this.income.summary(request.user.sub, 'all');
}
@Get('all/income-analytics')
@ApiOkResponse({ type: IncomeAnalyticsDto })
allAnalytics(
@Req() request: Request & { user: AccessPayload },
@Query() query: IncomeAnalyticsQueryDto,
) {
return this.income.analytics(request.user.sub, 'all', query.from, query.to);
}
@Get(':accountId/income-summary')
@ApiOkResponse({ type: IncomeSummaryDto })
summary(
+1 -1
View File
@@ -228,7 +228,7 @@ function instrumentKey(entry: BrokerageOperationDto): string {
if (entry.figi) return `figi:${entry.figi}`;
if (entry.ticker || entry.instrumentType)
return `ticker:${entry.ticker ?? ''}:${entry.instrumentType ?? ''}`;
return `operation:${entry.id}`;
return `operation:${entry.account ? `${entry.account.id}:` : ''}${entry.id}`;
}
function difference(
+25 -2
View File
@@ -1,4 +1,11 @@
import { add, decimal, formatDecimal, isZero, multiply } from './decimal';
import {
add,
decimal,
decimalFromString,
formatDecimal,
isZero,
multiply,
} from './decimal';
import { MoneyDto } from './portfolio.dto';
import { TbankPortfolio, TbankPositions } from './tbank.client';
@@ -34,4 +41,20 @@ function totalsByCurrency(
.map(([currency, value]) => ({ currency, value: formatDecimal(value) }));
}
export { totalsByCurrency };
function sumMoneyByCurrency(amounts: MoneyDto[]): MoneyDto[] {
const totals = new Map<string, ReturnType<typeof decimalFromString>>();
for (const amount of amounts) {
totals.set(
amount.currency,
add(
totals.get(amount.currency) ?? decimal({}),
decimalFromString(amount.value),
),
);
}
return [...totals]
.sort(([left], [right]) => left.localeCompare(right))
.map(([currency, value]) => ({ currency, value: formatDecimal(value) }));
}
export { totalsByCurrency, sumMoneyByCurrency };
+6
View File
@@ -15,6 +15,12 @@ import { PortfolioService } from './portfolio.service';
export class PortfolioController {
constructor(private readonly portfolio: PortfolioService) {}
@Get('all/portfolio')
@ApiOkResponse({ type: PortfolioDto })
all(@Req() request: Request & { user: AccessPayload }) {
return this.portfolio.getAll(request.user.sub);
}
@Get(':accountId/portfolio')
@ApiOkResponse({ type: PortfolioDto })
get(
+5 -1
View File
@@ -1,4 +1,5 @@
import { ApiProperty } from '@nestjs/swagger';
import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
import { AccountReferenceDto } from './account-reference.dto';
export class MoneyDto {
@ApiProperty()
@@ -9,6 +10,9 @@ export class MoneyDto {
}
export class PortfolioPositionDto {
@ApiPropertyOptional({ type: AccountReferenceDto })
account?: AccountReferenceDto;
@ApiProperty({ type: String, nullable: true })
figi: string | null;
+80 -40
View File
@@ -8,9 +8,12 @@ import {
multiply,
subtract,
} from './decimal';
import { totalsByCurrency } from './portfolio-totals';
import { totalsByCurrency, sumMoneyByCurrency } from './portfolio-totals';
import { MoneyDto, PortfolioDto, PortfolioPositionDto } from './portfolio.dto';
import { TbankAccessService } from './tbank-access.service';
import {
TbankAccessService,
TbankConnectionContext,
} from './tbank-access.service';
import { InstrumentMetadataService } from './instrument-metadata.service';
import { TbankClient, TbankMoney } from './tbank.client';
@@ -46,49 +49,86 @@ export class PortfolioService {
async get(userId: string, accountId: string): Promise<PortfolioDto> {
this.requestContext.set({ userId, brokerageAccountId: accountId });
const key = `${userId}:${accountId}`;
return this.access.withConnection(userId, async ({ token, generation }) => {
const cached = this.snapshots.get(key);
if (
cached?.generation === generation &&
Date.now() - cached.storedAt <= freshForMs
)
return cached.snapshot;
try {
const snapshot = await this.access.withAccountForConnection(
{ token, generation },
accountId,
async () => {
const [portfolio, positions] = await Promise.all([
this.tbank.getPortfolio(token, accountId),
this.tbank.getPositions(token, accountId),
]);
return this.normalize(
token,
portfolio.positions ?? [],
positions.money ?? [],
);
},
);
this.snapshots.set(key, {
generation,
snapshot,
storedAt: Date.now(),
return this.access.withConnection(userId, (connection) =>
this.snapshot(userId, accountId, connection),
);
}
async getAll(userId: string): Promise<PortfolioDto> {
return this.access.withSupportedAccounts(userId, async (context) => {
const portfolios: PortfolioDto[] = [];
for (const account of context.accounts) {
const portfolio = await this.snapshot(userId, account.id, context);
portfolios.push({
...portfolio,
positions: portfolio.positions.map((position) => ({
...position,
account: { id: account.id, name: account.name },
})),
});
return snapshot;
} catch (error) {
if (
error instanceof ServiceUnavailableException &&
cached?.generation === generation &&
Date.now() - cached.storedAt <= fallbackForMs
) {
return { ...cached.snapshot, isStale: true };
}
throw error;
}
return {
lastSuccessfulAt:
portfolios.map((portfolio) => portfolio.lastSuccessfulAt).sort()[0] ??
new Date().toISOString(),
isStale: portfolios.some((portfolio) => portfolio.isStale),
totalsByCurrency: sumMoneyByCurrency(
portfolios.flatMap((portfolio) => portfolio.totalsByCurrency),
),
cashBalances: sumMoneyByCurrency(
portfolios.flatMap((portfolio) => portfolio.cashBalances),
),
positions: portfolios.flatMap((portfolio) => portfolio.positions),
};
});
}
private async snapshot(
userId: string,
accountId: string,
{ token, generation }: TbankConnectionContext,
): Promise<PortfolioDto> {
const key = `${userId}:${accountId}`;
const cached = this.snapshots.get(key);
if (
cached?.generation === generation &&
Date.now() - cached.storedAt <= freshForMs
)
return cached.snapshot;
try {
const snapshot = await this.access.withAccountForConnection(
{ token, generation },
accountId,
async () => {
const [portfolio, positions] = await Promise.all([
this.tbank.getPortfolio(token, accountId),
this.tbank.getPositions(token, accountId),
]);
return this.normalize(
token,
portfolio.positions ?? [],
positions.money ?? [],
);
},
);
this.snapshots.set(key, {
generation,
snapshot,
storedAt: Date.now(),
});
return snapshot;
} catch (error) {
if (
error instanceof ServiceUnavailableException &&
cached?.generation === generation &&
Date.now() - cached.storedAt <= fallbackForMs
) {
return { ...cached.snapshot, isStale: true };
}
throw error;
}
}
invalidate(userId: string): void {
for (const key of this.snapshots.keys())
if (key.startsWith(`${userId}:`)) this.snapshots.delete(key);
+6
View File
@@ -15,6 +15,12 @@ import { UpcomingEventsService } from './upcoming-events.service';
export class UpcomingEventsController {
constructor(private readonly events: UpcomingEventsService) {}
@Get('all/events')
@ApiOkResponse({ type: UpcomingEventsDto })
all(@Req() request: Request & { user: AccessPayload }) {
return this.events.getAll(request.user.sub);
}
@Get(':accountId/events')
@ApiOkResponse({ type: UpcomingEventsDto })
get(
+5 -1
View File
@@ -1,7 +1,11 @@
import { ApiProperty } from '@nestjs/swagger';
import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
import { AccountReferenceDto } from './account-reference.dto';
import { MoneyDto } from './portfolio.dto';
export class UpcomingEventDto {
@ApiPropertyOptional({ type: AccountReferenceDto })
account?: AccountReferenceDto;
@ApiProperty({ enum: ['coupon', 'dividend', 'maturity', 'offer'] })
type: 'coupon' | 'dividend' | 'maturity' | 'offer';
+71 -37
View File
@@ -48,46 +48,80 @@ export class UpcomingEventsService {
async get(userId: string, accountId: string): Promise<UpcomingEventsDto> {
this.requestContext.set({ userId, brokerageAccountId: accountId });
return this.access.withAccount(userId, accountId, async ({ token }) => {
const portfolio = await this.tbank.getPortfolio(token, accountId);
const window = dateWindow();
const supportedPositions = (portfolio.positions ?? []).filter(supported);
const events: UpcomingEventDto[][] = [];
for (
let offset = 0;
offset < supportedPositions.length;
offset += positionsBatchSize
) {
const batch = supportedPositions.slice(
offset,
offset + positionsBatchSize,
);
events.push(
...(await Promise.all(
batch.map((position) => this.positionEvents(token, position)),
)),
);
}
const projected = events
.flat()
.filter(
(event) => event.date >= window.from && event.date <= window.to,
);
this.logger.info('upcoming-events.calculated', {
count: projected.length,
});
return {
accountId,
calculatedAt: new Date().toISOString(),
events: projected.sort(
return this.access.withAccount(userId, accountId, ({ token }) =>
this.forAccount(token, accountId),
);
}
async getAll(userId: string): Promise<UpcomingEventsDto> {
return this.access.withSupportedAccounts(
userId,
async ({ token, accounts }) => {
const events: UpcomingEventDto[] = [];
for (const account of accounts) {
const result = await this.forAccount(token, account.id);
events.push(
...result.events.map((event) => ({
...event,
account: { id: account.id, name: account.name },
})),
);
}
events.sort(
(left, right) =>
left.date.localeCompare(right.date) ||
left.type.localeCompare(right.type) ||
left.ticker.localeCompare(right.ticker),
),
incomeTotalsByCurrency: incomeTotals(projected),
};
left.account!.id.localeCompare(right.account!.id),
);
return {
accountId: 'all',
calculatedAt: new Date().toISOString(),
events,
incomeTotalsByCurrency: incomeTotals(events),
};
},
);
}
private async forAccount(
token: string,
accountId: string,
): Promise<UpcomingEventsDto> {
const portfolio = await this.tbank.getPortfolio(token, accountId);
const window = dateWindow();
const supportedPositions = (portfolio.positions ?? []).filter(supported);
const events: UpcomingEventDto[][] = [];
for (
let offset = 0;
offset < supportedPositions.length;
offset += positionsBatchSize
) {
const batch = supportedPositions.slice(
offset,
offset + positionsBatchSize,
);
events.push(
...(await Promise.all(
batch.map((position) => this.positionEvents(token, position)),
)),
);
}
const projected = events
.flat()
.filter((event) => event.date >= window.from && event.date <= window.to);
this.logger.info('upcoming-events.calculated', {
count: projected.length,
});
return {
accountId,
calculatedAt: new Date().toISOString(),
events: projected.sort(
(left, right) =>
left.date.localeCompare(right.date) ||
left.type.localeCompare(right.type) ||
left.ticker.localeCompare(right.ticker),
),
incomeTotalsByCurrency: incomeTotals(projected),
};
}
private async positionEvents(