codex-bot
31b0e15579
feat: unify income payment cards and drop period comparisons (WealthLedger/main#33)
636 lines
18 KiB
TypeScript
636 lines
18 KiB
TypeScript
import {
|
|
ChannelCredentials,
|
|
Client,
|
|
ClientUnaryCall,
|
|
connectivityState,
|
|
loadPackageDefinition,
|
|
Metadata,
|
|
ServiceError,
|
|
status,
|
|
} from '@grpc/grpc-js';
|
|
import { loadSync } from '@grpc/proto-loader';
|
|
import {
|
|
Injectable,
|
|
NotFoundException,
|
|
ServiceUnavailableException,
|
|
} from '@nestjs/common';
|
|
import { ConfigService } from '@nestjs/config';
|
|
import { readFileSync } from 'node:fs';
|
|
import { join } from 'node:path';
|
|
|
|
import type {
|
|
TbankBond,
|
|
TbankCandle,
|
|
TbankCoupon,
|
|
TbankFundamentals,
|
|
TbankLastPrice,
|
|
TbankShare,
|
|
} from './tbank-market.types';
|
|
|
|
import { StructuredLogger } from '../../common/logging/structured-logger.service';
|
|
import { MILLISECONDS_PER_SECOND } from '../../common/time';
|
|
import { Environment } from '../../config/env.validation';
|
|
import { TbankErrorCode } from './tbank-error-code';
|
|
import {
|
|
InstrumentIdType,
|
|
TbankAccountStatus,
|
|
TbankCandleInterval,
|
|
TbankInstrumentStatus,
|
|
TbankInstrumentType,
|
|
TbankOperationState,
|
|
} from './tbank-protocol';
|
|
|
|
const INITIAL_REQUEST_COUNT = 0;
|
|
const MIN_RATE_DELAY_MS = 1;
|
|
const FIRST_ATTEMPT = 1;
|
|
const ATTEMPT_INCREMENT = 1;
|
|
const NO_RPC_ATTEMPTS = 0;
|
|
const NEXT_DAY_OFFSET = 1;
|
|
|
|
const RATE_WINDOW_MS = 1_000;
|
|
const RPC_MAX_ATTEMPTS = 2;
|
|
const MAX_QUEUE_SIZE = 50;
|
|
const RPC_RETRY_DELAY_MS = 1_000;
|
|
|
|
const requestDeadlineMs = 15_000;
|
|
|
|
type Unary<Request, Response> = (
|
|
request: Request,
|
|
metadata: Metadata,
|
|
options: { deadline: Date },
|
|
callback: (error: null | ServiceError, response: Response) => void,
|
|
) => ClientUnaryCall;
|
|
type TbankClientService = Client & Record<string, Unary<never, never>>;
|
|
type RpcDiagnostics = {
|
|
channelState?: string;
|
|
headersReceived?: boolean;
|
|
queueWaitMs?: number;
|
|
trackingId?: string;
|
|
};
|
|
export type TbankDecimal = { nano?: number | string; units?: number | string };
|
|
export type TbankMoney = TbankDecimal & { currency?: string };
|
|
export type TbankAccount = {
|
|
id: string;
|
|
name: string;
|
|
openedDate?: { seconds?: number | string };
|
|
type: string;
|
|
};
|
|
export type TbankPortfolio = {
|
|
positions?: {
|
|
averagePositionPrice?: TbankMoney;
|
|
currentPrice?: TbankMoney;
|
|
figi?: string;
|
|
instrumentType?: string;
|
|
instrumentUid?: string;
|
|
quantity?: TbankDecimal;
|
|
ticker?: string;
|
|
}[];
|
|
};
|
|
export type TbankPositions = { money?: TbankMoney[] };
|
|
export type TbankInstrument = {
|
|
brand?: { logoName?: null | string };
|
|
figi?: string;
|
|
instrumentType?: string;
|
|
isin?: string;
|
|
name?: string;
|
|
ticker?: string;
|
|
uid?: string;
|
|
};
|
|
export type TbankDividend = {
|
|
dividendNet?: TbankMoney;
|
|
dividendType?: string;
|
|
lastBuyDate?: TbankTimestamp;
|
|
paymentDate?: TbankTimestamp;
|
|
recordDate?: TbankTimestamp;
|
|
};
|
|
export type TbankTimestamp = { nanos?: number; seconds?: number | string };
|
|
export type TbankOperationItem = {
|
|
accruedInt?: TbankMoney;
|
|
commission?: TbankMoney;
|
|
cursor?: string;
|
|
date?: TbankTimestamp;
|
|
description?: string;
|
|
figi?: string;
|
|
id?: string;
|
|
instrumentType?: string;
|
|
instrumentUid?: string;
|
|
name?: string;
|
|
payment?: TbankMoney;
|
|
price?: TbankMoney;
|
|
quantity?: number | string;
|
|
quantityDone?: number | string;
|
|
state?: string;
|
|
ticker?: string;
|
|
type?: string;
|
|
};
|
|
export type TbankOperationsByCursorRequest = {
|
|
accountId: string;
|
|
cursor?: string;
|
|
from?: TbankTimestamp;
|
|
instrumentId?: string;
|
|
limit?: number;
|
|
state: TbankOperationState.Executed;
|
|
to?: TbankTimestamp;
|
|
};
|
|
export type TbankOperationsByCursorResponse = {
|
|
hasNext?: boolean;
|
|
items?: TbankOperationItem[];
|
|
nextCursor?: string;
|
|
};
|
|
type TbankNamespace = {
|
|
tinkoff: {
|
|
public: { invest: { api: { contract: { v1: Record<string, unknown> } } } };
|
|
};
|
|
};
|
|
|
|
class RateLimiter {
|
|
private readonly waiting: (() => void)[] = [];
|
|
private starts = INITIAL_REQUEST_COUNT;
|
|
private windowStartedAt = Date.now();
|
|
|
|
constructor(private readonly perSecond: number) {}
|
|
|
|
get size(): number {
|
|
return this.waiting.length;
|
|
}
|
|
|
|
async run<T>(operation: () => Promise<T>): Promise<T> {
|
|
await new Promise<void>((resolve) => {
|
|
this.waiting.push(resolve);
|
|
this.drain();
|
|
});
|
|
try {
|
|
return await operation();
|
|
} finally {
|
|
this.drain();
|
|
}
|
|
}
|
|
|
|
private drain(): void {
|
|
const now = Date.now();
|
|
if (now - this.windowStartedAt >= RATE_WINDOW_MS) {
|
|
this.windowStartedAt = now;
|
|
this.starts = INITIAL_REQUEST_COUNT;
|
|
}
|
|
while (this.starts < this.perSecond && this.waiting.length) {
|
|
this.starts++;
|
|
this.waiting.shift()?.();
|
|
}
|
|
if (this.waiting.length && this.starts === this.perSecond) {
|
|
setTimeout(
|
|
() => this.drain(),
|
|
Math.max(MIN_RATE_DELAY_MS, RATE_WINDOW_MS - (now - this.windowStartedAt)),
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
@Injectable()
|
|
export class TbankClient {
|
|
private readonly queue: RateLimiter;
|
|
private readonly users: TbankClientService;
|
|
private readonly operations: TbankClientService;
|
|
private readonly instruments: TbankClientService;
|
|
private readonly sparseInstruments: TbankClientService;
|
|
private readonly marketData: TbankClientService;
|
|
|
|
constructor(
|
|
config: ConfigService<Environment, true>,
|
|
private readonly logger: StructuredLogger,
|
|
) {
|
|
const protoRoot = join(__dirname, 'proto/contracts');
|
|
const definition = loadSync(
|
|
[
|
|
'users.proto',
|
|
'operations.proto',
|
|
'instruments.proto',
|
|
'marketdata.proto',
|
|
].map((file) => join(protoRoot, file)),
|
|
{
|
|
includeDirs: [protoRoot],
|
|
keepCase: false,
|
|
longs: String,
|
|
enums: String,
|
|
defaults: true,
|
|
oneofs: true,
|
|
},
|
|
);
|
|
const services = (
|
|
loadPackageDefinition(definition) as unknown as TbankNamespace
|
|
).tinkoff.public.invest.api.contract.v1;
|
|
const caPath = config.get<string | undefined>('TBANK_CA_CERT_PATH');
|
|
const credentials = caPath
|
|
// eslint-disable-next-line security/detect-non-literal-fs-filename -- The CA path is trusted operator configuration, never supplied by an HTTP request.
|
|
? ChannelCredentials.createSsl(readFileSync(caPath))
|
|
: ChannelCredentials.createSsl();
|
|
const endpoint = config.getOrThrow<string>('TBANK_GRPC_ENDPOINT');
|
|
const create = (name: string) =>
|
|
new (
|
|
// eslint-disable-next-line security/detect-object-injection -- Service and method names are internal RPC selections, not request input.
|
|
services[name] as new (
|
|
address: string,
|
|
credentials: ChannelCredentials,
|
|
) => TbankClientService
|
|
)(endpoint, credentials);
|
|
this.users = create('UsersService');
|
|
this.operations = create('OperationsService');
|
|
this.instruments = create('InstrumentsService');
|
|
this.marketData = create('MarketDataService');
|
|
// Preserve field presence for fundamentals; the established clients retain their decoding contract.
|
|
const sparse = (
|
|
loadPackageDefinition(
|
|
loadSync(join(protoRoot, 'instruments.proto'), {
|
|
includeDirs: [protoRoot],
|
|
keepCase: false,
|
|
longs: String,
|
|
enums: String,
|
|
defaults: false,
|
|
oneofs: true,
|
|
}),
|
|
) as unknown as TbankNamespace
|
|
).tinkoff.public.invest.api.contract.v1;
|
|
this.sparseInstruments = new (
|
|
sparse.InstrumentsService as new (
|
|
address: string,
|
|
credentials: ChannelCredentials,
|
|
) => TbankClientService
|
|
)(endpoint, credentials);
|
|
this.queue = new RateLimiter(
|
|
config.getOrThrow<number>('TBANK_RATE_LIMIT_PER_SECOND'),
|
|
);
|
|
this.logger.info('tbank.client.configured');
|
|
}
|
|
|
|
validateToken(token: string): Promise<TbankAccount[]> {
|
|
return this.getOpenAccounts(token);
|
|
}
|
|
|
|
getOpenAccounts(token: string): Promise<TbankAccount[]> {
|
|
return this.call<{ accounts?: TbankAccount[] }>(
|
|
token,
|
|
this.users,
|
|
'getAccounts',
|
|
{ status: TbankAccountStatus.Open },
|
|
).then((response) => response.accounts ?? []);
|
|
}
|
|
|
|
getPortfolio(token: string, accountId: string): Promise<TbankPortfolio> {
|
|
return this.call(token, this.operations, 'getPortfolio', { accountId });
|
|
}
|
|
|
|
getPositions(token: string, accountId: string): Promise<TbankPositions> {
|
|
return this.call(token, this.operations, 'getPositions', { accountId });
|
|
}
|
|
|
|
getOperationsByCursor(
|
|
token: string,
|
|
request: TbankOperationsByCursorRequest,
|
|
): Promise<TbankOperationsByCursorResponse> {
|
|
return this.call(token, this.operations, 'getOperationsByCursor', request);
|
|
}
|
|
|
|
getInstrumentByFigi(
|
|
token: string,
|
|
figi: string,
|
|
): Promise<null | TbankInstrument> {
|
|
return this.getInstrument(token, InstrumentIdType.Figi, figi);
|
|
}
|
|
|
|
getInstrumentByUid(
|
|
token: string,
|
|
instrumentUid: string,
|
|
): Promise<null | TbankInstrument> {
|
|
return this.getInstrument(token, InstrumentIdType.Uid, instrumentUid);
|
|
}
|
|
|
|
getDividends(
|
|
token: string,
|
|
instrumentId: string,
|
|
from: string,
|
|
to: string,
|
|
): Promise<TbankDividend[]> {
|
|
return this.call<{ dividends?: TbankDividend[] }>(
|
|
token,
|
|
this.instruments,
|
|
'getDividends',
|
|
{
|
|
instrumentId,
|
|
from: { seconds: String(utcSeconds(from)) },
|
|
to: { seconds: String(utcSeconds(to, true)) },
|
|
},
|
|
).then((response) => response.dividends ?? []);
|
|
}
|
|
|
|
findShares(token: string, query: string): Promise<TbankShare[]> {
|
|
return this.call<{ instruments?: TbankShare[] }>(
|
|
token,
|
|
this.instruments,
|
|
'findInstrument',
|
|
{
|
|
query,
|
|
instrumentKind: TbankInstrumentType.Share,
|
|
},
|
|
).then((response) => response.instruments ?? []);
|
|
}
|
|
|
|
async getAllShares(token: string): Promise<TbankShare[]> {
|
|
const response = await this.call<{ instruments?: TbankShare[] }>(
|
|
token,
|
|
this.instruments,
|
|
'shares',
|
|
{ instrumentStatus: TbankInstrumentStatus.All },
|
|
);
|
|
const { instruments } = response;
|
|
if (
|
|
!Array.isArray(instruments) ||
|
|
!instruments.length ||
|
|
instruments.some((item) => !item?.uid)
|
|
) {
|
|
this.logger.event('error', 'tbank.shares.catalog.unusable', {
|
|
code: TbankErrorCode.Unavailable,
|
|
operation: 'shares',
|
|
});
|
|
throw new ServiceUnavailableException({
|
|
code: TbankErrorCode.Unavailable,
|
|
});
|
|
}
|
|
this.logger.event('info', 'tbank.shares.catalog.fetched', {
|
|
asOf: new Date().toISOString(),
|
|
count: instruments.length,
|
|
operation: 'shares',
|
|
});
|
|
return instruments;
|
|
}
|
|
|
|
findBonds(token: string, query: string): Promise<TbankBond[]> {
|
|
return this.call<{ instruments?: TbankBond[] }>(
|
|
token,
|
|
this.sparseInstruments,
|
|
'findInstrument',
|
|
{
|
|
query,
|
|
instrumentKind: TbankInstrumentType.Bond,
|
|
},
|
|
).then((response) => response.instruments ?? []);
|
|
}
|
|
|
|
getBond(token: string, uid: string): Promise<null | TbankBond> {
|
|
return this.call<{ instrument?: TbankBond }>(
|
|
token,
|
|
this.sparseInstruments,
|
|
'bondBy',
|
|
{
|
|
idType: InstrumentIdType.Uid,
|
|
id: uid,
|
|
},
|
|
).then((response) => response.instrument ?? null);
|
|
}
|
|
|
|
getBondCoupons(
|
|
token: string,
|
|
uid: string,
|
|
from: string,
|
|
to: string,
|
|
): Promise<TbankCoupon[]> {
|
|
return this.call<{ events?: TbankCoupon[] }>(
|
|
token,
|
|
this.sparseInstruments,
|
|
'getBondCoupons',
|
|
{
|
|
instrumentId: uid,
|
|
from: { seconds: String(utcSeconds(from)) },
|
|
to: { seconds: String(utcSeconds(to, true)) },
|
|
},
|
|
).then((response) => response.events ?? []);
|
|
}
|
|
|
|
getShare(token: string, uid: string): Promise<null | TbankShare> {
|
|
return this.call<{ instrument?: TbankShare }>(
|
|
token,
|
|
this.instruments,
|
|
'shareBy',
|
|
{
|
|
idType: InstrumentIdType.Uid,
|
|
id: uid,
|
|
},
|
|
).then((response) => response.instrument ?? null);
|
|
}
|
|
|
|
getAssetDescription(token: string, assetUid: string): Promise<null | string> {
|
|
return this.call<{ asset?: { description?: string } }>(
|
|
token,
|
|
this.instruments,
|
|
'getAssetBy',
|
|
{
|
|
id: assetUid,
|
|
},
|
|
).then((response) => response.asset?.description?.trim() || null);
|
|
}
|
|
|
|
getFundamentals(
|
|
token: string,
|
|
assetUid: string,
|
|
): Promise<null | TbankFundamentals> {
|
|
return this.call<{ fundamentals?: TbankFundamentals[] }>(
|
|
token,
|
|
this.sparseInstruments,
|
|
'getAssetFundamentals',
|
|
{
|
|
assets: [assetUid],
|
|
},
|
|
).then(
|
|
(response) =>
|
|
response.fundamentals?.find((item) => item.assetUid === assetUid) ??
|
|
null,
|
|
);
|
|
}
|
|
|
|
getLastPrices(token: string, uids: string[]): Promise<TbankLastPrice[]> {
|
|
return this.call<{ lastPrices?: TbankLastPrice[] }>(
|
|
token,
|
|
this.marketData,
|
|
'getLastPrices',
|
|
{ instrumentId: uids },
|
|
).then((response) => response.lastPrices ?? []);
|
|
}
|
|
|
|
getTradingStatus(token: string, uid: string): Promise<null | string> {
|
|
return this.call<{ tradingStatus?: string }>(
|
|
token,
|
|
this.marketData,
|
|
'getTradingStatus',
|
|
{
|
|
instrumentId: uid,
|
|
},
|
|
).then((response) => response.tradingStatus || null);
|
|
}
|
|
|
|
getCandles(
|
|
token: string,
|
|
uid: string,
|
|
from: Date,
|
|
to: Date,
|
|
): Promise<TbankCandle[]> {
|
|
return this.call<{ candles?: TbankCandle[] }>(
|
|
token,
|
|
this.marketData,
|
|
'getCandles',
|
|
{
|
|
instrumentId: uid,
|
|
from: {
|
|
seconds: String(Math.floor(from.valueOf() / MILLISECONDS_PER_SECOND)),
|
|
},
|
|
to: {
|
|
seconds: String(Math.floor(to.valueOf() / MILLISECONDS_PER_SECOND)),
|
|
},
|
|
interval: TbankCandleInterval.Day,
|
|
limit: 2400,
|
|
},
|
|
).then((response) => response.candles ?? []);
|
|
}
|
|
|
|
private async getInstrument(
|
|
token: string,
|
|
idType: InstrumentIdType,
|
|
id: string,
|
|
): Promise<null | TbankInstrument> {
|
|
try {
|
|
const response = await this.call<{ instrument?: TbankInstrument }>(
|
|
token,
|
|
this.instruments,
|
|
'getInstrumentBy',
|
|
{ idType, id },
|
|
);
|
|
return response.instrument ?? null;
|
|
} catch (error) {
|
|
if (error instanceof NotFoundException) return null;
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
private async call<Response>(
|
|
token: string,
|
|
client: TbankClientService,
|
|
method: string,
|
|
request: object,
|
|
): Promise<Response> {
|
|
const metadataCall = method === 'getInstrumentBy';
|
|
for (let attempts = FIRST_ATTEMPT; attempts <= RPC_MAX_ATTEMPTS; attempts += ATTEMPT_INCREMENT) {
|
|
const startedAt = performance.now();
|
|
const queueFull = this.queue.size >= MAX_QUEUE_SIZE;
|
|
const diagnostics: RpcDiagnostics = {};
|
|
try {
|
|
if (queueFull) {
|
|
throw new ServiceUnavailableException({ code: TbankErrorCode.Unavailable });
|
|
}
|
|
const response = await this.queue.run(() => {
|
|
diagnostics.queueWaitMs = Math.round(performance.now() - startedAt);
|
|
return this.request<Response>(
|
|
token, client, method, request, diagnostics,
|
|
);
|
|
});
|
|
this.logger.event('debug', 'tbank.rpc.completed', {
|
|
operation: method,
|
|
...diagnostics,
|
|
attempts,
|
|
durationMs: Math.round(performance.now() - startedAt),
|
|
});
|
|
return response;
|
|
} catch (error) {
|
|
const grpcCode = (error as ServiceError).code;
|
|
const failureContext = queueFull
|
|
? { code: TbankErrorCode.Unavailable }
|
|
: { grpcCode };
|
|
const retryable = !queueFull && [
|
|
status.DEADLINE_EXCEEDED,
|
|
status.RESOURCE_EXHAUSTED,
|
|
status.UNAVAILABLE,
|
|
].includes(grpcCode);
|
|
if (retryable && attempts < RPC_MAX_ATTEMPTS) {
|
|
this.logger.event('warn', 'tbank.rpc.retry', {
|
|
operation: method,
|
|
...diagnostics,
|
|
...failureContext,
|
|
attempts,
|
|
durationMs: Math.round(performance.now() - startedAt),
|
|
});
|
|
await new Promise((resolve) => setTimeout(resolve, RPC_RETRY_DELAY_MS));
|
|
continue;
|
|
}
|
|
this.logger.event('error', 'tbank.rpc.failed', {
|
|
operation: method,
|
|
...diagnostics,
|
|
...failureContext,
|
|
attempts: queueFull ? NO_RPC_ATTEMPTS : attempts,
|
|
durationMs: Math.round(performance.now() - startedAt),
|
|
});
|
|
if (
|
|
[status.PERMISSION_DENIED, status.UNAUTHENTICATED].includes(grpcCode)
|
|
) {
|
|
// eslint-disable-next-line preserve-caught-error -- Discard raw upstream authentication errors at the integration boundary.
|
|
throw new Error(TbankErrorCode.InvalidToken);
|
|
}
|
|
if (
|
|
grpcCode === status.NOT_FOUND &&
|
|
(method === 'shareBy' || metadataCall)
|
|
) {
|
|
throw new NotFoundException();
|
|
}
|
|
throw new ServiceUnavailableException({ code: TbankErrorCode.Unavailable });
|
|
}
|
|
}
|
|
throw new ServiceUnavailableException({ code: TbankErrorCode.Unavailable });
|
|
}
|
|
|
|
private request<Response>(
|
|
token: string,
|
|
client: TbankClientService,
|
|
method: string,
|
|
request: object,
|
|
diagnostics?: RpcDiagnostics,
|
|
): Promise<Response> {
|
|
return new Promise((resolve, reject) => {
|
|
const metadata = new Metadata();
|
|
metadata.set('Authorization', `Bearer ${token}`);
|
|
// A channel state is a snapshot, not proof that this particular call reached the server.
|
|
const channel = diagnostics && client.getChannel();
|
|
if (diagnostics) diagnostics.headersReceived = false;
|
|
// eslint-disable-next-line security/detect-object-injection -- Service and method names are internal RPC selections, not request input.
|
|
const call = client[method](
|
|
request as never,
|
|
metadata,
|
|
{ deadline: new Date(Date.now() + requestDeadlineMs) },
|
|
(error, response) => {
|
|
if (diagnostics && channel) {
|
|
diagnostics.channelState = connectivityState[
|
|
channel.getConnectivityState(false)
|
|
];
|
|
}
|
|
if (error) reject(error);
|
|
else resolve(response);
|
|
},
|
|
);
|
|
if (diagnostics) {
|
|
call.on('metadata', (headers: Metadata) => {
|
|
diagnostics.headersReceived = true;
|
|
const trackingId = headers.get('x-tracking-id')[0];
|
|
// Only a UUID may enter logs; never forward arbitrary upstream metadata.
|
|
if (
|
|
typeof trackingId === 'string' &&
|
|
/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(trackingId)
|
|
) {
|
|
diagnostics.trackingId = trackingId;
|
|
}
|
|
});
|
|
}
|
|
});
|
|
}
|
|
}
|
|
|
|
// Calendar owners supply inclusive UTC days; RPC timestamps use [from, next(to)).
|
|
function utcSeconds(date: string, nextDay = false): number {
|
|
const instant = new Date(`${date}T00:00:00Z`);
|
|
if (nextDay) instant.setUTCDate(instant.getUTCDate() + NEXT_DAY_OFFSET);
|
|
return Math.floor(instant.valueOf() / MILLISECONDS_PER_SECOND);
|
|
}
|