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)
2026-05-29 20:20:30 +09:00 · 2026-05-29 20:20:30 +09:00 · bee6dbb09a
commit bee6dbb09a
parent 6524afaa20
6 changed files with 557 additions and 233 deletions
--- 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
@ -1,92 +0,0 @@
-<!-- task=m-korea-daily-data-foundation/02+01_import_storage_pipeline plan=0 tag=KIS_IMPORT -->
-
-# 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-<milestone-slug>`이면 완료 이벤트 메타데이터를 보고한다.
- [ ] 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/...
-```
--- 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
@ -1,141 +0,0 @@
-<!-- task=m-korea-daily-data-foundation/02+01_import_storage_pipeline plan=0 tag=KIS_IMPORT -->
-
-# 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`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다.
--- a/services/worker/internal/jobs/marketdata_jobs.go
+++ 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
+	})
+}
--- a/services/worker/internal/jobs/marketdata_jobs_test.go
+++ 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")
+	}
+}
--- a/services/worker/internal/marketdata/importer/importer.go
+++ 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
+}
--- a/services/worker/internal/marketdata/importer/importer_test.go
+++ 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))
+	}
+}