toki
47109b4e42
- Add monthly bars aggregation in worker (services/worker/internal/marketdata/aggregation/) - Support monthly timeframe in operator runner and output - Add monthly aggregation test data and test cases - Update contracts (proto, Dart, Go) for monthly timeframe support - Sync parser_map, socket handlers, and worker client across services - Add code review and plan logs for multi-timeframe coverage milestones
399 lines
15 KiB
Go
399 lines
15 KiB
Go
package socket
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"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/packages/domain/market"
|
|
"git.toki-labs.com/toki/alt/services/worker/internal/marketdata/importer"
|
|
"git.toki-labs.com/toki/alt/services/worker/internal/storage"
|
|
)
|
|
|
|
// importDateLayout matches the YYYYMMDD date fields the import command carries,
|
|
// mirroring the job payload date contract so the same dates reach the provider.
|
|
const importDateLayout = "20060102"
|
|
|
|
// validImportTimeframes defines the set of timeframes accepted for import_daily_bars.
|
|
// Currently only daily is supported; minute values are intentionally rejected.
|
|
var validImportTimeframes = map[string]bool{
|
|
"daily": true,
|
|
"minute_1": true,
|
|
"minute_5": true,
|
|
"monthly": true,
|
|
}
|
|
|
|
const (
|
|
marketErrorInvalidRequest = "invalid_request"
|
|
marketErrorUnavailable = "unavailable"
|
|
marketErrorNotFound = "not_found"
|
|
marketErrorTimeout = "timeout"
|
|
marketErrorInternal = "internal"
|
|
)
|
|
|
|
func marketHandlers(deps Deps) []sessionHandler {
|
|
handlers := []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(deps, 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(deps, 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(deps, req)
|
|
},
|
|
)
|
|
},
|
|
},
|
|
}
|
|
if deps.MonthlyAggregator != nil {
|
|
handlers = append(handlers, sessionHandler{
|
|
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(deps, req)
|
|
},
|
|
)
|
|
},
|
|
})
|
|
}
|
|
return handlers
|
|
}
|
|
|
|
// importTimeframeNormalise resolves the proto Timeframe into a string accepted
|
|
// by the provider layer. Unspecified defaults to "daily"; known enum names are
|
|
// mapped explicitly. Unknown enum values return ("", unknownTimeframeErr) so
|
|
// that the caller can reject them as a typed invalid_request error instead of
|
|
// panicking on slice operations.
|
|
func importTimeframeNormalise(tf altv1.Timeframe) (string, error) {
|
|
if tf == altv1.Timeframe_TIMEFRAME_UNSPECIFIED {
|
|
return "daily", nil
|
|
}
|
|
// Map known proto enum values to their lowercase name (without TIMEFRAME_ prefix).
|
|
switch tf {
|
|
case altv1.Timeframe_TIMEFRAME_DAILY:
|
|
return "daily", nil
|
|
case altv1.Timeframe_TIMEFRAME_MINUTE_1:
|
|
return "minute_1", nil
|
|
case altv1.Timeframe_TIMEFRAME_MINUTE_5:
|
|
return "minute_5", nil
|
|
case altv1.Timeframe_TIMEFRAME_MONTHLY:
|
|
return "monthly", nil
|
|
default:
|
|
// Unknown enum value: return the raw string name (if available) so the
|
|
// caller can produce a stable error message. Do NOT panic on slice.
|
|
name := tf.String()
|
|
if strings.HasPrefix(name, "TIMEFRAME_") {
|
|
return strings.ToLower(name[len("TIMEFRAME_"):]), fmt.Errorf("unsupported timeframe %q", strings.ToLower(name[len("TIMEFRAME_"):]))
|
|
}
|
|
return strings.ToLower(name), fmt.Errorf("unsupported timeframe %q", strings.ToLower(name))
|
|
}
|
|
}
|
|
|
|
// handleImportDailyBars validates the import command, converts it into the
|
|
// importer request, and runs the import. The worker owns provider fetch and
|
|
// persistence; counts come back so the operator surface can confirm what landed.
|
|
func handleImportDailyBars(deps Deps, 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
|
|
}
|
|
|
|
// Normalise/validate the import timeframe. Daily (unspecified) is the only
|
|
// accepted value for import_daily_bars at this stage; minute values are
|
|
// rejected as a typed invalid_request so the operator sees a stable error.
|
|
tfStr, tfErr := importTimeframeNormalise(req.GetTimeframe())
|
|
if tfErr != nil {
|
|
return &altv1.ImportDailyBarsResponse{
|
|
Error: marketInvalidRequest(tfErr.Error()),
|
|
}, nil
|
|
}
|
|
if !validImportTimeframes[tfStr] {
|
|
return &altv1.ImportDailyBarsResponse{
|
|
Error: marketInvalidRequest(fmt.Sprintf("import_daily_bars supports timeframe daily; got %s", tfStr)),
|
|
}, nil
|
|
}
|
|
if tfStr != "daily" {
|
|
return &altv1.ImportDailyBarsResponse{
|
|
Error: marketInvalidRequest(fmt.Sprintf("import_daily_bars supports timeframe daily; got %s", tfStr)),
|
|
}, nil
|
|
}
|
|
|
|
mkt, _, err := marketFilterFromProto(req.GetMarket())
|
|
if err != nil {
|
|
return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest(err.Error())}, nil
|
|
}
|
|
venue, err := venueFromProto(req.GetVenue())
|
|
if err != nil {
|
|
return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest(err.Error())}, nil
|
|
}
|
|
from, to, err := parseImportDateRange(req.GetFromYyyymmdd(), req.GetToYyyymmdd())
|
|
if err != nil {
|
|
return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest(err.Error())}, nil
|
|
}
|
|
if deps.DailyBarImporter == nil {
|
|
return &altv1.ImportDailyBarsResponse{Error: marketUnavailableError("daily bar import is not available")}, nil
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), handlerTimeout)
|
|
defer cancel()
|
|
|
|
result, err := deps.DailyBarImporter.ImportDailyBars(ctx, importer.DailyBarRequest{
|
|
Provider: market.Provider(req.GetProvider()),
|
|
Selector: market.UniverseSelector{
|
|
Kind: market.UniverseSelectorKind(req.GetSelectorKind()),
|
|
Market: mkt,
|
|
Venue: venue,
|
|
Symbols: req.GetSymbols(),
|
|
Name: req.GetName(),
|
|
},
|
|
From: from,
|
|
To: to,
|
|
})
|
|
if err != nil {
|
|
return &altv1.ImportDailyBarsResponse{Error: marketBackendErrorInfo(err)}, nil
|
|
}
|
|
return &altv1.ImportDailyBarsResponse{
|
|
Provider: req.GetProvider(),
|
|
InstrumentCount: int32(result.Instruments),
|
|
BarCount: int32(result.Bars),
|
|
}, nil
|
|
}
|
|
|
|
// parseImportDateRange requires both endpoints of the import date range and
|
|
// returns them parsed for the importer request. The live KIS importer requires
|
|
// a bounded from/to range and rejects an inverted range, so the worker closes
|
|
// those operator-input errors here as invalid_request before reaching the
|
|
// importer instead of letting them surface as an internal backend failure.
|
|
func parseImportDateRange(fromValue, toValue string) (time.Time, time.Time, error) {
|
|
if fromValue == "" {
|
|
return time.Time{}, time.Time{}, errors.New("from_yyyymmdd is required")
|
|
}
|
|
if toValue == "" {
|
|
return time.Time{}, time.Time{}, errors.New("to_yyyymmdd is required")
|
|
}
|
|
from, err := time.Parse(importDateLayout, fromValue)
|
|
if err != nil {
|
|
return time.Time{}, time.Time{}, errors.New("from_yyyymmdd must be YYYYMMDD")
|
|
}
|
|
to, err := time.Parse(importDateLayout, toValue)
|
|
if err != nil {
|
|
return time.Time{}, time.Time{}, errors.New("to_yyyymmdd must be YYYYMMDD")
|
|
}
|
|
if from.After(to) {
|
|
return time.Time{}, time.Time{}, errors.New("from_yyyymmdd cannot be after to_yyyymmdd")
|
|
}
|
|
return from.UTC(), to.UTC(), nil
|
|
}
|
|
|
|
func handleListInstruments(deps Deps, req *altv1.ListInstrumentsRequest) (*altv1.ListInstrumentsResponse, error) {
|
|
marketFilter, hasMarketFilter, err := marketFilterFromProto(req.GetMarket())
|
|
if err != nil {
|
|
return &altv1.ListInstrumentsResponse{Error: marketInvalidRequest(err.Error())}, nil
|
|
}
|
|
if deps.Instruments == nil {
|
|
return &altv1.ListInstrumentsResponse{Error: marketUnavailableError("instrument store is not available")}, nil
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), handlerTimeout)
|
|
defer cancel()
|
|
|
|
insts, err := deps.Instruments.ListInstruments(ctx)
|
|
if err != nil {
|
|
return &altv1.ListInstrumentsResponse{Error: marketBackendErrorInfo(err)}, nil
|
|
}
|
|
filtered := filterInstruments(insts, marketFilter, hasMarketFilter, req.GetProvider())
|
|
return &altv1.ListInstrumentsResponse{Instruments: instrumentsToProto(filtered)}, nil
|
|
}
|
|
|
|
func handleListBars(deps Deps, req *altv1.ListBarsRequest) (*altv1.ListBarsResponse, error) {
|
|
if req.GetInstrumentId() == "" {
|
|
return &altv1.ListBarsResponse{Error: marketInvalidRequest("instrument_id is required")}, nil
|
|
}
|
|
timeframe, err := timeframeFromProto(req.GetTimeframe())
|
|
if err != nil {
|
|
return &altv1.ListBarsResponse{Error: marketInvalidRequest(err.Error())}, 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 deps.Bars == nil {
|
|
return &altv1.ListBarsResponse{Error: marketUnavailableError("bar store is not available")}, nil
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), handlerTimeout)
|
|
defer cancel()
|
|
|
|
from := time.UnixMilli(req.GetFromUnixMs()).UTC()
|
|
to := time.UnixMilli(req.GetToUnixMs()).UTC()
|
|
bars, err := deps.Bars.GetBars(ctx, market.InstrumentID(req.GetInstrumentId()), timeframe, from, to)
|
|
if err != nil {
|
|
return &altv1.ListBarsResponse{Error: marketBackendErrorInfo(err)}, nil
|
|
}
|
|
return &altv1.ListBarsResponse{Bars: barsToProto(bars)}, nil
|
|
}
|
|
|
|
func marketFilterFromProto(m altv1.Market) (market.Market, bool, error) {
|
|
switch m {
|
|
case altv1.Market_MARKET_UNSPECIFIED:
|
|
return "", false, nil
|
|
case altv1.Market_MARKET_KR, altv1.Market_MARKET_US:
|
|
converted, err := marketFromProto(m)
|
|
return converted, true, err
|
|
default:
|
|
return "", false, errors.New("market filter is unsupported")
|
|
}
|
|
}
|
|
|
|
func filterInstruments(insts []market.Instrument, marketFilter market.Market, hasMarketFilter bool, provider string) []market.Instrument {
|
|
if !hasMarketFilter && provider == "" {
|
|
return insts
|
|
}
|
|
out := make([]market.Instrument, 0, len(insts))
|
|
for _, inst := range insts {
|
|
if hasMarketFilter && inst.Market != marketFilter {
|
|
continue
|
|
}
|
|
if provider != "" && inst.ProviderSymbols[provider] == "" {
|
|
continue
|
|
}
|
|
out = append(out, inst)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func marketInvalidRequest(reason string) *altv1.ErrorInfo {
|
|
return errorInfo(marketErrorInvalidRequest, "invalid market request: "+reason)
|
|
}
|
|
|
|
func marketUnavailableError(message string) *altv1.ErrorInfo {
|
|
return errorInfo(marketErrorUnavailable, message)
|
|
}
|
|
|
|
func marketBackendErrorInfo(err error) *altv1.ErrorInfo {
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
switch {
|
|
case errors.Is(err, storage.ErrInstrumentNotFound):
|
|
return errorInfo(marketErrorNotFound, err.Error())
|
|
case errors.Is(err, ErrMonthlySourceBarsNotFound):
|
|
return errorInfo(marketErrorNotFound, err.Error())
|
|
case errors.Is(err, context.DeadlineExceeded):
|
|
return errorInfo(marketErrorTimeout, err.Error())
|
|
default:
|
|
return errorInfo(marketErrorInternal, err.Error())
|
|
}
|
|
}
|
|
|
|
// handleAggregateMonthlyBars validates the aggregation request, converts it
|
|
// into an AggregateMonthlyRequest and delegates to the MonthlyBarAggregator.
|
|
// The response carries instrument counts, source daily bar counts, monthly bar
|
|
// counts, and per-instrument provenance entries.
|
|
func handleAggregateMonthlyBars(deps Deps, 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
|
|
}
|
|
|
|
mkt, _, err := marketFilterFromProto(req.GetMarket())
|
|
if err != nil {
|
|
return &altv1.AggregateMonthlyBarsResponse{Error: marketInvalidRequest(err.Error())}, nil
|
|
}
|
|
venue, err := venueFromProto(req.GetVenue())
|
|
if err != nil {
|
|
return &altv1.AggregateMonthlyBarsResponse{Error: marketInvalidRequest(err.Error())}, nil
|
|
}
|
|
from, to, err := parseImportDateRange(req.GetFromYyyymmdd(), req.GetToYyyymmdd())
|
|
if err != nil {
|
|
return &altv1.AggregateMonthlyBarsResponse{Error: marketInvalidRequest(err.Error())}, nil
|
|
}
|
|
if deps.MonthlyAggregator == nil {
|
|
return &altv1.AggregateMonthlyBarsResponse{Error: marketUnavailableError("monthly aggregation is not available")}, nil
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), handlerTimeout)
|
|
defer cancel()
|
|
|
|
result, err := deps.MonthlyAggregator.AggregateMonthlyBars(ctx, AggregateMonthlyRequest{
|
|
Provider: market.Provider(req.GetProvider()),
|
|
Selector: market.UniverseSelector{
|
|
Kind: market.UniverseSelectorKind(req.GetSelectorKind()),
|
|
Market: mkt,
|
|
Venue: venue,
|
|
Symbols: req.GetSymbols(),
|
|
Name: req.GetName(),
|
|
},
|
|
From: from,
|
|
To: to,
|
|
})
|
|
if err != nil {
|
|
return &altv1.AggregateMonthlyBarsResponse{Error: marketBackendErrorInfo(err)}, nil
|
|
}
|
|
|
|
provenance := make([]*altv1.MonthlyProvenance, 0, len(result.Provenance))
|
|
for _, p := range result.Provenance {
|
|
provenance = append(provenance, &altv1.MonthlyProvenance{
|
|
InstrumentId: string(p.InstrumentID),
|
|
SourceDailyBarCount: int32(p.SourceDailyCount),
|
|
MonthlyBarCount: int32(p.MonthlyBarCount),
|
|
SourceStartYyyymmdd: p.SourceStart.Format(importDateLayout),
|
|
SourceEndYyyymmdd: p.SourceEnd.Format(importDateLayout),
|
|
AggregationRuleId: p.AggregationRuleID,
|
|
SourceTimeframe: altv1.Timeframe_TIMEFRAME_DAILY,
|
|
TargetTimeframe: altv1.Timeframe_TIMEFRAME_MONTHLY,
|
|
})
|
|
}
|
|
|
|
return &altv1.AggregateMonthlyBarsResponse{
|
|
Provider: req.GetProvider(),
|
|
InstrumentCount: int32(result.Instruments),
|
|
SourceDailyBarCount: int32(result.SourceDailyCount),
|
|
MonthlyBarCount: int32(result.MonthlyBarCount),
|
|
Provenance: provenance,
|
|
}, nil
|
|
}
|