toki
243a1293f4
- Add backtest matrix test cases (success, failed, mismatch, timeout) - Update runner with backtest scenario support - Update scenario.go with handoff validation - Update handoff tests for backtest scenarios - Remove obsolete plan/review documents (moved to archive)
1640 lines
50 KiB
Go
1640 lines
50 KiB
Go
package operator
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
altv1 "git.toki-labs.com/toki/alt/packages/contracts/gen/go/alt/v1"
|
|
)
|
|
|
|
// Run command exit codes. They form the terminal contract scripts rely on to
|
|
// tell a healthy run, a typed/expectation failure, bad input, and a dead
|
|
// transport apart.
|
|
const (
|
|
codeOK = 0 // every step matched its expectation
|
|
codeAppError = 1 // unexpected typed ErrorInfo or expectation mismatch
|
|
codeBadInput = 2 // unsupported action reached the runner
|
|
codeTransport = 3 // connect/disconnect/timeout before a typed response
|
|
)
|
|
|
|
// Step status values written to the output stream.
|
|
const (
|
|
statusOK = "ok"
|
|
statusError = "error"
|
|
statusMismatch = "mismatch"
|
|
statusTransportError = "transport_error"
|
|
)
|
|
|
|
// Run summary status values.
|
|
const (
|
|
summaryOK = "ok"
|
|
summaryFailed = "failed"
|
|
summaryTransport = "transport_error"
|
|
)
|
|
|
|
// Identity the CLI presents to the API in a hello handshake.
|
|
const (
|
|
clientName = "alt-operator-cli"
|
|
clientVersion = "dev"
|
|
altProtocolVersion = "alt.v1"
|
|
)
|
|
|
|
// marketByName maps scenario market strings onto the contract enum. The keys
|
|
// match scenario.go's validMarkets so a value accepted at validation always has
|
|
// a mapping here.
|
|
var marketByName = map[string]altv1.Market{
|
|
"": altv1.Market_MARKET_UNSPECIFIED,
|
|
"unspecified": altv1.Market_MARKET_UNSPECIFIED,
|
|
"kr": altv1.Market_MARKET_KR,
|
|
"us": altv1.Market_MARKET_US,
|
|
}
|
|
|
|
// timeframeByName maps scenario timeframe strings onto the contract enum.
|
|
var timeframeByName = map[string]altv1.Timeframe{
|
|
"daily": altv1.Timeframe_TIMEFRAME_DAILY,
|
|
"minute_1": altv1.Timeframe_TIMEFRAME_MINUTE_1,
|
|
"minute_5": altv1.Timeframe_TIMEFRAME_MINUTE_5,
|
|
}
|
|
|
|
// venueByName maps scenario venue strings onto the contract enum.
|
|
var venueByName = map[string]altv1.Venue{
|
|
"": altv1.Venue_VENUE_UNSPECIFIED,
|
|
"unspecified": altv1.Venue_VENUE_UNSPECIFIED,
|
|
"krx": altv1.Venue_VENUE_KRX,
|
|
"nasdaq": altv1.Venue_VENUE_NASDAQ,
|
|
"nyse": altv1.Venue_VENUE_NYSE,
|
|
}
|
|
|
|
// currencyByMarket derives the paper account currency from the request market so
|
|
// scenarios only declare a market and a starting cash amount. KR settles in KRW
|
|
// and US in USD; an unspecified market leaves the currency unspecified and the
|
|
// worker rejects it as an invalid request.
|
|
var currencyByMarket = map[string]altv1.Currency{
|
|
"kr": altv1.Currency_CURRENCY_KRW,
|
|
"us": altv1.Currency_CURRENCY_USD,
|
|
}
|
|
|
|
// RunOptions configures a scenario run.
|
|
type RunOptions struct {
|
|
// APIURL is the ws/wss endpoint of the API control plane.
|
|
APIURL string
|
|
// Timeout is the per-step request timeout. When zero, the scenario timeout
|
|
// is used, falling back to 5s.
|
|
Timeout time.Duration
|
|
// Format selects the output rendering.
|
|
Format OutputFormat
|
|
}
|
|
|
|
// RunScenario connects to the API, executes each step, writes machine-readable
|
|
// events to stdout, and returns the process exit code. A transport failure
|
|
// before any typed response yields codeTransport; an unexpected typed ErrorInfo
|
|
// or a failed expectation yields codeAppError.
|
|
func RunScenario(ctx context.Context, sc *Scenario, opts RunOptions, stdout io.Writer) int {
|
|
out := NewWriter(stdout, opts.Format)
|
|
|
|
timeout := opts.Timeout
|
|
if timeout <= 0 {
|
|
timeout = time.Duration(sc.Timeout)
|
|
}
|
|
if timeout <= 0 {
|
|
timeout = 5 * time.Second
|
|
}
|
|
|
|
dialCtx, cancelDial := context.WithTimeout(ctx, timeout)
|
|
client, err := Dial(dialCtx, opts.APIURL)
|
|
cancelDial()
|
|
if err != nil {
|
|
out.WriteStep(StepEvent{
|
|
Scenario: sc.Name,
|
|
Step: firstStepID(sc),
|
|
Action: firstStepAction(sc),
|
|
Status: statusTransportError,
|
|
Count: noCount,
|
|
InstrumentCount: noCount,
|
|
BarCount: noCount,
|
|
ErrorMessage: err.Error(),
|
|
})
|
|
out.WriteSummary(RunSummary{Scenario: sc.Name, Status: summaryTransport, Steps: len(sc.Steps), Passed: 0, ExitCode: codeTransport})
|
|
return codeTransport
|
|
}
|
|
defer client.Close()
|
|
|
|
runs, err := sc.MatrixRuns()
|
|
if err != nil {
|
|
out.WriteStep(StepEvent{Scenario: sc.Name, Status: statusError, ErrorCode: "invalid_matrix", ErrorMessage: err.Error()})
|
|
out.WriteSummary(RunSummary{Scenario: sc.Name, Status: summaryFailed, Steps: 0, Passed: 0, ExitCode: codeBadInput})
|
|
return codeBadInput
|
|
}
|
|
if len(runs) > 0 {
|
|
return runMatrixBatch(ctx, client, sc.Name, runs, timeout, out)
|
|
}
|
|
|
|
passed := 0
|
|
exitCode := codeOK
|
|
savedValues := make(map[string]string)
|
|
for _, step := range sc.Steps {
|
|
ev, code, runID := runStep(ctx, client, sc.Name, step, timeout, savedValues)
|
|
if runID != "" {
|
|
savedValues[step.ID] = runID
|
|
if step.SaveAs != "" {
|
|
savedValues[step.SaveAs] = runID
|
|
}
|
|
}
|
|
|
|
// Post-process step expectations for expected exit code/transport status
|
|
if step.Expect.TransportStatus != "" {
|
|
if ev.Status == step.Expect.TransportStatus {
|
|
ev.Status = statusOK
|
|
code = codeOK
|
|
}
|
|
}
|
|
|
|
if step.Expect.ExitCode != nil {
|
|
if code == *step.Expect.ExitCode {
|
|
ev.Status = statusOK
|
|
code = codeOK
|
|
} else {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected exit code %d, got %d", *step.Expect.ExitCode, code)
|
|
code = codeAppError
|
|
}
|
|
}
|
|
|
|
out.WriteStep(ev)
|
|
if code == codeOK {
|
|
passed++
|
|
} else if code > exitCode {
|
|
exitCode = code
|
|
}
|
|
if code == codeTransport {
|
|
// The socket dropped mid-run; remaining steps cannot be trusted.
|
|
break
|
|
}
|
|
}
|
|
|
|
out.WriteSummary(RunSummary{
|
|
Scenario: sc.Name,
|
|
Status: summaryStatus(exitCode),
|
|
Steps: len(sc.Steps),
|
|
Passed: passed,
|
|
ExitCode: exitCode,
|
|
})
|
|
return exitCode
|
|
}
|
|
|
|
func summaryStatus(exitCode int) string {
|
|
switch exitCode {
|
|
case codeOK:
|
|
return summaryOK
|
|
case codeTransport:
|
|
return summaryTransport
|
|
default:
|
|
return summaryFailed
|
|
}
|
|
}
|
|
|
|
func firstStepID(sc *Scenario) string {
|
|
if len(sc.Steps) > 0 {
|
|
return sc.Steps[0].ID
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func firstStepAction(sc *Scenario) string {
|
|
if len(sc.Steps) > 0 {
|
|
return string(sc.Steps[0].Action)
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// runStep executes one step and classifies the result into a StepEvent, an
|
|
// exit code, and an optional backtest run ID to save for interpolation.
|
|
func runStep(ctx context.Context, client *APIClient, scenario string, step Step, timeout time.Duration, savedValues map[string]string) (StepEvent, int, string) {
|
|
stepCtx, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
|
|
ev := StepEvent{
|
|
Scenario: scenario,
|
|
Step: step.ID,
|
|
Action: string(step.Action),
|
|
Count: noCount,
|
|
InstrumentCount: noCount,
|
|
BarCount: noCount,
|
|
MissingCount: noCount,
|
|
}
|
|
|
|
switch step.Action {
|
|
case ActionCollectionFreshness:
|
|
return runCollectionFreshness(stepCtx, client, scenario, step)
|
|
|
|
case ActionHello:
|
|
resp, err := client.Hello(stepCtx, &altv1.HelloRequest{
|
|
ClientName: clientName,
|
|
ClientVersion: clientVersion,
|
|
AltProtocolVersion: altProtocolVersion,
|
|
})
|
|
if err != nil {
|
|
return transportEvent3(ev, err)
|
|
}
|
|
ev2, code := evaluateHello(ev, step, resp)
|
|
return ev2, code, ""
|
|
|
|
case ActionListInstruments:
|
|
resp, err := client.ListInstruments(stepCtx, &altv1.ListInstrumentsRequest{
|
|
Market: marketByName[step.Request.Market],
|
|
})
|
|
if err != nil {
|
|
return transportEvent3(ev, err)
|
|
}
|
|
ev2, code := evaluateMarket(ev, step, resp.GetError(), len(resp.GetInstruments()))
|
|
return ev2, code, ""
|
|
|
|
case ActionImportDailyBars:
|
|
resp, err := client.ImportDailyBars(stepCtx, &altv1.ImportDailyBarsRequest{
|
|
Provider: step.Request.Provider,
|
|
SelectorKind: step.Request.SelectorKind,
|
|
Market: marketByName[step.Request.Market],
|
|
Venue: venueByName[step.Request.Venue],
|
|
Name: step.Request.Name,
|
|
Symbols: step.Request.Symbols,
|
|
FromYyyymmdd: step.Request.FromYYYYMMDD,
|
|
ToYyyymmdd: step.Request.ToYYYYMMDD,
|
|
})
|
|
if err != nil {
|
|
return transportEvent3(ev, err)
|
|
}
|
|
ev2, code := evaluateImport(ev, step, resp)
|
|
return ev2, code, ""
|
|
|
|
case ActionListBars:
|
|
resp, err := client.ListBars(stepCtx, &altv1.ListBarsRequest{
|
|
InstrumentId: step.Request.InstrumentID,
|
|
Timeframe: timeframeByName[step.Request.Timeframe],
|
|
FromUnixMs: step.Request.FromUnixMs,
|
|
ToUnixMs: step.Request.ToUnixMs,
|
|
})
|
|
if err != nil {
|
|
return transportEvent3(ev, err)
|
|
}
|
|
ev2, code := evaluateMarket(ev, step, resp.GetError(), len(resp.GetBars()))
|
|
return ev2, code, ""
|
|
|
|
case ActionStartBacktest:
|
|
resp, err := client.StartBacktest(stepCtx, &altv1.StartBacktestRequest{
|
|
Spec: &altv1.BacktestRunSpec{
|
|
StrategyId: step.Request.StrategyID,
|
|
Market: marketByName[step.Request.Market],
|
|
Timeframe: timeframeByName[step.Request.Timeframe],
|
|
FromUnixMs: step.Request.FromUnixMs,
|
|
ToUnixMs: step.Request.ToUnixMs,
|
|
Selector: &altv1.BacktestInputSelector{
|
|
InstrumentIds: step.Request.InstrumentIDs,
|
|
Symbols: step.Request.Symbols,
|
|
},
|
|
},
|
|
})
|
|
if err != nil {
|
|
return transportEvent3(ev, err)
|
|
}
|
|
return evaluateBacktest(ev, step, resp.GetError(), resp.GetRun(), nil, noCount)
|
|
|
|
case ActionListBacktestRuns:
|
|
var statusEnum altv1.BacktestRunStatus
|
|
if step.Request.Status != "" {
|
|
statusEnum = altv1.BacktestRunStatus(altv1.BacktestRunStatus_value["BACKTEST_RUN_STATUS_"+strings.ToUpper(step.Request.Status)])
|
|
}
|
|
resp, err := client.ListBacktestRuns(stepCtx, &altv1.ListBacktestRunsRequest{
|
|
Status: statusEnum,
|
|
})
|
|
if err != nil {
|
|
return transportEvent3(ev, err)
|
|
}
|
|
return evaluateBacktest(ev, step, resp.GetError(), nil, nil, len(resp.GetRuns()))
|
|
|
|
case ActionGetBacktestRunDetail:
|
|
runID := interpolate(step.Request.RunID, savedValues)
|
|
resp, err := client.GetBacktestRunDetail(stepCtx, &altv1.GetBacktestRunDetailRequest{
|
|
RunId: runID,
|
|
})
|
|
if err != nil {
|
|
return transportEvent3(ev, err)
|
|
}
|
|
return evaluateBacktest(ev, step, resp.GetError(), resp.GetRun(), resp.GetResult(), noCount)
|
|
|
|
case ActionGetBacktestResult:
|
|
runID := interpolate(step.Request.RunID, savedValues)
|
|
resp, err := client.GetBacktestResult(stepCtx, &altv1.GetBacktestResultRequest{
|
|
RunId: runID,
|
|
})
|
|
if err != nil {
|
|
return transportEvent3(ev, err)
|
|
}
|
|
return evaluateBacktest(ev, step, resp.GetError(), nil, resp.GetResult(), noCount)
|
|
|
|
case ActionCompareBacktestRuns:
|
|
runIDs := interpolateSlice(step.Request.RunIDs, savedValues)
|
|
resp, err := client.CompareBacktestRuns(stepCtx, &altv1.CompareBacktestRunsRequest{
|
|
RunIds: runIDs,
|
|
})
|
|
if err != nil {
|
|
return transportEvent3(ev, err)
|
|
}
|
|
return evaluateBacktest(ev, step, resp.GetError(), nil, nil, len(resp.GetResults()))
|
|
|
|
case ActionPollBacktestRun:
|
|
runID := interpolate(step.Request.RunID, savedValues)
|
|
interval := time.Duration(step.Request.PollingInterval)
|
|
if interval <= 0 {
|
|
interval = 500 * time.Millisecond
|
|
}
|
|
pollTimeout := time.Duration(step.Request.PollingTimeout)
|
|
if pollTimeout <= 0 {
|
|
pollTimeout = timeout
|
|
}
|
|
pollCtx, pollCancel := context.WithTimeout(ctx, pollTimeout)
|
|
defer pollCancel()
|
|
|
|
var lastResp *altv1.GetBacktestRunResponse
|
|
var lastErr error
|
|
|
|
for {
|
|
select {
|
|
case <-pollCtx.Done():
|
|
if lastResp != nil && lastResp.GetError() != nil {
|
|
return evaluateBacktest(ev, step, lastResp.GetError(), lastResp.GetRun(), nil, noCount)
|
|
}
|
|
ev.Status = statusTransportError
|
|
if lastErr != nil {
|
|
ev.ErrorMessage = fmt.Sprintf("polling timed out: %v (last error: %v)", pollCtx.Err(), lastErr)
|
|
} else {
|
|
ev.ErrorMessage = fmt.Sprintf("polling timed out: %v", pollCtx.Err())
|
|
}
|
|
return ev, codeTransport, ""
|
|
default:
|
|
}
|
|
|
|
resp, err := client.GetBacktestRun(pollCtx, &altv1.GetBacktestRunRequest{
|
|
RunId: runID,
|
|
})
|
|
if err != nil {
|
|
lastErr = err
|
|
select {
|
|
case <-pollCtx.Done():
|
|
case <-time.After(interval):
|
|
}
|
|
continue
|
|
}
|
|
|
|
lastResp = resp
|
|
if resp.GetError() != nil {
|
|
return evaluateBacktest(ev, step, resp.GetError(), resp.GetRun(), nil, noCount)
|
|
}
|
|
|
|
run := resp.GetRun()
|
|
if run != nil {
|
|
statusStr := strings.ToLower(strings.TrimPrefix(run.GetStatus().String(), "BACKTEST_RUN_STATUS_"))
|
|
isExpected := false
|
|
if step.Expect.RunStatus != "" {
|
|
isExpected = statusStr == strings.ToLower(step.Expect.RunStatus)
|
|
} else {
|
|
isExpected = statusStr == "succeeded" || statusStr == "failed" || statusStr == "canceled"
|
|
}
|
|
|
|
if isExpected {
|
|
return evaluateBacktest(ev, step, nil, run, nil, noCount)
|
|
}
|
|
}
|
|
|
|
select {
|
|
case <-pollCtx.Done():
|
|
case <-time.After(interval):
|
|
}
|
|
}
|
|
|
|
case ActionStartPaperTrading:
|
|
resp, err := client.StartPaperTrading(stepCtx, &altv1.StartPaperTradingRequest{
|
|
AccountId: step.Request.AccountID,
|
|
Spec: &altv1.BacktestRunSpec{
|
|
StrategyId: step.Request.StrategyID,
|
|
Market: marketByName[step.Request.Market],
|
|
Timeframe: timeframeByName[step.Request.Timeframe],
|
|
FromUnixMs: step.Request.FromUnixMs,
|
|
ToUnixMs: step.Request.ToUnixMs,
|
|
},
|
|
StartingCash: &altv1.Price{
|
|
Currency: currencyByMarket[step.Request.Market],
|
|
Amount: &altv1.Decimal{Value: step.Request.StartingCash},
|
|
},
|
|
})
|
|
if err != nil {
|
|
return transportEvent3(ev, err)
|
|
}
|
|
ev2, code := evaluatePaper(ev, step, resp.GetError(), resp.GetState())
|
|
return ev2, code, ""
|
|
|
|
case ActionGetPaperTradingState:
|
|
resp, err := client.GetPaperTradingState(stepCtx, &altv1.GetPaperTradingStateRequest{
|
|
AccountId: step.Request.AccountID,
|
|
})
|
|
if err != nil {
|
|
return transportEvent3(ev, err)
|
|
}
|
|
ev2, code := evaluatePaper(ev, step, resp.GetError(), resp.GetState())
|
|
return ev2, code, ""
|
|
|
|
case ActionSubmitPaperOrder:
|
|
req := &altv1.SubmitPaperOrderRequest{
|
|
AccountId: step.Request.AccountID,
|
|
InstrumentId: step.Request.InstrumentID,
|
|
Side: step.Request.Side,
|
|
Quantity: &altv1.Quantity{Amount: &altv1.Decimal{Value: step.Request.Quantity}},
|
|
Type: step.Request.OrderType,
|
|
}
|
|
if step.Request.OrderType == "limit" {
|
|
req.LimitPrice = &altv1.Price{
|
|
Currency: currencyByMarket[step.Request.Market],
|
|
Amount: &altv1.Decimal{Value: step.Request.LimitPrice},
|
|
}
|
|
}
|
|
resp, err := client.SubmitPaperOrder(stepCtx, req)
|
|
if err != nil {
|
|
return transportEvent3(ev, err)
|
|
}
|
|
return evaluatePaperOrder(ev, step, resp.GetError(), resp.GetOrder())
|
|
|
|
case ActionCancelPaperOrder:
|
|
orderID := interpolate(step.Request.OrderID, savedValues)
|
|
resp, err := client.CancelPaperOrder(stepCtx, &altv1.CancelPaperOrderRequest{
|
|
AccountId: step.Request.AccountID,
|
|
OrderId: orderID,
|
|
})
|
|
if err != nil {
|
|
return transportEvent3(ev, err)
|
|
}
|
|
return evaluatePaperOrder(ev, step, resp.GetError(), resp.GetOrder())
|
|
|
|
case ActionFillPaperOrder:
|
|
orderID := interpolate(step.Request.OrderID, savedValues)
|
|
fillReq := &altv1.FillPaperOrderRequest{
|
|
AccountId: step.Request.AccountID,
|
|
OrderId: orderID,
|
|
}
|
|
if step.Request.FillPrice != "" {
|
|
fillReq.FillPrice = &altv1.Price{
|
|
Currency: currencyByMarket[step.Request.Market],
|
|
Amount: &altv1.Decimal{Value: step.Request.FillPrice},
|
|
}
|
|
}
|
|
resp, err := client.FillPaperOrder(stepCtx, fillReq)
|
|
if err != nil {
|
|
return transportEvent3(ev, err)
|
|
}
|
|
return evaluatePaperOrder(ev, step, resp.GetError(), resp.GetOrder())
|
|
|
|
case ActionSubmitLiveOrder:
|
|
req := &altv1.SubmitLiveOrderRequest{
|
|
AccountId: step.Request.AccountID,
|
|
Intent: &altv1.LiveOrderIntent{
|
|
InstrumentId: step.Request.InstrumentID,
|
|
Side: step.Request.Side,
|
|
Type: step.Request.OrderType,
|
|
Quantity: &altv1.Quantity{Amount: &altv1.Decimal{Value: step.Request.Quantity}},
|
|
},
|
|
OperatorConfirmation: &altv1.OperatorConfirmation{
|
|
Confirmed: step.Request.OperatorConfirmed,
|
|
OperatorId: step.Request.OperatorID,
|
|
Reason: step.Request.ConfirmationReason,
|
|
},
|
|
IdempotencyKey: step.Request.IdempotencyKey,
|
|
}
|
|
if step.Request.LimitPrice != "" {
|
|
req.Intent.LimitPrice = &altv1.Price{
|
|
Currency: currencyByMarket[step.Request.Market],
|
|
Amount: &altv1.Decimal{Value: step.Request.LimitPrice},
|
|
}
|
|
}
|
|
sresp, serr := client.SubmitLiveOrder(stepCtx, req)
|
|
if serr != nil {
|
|
return transportEvent3(ev, serr)
|
|
}
|
|
return evaluateLiveOrder(ev, step, sresp.GetError(), sresp.GetOrder(), sresp.GetRiskDecision())
|
|
|
|
case ActionCancelLiveOrder:
|
|
liveOrderID := interpolate(step.Request.LiveOrderID, savedValues)
|
|
cresp, cerr := client.CancelLiveOrder(stepCtx, &altv1.CancelLiveOrderRequest{
|
|
AccountId: step.Request.AccountID,
|
|
LiveOrderId: liveOrderID,
|
|
})
|
|
if cerr != nil {
|
|
return transportEvent3(ev, cerr)
|
|
}
|
|
return evaluateLiveOrder(ev, step, cresp.GetError(), cresp.GetOrder(), nil)
|
|
|
|
case ActionGetLiveOrder:
|
|
liveOrderID := interpolate(step.Request.LiveOrderID, savedValues)
|
|
gresp, gerr := client.GetLiveOrder(stepCtx, &altv1.GetLiveOrderRequest{
|
|
AccountId: step.Request.AccountID,
|
|
LiveOrderId: liveOrderID,
|
|
})
|
|
if gerr != nil {
|
|
return transportEvent3(ev, gerr)
|
|
}
|
|
return evaluateLiveOrder(ev, step, gresp.GetError(), gresp.GetOrder(), nil)
|
|
|
|
case ActionGetLiveRiskPolicy:
|
|
rpresp, rperr := client.GetLiveRiskPolicy(stepCtx, &altv1.GetLiveRiskPolicyRequest{
|
|
AccountId: step.Request.AccountID,
|
|
})
|
|
if rperr != nil {
|
|
return transportEvent3(ev, rperr)
|
|
}
|
|
ev2, code := evaluateLiveRiskPolicy(ev, step, rpresp.GetError(), rpresp.GetPolicy())
|
|
return ev2, code, ""
|
|
|
|
case ActionGetLiveKillSwitch:
|
|
gksresp, gkserr := client.GetLiveKillSwitch(stepCtx, &altv1.GetLiveKillSwitchRequest{
|
|
AccountId: step.Request.AccountID,
|
|
})
|
|
if gkserr != nil {
|
|
return transportEvent3(ev, gkserr)
|
|
}
|
|
ev2, code := evaluateLiveKillSwitch(ev, step, gksresp.GetError(), gksresp.GetState())
|
|
return ev2, code, ""
|
|
|
|
case ActionSetLiveKillSwitch:
|
|
sksresp, skserr := client.SetLiveKillSwitch(stepCtx, &altv1.SetLiveKillSwitchRequest{
|
|
AccountId: step.Request.AccountID,
|
|
Halted: step.Request.Halted,
|
|
Reason: step.Request.KillSwitchReason,
|
|
})
|
|
if skserr != nil {
|
|
return transportEvent3(ev, skserr)
|
|
}
|
|
ev2, code := evaluateLiveKillSwitch(ev, step, sksresp.GetError(), sksresp.GetState())
|
|
return ev2, code, ""
|
|
|
|
case ActionSyncLiveAccount:
|
|
syncresp, syncerr := client.SyncLiveAccount(stepCtx, &altv1.SyncLiveAccountRequest{
|
|
AccountId: step.Request.AccountID,
|
|
})
|
|
if syncerr != nil {
|
|
return transportEvent3(ev, syncerr)
|
|
}
|
|
ev2, code := evaluateLiveAccount(ev, step, syncresp.GetError(), syncresp.GetSnapshot())
|
|
return ev2, code, ""
|
|
|
|
case ActionGetLiveAccountSnapshot:
|
|
getresp, geterr := client.GetLiveAccountSnapshot(stepCtx, &altv1.GetLiveAccountSnapshotRequest{
|
|
AccountId: step.Request.AccountID,
|
|
})
|
|
if geterr != nil {
|
|
return transportEvent3(ev, geterr)
|
|
}
|
|
ev2, code := evaluateLiveAccount(ev, step, getresp.GetError(), getresp.GetSnapshot())
|
|
return ev2, code, ""
|
|
|
|
case ActionListLiveAuditEvents:
|
|
auditresp, auditerr := client.ListLiveAuditEvents(stepCtx, &altv1.ListLiveAuditEventsRequest{
|
|
AccountId: step.Request.AccountID,
|
|
OrderId: step.Request.OrderID,
|
|
EventType: step.Request.EventType,
|
|
Limit: int32(step.Request.Limit),
|
|
})
|
|
if auditerr != nil {
|
|
return transportEvent3(ev, auditerr)
|
|
}
|
|
ev2, code := evaluateLiveAuditEvents(ev, step, auditresp.GetError(), auditresp.GetEvents())
|
|
return ev2, code, ""
|
|
|
|
default:
|
|
// Validation rejects unknown actions, so reaching here means a runner is
|
|
// missing for a registered action.
|
|
ev.Status = statusError
|
|
ev.ErrorCode = "unsupported_action"
|
|
ev.ErrorMessage = fmt.Sprintf("no runner for action %q", step.Action)
|
|
return ev, codeBadInput, ""
|
|
}
|
|
}
|
|
|
|
func transportEvent(ev StepEvent, err error) (StepEvent, int) {
|
|
ev.Status = statusTransportError
|
|
ev.Count = noCount
|
|
ev.ErrorMessage = err.Error()
|
|
return ev, codeTransport
|
|
}
|
|
|
|
func transportEvent3(ev StepEvent, err error) (StepEvent, int, string) {
|
|
ev.Status = statusTransportError
|
|
ev.Count = noCount
|
|
ev.ErrorMessage = err.Error()
|
|
return ev, codeTransport, ""
|
|
}
|
|
|
|
// evaluateHello checks a hello response against the step expectation. Hello has
|
|
// no typed ErrorInfo, so an expected "error" status is itself a mismatch.
|
|
func evaluateHello(ev StepEvent, step Step, resp *altv1.HelloResponse) (StepEvent, int) {
|
|
if step.Expect.Status == statusError {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = "hello cannot return a typed error but the step expected one"
|
|
return ev, codeAppError
|
|
}
|
|
if missing := missingCapabilities(step.Expect.Capabilities, resp.GetCapabilities()); missing != "" {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("missing expected capability %q", missing)
|
|
return ev, codeAppError
|
|
}
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
|
|
// evaluateMarket checks a market response. errInfo is the typed ErrorInfo (nil
|
|
// when the call succeeded) and count is the number of returned results.
|
|
func evaluateMarket(ev StepEvent, step Step, errInfo *altv1.ErrorInfo, count int) (StepEvent, int) {
|
|
expectsError := step.Expect.Status == statusError
|
|
|
|
if errInfo != nil {
|
|
ev.ErrorCode = errInfo.GetCode()
|
|
ev.ErrorMessage = errInfo.GetMessage()
|
|
if !expectsError {
|
|
ev.Status = statusError
|
|
return ev, codeAppError
|
|
}
|
|
if step.Expect.ErrorCode != "" && step.Expect.ErrorCode != errInfo.GetCode() {
|
|
ev.Status = statusMismatch
|
|
return ev, codeAppError
|
|
}
|
|
// Expected typed error arrived: the step passed.
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
|
|
// Successful response with data.
|
|
ev.Count = count
|
|
if expectsError {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = "expected a typed error but the request succeeded"
|
|
return ev, codeAppError
|
|
}
|
|
if step.Expect.MinCount != nil && count < *step.Expect.MinCount {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected at least %d results, got %d", *step.Expect.MinCount, count)
|
|
return ev, codeAppError
|
|
}
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
|
|
// evaluateImport checks an import response.
|
|
func evaluateImport(ev StepEvent, step Step, resp *altv1.ImportDailyBarsResponse) (StepEvent, int) {
|
|
expectsError := step.Expect.Status == statusError
|
|
errInfo := resp.GetError()
|
|
|
|
// Default counts to noCount (-1) so they don't print for error steps unless set.
|
|
ev.InstrumentCount = noCount
|
|
ev.BarCount = noCount
|
|
|
|
if errInfo != nil {
|
|
ev.ErrorCode = errInfo.GetCode()
|
|
ev.ErrorMessage = errInfo.GetMessage()
|
|
if !expectsError {
|
|
ev.Status = statusError
|
|
return ev, codeAppError
|
|
}
|
|
if step.Expect.ErrorCode != "" && step.Expect.ErrorCode != errInfo.GetCode() {
|
|
ev.Status = statusMismatch
|
|
return ev, codeAppError
|
|
}
|
|
// Expected typed error arrived: the step passed.
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
|
|
ev.Provider = resp.GetProvider()
|
|
ev.InstrumentCount = int(resp.GetInstrumentCount())
|
|
ev.BarCount = int(resp.GetBarCount())
|
|
|
|
if expectsError {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = "expected a typed error but the request succeeded"
|
|
return ev, codeAppError
|
|
}
|
|
if step.Expect.MinCount != nil && ev.BarCount < *step.Expect.MinCount {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected at least %d results, got %d", *step.Expect.MinCount, ev.BarCount)
|
|
return ev, codeAppError
|
|
}
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
|
|
// evaluateBacktest checks a backtest response against expectations.
|
|
func evaluateBacktest(ev StepEvent, step Step, errInfo *altv1.ErrorInfo, run *altv1.BacktestRun, result *altv1.BacktestResult, count int) (StepEvent, int, string) {
|
|
expectsError := step.Expect.Status == statusError
|
|
var runID string
|
|
if run != nil {
|
|
runID = run.GetId()
|
|
ev.RunID = run.GetId()
|
|
ev.RunStatus = strings.ToLower(strings.TrimPrefix(run.GetStatus().String(), "BACKTEST_RUN_STATUS_"))
|
|
}
|
|
if result != nil {
|
|
runID = result.GetRunId()
|
|
ev.RunID = result.GetRunId()
|
|
if result.GetSummary() != nil {
|
|
ev.StartingCash = result.GetSummary().GetStartingCash().GetAmount().GetValue()
|
|
ev.EndingEquity = result.GetSummary().GetEndingEquity().GetAmount().GetValue()
|
|
ev.TotalReturn = result.GetSummary().GetTotalReturn().GetValue()
|
|
ev.TradeCount = int(result.GetSummary().GetTradeCount())
|
|
}
|
|
}
|
|
|
|
if errInfo != nil {
|
|
ev.ErrorCode = errInfo.GetCode()
|
|
ev.ErrorMessage = errInfo.GetMessage()
|
|
if !expectsError {
|
|
ev.Status = statusError
|
|
return ev, codeAppError, runID
|
|
}
|
|
if step.Expect.ErrorCode != "" && step.Expect.ErrorCode != errInfo.GetCode() {
|
|
ev.Status = statusMismatch
|
|
return ev, codeAppError, runID
|
|
}
|
|
ev.Status = statusOK
|
|
return ev, codeOK, runID
|
|
}
|
|
|
|
if expectsError {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = "expected a typed error but the request succeeded"
|
|
return ev, codeAppError, runID
|
|
}
|
|
|
|
// Validate run status expectation if set
|
|
if step.Expect.RunStatus != "" && run != nil {
|
|
actualStatus := strings.ToLower(strings.TrimPrefix(run.GetStatus().String(), "BACKTEST_RUN_STATUS_"))
|
|
expectedStatus := strings.ToLower(step.Expect.RunStatus)
|
|
if actualStatus != expectedStatus {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected backtest run status %q, got %q", expectedStatus, actualStatus)
|
|
return ev, codeAppError, runID
|
|
}
|
|
}
|
|
|
|
if step.Expect.MinCount != nil && count < *step.Expect.MinCount {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected at least %d results, got %d", *step.Expect.MinCount, count)
|
|
return ev, codeAppError, runID
|
|
}
|
|
|
|
ev.Status = statusOK
|
|
if count >= 0 {
|
|
ev.Count = count
|
|
}
|
|
return ev, codeOK, runID
|
|
}
|
|
|
|
// evaluatePaper checks a paper trading response against expectations. errInfo is
|
|
// the typed ErrorInfo (nil when the call succeeded) and state carries the paper
|
|
// account snapshot rendered into stable text/JSONL fields.
|
|
func evaluatePaper(ev StepEvent, step Step, errInfo *altv1.ErrorInfo, state *altv1.PaperTradingState) (StepEvent, int) {
|
|
expectsError := step.Expect.Status == statusError
|
|
|
|
if errInfo != nil {
|
|
ev.ErrorCode = errInfo.GetCode()
|
|
ev.ErrorMessage = errInfo.GetMessage()
|
|
if !expectsError {
|
|
ev.Status = statusError
|
|
return ev, codeAppError
|
|
}
|
|
if step.Expect.ErrorCode != "" && step.Expect.ErrorCode != errInfo.GetCode() {
|
|
ev.Status = statusMismatch
|
|
return ev, codeAppError
|
|
}
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
|
|
if expectsError {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = "expected a typed error but the request succeeded"
|
|
return ev, codeAppError
|
|
}
|
|
|
|
if state != nil {
|
|
ev.AccountID = state.GetAccountId()
|
|
ev.Cash = state.GetCash().GetAmount().GetValue()
|
|
ev.PositionCount = len(state.GetPositions())
|
|
ev.FillCount = len(state.GetFills())
|
|
// Risk is derived from the worker-owned rejection summary: an empty list
|
|
// means every decided order cleared risk, a non-empty list means at least
|
|
// one order was blocked. The first reason is surfaced so blocked output
|
|
// carries a stable, human-readable cause without faking broker behaviour.
|
|
ev.Risk = "clear"
|
|
if rejections := state.GetRiskRejections(); len(rejections) > 0 {
|
|
ev.Risk = "blocked"
|
|
ev.RiskRejectedCount = len(rejections)
|
|
ev.RiskReason = rejections[0].GetReason()
|
|
}
|
|
if run := state.GetRun(); run != nil {
|
|
ev.RunID = run.GetId()
|
|
ev.RunStatus = strings.ToLower(strings.TrimPrefix(run.GetStatus().String(), "BACKTEST_RUN_STATUS_"))
|
|
}
|
|
// Summarise the equity curve so loop smoke output carries a stable
|
|
// point count and the latest equity value. An empty curve leaves both
|
|
// fields zero/empty so they are omitted from text and JSONL.
|
|
if curve := state.GetEquityCurve(); len(curve) > 0 {
|
|
ev.EquityPointCount = len(curve)
|
|
ev.LatestEquity = curve[len(curve)-1].GetEquity().GetAmount().GetValue()
|
|
}
|
|
}
|
|
|
|
if step.Expect.RunStatus != "" && state.GetRun() != nil {
|
|
actualStatus := strings.ToLower(strings.TrimPrefix(state.GetRun().GetStatus().String(), "BACKTEST_RUN_STATUS_"))
|
|
expectedStatus := strings.ToLower(step.Expect.RunStatus)
|
|
if actualStatus != expectedStatus {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected paper run status %q, got %q", expectedStatus, actualStatus)
|
|
return ev, codeAppError
|
|
}
|
|
}
|
|
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
|
|
// evaluatePaperOrder checks a paper order lifecycle response. order is the typed
|
|
// PaperOrder (nil on a typed error). The returned string is the order id, saved
|
|
// so a later cancel/fill step can interpolate {{steps.<id>.order.id}}. A risk or
|
|
// portfolio denial is a successful response carrying order_status=rejected, not a
|
|
// transport/typed error, so the operator sees the status transition.
|
|
func evaluatePaperOrder(ev StepEvent, step Step, errInfo *altv1.ErrorInfo, order *altv1.PaperOrder) (StepEvent, int, string) {
|
|
expectsError := step.Expect.Status == statusError
|
|
|
|
var orderID string
|
|
if order != nil {
|
|
orderID = order.GetOrderId()
|
|
ev.OrderID = order.GetOrderId()
|
|
ev.OrderStatus = order.GetStatus()
|
|
ev.OrderReason = order.GetReason()
|
|
if fill := order.GetFill(); fill != nil {
|
|
ev.OrderFillPrice = fill.GetPrice().GetAmount().GetValue()
|
|
ev.OrderFillCount = 1
|
|
}
|
|
}
|
|
|
|
if errInfo != nil {
|
|
ev.ErrorCode = errInfo.GetCode()
|
|
ev.ErrorMessage = errInfo.GetMessage()
|
|
if !expectsError {
|
|
ev.Status = statusError
|
|
return ev, codeAppError, orderID
|
|
}
|
|
if step.Expect.ErrorCode != "" && step.Expect.ErrorCode != errInfo.GetCode() {
|
|
ev.Status = statusMismatch
|
|
return ev, codeAppError, orderID
|
|
}
|
|
ev.Status = statusOK
|
|
return ev, codeOK, orderID
|
|
}
|
|
|
|
if expectsError {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = "expected a typed error but the request succeeded"
|
|
return ev, codeAppError, orderID
|
|
}
|
|
|
|
if step.Expect.OrderStatus != "" && order != nil {
|
|
expectedStatus := strings.ToLower(step.Expect.OrderStatus)
|
|
if order.GetStatus() != expectedStatus {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected order status %q, got %q", expectedStatus, order.GetStatus())
|
|
return ev, codeAppError, orderID
|
|
}
|
|
}
|
|
|
|
ev.Status = statusOK
|
|
return ev, codeOK, orderID
|
|
}
|
|
|
|
// evaluateLiveOrder checks a live order lifecycle response. order is the typed
|
|
// LiveOrder (nil on a typed error). riskDecision is non-nil only for
|
|
// submit_live_order steps when the risk guard blocked the submission.
|
|
// The returned string is the live order id, saved so a later cancel/get step
|
|
// can interpolate {{steps.<id>.live_order.id}}.
|
|
func evaluateLiveOrder(ev StepEvent, step Step, errInfo *altv1.ErrorInfo, order *altv1.LiveOrder, riskDecision *altv1.LiveRiskDecision) (StepEvent, int, string) {
|
|
expectsError := step.Expect.Status == statusError
|
|
|
|
var liveOrderID string
|
|
if order != nil {
|
|
liveOrderID = order.GetId()
|
|
ev.LiveOrderID = order.GetId()
|
|
ev.LiveOrderStatus = order.GetStatus()
|
|
ev.BrokerOrderID = order.GetBrokerId()
|
|
ev.BrokerStatus = order.GetBrokerStatus()
|
|
}
|
|
if riskDecision != nil {
|
|
ev.HasRiskDecision = true
|
|
ev.RiskDecisionAllow = riskDecision.GetAllowed()
|
|
ev.RiskDecisionReason = riskDecision.GetReason()
|
|
}
|
|
|
|
if errInfo != nil {
|
|
ev.ErrorCode = errInfo.GetCode()
|
|
ev.ErrorMessage = errInfo.GetMessage()
|
|
if !expectsError {
|
|
ev.Status = statusError
|
|
return ev, codeAppError, liveOrderID
|
|
}
|
|
if step.Expect.ErrorCode != "" && step.Expect.ErrorCode != errInfo.GetCode() {
|
|
ev.Status = statusMismatch
|
|
return ev, codeAppError, liveOrderID
|
|
}
|
|
ev.Status = statusOK
|
|
return ev, codeOK, liveOrderID
|
|
}
|
|
|
|
if expectsError {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = "expected a typed error but the request succeeded"
|
|
return ev, codeAppError, liveOrderID
|
|
}
|
|
|
|
if step.Expect.LiveOrderStatus != "" && order != nil {
|
|
if order.GetStatus() != step.Expect.LiveOrderStatus {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected live order status %q, got %q", step.Expect.LiveOrderStatus, order.GetStatus())
|
|
return ev, codeAppError, liveOrderID
|
|
}
|
|
}
|
|
|
|
ev.Status = statusOK
|
|
return ev, codeOK, liveOrderID
|
|
}
|
|
|
|
// missingCapabilities returns the first wanted capability absent from have, or
|
|
// "" when all are present.
|
|
func missingCapabilities(want, have []string) string {
|
|
if len(want) == 0 {
|
|
return ""
|
|
}
|
|
set := make(map[string]bool, len(have))
|
|
for _, c := range have {
|
|
set[c] = true
|
|
}
|
|
for _, c := range want {
|
|
if !set[c] {
|
|
return c
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func interpolate(val string, savedValues map[string]string) string {
|
|
for k, v := range savedValues {
|
|
val = strings.ReplaceAll(val, fmt.Sprintf("{{steps.%s.run.id}}", k), v)
|
|
val = strings.ReplaceAll(val, fmt.Sprintf("{{steps.%s.order.id}}", k), v)
|
|
val = strings.ReplaceAll(val, fmt.Sprintf("{{steps.%s.live_order.id}}", k), v)
|
|
}
|
|
return val
|
|
}
|
|
|
|
// evaluateLiveKillSwitch checks a get/set kill switch response against expectations.
|
|
func evaluateLiveKillSwitch(ev StepEvent, step Step, errInfo *altv1.ErrorInfo, state *altv1.LiveKillSwitchState) (StepEvent, int) {
|
|
expectsError := step.Expect.Status == statusError
|
|
|
|
if errInfo != nil {
|
|
ev.ErrorCode = errInfo.GetCode()
|
|
ev.ErrorMessage = errInfo.GetMessage()
|
|
if !expectsError {
|
|
ev.Status = statusError
|
|
return ev, codeAppError
|
|
}
|
|
if step.Expect.ErrorCode != "" && step.Expect.ErrorCode != errInfo.GetCode() {
|
|
ev.Status = statusMismatch
|
|
return ev, codeAppError
|
|
}
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
|
|
if expectsError {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = "expected a typed error but the request succeeded"
|
|
return ev, codeAppError
|
|
}
|
|
|
|
if state != nil {
|
|
ev.KillSwitchHalted = state.GetHalted()
|
|
ev.KillSwitchReason = state.GetReason()
|
|
}
|
|
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
|
|
// evaluateLiveRiskPolicy checks a get_live_risk_policy response against expectations.
|
|
func evaluateLiveRiskPolicy(ev StepEvent, step Step, errInfo *altv1.ErrorInfo, policy *altv1.LiveRiskPolicy) (StepEvent, int) {
|
|
expectsError := step.Expect.Status == statusError
|
|
|
|
if errInfo != nil {
|
|
ev.ErrorCode = errInfo.GetCode()
|
|
ev.ErrorMessage = errInfo.GetMessage()
|
|
if !expectsError {
|
|
ev.Status = statusError
|
|
return ev, codeAppError
|
|
}
|
|
if step.Expect.ErrorCode != "" && step.Expect.ErrorCode != errInfo.GetCode() {
|
|
ev.Status = statusMismatch
|
|
return ev, codeAppError
|
|
}
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
|
|
if expectsError {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = "expected a typed error but the request succeeded"
|
|
return ev, codeAppError
|
|
}
|
|
|
|
if policy != nil {
|
|
ev.MaxDailyOrders = int(policy.GetMaxDailyOrders())
|
|
ev.MaxOpenOrders = int(policy.GetMaxOpenOrders())
|
|
ev.AllowShortSelling = policy.GetAllowShortSelling()
|
|
}
|
|
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
|
|
// evaluateLiveAccount checks a sync/get_live_account_snapshot response against expectations.
|
|
func evaluateLiveAccount(ev StepEvent, step Step, errInfo *altv1.ErrorInfo, snap *altv1.LiveAccountSnapshot) (StepEvent, int) {
|
|
expectsError := step.Expect.Status == statusError
|
|
|
|
if errInfo != nil {
|
|
ev.ErrorCode = errInfo.GetCode()
|
|
ev.ErrorMessage = errInfo.GetMessage()
|
|
if !expectsError {
|
|
ev.Status = statusError
|
|
return ev, codeAppError
|
|
}
|
|
if step.Expect.ErrorCode != "" && step.Expect.ErrorCode != errInfo.GetCode() {
|
|
ev.Status = statusMismatch
|
|
return ev, codeAppError
|
|
}
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
|
|
if expectsError {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = "expected a typed error but the request succeeded"
|
|
return ev, codeAppError
|
|
}
|
|
|
|
if snap != nil {
|
|
ev.LiveAccountID = snap.GetAccountId()
|
|
ev.LiveBroker = snap.GetBroker()
|
|
ev.CashCount = len(snap.GetCash())
|
|
ev.LivePositionCount = len(snap.GetPositions())
|
|
ev.SyncedAtUnixMs = snap.GetSyncedAtUnixMs()
|
|
ev.LiveStale = snap.GetStale()
|
|
}
|
|
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
|
|
// evaluateLiveAuditEvents checks a list_live_audit_events response against expectations.
|
|
func evaluateLiveAuditEvents(ev StepEvent, step Step, errInfo *altv1.ErrorInfo, events []*altv1.LiveAuditEvent) (StepEvent, int) {
|
|
expectsError := step.Expect.Status == statusError
|
|
|
|
if errInfo != nil {
|
|
ev.ErrorCode = errInfo.GetCode()
|
|
ev.ErrorMessage = errInfo.GetMessage()
|
|
if !expectsError {
|
|
ev.Status = statusError
|
|
return ev, codeAppError
|
|
}
|
|
if step.Expect.ErrorCode != "" && step.Expect.ErrorCode != errInfo.GetCode() {
|
|
ev.Status = statusMismatch
|
|
return ev, codeAppError
|
|
}
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
|
|
if expectsError {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = "expected a typed error but the request succeeded"
|
|
return ev, codeAppError
|
|
}
|
|
|
|
ev.AuditEventCount = len(events)
|
|
ev.Count = len(events)
|
|
|
|
if step.Expect.MinCount != nil && len(events) < *step.Expect.MinCount {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected at least %d audit events, got %d", *step.Expect.MinCount, len(events))
|
|
return ev, codeAppError
|
|
}
|
|
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
|
|
func interpolateSlice(slice []string, savedValues map[string]string) []string {
|
|
if len(slice) == 0 {
|
|
return nil
|
|
}
|
|
res := make([]string, len(slice))
|
|
for i, v := range slice {
|
|
res[i] = interpolate(v, savedValues)
|
|
}
|
|
return res
|
|
}
|
|
|
|
func runCollectionFreshness(ctx context.Context, client *APIClient, scenario string, step Step) (StepEvent, int, string) {
|
|
ev := StepEvent{
|
|
Scenario: scenario,
|
|
Step: step.ID,
|
|
Action: string(step.Action),
|
|
Count: noCount,
|
|
InstrumentCount: noCount,
|
|
BarCount: noCount,
|
|
MissingCount: noCount,
|
|
GapCount: noCount,
|
|
DuplicateCount: noCount,
|
|
ProviderDelayDays: noCount,
|
|
}
|
|
|
|
resp, err := client.ListInstruments(ctx, &altv1.ListInstrumentsRequest{
|
|
Market: marketByName[step.Request.Market],
|
|
Provider: step.Request.Provider,
|
|
})
|
|
if err != nil {
|
|
ev.Status = statusTransportError
|
|
ev.ErrorMessage = err.Error()
|
|
return ev, codeTransport, ""
|
|
}
|
|
|
|
if resp.GetError() != nil {
|
|
return evaluateFreshnessError(ev, step, resp.GetError())
|
|
}
|
|
|
|
instBySymbol := make(map[string]*altv1.Instrument)
|
|
for _, inst := range resp.GetInstruments() {
|
|
if inst.GetSymbol() != "" {
|
|
instBySymbol[inst.GetSymbol()] = inst
|
|
}
|
|
if inst.GetProviderSymbols() != nil && step.Request.Provider != "" {
|
|
if psym, ok := inst.GetProviderSymbols()[step.Request.Provider]; ok && psym != "" {
|
|
instBySymbol[psym] = inst
|
|
}
|
|
}
|
|
}
|
|
|
|
var missingSymbols []string
|
|
var matchedInstruments []*altv1.Instrument
|
|
matchedSymbolMap := make(map[*altv1.Instrument]string)
|
|
|
|
for _, sym := range step.Request.Symbols {
|
|
if inst, ok := instBySymbol[sym]; ok {
|
|
matchedInstruments = append(matchedInstruments, inst)
|
|
matchedSymbolMap[inst] = sym
|
|
} else {
|
|
missingSymbols = append(missingSymbols, sym)
|
|
}
|
|
}
|
|
|
|
var latestUnixMs int64 = 0
|
|
symbolDateCounts := make(map[string]map[string]int)
|
|
|
|
for _, inst := range matchedInstruments {
|
|
barsResp, err := client.ListBars(ctx, &altv1.ListBarsRequest{
|
|
InstrumentId: inst.GetId(),
|
|
Timeframe: timeframeByName[step.Request.Timeframe],
|
|
FromUnixMs: step.Request.FromUnixMs,
|
|
ToUnixMs: step.Request.ToUnixMs,
|
|
})
|
|
if err != nil {
|
|
ev.Status = statusTransportError
|
|
ev.ErrorMessage = err.Error()
|
|
return ev, codeTransport, ""
|
|
}
|
|
|
|
if barsResp.GetError() != nil {
|
|
return evaluateFreshnessError(ev, step, barsResp.GetError())
|
|
}
|
|
|
|
instRequestSymbol := matchedSymbolMap[inst]
|
|
if instRequestSymbol == "" {
|
|
instRequestSymbol = inst.GetSymbol()
|
|
}
|
|
|
|
dateCounts := make(map[string]int)
|
|
if len(barsResp.GetBars()) == 0 {
|
|
missingSymbols = append(missingSymbols, instRequestSymbol)
|
|
} else {
|
|
for _, bar := range barsResp.GetBars() {
|
|
t := time.UnixMilli(bar.GetTimestampUnixMs()).UTC()
|
|
d := t.Format("20060102")
|
|
dateCounts[d]++
|
|
if bar.GetTimestampUnixMs() > latestUnixMs {
|
|
latestUnixMs = bar.GetTimestampUnixMs()
|
|
}
|
|
}
|
|
}
|
|
symbolDateCounts[instRequestSymbol] = dateCounts
|
|
}
|
|
|
|
ev.Universe = step.Request.Universe
|
|
ev.MissingSymbols = missingSymbols
|
|
ev.MissingCount = len(missingSymbols)
|
|
|
|
var latest string
|
|
if latestUnixMs > 0 {
|
|
t := time.UnixMilli(latestUnixMs).UTC()
|
|
latest = t.Format("20060102")
|
|
ev.LatestYYYYMMDD = latest
|
|
}
|
|
|
|
if len(missingSymbols) == 0 {
|
|
ev.FreshnessStatus = "fresh"
|
|
} else {
|
|
ev.FreshnessStatus = "missing"
|
|
}
|
|
|
|
if len(step.Request.ExpectedYYYYMMDD) > 0 {
|
|
gapCount := 0
|
|
duplicateCount := 0
|
|
gapDatesMap := make(map[string]bool)
|
|
dupDatesMap := make(map[string]bool)
|
|
|
|
for _, d := range step.Request.ExpectedYYYYMMDD {
|
|
if latest != "" && d > latest {
|
|
continue
|
|
}
|
|
for _, sym := range step.Request.Symbols {
|
|
dateCounts, matched := symbolDateCounts[sym]
|
|
if !matched {
|
|
gapCount++
|
|
gapDatesMap[d] = true
|
|
} else {
|
|
cnt := dateCounts[d]
|
|
if cnt == 0 {
|
|
gapCount++
|
|
gapDatesMap[d] = true
|
|
} else if cnt >= 2 {
|
|
duplicateCount++
|
|
dupDatesMap[d] = true
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
gapDates := sortMapKeys(gapDatesMap)
|
|
dupDates := sortMapKeys(dupDatesMap)
|
|
|
|
ev.GapCount = gapCount
|
|
ev.GapYYYYMMDD = gapDates
|
|
ev.DuplicateCount = duplicateCount
|
|
ev.DuplicateYYYYMMDD = dupDates
|
|
|
|
expectedLast := step.Request.ExpectedYYYYMMDD[len(step.Request.ExpectedYYYYMMDD)-1]
|
|
var delayDays int
|
|
if latest == "" {
|
|
firstDate := step.Request.ExpectedYYYYMMDD[0]
|
|
tFirst, _ := time.ParseInLocation("20060102", firstDate, time.UTC)
|
|
tLast, _ := time.ParseInLocation("20060102", expectedLast, time.UTC)
|
|
diff := int(tLast.Sub(tFirst) / (24 * time.Hour))
|
|
if diff < 0 {
|
|
diff = 0
|
|
}
|
|
delayDays = diff
|
|
} else {
|
|
latestIdx := -1
|
|
for idx, d := range step.Request.ExpectedYYYYMMDD {
|
|
if d == latest {
|
|
latestIdx = idx
|
|
break
|
|
}
|
|
}
|
|
if latestIdx != -1 {
|
|
delayDays = (len(step.Request.ExpectedYYYYMMDD) - 1) - latestIdx
|
|
} else {
|
|
tLast, _ := time.ParseInLocation("20060102", expectedLast, time.UTC)
|
|
tLatest, err := time.ParseInLocation("20060102", latest, time.UTC)
|
|
if err != nil {
|
|
delayDays = 0
|
|
} else {
|
|
diff := int(tLast.Sub(tLatest) / (24 * time.Hour))
|
|
if diff < 0 {
|
|
diff = 0
|
|
}
|
|
delayDays = diff
|
|
}
|
|
}
|
|
}
|
|
ev.ProviderDelayDays = delayDays
|
|
|
|
hasGap := gapCount > 0
|
|
hasDup := duplicateCount > 0
|
|
hasDelay := delayDays > 0
|
|
|
|
activeConditions := 0
|
|
if hasGap {
|
|
activeConditions++
|
|
}
|
|
if hasDup {
|
|
activeConditions++
|
|
}
|
|
if hasDelay {
|
|
activeConditions++
|
|
}
|
|
|
|
if activeConditions >= 2 {
|
|
ev.GapStatus = "mixed"
|
|
} else if hasGap {
|
|
ev.GapStatus = "gap"
|
|
} else if hasDup {
|
|
ev.GapStatus = "duplicate"
|
|
} else if hasDelay {
|
|
ev.GapStatus = "delayed"
|
|
} else {
|
|
ev.GapStatus = "clean"
|
|
}
|
|
}
|
|
|
|
expectsError := step.Expect.Status == statusError
|
|
if expectsError {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = "expected a typed error but the freshness check succeeded"
|
|
return ev, codeAppError, ""
|
|
}
|
|
|
|
if step.Expect.FreshnessStatus != "" && ev.FreshnessStatus != step.Expect.FreshnessStatus {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected freshness_status %q, got %q", step.Expect.FreshnessStatus, ev.FreshnessStatus)
|
|
return ev, codeAppError, ""
|
|
}
|
|
if step.Expect.MissingCount != nil && ev.MissingCount != *step.Expect.MissingCount {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected missing_count %d, got %d", *step.Expect.MissingCount, ev.MissingCount)
|
|
return ev, codeAppError, ""
|
|
}
|
|
if step.Expect.LatestYYYYMMDD != "" && ev.LatestYYYYMMDD != step.Expect.LatestYYYYMMDD {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected latest_yyyymmdd %q, got %q", step.Expect.LatestYYYYMMDD, ev.LatestYYYYMMDD)
|
|
return ev, codeAppError, ""
|
|
}
|
|
|
|
if step.Expect.GapStatus != "" && ev.GapStatus != step.Expect.GapStatus {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected gap_status %q, got %q", step.Expect.GapStatus, ev.GapStatus)
|
|
return ev, codeAppError, ""
|
|
}
|
|
if step.Expect.GapCount != nil && ev.GapCount != *step.Expect.GapCount {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected gap_count %d, got %d", *step.Expect.GapCount, ev.GapCount)
|
|
return ev, codeAppError, ""
|
|
}
|
|
if step.Expect.DuplicateCount != nil && ev.DuplicateCount != *step.Expect.DuplicateCount {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected duplicate_count %d, got %d", *step.Expect.DuplicateCount, ev.DuplicateCount)
|
|
return ev, codeAppError, ""
|
|
}
|
|
if step.Expect.ProviderDelayDays != nil && ev.ProviderDelayDays != *step.Expect.ProviderDelayDays {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected provider_delay_days %d, got %d", *step.Expect.ProviderDelayDays, ev.ProviderDelayDays)
|
|
return ev, codeAppError, ""
|
|
}
|
|
|
|
ev.Status = statusOK
|
|
return ev, codeOK, ""
|
|
}
|
|
|
|
func evaluateFreshnessError(ev StepEvent, step Step, errInfo *altv1.ErrorInfo) (StepEvent, int, string) {
|
|
expectsError := step.Expect.Status == statusError
|
|
ev.FreshnessStatus = "error"
|
|
ev.ErrorCode = errInfo.GetCode()
|
|
ev.ErrorMessage = errInfo.GetMessage()
|
|
|
|
if len(step.Request.ExpectedYYYYMMDD) > 0 {
|
|
ev.GapStatus = "error"
|
|
}
|
|
|
|
if !expectsError {
|
|
ev.Status = statusError
|
|
return ev, codeAppError, ""
|
|
}
|
|
if step.Expect.ErrorCode != "" && step.Expect.ErrorCode != errInfo.GetCode() {
|
|
ev.Status = statusMismatch
|
|
return ev, codeAppError, ""
|
|
}
|
|
if step.Expect.FreshnessStatus != "" && step.Expect.FreshnessStatus != "error" {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected freshness_status %q, got %q", step.Expect.FreshnessStatus, "error")
|
|
return ev, codeAppError, ""
|
|
}
|
|
|
|
if step.Expect.GapStatus != "" && step.Expect.GapStatus != "error" {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expected gap_status %q, got %q", step.Expect.GapStatus, "error")
|
|
return ev, codeAppError, ""
|
|
}
|
|
|
|
ev.Status = statusOK
|
|
return ev, codeOK, ""
|
|
}
|
|
|
|
// runMatrixBatch starts and polls each matrix run sequentially. Terminal
|
|
// succeeded/failed/canceled statuses are treated as lifecycle evidence (codeOK).
|
|
// A transport failure breaks the batch; a typed start error continues to the next run.
|
|
func runMatrixBatch(ctx context.Context, client *APIClient, scenario string, runs []MatrixRun, timeout time.Duration, out *Writer) int {
|
|
exitCode := codeOK
|
|
passed := 0
|
|
|
|
for _, run := range runs {
|
|
startCtx, cancelStart := context.WithTimeout(ctx, timeout)
|
|
startEv := StepEvent{
|
|
Scenario: scenario,
|
|
Step: run.ID,
|
|
Action: string(ActionStartBacktest),
|
|
Count: noCount,
|
|
InstrumentCount: noCount,
|
|
BarCount: noCount,
|
|
MissingCount: noCount,
|
|
MatrixRunID: run.ID,
|
|
Universe: run.Universe,
|
|
StrategyID: run.StrategyID,
|
|
Timeframe: run.Timeframe,
|
|
PeriodID: run.PeriodID,
|
|
}
|
|
|
|
startResp, err := client.StartBacktest(startCtx, &altv1.StartBacktestRequest{
|
|
Spec: &altv1.BacktestRunSpec{
|
|
StrategyId: run.StrategyID,
|
|
Market: marketByName[run.Market],
|
|
Timeframe: timeframeByName[run.Timeframe],
|
|
FromUnixMs: run.FromUnixMs,
|
|
ToUnixMs: run.ToUnixMs,
|
|
Selector: &altv1.BacktestInputSelector{Symbols: run.Symbols},
|
|
},
|
|
})
|
|
cancelStart()
|
|
|
|
if err != nil {
|
|
startEv.Status = statusTransportError
|
|
startEv.ErrorMessage = err.Error()
|
|
out.WriteStep(startEv)
|
|
if codeTransport > exitCode {
|
|
exitCode = codeTransport
|
|
}
|
|
break
|
|
}
|
|
|
|
if ei := startResp.GetError(); ei != nil {
|
|
startEv.Status = statusError
|
|
startEv.ErrorCode = ei.GetCode()
|
|
startEv.ErrorMessage = ei.GetMessage()
|
|
out.WriteStep(startEv)
|
|
if codeAppError > exitCode {
|
|
exitCode = codeAppError
|
|
}
|
|
continue
|
|
}
|
|
|
|
if r := startResp.GetRun(); r != nil {
|
|
startEv.RunID = r.GetId()
|
|
startEv.RunStatus = strings.ToLower(strings.TrimPrefix(r.GetStatus().String(), "BACKTEST_RUN_STATUS_"))
|
|
}
|
|
startEv.Status = statusOK
|
|
out.WriteStep(startEv)
|
|
|
|
runID := ""
|
|
if r := startResp.GetRun(); r != nil {
|
|
runID = r.GetId()
|
|
}
|
|
if runID == "" {
|
|
// No run ID to poll; treat as passed.
|
|
passed++
|
|
continue
|
|
}
|
|
|
|
pollEv, code := pollMatrixRun(ctx, client, scenario, run, runID, timeout)
|
|
out.WriteStep(pollEv)
|
|
if code == codeOK {
|
|
passed++
|
|
} else if code > exitCode {
|
|
exitCode = code
|
|
}
|
|
if code == codeTransport {
|
|
break
|
|
}
|
|
}
|
|
|
|
out.WriteSummary(RunSummary{
|
|
Scenario: scenario,
|
|
Status: summaryStatus(exitCode),
|
|
Steps: len(runs),
|
|
Passed: passed,
|
|
ExitCode: exitCode,
|
|
})
|
|
return exitCode
|
|
}
|
|
|
|
// pollMatrixRun polls a backtest run until it reaches a terminal status.
|
|
// Terminal succeeded/failed/canceled are returned as codeOK (lifecycle evidence).
|
|
// Polling timeout returns codeTransport.
|
|
func pollMatrixRun(ctx context.Context, client *APIClient, scenario string, run MatrixRun, runID string, timeout time.Duration) (StepEvent, int) {
|
|
ev := StepEvent{
|
|
Scenario: scenario,
|
|
Step: run.ID,
|
|
Action: string(ActionPollBacktestRun),
|
|
Count: noCount,
|
|
InstrumentCount: noCount,
|
|
BarCount: noCount,
|
|
MissingCount: noCount,
|
|
MatrixRunID: run.ID,
|
|
Universe: run.Universe,
|
|
StrategyID: run.StrategyID,
|
|
Timeframe: run.Timeframe,
|
|
PeriodID: run.PeriodID,
|
|
RunID: runID,
|
|
}
|
|
|
|
const interval = 500 * time.Millisecond
|
|
|
|
pollCtx, pollCancel := context.WithTimeout(ctx, timeout)
|
|
defer pollCancel()
|
|
|
|
var lastErr error
|
|
lastStatus := "running"
|
|
|
|
for {
|
|
select {
|
|
case <-pollCtx.Done():
|
|
ev.Status = statusTransportError
|
|
ev.RunStatus = lastStatus
|
|
if lastErr != nil {
|
|
ev.ErrorMessage = fmt.Sprintf("polling timed out: %v (last error: %v)", pollCtx.Err(), lastErr)
|
|
} else {
|
|
ev.ErrorMessage = fmt.Sprintf("polling timed out: %v", pollCtx.Err())
|
|
}
|
|
return ev, codeTransport
|
|
default:
|
|
}
|
|
|
|
resp, err := client.GetBacktestRun(pollCtx, &altv1.GetBacktestRunRequest{RunId: runID})
|
|
if err != nil {
|
|
lastErr = err
|
|
select {
|
|
case <-pollCtx.Done():
|
|
case <-time.After(interval):
|
|
}
|
|
continue
|
|
}
|
|
|
|
if ei := resp.GetError(); ei != nil {
|
|
ev.ErrorCode = ei.GetCode()
|
|
ev.ErrorMessage = ei.GetMessage()
|
|
ev.Status = statusError
|
|
return ev, codeAppError
|
|
}
|
|
|
|
if r := resp.GetRun(); r != nil {
|
|
statusStr := strings.ToLower(strings.TrimPrefix(r.GetStatus().String(), "BACKTEST_RUN_STATUS_"))
|
|
if statusStr != "" && statusStr != "unspecified" {
|
|
lastStatus = statusStr
|
|
}
|
|
if statusStr == "succeeded" || statusStr == "failed" || statusStr == "canceled" {
|
|
ev.RunStatus = statusStr
|
|
if run.ExpectedRunStatus != "" && statusStr != run.ExpectedRunStatus {
|
|
ev.Status = statusMismatch
|
|
ev.ErrorMessage = fmt.Sprintf("expectation mismatch: expected %s but got %s", run.ExpectedRunStatus, statusStr)
|
|
return ev, codeAppError
|
|
}
|
|
ev.Status = statusOK
|
|
return ev, codeOK
|
|
}
|
|
}
|
|
|
|
select {
|
|
case <-pollCtx.Done():
|
|
case <-time.After(interval):
|
|
}
|
|
}
|
|
}
|
|
|
|
func sortMapKeys(m map[string]bool) []string {
|
|
keys := make([]string, 0, len(m))
|
|
for k := range m {
|
|
keys = append(keys, k)
|
|
}
|
|
sort.Strings(keys)
|
|
return keys
|
|
}
|