feat: add portfolio
This commit is contained in:
+131
-76
@@ -18,31 +18,38 @@ import { existsSync, readFileSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
import { Environment } from '../config/env.validation';
|
||||
|
||||
type GetAccounts = (
|
||||
request: { status: string },
|
||||
const requestDeadlineMs = 15_000;
|
||||
|
||||
type Unary<Request, Response> = (
|
||||
request: Request,
|
||||
metadata: Metadata,
|
||||
options: { deadline: Date },
|
||||
callback: (
|
||||
error: ServiceError | null,
|
||||
response: { accounts?: TbankAccount[] },
|
||||
) => void,
|
||||
callback: (error: ServiceError | null, response: Response) => void,
|
||||
) => ClientUnaryCall;
|
||||
type UsersClient = Client & { getAccounts: GetAccounts };
|
||||
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;
|
||||
quantity?: TbankDecimal;
|
||||
currentPrice?: TbankMoney;
|
||||
averagePositionPrice?: TbankMoney;
|
||||
}[];
|
||||
};
|
||||
export type TbankPositions = { money?: TbankMoney[] };
|
||||
export type TbankInstrument = { ticker?: string; name?: string };
|
||||
type TbankNamespace = {
|
||||
tinkoff: {
|
||||
public: {
|
||||
invest: {
|
||||
api: {
|
||||
contract: { v1: { UsersService: unknown } };
|
||||
};
|
||||
};
|
||||
};
|
||||
public: { invest: { api: { contract: { v1: Record<string, unknown> } } } };
|
||||
};
|
||||
};
|
||||
|
||||
@@ -92,44 +99,52 @@ class RateLimiter {
|
||||
export class TbankClient {
|
||||
private readonly logger = new Logger(TbankClient.name);
|
||||
private readonly queue: RateLimiter;
|
||||
private readonly client: UsersClient;
|
||||
private readonly users: TbankClientService;
|
||||
private readonly operations: TbankClientService;
|
||||
private readonly instruments: TbankClientService;
|
||||
|
||||
constructor(config: ConfigService<Environment, true>) {
|
||||
const sourceProtoRoot = join(process.cwd(), 'src/tbank/proto/contracts');
|
||||
const builtProtoRoot = join(process.cwd(), 'dist/tbank/proto/contracts');
|
||||
const protoRoot = existsSync(sourceProtoRoot)
|
||||
? sourceProtoRoot
|
||||
: builtProtoRoot;
|
||||
const definition = loadSync(join(protoRoot, 'users.proto'), {
|
||||
includeDirs: [protoRoot],
|
||||
keepCase: false,
|
||||
longs: String,
|
||||
enums: String,
|
||||
defaults: true,
|
||||
oneofs: true,
|
||||
});
|
||||
const packageDefinition = loadPackageDefinition(
|
||||
definition,
|
||||
) as unknown as TbankNamespace;
|
||||
const Service = packageDefinition.tinkoff.public.invest.api.contract.v1
|
||||
.UsersService as new (
|
||||
address: string,
|
||||
credentials: ChannelCredentials,
|
||||
) => UsersClient;
|
||||
: join(process.cwd(), 'dist/tbank/proto/contracts');
|
||||
const definition = loadSync(
|
||||
['users.proto', 'operations.proto', 'instruments.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();
|
||||
this.client = new Service(
|
||||
config.getOrThrow<string>('TBANK_GRPC_ENDPOINT'),
|
||||
credentials,
|
||||
);
|
||||
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.queue = new RateLimiter(
|
||||
config.getOrThrow<number>('TBANK_RATE_LIMIT_PER_SECOND'),
|
||||
);
|
||||
this.logger.log({
|
||||
event: 'client_configured',
|
||||
endpoint: config.getOrThrow<string>('TBANK_GRPC_ENDPOINT'),
|
||||
endpoint,
|
||||
rateLimitPerSecond: config.getOrThrow<number>(
|
||||
'TBANK_RATE_LIMIT_PER_SECOND',
|
||||
),
|
||||
@@ -141,64 +156,104 @@ export class TbankClient {
|
||||
await this.getOpenAccounts(token);
|
||||
}
|
||||
|
||||
async getOpenAccounts(token: string): Promise<TbankAccount[]> {
|
||||
return this.call(token, true);
|
||||
getOpenAccounts(token: string): Promise<TbankAccount[]> {
|
||||
return this.call(token, this.users, 'getAccounts', {
|
||||
status: 'ACCOUNT_STATUS_OPEN',
|
||||
}).then(
|
||||
(response: { accounts?: TbankAccount[] }) => response.accounts ?? [],
|
||||
);
|
||||
}
|
||||
|
||||
private async call(token: string, retry: boolean): Promise<TbankAccount[]> {
|
||||
if (this.queue.size >= 50) {
|
||||
this.logger.warn({ event: 'queue_full', waiting: this.queue.size });
|
||||
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 });
|
||||
}
|
||||
|
||||
getInstrumentByFigi(
|
||||
token: string,
|
||||
figi: string,
|
||||
): Promise<TbankInstrument | null> {
|
||||
return this.call(token, this.instruments, 'getInstrumentBy', {
|
||||
idType: 'INSTRUMENT_ID_TYPE_FIGI',
|
||||
id: figi,
|
||||
}).then(
|
||||
(response: { instrument?: TbankInstrument }) =>
|
||||
response.instrument ?? null,
|
||||
);
|
||||
}
|
||||
|
||||
getInstrumentByUid(
|
||||
token: string,
|
||||
instrumentUid: string,
|
||||
): Promise<TbankInstrument | null> {
|
||||
return this.call(token, this.instruments, 'getInstrumentBy', {
|
||||
idType: 'INSTRUMENT_ID_TYPE_UID',
|
||||
id: instrumentUid,
|
||||
}).then(
|
||||
(response: { instrument?: TbankInstrument }) =>
|
||||
response.instrument ?? null,
|
||||
);
|
||||
}
|
||||
|
||||
private async call<Response>(
|
||||
token: string,
|
||||
client: TbankClientService,
|
||||
method: string,
|
||||
request: object,
|
||||
retry = true,
|
||||
): Promise<Response> {
|
||||
if (this.queue.size >= 50)
|
||||
throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' });
|
||||
}
|
||||
try {
|
||||
this.logger.debug({ event: 'get_accounts_started', retry: !retry });
|
||||
const accounts = await this.queue.run(() => this.request(token));
|
||||
this.logger.debug({ event: 'get_accounts_succeeded', retry: !retry });
|
||||
return accounts;
|
||||
return await this.queue.run(() =>
|
||||
this.request<Response>(token, client, method, request),
|
||||
);
|
||||
} catch (error) {
|
||||
const code = (error as ServiceError).code;
|
||||
const retryable = [
|
||||
status.UNAVAILABLE,
|
||||
status.DEADLINE_EXCEEDED,
|
||||
status.RESOURCE_EXHAUSTED,
|
||||
].includes(code);
|
||||
this.logger.warn({
|
||||
event: 'get_accounts_failed',
|
||||
event: 'tbank_rpc_failed',
|
||||
method,
|
||||
grpcCode: code,
|
||||
retry: !retry,
|
||||
retryable: [
|
||||
status.UNAVAILABLE,
|
||||
status.DEADLINE_EXCEEDED,
|
||||
status.RESOURCE_EXHAUSTED,
|
||||
].includes(code),
|
||||
retryable,
|
||||
...(method === 'getInstrumentBy'
|
||||
? {
|
||||
instrumentIdType: (request as { idType?: string }).idType,
|
||||
instrumentId: (request as { id?: string }).id,
|
||||
}
|
||||
: {}),
|
||||
});
|
||||
if (
|
||||
retry &&
|
||||
[
|
||||
status.UNAVAILABLE,
|
||||
status.DEADLINE_EXCEEDED,
|
||||
status.RESOURCE_EXHAUSTED,
|
||||
].includes(code)
|
||||
) {
|
||||
this.logger.log({
|
||||
event: 'get_accounts_retry_scheduled',
|
||||
delayMs: 1000,
|
||||
});
|
||||
if (retry && retryable) {
|
||||
await new Promise((resolve) => setTimeout(resolve, 1000));
|
||||
return this.call(token, false);
|
||||
return this.call(token, client, method, request, false);
|
||||
}
|
||||
if ([status.UNAUTHENTICATED, status.PERMISSION_DENIED].includes(code)) {
|
||||
if ([status.UNAUTHENTICATED, status.PERMISSION_DENIED].includes(code))
|
||||
throw new Error('INVALID_TBANK_TOKEN');
|
||||
}
|
||||
throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' });
|
||||
}
|
||||
}
|
||||
|
||||
private request(token: string): Promise<TbankAccount[]> {
|
||||
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}`);
|
||||
this.client.getAccounts(
|
||||
{ status: 'ACCOUNT_STATUS_OPEN' },
|
||||
client[method](
|
||||
request as never,
|
||||
metadata,
|
||||
{ deadline: new Date(Date.now() + 10_000) },
|
||||
(error, response) =>
|
||||
error ? reject(error) : resolve(response.accounts ?? []),
|
||||
{ deadline: new Date(Date.now() + requestDeadlineMs) },
|
||||
(error, response) => (error ? reject(error) : resolve(response)),
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user