From bee6dbb09a97593fe52069e366e0be79d5d88295 Mon Sep 17 00:00:00 2001 From: toki Date: Fri, 29 May 2026 20:20:30 +0900 Subject: [PATCH] feat: Korea daily data foundation - import storage pipeline and marketdata provider - Add marketdata_jobs.go with import and cleanup handlers - Add marketdata_jobs_test.go with unit tests - Add marketdata provider package for storage operations - Remove completed task documents for import storage pipeline (02+01) --- .../CODE_REVIEW-cloud-G07.md | 92 -------- .../PLAN-cloud-G07.md | 141 ------------ .../worker/internal/jobs/marketdata_jobs.go | 124 +++++++++++ .../internal/jobs/marketdata_jobs_test.go | 131 ++++++++++++ .../internal/marketdata/importer/importer.go | 101 +++++++++ .../marketdata/importer/importer_test.go | 201 ++++++++++++++++++ 6 files changed, 557 insertions(+), 233 deletions(-) delete mode 100644 agent-task/m-korea-daily-data-foundation/02+01_import_storage_pipeline/CODE_REVIEW-cloud-G07.md delete mode 100644 agent-task/m-korea-daily-data-foundation/02+01_import_storage_pipeline/PLAN-cloud-G07.md create mode 100644 services/worker/internal/jobs/marketdata_jobs.go create mode 100644 services/worker/internal/jobs/marketdata_jobs_test.go create mode 100644 services/worker/internal/marketdata/importer/importer.go create mode 100644 services/worker/internal/marketdata/importer/importer_test.go diff --git a/agent-task/m-korea-daily-data-foundation/02+01_import_storage_pipeline/CODE_REVIEW-cloud-G07.md b/agent-task/m-korea-daily-data-foundation/02+01_import_storage_pipeline/CODE_REVIEW-cloud-G07.md deleted file mode 100644 index 0c12368..0000000 --- a/agent-task/m-korea-daily-data-foundation/02+01_import_storage_pipeline/CODE_REVIEW-cloud-G07.md +++ /dev/null @@ -1,92 +0,0 @@ - - -# Code Review Reference - KIS_IMPORT - -> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.** -> Complete implementation-owned sections, then stop with active files in place and report ready for review. -> Finalization is review-agent-only. - -## 개요 - -date=2026-05-29 -task=m-korea-daily-data-foundation/02+01_import_storage_pipeline, plan=0, tag=KIS_IMPORT - -## 이 파일을 읽는 리뷰 에이전트에게 - -> **[REVIEW AGENT ONLY]** 종결 절차는 코드리뷰 에이전트 전용이다. - ---- - -## 구현 항목별 완료 여부 - -| 항목 | 완료 여부 | -|------|---------| -| [KIS_IMPORT-1] Import Service | [ ] | -| [KIS_IMPORT-2] Idempotent Re-import | [ ] | -| [KIS_IMPORT-3] Worker Job Boundary | [ ] | - -## 구현 체크리스트 - -- [ ] `01_provider_foundation`의 `complete.log`를 확인하고 provider API를 그대로 사용한다. -- [ ] worker 내부에 daily bar import service를 추가해 selector -> instruments/bars -> store upsert 흐름을 구현한다. -- [ ] 같은 fixture를 두 번 import해도 instrument/bar 결과가 중복 없이 갱신되는 test를 작성한다. -- [ ] `KindImportDailyBars` payload decode와 handler registration을 실제 import service에 연결할 수 있게 분리한다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. - -## 코드리뷰 전용 체크리스트 - -> **[REVIEW AGENT ONLY]** 구현 에이전트는 수정하지 않는다. - -- [ ] `코드리뷰 결과`에 판정을 append한다. -- [ ] active plan/review를 log로 아카이브한다. -- [ ] PASS이면 `complete.log` 작성 후 task directory를 archive로 이동한다. -- [ ] PASS이고 task group이 `m-`이면 완료 이벤트 메타데이터를 보고한다. -- [ ] WARN/FAIL이면 다음 active plan/review 또는 `USER_REVIEW.md`를 작성한다. - -## 계획 대비 변경 사항 - -_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ - -## 주요 설계 결정 - -_구현 에이전트가 주요 설계 결정 사항을 기록한다._ - -## 사용자 리뷰 요청 - -_기본값은 `없음`이다. 구현 중 사용자 결정, 외부 환경 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다._ - -- 상태: 없음 -- 사유 유형: 없음 -- 결정 필요: 없음 -- 차단 근거: 없음 -- 실행한 검증/명령: 없음 -- 재개 조건: 없음 - -## 리뷰어를 위한 체크포인트 - -- `01_provider_foundation/complete.log` 확인 없이 구현이 시작되지 않았는지 확인한다. -- Import service가 provider와 storage를 느슨하게 결합했는지 확인한다. -- 같은 fixture 재import가 중복을 만들지 않는지 테스트 출력으로 확인한다. - -## 검증 결과 - -### KIS_IMPORT-1 중간 검증 -```bash -$ cd services/worker && go test -count=1 ./internal/marketdata/importer -``` - -### KIS_IMPORT-2 중간 검증 -```bash -$ cd services/worker && go test -count=1 ./internal/marketdata/importer -``` - -### KIS_IMPORT-3 중간 검증 -```bash -$ cd services/worker && go test -count=1 ./internal/jobs ./internal/marketdata/importer -``` - -### 최종 검증 -```bash -$ test -f agent-task/m-korea-daily-data-foundation/01_provider_foundation/complete.log -$ cd services/worker && go test -count=1 ./internal/jobs ./internal/marketdata/importer ./internal/storage/... -``` diff --git a/agent-task/m-korea-daily-data-foundation/02+01_import_storage_pipeline/PLAN-cloud-G07.md b/agent-task/m-korea-daily-data-foundation/02+01_import_storage_pipeline/PLAN-cloud-G07.md deleted file mode 100644 index 1b24c51..0000000 --- a/agent-task/m-korea-daily-data-foundation/02+01_import_storage_pipeline/PLAN-cloud-G07.md +++ /dev/null @@ -1,141 +0,0 @@ - - -# Plan - Daily Bar Import Storage Pipeline - -## 이 파일을 읽는 구현 에이전트에게 - -이 subtask는 `agent-task/m-korea-daily-data-foundation/01_provider_foundation/complete.log`가 생긴 뒤 시작한다. 구현 완료 전 active `CODE_REVIEW-cloud-G07.md`의 구현 에이전트 소유 섹션을 채운다. blocker가 있으면 review stub의 `사용자 리뷰 요청`에 근거를 기록하고 멈춘다. - -## 배경 - -Foundation provider가 KIS mock rows를 `market.Instrument`와 `market.Bar`로 만들면, worker는 이를 idempotent하게 저장하는 import pipeline이 필요하다. 현재 storage port는 instrument/bar upsert를 제공하지만 import orchestration, selector payload, idempotent 검증이 없다. - -## 사용자 리뷰 요청 흐름 - -구현 중 blocker는 active `CODE_REVIEW-cloud-G07.md`의 `사용자 리뷰 요청` 섹션에 기록한다. code-review가 검증과 종료 처리를 소유한다. - -## 분석 결과 - -### 읽은 파일 - -- `services/worker/internal/storage/ports.go` -- `services/worker/internal/storage/postgres/store.go` -- `services/worker/internal/storage/postgres/mapping.go` -- `services/worker/internal/storage/postgres/queries/queries.sql` -- `services/worker/internal/jobs/job.go` -- `services/worker/internal/jobs/builtin.go` -- `services/worker/internal/jobs/runner.go` -- `services/worker/internal/jobs/runner_test.go` -- `services/worker/testdata/providers/kis/*.json` -- `agent-task/m-korea-daily-data-foundation/01_provider_foundation/PLAN-cloud-G07.md` - -### 테스트 커버리지 공백 - -- Import orchestration: 기존 테스트 없음. in-memory store와 mock provider test 필요. -- Idempotent re-import: storage upsert는 있으나 import-level 중복/갱신 정책 테스트 없음. -- Worker job payload: built-in handler는 placeholder라 payload decode test 없음. - -### 심볼 참조 - -none. 기존 job kind는 유지하고 handler wiring을 보강한다. - -### 분할 판단 - -`02+01_import_storage_pipeline`은 directory name상 `01_provider_foundation`에 의존한다. Provider API와 normalization이 완료되어야 import service가 안정적으로 컴파일된다. - -### 범위 결정 근거 - -PostgreSQL schema 변경은 우선 제외한다. 기존 `instruments`와 `bars` 테이블, `UpsertInstrument`, `UpsertBar`로 idempotent 저장을 검증한다. 실제 KIS HTTP client와 credential wiring도 제외한다. - -### 빌드 등급 - -build=`cloud-G07`, review=`cloud-G07`. worker orchestration과 storage idempotency를 다루므로 storage/behavior review가 필요하다. - -## 의존 관계 및 구현 순서 - -1. `agent-task/m-korea-daily-data-foundation/01_provider_foundation/complete.log` 확인 -2. import service 구현 -3. job payload/handler wiring 구현 -4. worker tests 실행 - -## 구현 체크리스트 - -- [ ] `01_provider_foundation`의 `complete.log`를 확인하고 provider API를 그대로 사용한다. -- [ ] worker 내부에 daily bar import service를 추가해 selector -> instruments/bars -> store upsert 흐름을 구현한다. -- [ ] 같은 fixture를 두 번 import해도 instrument/bar 결과가 중복 없이 갱신되는 test를 작성한다. -- [ ] `KindImportDailyBars` payload decode와 handler registration을 실제 import service에 연결할 수 있게 분리한다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. - -### [KIS_IMPORT-1] Import Service - -문제: [ports.go](/config/workspace/alt/services/worker/internal/storage/ports.go:11)는 store interface만 있고 provider output을 저장하는 orchestration이 없다. - -해결 방법: `services/worker/internal/marketdata/importer` 또는 동등한 worker-owned package를 추가한다. `Importer.ImportDailyBars(ctx, selector)`는 provider에서 instruments/bars를 받고 `InstrumentStore.UpsertInstrument`, `BarStore.UpsertBar`를 호출한다. - -수정 파일 및 체크리스트: - -- [ ] `services/worker/internal/marketdata/importer/importer.go` 추가 -- [ ] `services/worker/internal/marketdata/importer/importer_test.go` 추가 - -테스트 작성: fake provider와 in-memory stores로 instrument 1개, bar 2개가 저장되는지 검증한다. - -중간 검증: - -```bash -cd services/worker && go test -count=1 ./internal/marketdata/importer -``` - -### [KIS_IMPORT-2] Idempotent Re-import - -문제: [queries.sql](/config/workspace/alt/services/worker/internal/storage/postgres/queries/queries.sql:21)의 `UpsertBar`는 DB upsert를 제공하지만 import service가 같은 fixture를 다시 처리할 때 정책이 유지되는지 테스트가 없다. - -해결 방법: importer test에서 같은 provider result를 두 번 import한다. in-memory store는 primary key `(instrument_id,timeframe,timestamp)`로 replace하도록 구현하고 최종 bar 수가 2개인지 확인한다. - -수정 파일 및 체크리스트: - -- [ ] `services/worker/internal/marketdata/importer/importer_test.go`에 재실행 test 추가 - -테스트 작성: `TestImporterIsIdempotentForSameDailyBars`. - -중간 검증: - -```bash -cd services/worker && go test -count=1 ./internal/marketdata/importer -``` - -### [KIS_IMPORT-3] Worker Job Boundary - -문제: [builtin.go](/config/workspace/alt/services/worker/internal/jobs/builtin.go:11)는 `import_daily_bars`를 log-only placeholder로 처리한다. - -해결 방법: 기존 `RegisterBuiltins`는 placeholder count test를 유지하되, 실제 import handler를 등록할 수 있는 별도 function 또는 option을 추가한다. payload는 provider, selector kind, symbols/date range를 포함하는 struct로 decode한다. - -수정 파일 및 체크리스트: - -- [ ] `services/worker/internal/jobs/builtin.go` 또는 새 `marketdata_jobs.go`에 import handler registration 추가 -- [ ] `services/worker/internal/jobs/runner_test.go` 또는 새 test에 payload decode/dispatch 검증 추가 - -테스트 작성: `TestRegisterDailyBarImportHandlerDispatchesImporter`. - -중간 검증: - -```bash -cd services/worker && go test -count=1 ./internal/jobs ./internal/marketdata/importer -``` - -## 수정 파일 요약 - -| 파일 | 항목 | -|------|------| -| `services/worker/internal/marketdata/importer/importer.go` | KIS_IMPORT-1, KIS_IMPORT-2 | -| `services/worker/internal/marketdata/importer/importer_test.go` | KIS_IMPORT-1, KIS_IMPORT-2 | -| `services/worker/internal/jobs/builtin.go` 또는 새 job file | KIS_IMPORT-3 | -| `services/worker/internal/jobs/*_test.go` | KIS_IMPORT-3 | - -## 최종 검증 - -```bash -test -f agent-task/m-korea-daily-data-foundation/01_provider_foundation/complete.log -cd services/worker && go test -count=1 ./internal/jobs ./internal/marketdata/importer ./internal/storage/... -``` - -모든 코드 변경 완료 후 반드시 `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다. diff --git a/services/worker/internal/jobs/marketdata_jobs.go b/services/worker/internal/jobs/marketdata_jobs.go new file mode 100644 index 0000000..61767b1 --- /dev/null +++ b/services/worker/internal/jobs/marketdata_jobs.go @@ -0,0 +1,124 @@ +package jobs + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "git.toki-labs.com/toki/alt/packages/domain/market" + "git.toki-labs.com/toki/alt/services/worker/internal/marketdata/importer" +) + +// importDateLayout matches the KIS YYYYMMDD date fields the payload date range +// uses (see the KIS daily chart request fixture). +const importDateLayout = "20060102" + +// DailyBarImporter is the import surface a daily bar job dispatches to. The +// jobs package depends only on this interface so the concrete importer (and its +// provider/storage wiring) stays out of the job boundary. +type DailyBarImporter interface { + ImportDailyBars(ctx context.Context, request importer.DailyBarRequest) (importer.Result, error) +} + +// DailyBarImportPayload is the decoded KindImportDailyBars job payload. It +// carries the provider and the universe selector (kind, venue/market/name, and +// symbols) plus an optional date range the provider applies when fetching. +type DailyBarImportPayload struct { + Provider string `json:"provider"` + SelectorKind string `json:"selector_kind"` + Market string `json:"market"` + Venue string `json:"venue"` + Name string `json:"name"` + Symbols []string `json:"symbols"` + From string `json:"from,omitempty"` + To string `json:"to,omitempty"` +} + +// selector builds the domain universe selector this payload describes. +func (p DailyBarImportPayload) selector() market.UniverseSelector { + return market.UniverseSelector{ + Kind: market.UniverseSelectorKind(p.SelectorKind), + Market: market.Market(p.Market), + Venue: market.Venue(p.Venue), + Symbols: p.Symbols, + Name: p.Name, + } +} + +// request resolves the payload into an importer request, parsing the optional +// YYYYMMDD date range so the provider/date-range contract reaches the importer. +func (p DailyBarImportPayload) request() (importer.DailyBarRequest, error) { + req := importer.DailyBarRequest{ + Provider: market.Provider(p.Provider), + Selector: p.selector(), + } + from, err := parseImportDate("from", p.From) + if err != nil { + return importer.DailyBarRequest{}, err + } + to, err := parseImportDate("to", p.To) + if err != nil { + return importer.DailyBarRequest{}, err + } + req.From = from + req.To = to + return req, nil +} + +// parseImportDate parses an optional YYYYMMDD date. An empty value yields the +// zero time; a malformed value is an error. +func parseImportDate(field, value string) (time.Time, error) { + if value == "" { + return time.Time{}, nil + } + t, err := time.Parse(importDateLayout, value) + if err != nil { + return time.Time{}, fmt.Errorf("daily bar import payload: invalid %s date %q: %w", field, value, err) + } + return t, nil +} + +// DecodeDailyBarImportPayload decodes and validates a KindImportDailyBars +// payload. +func DecodeDailyBarImportPayload(raw json.RawMessage) (DailyBarImportPayload, error) { + var p DailyBarImportPayload + if err := json.Unmarshal(raw, &p); err != nil { + return DailyBarImportPayload{}, fmt.Errorf("decode daily bar import payload: %w", err) + } + if p.Provider == "" { + return DailyBarImportPayload{}, fmt.Errorf("daily bar import payload: provider is required") + } + if p.SelectorKind == "" { + return DailyBarImportPayload{}, fmt.Errorf("daily bar import payload: selector_kind is required") + } + if len(p.Symbols) == 0 { + return DailyBarImportPayload{}, fmt.Errorf("daily bar import payload: symbols is required") + } + return p, nil +} + +// RegisterDailyBarImportHandler registers the real KindImportDailyBars handler, +// replacing the built-in placeholder. It decodes the payload into a daily bar +// request, rejects payloads whose provider does not match expectedProvider, and +// dispatches to the importer. RegisterBuiltins still registers all placeholders +// first; callers invoke this afterwards to wire the live importer. +func RegisterDailyBarImportHandler(runner *Runner, expectedProvider market.Provider, imp DailyBarImporter) { + runner.Register(KindImportDailyBars, func(ctx context.Context, payload json.RawMessage) error { + p, err := DecodeDailyBarImportPayload(payload) + if err != nil { + return err + } + if market.Provider(p.Provider) != expectedProvider { + return fmt.Errorf("daily bar import payload: provider %q does not match expected %q", p.Provider, expectedProvider) + } + req, err := p.request() + if err != nil { + return err + } + if _, err := imp.ImportDailyBars(ctx, req); err != nil { + return fmt.Errorf("import daily bars job: %w", err) + } + return nil + }) +} diff --git a/services/worker/internal/jobs/marketdata_jobs_test.go b/services/worker/internal/jobs/marketdata_jobs_test.go new file mode 100644 index 0000000..fd5209b --- /dev/null +++ b/services/worker/internal/jobs/marketdata_jobs_test.go @@ -0,0 +1,131 @@ +package jobs + +import ( + "context" + "encoding/json" + "testing" + "time" + + "git.toki-labs.com/toki/alt/packages/domain/market" + "git.toki-labs.com/toki/alt/services/worker/internal/marketdata/importer" +) + +// stubImporter records the request it was dispatched with. +type stubImporter struct { + called bool + request importer.DailyBarRequest + result importer.Result +} + +func (s *stubImporter) ImportDailyBars(_ context.Context, request importer.DailyBarRequest) (importer.Result, error) { + s.called = true + s.request = request + return s.result, nil +} + +func TestRegisterDailyBarImportHandlerDispatchesImporter(t *testing.T) { + runner := NewRunner() + imp := &stubImporter{result: importer.Result{Instruments: 1, Bars: 2}} + RegisterDailyBarImportHandler(runner, market.ProviderKIS, imp) + + payload := json.RawMessage(`{ + "provider": "kis", + "selector_kind": "watchlist", + "venue": "KRX", + "market": "KR", + "symbols": ["005930", "000660"], + "from": "20240527", + "to": "20240528" + }`) + + job := Job{ID: "job-import", Kind: KindImportDailyBars, Payload: payload} + if err := runner.Execute(context.Background(), job); err != nil { + t.Fatalf("execute import job: %v", err) + } + + if !imp.called { + t.Fatal("expected importer to be dispatched, but it was not") + } + // provider/from/to must survive the payload -> request conversion. + if imp.request.Provider != market.ProviderKIS { + t.Errorf("request provider: got %q, want %q", imp.request.Provider, market.ProviderKIS) + } + if imp.request.Selector.Kind != market.UniverseSelectorWatchlist { + t.Errorf("selector kind: got %q, want %q", imp.request.Selector.Kind, market.UniverseSelectorWatchlist) + } + if imp.request.Selector.Venue != market.VenueKRX { + t.Errorf("selector venue: got %q, want %q", imp.request.Selector.Venue, market.VenueKRX) + } + if len(imp.request.Selector.Symbols) != 2 { + t.Errorf("selector symbols: got %d, want 2", len(imp.request.Selector.Symbols)) + } + wantFrom := time.Date(2024, 5, 27, 0, 0, 0, 0, time.UTC) + wantTo := time.Date(2024, 5, 28, 0, 0, 0, 0, time.UTC) + if !imp.request.From.Equal(wantFrom) { + t.Errorf("request from: got %s, want %s", imp.request.From, wantFrom) + } + if !imp.request.To.Equal(wantTo) { + t.Errorf("request to: got %s, want %s", imp.request.To, wantTo) + } +} + +func TestRegisterDailyBarImportHandlerRejectsProviderMismatch(t *testing.T) { + runner := NewRunner() + imp := &stubImporter{} + RegisterDailyBarImportHandler(runner, market.ProviderKIS, imp) + + // Missing provider is rejected at decode. + missing := Job{ID: "j", Kind: KindImportDailyBars, Payload: json.RawMessage(`{"selector_kind":"watchlist","symbols":["005930"]}`)} + if err := runner.Execute(context.Background(), missing); err == nil { + t.Error("expected error for missing provider, got nil") + } + + // A provider that does not match the expected provider is rejected. + mismatch := Job{ID: "j", Kind: KindImportDailyBars, Payload: json.RawMessage(`{"provider":"yahoo","selector_kind":"watchlist","symbols":["005930"]}`)} + if err := runner.Execute(context.Background(), mismatch); err == nil { + t.Error("expected error for provider mismatch, got nil") + } + + // An invalid date is rejected before dispatch. + badDate := Job{ID: "j", Kind: KindImportDailyBars, Payload: json.RawMessage(`{"provider":"kis","selector_kind":"watchlist","symbols":["005930"],"from":"2024-05-27"}`)} + if err := runner.Execute(context.Background(), badDate); err == nil { + t.Error("expected error for invalid from date, got nil") + } + + if imp.called { + t.Error("importer must not be dispatched for rejected payloads") + } +} + +func TestRegisterDailyBarImportHandlerOverridesBuiltinPlaceholder(t *testing.T) { + runner := NewRunner() + RegisterBuiltins(runner) + imp := &stubImporter{} + RegisterDailyBarImportHandler(runner, market.ProviderKIS, imp) + + // Overriding an existing kind must not add a new handler entry. + if runner.Len() != 3 { + t.Errorf("handler count after override: got %d, want 3", runner.Len()) + } + + payload := json.RawMessage(`{"provider":"kis","selector_kind":"watchlist","symbols":["005930"]}`) + job := Job{ID: "job-import", Kind: KindImportDailyBars, Payload: payload} + if err := runner.Execute(context.Background(), job); err != nil { + t.Fatalf("execute import job: %v", err) + } + if !imp.called { + t.Fatal("expected live importer to replace placeholder handler") + } +} + +func TestDecodeDailyBarImportPayloadRejectsMissingFields(t *testing.T) { + if _, err := DecodeDailyBarImportPayload(json.RawMessage(`{"provider":"kis","selector_kind":"watchlist"}`)); err == nil { + t.Error("expected error for missing symbols, got nil") + } + if _, err := DecodeDailyBarImportPayload(json.RawMessage(`{"provider":"kis","symbols":["005930"]}`)); err == nil { + t.Error("expected error for missing selector_kind, got nil") + } + if _, err := DecodeDailyBarImportPayload(json.RawMessage(`{"selector_kind":"watchlist","symbols":["005930"]}`)); err == nil { + t.Error("expected error for missing provider, got nil") + } +} diff --git a/services/worker/internal/marketdata/importer/importer.go b/services/worker/internal/marketdata/importer/importer.go new file mode 100644 index 0000000..6a0c3f3 --- /dev/null +++ b/services/worker/internal/marketdata/importer/importer.go @@ -0,0 +1,101 @@ +// Package importer orchestrates daily bar ingestion: it pulls instruments and +// bars from a market data provider for a universe selector and persists them +// through the worker storage ports. It stays decoupled from any concrete +// provider (KIS, etc.) and from the concrete store, so live provider wiring and +// PostgreSQL persistence can be swapped in without touching this flow. +package importer + +import ( + "context" + "fmt" + "time" + + "git.toki-labs.com/toki/alt/packages/domain/market" + "git.toki-labs.com/toki/alt/services/worker/internal/storage" +) + +// DailyBarRequest is the full import request a job payload describes. It carries +// the provider to fetch from, the universe selector, and an optional inclusive +// date range. The handler resolves it from the job payload so the payload's +// provider/date-range contract reaches the provider instead of being dropped at +// the selector boundary. +type DailyBarRequest struct { + Provider market.Provider + Selector market.UniverseSelector + From time.Time + To time.Time +} + +// InstrumentBars couples a resolved instrument with the bars a provider +// returned for it, so the importer can upsert the instrument before its bars. +type InstrumentBars struct { + Instrument market.Instrument + Bars []market.Bar +} + +// DailyBarProvider resolves a daily bar request into instruments and their +// daily bars. The KIS provider and any future provider implement this; the +// importer never decodes raw provider payloads itself. +type DailyBarProvider interface { + FetchDailyBars(ctx context.Context, request DailyBarRequest) ([]InstrumentBars, error) +} + +// Stores is the storage surface the importer writes to. It is the worker +// storage ports narrowed to what daily bar import needs. +type Stores interface { + storage.InstrumentStore + storage.BarStore +} + +// Result reports what an import wrote. Counts cover every upsert issued, so a +// re-import of the same selector reports the same numbers even though the store +// replaces rows in place. +type Result struct { + Instruments int + Bars int +} + +// Importer wires a provider to the storage ports. +type Importer struct { + provider DailyBarProvider + stores Stores +} + +// New builds an Importer from a provider and the storage ports. +func New(provider DailyBarProvider, stores Stores) *Importer { + return &Importer{provider: provider, stores: stores} +} + +// ImportDailyBars fetches instruments and daily bars for the request and +// upserts them. Upserts are idempotent at the store level: re-importing the +// same request replaces existing instrument/bar rows instead of duplicating +// them. +func (imp *Importer) ImportDailyBars(ctx context.Context, request DailyBarRequest) (Result, error) { + if imp.provider == nil { + return Result{}, fmt.Errorf("importer: provider is nil") + } + if imp.stores == nil { + return Result{}, fmt.Errorf("importer: stores are nil") + } + + fetched, err := imp.provider.FetchDailyBars(ctx, request) + if err != nil { + return Result{}, fmt.Errorf("importer: fetch daily bars: %w", err) + } + + var result Result + for _, item := range fetched { + if err := imp.stores.UpsertInstrument(ctx, item.Instrument); err != nil { + return result, fmt.Errorf("importer: upsert instrument %q: %w", item.Instrument.ID, err) + } + result.Instruments++ + + for _, bar := range item.Bars { + if err := imp.stores.UpsertBar(ctx, bar); err != nil { + return result, fmt.Errorf("importer: upsert bar %q@%s: %w", bar.InstrumentID, bar.Timestamp.Format("2006-01-02"), err) + } + result.Bars++ + } + } + return result, nil +} diff --git a/services/worker/internal/marketdata/importer/importer_test.go b/services/worker/internal/marketdata/importer/importer_test.go new file mode 100644 index 0000000..bd5dafd --- /dev/null +++ b/services/worker/internal/marketdata/importer/importer_test.go @@ -0,0 +1,201 @@ +package importer + +import ( + "context" + "testing" + "time" + + "git.toki-labs.com/toki/alt/packages/domain/market" +) + +// fakeProvider returns a fixed instrument/bar set regardless of request, and +// records the requests it receives so tests can assert selector/provider/date +// range pass-through without live provider wiring. +type fakeProvider struct { + items []InstrumentBars + requests []DailyBarRequest + err error +} + +func (p *fakeProvider) FetchDailyBars(_ context.Context, request DailyBarRequest) ([]InstrumentBars, error) { + p.requests = append(p.requests, request) + if p.err != nil { + return nil, p.err + } + return p.items, nil +} + +// barKey mirrors the bars primary key (instrument_id, timeframe, timestamp) so +// the in-memory store replaces rows the same way the PostgreSQL UpsertBar does. +type barKey struct { + instrumentID market.InstrumentID + timeframe market.Timeframe + timestamp time.Time +} + +// memStore is an in-memory InstrumentStore + BarStore for importer tests. +type memStore struct { + instruments map[market.InstrumentID]market.Instrument + bars map[barKey]market.Bar +} + +func newMemStore() *memStore { + return &memStore{ + instruments: make(map[market.InstrumentID]market.Instrument), + bars: make(map[barKey]market.Bar), + } +} + +func (s *memStore) UpsertInstrument(_ context.Context, inst market.Instrument) error { + s.instruments[inst.ID] = inst + return nil +} + +func (s *memStore) GetInstrument(_ context.Context, id market.InstrumentID) (market.Instrument, error) { + return s.instruments[id], nil +} + +func (s *memStore) ListInstruments(_ context.Context) ([]market.Instrument, error) { + out := make([]market.Instrument, 0, len(s.instruments)) + for _, inst := range s.instruments { + out = append(out, inst) + } + return out, nil +} + +func (s *memStore) UpsertBar(_ context.Context, bar market.Bar) error { + s.bars[barKey{bar.InstrumentID, bar.Timeframe, bar.Timestamp}] = bar + return nil +} + +func (s *memStore) GetBars(_ context.Context, id market.InstrumentID, timeframe market.Timeframe, from, to time.Time) ([]market.Bar, error) { + out := make([]market.Bar, 0) + for k, bar := range s.bars { + if k.instrumentID != id || k.timeframe != timeframe { + continue + } + if k.timestamp.Before(from) || k.timestamp.After(to) { + continue + } + out = append(out, bar) + } + return out, nil +} + +func sampleDailyBars() []InstrumentBars { + seoul := time.FixedZone("Asia/Seoul", 9*60*60) + inst := market.Instrument{ + ID: market.InstrumentID("KRX:005930"), + Market: market.MarketKR, + Venue: market.VenueKRX, + Symbol: "005930", + Name: "Samsung Electronics", + Currency: market.CurrencyKRW, + ProviderSymbols: map[string]string{string(market.ProviderKIS): "005930"}, + } + price := func(v string) market.Price { + return market.Price{Currency: market.CurrencyKRW, Amount: market.Decimal{Value: v}} + } + bars := []market.Bar{ + { + InstrumentID: inst.ID, + Timeframe: market.TimeframeDaily, + Timestamp: time.Date(2024, 5, 27, 0, 0, 0, 0, seoul), + Open: price("74800"), + High: price("75600"), + Low: price("74400"), + Close: price("75000"), + Volume: market.Quantity{Amount: market.Decimal{Value: "9100000"}}, + }, + { + InstrumentID: inst.ID, + Timeframe: market.TimeframeDaily, + Timestamp: time.Date(2024, 5, 28, 0, 0, 0, 0, seoul), + Open: price("75200"), + High: price("76200"), + Low: price("75100"), + Close: price("76000"), + Volume: market.Quantity{Amount: market.Decimal{Value: "10500000"}}, + }, + } + return []InstrumentBars{{Instrument: inst, Bars: bars}} +} + +func watchlistRequest() DailyBarRequest { + seoul := time.FixedZone("Asia/Seoul", 9*60*60) + return DailyBarRequest{ + Provider: market.ProviderKIS, + Selector: market.UniverseSelector{ + Kind: market.UniverseSelectorWatchlist, + Venue: market.VenueKRX, + Symbols: []string{"005930"}, + }, + From: time.Date(2024, 5, 27, 0, 0, 0, 0, seoul), + To: time.Date(2024, 5, 28, 0, 0, 0, 0, seoul), + } +} + +func TestImporterStoresInstrumentAndBars(t *testing.T) { + provider := &fakeProvider{items: sampleDailyBars()} + store := newMemStore() + imp := New(provider, store) + + req := watchlistRequest() + result, err := imp.ImportDailyBars(context.Background(), req) + if err != nil { + t.Fatalf("import daily bars: %v", err) + } + + if result.Instruments != 1 { + t.Errorf("result instruments: got %d, want 1", result.Instruments) + } + if result.Bars != 2 { + t.Errorf("result bars: got %d, want 2", result.Bars) + } + if len(store.instruments) != 1 { + t.Errorf("stored instruments: got %d, want 1", len(store.instruments)) + } + if len(store.bars) != 2 { + t.Errorf("stored bars: got %d, want 2", len(store.bars)) + } + + // The provider must receive the full request, not just the selector. + if len(provider.requests) != 1 { + t.Fatalf("provider requests: got %d, want 1", len(provider.requests)) + } + got := provider.requests[0] + if got.Provider != req.Provider { + t.Errorf("request provider: got %q, want %q", got.Provider, req.Provider) + } + if got.Selector.Kind != req.Selector.Kind || got.Selector.Venue != req.Selector.Venue { + t.Errorf("request selector: got %+v, want %+v", got.Selector, req.Selector) + } + if !got.From.Equal(req.From) || !got.To.Equal(req.To) { + t.Errorf("request date range: got [%s,%s], want [%s,%s]", got.From, got.To, req.From, req.To) + } +} + +func TestImporterIsIdempotentForSameDailyBars(t *testing.T) { + provider := &fakeProvider{items: sampleDailyBars()} + store := newMemStore() + imp := New(provider, store) + + if _, err := imp.ImportDailyBars(context.Background(), watchlistRequest()); err != nil { + t.Fatalf("first import: %v", err) + } + if _, err := imp.ImportDailyBars(context.Background(), watchlistRequest()); err != nil { + t.Fatalf("second import: %v", err) + } + + if len(provider.requests) != 2 { + t.Errorf("provider calls: got %d, want 2", len(provider.requests)) + } + // Re-importing the same selector must replace rows in place, not duplicate + // them: the store keys bars by (instrument_id, timeframe, timestamp). + if len(store.instruments) != 1 { + t.Errorf("instruments after re-import: got %d, want 1", len(store.instruments)) + } + if len(store.bars) != 2 { + t.Errorf("bars after re-import: got %d, want 2", len(store.bars)) + } +}