diff --git a/CHANGELOG.md b/CHANGELOG.md index 1651faf..db66732 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,30 @@ # Changelog +## [2.4.0](https://git.personal.ksv741.ru/WealthLedger/backend/compare/v2.3.1...v2.4.0) (2026-11-01) + +### Features + +* extend account list portfolio summary ([#138](https://git.personal.ksv741.ru/WealthLedger/backend/issues/138)) ([97fc41f](https://git.personal.ksv741.ru/WealthLedger/backend/commit/97fc41fa295a7a943a27171fa41cdaf286632956)) +* **portfolio:** store and publish instrument sector and issuer (WealthLedger/main[#48](https://git.personal.ksv741.ru/WealthLedger/backend/issues/48)) ([#148](https://git.personal.ksv741.ru/WealthLedger/backend/issues/148)) ([17432df](https://git.personal.ksv741.ru/WealthLedger/backend/commit/17432dfee3d7d06841caea4419a062e2a4bfc82c)) + +### Bug Fixes + +* count brokerage cash once and skip short-position return percent (WealthLedger/backend[#140](https://git.personal.ksv741.ru/WealthLedger/backend/issues/140)) ([f2843e3](https://git.personal.ksv741.ru/WealthLedger/backend/commit/f2843e30bda5442158ec145cc63a5491506bb60f)) +* reset portfolio value points inflated by double-counted cash (WealthLedger/backend[#140](https://git.personal.ksv741.ru/WealthLedger/backend/issues/140)) ([2325b63](https://git.personal.ksv741.ru/WealthLedger/backend/commit/2325b638839240bf5d6d4f7e6e1f8eaa82d4d7fd)) +* scope income and ledger sync to requested intervals (WealthLedger/backend[#57](https://git.personal.ksv741.ru/WealthLedger/backend/issues/57)) ([#147](https://git.personal.ksv741.ru/WealthLedger/backend/issues/147)) ([5e84e1e](https://git.personal.ksv741.ru/WealthLedger/backend/commit/5e84e1e5f1943b5af1992a3a946194b27fc1a1d5)) +* use available Debian packages in image build ([#129](https://git.personal.ksv741.ru/WealthLedger/backend/issues/129)) ([a250142](https://git.personal.ksv741.ru/WealthLedger/backend/commit/a250142a86e07ec7047e642afa16492fa4987f9a)) + +### Refactoring + +* expose tbank-sync public api and guard its boundary (WealthLedger/backend[#118](https://git.personal.ksv741.ru/WealthLedger/backend/issues/118)) ([#143](https://git.personal.ksv741.ru/WealthLedger/backend/issues/143)) ([5295945](https://git.personal.ksv741.ru/WealthLedger/backend/commit/5295945ab237a6e94aa64d7bb54bfd04554ca9f9)) +* publish typed problem detail codes (WealthLedger/backend[#117](https://git.personal.ksv741.ru/WealthLedger/backend/issues/117)) ([#144](https://git.personal.ksv741.ru/WealthLedger/backend/issues/144)) ([d37fa2c](https://git.personal.ksv741.ru/WealthLedger/backend/commit/d37fa2c9034e90e9d22d4022c45fc36c22dd2ab3)) +* remove tbank-sync operations import cycle (WealthLedger/backend[#116](https://git.personal.ksv741.ru/WealthLedger/backend/issues/116)) ([#142](https://git.personal.ksv741.ru/WealthLedger/backend/issues/142)) ([22668da](https://git.personal.ksv741.ru/WealthLedger/backend/commit/22668da3fa0666bdd22737913379c99d4cf01993)) + +### Tests + +* cover portfolio publish rollback and claim fence (WealthLedger/backend[#105](https://git.personal.ksv741.ru/WealthLedger/backend/issues/105)) ([#146](https://git.personal.ksv741.ru/WealthLedger/backend/issues/146)) ([e350146](https://git.personal.ksv741.ru/WealthLedger/backend/commit/e350146749293f2eaf8060f7626ff3c1cd8edf2c)) +* replace operations sync delays with explicit barriers (WealthLedger/backend[#104](https://git.personal.ksv741.ru/WealthLedger/backend/issues/104)) ([#145](https://git.personal.ksv741.ru/WealthLedger/backend/issues/145)) ([99b137e](https://git.personal.ksv741.ru/WealthLedger/backend/commit/99b137ed0b39ab73b6a28d1a494d2d98bf8fa380)) + ## [2.3.1](https://git.personal.ksv741.ru/WealthLedger/backend/compare/v2.3.0...v2.3.1) (2026-10-24) ### Bug Fixes diff --git a/eslint-suppressions.json b/eslint-suppressions.json index 4506661..af6f170 100644 --- a/eslint-suppressions.json +++ b/eslint-suppressions.json @@ -92,11 +92,8 @@ } }, "src/modules/investments/investment-sync-plan.ts": { - "sonarjs/cognitive-complexity": { - "count": 1 - }, "sonarjs/no-nested-conditional": { - "count": 4 + "count": 3 } }, "src/modules/investments/operations/application/brokerage-operations.service.ts": { diff --git a/eslint.config.mjs b/eslint.config.mjs index e453da7..d5a9e23 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -56,6 +56,7 @@ export default defineConfig( 'after', { overrides: { '?': 'before', ':': 'before' } }, ], + 'import-x/no-cycle': 'error', 'import-x/no-duplicates': ['error', { 'prefer-inline': true }], }, }, @@ -211,4 +212,24 @@ export default defineConfig( '@typescript-eslint/no-unsafe-enum-comparison': 'error', }, }, + { + files: ['src/**/*.ts'], + ignores: [...tests, 'src/modules/investments/tbank-sync/**'], + rules: { + // Separate rule so it does not replace layer-specific no-restricted-imports configs. + 'import-x/no-restricted-paths': ['error', { + zones: [{ + target: './src', + from: './src/modules/investments/tbank-sync', + except: [ + './tbank-sync.api.ts', + './tbank-sync.dto.ts', + './tbank-sync.module.ts', + './tbank-sync-repository.module.ts', + ], + message: 'Use tbank-sync.api for cross-module access.', + }], + }], + }, + }, ); diff --git a/package-lock.json b/package-lock.json index c7c0d90..8d9e113 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "wealth-ledger-backend", - "version": "2.3.1", + "version": "2.4.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "wealth-ledger-backend", - "version": "2.3.1", + "version": "2.4.0", "license": "UNLICENSED", "dependencies": { "@grpc/grpc-js": "1.14.6", diff --git a/package.json b/package.json index 56e6e99..822929f 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "wealth-ledger-backend", - "version": "2.3.1", + "version": "2.4.0", "description": "", "author": "", "private": true, diff --git a/prisma/migrations/20261028080000_reset_inflated_portfolio_value_points/migration.sql b/prisma/migrations/20261028080000_reset_inflated_portfolio_value_points/migration.sql new file mode 100644 index 0000000..74c1101 --- /dev/null +++ b/prisma/migrations/20261028080000_reset_inflated_portfolio_value_points/migration.sql @@ -0,0 +1,4 @@ +-- Points written before WealthLedger/backend#140 counted cash twice (cash balance +-- plus the mirroring CURRENCY position). Historical compositions and FX rates are +-- not stored, so the points cannot be recomputed exactly; history restarts here. +DELETE FROM "TbankPortfolioValuePoint"; diff --git a/prisma/migrations/20261031150000_add_instrument_classification/migration.sql b/prisma/migrations/20261031150000_add_instrument_classification/migration.sql new file mode 100644 index 0000000..c89280b --- /dev/null +++ b/prisma/migrations/20261031150000_add_instrument_classification/migration.sql @@ -0,0 +1,11 @@ +CREATE TABLE "TbankInstrumentClassification" ( + "provider" TEXT NOT NULL, + "instrumentKey" TEXT NOT NULL, + "sector" TEXT, + "sectorFetchedAt" DATETIME, + "assetUid" TEXT, + "brandUid" TEXT, + "brandName" TEXT, + "brandFetchedAt" DATETIME, + PRIMARY KEY ("provider","instrumentKey") +); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 734064d..4bcbf80 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -186,6 +186,20 @@ model TbankInstrumentLookup { @@id([provider, identifierType, identifier]) } +// Public per-instrument classification; no holdings or credentials. +model TbankInstrumentClassification { + provider String + instrumentKey String + sector String? + sectorFetchedAt DateTime? + assetUid String? + brandUid String? + brandName String? + brandFetchedAt DateTime? + + @@id([provider, instrumentKey]) +} + // Source records contain no holdings, personal estimates or credentials errors. model TbankInstrumentSource { provider String diff --git a/src/integrations/moex/moex.client.ts b/src/integrations/moex/moex.client.ts index 61f1f0e..df037e0 100644 --- a/src/integrations/moex/moex.client.ts +++ b/src/integrations/moex/moex.client.ts @@ -1,7 +1,7 @@ import { Injectable, ServiceUnavailableException } from '@nestjs/common'; import { constants as httpStatus } from 'node:http2'; -enum MoexErrorCode { Unavailable = 'MOEX_UNAVAILABLE' } +export enum MoexErrorCode { Unavailable = 'MOEX_UNAVAILABLE' } const MIN_PRICE_PERCENT_EXCLUSIVE = 0; const FIRST_PAGE_OFFSET = 0; diff --git a/src/integrations/ncr/ncr-ratings.client.ts b/src/integrations/ncr/ncr-ratings.client.ts index a5f110e..140545a 100644 --- a/src/integrations/ncr/ncr-ratings.client.ts +++ b/src/integrations/ncr/ncr-ratings.client.ts @@ -4,7 +4,7 @@ import { LRUCache } from 'lru-cache'; import { ISO_DATE_LENGTH, ISO_DATE_START } from '../../common/time'; -enum NcrErrorCode { Unavailable = 'NCR_UNAVAILABLE' } +export enum NcrErrorCode { Unavailable = 'NCR_UNAVAILABLE' } const EXPECTED_REGISTRY_TABLES = 1; const RATING_COLUMN = 2; diff --git a/src/integrations/tbank/tbank.client.spec.ts b/src/integrations/tbank/tbank.client.spec.ts index 4f73c6b..b6d52be 100644 --- a/src/integrations/tbank/tbank.client.spec.ts +++ b/src/integrations/tbank/tbank.client.spec.ts @@ -236,6 +236,16 @@ describe('TbankClient metadata retries', () => { jest.useRealTimers(); }); + it.each([ + ['bondBy', (client: TbankClient, optIn: boolean) => client.getBond('token', 'uid', undefined, optIn)], + ['etfBy', (client: TbankClient, optIn: boolean) => client.getEtf('token', 'uid', undefined, optIn)], + ['getAssetBy', (client: TbankClient, optIn: boolean) => optIn ? client.getAssetBrand('token', 'uid') : client.getAssetDescription('token', 'uid')], + ])('%s reports NOT_FOUND only for classification lookups', async (_method, invoke) => { + requestRpc.mockRejectedValue({ code: status.NOT_FOUND }); + await expect(invoke(client, false)).rejects.toBeInstanceOf(ServiceUnavailableException); + await expect(invoke(client, true)).rejects.toBeInstanceOf(NotFoundException); + }); + it.each([ status.UNAVAILABLE, status.DEADLINE_EXCEEDED, diff --git a/src/integrations/tbank/tbank.client.ts b/src/integrations/tbank/tbank.client.ts index e3be72a..24cd687 100644 --- a/src/integrations/tbank/tbank.client.ts +++ b/src/integrations/tbank/tbank.client.ts @@ -375,18 +375,29 @@ export class TbankClient { ).then((response) => response.instruments ?? []); } - getBond(token: string, uid: string): Promise { + getBond(token: string, uid: string, idType = InstrumentIdType.Uid, notFound = false): Promise { return this.call<{ instrument?: TbankBond }>( token, this.sparseInstruments, 'bondBy', { - idType: InstrumentIdType.Uid, + idType, id: uid, }, + notFound, ).then((response) => response.instrument ?? null); } + getEtf( + token: string, + id: string, + idType = InstrumentIdType.Uid, + notFound = false, + ): Promise> { + return this.call<{ instrument?: TbankShare }>(token, this.sparseInstruments, 'etfBy', { idType, id }, notFound) + .then((response) => response.instrument ?? null); + } + getBondCoupons( token: string, uid: string, @@ -405,18 +416,25 @@ export class TbankClient { ).then((response) => response.events ?? []); } - getShare(token: string, uid: string): Promise { + getShare(token: string, uid: string, idType = InstrumentIdType.Uid, notFound = false): Promise { return this.call<{ instrument?: TbankShare }>( token, this.instruments, 'shareBy', { - idType: InstrumentIdType.Uid, + idType, id: uid, }, + notFound, ).then((response) => response.instrument ?? null); } + getAssetBrand(token: string, assetUid: string): Promise { + return this.call<{ asset?: { brand?: { name?: string; uid?: string }; uid?: string } }>( + token, this.instruments, 'getAssetBy', { id: assetUid }, true, + ).then((response) => response.asset ? { assetUid: response.asset.uid, brand: response.asset.brand } : null); + } + getAssetDescription(token: string, assetUid: string): Promise { return this.call<{ asset?: { description?: string } }>( token, @@ -514,6 +532,7 @@ export class TbankClient { client: TbankClientService, method: string, request: object, + notFound = false, ): Promise { const metadataCall = method === 'getInstrumentBy'; for (let attempts = FIRST_ATTEMPT; attempts <= RPC_MAX_ATTEMPTS; attempts += ATTEMPT_INCREMENT) { @@ -576,7 +595,7 @@ export class TbankClient { } if ( grpcCode === status.NOT_FOUND && - (method === 'shareBy' || metadataCall) + (notFound || method === 'shareBy' || metadataCall) ) { throw new NotFoundException(); } diff --git a/src/modules/investments/brokerage-accounts/brokerage-account.dto.ts b/src/modules/investments/brokerage-accounts/brokerage-account.dto.ts index 31fe0fb..a661d1b 100644 --- a/src/modules/investments/brokerage-accounts/brokerage-account.dto.ts +++ b/src/modules/investments/brokerage-accounts/brokerage-account.dto.ts @@ -2,12 +2,39 @@ import { ApiProperty } from '@nestjs/swagger'; import { MoneyDto } from '../../../common/money.dto'; import { InvestmentResourceMetaDto } from '../investment-resource.dto'; +import { PortfolioStructureDto } from '../portfolio/portfolio.api'; import { HourlySyncStatusDto } from '../tbank-sync/tbank-sync.dto'; +export class UnrealizedReturnDto { + @ApiProperty() + declare currency: string; + + @ApiProperty({ example: '12.345678901' }) + declare absolute: string; + + @ApiProperty({ example: '100.00' }) + declare costBasis: string; + + @ApiProperty({ type: String, nullable: true, example: '12.34' }) + declare percent: null | string; + + @ApiProperty() + declare partial: boolean; +} + export class PortfolioSummaryDto { @ApiProperty({ type: MoneyDto, isArray: true }) declare totalsByCurrency: MoneyDto[]; + @ApiProperty({ type: MoneyDto, isArray: true }) + declare cashBalances: MoneyDto[]; + + @ApiProperty({ type: PortfolioStructureDto }) + declare structure: PortfolioStructureDto; + + @ApiProperty({ type: UnrealizedReturnDto, isArray: true }) + declare unrealizedReturnByCurrency: UnrealizedReturnDto[]; + @ApiProperty({ type: 'integer', minimum: 0, @@ -21,6 +48,20 @@ export class PortfolioSummaryDto { description: 'Number of non-zero bond positions, not the number of units.', }) declare bondPositionCount: number; + + @ApiProperty({ + type: 'integer', + minimum: 0, + description: 'Number of non-zero fund positions, not the number of units.', + }) + declare fundPositionCount: number; + + @ApiProperty({ + type: 'integer', + minimum: 0, + description: 'Number of all non-zero positions, not the number of units.', + }) + declare totalPositionCount: number; } export class BrokerageAccountDto { diff --git a/src/modules/investments/brokerage-accounts/brokerage-accounts.service.spec.ts b/src/modules/investments/brokerage-accounts/brokerage-accounts.service.spec.ts index bfb01ba..44c54a9 100644 --- a/src/modules/investments/brokerage-accounts/brokerage-accounts.service.spec.ts +++ b/src/modules/investments/brokerage-accounts/brokerage-accounts.service.spec.ts @@ -33,6 +33,7 @@ describe('saved brokerage account projection', () => { }); it('keeps exact saved totals and independent missing summaries', async () => { + const structure = { currencies: [], unvaluedCategories: [] }; portfolio.saved.mockResolvedValue({ membership: meta, accounts: [ @@ -43,10 +44,12 @@ describe('saved brokerage account projection', () => { openedOn: null, meta, snapshot: { + cashBalances: [{ currency: 'RUB', value: '10.000000001' }], + structure, totalsByCurrency: [ { currency: 'RUB', value: '100000000.000000001' }, ], - positions: [{ type: 'STOCK' }, { type: 'BOND' }], + positions: [{ type: 'STOCK', quantity: '1' }, { type: 'BOND', quantity: '2' }], }, }, { @@ -61,15 +64,91 @@ describe('saved brokerage account projection', () => { }); const response = await service.list('owner'); expect(response.data?.[0].portfolioSummary).toEqual({ + cashBalances: [{ currency: 'RUB', value: '10.000000001' }], + structure, totalsByCurrency: [{ currency: 'RUB', value: '100000000.000000001' }], stockPositionCount: 1, bondPositionCount: 1, + fundPositionCount: 0, + totalPositionCount: 2, + unrealizedReturnByCurrency: [], }); expect(response.data?.[1].portfolioSummary).toBeNull(); expect(response.meta.availability).toBe('available'); expect(response.meta.completeness).toBe('complete'); }); + it('returns an empty summary from an empty saved portfolio', async () => { + const structure = { currencies: [], unvaluedPositions: [], unknownValues: [] }; + portfolio.saved.mockResolvedValueOnce({ + membership: meta, + accounts: [{ + id: 'empty', name: 'Empty', type: 'BROKERAGE', openedOn: null, meta, + snapshot: { cashBalances: [], structure, totalsByCurrency: [], positions: [] }, + }], + }); + + expect((await service.list('owner')).data?.[0].portfolioSummary).toEqual({ + cashBalances: [], structure, totalsByCurrency: [], + stockPositionCount: 0, bondPositionCount: 0, + fundPositionCount: 0, totalPositionCount: 0, + unrealizedReturnByCurrency: [], + }); + }); + + it('calculates exact returns by currency, partial data, and zero cost basis', async () => { + portfolio.saved.mockResolvedValueOnce({ + membership: meta, + accounts: [{ + id: 'a', name: 'A', type: 'BROKERAGE', openedOn: null, meta, + snapshot: { + cashBalances: [], structure: {}, totalsByCurrency: [ + { currency: 'RUB', value: '100' }, { currency: 'USD', value: '20' }, + ], + positions: [ + { type: 'STOCK', quantity: '2', averagePurchasePrice: { currency: 'RUB', value: '10.000000001' }, absoluteReturn: { currency: 'RUB', value: '0.000000003' } }, + { type: 'BOND', quantity: '0.5', averagePurchasePrice: { currency: 'RUB', value: '20' }, absoluteReturn: { currency: 'RUB', value: '1.25' } }, + { type: 'FUND', quantity: '3', averagePurchasePrice: { currency: 'USD', value: '2.5' }, absoluteReturn: null }, + { type: 'STOCK', quantity: '4', averagePurchasePrice: { currency: 'USD', value: '0' }, absoluteReturn: { currency: 'USD', value: '2.5' } }, + { type: 'CURRENCY', quantity: '100', averagePurchasePrice: { currency: 'EUR', value: '1' }, absoluteReturn: { currency: 'EUR', value: '5' } }, + { type: 'BOND', quantity: '1', currentValue: { currency: 'CNY', value: '7' }, averagePurchasePrice: null, absoluteReturn: null }, + ], + }, + }], + }); + + expect((await service.list('owner')).data?.[0].portfolioSummary).toMatchObject({ + stockPositionCount: 2, bondPositionCount: 2, + fundPositionCount: 1, totalPositionCount: 6, + unrealizedReturnByCurrency: [ + { currency: 'CNY', absolute: '0', costBasis: '0', percent: null, partial: true }, + { currency: 'RUB', absolute: '1.250000003', costBasis: '30.000000002', percent: '4.17', partial: false }, + { currency: 'USD', absolute: '2.5', costBasis: '0', percent: null, partial: true }, + ], + }); + }); + + it('omits return percent for a currency with short positions', async () => { + portfolio.saved.mockResolvedValueOnce({ + membership: meta, + accounts: [{ + id: 'a', name: 'A', type: 'BROKERAGE', openedOn: null, meta, + snapshot: { + cashBalances: [], structure: {}, totalsByCurrency: [], + positions: [ + { type: 'STOCK', quantity: '-2', averagePurchasePrice: { currency: 'RUB', value: '100' }, absoluteReturn: { currency: 'RUB', value: '-20' } }, + { type: 'STOCK', quantity: '1', averagePurchasePrice: { currency: 'USD', value: '10' }, absoluteReturn: { currency: 'USD', value: '1' } }, + ], + }, + }], + }); + + expect((await service.list('owner')).data?.[0]?.portfolioSummary?.unrealizedReturnByCurrency).toEqual([ + { currency: 'RUB', absolute: '-20', costBasis: '-200', percent: null, partial: false }, + { currency: 'USD', absolute: '1', costBasis: '10', percent: '10', partial: false }, + ]); + }); + it('observes the latest hourly walk without starting another one', async () => { portfolio.saved.mockResolvedValue({ membership: meta, accounts: [] }); sync.hourly.mockResolvedValue({ diff --git a/src/modules/investments/brokerage-accounts/brokerage-accounts.service.ts b/src/modules/investments/brokerage-accounts/brokerage-accounts.service.ts index 8ad93bd..1e7e5c0 100644 --- a/src/modules/investments/brokerage-accounts/brokerage-accounts.service.ts +++ b/src/modules/investments/brokerage-accounts/brokerage-accounts.service.ts @@ -1,10 +1,82 @@ import { Injectable } from '@nestjs/common'; import { createHash } from 'node:crypto'; +import { + decimalFromString, + divide, + formatDecimal, + isZero, + multiply, +} from '../../../common/decimal'; import { investmentResourceMeta } from '../investment-resource.mapper'; -import { PortfolioService } from '../portfolio/portfolio.api'; -import { hourlyAttempt, TbankSyncRepository } from '../tbank-sync/tbank-sync.repository'; -import { BrokerageAccountDto, BrokerageAccountsResourceDto } from './brokerage-account.dto'; +import { sumMoneyByCurrency } from '../portfolio/domain/portfolio-totals'; +import { PortfolioPositionDto, PortfolioService } from '../portfolio/portfolio.api'; +import { hourlyAttempt, TbankSyncRepository } from '../tbank-sync/tbank-sync.api'; +import { + BrokerageAccountDto, + BrokerageAccountsResourceDto, + UnrealizedReturnDto, +} from './brokerage-account.dto'; + +const PERCENT_DECIMAL_PLACES = 2; +const ZERO_QUANTITY = 0n; + +function unrealizedReturns(positions: PortfolioPositionDto[]): UnrealizedReturnDto[] { + // Cash-like CURRENCY positions must not duplicate the separately returned cash balances. + const investments = positions.filter((position) => position.type !== 'CURRENCY'); + const absolute = sumMoneyByCurrency( + investments.flatMap((position) => position.absoluteReturn ? [position.absoluteReturn] : []), + ); + const costs = sumMoneyByCurrency( + investments.flatMap((position) => + position.absoluteReturn && position.averagePurchasePrice + ? [{ + currency: position.absoluteReturn.currency, + value: formatDecimal(multiply( + decimalFromString(position.averagePurchasePrice.value), + decimalFromString(position.quantity), + )), + }] + : [], + ), + ); + const partial = new Set( + investments.flatMap((position) => { + const money = position.absoluteReturn + ? null + : position.averagePurchasePrice ?? position.currentValue; + return money ? [money.currency] : []; + }), + ); + // A short position has negative cost basis, so a percent would flip sign or net out. + const withShorts = new Set( + investments.flatMap((position) => { + const money = position.absoluteReturn ?? position.averagePurchasePrice ?? position.currentValue; + return money && decimalFromString(position.quantity).value < ZERO_QUANTITY ? [money.currency] : []; + }), + ); + const currencies = new Set([...absolute, ...costs].map((money) => money.currency)); + return [...new Set([...currencies, ...partial])] + .sort((left, right) => left.localeCompare(right)) + .map((currency) => { + const absoluteReturn = absolute.find((money) => money.currency === currency)?.value ?? '0'; + const costBasis = costs.find((money) => money.currency === currency)?.value ?? '0'; + const cost = decimalFromString(costBasis); + return { + currency, + absolute: absoluteReturn, + costBasis, + percent: isZero(cost) || withShorts.has(currency) + ? null + : formatDecimal(divide( + multiply(decimalFromString(absoluteReturn), decimalFromString('100')), + cost, + PERCENT_DECIMAL_PLACES, + )), + partial: partial.has(currency), + }; + }); +} @Injectable() export class BrokerageAccountsService { @@ -26,12 +98,20 @@ export class BrokerageAccountsService { portfolioSummary: account.snapshot ? { totalsByCurrency: account.snapshot.totalsByCurrency, + cashBalances: account.snapshot.cashBalances, + structure: account.snapshot.structure, + unrealizedReturnByCurrency: unrealizedReturns(account.snapshot.positions), stockPositionCount: account.snapshot.positions.filter( (position) => position.type === 'STOCK', ).length, bondPositionCount: account.snapshot.positions.filter( (position) => position.type === 'BOND', ).length, + fundPositionCount: account.snapshot.positions.filter( + (position) => position.type === 'FUND', + ).length, + // The snapshot mapper already drops zero-quantity positions. + totalPositionCount: account.snapshot.positions.length, } : null, })); diff --git a/src/modules/investments/income/income-period.ts b/src/modules/investments/income/income-period.ts index cf4a298..43d8bcf 100644 --- a/src/modules/investments/income/income-period.ts +++ b/src/modules/investments/income/income-period.ts @@ -8,7 +8,7 @@ import { utcDate, } from '../investment-calendar'; -enum IncomeErrorCode { InvalidPeriod = 'INVALID_INCOME_PERIOD' } +export enum IncomeErrorCode { InvalidPeriod = 'INVALID_INCOME_PERIOD' } const MONDAY_WEEKDAY_INDEX = 1; const INCLUSIVE_DAY_OFFSET = 1; diff --git a/src/modules/investments/instruments/application/instrument-position.spec.ts b/src/modules/investments/instruments/application/instrument-position.spec.ts index fdf43ce..eaf75dd 100644 --- a/src/modules/investments/instruments/application/instrument-position.spec.ts +++ b/src/modules/investments/instruments/application/instrument-position.spec.ts @@ -22,6 +22,8 @@ const row = ( isin: null, ticker: null, logoName: null, + sector: null, + issuer: null, }); const portfolio = (positions: PortfolioPositionDto[]): PortfolioDto => ({ positions, diff --git a/src/modules/investments/investment-sync-plan.spec.ts b/src/modules/investments/investment-sync-plan.spec.ts index 8cb4dd1..09793d2 100644 --- a/src/modules/investments/investment-sync-plan.spec.ts +++ b/src/modules/investments/investment-sync-plan.spec.ts @@ -1,196 +1,16 @@ -import { - accountHistory, - ledgerRange, - planInvestmentScreen, -} from './investment-sync-plan'; +import { accountHistory, ledgerRange } from './investment-sync-plan'; -describe('screen dependency and history planning', () => { +// Missing opening dates must not silently narrow the reported history boundary. +describe('account history boundaries', () => { const now = new Date('2026-09-25T12:00:00Z'); - const context = { - membershipKnown: true, - accounts: [{ accountId: 'a', openedOn: '2020-01-01' }], - instruments: [ - { - uid: 'uid', - type: 'STOCK' as const, - historyFrom: new Date('2020-01-01Z'), - }, - ], - }; - it('prioritizes discovery, the selected period, then history; Moscow dates are converted once', () => { - const plan = planInvestmentScreen( - { - screen: 'income', - accountId: 'a', - from: '2024-02-01', - to: '2024-02-29', - }, - context, - now, - ); - expect(plan.sources[0].kind).toBe('membership'); - const selected = plan.sources.filter( - (source) => source.kind === 'operations' && source.priority === 'screen', - ); - // The single selected range is chunked into month-aligned intervals. - expect(selected).toHaveLength(2); - expect( - selected.every( - (source) => - source.from != null && - source.from.toISOString() >= '2024-01-31T21:00:00.000Z' && - source.to != null && - source.to.toISOString() <= '2024-02-29T21:00:00.000Z', - ), - ).toBe(true); - expect( - selected.some( - (source) => source.from?.toISOString() === '2024-01-31T21:00:00.000Z', - ), - ).toBe(true); - expect( - selected.some( - (source) => source.to?.toISOString() === '2024-02-29T21:00:00.000Z', - ), - ).toBe(true); - expect(plan.unavailable).toEqual([]); - const firstHistory = plan.sources.findIndex( - (source) => source.priority === 'history', - ); - expect( - plan.sources - .slice(firstHistory) - .every((source) => source.priority === 'history'), - ).toBe(true); - }); - it('force contains only shown dependencies and retains instrument reconciliation predicates', () => { - const plan = planInvestmentScreen( - { - screen: 'stock', - instrumentUid: 'uid', - from: '2026-01-01', - to: '2026-02-01', - candlePeriod: 'year', - }, - context, - now, - true, - ); - expect(plan.sources.map((source) => source.kind)).toEqual( - expect.arrayContaining([ - 'membership', - 'portfolio', - 'operations', - 'metadata', - 'quote', - 'company', - 'calendar', - 'candles', - ]), - ); - expect(plan.sources.every((source) => source.priority !== 'history')).toBe( - true, - ); - expect( - plan.sources - .filter((source) => source.kind === 'operations') - .every((source) => source.instrumentUid === 'uid'), - ).toBe(true); - const accounts = planInvestmentScreen( - { screen: 'accounts' }, - context, - now, - true, - ); - expect( - accounts.sources.some((source) => source.kind === 'operations'), - ).toBe(false); - }); - it('never invents an account or candle history boundary; discovery gates unknown membership', () => { - const unknown = { - ...context, - accounts: [{ accountId: 'a', openedOn: null }], - instruments: [], - }; - expect(accountHistory(unknown.accounts, now)[0].from.toISOString()).toBe( + const accounts = [{ accountId: 'a', openedOn: null }]; + + it('retains the unknown opening boundary for history and instrument ledger', () => { + expect(accountHistory(accounts, now)[0].from.toISOString()).toBe( '1970-01-01T00:00:00.000Z', ); expect( - ledgerRange( - undefined, - undefined, - 'uid', - unknown.accounts, - now, - ).from.toISOString(), + ledgerRange(undefined, undefined, 'uid', accounts, now).from.toISOString(), ).toBe('1970-01-01T00:00:00.000Z'); - const plan = planInvestmentScreen( - { screen: 'bond', instrumentUid: 'uid', candlePeriod: 'all' }, - unknown, - now, - ); - expect(plan.sources.some((source) => source.kind === 'candles')).toBe( - false, - ); - expect(plan.unavailable[0].reason).toBe('history_start_unavailable'); - expect( - planInvestmentScreen( - { screen: 'account' }, - { ...context, membershipKnown: false }, - now, - ).sources, - ).toEqual([{ kind: 'membership', priority: 'discovery' }]); - }); - it('keeps UTC event days separate from Moscow instrument calendar days at year rollover', () => { - const rollover = new Date('2026-12-31T22:00:00Z'); - const events = planInvestmentScreen( - { screen: 'events' }, - context, - rollover, - true, - ); - expect( - events.sources.find((source) => source.kind === 'calendar')?.from, - ).toEqual(new Date('2026-12-31Z')); - const stock = planInvestmentScreen( - { screen: 'stock', instrumentUid: 'uid' }, - context, - rollover, - true, - ); - expect( - stock.sources.find((source) => source.kind === 'calendar'), - ).toMatchObject({ - from: new Date('2022-01-01Z'), - to: new Date('2028-01-01Z'), - }); - expect(rollover.toISOString()).toBe('2026-12-31T22:00:00.000Z'); - }); - - it('keeps search queries separate and events scoped to their inclusive period', () => { - const search = planInvestmentScreen( - { screen: 'stocks', query: ' sber ' }, - context, - now, - true, - ); - expect( - search.sources.find((source) => source.kind === 'catalog')?.query, - ).toBe('SBER'); - const events = planInvestmentScreen( - { screen: 'events', from: '2026-10-01', to: '2026-10-31' }, - context, - now, - true, - ); - expect( - events.sources.find((source) => source.kind === 'calendar'), - ).toMatchObject({ - from: new Date('2026-10-01Z'), - to: new Date('2026-10-31Z'), - }); - expect(events.sources.some((source) => source.kind === 'operations')).toBe( - false, - ); }); }); diff --git a/src/modules/investments/investment-sync-plan.ts b/src/modules/investments/investment-sync-plan.ts index 53a4560..e3807b4 100644 --- a/src/modules/investments/investment-sync-plan.ts +++ b/src/modules/investments/investment-sync-plan.ts @@ -1,267 +1,11 @@ -import { ISO_DATE_LENGTH, ISO_DATE_START } from '../../common/time'; -import { planIncomePeriod } from './income/income.api'; -import { dateOnlyBounds, moscowDate } from './investment-calendar'; -import { - type OperationInterval, - operationIntervals, -} from './operations/domain/operations-sync-plan'; +import type { OperationInterval } from './operations/domain/operations-sync-plan'; -const FORECAST_YEARS = 1; const JANUARY_INDEX = 0; const FIRST_DAY_OF_MONTH = 1; const EXCLUSIVE_END_OFFSET_MS = 1; const NO_END_OFFSET_MS = 0; -export type InvestmentScreen = { - accountId?: string; - candlePeriod?: 'all' | 'halfYear' | 'month' | 'year'; - candleRange?: { from: Date; to: Date }; - from?: string; - instrumentUid?: string; - query?: string; - screen: - | 'account' | - 'accounts' | - 'bond' | - 'bonds' | - 'events' | - 'income' | - 'operations' | - 'stock' | - 'stocks'; - to?: string; -}; -const defaultEventDays = 30; -const incomeHistoryYears = 5; -const candleMonths = { month: 1, halfYear: 6, year: 12 }; - type Account = { accountId: string; openedOn: null | string }; -type Instrument = { - historyFrom: Date | null; - type: 'BOND' | 'STOCK'; - uid: string; -}; -type Source = { - accountId?: string; - from?: Date; - instrumentType?: 'BOND' | 'STOCK'; - instrumentUid?: string; - kind: - | 'calendar' | - 'candles' | - 'catalog' | - 'company' | - 'membership' | - 'metadata' | - 'metrics' | - 'operations' | - 'portfolio' | - 'quote'; - priority: 'discovery' | 'history' | 'screen'; - query?: string; - to?: Date; -}; - -// Expands screen scope, never credentials or arbitrary caller-supplied jobs. -// Owners execute their own sources and expand instruments after discovery. -export function planInvestmentScreen( - screen: InvestmentScreen, - context: { - accounts: Account[]; - instruments: Instrument[]; - membershipKnown: boolean; - }, - now: Date, - force = false, -) { - const sources: Source[] = [{ kind: 'membership', priority: 'discovery' }]; - const accounts = context.accounts.filter( - (account) => - !screen.accountId || - screen.accountId === 'all' || - account.accountId === screen.accountId, - ); - const unavailable: { reason: string; resourceKey: string }[] = []; - const operations = (from: Date, to: Date, instrumentUid?: string) => { - const targets = accounts.flatMap((account) => - from < to - ? [ - { - accountId: account.accountId, - from, - to, - ...(instrumentUid ? { instrumentUid } : {}), - }, - ] - : [], - ); - sources.push( - ...operationIntervals(targets).map((range) => ({ - ...range, - kind: 'operations' as const, - priority: 'screen' as const, - })), - ); - }; - const portfolio = () => - sources.push( - ...accounts.map(({ accountId }) => ({ - kind: 'portfolio' as const, - accountId, - priority: 'screen' as const, - })), - ); - if (!context.membershipKnown) return { sources, unavailable, now }; - if ( - ['account', 'accounts', 'bond', 'events', 'stock'].includes(screen.screen) - ) { - portfolio(); - } - if ( - screen.screen === 'operations' || - screen.screen === 'stock' || - screen.screen === 'bond' - ) { - const range = ledgerRange( - screen.from, - screen.to, - screen.instrumentUid, - accounts, - now, - ); - operations(range.from, range.to, force ? screen.instrumentUid : undefined); - } - if (screen.screen === 'account' || screen.screen === 'income') { - const period = planIncomePeriod( - screen.screen === 'income' ? screen.from : undefined, - screen.screen === 'income' ? screen.to : undefined, - ); - const bounds = dateOnlyBounds(period.selected, now); - operations(bounds.from, bounds.to); - if (screen.screen === 'account') { - const range = ledgerRange( - screen.from, - screen.to, - undefined, - accounts, - now, - ); - operations(range.from, range.to); - } - } - if (screen.screen === 'stocks' || screen.screen === 'bonds') { - sources.push({ - kind: 'catalog', - instrumentType: screen.screen === 'stocks' ? 'STOCK' : 'BOND', - query: screen.query?.trim().toUpperCase() ?? '', - priority: 'screen', - }); - } - const instruments = context.instruments.filter( - (instrument) => - !screen.instrumentUid || instrument.uid === screen.instrumentUid, - ); - if ( - screen.instrumentUid && - !instruments.length && - (screen.screen === 'stock' || screen.screen === 'bond') - ) { - instruments.push({ - uid: screen.instrumentUid, - type: screen.screen === 'stock' ? 'STOCK' : 'BOND', - historyFrom: null, - }); - } - for (const instrument of instruments) { - const identity = { - instrumentUid: instrument.uid, - instrumentType: instrument.type, - }; - sources.push({ ...identity, kind: 'metadata', priority: 'screen' }); - if (['bond', 'bonds', 'stock', 'stocks'].includes(screen.screen)) { - sources.push({ ...identity, kind: 'quote', priority: 'screen' }); - } - if (screen.screen === 'stock' || screen.screen === 'bond') { - sources.push({ - ...identity, - kind: screen.screen === 'stock' ? 'company' : 'metrics', - priority: 'screen', - }); - const from = new Date(now); - const period = screen.candlePeriod ?? 'month'; - if (period !== 'all') { - // eslint-disable-next-line security/detect-object-injection -- The period belongs to the validated candle-period union. - from.setUTCMonth(from.getUTCMonth() - candleMonths[period]); - } - const range = - screen.candleRange ?? - (period === 'all' - ? instrument.historyFrom - ? { from: instrument.historyFrom, to: now } - : null - : { from, to: now }); - if (range) { - sources.push({ - ...identity, - kind: 'candles', - ...range, - priority: 'screen', - }); - } else { - unavailable.push({ - resourceKey: `candles/${instrument.uid}/all`, - reason: 'history_start_unavailable', - }); - } - } - if (['account', 'bond', 'events', 'stock'].includes(screen.screen)) { - const day = - screen.screen === 'stock' || screen.screen === 'bond' - ? moscowDate(now) - : now.toISOString().slice(ISO_DATE_START, ISO_DATE_LENGTH); - const from = - screen.screen === 'events' && screen.from - ? new Date(`${screen.from}T00:00:00Z`) - : new Date(`${day}T00:00:00Z`); - const to = - screen.screen === 'events' && screen.to - ? new Date(`${screen.to}T00:00:00Z`) - : new Date(from); - if ( - !(screen.screen === 'events' && screen.to) && - screen.screen !== 'stock' && - screen.screen !== 'bond' - ) { - to.setUTCDate(to.getUTCDate() + defaultEventDays); - } - if (screen.screen === 'stock' || screen.screen === 'bond') { - from.setUTCFullYear(from.getUTCFullYear() - incomeHistoryYears); - to.setUTCFullYear(to.getUTCFullYear() + FORECAST_YEARS); - } - sources.push({ - ...identity, - kind: 'calendar', - from, - to, - priority: 'screen', - }); - } - } - if (!force && !['bonds', 'stocks'].includes(screen.screen)) { - sources.push( - ...accountHistory(accounts, now).map((range) => ({ - ...range, - kind: 'operations' as const, - priority: 'history' as const, - })), - ); - } - const priorities = { discovery: 0, screen: 1, history: 2 }; - sources.sort( - (left, right) => priorities[left.priority] - priorities[right.priority], - ); - return { sources, unavailable, now }; -} export function accountHistory( accounts: Account[], diff --git a/src/modules/investments/operation-category.ts b/src/modules/investments/operation-category.ts new file mode 100644 index 0000000..8335697 --- /dev/null +++ b/src/modules/investments/operation-category.ts @@ -0,0 +1,10 @@ +export const operationCategories = [ + 'trade', + 'income', + 'tax', + 'fee', + 'transfer', + 'other', +] as const; + +export type OperationCategory = (typeof operationCategories)[number]; diff --git a/src/modules/investments/operations/application/brokerage-operations-sync.service.ts b/src/modules/investments/operations/application/brokerage-operations-sync.service.ts index ca3dbf3..bc0a98b 100644 --- a/src/modules/investments/operations/application/brokerage-operations-sync.service.ts +++ b/src/modules/investments/operations/application/brokerage-operations-sync.service.ts @@ -6,13 +6,11 @@ import { MILLISECONDS_PER_SECOND } from '../../../../common/time'; import { isTbankErrorCode, TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code'; import { TbankOperationState } from '../../../../integrations/tbank/tbank-protocol'; import { TbankAccessService, TbankClient } from '../../../../integrations/tbank/tbank.api'; -import { accountHistory } from '../../investment-sync-plan'; import { type OperationInterval, operationIntervals, operationsSyncPriority, planOperationRead, - uncoveredOperationIntervals, } from '../domain/operations-sync-plan'; import { BrokerageOperationsRepository, @@ -48,7 +46,6 @@ export class BrokerageOperationsSyncService { intervals: OperationInterval[], options: { force?: boolean; - history?: boolean; movingEnd?: boolean; observe?: boolean; revision?: string; @@ -99,30 +96,6 @@ export class BrokerageOperationsSyncService { connection.revision, ) : null; - if (!options.observe && !options.force && options.history) { - const accountIds = new Set( - intervals.map((interval) => interval.accountId), - ); - const history = accountHistory( - connection.accounts.filter((account) => - accountIds.has(account.accountId), - ), - now, - ); - const missing = planOperationRead(history, coverage, now, { - movingEnd: true, - }).load; - const remaining = uncoveredOperationIntervals(missing, targets); - if (remaining.length) { - await this.request( - userId, - remaining, - false, - 'history', - connection.revision, - ); - } - } return { plan, handle }; } diff --git a/src/modules/investments/operations/application/brokerage-operations.service.ts b/src/modules/investments/operations/application/brokerage-operations.service.ts index 2a1ef46..fcabee2 100644 --- a/src/modules/investments/operations/application/brokerage-operations.service.ts +++ b/src/modules/investments/operations/application/brokerage-operations.service.ts @@ -6,21 +6,16 @@ import { InstrumentMetadataService, metadataKey, TbankAccessService, - TbankMembershipService, type TbankOperationItem, tbankSourceTtlMs, } from '../../../../integrations/tbank/tbank.api'; import { dateOnlyBounds, DateOnlyRange } from '../../investment-calendar'; import { investmentResourceMeta } from '../../investment-resource.mapper'; -import { - accountHistory, - ledgerRange, - planInvestmentScreen, -} from '../../investment-sync-plan'; +import { accountHistory, ledgerRange } from '../../investment-sync-plan'; import { hourlyAttempt, TbankSyncRepository, -} from '../../tbank-sync/tbank-sync.repository'; +} from '../../tbank-sync/tbank-sync.api'; import { OperationErrorCode } from '../domain/operation-error-code'; import { confirmedOperationRanges, @@ -37,7 +32,6 @@ import { } from '../presentation/brokerage-operation.dto'; import { BrokerageOperationsSyncService } from './brokerage-operations-sync.service'; -const EXCLUSIVE_END_OFFSET_MS = 1; const FIRST_RESULT_OFFSET = 0; const MISSING_MEMBERSHIP_REVISION = 0; const STALE_FRESHNESS_MS = 0; @@ -61,7 +55,6 @@ export class BrokerageOperationsService { private readonly metadata: InstrumentMetadataService, private readonly repository: BrokerageOperationsRepository, private readonly sync: BrokerageOperationsSyncService, - private readonly membership: TbankMembershipService, private readonly config: ConfigService, private readonly hourlySync: TbankSyncRepository, ) {} @@ -117,35 +110,14 @@ export class BrokerageOperationsService { cursor, }; if (cursor) await this.repository.readPage(readInput); - const screen = planInvestmentScreen( - { - screen: 'operations', - accountId, - from: from.toISOString(), - to: new Date(to.valueOf() - EXCLUSIVE_END_OFFSET_MS).toISOString(), - }, - { - accounts: selected, - instruments: [], - membershipKnown: Boolean(connection.membershipAt), - }, - now, - ); - const intervals = screen.sources.flatMap((source) => - source.kind === 'operations' && - source.priority === 'screen' && - source.accountId && - source.from && - source.to - ? [{ accountId: source.accountId, from: source.from, to: source.to }] - : [], + const intervals = selected.flatMap((account) => + from < to ? [{ accountId: account.accountId, from, to }] : [], ); // The ledger only starts the first-time load of missing intervals; the // mutable tail belongs to the hourly walk and membership is not rechecked. if (intervals.length) { await this.sync.readPlan(userId, intervals, { movingEnd, - history: true, revision: connection.revision, }); } @@ -406,32 +378,19 @@ export class BrokerageOperationsService { periods: ranges, }; let saved = await this.repository.readPage(input); - const schedule = async () => { - const current = await this.access.local(userId, accountId); - if (current.revision !== connection.revision) return; - const snapshot = await this.repository.readPage(input); - if (!snapshot?.connection.membershipAt) return; - const targets = snapshot.connection.accounts.flatMap((account) => - (snapshot.periods ?? []).map((period) => ({ + if (!observe) { + const targets = (saved?.connection.accounts ?? []).flatMap((account) => + (saved?.periods ?? []).map((period) => ({ accountId: account.accountId, ...period, })), ); if (targets.length) { await this.sync.readPlan(userId, targets, { - movingEnd: snapshot.periods?.some((period) => period.movingEnd), - history: true, + movingEnd: saved?.periods?.some((period) => period.movingEnd), revision: connection.revision, }); } - }; - if (!observe) { - const discovery = await this.membership.request( - userId, - connection.revision, - ); - if (discovery) void discovery.done.then(schedule).catch(() => undefined); - else await schedule(); saved = await this.repository.readPage(input); } const now = new Date(); diff --git a/src/modules/investments/operations/infrastructure/brokerage-operation.mapper.ts b/src/modules/investments/operations/infrastructure/brokerage-operation.mapper.ts index 5d2a752..df9cf0b 100644 --- a/src/modules/investments/operations/infrastructure/brokerage-operation.mapper.ts +++ b/src/modules/investments/operations/infrastructure/brokerage-operation.mapper.ts @@ -3,10 +3,8 @@ import { MILLISECONDS_PER_SECOND } from '../../../../common/time'; import { decimal } from '../../../../integrations/tbank/decimal'; import { TbankOperationType } from '../../../../integrations/tbank/tbank-protocol'; import { TbankMoney, TbankOperationItem } from '../../../../integrations/tbank/tbank.api'; -import { - BrokerageOperationDto, - OperationCategory, -} from '../presentation/brokerage-operation.dto'; +import { type OperationCategory } from '../../operation-category'; +import { BrokerageOperationDto } from '../presentation/brokerage-operation.dto'; const MISSING_TIMESTAMP_COMPONENT = 0; diff --git a/src/modules/investments/operations/operations.api.ts b/src/modules/investments/operations/operations.api.ts index e197d9d..c26238c 100644 --- a/src/modules/investments/operations/operations.api.ts +++ b/src/modules/investments/operations/operations.api.ts @@ -5,4 +5,3 @@ export type { OperationInterval } from './domain/operations-sync-plan'; // Public cross-module surface. Nest providers are exported by the matching module. export { BrokerageOperationDto } from './presentation/brokerage-operation.dto'; -export { operationCategories } from './presentation/brokerage-operation.dto'; diff --git a/src/modules/investments/operations/presentation/brokerage-operation.dto.ts b/src/modules/investments/operations/presentation/brokerage-operation.dto.ts index c808477..c349c7d 100644 --- a/src/modules/investments/operations/presentation/brokerage-operation.dto.ts +++ b/src/modules/investments/operations/presentation/brokerage-operation.dto.ts @@ -16,6 +16,7 @@ import { import { MoneyDto } from '../../../../common/money.dto'; import { AccountReferenceDto } from '../../account-reference.dto'; import { InvestmentResourceMetaDto } from '../../investment-resource.dto'; +import { operationCategories, type OperationCategory } from '../../operation-category'; const MIN_OPERATIONS_PAGE_SIZE = 1; @@ -23,16 +24,6 @@ const MAX_CURSOR_LENGTH = 4096; const MAX_OPERATIONS_PAGE_SIZE = 1000; export const DEFAULT_OPERATIONS_PAGE_SIZE = 100; -export const operationCategories = [ - 'trade', - 'income', - 'tax', - 'fee', - 'transfer', - 'other', -] as const; -export type OperationCategory = (typeof operationCategories)[number]; - export class BrokerageOperationsQueryDto { @ApiPropertyOptional({ format: 'uuid', diff --git a/src/modules/investments/portfolio/application/portfolio.service.spec.ts b/src/modules/investments/portfolio/application/portfolio.service.spec.ts index 008c160..448e2e6 100644 --- a/src/modules/investments/portfolio/application/portfolio.service.spec.ts +++ b/src/modules/investments/portfolio/application/portfolio.service.spec.ts @@ -6,9 +6,10 @@ import { PortfolioDto } from '../presentation/portfolio.dto'; import { PortfolioService } from './portfolio.service'; describe('persistent portfolio projection', () => { - const access = { local: jest.fn(), tokenForAttempt: jest.fn() }; + const access = { local: jest.fn(), tokenForAttempt: jest.fn(), rejectMembershipToken: jest.fn() }; const tbank = { getPortfolio: jest.fn(), getPositions: jest.fn() }; const metadata = { savedMany: jest.fn(), request: jest.fn() }; + const classification = { refresh: jest.fn() }; const repository = { read: jest.fn(), reserve: jest.fn(), @@ -23,6 +24,7 @@ describe('persistent portfolio projection', () => { access as never, tbank as never, metadata as never, + classification as never, repository as never, { request: jest.fn().mockResolvedValue(null) } as never, { event: jest.fn() } as never, @@ -34,6 +36,7 @@ describe('persistent portfolio projection', () => { accounts: [{ accountId: 'a' }, { accountId: 'b' }], }); access.tokenForAttempt.mockResolvedValue('synthetic'); + access.rejectMembershipToken.mockResolvedValue(undefined); repository.reserve.mockResolvedValue(null); repository.read.mockResolvedValue({ membershipAt, @@ -44,6 +47,37 @@ describe('persistent portfolio projection', () => { repository.publish.mockResolvedValue(true); metadata.savedMany.mockResolvedValue(new Map()); metadata.request.mockResolvedValue(null); + classification.refresh.mockResolvedValue(new Map()); + }); + it('refreshes classification before publishing and passes the hourly walk time', async () => { + const walkStartedAt = new Date('2026-10-31T12:00:00Z'); + const position = { instrumentType: 'share', instrumentUid: 'uid', figi: 'figi', quantity: { units: '1' } }; + repository.reserve.mockResolvedValue({ launch: true, work: { id: 'work', attemptId: 'attempt' } }); + tbank.getPortfolio.mockResolvedValue({ positions: [position] }); + tbank.getPositions.mockResolvedValue({ money: [] }); + classification.refresh.mockResolvedValue(new Map([[JSON.stringify(['uid', 'figi']), { + sector: 'Energy', issuer: { id: 'issuer', name: 'Issuer' }, + }]])); + const handle = await service.request('user', 'private', 'a', true, walkStartedAt); + assert(handle); + await handle.done; + expect(classification.refresh).toHaveBeenCalledWith('synthetic', [position], walkStartedAt); + expect(classification.refresh.mock.invocationCallOrder[0]).toBeLessThan(repository.publish.mock.invocationCallOrder[0]); + const snapshot = (repository.publish.mock.calls as [unknown, string, PortfolioDto][])[0][2]; + expect(snapshot.positions[0]).toEqual(expect.objectContaining({ + sector: 'Energy', issuer: { id: 'issuer', name: 'Issuer' }, + })); + }); + it('rejects an invalid classification token without publishing the snapshot', async () => { + repository.reserve.mockResolvedValue({ launch: true, work: { id: 'work', attemptId: 'attempt' } }); + tbank.getPortfolio.mockResolvedValue({ positions: [] }); + tbank.getPositions.mockResolvedValue({ money: [] }); + classification.refresh.mockRejectedValue(new Error('INVALID_TBANK_TOKEN')); + const handle = await service.request('user', 'private', 'a'); + assert(handle); + await handle.done; + expect(repository.publish).not.toHaveBeenCalled(); + expect(access.rejectMembershipToken).toHaveBeenCalled(); }); it('normalizes exact signed money, order, UID metadata and unknown types without metadata RPC', async () => { repository.reserve.mockResolvedValueOnce({ @@ -159,6 +193,31 @@ describe('persistent portfolio projection', () => { expect(snapshot.positions[1].averagePurchasePrice).toEqual({ currency: 'RUB', value: '2' }); }); + it('counts cash once when T-Bank also returns it as a CURRENCY position', async () => { + repository.reserve.mockResolvedValueOnce({ + launch: true, + work: { id: 'work', attemptId: 'attempt' }, + }); + tbank.getPortfolio.mockResolvedValue({ positions: [ + { figi: 'FUND', instrumentType: 'etf', quantity: { units: '1' }, currentPrice: { currency: 'RUB', units: '120', nano: 240000000 } }, + { figi: 'RUB000UTSTOM', instrumentType: 'currency', quantity: { units: '1', nano: 250000000 }, currentPrice: { currency: 'RUB', units: '1' } }, + ] }); + tbank.getPositions.mockResolvedValue({ money: [{ currency: 'RUB', units: '1', nano: 250000000 }] }); + repository.read.mockResolvedValue({ + membershipAt, + membershipRevision: 1, + accounts: [{ accountId: 'a', name: 'A', type: 'BROKERAGE', openedOn: null, portfolio: null }], + work: [], + }); + const handle = await service.request('user', 'private', 'a'); + assert(handle); + await handle.done; + const snapshot = (repository.publish.mock.calls as [unknown, string, PortfolioDto][])[0][2]; + expect(snapshot.totalsByCurrency).toEqual([{ currency: 'RUB', value: '121.49' }]); + expect(snapshot.positions.map((position) => position.type)).toEqual(['FUND', 'CURRENCY']); + expect(snapshot.structure.currencies[0].totalValue).toBe('121.49'); + }); + it('does not turn a missing account snapshot into aggregate zero; summaries stay independent', async () => { const snapshot: PortfolioDto = { lastSuccessfulAt: membershipAt.toISOString(), diff --git a/src/modules/investments/portfolio/application/portfolio.service.ts b/src/modules/investments/portfolio/application/portfolio.service.ts index 61852ff..b001098 100644 --- a/src/modules/investments/portfolio/application/portfolio.service.ts +++ b/src/modules/investments/portfolio/application/portfolio.service.ts @@ -20,7 +20,7 @@ import { investmentResourceMeta } from '../../investment-resource.mapper'; import { hourlyAttempt, TbankSyncRepository, -} from '../../tbank-sync/tbank-sync.repository'; +} from '../../tbank-sync/tbank-sync.api'; import { portfolioStructure } from '../domain/portfolio-structure'; import { sumMoneyByCurrency } from '../domain/portfolio-totals'; import { @@ -30,6 +30,7 @@ import { } from '../infrastructure/portfolio.repository'; import { mapPortfolio } from '../infrastructure/tbank-portfolio.mapper'; import { PortfolioDto } from '../presentation/portfolio.dto'; +import { type Classification, PositionClassificationService } from './position-classification.service'; const MISSING_DATA_REVISION = 0; const NO_ACCOUNTS = 0; @@ -46,6 +47,7 @@ export class PortfolioService { private readonly access: TbankAccessService, private readonly tbank: TbankClient, private readonly metadata: InstrumentMetadataService, + private readonly classification: PositionClassificationService, private readonly repository: PortfolioRepository, private readonly membership: TbankMembershipService, private readonly logger: StructuredLogger, @@ -88,6 +90,11 @@ export class PortfolioService { snapshot: account.portfolio ? { ...(account.portfolio.payload as unknown as PortfolioDto), + positions: (account.portfolio.payload as unknown as PortfolioDto).positions.map((position) => ({ + ...position, + sector: position.sector ?? null, + issuer: position.issuer ?? null, + })), lastSuccessfulAt: account.portfolio.fetchedAt.toISOString(), isStale: Date.now() - account.portfolio.fetchedAt.valueOf() >= @@ -231,8 +238,8 @@ export class PortfolioService { }; } - request(userId: string, revision: string, accountId: string, force = false) { - return this.queueSnapshot(userId, revision, accountId, force); + request(userId: string, revision: string, accountId: string, force = false, walkStartedAt?: Date) { + return this.queueSnapshot(userId, revision, accountId, force, walkStartedAt); } private queueSnapshot( @@ -240,6 +247,7 @@ export class PortfolioService { revision: string, accountId: string, force = false, + walkStartedAt = new Date(), ) { const reserved = this.reservations.then(async () => { const requested = await this.repository.reserve( @@ -268,9 +276,11 @@ export class PortfolioService { this.tbank.getPortfolio(token, accountId), this.tbank.getPositions(token, accountId), ]); + const classifications = await this.classification.refresh(token, portfolio.positions ?? [], walkStartedAt); const snapshot = await this.normalize( portfolio.positions ?? [], positions.money ?? [], + classifications, ); const published = await this.repository.publish( attempt, @@ -309,9 +319,10 @@ export class PortfolioService { private async normalize( upstream: NonNullable, money: TbankMoney[], + classifications: ReadonlyMap, ): Promise { const lastSuccessfulAt = new Date().toISOString().replace(/\.\d{3}Z$/, 'Z'); const savedMetadata = await this.metadata.savedMany(upstream); - return mapPortfolio(upstream, money, savedMetadata, lastSuccessfulAt); + return mapPortfolio(upstream, money, savedMetadata, lastSuccessfulAt, classifications); } } diff --git a/src/modules/investments/portfolio/application/position-classification.service.spec.ts b/src/modules/investments/portfolio/application/position-classification.service.spec.ts new file mode 100644 index 0000000..4b26d44 --- /dev/null +++ b/src/modules/investments/portfolio/application/position-classification.service.spec.ts @@ -0,0 +1,172 @@ +import { NotFoundException } from '@nestjs/common'; + +jest.mock('../../../../prisma/prisma.service', () => ({ PrismaService: class {} })); +import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code'; +import { InstrumentIdType } from '../../../../integrations/tbank/tbank-protocol'; +import { PositionClassificationService } from './position-classification.service'; + +describe('PositionClassificationService', () => { + const position = { instrumentType: 'share', instrumentUid: 'uid', figi: 'figi', quantity: { units: 1 } }; + const saved = { sector: 'Old', sectorFetchedAt: new Date('2026-10-30T12:00:00Z'), assetUid: 'asset', brandUid: 'brand', brandName: 'Old brand', brandFetchedAt: new Date('2026-10-30T12:00:00Z') }; + const tbank = { getShare: jest.fn(), getAssetBrand: jest.fn(), getBond: jest.fn(), getEtf: jest.fn() }; + const repository = { read: jest.fn(), save: jest.fn() }; + const logger = { event: jest.fn() }; + const service = new PositionClassificationService(tbank as never, repository as never, logger as never); + + beforeEach(() => { + jest.resetAllMocks(); + repository.read.mockResolvedValue(saved); + repository.save.mockImplementation((_key: string, update: Record) => Promise.resolve({ ...saved, ...update })); + tbank.getAssetBrand.mockResolvedValue({ assetUid: 'asset', brand: { uid: 'new-brand', name: 'New brand' } }); + }); + + it('does not fetch a value at 23:59, and refreshes it at 24:00', async () => { + await service.refresh('token', [position], new Date('2026-10-31T11:59:00Z')); + expect(tbank.getShare).not.toHaveBeenCalled(); + tbank.getShare.mockResolvedValue({ uid: 'uid', figi: 'figi', sector: ' Tech ', assetUid: 'asset' }); + const result = await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + expect(result.values().next().value).toEqual({ sector: 'Tech', issuer: { id: 'new-brand', name: 'New brand' } }); + expect(repository.save).toHaveBeenCalledWith(expect.any(String), { + sector: 'Tech', sectorFetchedAt: new Date('2026-10-31T12:00:00Z'), + brandUid: 'new-brand', brandName: 'New brand', brandFetchedAt: new Date('2026-10-31T12:00:00Z'), + }); + }); + + it('accepts a matching FIGI after UID NOT_FOUND', async () => { + tbank.getShare.mockRejectedValueOnce(new NotFoundException()).mockResolvedValueOnce({ uid: 'different', figi: 'figi', sector: 'Energy', assetUid: 'asset' }); + const result = await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + expect(tbank.getShare).toHaveBeenNthCalledWith(2, 'token', 'figi', InstrumentIdType.Figi, true); + expect(result.values().next().value?.sector).toBe('Energy'); + }); + + it('preserves saved values on mismatched instrument and returns refreshed values on database failure', async () => { + tbank.getShare.mockResolvedValue({ uid: 'wrong', sector: 'Energy' }); + const result = await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + expect(result.values().next().value?.sector).toBe('Old'); + repository.save.mockRejectedValue(new Error('db unavailable')); + tbank.getShare.mockResolvedValue({ uid: 'uid', sector: 'Energy', assetUid: 'asset' }); + const fallback = await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + expect(fallback.values().next().value?.sector).toBe('Energy'); + expect(logger.event).toHaveBeenCalled(); + }); + + it('does not retain an issuer from a previous asset when the asset changes', async () => { + repository.read.mockResolvedValue({ ...saved, brandFetchedAt: new Date('2026-10-31T12:00:00Z') }); + tbank.getShare.mockResolvedValue({ uid: 'uid', sector: 'Energy', assetUid: 'other' }); + tbank.getAssetBrand.mockResolvedValue({ assetUid: 'other', brand: { uid: 'other-brand', name: 'Other brand' } }); + const result = await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + expect(tbank.getAssetBrand).toHaveBeenCalledWith('token', 'other'); + expect(result.values().next().value?.issuer).toEqual({ id: 'other-brand', name: 'Other brand' }); + }); + + it('does not restart the brand clock when the brand fetch fails after an asset change', async () => { + repository.read.mockResolvedValue({ ...saved, brandFetchedAt: new Date('2026-10-31T12:00:00Z') }); + tbank.getShare.mockResolvedValue({ uid: 'uid', sector: 'Energy', assetUid: 'other' }); + tbank.getAssetBrand.mockRejectedValue(new Error('upstream unavailable')); + const result = await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + expect(result.values().next().value?.issuer).toBeNull(); + expect(repository.save).toHaveBeenCalledWith(expect.any(String), expect.objectContaining({ + assetUid: 'other', brandUid: null, brandName: null, brandFetchedAt: null, + })); + }); + + it('does not return the previous asset brand when saving a changed asset fails', async () => { + repository.save.mockRejectedValue(new Error('db unavailable')); + tbank.getShare.mockResolvedValue({ uid: 'uid', sector: 'Energy', assetUid: 'other' }); + tbank.getAssetBrand.mockRejectedValue(new Error('upstream unavailable')); + const result = await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + expect(result.values().next().value).toEqual({ sector: 'Energy', issuer: null }); + expect(repository.save).toHaveBeenCalledWith(expect.any(String), expect.objectContaining({ + assetUid: 'other', brandUid: null, brandName: null, brandFetchedAt: null, + })); + expect(logger.event).toHaveBeenCalled(); + }); + + it('records a fetched brand as null when a fresh instrument has no asset_uid', async () => { + repository.read.mockResolvedValue(null); + tbank.getShare.mockResolvedValue({ uid: 'uid', sector: 'Energy' }); + await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + expect(repository.save).toHaveBeenCalledWith(expect.any(String), expect.objectContaining({ + brandUid: null, brandName: null, brandFetchedAt: new Date('2026-10-31T12:00:00Z'), + })); + }); + + it('treats a NOT_FOUND answer as fetched and does not refetch it an hour later', async () => { + tbank.getShare.mockRejectedValue(new NotFoundException()); + await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + expect(repository.save).toHaveBeenCalledWith(expect.any(String), expect.objectContaining({ + sector: null, sectorFetchedAt: new Date('2026-10-31T12:00:00Z'), + })); + repository.read.mockResolvedValue({ ...saved, sector: null, sectorFetchedAt: new Date('2026-10-31T12:00:00Z'), assetUid: null, brandUid: null, brandName: null, brandFetchedAt: new Date('2026-10-31T12:00:00Z') }); + tbank.getShare.mockClear(); + await service.refresh('token', [position], new Date('2026-10-31T13:00:00Z')); + expect(tbank.getShare).not.toHaveBeenCalled(); + }); + + it('records a missing brand and restarts its clock when NOT_FOUND has no saved asset UID', async () => { + repository.read.mockResolvedValue(null); + tbank.getShare.mockRejectedValue(new NotFoundException()); + const startedAt = new Date('2026-10-31T12:00:00Z'); + const result = await service.refresh('token', [position], startedAt); + expect(result.values().next().value).toEqual({ sector: null, issuer: null }); + expect(repository.save).toHaveBeenCalledWith(expect.any(String), expect.objectContaining({ + brandUid: null, brandName: null, brandFetchedAt: startedAt, + })); + expect(tbank.getAssetBrand).not.toHaveBeenCalled(); + }); + + it('does not leak one refresh call\'s invalid token into another call', async () => { + tbank.getShare.mockRejectedValueOnce(new Error(TbankErrorCode.InvalidToken)) + .mockResolvedValueOnce({ uid: 'uid', figi: 'figi', sector: 'Energy', assetUid: 'asset' }); + await expect(service.refresh('first', [position], new Date('2026-10-31T12:00:00Z'))).rejects.toThrow(); + const result = await service.refresh('second', [position], new Date('2026-10-31T12:00:00Z')); + expect(result.values().next().value?.sector).toBe('Energy'); + }); + + it('propagates an invalid token before publishing a classification', async () => { + tbank.getShare.mockRejectedValue(new Error(TbankErrorCode.InvalidToken)); + const result = service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + await expect(result).rejects.toThrow(TbankErrorCode.InvalidToken); + expect(repository.save).not.toHaveBeenCalled(); + }); + + it('leaves a fresh brand untouched when the sector refresh finds the same asset', async () => { + repository.read.mockResolvedValue({ ...saved, brandFetchedAt: new Date('2026-10-31T12:00:00Z') }); + tbank.getShare.mockResolvedValue({ uid: 'uid', sector: 'Energy', assetUid: 'asset' }); + await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + expect(tbank.getAssetBrand).not.toHaveBeenCalled(); + }); + + it('skips unsupported types, zero-quantity positions, and shares a brand lookup across instruments', async () => { + repository.read.mockResolvedValue(null); + tbank.getShare.mockResolvedValue({ uid: 'uid', sector: 'Energy', assetUid: 'asset' }); + tbank.getBond.mockResolvedValue({ uid: 'bond', sector: 'Energy', assetUid: 'asset' }); + const result = await service.refresh('token', [ + { instrumentType: 'currency', instrumentUid: 'cash', figi: 'cash', quantity: { units: 1 } }, + { instrumentType: 'share', instrumentUid: 'empty', figi: 'empty', quantity: {} }, + { instrumentType: 'Share', instrumentUid: 'upper', figi: 'upper', quantity: { units: 1 } }, + position, + { instrumentType: 'bond', instrumentUid: 'bond', figi: 'bond-figi', quantity: { units: 1 } }, + ], new Date('2026-10-31T12:00:00Z')); + expect(result.size).toBe(3); + expect(tbank.getAssetBrand).toHaveBeenCalledTimes(1); + expect(tbank.getEtf).not.toHaveBeenCalled(); + }); + + it('skips duplicate instrument keys and retries a failed asset lookup for another instrument in the same walk', async () => { + repository.read.mockResolvedValue(null); + tbank.getShare.mockResolvedValue({ uid: 'uid', sector: 'Energy', assetUid: 'asset' }); + tbank.getBond.mockResolvedValue({ uid: 'bond', sector: 'Energy', assetUid: 'asset' }); + tbank.getAssetBrand.mockRejectedValueOnce(new Error('upstream unavailable')) + .mockResolvedValueOnce({ assetUid: 'asset', brand: { uid: 'new-brand', name: 'New brand' } }); + const result = await service.refresh('token', [position, position, + { instrumentType: 'bond', instrumentUid: 'bond', figi: 'bond-figi', quantity: { units: 1 } }, + ], new Date('2026-10-31T12:00:00Z')); + expect(result.size).toBe(2); + expect(tbank.getShare).toHaveBeenCalledTimes(1); + expect(tbank.getAssetBrand).toHaveBeenCalledTimes(2); + expect([...result.values()].map(({ issuer }) => issuer)).toEqual([ + null, { id: 'new-brand', name: 'New brand' }, + ]); + }); +}); diff --git a/src/modules/investments/portfolio/application/position-classification.service.ts b/src/modules/investments/portfolio/application/position-classification.service.ts new file mode 100644 index 0000000..1832f6a --- /dev/null +++ b/src/modules/investments/portfolio/application/position-classification.service.ts @@ -0,0 +1,184 @@ +import { Injectable, NotFoundException } from '@nestjs/common'; + +import { isZero } from '../../../../common/decimal'; +import { StructuredLogger } from '../../../../common/logging/structured-logger.service'; +import { MILLISECONDS_PER_DAY } from '../../../../common/time'; +import { decimal } from '../../../../integrations/tbank/decimal'; +import { isTbankErrorCode, TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code'; +import { InstrumentIdType } from '../../../../integrations/tbank/tbank-protocol'; +import { metadataKey, TbankClient, type TbankPortfolio } from '../../../../integrations/tbank/tbank.api'; +import { PositionClassificationRepository } from '../infrastructure/position-classification.repository'; + +export type Classification = { issuer: null | { id: string; name: string }; sector: null | string }; +type Position = NonNullable[number]; +type Instrument = { assetUid?: string; figi?: string; sector?: string; uid?: string }; +type Row = NonNullable>>; +type Update = Parameters[1]; + +@Injectable() +export class PositionClassificationService { + constructor( + private readonly tbank: TbankClient, + private readonly repository: PositionClassificationRepository, + private readonly logger: StructuredLogger, + ) {} + + async refresh(token: string, positions: Position[], walkStartedAt: Date): Promise> { + const result = new Map(); + const inFlight = new Map>(); + for (const position of positions) { + if (!['bond', 'etf', 'share'].includes((position.instrumentType ?? '').toLowerCase())) continue; + if (isZero(decimal(position.quantity ?? {}))) continue; + const key = metadataKey(position); + if (result.has(key)) continue; + result.set(key, await this.refreshOne(token, position, key, walkStartedAt, inFlight)); + } + return result; + } + + private async refreshOne(token: string, position: Position, key: string, startedAt: Date, + inFlight: Map>): Promise { + let saved: null | Row = null; + const update: Update = {}; + try { + saved = await this.repository.read(key); + const due = (at: Date | null | undefined) => !at || startedAt.valueOf() - at.valueOf() >= MILLISECONDS_PER_DAY; + const sectorDue = due(saved?.sectorFetchedAt); + const brandDue = due(saved?.brandFetchedAt); + let assetUid = saved?.assetUid ?? null; + let instrumentFresh = false; + if (sectorDue || (brandDue && !saved?.assetUid)) { + const instrument = await this.refreshInstrument(token, position, sectorDue, assetUid, update, startedAt); + assetUid = instrument.assetUid; + instrumentFresh = instrument.fresh; + } + const assetChanged = 'assetUid' in update && update.assetUid !== saved?.assetUid; + if (assetChanged) { + update.brandUid = null; + update.brandName = null; + update.brandFetchedAt = null; + } + if (brandDue || assetChanged) { + await this.refreshBrand(token, assetUid, update, startedAt, inFlight, instrumentFresh); + } + if (Object.keys(update).length) saved = await this.repository.save(key, update); + } catch (error) { + if (this.invalidToken(error)) throw error; + this.failed(error); + } + const current = { ...saved, ...update }; + return { + issuer: current.brandUid && current.brandName ? { id: current.brandUid, name: current.brandName } : null, + sector: current.sector ?? null, + }; + } + + private async refreshInstrument(token: string, position: Position, sectorDue: boolean, savedUid: null | string, + update: Update, startedAt: Date): Promise<{ assetUid: null | string; fresh: boolean }> { + try { + const lookup = await this.instrument(token, position); + if (lookup && !this.matches(lookup)) return { assetUid: savedUid, fresh: false }; + if (lookup) return { assetUid: this.acceptInstrument(lookup.found, savedUid, sectorDue, update, startedAt), fresh: true }; + if (sectorDue) this.clearSector(update, startedAt); + } catch (error) { + if (this.invalidToken(error)) throw error; + if (error instanceof NotFoundException) { + if (sectorDue) this.clearSector(update, startedAt); + } else this.failed(error); + } + return { assetUid: savedUid, fresh: false }; + } + + private clearSector(update: Update, startedAt: Date) { + update.sector = null; + update.sectorFetchedAt = startedAt; + } + + private acceptInstrument(found: Instrument, savedUid: null | string, sectorDue: boolean, update: Update, startedAt: Date) { + const assetUid = found.assetUid?.trim() || null; + if (assetUid !== savedUid) update.assetUid = assetUid; + if (sectorDue) this.setSector(update, found.sector, startedAt); + return assetUid; + } + + private setSector(update: Update, sector: string | undefined, startedAt: Date) { + update.sector = sector?.trim() || null; + update.sectorFetchedAt = startedAt; + } + + private async refreshBrand(token: string, assetUid: null | string, update: Update, startedAt: Date, + inFlight: Map>, instrumentFresh: boolean) { + if (!assetUid) { + if (instrumentFresh || 'assetUid' in update || 'sectorFetchedAt' in update) this.clearBrand(update, startedAt); + return; + } + try { + const asset = await this.dedupe(inFlight, assetUid, () => this.tbank.getAssetBrand(token, assetUid)); + if (asset && asset.assetUid !== assetUid) return; + const uid = asset?.brand?.uid?.trim() || null; + const name = asset?.brand?.name?.trim() || null; + update.brandUid = uid && name ? uid : null; + update.brandName = uid && name ? name : null; + update.brandFetchedAt = startedAt; + } catch (error) { + if (this.invalidToken(error)) throw error; + if (error instanceof NotFoundException) this.clearBrand(update, startedAt); + else this.failed(error); + } + } + + private clearBrand(update: Update, startedAt: Date) { + update.brandUid = null; + update.brandName = null; + update.brandFetchedAt = startedAt; + } + + private matches(lookup: { found: Instrument; id: string; idType: InstrumentIdType }) { + return (lookup.idType === InstrumentIdType.Uid ? lookup.found.uid : lookup.found.figi) === lookup.id; + } + + private invalidToken(error: unknown) { + return error instanceof Error && isTbankErrorCode(error.message, TbankErrorCode.InvalidToken); + } + + private async instrument(token: string, position: Position) { + const type = (position.instrumentType ?? '').toLowerCase(); + const lookup = (id: string, idType: InstrumentIdType) => { + switch (type) { + case 'bond': return this.tbank.getBond(token, id, idType, true); + case 'share': return this.tbank.getShare(token, id, idType, true); + default: return this.tbank.getEtf(token, id, idType, true); + } + }; + const uid = position.instrumentUid?.trim(); + const figi = position.figi?.trim(); + if (uid) { + try { + const found = await lookup(uid, InstrumentIdType.Uid); + if (found) return { found, id: uid, idType: InstrumentIdType.Uid }; + } catch (error) { + if (!(error instanceof NotFoundException)) throw error; + } + } + if (figi) { + const found = await lookup(figi, InstrumentIdType.Figi); + return found ? { found, id: figi, idType: InstrumentIdType.Figi } : null; + } + return null; + } + + private dedupe(map: Map>, key: string, run: () => Promise): Promise { + const pending = map.get(key); + if (pending) return pending as Promise; + const promise = run().catch((error: unknown) => { + map.delete(key); + throw error; + }); + map.set(key, promise); + return promise; + } + + private failed(error: unknown) { + this.logger.event('warn', 'portfolio.classification.failed', { code: error instanceof Error ? error.name : 'Unknown' }); + } +} diff --git a/src/modules/investments/portfolio/domain/portfolio-structure.spec.ts b/src/modules/investments/portfolio/domain/portfolio-structure.spec.ts index 4264872..c7ff8b7 100644 --- a/src/modules/investments/portfolio/domain/portfolio-structure.spec.ts +++ b/src/modules/investments/portfolio/domain/portfolio-structure.spec.ts @@ -14,12 +14,14 @@ const position = ( isin: null, ticker: null, logoName: null, + sector: null, + issuer: null, absoluteReturn: null, averagePurchasePrice: null, }); describe('portfolio structure', () => { - it('includes cash once and keeps currency instruments among other assets', () => { + it('counts cash once and skips CURRENCY positions that mirror it', () => { const result = portfolioStructure( [ position('STOCK', '60'), @@ -30,13 +32,13 @@ describe('portfolio structure', () => { ); expect(result.currencies[0]).toEqual({ currency: 'RUB', - totalValue: '100', + totalValue: '90', hasNegativeValues: false, categories: [ - { type: 'STOCK', value: '60', sharePercent: '60' }, - { type: 'BOND', value: '20', sharePercent: '20' }, - { type: 'OTHER', value: '10', sharePercent: '10' }, - { type: 'CASH', value: '10', sharePercent: '10' }, + { type: 'STOCK', value: '60', sharePercent: '66.666667' }, + { type: 'BOND', value: '20', sharePercent: '22.222222' }, + { type: 'OTHER', value: '0', sharePercent: '0' }, + { type: 'CASH', value: '10', sharePercent: '11.111111' }, ], }); }); diff --git a/src/modules/investments/portfolio/domain/portfolio-structure.ts b/src/modules/investments/portfolio/domain/portfolio-structure.ts index 6b01a9a..7498ff6 100644 --- a/src/modules/investments/portfolio/domain/portfolio-structure.ts +++ b/src/modules/investments/portfolio/domain/portfolio-structure.ts @@ -43,6 +43,8 @@ export function portfolioStructure( currencies.set(money.currency, group); }; for (const position of positions) { + // CURRENCY positions mirror cash, which is counted once as CASH below. + if (position.type === 'CURRENCY') continue; const type = ['BOND', 'STOCK'].includes(position.type) ? position.type : 'OTHER'; diff --git a/src/modules/investments/portfolio/infrastructure/position-classification.repository.ts b/src/modules/investments/portfolio/infrastructure/position-classification.repository.ts new file mode 100644 index 0000000..937cc18 --- /dev/null +++ b/src/modules/investments/portfolio/infrastructure/position-classification.repository.ts @@ -0,0 +1,29 @@ +import { Injectable } from '@nestjs/common'; + +import { PrismaService } from '../../../../prisma/prisma.service'; + +@Injectable() +export class PositionClassificationRepository { + constructor(private readonly prisma: PrismaService) {} + + read(key: string) { + return this.prisma.tbankInstrumentClassification.findUnique({ + where: { provider_instrumentKey: { provider: 'tbank', instrumentKey: key } }, + }); + } + + save(key: string, data: { + assetUid?: null | string; + brandFetchedAt?: Date | null; + brandName?: null | string; + brandUid?: null | string; + sector?: null | string; + sectorFetchedAt?: Date; + }) { + return this.prisma.tbankInstrumentClassification.upsert({ + where: { provider_instrumentKey: { provider: 'tbank', instrumentKey: key } }, + create: { provider: 'tbank', instrumentKey: key, ...data }, + update: data, + }); + } +} diff --git a/src/modules/investments/portfolio/infrastructure/tbank-portfolio.mapper.ts b/src/modules/investments/portfolio/infrastructure/tbank-portfolio.mapper.ts index 08c8873..47d3ae6 100644 --- a/src/modules/investments/portfolio/infrastructure/tbank-portfolio.mapper.ts +++ b/src/modules/investments/portfolio/infrastructure/tbank-portfolio.mapper.ts @@ -1,4 +1,5 @@ import type { Money } from '../../../../common/money.types'; +import type { Classification } from '../application/position-classification.service'; import type { PortfolioDto, PortfolioPositionDto } from '../presentation/portfolio.dto'; import { @@ -35,9 +36,10 @@ function mapPortfolio( cash: TbankMoney[], metadata: ReadonlyMap, lastSuccessfulAt: string, + classification: ReadonlyMap = new Map(), ): PortfolioDto { const positions = upstream - .map((position) => mapPosition(position, metadata.get(metadataKey(position))?.data ?? null)) + .map((position) => mapPosition(position, metadata.get(metadataKey(position))?.data ?? null, classification.get(metadataKey(position)))) .filter((position): position is PortfolioPositionDto => position !== null); const balances = new Map>(); for (const value of cash) { @@ -63,6 +65,7 @@ function mapPortfolio( function mapPosition( position: NonNullable[number], metadata: null | TbankInstrument, + classification?: Classification, ): null | PortfolioPositionDto { const quantity = decimal(position.quantity ?? {}); if (isZero(quantity)) return null; @@ -96,6 +99,8 @@ function mapPosition( isin: metadata?.isin?.trim() || null, ticker: metadata?.ticker?.trim() || null, logoName: metadata?.brand?.logoName?.trim() || null, + sector: classification?.sector ?? null, + issuer: classification?.issuer ?? null, }; } @@ -126,7 +131,8 @@ function totalsByCurrency( for (const value of money) addTotal(value.currency, decimal(value)); for (const position of positions) { const price = position.currentPrice; - if (!price) continue; + // CURRENCY positions mirror the cash balances already added above. + if (!price || types[position.instrumentType?.toLowerCase() ?? ''] === 'CURRENCY') continue; addTotal( price.currency, multiply(decimal(position.quantity ?? {}), decimal(price)), diff --git a/src/modules/investments/portfolio/portfolio.api.ts b/src/modules/investments/portfolio/portfolio.api.ts index fc567da..6deff8a 100644 --- a/src/modules/investments/portfolio/portfolio.api.ts +++ b/src/modules/investments/portfolio/portfolio.api.ts @@ -5,4 +5,5 @@ export { PortfolioService } from './application/portfolio.service'; export { PortfolioDto, PortfolioPositionDto, + PortfolioStructureDto, } from './presentation/portfolio.dto'; diff --git a/src/modules/investments/portfolio/portfolio.module.ts b/src/modules/investments/portfolio/portfolio.module.ts index 5b91496..05c3a6e 100644 --- a/src/modules/investments/portfolio/portfolio.module.ts +++ b/src/modules/investments/portfolio/portfolio.module.ts @@ -5,8 +5,10 @@ import { AuthModule } from '../../auth/auth.module'; import { TbankSyncRepositoryModule } from '../tbank-sync/tbank-sync-repository.module'; import { ForecastService } from './application/forecast.service'; import { PortfolioService } from './application/portfolio.service'; +import { PositionClassificationService } from './application/position-classification.service'; import { ForecastRepository } from './infrastructure/forecast.repository'; import { PortfolioRepository } from './infrastructure/portfolio.repository'; +import { PositionClassificationRepository } from './infrastructure/position-classification.repository'; import { PortfolioController } from './presentation/portfolio.controller'; @Module({ @@ -15,6 +17,8 @@ import { PortfolioController } from './presentation/portfolio.controller'; providers: [ PortfolioService, PortfolioRepository, + PositionClassificationService, + PositionClassificationRepository, ForecastService, ForecastRepository, ], diff --git a/src/modules/investments/portfolio/presentation/portfolio.dto.ts b/src/modules/investments/portfolio/presentation/portfolio.dto.ts index a2bcde6..fdd6bd9 100644 --- a/src/modules/investments/portfolio/presentation/portfolio.dto.ts +++ b/src/modules/investments/portfolio/presentation/portfolio.dto.ts @@ -4,6 +4,14 @@ import { MoneyDto } from '../../../../common/money.dto'; import { AccountReferenceDto } from '../../account-reference.dto'; import { InvestmentResourceMetaDto } from '../../investment-resource.dto'; +export class PortfolioIssuerDto { + @ApiProperty() + declare id: string; + + @ApiProperty() + declare name: string; +} + export class PortfolioPositionDto { @ApiPropertyOptional({ type: String, nullable: true }) instrumentUid?: null | string; @@ -52,6 +60,12 @@ export class PortfolioPositionDto { @ApiProperty({ type: String, nullable: true }) declare logoName: null | string; + + @ApiProperty({ type: String, nullable: true }) + declare sector: null | string; + + @ApiProperty({ type: PortfolioIssuerDto, nullable: true }) + declare issuer: null | PortfolioIssuerDto; } export class PortfolioStructureCategoryDto { diff --git a/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts b/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts index 8f89f01..1ac7b2a 100644 --- a/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts +++ b/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts @@ -132,6 +132,7 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { // upstream call inside one. The attempt outcome is derived from durable // step rows by the repository, never from walk-local state. private async walk(attempt: TbankSyncAttempt) { + const walkStartedAt = new Date(); this.active.add(attempt.attemptId); try { const { userId, revision } = attempt; @@ -155,6 +156,7 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { accounts, dispatch, dispatchProblems, + walkStartedAt, ); await this.waitWorks(userId, children.map((child) => child.id)); for (const account of accounts) { @@ -239,6 +241,7 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { accounts: Accounts, dispatch: DispatchStep, dispatchProblems: HourlySection[], + walkStartedAt: Date, ) { for (const account of accounts) { await this.dispatchPortfolio( @@ -247,6 +250,7 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { account.accountId, dispatch, dispatchProblems, + walkStartedAt, ); } for (const account of accounts) { @@ -274,10 +278,11 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { accountId: string, dispatch: DispatchStep, dispatchProblems: HourlySection[], + walkStartedAt: Date, ) { try { const work = await this.withBound( - this.portfolio.request(userId, revision, accountId, true), + this.portfolio.request(userId, revision, accountId, true, walkStartedAt), ); if (work && work !== BOUND_EXPIRED) { await dispatch({ accountId, id: work.id, section: 'portfolio' }); diff --git a/src/modules/investments/tbank-sync/tbank-sync.api.ts b/src/modules/investments/tbank-sync/tbank-sync.api.ts new file mode 100644 index 0000000..3350ac5 --- /dev/null +++ b/src/modules/investments/tbank-sync/tbank-sync.api.ts @@ -0,0 +1 @@ +export { hourlyAttempt, TbankSyncRepository } from './tbank-sync.repository'; diff --git a/src/modules/investments/tbank-sync/tbank-sync.dto.ts b/src/modules/investments/tbank-sync/tbank-sync.dto.ts index cfb5d78..d3a28f7 100644 --- a/src/modules/investments/tbank-sync/tbank-sync.dto.ts +++ b/src/modules/investments/tbank-sync/tbank-sync.dto.ts @@ -23,7 +23,7 @@ import { import { MAX_INSTRUMENT_QUERY_LENGTH, MIN_INSTRUMENT_QUERY_LENGTH } from '../instrument-query'; import { InvestmentAttemptProblemDto } from '../investment-resource.dto'; -import { operationCategories } from '../operations/operations.api'; +import { operationCategories } from '../operation-category'; const MIN_ACCOUNT_ID_LENGTH = 1; const MIN_SYNC_PAGE_SIZE = 1; diff --git a/src/modules/investments/upcoming-events/upcoming-events.service.ts b/src/modules/investments/upcoming-events/upcoming-events.service.ts index a3f905b..e36d9ae 100644 --- a/src/modules/investments/upcoming-events/upcoming-events.service.ts +++ b/src/modules/investments/upcoming-events/upcoming-events.service.ts @@ -24,7 +24,7 @@ import { import { hourlyAttempt, TbankSyncRepository, -} from '../tbank-sync/tbank-sync.repository'; +} from '../tbank-sync/tbank-sync.api'; import { bondEvent, dividendEvent } from './upcoming-event.mapper'; import { UpcomingEventDto, @@ -32,7 +32,7 @@ import { UpcomingEventsQueryDto, } from './upcoming-events.dto'; -enum EventsErrorCode { InvalidPeriod = 'INVALID_EVENTS_PERIOD' } +export enum EventsErrorCode { InvalidPeriod = 'INVALID_EVENTS_PERIOD' } const FIRST_BATCH_OFFSET = 0; diff --git a/src/openapi.ts b/src/openapi.ts index 3ea3fda..0857d4d 100644 --- a/src/openapi.ts +++ b/src/openapi.ts @@ -1,12 +1,38 @@ import { INestApplication } from '@nestjs/common'; import { DocumentBuilder, SwaggerModule } from '@nestjs/swagger'; +import { HttpErrorCode } from './common/error-code'; +import { MoexErrorCode } from './integrations/moex/moex.client'; +import { NcrErrorCode } from './integrations/ncr/ncr-ratings.client'; +import { TbankErrorCode } from './integrations/tbank/tbank-error-code'; +import { AuthErrorCode } from './modules/auth/application/auth-error-code'; +import { BankErrorCode } from './modules/banking/bank-cards/application/bank-card-input'; +import { IncomeErrorCode } from './modules/investments/income/income-period'; +import { InstrumentErrorCode } from './modules/investments/instruments/domain/instrument-error-code'; +import { OperationErrorCode } from './modules/investments/operations/domain/operation-error-code'; +import { EventsErrorCode } from './modules/investments/upcoming-events/upcoming-events.service'; + export const API_PREFIX = 'api/v1'; // Nest Swagger cannot express additionalProperties on a class, so closed request // bodies are restored here after generation; the runtime allowlist stays in the controller. const CLOSED_SCHEMAS = ['StockScreenerCriteriaDto']; +// Every code HttpExceptionFilter can emit. Assembled at the composition root so +// common/ProblemDetailsDto does not depend on modules. +export const PROBLEM_CODES = [ + ...Object.values(HttpErrorCode), + ...Object.values(AuthErrorCode), + ...Object.values(OperationErrorCode), + ...Object.values(InstrumentErrorCode), + ...Object.values(TbankErrorCode), + ...Object.values(BankErrorCode), + ...Object.values(MoexErrorCode), + ...Object.values(NcrErrorCode), + ...Object.values(EventsErrorCode), + ...Object.values(IncomeErrorCode), +]; + export function createOpenApiDocument(app: INestApplication) { const document = SwaggerModule.createDocument( app, @@ -18,5 +44,8 @@ export function createOpenApiDocument(app: INestApplication) { schema.additionalProperties = false; } } + const problem = schemas.ProblemDetailsDto; + const problemCode = problem && 'properties' in problem ? problem.properties?.code : undefined; + if (problemCode && !('$ref' in problemCode)) problemCode.enum = PROBLEM_CODES; return document; } diff --git a/test/openapi-contract.e2e-spec.ts b/test/openapi-contract.e2e-spec.ts index f9eef12..816eb0f 100644 --- a/test/openapi-contract.e2e-spec.ts +++ b/test/openapi-contract.e2e-spec.ts @@ -121,6 +121,55 @@ describe('OpenAPI contract (integration)', () => { expect(analytics.properties).not.toHaveProperty('yearComparison'); }); + it('publishes every emitted problem detail code as a named enum', () => { + const schemas = createOpenApiDocument(app).components?.schemas; + const expectedCodes = [ + 'BAD_REQUEST', + 'INTERNAL_SERVER_ERROR', + 'NOT_FOUND', + 'UNAUTHORIZED', + 'UNSUPPORTED_MEDIA_TYPE', + 'VALIDATION_ERROR', + 'EMAIL_ALREADY_REGISTERED', + 'INVALID_CREDENTIALS', + 'INVALID_OPERATIONS_CURSOR', + 'BOND_HISTORY_START_UNAVAILABLE', + 'BOND_TICKER_AMBIGUOUS', + 'STOCK_HISTORY_START_UNAVAILABLE', + 'STOCK_TICKER_AMBIGUOUS', + 'INVALID_TBANK_TOKEN', + 'TBANK_NOT_CONNECTED', + 'TBANK_UNAVAILABLE', + 'BANK_CARD_ARCHIVED', + 'BANK_DEPENDENT_REFUNDS', + 'BANK_DUPLICATE_CATEGORY', + 'BANK_IMMUTABLE_OPERATION', + 'BANK_INTENT_CONFLICT', + 'BANK_INVALID_DATE', + 'BANK_INVALID_INPUT', + 'BANK_INVALID_MONEY', + 'BANK_INVALID_REFUND', + 'BANK_NOT_FOUND', + 'BANK_OPENING_DATE_CONFLICT', + 'BANK_OPERATION_DELETED', + 'BANK_REFUND_LIMIT', + 'BANK_VERSION_CONFLICT', + 'BANK_WRITE_CONFLICT', + 'MOEX_UNAVAILABLE', + 'NCR_UNAVAILABLE', + 'INVALID_EVENTS_PERIOD', + 'INVALID_INCOME_PERIOD', + ]; + + const problem = schemas?.ProblemDetailsDto; + assert(problem && !('$ref' in problem)); + const codeSchema = problem.properties?.code; + assert(codeSchema && !('$ref' in codeSchema)); + expect(codeSchema.type).toBe('string'); + expect(codeSchema.enum).toEqual(expect.arrayContaining(expectedCodes)); + expect(codeSchema.enum).toHaveLength(expectedCodes.length); + }); + it('generates the runtime contract without server startup or deployment configuration', async () => { const schemas = createOpenApiDocument(app).components?.schemas; expect(schemas?.StockScreenerCriteriaDto).toMatchObject({ diff --git a/test/tbank-income-local.e2e-spec.ts b/test/tbank-income-local.e2e-spec.ts index c571d1d..6484605 100644 --- a/test/tbank-income-local.e2e-spec.ts +++ b/test/tbank-income-local.e2e-spec.ts @@ -297,6 +297,58 @@ describe('local income completeness and exact money (SQLite)', () => { expect(result.data?.calculatedAt).toBe('2025-02-01T00:00:00.000Z'); expect(upstream.getOperationsByCursor).not.toHaveBeenCalled(); }); + it('ordinary covered income reads local data without membership or operations RPC', async () => { + await prisma.tbankConnection.update({ + where: { userId: 'income-owner' }, + data: { membershipAt: new Date('2020-01-01Z') }, + }); + const response = await get('a', undefined, false).expect(200); + const result = response.body as IncomeAnalyticsResourceDto; + expect(result.data).not.toBeNull(); + expect(result.meta.completeness).toBe('complete'); + expect(upstream.getOpenAccounts).not.toHaveBeenCalled(); + expect(upstream.getOperationsByCursor).not.toHaveBeenCalled(); + }); + it('cold income imports only the missing requested interval', async () => { + await prisma.tbankConnection.update({ + where: { userId: 'income-owner' }, + data: { membershipAt: new Date('2020-01-01Z') }, + }); + let release: () => void = () => { + throw new Error('Upstream not initialized'); + }; + const blocked = new Promise<{ hasNext: boolean; items: never[] }>((resolve) => { + release = () => resolve({ hasNext: false, items: [] }); + }); + upstream.getOperationsByCursor.mockReturnValue(blocked); + const first = await get('a', 'from=2025-02-01&to=2025-02-28', false).expect(200); + const cold = first.body as IncomeAnalyticsResourceDto; + release(); + expect(cold.data).toBeNull(); + expect(cold.meta.completeness).toBe('incomplete'); + let result: IncomeAnalyticsResourceDto = cold; + for (let attempt = 0; attempt < 100; attempt++) { + const response = await get('a', 'from=2025-02-01&to=2025-02-28').expect(200); + result = response.body as IncomeAnalyticsResourceDto; + if (result.meta.completeness === 'complete') break; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + expect(result.data).not.toBeNull(); + expect(result.meta.completeness).toBe('complete'); + expect(upstream.getOpenAccounts).not.toHaveBeenCalled(); + expect(upstream.getOperationsByCursor.mock.calls).toEqual([ + ['synthetic', expect.objectContaining({ + accountId: 'a', + from: { seconds: '1738357199' }, + to: { seconds: '1738368001' }, + })], + ['synthetic', expect.objectContaining({ + accountId: 'a', + from: { seconds: '1738367999' }, + to: { seconds: '1740776401' }, + })], + ]); + }); it('returns while a cold upstream is blocked and validates access/period before scheduling', async () => { await get('absent', undefined, false).expect(404); await get('a', 'from=2025-02-31&to=2025-03-01', false).expect(400); diff --git a/test/tbank-ledger-local.e2e-spec.ts b/test/tbank-ledger-local.e2e-spec.ts index 04813bd..9d35709 100644 --- a/test/tbank-ledger-local.e2e-spec.ts +++ b/test/tbank-ledger-local.e2e-spec.ts @@ -113,6 +113,50 @@ describe('persistent ledger HTTP and cursor (SQLite)', () => { } } }); + it.each([ + 'from=2099-01-01', + 'from=2025-03-01&to=2025-02-01', + ])('empty ledger range %s returns 200 without upstream RPC', async (query) => { + const response = await request(app.getHttpServer()) + .get(`/api/v1/brokerage-accounts/a/operations?${query}`) + .set('Authorization', auth()) + .expect(200); + const result = response.body as BrokerageOperationsResourceDto; + expect(result.data?.items).toEqual([]); + expect(upstream.getOperationsByCursor).not.toHaveBeenCalled(); + }); + it('cold ledger imports only the missing requested interval', async () => { + upstream.getOperationsByCursor.mockResolvedValue({ + hasNext: false, + items: [], + }); + await put().expect(200); + await request(app.getHttpServer()) + .get('/api/v1/brokerage-accounts/a/operations?from=2025-02-01&to=2025-02-28') + .set('Authorization', auth()) + .expect(200); + let complete = false; + for (let attempt = 0; attempt < 100; attempt++) { + const response = await request(app.getHttpServer()) + .get('/api/v1/brokerage-accounts/a/operations?from=2025-02-01&to=2025-02-28&sync=observe') + .set('Authorization', auth()) + .expect(200); + const result = response.body as BrokerageOperationsResourceDto; + complete = result.meta.completeness === 'complete'; + if (complete) break; + await delay(10); + } + expect(complete).toBe(true); + expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1); + expect(upstream.getOperationsByCursor).toHaveBeenCalledWith( + 'synthetic', + expect.objectContaining({ + accountId: 'a', + from: { seconds: '1738367999' }, + to: { seconds: '1740787201' }, + }), + ); + }); it('reads ordered pages with identical IDs/timestamps across accounts and no upstream RPC, including after reopen', async () => { let cursor: null | string = null; const ids: string[] = []; @@ -409,7 +453,7 @@ describe('persistent ledger HTTP and cursor (SQLite)', () => { const result = (await get('sync=observe', 'a').expect(200)) .body as BrokerageOperationsResourceDto; expect(result.meta.coverage?.historyBoundary).toBe('unknown'); - expect(result.meta.coverage?.historyState).toBe('loading'); + expect(result.meta.coverage?.historyState).toBe('missing'); }, 30_000); it('returns cold missing while upstream is blocked, then confirms an empty interval; observe never retries failure', async () => { diff --git a/test/tbank-local-reads.e2e-spec.ts b/test/tbank-local-reads.e2e-spec.ts index bb18ca4..58aa495 100644 --- a/test/tbank-local-reads.e2e-spec.ts +++ b/test/tbank-local-reads.e2e-spec.ts @@ -216,7 +216,27 @@ describe('local accounts and portfolio HTTP (SQLite)', () => { ).toEqual({ stockPositionCount: 1, bondPositionCount: 0, + fundPositionCount: 0, + totalPositionCount: 1, totalsByCurrency: [{ currency: 'RUB', value: '21.100000002' }], + cashBalances: [{ currency: 'RUB', value: '1.1' }], + structure: { + currencies: [{ + currency: 'RUB', + totalValue: '21.100000002', + hasNegativeValues: false, + categories: [ + { type: 'STOCK', value: '20.000000002', sharePercent: '94.78673' }, + { type: 'BOND', value: '0', sharePercent: '0' }, + { type: 'OTHER', value: '0', sharePercent: '0' }, + { type: 'CASH', value: '1.1', sharePercent: '5.21327' }, + ], + }], + unvaluedCategories: [], + }, + unrealizedReturnByCurrency: [ + { currency: 'RUB', absolute: '0', costBasis: '0', percent: null, partial: true }, + ], }); const portfolio = await get('brokerage-accounts/account/portfolio').expect( 200, @@ -492,9 +512,18 @@ describe('local accounts and portfolio HTTP (SQLite)', () => { } const summaries = (await get('brokerage-accounts?sync=observe').expect(200)) .body as BrokerageAccountsResourceDto; - expect(summaries.data?.[0].portfolioSummary?.totalsByCurrency).toEqual([ - { currency: 'RUB', value: '21.1' }, - ]); + expect(summaries.data?.[0].portfolioSummary).toMatchObject({ + totalsByCurrency: [{ currency: 'RUB', value: '21.1' }], + cashBalances: [{ currency: 'RUB', value: '1.1' }], + structure: { currencies: [{ currency: 'RUB' }] }, + unrealizedReturnByCurrency: [ + { currency: 'RUB', absolute: '0', costBasis: '0', percent: null, partial: true }, + ], + stockPositionCount: 1, + bondPositionCount: 0, + fundPositionCount: 0, + totalPositionCount: 1, + }); expect(summaries.data?.[1].portfolioSummary).toBeNull(); expect(summaries.meta.availability).toBe('available'); expect(summaries.meta.completeness).toBe('complete'); diff --git a/test/tbank-operations-sync.e2e-spec.ts b/test/tbank-operations-sync.e2e-spec.ts index aad2971..11af629 100644 --- a/test/tbank-operations-sync.e2e-spec.ts +++ b/test/tbank-operations-sync.e2e-spec.ts @@ -76,7 +76,7 @@ describe('Request-driven operations synchronization (SQLite)', () => { where: { id }, }); if (work.state === expected) return work; - await delay(2); + await delay(2); // Backoff between exact DB-state polls; not a timing barrier. } throw new Error(`Sync did not reach ${expected}`); } @@ -113,6 +113,7 @@ describe('Request-driven operations synchronization (SQLite)', () => { }); service = createService(); }); + afterEach(() => jest.restoreAllMocks()); afterAll(async () => prisma.$disconnect()); it.each([[false, 1], [true, 1], [false, 2], [true, 2]] as const)( @@ -186,13 +187,17 @@ describe('Request-driven operations synchronization (SQLite)', () => { it('publishes the forced visible period before unrelated pending history and retains it if history fails', async () => { const gate = deferred(); + const started = deferred(); const march = { ...january, from: new Date('2026-03-01Z'), to: new Date('2026-04-01Z'), }; upstream.getOperationsByCursor - .mockReturnValueOnce(gate.promise) + .mockImplementationOnce(() => { + started.resolve(); + return gate.promise; + }) .mockImplementation((_token, request) => { if (Number(request.from?.seconds) + 1 === march.from.valueOf() / 1000) { return Promise.resolve({ @@ -205,7 +210,9 @@ describe('Request-driven operations synchronization (SQLite)', () => { const history = requiredHandle( await service.request('owner', [february, march], false, 'history'), ); - await delay(5); + // RPC starts after reserve/claim commits and pending reads the history plan. + // Waiting for state=running alone would not ensure that pending was read. + await started.promise; const screen = requiredHandle( await service.request('owner', [march], true), ); @@ -225,8 +232,12 @@ describe('Request-driven operations synchronization (SQLite)', () => { it('moves a newly visible interval ahead of pending backfill and records an unknown requested floor', async () => { const gate = deferred(); + const started = deferred(); upstream.getOperationsByCursor - .mockReturnValueOnce(gate.promise) + .mockImplementationOnce(() => { + started.resolve(); + return gate.promise; + }) .mockResolvedValue({ hasNext: false, items: [] }); const march = { ...january, @@ -241,7 +252,7 @@ describe('Request-driven operations synchronization (SQLite)', () => { 'history', ), ); - await delay(5); + await started.promise; await service.request('owner', [march], false); gate.resolve({ hasNext: false, items: [] }); await state(history, 'complete'); @@ -471,7 +482,7 @@ describe('Request-driven operations synchronization (SQLite)', () => { expect(saved.state).toBe( checkpoint === 'after-publish' ? 'complete' : 'running', ); - await delay(10); + await delay(10); // Bounded pause; absence of a recovery RPC cannot be awaited. expect(upstream.getOperationsByCursor).not.toHaveBeenCalled(); if (checkpoint === 'after-publish') { // eslint-disable-next-line jest/no-conditional-expect -- The parameterized fixture chooses the scenario; each branch asserts its outcome and the test also asserts common invariants. @@ -504,13 +515,17 @@ describe('Request-driven operations synchronization (SQLite)', () => { it('joins a request bridging active groups and keeps both original handles observable', async () => { const blocked = deferred(); + const started = deferred(); upstream.getOperationsByCursor - .mockReturnValueOnce(blocked.promise) + .mockImplementationOnce(() => { + started.resolve(); + return blocked.promise; + }) .mockResolvedValue({ hasNext: false, items: [] }); const early = { ...january, to: new Date('2026-01-10Z') }; const late = { ...january, from: new Date('2026-01-20Z') }; const first = requiredHandle(await service.request('owner', [early], true)); - await delay(5); + await started.promise; const second = requiredHandle(await service.request('owner', [late], true)); const merged = requiredHandle( await service.request('owner', [january], true), @@ -526,25 +541,28 @@ describe('Request-driven operations synchronization (SQLite)', () => { it('extends an overlapping atomic group without rereading its intersection or publishing a partial result', async () => { const first = deferred(); + const started = deferred(); const remainder = deferred(); + const remainderStarted = deferred(); const early = { ...january, to: new Date('2026-01-20Z') }; const late = { ...january, from: new Date('2026-01-10Z') }; upstream.getOperationsByCursor - .mockReturnValueOnce(first.promise) - .mockReturnValueOnce(remainder.promise); + .mockImplementationOnce(() => { + started.resolve(); + return first.promise; + }) + .mockImplementationOnce(() => { + remainderStarted.resolve(); + return remainder.promise; + }); const handle = requiredHandle( await service.request('owner', [early], true), ); - await delay(5); + await started.promise; + expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1); expect(await service.request('owner', [late], true)).toBe(handle); first.resolve({ hasNext: false, items: [] }); - for ( - let index = 0; - index < 500 && upstream.getOperationsByCursor.mock.calls.length < 2; - index += 1 - ) { - await delay(2); - } + await remainderStarted.promise; expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(2); expect(upstream.getOperationsByCursor.mock.calls[1][1].from).toEqual({ seconds: String(early.to.valueOf() / 1000 - 1), @@ -561,13 +579,17 @@ describe('Request-driven operations synchronization (SQLite)', () => { items: never[]; nextCursor: string; }>(); + const started = deferred(); upstream.getOperationsByCursor - .mockReturnValueOnce(first.promise) + .mockImplementationOnce(() => { + started.resolve(); + return first.promise; + }) .mockResolvedValue({ hasNext: false, items: [] }); const history = requiredHandle( await service.request('owner', [january], false, 'history'), ); - await delay(5); + await started.promise; const screen = requiredHandle( await service.request('owner', [february], true), ); @@ -583,19 +605,17 @@ describe('Request-driven operations synchronization (SQLite)', () => { it('keeps published initial windows on recovery and restarts only the unfinished month', async () => { const blocked = deferred(); + const blockedStarted = deferred(); upstream.getOperationsByCursor .mockResolvedValueOnce(complete) - .mockReturnValueOnce(blocked.promise); + .mockImplementationOnce(() => { + blockedStarted.resolve(); + return blocked.promise; + }); const handle = requiredHandle( await service.request('owner', [january, february], false), ); - for ( - let index = 0; - index < 500 && upstream.getOperationsByCursor.mock.calls.length < 2; - index += 1 - ) { - await delay(2); - } + await blockedStarted.promise; expect(await prisma.tbankOperation.count()).toBe(1); const before = await prisma.tbankOperationCoverage.findMany(); expect(before).toHaveLength(1); @@ -616,8 +636,15 @@ describe('Request-driven operations synchronization (SQLite)', () => { }) ).fetchedAt, ).toEqual(before[0].fetchedAt); + const stalePage = deferred>(); + const stagePage = repository.stagePage.bind(repository); + jest.spyOn(repository, 'stagePage').mockImplementation((...args) => { + const result = stagePage(...args); + stalePage.resolve(result); + return result; + }); blocked.resolve(complete); - await delay(10); + expect(await stalePage.promise).toBe(false); expect(await prisma.tbankOperation.count()).toBe(1); expect(await prisma.tbankOperationCoverage.count()).toBe(2); }); @@ -651,7 +678,11 @@ describe('Request-driven operations synchronization (SQLite)', () => { it('coalesces twenty requests, observes without RPC and completes only after the last page', async () => { const blocked = deferred(); - upstream.getOperationsByCursor.mockReturnValueOnce(blocked.promise); + const started = deferred(); + upstream.getOperationsByCursor.mockImplementationOnce(() => { + started.resolve(); + return blocked.promise; + }); const handles = await Promise.all( Array.from({ length: 20 }, () => service.request('owner', [january], true), @@ -659,7 +690,7 @@ describe('Request-driven operations synchronization (SQLite)', () => { ); expect(new Set(handles).size).toBe(1); const handle = requiredHandle(handles[0]); - await delay(5); + await started.promise; expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1); expect((await service.observe('owner', handle)).state).toBe('running'); expect(await prisma.tbankOperation.count()).toBe(0); @@ -699,6 +730,7 @@ describe('Request-driven operations synchronization (SQLite)', () => { expect(await prisma.tbankOperationCoverage.findMany()).toEqual(oldCoverage); const calls = upstream.getOperationsByCursor.mock.calls.length; await service.observe('owner', handle); + // Bounded pause; absence of a background retry cannot be awaited. await delay(10); expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(calls); upstream.getOperationsByCursor @@ -713,29 +745,29 @@ describe('Request-driven operations synchronization (SQLite)', () => { it('starts an unfinished interval from page one after a new run request and ignores late old pages', async () => { const blocked = deferred(); + const blockedStarted = deferred(); upstream.getOperationsByCursor .mockResolvedValueOnce({ hasNext: true, nextCursor: 'provider-cursor', items: [operation()], }) - .mockReturnValueOnce(blocked.promise); + .mockImplementationOnce(() => { + blockedStarted.resolve(); + return blocked.promise; + }); const handle = requiredHandle( await service.request('owner', [january], true), ); - for ( - let index = 0; - index < 500 && upstream.getOperationsByCursor.mock.calls.length < 2; - index += 1 - ) { - await delay(2); - } - expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(2); + await blockedStarted.promise; const before = await prisma.tbankSyncWork.findUniqueOrThrow({ where: { id: handle }, }); + expect(await prisma.tbankSyncStage.count()).toBeGreaterThan(0); + service = createService(); await service.observe('owner', handle); + // Bounded pause; absence of work adoption by observe cannot be awaited. await delay(5); expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(2); upstream.getOperationsByCursor.mockResolvedValueOnce({ @@ -752,8 +784,15 @@ describe('Request-driven operations synchronization (SQLite)', () => { expect( upstream.getOperationsByCursor.mock.calls[2][1].cursor, ).toBeUndefined(); + const stalePage = deferred>(); + const stagePage = repository.stagePage.bind(repository); + jest.spyOn(repository, 'stagePage').mockImplementation((...args) => { + const result = stagePage(...args); + stalePage.resolve(result); + return result; + }); blocked.resolve(complete); - await delay(10); + expect(await stalePage.promise).toBe(false); expect((await prisma.tbankOperation.findFirstOrThrow()).payload).toEqual( operation('30'), ); diff --git a/test/tbank-portfolio-publish.e2e-spec.ts b/test/tbank-portfolio-publish.e2e-spec.ts index 4c877e7..911be8d 100644 --- a/test/tbank-portfolio-publish.e2e-spec.ts +++ b/test/tbank-portfolio-publish.e2e-spec.ts @@ -1,6 +1,9 @@ import { ConfigService } from '@nestjs/config'; import assert from 'node:assert/strict'; +import type { Prisma } from '../generated/prisma/client'; +import type { PortfolioDto } from '../src/modules/investments/portfolio/presentation/portfolio.dto'; + import { PortfolioRepository } from '../src/modules/investments/portfolio/infrastructure/portfolio.repository'; import { PrismaService } from '../src/prisma/prisma.service'; @@ -16,9 +19,41 @@ describe('portfolio publication (SQLite)', () => { totalsByCurrency: [], positions: [], cashBalances: [], - structure: {}, - } as never; + structure: { currencies: [], unvaluedCategories: [] }, + } satisfies PortfolioDto; const accountId = 'account'; + const valuedSnapshot = { + ...snapshot, + totalsByCurrency: [ + { currency: 'RUB', value: '200' }, + { currency: 'USD', value: '10' }, + ], + } satisfies PortfolioDto; + const createRunningAttempt = () => + prisma.tbankSyncWork.create({ + data: { + id: 'live', + userId: 'owner', + revision: 'current', + scopeKey: `portfolio:${accountId}`, + intent: {}, + state: 'running', + attemptId: 'live-attempt', + processRunId: 'run', + priority: 1, + progress: {}, + }, + }); + const interceptTransaction = (intercept: (tx: Prisma.TransactionClient) => void) => { + const transaction = prisma.$transaction.bind(prisma); + jest.spyOn(prisma, '$transaction').mockImplementation( + (callback: (tx: Prisma.TransactionClient) => Promise) => + transaction((tx) => { + intercept(tx); + return callback(tx); + }), + ); + }; beforeAll(async () => { await prisma.$connect(); @@ -27,7 +62,9 @@ describe('portfolio publication (SQLite)', () => { }); }); afterAll(async () => prisma.$disconnect()); + afterEach(() => jest.restoreAllMocks()); beforeEach(async () => { + await prisma.tbankPortfolioValueHistory.deleteMany(); await prisma.tbankConnection.deleteMany(); await prisma.tbankConnection.create({ data: { @@ -48,20 +85,7 @@ describe('portfolio publication (SQLite)', () => { }); it('publishes the same snapshot for the running attempt of the current revision', async () => { - await prisma.tbankSyncWork.create({ - data: { - id: 'live', - userId: 'owner', - revision: 'current', - scopeKey: `portfolio:${accountId}`, - intent: {}, - state: 'running', - attemptId: 'live-attempt', - processRunId: 'run', - priority: 1, - progress: {}, - }, - }); + await createRunningAttempt(); const published = await repository.publish( { id: 'live', attemptId: 'live-attempt', userId: 'owner', revision: 'current' }, accountId, @@ -70,4 +94,99 @@ describe('portfolio publication (SQLite)', () => { assert.equal(published, true); expect(await prisma.tbankPortfolio.count()).toBe(1); }); + + it('rolls back the snapshot, first value point and claim when the second point fails', async () => { + const previousFetchedAt = new Date('2025-12-31T00:00:00.000Z'); + const previousCapturedAt = new Date('2025-12-31T01:00:00.000Z'); + await prisma.tbankPortfolio.create({ + data: { + userId: 'owner', + revision: 'current', + accountId, + payload: snapshot, + fetchedAt: previousFetchedAt, + }, + }); + await prisma.tbankPortfolioValueHistory.create({ + data: { + userId: 'owner', + accountId, + points: { + create: { + attemptId: 'previous-attempt', + currency: 'RUB', + value: '100', + capturedAt: previousCapturedAt, + }, + }, + }, + }); + await createRunningAttempt(); + interceptTransaction((tx) => { + const upsert = tx.tbankPortfolioValuePoint.upsert.bind(tx.tbankPortfolioValuePoint); + jest + .spyOn(tx.tbankPortfolioValuePoint, 'upsert') + .mockImplementationOnce((args) => upsert(args)) + .mockRejectedValueOnce(new Error('injected second point failure')); + }); + + await expect( + repository.publish( + { id: 'live', attemptId: 'live-attempt', userId: 'owner', revision: 'current' }, + accountId, + valuedSnapshot, + ), + ).rejects.toThrow('injected second point failure'); + + expect( + await prisma.tbankPortfolio.findUnique({ + where: { userId_revision_accountId: { userId: 'owner', revision: 'current', accountId } }, + }), + ).toMatchObject({ payload: snapshot, fetchedAt: previousFetchedAt, dataRevision: 1 }); + expect( + await prisma.tbankPortfolioValuePoint.findMany({ + select: { attemptId: true, value: true, capturedAt: true }, + }), + ).toEqual([ + { attemptId: 'previous-attempt', value: '100', capturedAt: previousCapturedAt }, + ]); + expect(await prisma.tbankSyncWork.findUniqueOrThrow({ where: { id: 'live' } })) + .toMatchObject({ state: 'running', attemptId: 'live-attempt' }); + }); + + it.each(['stale attempt', 'complete work', 'invalid token'] as const)( + 'does not publish for %s', + async (scenario) => { + await createRunningAttempt(); + if (scenario === 'complete work') { + await prisma.tbankSyncWork.update({ + where: { id: 'live' }, + data: { state: 'complete' }, + }); + } + if (scenario === 'invalid token') { + await prisma.tbankConnection.update({ + where: { userId: 'owner' }, + data: { invalidToken: true }, + }); + } + + await expect( + repository.publish( + { + id: 'live', + attemptId: scenario === 'stale attempt' ? 'stale-attempt' : 'live-attempt', + userId: 'owner', + revision: 'current', + }, + accountId, + valuedSnapshot, + ), + ).resolves.toBe(false); + + expect(await prisma.tbankPortfolio.count()).toBe(0); + expect(await prisma.tbankPortfolioValueHistory.count()).toBe(0); + expect(await prisma.tbankPortfolioValuePoint.count()).toBe(0); + }, + ); }); diff --git a/test/tbank-value-history.e2e-spec.ts b/test/tbank-value-history.e2e-spec.ts index e47eae2..92dfbd1 100644 --- a/test/tbank-value-history.e2e-spec.ts +++ b/test/tbank-value-history.e2e-spec.ts @@ -298,6 +298,8 @@ describe('T-Bank value history persistence and visibility (SQLite)', () => { isin: null, ticker: null, logoName: null, + sector: null, + issuer: null, }, ], cashBalances: [{ currency: 'USD', value: '50' }],