VWAP Demo — Quick Start Guide

-- Create dummy table for time window calculations
CREATE TABLE dummy (x VARCHAR(1) NOT NULL, PRIMARY KEY (x));

-- Create the OHLCV table with VWAP
CREATE TABLE stocktick_vmap
(symbol varchar(10) not null
,tickdate timestamp not null
,timescale varchar(10) not null
,open_price decimal not null
,high_price decimal not null
,low_price decimal not null
,close_price decimal not null
,volume decimal not null
,total_value decimal not null
,vmap decimal
,primary key (symbol,tickdate,timescale));

PARTITION TABLE stocktick_vmap ON COLUMN symbol;

-- Session VWAP state table
CREATE TABLE vwap_state (
    symbol VARCHAR(10) NOT NULL,
    session_start TIMESTAMP NOT NULL,
    cumulative_value_volume DECIMAL,
    cumulative_volume DECIMAL,
    last_update TIMESTAMP,
    PRIMARY KEY (symbol, session_start)
);

PARTITION TABLE vwap_state ON COLUMN symbol;

-- Create stream for tick data export
CREATE STREAM ALL_TICKS
  PARTITION ON COLUMN SYMBOL
  EXPORT TO TARGET ALERTS (
    Symbol varchar(10) NOT NULL,
    Tickdate timestamp NOT NULL,
    Value decimal NOT NULL,
    Volume decimal NOT NULL
);

-- Summary view for monitoring
CREATE VIEW TICKS_SUMMARY (
  Symbol, Ticks_count
) AS SELECT Symbol, count(*)
   FROM ALL_TICKS GROUP BY symbol;

-- Register the stored procedure
CREATE PROCEDURE
   PARTITION ON TABLE stocktick_vmap COLUMN symbol
   FROM CLASS com.voltactivedata.voltdb.storedProcedures
              .ReportTickSessionAnchor;

-- Initialize dummy table (required for TIME_WINDOW calculations)
INSERT INTO dummy VALUES ('X');

/*
 * Copyright (C) 2025 Volt Active Data Inc.
 */
package com.voltactivedata.voltdb.storedProcedures;

import java.math.RoundingMode;
import java.math.BigDecimal;
import org.voltdb.SQLStmt;
import org.voltdb.VoltProcedure;
import org.voltdb.VoltTable;
import org.voltdb.VoltTableRow;
import org.voltdb.types.TimestampType;

public class ReportTickSessionAnchor extends VoltProcedure {

    private static final String STOCKTICK_COLUMNS = "(symbol, tickdate, timescale, open_price"
            + ", high_price, low_price, close_price, volume, total_value, vmap) ";
    private static final String MINUTE = "MINUTES_1";
    private static final String HOUR = "MINUTES_60";
    private static final String MINUTES_15 = "MINUTES_15";
    private static final String INSERT_TICK_START = "INSERT INTO stocktick_vmap " + STOCKTICK_COLUMNS
            + "VALUES (?,TIME_WINDOW(";

    private static final String INSERT_TICK_END = ",?,?,?,?,?,?,?,?);";
    private static final String MINUTE_INSERT = "MINUTE,1,?)";
    private static final String MINUTES_15_INSERT = "MINUTE,15,?)";
    private static final String HOUR_INSERT = "MINUTE,60,?)";

    public static final SQLStmt getRollingMinute = new SQLStmt(
            "SELECT TIME_WINDOW(MINUTE,1,?) TICKDATE FROM dummy WHERE x = 'X';");

    public static final SQLStmt getSessionTime = new SQLStmt(
            "SELECT DATEADD(MINUTE,570,TIME_WINDOW(DAY, 1, ?)) tickdate FROM dummy WHERE x = 'X';");

    public static final SQLStmt get15Minutes = new SQLStmt(
            "SELECT * FROM stocktick_vmap WHERE symbol = ? AND tickdate = TIME_WINDOW(MINUTE,15,?) AND timescale = '"+MINUTES_15+"';");

    public static final SQLStmt getMinute = new SQLStmt(
            "SELECT * FROM stocktick_vmap WHERE symbol = ? AND tickdate = TIME_WINDOW(MINUTE,1,?) AND timescale = '"+MINUTE+"';");

    public static final SQLStmt getHour = new SQLStmt(
            "SELECT * FROM stocktick_vmap WHERE symbol = ? AND tickdate = TIME_WINDOW(MINUTE,60,?) AND timescale = '"+HOUR+"';");

    public static final SQLStmt insertTickMinute = new SQLStmt(
            INSERT_TICK_START+MINUTE_INSERT+INSERT_TICK_END);

    public static final SQLStmt insertTick15Minutes = new SQLStmt(
            INSERT_TICK_START+MINUTES_15_INSERT+INSERT_TICK_END);

    public static final SQLStmt insertTickHour = new SQLStmt(
            INSERT_TICK_START+HOUR_INSERT+INSERT_TICK_END);

    public static final SQLStmt updateTick = new SQLStmt(
            "UPDATE stocktick_vmap "
                    + "SET low_price = ?, high_price = ?, close_price = ?, volume = ?, total_value = ?, vmap = ? "
                    + "WHERE symbol = ? AND tickdate = ? AND timescale = ? ;");

    public final SQLStmt SELECT_VWAP_STATE = new SQLStmt(
            "SELECT symbol, session_start, cumulative_value_volume, cumulative_volume " +
                    "FROM vwap_state WHERE symbol = ? and session_start = (SELECT DATEADD(MINUTE,570,TIME_WINDOW(DAY, 1, ?)) tickdate FROM dummy WHERE x = 'X');"
    );

    public final SQLStmt UPSERT_VWAP_STATE = new SQLStmt(
            "UPSERT INTO vwap_state (symbol, session_start, cumulative_value_volume, " +
                    "cumulative_volume, last_update) " +
                    "VALUES (?, ?, ?, ?, ?);"
    );

    public final SQLStmt INSERT_ALLTICKS_STREAM = new SQLStmt("INSERT INTO ALL_TICKS (Symbol, Tickdate, Value, Volume) VALUES (?, ?, ?, ?);");

    final String[] theTimes = {HOUR, MINUTES_15, MINUTE};
    final SQLStmt[] theGets = {getHour, get15Minutes, getMinute,};
    final SQLStmt[] theInserts = {insertTickHour, insertTick15Minutes, insertTickMinute};

    private BigDecimal calculateVWAP(BigDecimal cumulativeValueVolume, BigDecimal cumulativeVolume) {
        if (cumulativeVolume.compareTo(BigDecimal.ZERO) == 0) {
            return BigDecimal.ZERO;
        }
        return cumulativeValueVolume.divide(cumulativeVolume, 6, RoundingMode.HALF_UP);
    }

    private BigDecimal getAndUpdateSessionVWAP(TimestampType sessionTime, VoltTable[] vwap_state, String symbol, TimestampType tickdate, BigDecimal value, BigDecimal volume) {
        TimestampType currentSessionStart = sessionTime;
        VoltTable[] results = vwap_state;

        BigDecimal sessionCumulativeValueVolume = BigDecimal.ZERO;
        BigDecimal sessionCumulativeVolume = BigDecimal.ZERO;
        TimestampType storedSessionStart = null;

        if (results[0].getRowCount() > 0) {
            VoltTableRow row = results[0].fetchRow(0);
            storedSessionStart = row.getTimestampAsTimestamp(1);

            if (storedSessionStart.equals(currentSessionStart)) {
                sessionCumulativeValueVolume = row.getDecimalAsBigDecimal(2);
                sessionCumulativeVolume = row.getDecimalAsBigDecimal(3);
            }
        }

        BigDecimal tradeValueVolume = value.multiply(volume);
        sessionCumulativeValueVolume = sessionCumulativeValueVolume.add(tradeValueVolume);
        sessionCumulativeVolume = sessionCumulativeVolume.add(volume);

        BigDecimal sessionVWAP = calculateVWAP(sessionCumulativeValueVolume, sessionCumulativeVolume);

        voltQueueSQL(UPSERT_VWAP_STATE, symbol, currentSessionStart,
                sessionCumulativeValueVolume, sessionCumulativeVolume, tickdate);

        return sessionVWAP;
    }

    public VoltTable[] run(String symbol, TimestampType tickdate, BigDecimal value, BigDecimal volume)
            throws VoltAbortException {

        voltQueueSQL(SELECT_VWAP_STATE, symbol, tickdate);
        for (int i = 0; i < theTimes.length; i++) {
            voltQueueSQL(theGets[i], symbol, tickdate);
        }
        voltQueueSQL(getRollingMinute, tickdate);
        voltQueueSQL(getSessionTime, tickdate);

        VoltTable[] currentTicks = voltExecuteSQL();

        currentTicks[currentTicks.length - 2].advanceRow();
        final TimestampType rollingTickdate = currentTicks[currentTicks.length - 2].getTimestampAsTimestamp("TICKDATE");

        currentTicks[currentTicks.length - 1].advanceRow();
        final TimestampType sessionTime = currentTicks[currentTicks.length - 1].getTimestampAsTimestamp("TICKDATE");

        BigDecimal sessionVWAP = getAndUpdateSessionVWAP(sessionTime, currentTicks, symbol, tickdate, value, volume);

        for (int i = 1; i < theTimes.length + 1; i++) {
            if (!currentTicks[i].advanceRow()) {
                BigDecimal vmap = sessionVWAP;
                voltQueueSQL(theInserts[i - 1], symbol, tickdate, theTimes[i - 1], value, value, value, value, volume,
                        value.multiply(volume), vmap);
            } else {
                final TimestampType actualTickdate = currentTicks[i].getTimestampAsTimestamp("TICKDATE");
                BigDecimal low = currentTicks[i].getDecimalAsBigDecimal("LOW_PRICE");
                if (value.compareTo(low) == -1) {
                    low = value;
                }
                BigDecimal high = currentTicks[i].getDecimalAsBigDecimal("HIGH_PRICE");
                if (value.compareTo(high) == 1) {
                    high = value;
                }
                final BigDecimal newEntryVolume = volume.add(currentTicks[i].getDecimalAsBigDecimal("VOLUME"));
                final BigDecimal extraTotalValue = value.multiply(volume);
                final BigDecimal newEntryTotalValue = extraTotalValue
                        .add(currentTicks[i].getDecimalAsBigDecimal("TOTAL_VALUE"));
                BigDecimal newVmap = sessionVWAP;
                voltQueueSQL(updateTick, low, high, value, newEntryVolume, newEntryTotalValue, newVmap, symbol,
                        actualTickdate, theTimes[i - 1]);
            }
        }
        voltQueueSQL(INSERT_ALLTICKS_STREAM, symbol, tickdate, value, volume);

        return voltExecuteSQL(true);
    }
}

# config.yaml - VoltSP Pipeline Configuration
version: 1
name: "kafka-to-voltdb-with-js"

resources:
  - name: voltdb
    voltdb-client:
      servers:
        - localhost:21212

source:
  kafka:
    bootstrapServers:
      - "localhost:9092"
    topicNames: "ticker-data"
    groupId: "vwap-consumer"
    startingOffset: "EARLIEST"
    keyDeserializer: "org.apache.kafka.common.serialization.StringDeserializer"
    valueDeserializer: "com.voltactivedata.vwapdemo.kafka.TickerDataDeserializer"

processors:
  - javascript:
      script: |
        function process(input) {
          var tickerData = input.getValue();

          // Import Java types
          var VoltProcedureRequest = Java.type(
            'org.voltdb.stream.plugin.volt.api.VoltProcedureRequest');
          var VoltTimestamp = Java.type('org.voltdb.types.TimestampType');
          var BigDecimal = Java.type('java.math.BigDecimal');

          // Build procedure call
          return new VoltProcedureRequest(
            "ReportTickSessionAnchor",
            [
              tickerData.getSymbol(),
              new VoltTimestamp(+tickerData.getEventDate()),
              new BigDecimal(tickerData.getPrice()),
              new BigDecimal(tickerData.getVolume())
            ]
          );
        }

sink:
  voltdb-procedure:
    voltClientResource: "voltdb"

# Run with:
# ./voltsp -l ~/license.xml --metrics-port 11799 --parallelism=1 config.yaml

-- View latest candlesticks with VWAP
SELECT symbol, timescale, tickdate, open_price, high_price,
       low_price, close_price, volume, vmap AS session_vwap
FROM stocktick_vmap
WHERE timescale = 'MINUTES_1'
ORDER BY tickdate DESC
LIMIT 20;

-- View session VWAP state
SELECT symbol, session_start, cumulative_value_volume,
       cumulative_volume, last_update
FROM vwap_state
ORDER BY last_update DESC;

-- Check tick throughput
SELECT Symbol, Ticks_count
FROM TICKS_SUMMARY
ORDER BY Ticks_count DESC;