snowflake_setup.sql

-- Generic ingestion control plane and stock-market reference model.
-- This script is rerunnable and preserves previously ingested data.

CREATE DATABASE IF NOT EXISTS STOCK_MARKET;
CREATE SCHEMA IF NOT EXISTS STOCK_MARKET.PUBLIC;
CREATE SCHEMA IF NOT EXISTS STOCK_MARKET.ANALYTICS;
USE DATABASE STOCK_MARKET;
USE SCHEMA PUBLIC;

CREATE TABLE IF NOT EXISTS INGESTION_WATERMARKS (
    JOB_NAME VARCHAR NOT NULL,
    CURSOR_VALUE VARIANT,
    UPDATED_AT TIMESTAMP_TZ NOT NULL DEFAULT CURRENT_TIMESTAMP(),
    CONSTRAINT PK_INGESTION_WATERMARKS PRIMARY KEY (JOB_NAME)
);

CREATE TABLE IF NOT EXISTS DIM_SECTOR (
    SECTOR VARCHAR(100) NOT NULL,
    LOADED_AT TIMESTAMP_TZ NOT NULL,
    CONSTRAINT PK_DIM_SECTOR PRIMARY KEY (SECTOR)
);

CREATE TABLE IF NOT EXISTS DIM_COMPANY (
    SYMBOL VARCHAR(20) NOT NULL,
    COMPANY_NAME VARCHAR(250),
    SECTOR VARCHAR(100),
    SUB_INDUSTRY VARCHAR(250),
    HEADQUARTERS VARCHAR(250),
    DATE_ADDED DATE,
    CIK VARCHAR(20),
    SOURCE_UPDATED_AT TIMESTAMP_TZ,
    LOADED_AT TIMESTAMP_TZ NOT NULL,
    CONSTRAINT PK_DIM_COMPANY PRIMARY KEY (SYMBOL),
    CONSTRAINT FK_COMPANY_SECTOR FOREIGN KEY (SECTOR) REFERENCES DIM_SECTOR (SECTOR)
);

CREATE TABLE IF NOT EXISTS FACT_MARKET_BARS (
    SYMBOL VARCHAR(20) NOT NULL,
    OBSERVED_AT TIMESTAMP_TZ NOT NULL,
    TRADE_DATE DATE NOT NULL,
    OPEN NUMBER(20, 6),
    HIGH NUMBER(20, 6),
    LOW NUMBER(20, 6),
    CLOSE NUMBER(20, 6),
    ADJUSTED_CLOSE NUMBER(20, 6),
    VOLUME NUMBER(38, 0),
    DIVIDENDS NUMBER(20, 8),
    STOCK_SPLITS NUMBER(20, 8),
    SOURCE_INTERVAL VARCHAR(10) NOT NULL,
    EXTRACTED_AT TIMESTAMP_TZ NOT NULL,
    LOADED_AT TIMESTAMP_TZ NOT NULL,
    CONSTRAINT PK_FACT_MARKET_BARS
        PRIMARY KEY (SYMBOL, OBSERVED_AT, SOURCE_INTERVAL),
    CONSTRAINT FK_MARKET_BAR_COMPANY FOREIGN KEY (SYMBOL) REFERENCES DIM_COMPANY (SYMBOL)
);

CREATE TABLE IF NOT EXISTS DIM_DATE (
    DATE_DAY DATE NOT NULL,
    CALENDAR_YEAR NUMBER(4, 0) NOT NULL,
    CALENDAR_QUARTER NUMBER(1, 0) NOT NULL,
    MONTH_NUMBER NUMBER(2, 0) NOT NULL,
    MONTH_NAME VARCHAR(9) NOT NULL,
    WEEK_OF_YEAR NUMBER(2, 0) NOT NULL,
    DAY_OF_WEEK NUMBER(1, 0) NOT NULL,
    DAY_NAME VARCHAR(9) NOT NULL,
    IS_WEEKEND BOOLEAN NOT NULL,
    CONSTRAINT PK_DIM_DATE PRIMARY KEY (DATE_DAY)
);

MERGE INTO DIM_DATE AS target
USING (
    SELECT
        DATEADD(DAY, SEQ4(), '2000-01-01'::DATE) AS DATE_DAY
    FROM TABLE(GENERATOR(ROWCOUNT => 21915))
) AS source
ON target.DATE_DAY = source.DATE_DAY
WHEN NOT MATCHED THEN INSERT (
    DATE_DAY, CALENDAR_YEAR, CALENDAR_QUARTER, MONTH_NUMBER, MONTH_NAME,
    WEEK_OF_YEAR, DAY_OF_WEEK, DAY_NAME, IS_WEEKEND
) VALUES (
    source.DATE_DAY,
    YEAR(source.DATE_DAY),
    QUARTER(source.DATE_DAY),
    MONTH(source.DATE_DAY),
    MONTHNAME(source.DATE_DAY),
    WEEKOFYEAR(source.DATE_DAY),
    DAYOFWEEKISO(source.DATE_DAY),
    DAYNAME(source.DATE_DAY),
    DAYOFWEEKISO(source.DATE_DAY) IN (6, 7)
);