- 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
365 lines
9.5 KiB
Go
365 lines
9.5 KiB
Go
package operator
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
altv1 "git.toki-labs.com/toki/alt/packages/contracts/gen/go/alt/v1"
|
|
)
|
|
|
|
// monthlyAggregateResp is the deterministic canned response the fake API returns
|
|
// for the monthly aggregation evidence: one instrument, 60 source daily bars
|
|
// aggregated into 3 monthly bars under a single rule id, matching SDD S04.
|
|
func monthlyAggregateResp() *altv1.AggregateMonthlyBarsResponse {
|
|
return &altv1.AggregateMonthlyBarsResponse{
|
|
Provider: "kis",
|
|
InstrumentCount: 1,
|
|
SourceDailyBarCount: 60,
|
|
MonthlyBarCount: 3,
|
|
Provenance: []*altv1.MonthlyProvenance{
|
|
{
|
|
InstrumentId: "KRX:005930",
|
|
SourceDailyBarCount: 60,
|
|
MonthlyBarCount: 3,
|
|
SourceStartYyyymmdd: "20240101",
|
|
SourceEndYyyymmdd: "20240331",
|
|
AggregationRuleId: "ohlcv-month-v1",
|
|
SourceTimeframe: altv1.Timeframe_TIMEFRAME_DAILY,
|
|
TargetTimeframe: altv1.Timeframe_TIMEFRAME_MONTHLY,
|
|
},
|
|
},
|
|
}
|
|
}
|
|
|
|
// TestRunScenarioAggregateMonthlyBars proves the operator maps a monthly
|
|
// aggregation scenario onto the AggregateMonthlyBars API call, forwards the
|
|
// shared market vocabulary, and surfaces the monthly counts plus provenance
|
|
// summary fields the SDD S04 evidence relies on.
|
|
func TestRunScenarioAggregateMonthlyBars(t *testing.T) {
|
|
api := &fakeAPI{aggregateMonthlyResp: monthlyAggregateResp()}
|
|
url := startFakeAPIServer(t, api)
|
|
sc := &Scenario{
|
|
Name: "monthly_bars_aggregation",
|
|
Steps: []Step{{
|
|
ID: "aggregate_kr_monthly",
|
|
Action: ActionAggregateMonthlyBars,
|
|
Request: Request{
|
|
Provider: "kis",
|
|
SelectorKind: "watchlist",
|
|
Market: "kr",
|
|
Venue: "krx",
|
|
Name: "kr-core",
|
|
Symbols: []string{"005930"},
|
|
FromYYYYMMDD: "20240101",
|
|
ToYYYYMMDD: "20240331",
|
|
},
|
|
Expect: Expect{Status: "ok", MinCount: intPtr(3)},
|
|
}},
|
|
}
|
|
if err := sc.Validate(); err != nil {
|
|
t.Fatalf("scenario validate: %v", err)
|
|
}
|
|
|
|
out, code := runScenario(t, sc, url)
|
|
if code != codeOK {
|
|
t.Fatalf("exit code = %d, want 0 (out=%q)", code, out)
|
|
}
|
|
for _, want := range []string{
|
|
"scenario=monthly_bars_aggregation",
|
|
"step=aggregate_kr_monthly",
|
|
"action=aggregate_monthly_bars",
|
|
"status=ok",
|
|
"provider=kis",
|
|
"instrument_count=1",
|
|
"source_daily_bar_count=60",
|
|
"monthly_bar_count=3",
|
|
"provenance_count=1",
|
|
"aggregation_rule_id=ohlcv-month-v1",
|
|
} {
|
|
if !strings.Contains(out, want) {
|
|
t.Errorf("output %q missing %q", out, want)
|
|
}
|
|
}
|
|
|
|
req := api.lastAggregateMonthlyReq()
|
|
if req == nil {
|
|
t.Fatal("server did not receive an aggregate_monthly_bars request")
|
|
}
|
|
if req.GetProvider() != "kis" || req.GetSelectorKind() != "watchlist" {
|
|
t.Errorf("request provider/selector = %q/%q, want kis/watchlist", req.GetProvider(), req.GetSelectorKind())
|
|
}
|
|
if req.GetMarket() != altv1.Market_MARKET_KR {
|
|
t.Errorf("request market = %v, want MARKET_KR", req.GetMarket())
|
|
}
|
|
if req.GetVenue() != altv1.Venue_VENUE_KRX {
|
|
t.Errorf("request venue = %v, want VENUE_KRX", req.GetVenue())
|
|
}
|
|
if !equalStrings(req.GetSymbols(), []string{"005930"}) {
|
|
t.Errorf("request symbols = %v, want [005930]", req.GetSymbols())
|
|
}
|
|
if req.GetFromYyyymmdd() != "20240101" || req.GetToYyyymmdd() != "20240331" {
|
|
t.Errorf("request range = %q..%q, want 20240101..20240331", req.GetFromYyyymmdd(), req.GetToYyyymmdd())
|
|
}
|
|
}
|
|
|
|
// TestRunScenarioAggregateMonthlyBarsTypedError proves a typed aggregation error
|
|
// is surfaced (not degraded to a transport error) and that the monthly count
|
|
// fields are omitted from the error output so the evidence stays clean.
|
|
func TestRunScenarioAggregateMonthlyBarsTypedError(t *testing.T) {
|
|
url := startFakeAPIServer(t, &fakeAPI{
|
|
aggregateMonthlyResp: &altv1.AggregateMonthlyBarsResponse{
|
|
Error: &altv1.ErrorInfo{Code: "invalid_request", Message: "no source daily bars"},
|
|
},
|
|
})
|
|
sc := &Scenario{
|
|
Name: "monthly_typed_error",
|
|
Steps: []Step{{
|
|
ID: "aggregate1",
|
|
Action: ActionAggregateMonthlyBars,
|
|
Request: Request{
|
|
Provider: "kis",
|
|
SelectorKind: "watchlist",
|
|
Market: "kr",
|
|
Symbols: []string{"005930"},
|
|
FromYYYYMMDD: "20240101",
|
|
ToYYYYMMDD: "20240331",
|
|
},
|
|
Expect: Expect{Status: "error", ErrorCode: "invalid_request"},
|
|
}},
|
|
}
|
|
|
|
out, code := runScenario(t, sc, url)
|
|
if code != codeOK {
|
|
t.Fatalf("exit code = %d, want 0 (out=%q)", code, out)
|
|
}
|
|
if !strings.Contains(out, "error.code=invalid_request") {
|
|
t.Errorf("output %q missing error.code=invalid_request", out)
|
|
}
|
|
if strings.Contains(out, "monthly_bar_count") {
|
|
t.Errorf("output %q should not contain monthly_bar_count on an error step", out)
|
|
}
|
|
}
|
|
|
|
// TestValidateAggregateMonthlyBars covers the dry-run validation contract: a
|
|
// well-formed scenario parses, and each missing/invalid field is rejected before
|
|
// any socket is opened.
|
|
func TestValidateAggregateMonthlyBars(t *testing.T) {
|
|
valid := `
|
|
name: valid_monthly
|
|
timeout: 5s
|
|
steps:
|
|
- id: step1
|
|
action: aggregate_monthly_bars
|
|
request:
|
|
provider: kis
|
|
selector_kind: watchlist
|
|
market: kr
|
|
venue: krx
|
|
symbols: ["005930"]
|
|
from_yyyymmdd: "20240101"
|
|
to_yyyymmdd: "20240331"
|
|
expect:
|
|
status: ok
|
|
`
|
|
sc, err := ParseScenario([]byte(valid))
|
|
if err != nil {
|
|
t.Fatalf("unexpected error parsing valid monthly scenario: %v", err)
|
|
}
|
|
if sc.Steps[0].Action != ActionAggregateMonthlyBars {
|
|
t.Fatalf("action = %q, want %q", sc.Steps[0].Action, ActionAggregateMonthlyBars)
|
|
}
|
|
|
|
cases := []struct {
|
|
name string
|
|
yaml string
|
|
want string
|
|
}{
|
|
{
|
|
name: "missing provider",
|
|
yaml: `
|
|
name: m
|
|
steps:
|
|
- id: s1
|
|
action: aggregate_monthly_bars
|
|
request:
|
|
selector_kind: watchlist
|
|
market: kr
|
|
symbols: ["005930"]
|
|
from_yyyymmdd: "20240101"
|
|
to_yyyymmdd: "20240331"
|
|
`,
|
|
want: "requires request.provider",
|
|
},
|
|
{
|
|
name: "empty symbols",
|
|
yaml: `
|
|
name: m
|
|
steps:
|
|
- id: s1
|
|
action: aggregate_monthly_bars
|
|
request:
|
|
provider: kis
|
|
selector_kind: watchlist
|
|
market: kr
|
|
symbols: []
|
|
from_yyyymmdd: "20240101"
|
|
to_yyyymmdd: "20240331"
|
|
`,
|
|
want: "requires request.symbols",
|
|
},
|
|
{
|
|
name: "blank symbol",
|
|
yaml: `
|
|
name: m
|
|
steps:
|
|
- id: s1
|
|
action: aggregate_monthly_bars
|
|
request:
|
|
provider: kis
|
|
selector_kind: watchlist
|
|
market: kr
|
|
symbols: [" "]
|
|
from_yyyymmdd: "20240101"
|
|
to_yyyymmdd: "20240331"
|
|
`,
|
|
want: "cannot contain empty or blank values",
|
|
},
|
|
{
|
|
name: "invalid market",
|
|
yaml: `
|
|
name: m
|
|
steps:
|
|
- id: s1
|
|
action: aggregate_monthly_bars
|
|
request:
|
|
provider: kis
|
|
selector_kind: watchlist
|
|
market: jp
|
|
symbols: ["005930"]
|
|
from_yyyymmdd: "20240101"
|
|
to_yyyymmdd: "20240331"
|
|
`,
|
|
want: "unsupported market",
|
|
},
|
|
{
|
|
name: "invalid venue",
|
|
yaml: `
|
|
name: m
|
|
steps:
|
|
- id: s1
|
|
action: aggregate_monthly_bars
|
|
request:
|
|
provider: kis
|
|
selector_kind: watchlist
|
|
market: kr
|
|
venue: lse
|
|
symbols: ["005930"]
|
|
from_yyyymmdd: "20240101"
|
|
to_yyyymmdd: "20240331"
|
|
`,
|
|
want: "unsupported venue",
|
|
},
|
|
{
|
|
name: "missing from date",
|
|
yaml: `
|
|
name: m
|
|
steps:
|
|
- id: s1
|
|
action: aggregate_monthly_bars
|
|
request:
|
|
provider: kis
|
|
selector_kind: watchlist
|
|
market: kr
|
|
symbols: ["005930"]
|
|
to_yyyymmdd: "20240331"
|
|
`,
|
|
want: "requires request.from_yyyymmdd",
|
|
},
|
|
{
|
|
name: "bad date format",
|
|
yaml: `
|
|
name: m
|
|
steps:
|
|
- id: s1
|
|
action: aggregate_monthly_bars
|
|
request:
|
|
provider: kis
|
|
selector_kind: watchlist
|
|
market: kr
|
|
symbols: ["005930"]
|
|
from_yyyymmdd: "2024-01-01"
|
|
to_yyyymmdd: "20240331"
|
|
`,
|
|
want: "must be in YYYYMMDD format",
|
|
},
|
|
{
|
|
name: "from after to",
|
|
yaml: `
|
|
name: m
|
|
steps:
|
|
- id: s1
|
|
action: aggregate_monthly_bars
|
|
request:
|
|
provider: kis
|
|
selector_kind: watchlist
|
|
market: kr
|
|
symbols: ["005930"]
|
|
from_yyyymmdd: "20240401"
|
|
to_yyyymmdd: "20240331"
|
|
`,
|
|
want: "cannot be after",
|
|
},
|
|
}
|
|
|
|
for _, tc := range cases {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
_, err := ParseScenario([]byte(tc.yaml))
|
|
if err == nil {
|
|
t.Fatalf("expected validation error containing %q, got nil", tc.want)
|
|
}
|
|
if !strings.Contains(err.Error(), tc.want) {
|
|
t.Errorf("error = %q, want it to contain %q", err.Error(), tc.want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestMonthlyBarsFixture guards the deterministic monthly aggregation evidence:
|
|
// the YAML passes dry-run validation, and running it against the canned response
|
|
// reproduces the committed expected JSONL fixture byte-for-byte so the SDD S04
|
|
// evidence is stable across repeated runs.
|
|
func TestMonthlyBarsFixture(t *testing.T) {
|
|
yamlPath := filepath.Join("..", "..", "testdata", "operator", "monthly_bars_aggregation.yaml")
|
|
sc, err := LoadScenario(yamlPath)
|
|
if err != nil {
|
|
t.Fatalf("monthly_bars_aggregation.yaml failed validation: %v", err)
|
|
}
|
|
|
|
api := &fakeAPI{aggregateMonthlyResp: monthlyAggregateResp()}
|
|
url := startFakeAPIServer(t, api)
|
|
|
|
var stdout bytes.Buffer
|
|
code := RunScenario(context.Background(), sc, RunOptions{
|
|
APIURL: url,
|
|
Timeout: 2 * time.Second,
|
|
Format: OutputJSONL,
|
|
}, &stdout)
|
|
if code != codeOK {
|
|
t.Fatalf("exit code = %d, want 0 (out=%q)", code, stdout.String())
|
|
}
|
|
|
|
expectedPath := filepath.Join("..", "..", "testdata", "operator", "expected", "monthly_bars_aggregation.jsonl")
|
|
want, err := os.ReadFile(expectedPath)
|
|
if err != nil {
|
|
t.Fatalf("read expected fixture: %v", err)
|
|
}
|
|
|
|
if got := stdout.String(); got != string(want) {
|
|
t.Errorf("monthly aggregation JSONL output drift:\n got=%q\nwant=%q", got, string(want))
|
|
}
|
|
}
|