+------------------+
|   Source Systems  |
|  (OLTP, APIs)    |
+--------+---------+
         |
         v
+------------------+
|   ETL Pipeline   |
| (Extract,        |
|  Transform, Load)|
+--------+---------+
         |
         v
+------------------+
|   Data Warehouse |
|   (Star Schema)  |
+--------+---------+
         |
         v
+------------------+
|   OLAP Cubes     |
|   (Multi-        |
|    dimensional)  |
+--------+---------+
         |
         v
+------------------+
|   Reporting &    |
|   Analytics      |
+------------------+

                    +------------------+
                    |   Time Dimension |
                    |   (Date, Month,  |
                    |    Quarter, Year)|
                    +--------+---------+
                             |
+------------------+         |         +------------------+
|   Customer       |---------+---------|   Trade Fact     |
|   Dimension      |         |         |   (Trade_ID,     |
|   (Customer_ID,  |         |         |    Date_ID,      |
|    Name, Region) |         |         |    Customer_ID,  |
+------------------+         |         |    Instrument_ID,|
                             |         |    Quantity,     |
+------------------+         |         |    Price, Amount)|
|   Instrument     |---------+         +--------+---------+
|   Dimension      |         |                  |
|   (Instrument_ID,|         |                  |
|    Symbol, Type) |         |         +--------+---------+
+------------------+         |         |   Account         |
                             |         |   Dimension       |
                             +---------|   (Account_ID,    |
                                       |    Account_Type,  |
                                       |    Currency)      |
                                       +------------------+

-- Trade Fact Table (central fact)
CREATE TABLE trade_fact (
    trade_id BIGINT PRIMARY KEY,
    date_id INTEGER NOT NULL,
    customer_id INTEGER NOT NULL,
    account_id INTEGER NOT NULL,
    instrument_id INTEGER NOT NULL,
    trade_type_id INTEGER NOT NULL,
    side VARCHAR(4) NOT NULL,
    quantity INTEGER NOT NULL,
    price DECIMAL(18,4) NOT NULL,
    amount DECIMAL(18,2) NOT NULL,
    commission DECIMAL(18,2) DEFAULT 0.00,
    execution_latency_ms INTEGER,
    FOREIGN KEY (date_id) REFERENCES time_dimension(date_id),
    FOREIGN KEY (customer_id) REFERENCES customer_dimension(customer_id),
    FOREIGN KEY (account_id) REFERENCES account_dimension(account_id),
    FOREIGN KEY (instrument_id) REFERENCES instrument_dimension(instrument_id),
    FOREIGN KEY (trade_type_id) REFERENCES trade_type_dimension(trade_type_id)
);

-- Daily aggregate fact (for performance)
CREATE TABLE daily_aggregate_fact (
    date_id INTEGER NOT NULL,
    instrument_id INTEGER NOT NULL,
    total_volume INTEGER,
    total_value DECIMAL(18,2),
    trade_count INTEGER,
    vwap DECIMAL(18,4),
    open DECIMAL(18,4),
    high DECIMAL(18,4),
    low DECIMAL(18,4),
    close DECIMAL(18,4),
    PRIMARY KEY (date_id, instrument_id)
);

-- Time Dimension
CREATE TABLE time_dimension (
    date_id INTEGER PRIMARY KEY,
    full_date DATE NOT NULL,
    year INTEGER NOT NULL,
    quarter INTEGER NOT NULL,
    month INTEGER NOT NULL,
    month_name VARCHAR(20),
    day INTEGER NOT NULL,
    day_of_week INTEGER NOT NULL,
    day_name VARCHAR(20),
    week_of_year INTEGER,
    is_weekend BOOLEAN DEFAULT FALSE,
    is_holiday BOOLEAN DEFAULT FALSE
);

-- Customer Dimension
CREATE TABLE customer_dimension (
    customer_id INTEGER PRIMARY KEY,
    customer_key VARCHAR(50) UNIQUE,
    first_name VARCHAR(100),
    last_name VARCHAR(100),
    email VARCHAR(255),
    region VARCHAR(50),
    country VARCHAR(50),
    customer_type VARCHAR(20),
    risk_tier INTEGER,
    acquisition_date DATE,
    status VARCHAR(20),
    valid_from DATE,
    valid_to DATE,
    is_current BOOLEAN DEFAULT TRUE
);

-- Instrument Dimension
CREATE TABLE instrument_dimension (
    instrument_id INTEGER PRIMARY KEY,
    symbol VARCHAR(20) NOT NULL,
    name VARCHAR(255),
    instrument_type VARCHAR(20),
    sector VARCHAR(50),
    industry VARCHAR(50),
    exchange VARCHAR(20),
    currency VARCHAR(3),
    isin VARCHAR(12),
    market_cap DECIMAL(18,2)
);

-- Trade Type Dimension
CREATE TABLE trade_type_dimension (
    trade_type_id INTEGER PRIMARY KEY,
    trade_type VARCHAR(30),
    order_type VARCHAR(20),
    time_in_force VARCHAR(20),
    venue VARCHAR(50)
);

                    +------------------+
                    |   Time Dimension |
                    +--------+---------+
                             |
+------------------+         |         +------------------+
|   Customer       |---------+---------|   Trade Fact     |
|   Dimension      |         |         |                  |
+--------+---------+         |         +------------------+
         |                   |                   |
         v                   |                   v
+------------------+         |         +------------------+
|   Region         |         |         |   Account        |
|   Dimension      |---------+         |   Dimension      |
+------------------+                   +--------+---------+
                                                   |
                                                   v
                                        +------------------+
                                        |   Currency       |
                                        |   Dimension      |
                                        +------------------+

import pandas as pd
from sqlalchemy import create_engine

class OLAPCube:
    """
    Implements an OLAP cube for financial data analysis.
    """
    
    def __init__(self, connection_string):
        self.engine = create_engine(connection_string)
        self.data = None
        self.dimensions = {}
        self.measures = {}
    
    def build_cube(self, fact_table, dimensions, measures, date_range=None):
        """
        Build an OLAP cube from the fact table.
        """
        # Build query
        query = f"SELECT * FROM {fact_table}"
        if date_range:
            query += f" WHERE date_id BETWEEN {date_range[0]} AND {date_range[1]}"
        
        # Load data
        self.data = pd.read_sql(query, self.engine)
        self.dimensions = dimensions
        self.measures = measures
        
        print(f"Cube built with {len(self.data)} records")
    
    def roll_up(self, dimension, level):
        """
        Roll up to a higher level of aggregation.
        """
        # Group by the selected dimension
        if dimension not in self.dimensions:
            raise ValueError(f"Dimension {dimension} not found")
        
        # Aggregate measures
        agg_dict = {measure: ['sum', 'avg', 'count'] for measure in self.measures}
        result = self.data.groupby(dimension).agg(agg_dict)
        result.columns = ['_'.join(col).strip() for col in result.columns.values]
        return result
    
    def drill_down(self, dimension, value, sub_dimension):
        """
        Drill down to a more detailed level.
        """
        filtered = self.data[self.data[dimension] == value]
        if sub_dimension in self.dimensions:
            return filtered.groupby(sub_dimension).agg({
                measure: ['sum', 'avg'] for measure in self.measures
            })
        return filtered
    
    def slice(self, dimension, value):
        """
        Slice the cube by selecting a single value from a dimension.
        """
        return self.data[self.data[dimension] == value]
    
    def dice(self, **conditions):
        """
        Dice the cube by selecting multiple values from multiple dimensions.
        """
        result = self.data.copy()
        for dim, values in conditions.items():
            result = result[result[dim].isin(values)]
        return result
    
    def pivot(self, rows, columns, values, aggfunc='sum'):
        """
        Create a pivot table for analysis.
        """
        pivot = pd.pivot_table(
            self.data,
            values=values,
            index=rows,
            columns=columns,
            aggfunc=aggfunc,
            fill_value=0
        )
        return pivot

# Usage
cube = OLAPCube('postgresql://user:pass@localhost/warehouse')
cube.build_cube(
    fact_table='trade_fact',
    dimensions=['date_id', 'customer_id', 'instrument_id'],
    measures=['amount', 'commission']
)

# Roll up by month
monthly_trades = cube.roll_up('date_id', 'month')

# Slice by customer
customer_trades = cube.slice('customer_id', 12345)

# Pivot table
pivot_table = cube.pivot(
    rows=['date_id'],
    columns=['instrument_id'],
    values=['amount']
)

-- Roll-up: Total trading volume by month and instrument
SELECT 
    d.year,
    d.month,
    i.symbol,
    SUM(t.quantity) AS total_volume,
    SUM(t.amount) AS total_value,
    COUNT(*) AS trade_count
FROM trade_fact t
JOIN time_dimension d ON t.date_id = d.date_id
JOIN instrument_dimension i ON t.instrument_id = i.instrument_id
GROUP BY GROUPING SETS (
    (d.year, d.month, i.symbol),
    (d.year, d.month),
    (d.year),
    ()
)
ORDER BY d.year, d.month, i.symbol;

-- Drill-down: Detailed trades for a specific date
SELECT 
    t.trade_id,
    c.customer_name,
    a.account_number,
    i.symbol,
    t.side,
    t.quantity,
    t.price,
    t.amount
FROM trade_fact t
JOIN customer_dimension c ON t.customer_id = c.customer_id
JOIN account_dimension a ON t.account_id = a.account_id
JOIN instrument_dimension i ON t.instrument_id = i.instrument_id
WHERE t.date_id = 20240115;

-- Slice and dice: Trades for specific criteria
SELECT 
    d.year,
    d.month,
    i.sector,
    SUM(t.amount) AS total_amount
FROM trade_fact t
JOIN time_dimension d ON t.date_id = d.date_id
JOIN instrument_dimension i ON t.instrument_id = i.instrument_id
WHERE i.sector IN ('Technology', 'Finance')
AND d.year = 2024
GROUP BY d.year, d.month, i.sector
ORDER BY d.month, i.sector;

-- Customer dimension with history tracking
CREATE TABLE customer_dimension_scd (
    customer_id INTEGER NOT NULL,
    customer_key VARCHAR(50) NOT NULL,
    customer_name VARCHAR(255) NOT NULL,
    email VARCHAR(255),
    region VARCHAR(50),
    risk_tier INTEGER,
    status VARCHAR(20),
    valid_from DATE NOT NULL,
    valid_to DATE,
    is_current BOOLEAN DEFAULT TRUE,
    version INTEGER DEFAULT 1,
    PRIMARY KEY (customer_id, valid_from)
);

-- Function to handle SCD Type 2 updates
CREATE OR REPLACE FUNCTION update_customer_dimension()
RETURNS TRIGGER AS $$
BEGIN
    -- Close the current record
    UPDATE customer_dimension_scd
    SET valid_to = CURRENT_DATE,
        is_current = FALSE
    WHERE customer_id = NEW.customer_id
    AND is_current = TRUE;
    
    -- Insert the new record
    INSERT INTO customer_dimension_scd (
        customer_id, customer_key, customer_name, email, region, risk_tier, status,
        valid_from, valid_to, is_current, version
    )
    VALUES (
        NEW.customer_id, NEW.customer_key, NEW.customer_name, NEW.email, NEW.region,
        NEW.risk_tier, NEW.status, CURRENT_DATE, NULL, TRUE,
        (SELECT COALESCE(MAX(version), 0) + 1 FROM customer_dimension_scd 
         WHERE customer_id = NEW.customer_id)
    );
    
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- Trigger for automatic history tracking
CREATE TRIGGER customer_dimension_update
BEFORE INSERT ON customer_dimension_scd
FOR EACH ROW
EXECUTE FUNCTION update_customer_dimension();

-- Simple update (no history)
CREATE OR REPLACE FUNCTION update_customer_scd1()
RETURNS TRIGGER AS $$
BEGIN
    -- Overwrite the existing record
    UPDATE customer_dimension
    SET 
        customer_name = NEW.customer_name,
        email = NEW.email,
        region = NEW.region,
        risk_tier = NEW.risk_tier,
        status = NEW.status,
        updated_at = CURRENT_TIMESTAMP
    WHERE customer_id = NEW.customer_id;
    
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- Fact: Trade Execution
CREATE TABLE fact_trade_execution (
    execution_id BIGINT PRIMARY KEY,
    date_id INTEGER NOT NULL,
    time_id INTEGER NOT NULL,
    customer_id INTEGER NOT NULL,
    account_id INTEGER NOT NULL,
    instrument_id INTEGER NOT NULL,
    trade_type_id INTEGER NOT NULL,
    venue_id INTEGER NOT NULL,
    side VARCHAR(4) NOT NULL,
    quantity INTEGER NOT NULL,
    price DECIMAL(18,4) NOT NULL,
    amount DECIMAL(18,2) NOT NULL,
    commission DECIMAL(18,2),
    execution_latency_ms INTEGER,
    FOREIGN KEY (date_id) REFERENCES dim_time(date_id),
    FOREIGN KEY (time_id) REFERENCES dim_time_of_day(time_id),
    FOREIGN KEY (customer_id) REFERENCES dim_customer(customer_id),
    FOREIGN KEY (account_id) REFERENCES dim_account(account_id),
    FOREIGN KEY (instrument_id) REFERENCES dim_instrument(instrument_id),
    FOREIGN KEY (trade_type_id) REFERENCES dim_trade_type(trade_type_id),
    FOREIGN KEY (venue_id) REFERENCES dim_venue(venue_id)
);

-- Fact: Daily Position
CREATE TABLE fact_daily_position (
    position_id BIGINT PRIMARY KEY,
    date_id INTEGER NOT NULL,
    account_id INTEGER NOT NULL,
    instrument_id INTEGER NOT NULL,
    quantity INTEGER NOT NULL,
    average_cost DECIMAL(18,4),
    mark_price DECIMAL(18,4),
    unrealized_pnl DECIMAL(18,2),
    realized_pnl DECIMAL(18,2),
    FOREIGN KEY (date_id) REFERENCES dim_time(date_id),
    FOREIGN KEY (account_id) REFERENCES dim_account(account_id),
    FOREIGN KEY (instrument_id) REFERENCES dim_instrument(instrument_id)
);

-- Fact: Risk Metrics
CREATE TABLE fact_risk_metrics (
    risk_id BIGINT PRIMARY KEY,
    date_id INTEGER NOT NULL,
    account_id INTEGER NOT NULL,
    portfolio_id INTEGER,
    var_95 DECIMAL(18,2),
    var_99 DECIMAL(18,2),
    expected_shortfall DECIMAL(18,2),
    leverage_ratio DECIMAL(10,4),
    concentration_ratio DECIMAL(10,4),
    FOREIGN KEY (date_id) REFERENCES dim_time(date_id),
    FOREIGN KEY (account_id) REFERENCES dim_account(account_id)
);

-- Dimension: Time of Day
CREATE TABLE dim_time_of_day (
    time_id INTEGER PRIMARY KEY,
    hour INTEGER NOT NULL,
    minute INTEGER NOT NULL,
    second INTEGER DEFAULT 0,
    time_of_day VARCHAR(20),
    trading_session VARCHAR(20)
);

-- Dimension: Venue
CREATE TABLE dim_venue (
    venue_id INTEGER PRIMARY KEY,
    venue_name VARCHAR(50) NOT NULL,
    venue_type VARCHAR(20),
    country VARCHAR(50),
    timezone VARCHAR(50)
);

-- Dimension: Portfolio
CREATE TABLE dim_portfolio (
    portfolio_id INTEGER PRIMARY KEY,
    portfolio_name VARCHAR(100),
    portfolio_type VARCHAR(20),
    manager_id INTEGER,
    strategy VARCHAR(50)
);

import pandas as pd
from datetime import datetime
import psycopg2

class DataWarehouseETL:
    """
    ETL process for loading the financial data warehouse.
    """
    
    def __init__(self, source_conn, target_conn):
        self.source = source_conn
        self.target = target_conn
        self.batch_size = 10000
    
    def extract_trades(self, last_etl_date):
        """
        Extract trades from the source system.
        """
        query = f"""
        SELECT 
            trade_id,
            trade_date,
            trade_time,
            customer_id,
            account_id,
            instrument_id,
            side,
            quantity,
            price,
            amount,
            commission,
            execution_latency_ms
        FROM source_trades
        WHERE trade_date > '{last_etl_date}'
        """
        return pd.read_sql(query, self.source)
    
    def transform_data(self, df):
        """
        Transform data for the warehouse.
        """
        # Convert dates
        df['date_id'] = pd.to_datetime(df['trade_date']).dt.strftime('%Y%m%d').astype(int)
        
        # Create time_id
        df['time_id'] = (pd.to_datetime(df['trade_time']).dt.hour * 3600 + 
                        pd.to_datetime(df['trade_time']).dt.minute * 60 +
                        pd.to_datetime(df['trade_time']).dt.second)
        
        # Map trade_type_id (example mapping)
        trade_type_map = {
            'MARKET': 1,
            'LIMIT': 2,
            'STOP': 3,
            'STOP_LIMIT': 4
        }
        df['trade_type_id'] = df['order_type'].map(trade_type_map)
        
        # Map venue_id
        venue_map = {
            'NASDAQ': 1,
            'NYSE': 2,
            'CME': 3,
            'ICE': 4
        }
        df['venue_id'] = df['venue'].map(venue_map)
        
        # Clean data
        df = df.dropna(subset=['trade_id', 'date_id'])
        df = df[df['quantity'] > 0]
        
        return df
    
    def load_data(self, df, table_name):
        """
        Load data into the warehouse.
        """
        cursor = self.target.cursor()
        
        for i in range(0, len(df), self.batch_size):
            batch = df.iloc[i:i+self.batch_size]
            records = batch.to_dict('records')
            
            # Build insert statement
            columns = batch.columns.tolist()
            placeholders = ', '.join(['%s'] * len(columns))
            column_names = ', '.join(columns)
            
            query = f"INSERT INTO {table_name} ({column_names}) VALUES ({placeholders})"
            
            # Execute batch insert
            cursor.executemany(query, [tuple(record.values()) for record in records])
            
        self.target.commit()
        cursor.close()
    
    def run_etl(self, last_etl_date):
        """
        Run the complete ETL process.
        """
        print(f"Starting ETL from {last_etl_date}")
        
        # Extract
        print("Extracting data...")
        df = self.extract_trades(last_etl_date)
        print(f"Extracted {len(df)} records")
        
        if len(df) == 0:
            print("No new data to process")
            return
        
        # Transform
        print("Transforming data...")
        df = self.transform_data(df)
        print(f"Transformed {len(df)} records")
        
        # Load
        print("Loading data into warehouse...")
        self.load_data(df, 'fact_trade_execution')
        
        print(f"ETL completed successfully. Loaded {len(df)} records")

-- Daily instrument summary
CREATE TABLE agg_daily_instrument_summary AS
SELECT 
    date_id,
    instrument_id,
    SUM(quantity) AS total_volume,
    SUM(amount) AS total_value,
    COUNT(*) AS trade_count,
    AVG(price) AS vwap,
    SUM(commission) AS total_commission
FROM fact_trade_execution
GROUP BY date_id, instrument_id;

-- Monthly customer summary
CREATE TABLE agg_monthly_customer_summary AS
SELECT 
    d.year,
    d.month,
    c.customer_id,
    c.customer_name,
    COUNT(*) AS trade_count,
    SUM(t.amount) AS total_value,
    SUM(t.commission) AS total_commission,
    AVG(t.quantity) AS avg_trade_size
FROM fact_trade_execution t
JOIN dim_time d ON t.date_id = d.date_id
JOIN dim_customer c ON t.customer_id = c.customer_id
GROUP BY d.year, d.month, c.customer_id, c.customer_name;

-- Refresh aggregates
CREATE OR REPLACE FUNCTION refresh_aggregates()
RETURNS VOID AS $$
BEGIN
    TRUNCATE agg_daily_instrument_summary;
    INSERT INTO agg_daily_instrument_summary
    SELECT 
        date_id,
        instrument_id,
        SUM(quantity) AS total_volume,
        SUM(amount) AS total_value,
        COUNT(*) AS trade_count,
        AVG(price) AS vwap,
        SUM(commission) AS total_commission
    FROM fact_trade_execution
    GROUP BY date_id, instrument_id;
END;
$$ LANGUAGE plpgsql;

-- Bitmap indexes for low-cardinality columns
CREATE INDEX idx_fact_trade_side ON fact_trade_execution USING BITMAP (side);
CREATE INDEX idx_dim_customer_type ON dim_customer USING BITMAP (customer_type);
CREATE INDEX idx_dim_instrument_sector ON dim_instrument USING BITMAP (sector);

-- Composite indexes for common query patterns
CREATE INDEX idx_fact_trade_date_instrument ON fact_trade_execution(date_id, instrument_id);
CREATE INDEX idx_fact_trade_date_customer ON fact_trade_execution(date_id, customer_id);
CREATE INDEX idx_fact_trade_customer_instrument ON fact_trade_execution(customer_id, instrument_id);

-- Covering indexes for aggregation queries
CREATE INDEX idx_fact_trade_cover ON fact_trade_execution(date_id, instrument_id) 
INCLUDE (quantity, amount);