# 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

```sql
-- 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

```sql
-- 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

```sql
-- 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

```sql
-- 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

```sql
-- 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

```javascript
// 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

```javascript
// 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

```javascript
// 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

```javascript
// 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)

```javascript
// 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:

```sql
-- 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)

```sql
-- 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

```sql
-- 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

```javascript
// 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

```javascript
// 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

```javascript
// 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

```javascript
// 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

```javascript
// 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!