import { Injectable, NotFoundException } from '@nestjs/common'; import { isZero } from '#src/common/decimal'; import { StructuredLogger } from '#src/common/logging/structured-logger.service'; import { MILLISECONDS_PER_DAY } from '#src/common/time'; import { decimal } from '#src/integrations/tbank/decimal'; import { isTbankErrorCode, TbankErrorCode } from '#src/integrations/tbank/tbank-error-code'; import { InstrumentIdType } from '#src/integrations/tbank/tbank-protocol'; import { metadataKey, TbankClient, type TbankPortfolio } from '#src/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' }); } }