toki
bdbfe8228e
- Add parser_map for market data refresh configuration propagation (cli/worker/api) - Implement refresh status model with SQLite persistence (status_store.go) - Add scheduler refresh status headless output to CLI operator - Extend backfill scheduler with status tracking (start/complete/fail) - Add socket events for scheduler refresh status (start/complete/fail) - Update proto definitions for refresh status - Add generated PB files for client
317 lines
12 KiB
Go
317 lines
12 KiB
Go
package socket
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"time"
|
|
|
|
protoSocket "git.toki-labs.com/toki/proto-socket/go"
|
|
|
|
altv1 "git.toki-labs.com/toki/alt/packages/contracts/gen/go/alt/v1"
|
|
"git.toki-labs.com/toki/alt/services/api/internal/workerclient"
|
|
)
|
|
|
|
// importDateLayout matches the YYYYMMDD date fields the import command carries.
|
|
// The API only validates the shape so a malformed date is rejected before a
|
|
// worker connection is opened.
|
|
const importDateLayout = "20060102"
|
|
|
|
const (
|
|
marketErrorInvalidRequest = "invalid_request"
|
|
marketErrorUnavailable = "unavailable"
|
|
marketErrorTimeout = "timeout"
|
|
marketErrorInternal = "internal"
|
|
)
|
|
|
|
// marketHandlers returns the API session handlers for market data reads. The
|
|
// API validates query shape and forwards the request unchanged to the worker.
|
|
func marketHandlers(worker workerclient.WorkerClient) []sessionHandler {
|
|
return []sessionHandler{
|
|
{
|
|
requestType: protoSocket.TypeNameOf(&altv1.ListInstrumentsRequest{}),
|
|
register: func(client *protoSocket.WsClient) {
|
|
protoSocket.AddRequestListenerTyped[*altv1.ListInstrumentsRequest, *altv1.ListInstrumentsResponse](&client.Communicator, func(req *altv1.ListInstrumentsRequest) (*altv1.ListInstrumentsResponse, error) {
|
|
return handleListInstruments(worker, req)
|
|
})
|
|
},
|
|
},
|
|
{
|
|
requestType: protoSocket.TypeNameOf(&altv1.ListBarsRequest{}),
|
|
register: func(client *protoSocket.WsClient) {
|
|
protoSocket.AddRequestListenerTyped[*altv1.ListBarsRequest, *altv1.ListBarsResponse](&client.Communicator, func(req *altv1.ListBarsRequest) (*altv1.ListBarsResponse, error) {
|
|
return handleListBars(worker, req)
|
|
})
|
|
},
|
|
},
|
|
{
|
|
requestType: protoSocket.TypeNameOf(&altv1.ImportDailyBarsRequest{}),
|
|
register: func(client *protoSocket.WsClient) {
|
|
protoSocket.AddRequestListenerTyped[*altv1.ImportDailyBarsRequest, *altv1.ImportDailyBarsResponse](&client.Communicator, func(req *altv1.ImportDailyBarsRequest) (*altv1.ImportDailyBarsResponse, error) {
|
|
return handleImportDailyBars(worker, req)
|
|
})
|
|
},
|
|
},
|
|
{
|
|
requestType: protoSocket.TypeNameOf(&altv1.AggregateMonthlyBarsRequest{}),
|
|
register: func(client *protoSocket.WsClient) {
|
|
protoSocket.AddRequestListenerTyped[*altv1.AggregateMonthlyBarsRequest, *altv1.AggregateMonthlyBarsResponse](&client.Communicator, func(req *altv1.AggregateMonthlyBarsRequest) (*altv1.AggregateMonthlyBarsResponse, error) {
|
|
return handleAggregateMonthlyBars(worker, req)
|
|
})
|
|
},
|
|
},
|
|
{
|
|
requestType: protoSocket.TypeNameOf(&altv1.SchedulerRefreshStatusRequest{}),
|
|
register: func(client *protoSocket.WsClient) {
|
|
protoSocket.AddRequestListenerTyped[*altv1.SchedulerRefreshStatusRequest, *altv1.SchedulerRefreshStatusResponse](&client.Communicator, func(req *altv1.SchedulerRefreshStatusRequest) (*altv1.SchedulerRefreshStatusResponse, error) {
|
|
return handleSchedulerRefreshStatus(worker, req)
|
|
})
|
|
},
|
|
},
|
|
}
|
|
}
|
|
|
|
// handleImportDailyBars validates the import command shape, then forwards it to
|
|
// the worker which owns provider fetch and persistence. The API stays a thin
|
|
// control plane: it rejects malformed commands before opening a worker
|
|
// connection and never runs the import itself.
|
|
func handleImportDailyBars(worker workerclient.WorkerClient, req *altv1.ImportDailyBarsRequest) (*altv1.ImportDailyBarsResponse, error) {
|
|
if req.GetProvider() == "" {
|
|
return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest("provider is required")}, nil
|
|
}
|
|
if req.GetSelectorKind() == "" {
|
|
return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest("selector_kind is required")}, nil
|
|
}
|
|
if len(req.GetSymbols()) == 0 {
|
|
return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest("symbols is required")}, nil
|
|
}
|
|
if !isValidMarketFilter(req.GetMarket()) {
|
|
return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest("market is unsupported")}, nil
|
|
}
|
|
if !isValidVenueFilter(req.GetVenue()) {
|
|
return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest("venue is unsupported")}, nil
|
|
}
|
|
if err := validateImportDateRange(req.GetFromYyyymmdd(), req.GetToYyyymmdd()); err != nil {
|
|
return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest(err.Error())}, nil
|
|
}
|
|
if worker == nil {
|
|
return &altv1.ImportDailyBarsResponse{Error: marketWorkerUnavailable()}, nil
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout)
|
|
defer cancel()
|
|
|
|
if err := worker.Connect(ctx); err != nil {
|
|
return &altv1.ImportDailyBarsResponse{Error: marketWorkerErrorInfo(err)}, nil
|
|
}
|
|
|
|
res, err := worker.ImportDailyBars(ctx, req)
|
|
if err != nil {
|
|
return &altv1.ImportDailyBarsResponse{Error: marketWorkerErrorInfo(err)}, nil
|
|
}
|
|
if res == nil {
|
|
return &altv1.ImportDailyBarsResponse{Error: marketInternalError("worker returned no import daily bars response")}, nil
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
func handleListInstruments(worker workerclient.WorkerClient, req *altv1.ListInstrumentsRequest) (*altv1.ListInstrumentsResponse, error) {
|
|
if !isValidMarketFilter(req.GetMarket()) {
|
|
return &altv1.ListInstrumentsResponse{Error: marketInvalidRequest("market filter is unsupported")}, nil
|
|
}
|
|
if worker == nil {
|
|
return &altv1.ListInstrumentsResponse{Error: marketWorkerUnavailable()}, nil
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout)
|
|
defer cancel()
|
|
|
|
if err := worker.Connect(ctx); err != nil {
|
|
return &altv1.ListInstrumentsResponse{Error: marketWorkerErrorInfo(err)}, nil
|
|
}
|
|
|
|
res, err := worker.ListInstruments(ctx, req)
|
|
if err != nil {
|
|
return &altv1.ListInstrumentsResponse{Error: marketWorkerErrorInfo(err)}, nil
|
|
}
|
|
if res == nil {
|
|
return &altv1.ListInstrumentsResponse{Error: marketInternalError("worker returned no list instruments response")}, nil
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
func handleListBars(worker workerclient.WorkerClient, req *altv1.ListBarsRequest) (*altv1.ListBarsResponse, error) {
|
|
if req.GetInstrumentId() == "" {
|
|
return &altv1.ListBarsResponse{Error: marketInvalidRequest("instrument_id is required")}, nil
|
|
}
|
|
if !isValidTimeframe(req.GetTimeframe()) {
|
|
return &altv1.ListBarsResponse{Error: marketInvalidRequest("timeframe is unsupported")}, nil
|
|
}
|
|
if req.GetFromUnixMs() == 0 {
|
|
return &altv1.ListBarsResponse{Error: marketInvalidRequest("from_unix_ms is required")}, nil
|
|
}
|
|
if req.GetToUnixMs() == 0 {
|
|
return &altv1.ListBarsResponse{Error: marketInvalidRequest("to_unix_ms is required")}, nil
|
|
}
|
|
if req.GetFromUnixMs() > req.GetToUnixMs() {
|
|
return &altv1.ListBarsResponse{Error: marketInvalidRequest("from_unix_ms cannot be after to_unix_ms")}, nil
|
|
}
|
|
if worker == nil {
|
|
return &altv1.ListBarsResponse{Error: marketWorkerUnavailable()}, nil
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout)
|
|
defer cancel()
|
|
|
|
if err := worker.Connect(ctx); err != nil {
|
|
return &altv1.ListBarsResponse{Error: marketWorkerErrorInfo(err)}, nil
|
|
}
|
|
|
|
res, err := worker.ListBars(ctx, req)
|
|
if err != nil {
|
|
return &altv1.ListBarsResponse{Error: marketWorkerErrorInfo(err)}, nil
|
|
}
|
|
if res == nil {
|
|
return &altv1.ListBarsResponse{Error: marketInternalError("worker returned no list bars response")}, nil
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
func isValidMarketFilter(m altv1.Market) bool {
|
|
switch m {
|
|
case altv1.Market_MARKET_UNSPECIFIED, altv1.Market_MARKET_KR, altv1.Market_MARKET_US:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
// isValidVenueFilter accepts the unspecified venue (the import selector may not
|
|
// pin a venue) and the known venues, rejecting only out-of-range enum values.
|
|
func isValidVenueFilter(v altv1.Venue) bool {
|
|
switch v {
|
|
case altv1.Venue_VENUE_UNSPECIFIED, altv1.Venue_VENUE_KRX, altv1.Venue_VENUE_NASDAQ, altv1.Venue_VENUE_NYSE:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
// validateImportDateRange requires both endpoints of the import date range. The
|
|
// live KIS importer requires a bounded from/to range and rejects an inverted
|
|
// range, so the API closes those operator-input errors here as typed
|
|
// invalid_request instead of letting them surface as worker-internal failures.
|
|
func validateImportDateRange(fromValue, toValue string) error {
|
|
if fromValue == "" {
|
|
return errors.New("from_yyyymmdd is required")
|
|
}
|
|
if toValue == "" {
|
|
return errors.New("to_yyyymmdd is required")
|
|
}
|
|
from, err := time.Parse(importDateLayout, fromValue)
|
|
if err != nil {
|
|
return errors.New("from_yyyymmdd must be YYYYMMDD")
|
|
}
|
|
to, err := time.Parse(importDateLayout, toValue)
|
|
if err != nil {
|
|
return errors.New("to_yyyymmdd must be YYYYMMDD")
|
|
}
|
|
if from.After(to) {
|
|
return errors.New("from_yyyymmdd cannot be after to_yyyymmdd")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func isValidTimeframe(t altv1.Timeframe) bool {
|
|
switch t {
|
|
case altv1.Timeframe_TIMEFRAME_MONTHLY, altv1.Timeframe_TIMEFRAME_DAILY, altv1.Timeframe_TIMEFRAME_MINUTE_1, altv1.Timeframe_TIMEFRAME_MINUTE_5:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
// handleAggregateMonthlyBars validates the aggregation request shape and forwards
|
|
// it to the worker which owns the aggregation and storage logic.
|
|
func handleAggregateMonthlyBars(worker workerclient.WorkerClient, req *altv1.AggregateMonthlyBarsRequest) (*altv1.AggregateMonthlyBarsResponse, error) {
|
|
if req.GetProvider() == "" {
|
|
return &altv1.AggregateMonthlyBarsResponse{Error: marketInvalidRequest("provider is required")}, nil
|
|
}
|
|
if req.GetSelectorKind() == "" {
|
|
return &altv1.AggregateMonthlyBarsResponse{Error: marketInvalidRequest("selector_kind is required")}, nil
|
|
}
|
|
if len(req.GetSymbols()) == 0 {
|
|
return &altv1.AggregateMonthlyBarsResponse{Error: marketInvalidRequest("symbols is required")}, nil
|
|
}
|
|
if !isValidMarketFilter(req.GetMarket()) {
|
|
return &altv1.AggregateMonthlyBarsResponse{Error: marketInvalidRequest("market is unsupported")}, nil
|
|
}
|
|
if !isValidVenueFilter(req.GetVenue()) {
|
|
return &altv1.AggregateMonthlyBarsResponse{Error: marketInvalidRequest("venue is unsupported")}, nil
|
|
}
|
|
if err := validateImportDateRange(req.GetFromYyyymmdd(), req.GetToYyyymmdd()); err != nil {
|
|
return &altv1.AggregateMonthlyBarsResponse{Error: marketInvalidRequest(err.Error())}, nil
|
|
}
|
|
if worker == nil {
|
|
return &altv1.AggregateMonthlyBarsResponse{Error: marketWorkerUnavailable()}, nil
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout)
|
|
defer cancel()
|
|
|
|
if err := worker.Connect(ctx); err != nil {
|
|
return &altv1.AggregateMonthlyBarsResponse{Error: marketWorkerErrorInfo(err)}, nil
|
|
}
|
|
|
|
res, err := worker.AggregateMonthlyBars(ctx, req)
|
|
if err != nil {
|
|
return &altv1.AggregateMonthlyBarsResponse{Error: marketWorkerErrorInfo(err)}, nil
|
|
}
|
|
if res == nil {
|
|
return &altv1.AggregateMonthlyBarsResponse{Error: marketInternalError("worker returned no aggregate monthly bars response")}, nil
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
func handleSchedulerRefreshStatus(worker workerclient.WorkerClient, req *altv1.SchedulerRefreshStatusRequest) (*altv1.SchedulerRefreshStatusResponse, error) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
|
|
if err := worker.Connect(ctx); err != nil {
|
|
return &altv1.SchedulerRefreshStatusResponse{Error: marketWorkerErrorInfo(err)}, nil
|
|
}
|
|
|
|
res, err := worker.SchedulerRefreshStatus(ctx, req)
|
|
if err != nil {
|
|
return &altv1.SchedulerRefreshStatusResponse{Error: marketWorkerErrorInfo(err)}, nil
|
|
}
|
|
if res == nil {
|
|
return &altv1.SchedulerRefreshStatusResponse{Error: marketInternalError("worker returned no scheduler refresh status response")}, nil
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
func marketInvalidRequest(reason string) *altv1.ErrorInfo {
|
|
return errorInfo(marketErrorInvalidRequest, "invalid market request: "+reason)
|
|
}
|
|
|
|
func marketWorkerUnavailable() *altv1.ErrorInfo {
|
|
return errorInfo(marketErrorUnavailable, "market worker unavailable")
|
|
}
|
|
|
|
func marketInternalError(message string) *altv1.ErrorInfo {
|
|
return errorInfo(marketErrorInternal, message)
|
|
}
|
|
|
|
func marketWorkerErrorInfo(err error) *altv1.ErrorInfo {
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
switch {
|
|
case errors.Is(err, workerclient.ErrUnavailable):
|
|
return errorInfo(marketErrorUnavailable, "market worker unavailable")
|
|
case errors.Is(err, workerclient.ErrTimeout):
|
|
return errorInfo(marketErrorTimeout, "market worker timeout")
|
|
default:
|
|
return errorInfo(marketErrorInternal, err.Error())
|
|
}
|
|
}
|