institutional-trader/README/QUERY_ENHANCEMENT_ANALYSIS.md

Query Enhancement Analysis & Recommendations

Institutional Trading Platform - Next-Level Optimizations

---

Executive Summary

Your queries are well-structured with sophisticated CTEs and window functions. To reach institutional-grade performance, focus on:

1. Strategic Indexing - Composite indexes for multi-column filters
2. Query Architecture - Materialized views for expensive aggregations
3. Performance Monitoring - Query explain plans and execution time tracking
4. Advanced Caching - Multi-tier caching with intelligent invalidation
5. Query Optimization - LATERAL join optimizations and partition pruning

---

1. CRITICAL INDEX OPTIMIZATIONS

Current State

• Basic single-column indexes exist
• Missing composite indexes for common query patterns
• No partial indexes for filtered queries

Recommended Indexes

For OptionsFlow_monthly Table

-- Composite index for date range + symbol queries (most common pattern)
CREATE INDEX IF NOT EXISTS idx_ofm_date_symbol_premium 
ON "OptionsFlow_monthly"("CreatedDate", "Symbol", "Premium" DESC)
WHERE "Premium" IS NOT NULL AND "Premium"::numeric > 0;

-- Composite index for expiration + symbol + strike (for moneyness calculations)
CREATE INDEX IF NOT EXISTS idx_ofm_exp_symbol_strike 
ON "OptionsFlow_monthly"("ExpirationDate", "Symbol", "Strike"::numeric);

-- Partial index for high-premium flows (filters 80% of data)
CREATE INDEX IF NOT EXISTS idx_ofm_high_premium 
ON "OptionsFlow_monthly"("CreatedDate", "Symbol", "Premium" DESC)
WHERE "Premium"::numeric > 80000;

-- Index for CallPut + Side normalization (used in every query)
CREATE INDEX IF NOT EXISTS idx_ofm_cp_side 
ON "OptionsFlow_monthly"(UPPER(TRIM("CallPut")), UPPER(TRIM("Side")));

-- GIN index for text search on Symbol (if doing fuzzy matching)
CREATE INDEX IF NOT EXISTS idx_ofm_symbol_gin 
ON "OptionsFlow_monthly" USING gin("Symbol" gin_trgm_ops);

For prices_intraday_1m Table

-- Composite index for symbol + timestamp (used in LATERAL joins)
CREATE INDEX IF NOT EXISTS idx_prices_symbol_ts_desc 
ON prices_intraday_1m(symbol, ts DESC)
WHERE symbol IS NOT NULL;

-- Partial index for recent prices (last 7 days - most queries)
CREATE INDEX IF NOT EXISTS idx_prices_recent 
ON prices_intraday_1m(symbol, ts DESC)
WHERE ts >= NOW() - INTERVAL '7 days';

-- Index for session-based queries (RTH, PRE, POST)
CREATE INDEX IF NOT EXISTS idx_prices_session 
ON prices_intraday_1m(symbol, ts)
WHERE EXTRACT(HOUR FROM ts AT TIME ZONE 'America/Chicago') BETWEEN 4 AND 20;

For AlertStream_monthly Table

-- Composite index for ticker + event time (alert matching)
CREATE INDEX IF NOT EXISTS idx_alert_ticker_time 
ON "AlertStream_monthly"("ticker", "date", "timestamp");

-- Partial index for recent alerts (within 24 hours)
CREATE INDEX IF NOT EXISTS idx_alert_recent 
ON "AlertStream_monthly"("ticker", "date", "timestamp")
WHERE "date" >= CURRENT_DATE - INTERVAL '1 day';

-- Index for alert type filtering
CREATE INDEX IF NOT EXISTS idx_alert_type 
ON "AlertStream_monthly"("type", "date", "ticker");

Index Maintenance

-- Analyze tables after index creation
ANALYZE "OptionsFlow_monthly";
ANALYZE prices_intraday_1m;
ANALYZE "AlertStream_monthly";

-- Check index usage (run periodically)
SELECT 
  schemaname,
  tablename,
  indexname,
  idx_scan as index_scans,
  pg_size_pretty(pg_relation_size(indexrelid)) as index_size
FROM pg_stat_user_indexes
WHERE schemaname = 'public'
ORDER BY idx_scan DESC;

---

2. MATERIALIZED VIEWS FOR EXPENSIVE AGGREGATIONS

Problem

Your optionsFlowQuery recalculates running sums, badges, and scores for every request. These can be pre-computed.

Solution: Materialized Views

-- Materialized view for daily flow aggregations
CREATE MATERIALIZED VIEW IF NOT EXISTS mv_daily_flow_agg AS
SELECT
  (flow_ts_local)::date AS flow_date,
  symbol_norm,
  exp_date,
  
  -- Aggregated metrics
  SUM(CASE WHEN cp_norm='CALL' AND side_norm='BUY' AND moneyness='OTM' THEN premium_num ELSE 0 END) AS prem_cb_otm_total,
  SUM(CASE WHEN cp_norm='CALL' AND side_norm='BUY' AND moneyness='ITM' THEN premium_num ELSE 0 END) AS prem_cb_itm_total,
  SUM(CASE WHEN cp_norm='PUT' AND side_norm='BUY' AND moneyness='OTM' THEN premium_num ELSE 0 END) AS prem_pb_otm_total,
  SUM(CASE WHEN cp_norm='PUT' AND side_norm='BUY' AND moneyness='ITM' THEN premium_num ELSE 0 END) AS prem_pb_itm_total,
  
  SUM(vol_num) AS vol_total,
  SUM(oi_num) AS oi_total,
  
  COUNT(*) AS flow_count,
  MAX(flow_ts_utc) AS last_flow_time
  
FROM (
  -- Your base CTE logic here (simplified)
  SELECT 
    symbol_norm,
    exp_date,
    cp_norm,
    side_norm,
    premium_num,
    vol_num,
    oi_num,
    flow_ts_local,
    flow_ts_utc,
    CASE
      WHEN cp_norm='CALL' AND strike_num > spot_num THEN 'OTM'
      WHEN cp_norm='CALL' AND strike_num <= spot_num THEN 'ITM'
      WHEN cp_norm='PUT' AND strike_num < spot_num THEN 'OTM'
      WHEN cp_norm='PUT' AND strike_num >= spot_num THEN 'ITM'
    END AS moneyness
  FROM "OptionsFlow_monthly" ofm
  WHERE ofm."Premium" IS NOT NULL
    AND ofm."StockEtf" = 'STOCK'
) base
GROUP BY (flow_ts_local)::date, symbol_norm, exp_date;

-- Index on materialized view
CREATE INDEX IF NOT EXISTS idx_mv_daily_flow_date_symbol 
ON mv_daily_flow_agg(flow_date DESC, symbol_norm);

-- Refresh strategy (run every 15 minutes during market hours)
CREATE OR REPLACE FUNCTION refresh_daily_flow_agg()
RETURNS void AS $$
BEGIN
  REFRESH MATERIALIZED VIEW CONCURRENTLY mv_daily_flow_agg;
END;
$$ LANGUAGE plpgsql;

Usage in Queries

// Instead of recalculating, join with materialized view
const query = `
  WITH base AS (
    SELECT * FROM mv_daily_flow_agg
    WHERE flow_date BETWEEN $1::date AND $2::date
  )
  SELECT 
    b.*,
    -- Add per-row calculations here
    CASE WHEN b.prem_cb_itm_total > b.prem_pb_itm_total THEN '🟢' ELSE '🔴' END AS badge_round
  FROM base b
  WHERE b.prem_cb_itm_total + b.prem_pb_itm_total > $3::numeric
`;

---

3. QUERY PERFORMANCE MONITORING

Add Query Explain Plan Endpoint

// backend/src/routes/performance.js (add this)
router.post('/explain', async (req, res) => {
  try {
    const { query, params = [] } = req.body;
    
    if (!query) {
      return res.status(400).json({ error: 'Query is required' });
    }
    
    // Get explain plan
    const explainQuery = `EXPLAIN (ANALYZE, BUFFERS, VERBOSE, FORMAT JSON) ${query}`;
    const explainResult = await rawQuery(explainQuery, params);
    
    // Get execution time
    const timingQuery = `EXPLAIN (ANALYZE, TIMING, FORMAT JSON) ${query}`;
    const timingResult = await rawQuery(timingQuery, params);
    
    res.json({
      success: true,
      explain: explainResult[0]?.query_plan || explainResult,
      timing: timingResult[0]?.query_plan || timingResult,
      recommendations: analyzeExplainPlan(explainResult)
    });
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
});

function analyzeExplainPlan(plan) {
  const recommendations = [];
  const planStr = JSON.stringify(plan);
  
  // Check for sequential scans
  if (planStr.includes('Seq Scan')) {
    recommendations.push({
      severity: 'HIGH',
      issue: 'Sequential scan detected',
      fix: 'Add appropriate index or use index hint'
    });
  }
  
  // Check for high cost
  if (planStr.includes('"Total Cost"') && parseFloat(planStr.match(/"Total Cost":\s*(\d+)/)?.[1]) > 100000) {
    recommendations.push({
      severity: 'MEDIUM',
      issue: 'High query cost',
      fix: 'Consider materialized view or query optimization'
    });
  }
  
  return recommendations;
}

Query Execution Time Tracking

// backend/src/utils/queryProfiler.js
export class QueryProfiler {
  static async profile(queryFn, queryName) {
    const start = process.hrtime.bigint();
    const startMemory = process.memoryUsage().heapUsed;
    
    try {
      const result = await queryFn();
      const end = process.hrtime.bigint();
      const endMemory = process.memoryUsage().heapUsed;
      
      const duration = Number(end - start) / 1_000_000; // milliseconds
      const memoryDelta = (endMemory - startMemory) / 1024 / 1024; // MB
      
      // Log slow queries
      if (duration > 1000) {
        console.warn(`⚠️  Slow query detected: ${queryName} took ${duration.toFixed(2)}ms`);
      }
      
      return {
        result,
        metrics: {
          duration,
          memoryDelta,
          queryName
        }
      };
    } catch (error) {
      const end = process.hrtime.bigint();
      const duration = Number(end - start) / 1_000_000;
      
      console.error(`❌ Query failed: ${queryName} after ${duration.toFixed(2)}ms`, error);
      throw error;
    }
  }
}

// Usage in routes
import { QueryProfiler } from '../utils/queryProfiler.js';

const { result: rawData, metrics } = await QueryProfiler.profile(
  () => rawQuery(optionsFlowQuery, [startDate, endDate]),
  'optionsFlowQuery'
);

---

4. ADVANCED CACHING STRATEGY

Current State

• Basic 30-second cache exists
• No cache invalidation strategy
• No cache warming

Enhanced Caching Implementation

// backend/src/middleware/cache.js (enhanced)
import NodeCache from 'node-cache';
import { rawQuery } from '../db.js';

const cache = new NodeCache({ 
  stdTTL: 60, // 60 seconds default
  checkperiod: 30,
  useClones: false, // Better performance for large objects
  maxKeys: 1000
});

// Cache with intelligent TTL based on market hours
export function smartCacheMiddleware() {
  return (req, res, next) => {
    if (req.method !== 'GET') {
      return next();
    }
    
    const key = generateCacheKey(req);
    const cached = cache.get(key);
    
    if (cached) {
      res.set('X-Cache', 'HIT');
      return res.json(cached);
    }
    
    res.set('X-Cache', 'MISS');
    res.originalJson = res.json;
    res.json = (body) => {
      const ttl = getCacheTTL(req);
      cache.set(key, body, ttl);
      res.originalJson(body);
    };
    
    next();
  };
}

function generateCacheKey(req) {
  const { startDate, endDate, minPremium, ...filters } = req.query;
  return `query:${req.path}:${JSON.stringify(filters)}`;
}

function getCacheTTL(req) {
  const now = new Date();
  const hour = now.getHours();
  const isMarketHours = hour >= 9 && hour < 16;
  
  // Shorter cache during market hours (15s), longer after hours (5min)
  return isMarketHours ? 15 : 300;
}

// Cache warming for common queries
export async function warmCache() {
  const today = new Date().toISOString().split('T')[0];
  const yesterday = new Date(Date.now() - 86400000).toISOString().split('T')[0];
  
  // Pre-fetch common queries
  const commonQueries = [
    { path: '/api/options/flow', params: { startDate: today, endDate: today } },
    { path: '/api/scanner/multi-signal', params: {} }
  ];
  
  // This would be called by a cron job or on server start
  console.log('🔥 Warming cache...');
}

Redis Integration (Optional - for production scale)

// backend/src/middleware/redisCache.js
import Redis from 'ioredis';

const redis = new Redis(process.env.REDIS_URL);

export async function redisCache(key, ttl, fetchFn) {
  const cached = await redis.get(key);
  if (cached) {
    return JSON.parse(cached);
  }
  
  const data = await fetchFn();
  await redis.setex(key, ttl, JSON.stringify(data));
  return data;
}

---

5. QUERY OPTIMIZATION TECHNIQUES

A. Optimize LATERAL Joins

Your price_ctx CTE uses multiple LATERAL joins. Optimize with:

-- Instead of multiple LATERAL joins, use a single subquery with window functions
price_ctx_optimized AS (
  SELECT
    c.rid,
    c.symbol_norm,
    c.flow_ts_utc,
    
    -- Use window functions to get nearest price in one pass
    (SELECT p.close 
     FROM prices_intraday_1m p
     WHERE UPPER(p.symbol) = c.symbol_norm
       AND p.ts <= c.flow_ts_utc
     ORDER BY p.ts DESC
     LIMIT 1) AS u_close,
    
    -- Use array aggregation for multiple lookups
    (SELECT array_agg(p.close ORDER BY p.ts DESC)
     FROM prices_intraday_1m p
     WHERE UPPER(p.symbol) = c.symbol_norm
       AND p.ts <= c.flow_ts_utc
       AND p.ts >= c.flow_ts_utc - INTERVAL '15 minutes'
     LIMIT 5) AS recent_closes
  FROM rocketize c
)

B. Partition Pruning (if using table partitioning)

-- If OptionsFlow_monthly is partitioned by month
-- Add partition pruning hints
SELECT /*+ USE_PARTITION_HINT */ * 
FROM "OptionsFlow_monthly"
WHERE "CreatedDate" BETWEEN $1 AND $2;

C. Query Hints for Complex Joins

-- Force index usage for specific patterns
SET enable_seqscan = off; -- For specific query session
-- Your query here
SET enable_seqscan = on;

---

6. QUERY MODULARITY & MAINTAINABILITY

Current Issue

• 662-line monolithic query in optionsFlowQuery.js
• Hard to test individual components
• Difficult to optimize specific parts

Solution: Query Builder Pattern

// backend/src/queries/builders/optionsFlowBuilder.js
export class OptionsFlowQueryBuilder {
  constructor() {
    this.ctes = [];
    this.filters = [];
    this.selects = [];
  }
  
  withBase() {
    this.ctes.push(`
      base AS (
        SELECT
          ofm.ctid AS rid,
          ofm.*,
          UPPER(TRIM(ofm."Symbol")) AS symbol_norm,
          -- ... base logic
        FROM public."OptionsFlow_monthly" ofm
        WHERE ofm."Premium" IS NOT NULL
      )
    `);
    return this;
  }
  
  withFlow(startDate, endDate) {
    this.ctes.push(`
      flow AS (
        SELECT b.*, (b.flow_ts_local)::date AS flow_date_cst
        FROM base b
        WHERE (b.flow_ts_local)::date BETWEEN $1::date AND $2::date
      )
    `);
    return this;
  }
  
  withPriceContext() {
    this.ctes.push(`
      price_ctx AS (
        -- Price context logic
      )
    `);
    return this;
  }
  
  build() {
    return `
      WITH ${this.ctes.join(',\n')}
      SELECT ${this.selects.join(',\n')}
      FROM ${this.getFinalCTE()}
      ${this.buildWhere()}
      ${this.buildOrderBy()}
      LIMIT $3::integer
    `;
  }
  
  buildWhere() {
    if (this.filters.length === 0) return '';
    return `WHERE ${this.filters.join(' AND ')}`;
  }
  
  buildOrderBy() {
    return 'ORDER BY flow_ts_utc DESC, rid DESC';
  }
}

// Usage
const query = new OptionsFlowQueryBuilder()
  .withBase()
  .withFlow(startDate, endDate)
  .withPriceContext()
  .build();

---

7. CONNECTION POOLING & QUERY TIMEOUTS

Enhanced Database Configuration

// backend/src/db.js (enhancements)
import { Pool } from 'pg';

const pgPool = new Pool({
  connectionString: process.env.DATABASE_URL,
  max: 20, // Maximum pool size
  min: 5,  // Minimum pool size
  idleTimeoutMillis: 30000,
  connectionTimeoutMillis: 10000,
  
  // Query timeout (prevent runaway queries)
  statement_timeout: 30000, // 30 seconds
  
  // Application name for monitoring
  application_name: 'institutional_trader_backend'
});

// Add query timeout wrapper
export async function rawQueryWithTimeout(sql, params = [], timeoutMs = 30000) {
  const client = await pgPool.connect();
  
  try {
    // Set statement timeout for this query
    await client.query(`SET statement_timeout = ${timeoutMs}`);
    
    const result = await client.query(sql, params);
    return result.rows;
  } catch (error) {
    if (error.code === '57014') { // Statement timeout
      throw new Error(`Query timeout after ${timeoutMs}ms`);
    }
    throw error;
  } finally {
    await client.query('RESET statement_timeout');
    client.release();
  }
}

---

8. QUERY RESULT STREAMING (for large datasets)

Current Issue

• Loading all results into memory
• High memory usage for large date ranges

Solution: Streaming Results

// backend/src/utils/queryStream.js
import { Readable } from 'stream';

export async function* streamQuery(sql, params) {
  const client = await pgPool.connect();
  
  try {
    const query = new QueryStream(sql, params);
    const stream = client.query(query);
    
    for await (const row of stream) {
      yield row;
    }
  } finally {
    client.release();
  }
}

// Usage in route
router.get('/flow/stream', async (req, res) => {
  res.setHeader('Content-Type', 'application/json');
  res.write('[');
  
  let first = true;
  for await (const row of streamQuery(optionsFlowQuery, [startDate, endDate])) {
    if (!first) res.write(',');
    res.write(JSON.stringify(row));
    first = false;
  }
  
  res.write(']');
  res.end();
});

---

9. QUERY VALIDATION & ERROR HANDLING

Enhanced Error Handling

// backend/src/utils/queryValidator.js
export function validateQueryParams(params) {
  const errors = [];
  
  if (params.startDate && !isValidDate(params.startDate)) {
    errors.push('Invalid startDate format. Use YYYY-MM-DD');
  }
  
  if (params.endDate && !isValidDate(params.endDate)) {
    errors.push('Invalid endDate format. Use YYYY-MM-DD');
  }
  
  if (params.startDate && params.endDate) {
    const start = new Date(params.startDate);
    const end = new Date(params.endDate);
    const daysDiff = (end - start) / (1000 * 60 * 60 * 24);
    
    if (daysDiff > 90) {
      errors.push('Date range cannot exceed 90 days');
    }
    
    if (start > end) {
      errors.push('startDate must be before endDate');
    }
  }
  
  return errors;
}

// Usage
const errors = validateQueryParams(req.query);
if (errors.length > 0) {
  return res.status(400).json({ errors });
}

---

10. IMPLEMENTATION PRIORITY

Phase 1: Critical (Immediate Impact)

1. ✅ Add composite indexes (Section 1)
2. ✅ Implement query profiling (Section 3)
3. ✅ Add query timeouts (Section 7)

Phase 2: High Value (This Week)

4. ✅ Enhanced caching (Section 4)
5. ✅ Optimize LATERAL joins (Section 5A)
6. ✅ Query validation (Section 9)

Phase 3: Architecture (Next Sprint)

7. ✅ Materialized views (Section 2)
8. ✅ Query builder pattern (Section 6)
9. ✅ Result streaming (Section 8)

---

11. MONITORING & ALERTS

Query Performance Dashboard

// backend/src/routes/performance.js (add)
router.get('/metrics', async (req, res) => {
  const metrics = {
    poolStats: pgPool.totalCount,
    idleConnections: pgPool.idleCount,
    waitingCount: pgPool.waitingCount,
    cacheStats: getCacheStats(),
    slowQueries: getSlowQueries(), // Track queries > 1s
    errorRate: getErrorRate()
  };
  
  res.json(metrics);
});

---

Expected Performance Improvements

Optimization	Expected Improvement	Implementation Effort
Composite Indexes	50-80% faster queries	Low (30 min)
Materialized Views	90% faster aggregations	Medium (2-3 hours)
Query Profiling	Identify bottlenecks	Low (1 hour)
Enhanced Caching	70% reduction in DB load	Medium (2 hours)
LATERAL Join Optimization	30-40% faster price lookups	Medium (2 hours)
Connection Pooling	Better concurrency	Low (30 min)

---

Next Steps

1. Run index creation scripts (Section 1) - Test in dev first
2. Add query profiling to identify actual bottlenecks
3. Implement enhanced caching for immediate wins
4. Create materialized views for daily aggregations
5. Monitor and iterate based on real performance data

---

Questions?

If you need help implementing any of these, I can:

• Generate the exact SQL for your schema
• Create the JavaScript modules
• Set up monitoring dashboards
• Optimize specific slow queries

Let me know which area you want to tackle first!