toki
1c11a17eff
- Implement backtest workflow validation for headless operator mode - Add handoff evidence tracking for backtest runs - Update operator client, runner, scenario, and output modules - Add test files for handoff, backtest runner, and error handling - Add test data for backtest scenarios and validation
533 lines
16 KiB
Go
533 lines
16 KiB
Go
package operator
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"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,
|
|
}
|
|
|
|
// 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,
|
|
ErrorMessage: err.Error(),
|
|
})
|
|
out.WriteSummary(RunSummary{Scenario: sc.Name, Status: summaryTransport, Steps: len(sc.Steps), Passed: 0, ExitCode: codeTransport})
|
|
return codeTransport
|
|
}
|
|
defer client.Close()
|
|
|
|
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}
|
|
|
|
switch step.Action {
|
|
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 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,
|
|
},
|
|
})
|
|
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):
|
|
}
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
// 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 {
|
|
placeholder := fmt.Sprintf("{{steps.%s.run.id}}", k)
|
|
val = strings.ReplaceAll(val, placeholder, v)
|
|
}
|
|
return val
|
|
}
|
|
|
|
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
|
|
}
|