ksv741
33592c1d5f
Use owner enums for error and T-Bank protocol values with ESLint regression checks. Preserve open wire fallbacks and keep protocol imports independent of DI and Prisma. Verified npm ci, verify, 226 unit tests, 262 isolated e2e tests and unchanged OpenAPI.
585 lines
17 KiB
TypeScript
585 lines
17 KiB
TypeScript
import {
|
|
ChannelCredentials,
|
|
Client,
|
|
ClientUnaryCall,
|
|
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,
|
|
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 METADATA_MAX_ATTEMPTS = 3;
|
|
const RPC_MAX_ATTEMPTS = 2;
|
|
const MAX_QUEUE_SIZE = 50;
|
|
const METADATA_RETRY_DELAY_MS = 2_000;
|
|
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>>;
|
|
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 ?? []);
|
|
}
|
|
|
|
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 ?? []);
|
|
}
|
|
|
|
getLastPrice(token: string, uid: string): Promise<null | TbankLastPrice> {
|
|
return this.getLastPrices(token, [uid]).then(
|
|
(prices) => prices.find((item) => item.instrumentUid === uid) ?? null,
|
|
);
|
|
}
|
|
|
|
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';
|
|
const maxAttempts = metadataCall ? METADATA_MAX_ATTEMPTS : RPC_MAX_ATTEMPTS;
|
|
for (let attempts = FIRST_ATTEMPT; attempts <= maxAttempts; attempts += ATTEMPT_INCREMENT) {
|
|
const startedAt = performance.now();
|
|
const queueFull = this.queue.size >= MAX_QUEUE_SIZE;
|
|
try {
|
|
if (queueFull) {
|
|
throw new ServiceUnavailableException({ code: TbankErrorCode.Unavailable });
|
|
}
|
|
const response = await this.queue.run(() =>
|
|
this.request<Response>(token, client, method, request),
|
|
);
|
|
this.logger.event('debug', 'tbank.rpc.completed', {
|
|
operation: method,
|
|
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
|
|
? metadataCall
|
|
: [
|
|
status.DEADLINE_EXCEEDED,
|
|
status.RESOURCE_EXHAUSTED,
|
|
status.UNAVAILABLE,
|
|
].includes(grpcCode);
|
|
if (retryable && attempts < maxAttempts) {
|
|
this.logger.event('warn', 'tbank.rpc.retry', {
|
|
operation: method,
|
|
...failureContext,
|
|
attempts,
|
|
durationMs: Math.round(performance.now() - startedAt),
|
|
});
|
|
await new Promise((resolve) =>
|
|
setTimeout(
|
|
resolve,
|
|
metadataCall
|
|
? attempts * METADATA_RETRY_DELAY_MS
|
|
: RPC_RETRY_DELAY_MS,
|
|
),
|
|
);
|
|
continue;
|
|
}
|
|
this.logger.event('error', 'tbank.rpc.failed', {
|
|
operation: method,
|
|
...failureContext,
|
|
attempts: queueFull && !metadataCall ? 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,
|
|
): Promise<Response> {
|
|
return new Promise((resolve, reject) => {
|
|
const metadata = new Metadata();
|
|
metadata.set('Authorization', `Bearer ${token}`);
|
|
// eslint-disable-next-line security/detect-object-injection -- Service and method names are internal RPC selections, not request input.
|
|
client[method](
|
|
request as never,
|
|
metadata,
|
|
{ deadline: new Date(Date.now() + requestDeadlineMs) },
|
|
(error, response) => (error ? reject(error) : resolve(response)),
|
|
);
|
|
});
|
|
}
|
|
}
|
|
|
|
// 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);
|
|
}
|