KArtSell.Aegis/docs/contracts/data/VS-03_DATA_CONTRACT.md

VS-03: Market Data Ingestion - Data Contract

Slice ID: VS-03
Phase: Data Layer (write model)
Status: Specification Ready

---

Write Model (Normalized, 3NF)

Table: market_data.daily_prices (Core)

CREATE TABLE market_data.daily_prices (
    -- Identity
    price_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    symbol VARCHAR(20) NOT NULL,
    trading_date DATE NOT NULL,
    
    -- OHLCV
    open_price DECIMAL(10, 2) NOT NULL CHECK (open_price > 0),
    high_price DECIMAL(10, 2) NOT NULL CHECK (high_price > 0),
    low_price DECIMAL(10, 2) NOT NULL CHECK (low_price > 0),
    close_price DECIMAL(10, 2) NOT NULL CHECK (close_price > 0),
    adjusted_close DECIMAL(10, 2),
    volume BIGINT NOT NULL CHECK (volume >= 0),
    
    -- PIT (Point-in-Time) Compliance
    published_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    revision INT NOT NULL DEFAULT 1,
    
    -- Audit
    data_source VARCHAR(50) NOT NULL, -- 'KRX', 'OpenDart', 'Stub'
    ingestion_job_id UUID,
    correlation_id UUID,
    
    -- Soft-delete (never delete, only version)
    removed_at TIMESTAMP,
    
    CONSTRAINT unique_daily_price UNIQUE (symbol, trading_date, revision),
    CONSTRAINT valid_prices CHECK (low_price <= open_price AND open_price <= high_price)
);

CREATE INDEX idx_daily_prices_symbol_date ON market_data.daily_prices(symbol, trading_date DESC);
CREATE INDEX idx_daily_prices_published ON market_data.daily_prices(published_at DESC);

Table: market_data.indices (Supplementary)

CREATE TABLE market_data.indices (
    -- Identity
    index_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    index_code VARCHAR(20) NOT NULL, -- 'KOSPI', 'KRX200', 'KOSDAQ'
    trading_date DATE NOT NULL,
    
    -- OHLCV
    open_value DECIMAL(10, 2) NOT NULL,
    high_value DECIMAL(10, 2) NOT NULL,
    low_value DECIMAL(10, 2) NOT NULL,
    close_value DECIMAL(10, 2) NOT NULL,
    change_percent DECIMAL(5, 2),
    volume BIGINT,
    
    -- PIT
    published_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    revision INT NOT NULL DEFAULT 1,
    
    -- Audit
    data_source VARCHAR(50) NOT NULL,
    correlation_id UUID,
    
    removed_at TIMESTAMP,
    
    CONSTRAINT unique_index UNIQUE (index_code, trading_date, revision)
);

CREATE INDEX idx_indices_code_date ON market_data.indices(index_code, trading_date DESC);

Table: market_data.companies (Master)

CREATE TABLE market_data.companies (
    -- Identity
    company_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    symbol VARCHAR(20) NOT NULL UNIQUE,
    
    -- Master Data
    korean_name VARCHAR(100) NOT NULL,
    english_name VARCHAR(100),
    sector VARCHAR(50),
    industry VARCHAR(100),
    listing_date DATE,
    
    -- Status
    listing_status VARCHAR(20) NOT NULL DEFAULT 'Active', -- Active, Suspended, Delisted
    market VARCHAR(20) NOT NULL, -- 'KOSPI', 'KOSDAQ', 'KONEX'
    
    -- PIT
    published_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    revision INT NOT NULL DEFAULT 1,
    removed_at TIMESTAMP,
    
    -- Audit
    last_updated TIMESTAMP,
    data_source VARCHAR(50),
    
    CONSTRAINT unique_company UNIQUE (symbol, revision)
);

CREATE INDEX idx_companies_symbol ON market_data.companies(symbol);

Table: market_data.ingestion_jobs (Audit)

CREATE TABLE market_data.ingestion_jobs (
    -- Identity
    job_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    job_run_id UUID NOT NULL, -- Hangfire RunId
    
    -- Input
    data_source VARCHAR(50) NOT NULL,
    from_date DATE NOT NULL,
    to_date DATE NOT NULL,
    
    -- Progress
    status VARCHAR(50) NOT NULL DEFAULT 'Queued', -- Queued, Running, Completed, Failed
    rows_processed INT DEFAULT 0,
    rows_failed INT DEFAULT 0,
    rows_skipped INT DEFAULT 0,
    
    -- Timing
    started_at TIMESTAMP,
    completed_at TIMESTAMP,
    duration_seconds INT,
    
    -- Error Handling
    last_error_message TEXT,
    retry_count INT DEFAULT 0,
    
    -- Traceability
    correlation_id UUID NOT NULL,
    triggered_by VARCHAR(100), -- 'Scheduler', 'Manual', 'API'
    
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    
    CONSTRAINT unique_job_run UNIQUE (job_run_id)
);

CREATE INDEX idx_ingestion_jobs_status ON market_data.ingestion_jobs(status);
CREATE INDEX idx_ingestion_jobs_dates ON market_data.ingestion_jobs(from_date, to_date);

---

Read Model (Denormalized Projections)

View: market_data.latest_prices (Cache)

CREATE VIEW market_data.latest_prices AS
SELECT DISTINCT ON (symbol)
    symbol,
    trading_date,
    close_price,
    volume,
    published_at
FROM market_data.daily_prices
WHERE removed_at IS NULL
  AND published_at <= CURRENT_TIMESTAMP
ORDER BY symbol, trading_date DESC;

---

PIT (Point-in-Time) Query Pattern

-- Fetch prices as of 2026-06-30
SELECT symbol, open_price, close_price, volume
FROM market_data.daily_prices
WHERE trading_date <= '2026-06-30'
  AND published_at <= '2026-06-30'::timestamp
  AND removed_at IS NULL
ORDER BY symbol, trading_date DESC
LIMIT 1 PER symbol;

---

Migration Strategy

1. 0033_market_data_schema.sql

   • Create market_data schema
   • Define daily_prices, indices, companies, ingestion_jobs tables
   • Add PK, FK, constraints

2. 0034_market_data_indexes.sql

   • Create performance indexes
   • Partition by year (optional, if 10M+ rows/year)

3. 0035_market_data_audit.sql

   • Create audit trigger (log all writes)
   • Set up row-level security (market access control)

---

Data Dictionary

Column	Type	Purpose
symbol	VARCHAR(20)	Stock ticker (e.g., '005930' for Samsung)
trading_date	DATE	Market trading date (YYYY-MM-DD)
open_price	DECIMAL(10,2)	Opening price
close_price	DECIMAL(10,2)	Closing price
volume	BIGINT	Trading volume (shares)
published_at	TIMESTAMP	PIT anchor (when row became "true")
revision	INT	Version number (immutable history)
removed_at	TIMESTAMP	Soft-delete marker (NULL = active)
correlation_id	UUID	Trace this data ingestion back to job

---

Idempotency & Upsert Strategy

Idempotency Key: (symbol, trading_date)

Upsert SQL:

INSERT INTO market_data.daily_prices (symbol, trading_date, open_price, high_price, low_price, close_price, volume, published_at, revision, correlation_id, data_source)
VALUES (@symbol, @date, @open, @high, @low, @close, @volume, CURRENT_TIMESTAMP, 1, @corrId, @source)
ON CONFLICT (symbol, trading_date, revision) DO UPDATE SET
  open_price = EXCLUDED.open_price,
  close_price = EXCLUDED.close_price,
  volume = EXCLUDED.volume,
  published_at = CURRENT_TIMESTAMP,
  revision = market_data.daily_prices.revision + 1
WHERE EXCLUDED.published_at > market_data.daily_prices.published_at;

Effect: Same-day re-ingestion updates the row; older data is immutable (PIT principle).

---

Testing & Validation

Unit Tests (SQL):

• Constraints enforced (negative prices rejected)
• Unique keys prevent duplicates
• Soft-delete preserves history
• PIT query returns correct version

Integration Tests:

• Ingest 100 rows, verify count
• Duplicate ingestion (same date/symbol) increments revision
• Upsert with newer timestamp overwrites