Insurance OLAP Analytics Pipeline

-- Insurance Star Schema
CREATE TABLE fact_claims (
    claim_id BIGINT PRIMARY KEY,
    policy_id BIGINT,
    claim_amount DECIMAL(15,2),
    claim_date DATE,
    settlement_date DATE,
    claim_status VARCHAR(50),
    claim_type VARCHAR(100),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE dim_policy (
    policy_id BIGINT PRIMARY KEY,
    customer_id BIGINT,
    product_id BIGINT,
    policy_number VARCHAR(50),
    start_date DATE,
    end_date DATE,
    premium_amount DECIMAL(15,2),
    coverage_amount DECIMAL(15,2),
    status VARCHAR(20)
);

CREATE TABLE dim_customer (
    customer_id BIGINT PRIMARY KEY,
    first_name VARCHAR(100),
    last_name VARCHAR(100),
    date_of_birth DATE,
    risk_profile VARCHAR(20),
    credit_score INTEGER,
    address_id BIGINT
);

-- Create indexes for OLAP performance
CREATE INDEX idx_claims_policy_date ON fact_claims(policy_id, claim_date);
CREATE INDEX idx_claims_amount ON fact_claims(claim_amount);
CREATE INDEX idx_policy_customer ON dim_policy(customer_id);

-- OLAP Cube Views
CREATE VIEW v_claims_analytics AS
SELECT 
    DATE_TRUNC('month', c.claim_date) as month,
    p.product_line,
    p.coverage_type,
    c.claim_type,
    COUNT(*) as claim_count,
    SUM(c.claim_amount) as total_claims,
    AVG(c.claim_amount) as avg_claim_amount,
    SUM(c.claim_amount) / NULLIF(SUM(p.premium_amount), 0) as loss_ratio
FROM fact_claims c
JOIN dim_policy p ON c.policy_id = p.policy_id
GROUP BY 
    DATE_TRUNC('month', c.claim_date),
    p.product_line,
    p.coverage_type,
    c.claim_type;

-- Materialized view for performance
CREATE MATERIALIZED VIEW mv_monthly_claims_summary AS
SELECT * FROM v_claims_analytics;

-- Refresh materialized view
REFRESH MATERIALIZED VIEW mv_monthly_claims_summary;

// Real-time Claims Processing with Flink
public class ClaimsStreamProcessor extends KeyedProcessFunction<String, ClaimEvent, FraudAlert> {
    
    private ValueState<ClaimHistory> claimHistoryState;
    private ValueState<CustomerProfile> customerProfileState;
    
    @Override
    public void processElement(ClaimEvent claim, Context ctx, Collector<FraudAlert> out) throws Exception {
        // Get historical data
        ClaimHistory history = claimHistoryState.value();
        CustomerProfile profile = customerProfileState.value();
        
        // Real-time fraud detection
        FraudScore score = calculateFraudScore(claim, history, profile);
        
        if (score.getRiskLevel() == RiskLevel.HIGH) {
            FraudAlert alert = FraudAlert.builder()
                .claimId(claim.getClaimId())
                .customerId(claim.getCustomerId())
                .riskScore(score.getScore())
                .indicators(score.getIndicators())
                .timestamp(Instant.now())
                .build();
                
            out.collect(alert);
        }
        
        // Update state
        updateClaimHistory(claim, history);
    }
    
    private FraudScore calculateFraudScore(ClaimEvent claim, ClaimHistory history, CustomerProfile profile) {
        double score = 0.0;
        List<String> indicators = new ArrayList<>();
        
        // Frequency analysis
        if (history.getClaimCount() > profile.getExpectedClaims()) {
            score += 30.0;
            indicators.add("HIGH_CLAIM_FREQUENCY");
        }
        
        // Amount analysis
        if (claim.getAmount() > profile.getAverageClaimAmount() * 3) {
            score += 25.0;
            indicators.add("UNUSUAL_CLAIM_AMOUNT");
        }
        
        // Timing analysis
        if (isSuspiciousTiming(claim, history)) {
            score += 20.0;
            indicators.add("SUSPICIOUS_TIMING");
        }
        
        return new FraudScore(score, indicators);
    }
}

-- Data Lineage Tracking
CREATE TABLE data_lineage (
    lineage_id BIGINT PRIMARY KEY,
    source_table VARCHAR(255),
    target_table VARCHAR(255),
    transformation_type VARCHAR(100),
    transformation_sql TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    created_by VARCHAR(100)
);

-- Data Quality Monitoring
CREATE TABLE data_quality_checks (
    check_id BIGINT PRIMARY KEY,
    table_name VARCHAR(255),
    column_name VARCHAR(255),
    check_type VARCHAR(100),
    check_sql TEXT,
    threshold_value DECIMAL(10,2),
    current_value DECIMAL(10,2),
    status VARCHAR(20),
    last_run TIMESTAMP,
    alert_sent BOOLEAN DEFAULT FALSE
);

-- Compliance Monitoring
CREATE VIEW v_compliance_status AS
SELECT 
    'GDPR' as regulation,
    COUNT(CASE WHEN consent_status = 'ACTIVE' THEN 1 END) as compliant_records,
    COUNT(*) as total_records,
    ROUND(COUNT(CASE WHEN consent_status = 'ACTIVE' THEN 1 END) * 100.0 / COUNT(*), 2) as compliance_rate
FROM customer_consents
WHERE consent_type = 'DATA_PROCESSING'
UNION ALL
SELECT 
    'Data Retention' as regulation,
    COUNT(CASE WHEN retention_status = 'COMPLIANT' THEN 1 END) as compliant_records,
    COUNT(*) as total_records,
    ROUND(COUNT(CASE WHEN retention_status = 'COMPLIANT' THEN 1 END) * 100.0 / COUNT(*), 2) as compliance_rate
FROM data_retention_policies;