ksv741
d6e6c4ace5
Backend security / security (push) Successful in 8s
Backend verify / verify (push) Failing after 10m45s
Conventional commits / commits (pull_request) Successful in 29s
Backend security / security (pull_request) Successful in 8s
Backend verify / verify (pull_request) Failing after 10m37s
Store account membership, portfolio, operations, forecasts and market sources in SQLite. Coalesce scoped refreshes, fence connection revisions, preserve complete financial snapshots and stabilize operation period state. BREAKING CHANGE: investment reads return data/meta resource envelopes and locally persisted snapshots; consumers must use the matching OpenAPI contract and explicit synchronization endpoints.
558 lines
16 KiB
TypeScript
558 lines
16 KiB
TypeScript
import { MILLISECONDS_PER_SECOND } from '../../common/time';
|
|
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 { StructuredLogger } from '../../common/logging/structured-logger.service';
|
|
import { Environment } from '../../config/env.validation';
|
|
|
|
import type {
|
|
TbankBond,
|
|
TbankCandle,
|
|
TbankCoupon,
|
|
TbankFundamentals,
|
|
TbankLastPrice,
|
|
TbankShare,
|
|
} from './tbank-market.types';
|
|
|
|
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: ServiceError | null, response: Response) => void,
|
|
) => ClientUnaryCall;
|
|
type TbankClientService = Client & Record<string, Unary<never, never>>;
|
|
export type TbankDecimal = { units?: string | number; nano?: string | number };
|
|
export type TbankMoney = TbankDecimal & { currency?: string };
|
|
export type TbankAccount = {
|
|
id: string;
|
|
type: string;
|
|
name: string;
|
|
openedDate?: { seconds?: string | number };
|
|
};
|
|
export type TbankPortfolio = {
|
|
positions?: {
|
|
figi?: string;
|
|
instrumentUid?: string;
|
|
instrumentType?: string;
|
|
ticker?: string;
|
|
quantity?: TbankDecimal;
|
|
currentPrice?: TbankMoney;
|
|
averagePositionPrice?: TbankMoney;
|
|
}[];
|
|
};
|
|
export type TbankPositions = { money?: TbankMoney[] };
|
|
export type TbankInstrument = {
|
|
uid?: string;
|
|
figi?: string;
|
|
instrumentType?: string;
|
|
ticker?: string;
|
|
name?: string;
|
|
isin?: string;
|
|
brand?: { logoName?: string | null };
|
|
};
|
|
export type TbankDividend = {
|
|
dividendNet?: TbankMoney;
|
|
recordDate?: TbankTimestamp;
|
|
paymentDate?: TbankTimestamp;
|
|
lastBuyDate?: TbankTimestamp;
|
|
dividendType?: string;
|
|
};
|
|
export type TbankTimestamp = { seconds?: string | number; nanos?: number };
|
|
export type TbankOperationItem = {
|
|
state?: string;
|
|
cursor?: string;
|
|
id?: string;
|
|
name?: string;
|
|
date?: TbankTimestamp;
|
|
type?: string;
|
|
description?: string;
|
|
figi?: string;
|
|
instrumentUid?: string;
|
|
instrumentType?: string;
|
|
ticker?: string;
|
|
payment?: TbankMoney;
|
|
price?: TbankMoney;
|
|
commission?: TbankMoney;
|
|
accruedInt?: TbankMoney;
|
|
quantity?: string | number;
|
|
quantityDone?: string | number;
|
|
};
|
|
export type TbankOperationsByCursorRequest = {
|
|
accountId: string;
|
|
instrumentId?: string;
|
|
from?: TbankTimestamp;
|
|
to?: TbankTimestamp;
|
|
cursor?: string;
|
|
limit?: number;
|
|
state: 'OPERATION_STATE_EXECUTED';
|
|
};
|
|
export type TbankOperationsByCursorResponse = {
|
|
hasNext?: boolean;
|
|
nextCursor?: string;
|
|
items?: TbankOperationItem[];
|
|
};
|
|
type TbankNamespace = {
|
|
tinkoff: {
|
|
public: { invest: { api: { contract: { v1: Record<string, unknown> } } } };
|
|
};
|
|
};
|
|
|
|
class RateLimiter {
|
|
private readonly waiting: (() => void)[] = [];
|
|
private starts = 0;
|
|
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 = 0;
|
|
}
|
|
while (this.starts < this.perSecond && this.waiting.length) {
|
|
this.starts += 1;
|
|
this.waiting.shift()?.();
|
|
}
|
|
if (this.waiting.length && this.starts === this.perSecond) {
|
|
setTimeout(
|
|
() => this.drain(),
|
|
Math.max(1, 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
|
|
? ChannelCredentials.createSsl(readFileSync(caPath))
|
|
: ChannelCredentials.createSsl();
|
|
const endpoint = config.getOrThrow<string>('TBANK_GRPC_ENDPOINT');
|
|
const create = (name: string) =>
|
|
new (
|
|
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: 'ACCOUNT_STATUS_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<TbankInstrument | null> {
|
|
return this.getInstrument(token, 'INSTRUMENT_ID_TYPE_FIGI', figi);
|
|
}
|
|
|
|
getInstrumentByUid(
|
|
token: string,
|
|
instrumentUid: string,
|
|
): Promise<TbankInstrument | null> {
|
|
return this.getInstrument(token, 'INSTRUMENT_ID_TYPE_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)) },
|
|
},
|
|
).then((response) => response.dividends ?? []);
|
|
}
|
|
|
|
findShares(token: string, query: string): Promise<TbankShare[]> {
|
|
return this.call<{ instruments?: TbankShare[] }>(
|
|
token,
|
|
this.instruments,
|
|
'findInstrument',
|
|
{
|
|
query,
|
|
instrumentKind: 'INSTRUMENT_TYPE_SHARE',
|
|
},
|
|
).then((response) => response.instruments ?? []);
|
|
}
|
|
|
|
findBonds(token: string, query: string): Promise<TbankBond[]> {
|
|
return this.call<{ instruments?: TbankBond[] }>(
|
|
token,
|
|
this.sparseInstruments,
|
|
'findInstrument',
|
|
{
|
|
query,
|
|
instrumentKind: 'INSTRUMENT_TYPE_BOND',
|
|
},
|
|
).then((response) => response.instruments ?? []);
|
|
}
|
|
getBond(token: string, uid: string): Promise<TbankBond | null> {
|
|
return this.call<{ instrument?: TbankBond }>(
|
|
token,
|
|
this.sparseInstruments,
|
|
'bondBy',
|
|
{
|
|
idType: 'INSTRUMENT_ID_TYPE_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)) },
|
|
},
|
|
).then((response) => response.events ?? []);
|
|
}
|
|
|
|
getShare(token: string, uid: string): Promise<TbankShare | null> {
|
|
return this.call<{ instrument?: TbankShare }>(
|
|
token,
|
|
this.instruments,
|
|
'shareBy',
|
|
{
|
|
idType: 'INSTRUMENT_ID_TYPE_UID',
|
|
id: uid,
|
|
},
|
|
).then((response) => response.instrument ?? null);
|
|
}
|
|
|
|
getAssetDescription(token: string, assetUid: string): Promise<string | null> {
|
|
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<TbankFundamentals | null> {
|
|
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<TbankLastPrice | null> {
|
|
return this.getLastPrices(token, [uid]).then(
|
|
(prices) => prices.find((item) => item.instrumentUid === uid) ?? null,
|
|
);
|
|
}
|
|
|
|
getTradingStatus(token: string, uid: string): Promise<string | null> {
|
|
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: 'CANDLE_INTERVAL_DAY',
|
|
limit: 2400,
|
|
},
|
|
).then((response) => response.candles ?? []);
|
|
}
|
|
|
|
private async getInstrument(
|
|
token: string,
|
|
idType: 'INSTRUMENT_ID_TYPE_UID' | 'INSTRUMENT_ID_TYPE_FIGI',
|
|
id: string,
|
|
): Promise<TbankInstrument | null> {
|
|
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 = 1; attempts <= maxAttempts; attempts += 1) {
|
|
const startedAt = performance.now();
|
|
const queueFull = this.queue.size >= MAX_QUEUE_SIZE;
|
|
try {
|
|
if (queueFull)
|
|
throw new ServiceUnavailableException({ code: 'TBANK_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: 'TBANK_UNAVAILABLE' }
|
|
: { grpcCode };
|
|
const retryable = queueFull
|
|
? metadataCall
|
|
: [
|
|
status.UNAVAILABLE,
|
|
status.DEADLINE_EXCEEDED,
|
|
status.RESOURCE_EXHAUSTED,
|
|
].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 ? 0 : attempts,
|
|
durationMs: Math.round(performance.now() - startedAt),
|
|
});
|
|
if (
|
|
[status.UNAUTHENTICATED, status.PERMISSION_DENIED].includes(grpcCode)
|
|
)
|
|
// eslint-disable-next-line preserve-caught-error -- Discard raw upstream authentication errors at the integration boundary.
|
|
throw new Error('INVALID_TBANK_TOKEN');
|
|
if (
|
|
grpcCode === status.NOT_FOUND &&
|
|
(method === 'shareBy' || metadataCall)
|
|
)
|
|
throw new NotFoundException();
|
|
throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' });
|
|
}
|
|
}
|
|
throw new ServiceUnavailableException({ code: 'TBANK_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}`);
|
|
client[method](
|
|
request as never,
|
|
metadata,
|
|
{ deadline: new Date(Date.now() + requestDeadlineMs) },
|
|
(error, response) => (error ? reject(error) : resolve(response)),
|
|
);
|
|
});
|
|
}
|
|
}
|
|
|
|
function utcSeconds(date: string): number {
|
|
return Math.floor(Date.parse(`${date}T00:00:00Z`) / MILLISECONDS_PER_SECOND);
|
|
}
|