Compare commits

...
1 Commits
Author SHA1 Message Date
arham ba5fff0168 add: ping service to sdk 2026-06-01 10:28:06 +00:00
5 changed files with 42 additions and 1 deletions

No files matched your search

@@ -8,6 +8,7 @@ using bts.api.core;
using bts.api.orderman;
using bts.api.position;
using bts.co2;
using bts.metrics;
using bts.md;
using Microsoft.Extensions.Configuration;
@@ -86,6 +87,7 @@ public static class SparkApiServices {
services.AddSparkdService<OrderService>();
services.AddSingleton<IOrdersClient, OrdersClient>();
}
services.AddSparkdService<ShellLatencyService>();
if (configurationSection.GetValue("EnablePosition", true)) {
services.TryAddSingleton<ISecurityResolver>(ActivatorUtilities.GetServiceOrCreateInstance<PassiveSecurityResolver>);
services.AddSparkdService<PositionService>()
@@ -176,6 +178,7 @@ public static class SparkApiServices {
services.AddSparkdService<OrderService>();
services.AddSingleton<IOrdersClient, OrdersClient>();
}
services.AddSparkdService<ShellLatencyService>();
if (isPositionsEnabled) {
services.AddSparkdService<PositionService>();
}
@@ -3,6 +3,7 @@ namespace Bts.Spark.Sdk;
using bts.api;
using bts.api.core;
using bts.api.orderman;
using bts.metrics;
using global::Spark.Api.Admin;
using global::Spark.Api.MarketData;
@@ -28,6 +29,7 @@ public class AuthenticatedSimpleSparkClient : ISimpleSparkClient, IHostedService
readonly IOrdersClient ordersClient;
readonly IAdminClient adminClient;
readonly ISecurityResolver securityResolver;
readonly ShellLatencyService shellLatencyService;
readonly ISparkLoginService loginService;
readonly SimpleSparkClientConfig config;
readonly ILogger<AuthenticatedSimpleSparkClient> logger;
@@ -39,6 +41,7 @@ public class AuthenticatedSimpleSparkClient : ISimpleSparkClient, IHostedService
IOrdersClient ordersClient,
IAdminClient adminClient,
ISecurityResolver securityResolver,
ShellLatencyService shellLatencyService,
ISparkLoginService loginService,
IOptions<SimpleSparkClientConfig> config,
ILogger<AuthenticatedSimpleSparkClient> logger) {
@@ -46,6 +49,7 @@ public class AuthenticatedSimpleSparkClient : ISimpleSparkClient, IHostedService
this.ordersClient = ordersClient;
this.adminClient = adminClient;
this.securityResolver = securityResolver;
this.shellLatencyService = shellLatencyService;
this.loginService = loginService;
this.config = config.Value;
this.logger = logger;
@@ -61,7 +65,7 @@ public class AuthenticatedSimpleSparkClient : ISimpleSparkClient, IHostedService
}
cancellationToken.ThrowIfCancellationRequested();
var user = loginRes.user.Value;
simpleClient = new(user.name, user.id, mdClient, ordersClient, adminClient, securityResolver, config.OrderPoolCapacity, config.OrderPoolReplenishInterval);
simpleClient = new(user.name, user.id, mdClient, ordersClient, shellLatencyService, adminClient, securityResolver, config.OrderPoolCapacity, config.OrderPoolReplenishInterval);
simpleClientSubject.OnNext(simpleClient);
logger.LogInformation("SimpleSparkClient started");
}
@@ -97,6 +101,9 @@ public class AuthenticatedSimpleSparkClient : ISimpleSparkClient, IHostedService
public IObservable<TableUpdate<PriceTicker, uint>> PriceTicker(SecurityCriteria? securityCriteria) =>
simpleClient?.PriceTicker(securityCriteria) ?? simpleClientSubject.Take(1).Select(c => c.PriceTicker(securityCriteria)).FlattenSeries();
public Task<bool> PingAsync() =>
simpleClient?.PingAsync() ?? pingBeforeStartAsync();
public Task<Response> CreateOrder(OrderDefinition orderDef, double quantity, double price) =>
orderOp(cli => cli.CreateOrder(orderDef, quantity, price));
@@ -134,4 +141,15 @@ public class AuthenticatedSimpleSparkClient : ISimpleSparkClient, IHostedService
async Task<T> LateImpl() =>
await op(await simpleClientSubject.FirstAsync());
}
async Task<bool> pingBeforeStartAsync() {
var request = new BufferMessage() {
Sequence = (ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
Data = "spark-sdk-ping"
};
var result = await shellLatencyService.ping_async(request);
return result.success &&
result.value.Sequence == request.Sequence &&
result.value.Data == request.Data;
}
}
@@ -48,6 +48,9 @@ public class ExecutionFacadeSimpleSparkClient : ISimpleSparkClient {
public IObservable<TableUpdate<PriceTicker, uint>> PriceTicker(SecurityCriteria? securityCriteria) =>
sparkClient.PriceTicker(securityCriteria);
public Task<bool> PingAsync() =>
sparkClient.PingAsync();
public Task<Response> CreateOrder(OrderDefinition orderDef, double quantity, double price) {
logger.LogInformation("CreateOrder {account} {tag} {securityId} {side} {quantity} @ {price}", orderDef.Account, orderDef.Tag, orderDef.SecurityKey, orderDef.Side, quantity, price);
return Task.FromResult(facadeResponse());
@@ -12,6 +12,8 @@ using mbts.spark;
using System;
public interface ISimpleSparkClient {
Task<bool> PingAsync();
// Market Data
IObservable<TableUpdate<SecurityImmutable, uint>> Securities(SecurityCriteria? securityCriteria);
IObservable<TableUpdate<BookRecord, uint>> Books(SecurityCriteria? securityCriteria);
+15
View File
@@ -16,12 +16,14 @@ using global::Spark.Core.Disposables;
using global::Spark.Core.Reactive;
using mbts.spark;
using bts.metrics;
using System.Reactive.Linq;
public class SimpleSparkClient : ISimpleSparkClient, IDisposable {
readonly IMarketDataClient mdClient;
readonly IOrdersClient ordersClient;
readonly ShellLatencyService shellLatencyService;
readonly OrderPool limitOrderPool;
readonly OrderPool marketOrderPool;
readonly TableUpdateDemuxer<OrderRecordTupleWithRoot, uint, uint> orderSecurityDmx;
@@ -46,6 +48,7 @@ public class SimpleSparkClient : ISimpleSparkClient, IDisposable {
uint userId,
IMarketDataClient mdClient,
IOrdersClient ordersClient,
ShellLatencyService shellLatencyService,
IAdminClient adminClient,
ISecurityResolver securityResolver,
int orderPoolCapacity,
@@ -53,6 +56,7 @@ public class SimpleSparkClient : ISimpleSparkClient, IDisposable {
) {
this.mdClient = mdClient;
this.ordersClient = ordersClient;
this.shellLatencyService = shellLatencyService;
limitOrderPool = new(orderPoolCapacity, () => new() { User = userName, UserId = userId, TimeInForce = TimeInForce.Gtc, CustomDef = new() { Limit = new() } });
marketOrderPool = new(orderPoolCapacity, () => new() { User = userName, UserId = userId, TimeInForce = TimeInForce.Gtc, CustomDef = new() { Market = new() } });
orderSecurityDmx = new TableUpdateDemuxer<OrderRecordTupleWithRoot, uint, uint>(o => o.order.order.Key, o => o.security_id);
@@ -140,6 +144,17 @@ public class SimpleSparkClient : ISimpleSparkClient, IDisposable {
}
}
public async Task<bool> PingAsync() {
var request = new BufferMessage() {
Sequence = (ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
Data = "spark-sdk-ping"
};
var result = await shellLatencyService.ping_async(request);
return result.success &&
result.value.Sequence == request.Sequence &&
result.value.Data == request.Data;
}
public IObservable<TableUpdate<BookRecord, uint>> Books(SecurityCriteria? securityCriteria) {
var booksSource = this.booksSource;
var secFilter = securityCriteria?.ToPredicate();