alx/ATS_Esper
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

add OHLC plugin

Browse Source
This commit is contained in:
Seth Ladygo
2018-05-26 23:19:12 -07:00
parent 476f3708f4
commit 26b199a4f9
9 changed files with 734 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
src/main/java/EsperProcessor.java
View File

@ -1,17 +1,24 @@
import ats.plugin.OHLCPlugInViewFactory;
import ats.plugin.OHLCTick;
import ats.plugin.OHLCUpdateListener;
import ats.plugin.OHLCValue;
import com.espertech.esper.client.Configuration; import com.espertech.esper.client.Configuration;
import com.espertech.esper.client.EPServiceProvider; import com.espertech.esper.client.EPServiceProvider;
import com.espertech.esper.client.EPServiceProviderManager; import com.espertech.esper.client.EPServiceProviderManager;
import com.espertech.esper.client.EPStatement; import com.espertech.esper.client.EPStatement;
import java.math.BigDecimal; import com.espertech.esper.client.EventBean;
import java.io.File; import com.espertech.esper.client.StatementAwareUpdateListener;
import com.espertech.esper.client.UpdateListener;
import com.espertech.esper.client.time.CurrentTimeEvent; import com.espertech.esper.client.time.CurrentTimeEvent;
import org.slf4j.Logger; import java.io.File;
import org.slf4j.LoggerFactory;
import java.lang.reflect.Array; import java.lang.reflect.Array;
import java.math.BigDecimal;
import java.util.Arrays; import java.util.Arrays;
import java.util.stream.Stream; import java.util.Map;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import org.joda.time.DateTime; import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class EsperProcessor implements TickProcessor { public class EsperProcessor implements TickProcessor {
@ -36,12 +43,6 @@ public class EsperProcessor implements TickProcessor {
addStatements(epl); addStatements(epl);
// Map<String,ConfigurationVariable> vars = config.getVariables();
// log.debug("VAR MAP: {}", vars);
// for (Map.Entry<String, ConfigurationVariable> entry : vars.entrySet()) {
// log.debug("VAR: {} = {}", entry.getKey(), entry.getValue());
// }
addStatement("select * from TickEvent", addStatement("select * from TickEvent",
(newData, oldData) -> { (newData, oldData) -> {
log.debug("Tick: {}", newData[0].getUnderlying()); log.debug("Tick: {}", newData[0].getUnderlying());
@ -110,6 +111,13 @@ public class EsperProcessor implements TickProcessor {
return str.startsWith("//") || str.startsWith("--"); return str.startsWith("//") || str.startsWith("--");
} }
/**
* Return true if the given string is only whitespace.
*/
private static boolean isEmpty(String str) {
return str.matches("^\\s*$");
}
/** /**
* Remove comment lines from the string. * Remove comment lines from the string.
*/ */
@ -119,15 +127,38 @@ public class EsperProcessor implements TickProcessor {
.collect(Collectors.joining("\n")); .collect(Collectors.joining("\n"));
} }
/**
* Given a string with (possibly) multiple statements, split and
* add individually.
*/
private void addStatements(String epl) { private void addStatements(String epl) {
for (String s : splitStatements(filterComments(epl))) { for (String s : splitStatements(filterComments(epl))) {
if (!isEmpty(s))
addStatement(s); addStatement(s);
} }
} }
/**
* Add a single EPL statement to the Esper engine.
*/
private EPStatement addStatement(String epl) { private EPStatement addStatement(String epl) {
logger.debug("Adding statement: {}", epl); return addStatement(epl, (UpdateListener)null);
return engine.getEPAdministrator().createEPL(epl); }
/**
* Add a single EPL statement to the Esper engine with a listener
* to respond to the Statement.
*/
private EPStatement addStatement(String epl, UpdateListener listener) {
// log.debug("Adding statement: {}", epl);
EPStatement statement = engine.getEPAdministrator().createEPL(epl);
if (listener != null) {
statement.addListener(listener);
}
return statement;
} }
/** /**

275
src/main/java/ats/plugin/OHLCPlugInView.java Normal file
View File

@ -0,0 +1,275 @@
package ats.plugin;
import com.espertech.esper.client.EventBean;
import com.espertech.esper.client.EventType;
import com.espertech.esper.core.context.util.AgentInstanceViewFactoryChainContext;
import com.espertech.esper.core.service.EPStatementHandleCallback;
import com.espertech.esper.core.service.EngineLevelExtensionServicesContext;
import com.espertech.esper.epl.expression.core.ExprNode;
import com.espertech.esper.event.EventAdapterService;
import com.espertech.esper.schedule.ScheduleHandleCallback;
import com.espertech.esper.view.ViewSupport;
import java.util.Calendar;
import java.util.GregorianCalendar;
import java.util.Iterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.joda.time.Period;
import org.joda.time.format.PeriodFormatter;
import org.joda.time.format.PeriodFormatterBuilder;
import com.espertech.esper.epl.expression.core.ExprEvaluator;
import com.espertech.esper.schedule.SchedulingService;
import org.joda.time.Duration;
import org.joda.time.format.ISODateTimeFormat;
import java.util.TimeZone;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.joda.time.Instant;
import org.joda.time.PeriodType;
/**
* OHLCPlugInView computes OHLC bars for a given time interval.
*/
public class OHLCPlugInView extends ViewSupport {
private static final Logger log = LoggerFactory.getLogger(OHLCPlugInView.class);
private static final int LATE_EVENT_SLACK_SECONDS = 5;
private static final PeriodFormatter periodFormatter = new PeriodFormatterBuilder()
.appendDays().appendSuffix("d ")
.appendHours().appendSuffix("h ")
.appendMinutes().appendSuffix("m ")
.appendSeconds().appendSuffix("s")
.toFormatter();
private final AgentInstanceViewFactoryChainContext agentContext;
private final long scheduleSlot;
private EPStatementHandleCallback handle;
private final Duration interval;
private final ExprNode timestampExpression;
private final ExprNode valueExpression;
private DateTime windowStartTime;
private DateTime windowEndTime;
private Double open;
private Double close;
private Double high;
private Double low;
private EventBean[] lastData;
public OHLCPlugInView(AgentInstanceViewFactoryChainContext context,
ExprNode intervalExpression,
ExprNode timestampExpression,
ExprNode valueExpression)
{
agentContext = context;
scheduleSlot = context.getStatementContext().getScheduleBucket().allocateSlot();
interval = parseInterval(intervalExpression);
log.info("Interval is {}", interval);
this.timestampExpression = timestampExpression;
this.valueExpression = valueExpression;
}
/**
* Return the time period specified by the given expression value.
*/
private Duration parseInterval(ExprNode interval) {
ExprEvaluator evaluator = interval.getForge().getExprEvaluator();
String intervalStr = (String)evaluator.evaluate(null, true, agentContext);
return periodFormatter.parsePeriod(intervalStr).toStandardDuration();
}
/**
* Return the timestamp for the event.
*/
private DateTime getTimestamp(EventBean event) {
ExprEvaluator evaluator = timestampExpression.getForge().getExprEvaluator();
//Long l = (Long)evaluator.evaluate(new EventBean[] {event}, true, agentContext);
return (DateTime)evaluator.evaluate(new EventBean[] {event}, true, agentContext);
//return toDateTime(l);
}
/**
* Convert a bare long value to a proper DateTime entity. Assumes
* UTC time zone.
*/
public static DateTime toDateTime(long l) {
return new DateTime(l, DateTimeZone.UTC);
}
/**
* Return the value for the event.
*/
private double getValue(EventBean event) {
ExprEvaluator evaluator = valueExpression.getForge().getExprEvaluator();
return (double)evaluator.evaluate(new EventBean[] {event}, true, agentContext);
}
/**
* Notify that data has been added or removed from the Viewable parent.
*/
@Override
public void update(EventBean[] newData, EventBean[] oldData) {
if (newData == null) return;
for (EventBean event : newData) {
DateTime timestamp = getTimestamp(event);
double value = getValue(event);
ensureWindow(timestamp);
applyValue(value);
}
}
/**
* Make sure our window times are set up and current.
*/
private void ensureWindow(DateTime timestamp) {
if (timestamp == null) return;
if (windowStartTime == null) {
// create open window
windowStartTime = makeWindowStartTime(timestamp, interval);
windowEndTime = makeWindowEndTime(windowStartTime, interval);
}
if (!inWindow(timestamp)) {
// past current window.
// post and create a new one.
postData();
scheduleCallback();
windowStartTime = makeWindowStartTime(timestamp, interval);
windowEndTime = makeWindowEndTime(windowStartTime, interval);
}
}
public static DateTime makeWindowStartTime(DateTime timestamp,
Duration interval)
{
DateTime today = timestamp.withTimeAtStartOfDay();
// log.info("Timestamp is {}", timestamp);
// log.info("Day start is {}", today);
Duration intoToday = new Duration(today, timestamp);
// then modulo timestamp to find how far into the current window we'd be in
long intoPeriod = intoToday.getMillis() % interval.getMillis();
// then subtract modulo from current time to get window start
return timestamp.minus(intoPeriod);
}
private static DateTime makeWindowEndTime(DateTime startTime, Duration interval) {
if (startTime == null || interval == null) return null;
return startTime.plus(interval.getMillis());
}
/**
* Return true if the timestamp is within the current time window.
*/
private boolean inWindow(DateTime timestamp) {
if (timestamp == null) return false;
return timestamp.compareTo(windowStartTime) >= 0 &&
timestamp.compareTo(windowEndTime) < 0;
}
private void applyValue(double value) {
if (open == null) {
open = value;
}
close = value;
if (low == null) {
low = value;
} else if (low.compareTo(value) > 0) {
low = value;
}
if (high == null) {
high = value;
} else if (high.compareTo(value) < 0) {
high = value;
}
}
/**
* Set up a callback to post an event when our time window expires.
*/
private void scheduleCallback() {
SchedulingService sched = agentContext.getStatementContext().getSchedulingService();
if (handle != null) {
// remove old schedule
sched.remove(handle, scheduleSlot);
handle = null;
}
DateTime currentTime = toDateTime(sched.getTime());
DateTime targetTime = windowEndTime.plusSeconds(LATE_EVENT_SLACK_SECONDS);
long callbackTime = targetTime.getMillis() - currentTime.getMillis();
ScheduleHandleCallback callback = new ScheduleHandleCallback() {
public void scheduledTrigger(EngineLevelExtensionServicesContext esc) {
handle = null; // clear out schedule handle
OHLCPlugInView.this.postData();
}
};
handle = new EPStatementHandleCallback(agentContext.getEpStatementAgentInstanceHandle(), callback);
sched.add(callbackTime, handle, scheduleSlot);
}
/**
* Update listeners with our new value.
*/
private void postData() {
if (open == null) return;
OHLCValue value = new OHLCValue(windowStartTime, open, high, low, close);
// send
EventAdapterService service = agentContext.getStatementContext().getEventAdapterService();
EventBean[] newBeans = new EventBean[] {service.adapterForBean(value)};
updateChildren(newBeans, lastData);
// reset for next post
lastData = newBeans;
open = null;
close = null;
high = null;
low = null;
}
//
// for the EventCollection interface
//
/**
* Provides metadata information about the type of object the
* event collection contains.
*/
@Override
public EventType getEventType() {
return getEventType(agentContext.getStatementContext().getEventAdapterService());
}
protected static EventType getEventType(EventAdapterService service)
{
return service.addBeanType(OHLCValue.class.getName(),
OHLCValue.class,
false, false, false);
}
/**
* (Don't) allow iteration through all elements in this event collection.
*/
@Override
public Iterator<EventBean> iterator() {
throw new UnsupportedOperationException("Can't iterate.");
}
}

106
src/main/java/ats/plugin/OHLCPlugInViewFactory.java Normal file
View File

@ -0,0 +1,106 @@
package ats.plugin;
import com.espertech.esper.client.EventType;
import com.espertech.esper.core.context.util.AgentInstanceViewFactoryChainContext;
import com.espertech.esper.core.service.StatementContext;
import com.espertech.esper.epl.expression.core.ExprNode;
import com.espertech.esper.event.EventAdapterService;
import com.espertech.esper.view.View;
import com.espertech.esper.view.ViewFactory;
import com.espertech.esper.view.ViewFactoryContext;
import com.espertech.esper.view.ViewFactorySupport;
import com.espertech.esper.view.ViewParameterException;
import java.math.BigDecimal;
import java.util.Date;
import java.util.List;
import org.joda.time.DateTime;
/**
* OHLCPlugInViewFactory manages creation of OHLC plugin views.
*/
public class OHLCPlugInViewFactory extends ViewFactorySupport {
private EventAdapterService eventAdapterService;
private List<ExprNode> params;
private ExprNode interval;
private ExprNode timestamp;
private ExprNode value;
/**
* Pass in EPL query view params.
*/
public void setViewParameters(ViewFactoryContext context,
List<ExprNode> params)
throws ViewParameterException
{
eventAdapterService = context.getEventAdapterService();
if (params.size() != 3) {
throw new ViewParameterException("OHLC view takes three parameters: time interval, timestamp expression, and mid price expression.");
}
this.params = params;
}
/**
* Attach the factory to a parent event type such that the
* factory can validate attach requirements and determine an event
* type for resulting views.
*/
public void attach(EventType parentEventType,
StatementContext statementContext,
ViewFactory optionalParentFactory,
List<ViewFactory> parentViewFactories)
throws ViewParameterException
{
ExprNode[] validatedNodes = ViewFactorySupport.validate(getViewName(),
parentEventType,
statementContext,
params, true);
interval = validatedNodes[0];
timestamp = validatedNodes[1];
value = validatedNodes[2];
Class intervalClass = interval.getForge().getEvaluationType();
if ((intervalClass != String.class))
{
throw new ViewParameterException("OHLC view needs String-typed interval value for parameter 1.");
}
Class timestampClass = timestamp.getForge().getEvaluationType();
if ((timestampClass != DateTime.class))
{
throw new ViewParameterException("OHLC view needs DateTime typed timestamp values for parameter 2");
}
Class valueClass = value.getForge().getEvaluationType();
if ((valueClass != double.class) &&
(valueClass != Double.class) &&
(valueClass != BigDecimal.class))
{
throw new ViewParameterException("OHLC view needs double or BigDecimal values for parameter 3");
}
}
/**
* Create a new view using already-passed context and params.
*/
public View makeView(AgentInstanceViewFactoryChainContext agentContext) {
return new OHLCPlugInView(agentContext, interval, timestamp, value);
}
/**
* Returns the event type that the view that is created by the
* view factory would create for events posted by the view.
*/
public EventType getEventType() {
return OHLCPlugInView.getEventType(eventAdapterService);
}
/**
* Return the pretty name of our view type.
*/
public String getViewName() {
return OHLCPlugInView.class.getSimpleName();
}
}

47
src/main/java/ats/plugin/OHLCTick.java Normal file
View File

@ -0,0 +1,47 @@
/*
***************************************************************************************
* Copyright (C) 2006 EsperTech, Inc. All rights reserved. *
* http://www.espertech.com/esper *
* http://www.espertech.com *
* ---------------------------------------------------------------------------------- *
* The software in this package is published under the terms of the GPL license *
* a copy of which has been included with this distribution in the license.txt file. *
***************************************************************************************
*/
package ats.plugin;
import java.util.Date;
public class OHLCTick {
private String ticker;
private double price;
private long timestamp;
public OHLCTick(String ticker, double price, long timestamp) {
this.ticker = ticker;
this.price = price;
this.timestamp = timestamp;
}
public String getTicker() {
return ticker;
}
public double getPrice() {
return price;
}
public long getTimestamp() {
return timestamp;
}
public String toString() {
return "ticker " + ticker +
" price " + price +
" timestamp " + printTime(timestamp);
}
private String printTime(long timestamp) {
return new Date(timestamp).toString();
}
}

50
src/main/java/ats/plugin/OHLCUpdateListener.java Normal file
View File

@ -0,0 +1,50 @@
/*
***************************************************************************************
* Copyright (C) 2006 EsperTech, Inc. All rights reserved. *
* http://www.espertech.com/esper *
* http://www.espertech.com *
* ---------------------------------------------------------------------------------- *
* The software in this package is published under the terms of the GPL license *
* a copy of which has been included with this distribution in the license.txt file. *
***************************************************************************************
*/
package ats.plugin;
import com.espertech.esper.client.EPServiceProvider;
import com.espertech.esper.client.EPStatement;
import com.espertech.esper.client.EventBean;
import com.espertech.esper.client.StatementAwareUpdateListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Date;
public class OHLCUpdateListener implements StatementAwareUpdateListener {
private final static Logger log = LoggerFactory.getLogger(OHLCUpdateListener.class);
public void update(EventBean[] newData, EventBean[] oldData, EPStatement epStatement, EPServiceProvider epServiceProvider) {
for (int i = 0; i < newData.length; i++) {
//if (log.isInfoEnabled()) {
log.info("Statement " + String.format("%s", epStatement.getName()) + " produced: " + getProperties(newData[i]));
//}
}
}
private String getProperties(EventBean theEvent) {
StringBuilder buf = new StringBuilder();
for (String name : theEvent.getEventType().getPropertyNames()) {
Object value = theEvent.get(name);
buf.append(name);
buf.append("=");
if (name.contains("timestamp")) {
buf.append(new Date((Long) value));
} else {
buf.append(value);
}
buf.append(" ");
}
return buf.toString();
}
}

54
src/main/java/ats/plugin/OHLCValue.java Normal file
View File

@ -0,0 +1,54 @@
package ats.plugin;
import org.joda.time.DateTime;
/**
* OHLCValue stores one bar of OHLC info.
*/
public class OHLCValue {
private DateTime time;
private double open;
private double high;
private double low;
private double close;
public OHLCValue(DateTime time,
double open, double high,
double low, double close)
{
this.time = time;
this.open = open;
this.high = high;
this.low = low;
this.close = close;
}
public DateTime getTime() {
return time;
}
public Double getOpen() {
return open;
}
public Double getHigh() {
return high;
}
public Double getLow() {
return low;
}
public Double getClose() {
return close;
}
/**
* Return a human readable representation of this event.
*/
public String toString() {
return String.format("OHLCValue[%s,open=%s,high=%s,low=%s,close=%s]",
time, open, high, low, close);
}
}

12
src/test/java/AppTest.java
View File

@ -1,12 +0,0 @@
/*
* This Java source file was generated by the Gradle 'init' task.
*/
import org.junit.Test;
//import static org.junit.Assert.*;
public class AppTest {
@Test public void testAppHasAGreeting() {
//App classUnderTest = new App();
//assertNotNull("app should have a greeting", classUnderTest.getGreeting());
}
}

88
src/test/java/ats/plugin/OHLCPlugInViewTest.java Normal file
View File

@ -0,0 +1,88 @@
import ats.plugin.OHLCPlugInView;
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.joda.time.Period;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.junit.Assert.*;
import org.joda.time.format.DateTimeFormatter;
import org.joda.time.format.ISODateTimeFormat;
import org.joda.time.DateTime;
public class OHLCPlugInViewTest {
private static final Logger log = LoggerFactory.getLogger(OHLCPlugInViewTest.class);
private static DateTimeFormatter dateFormatter = ISODateTimeFormat.dateTime();
@Test
public void testMakeWindowStartTime() {
DateTime now = DateTime.parse("2018-05-15T13:23:25.300-00:00");
assertInterval(now, Duration.standardSeconds(1), "2018-05-15T13:23:25");
assertInterval(now, Duration.standardSeconds(2), "2018-05-15T13:23:24");
assertInterval(now, Duration.standardSeconds(5), "2018-05-15T13:23:25");
assertInterval(now, Duration.standardSeconds(10), "2018-05-15T13:23:20");
assertInterval(now, Duration.standardMinutes(1), "2018-05-15T13:23:00");
assertInterval(now, Duration.standardMinutes(2), "2018-05-15T13:22:00");
assertInterval(now, Duration.standardMinutes(5), "2018-05-15T13:20:00");
// assertInterval(now, Duration.standardMinutes(15), "2018-05-15T13:15:00");
// assertInterval(now, Duration.standardMinutes(45), "2018-05-15T13:15:00");
assertInterval(now, Duration.standardHours(1), "2018-05-15T13:00:00");
assertInterval(now, Duration.standardHours(2), "2018-05-15T12:00:00");
assertInterval(now, Duration.standardHours(10), "2018-05-15T10:00:00");
}
private void assertInterval(DateTime currentTime,
Duration interval,
String expected)
{
DateTime l = OHLCPlugInView.makeWindowStartTime(currentTime, interval);
String start = dateFormatter.print(l);
expected += ".000Z";
log.info(" duration: {}", interval);
log.info(" cur time: {}", dateFormatter.print(currentTime));
log.info("start time: {}", start);
log.info(" expected: {}", expected);
//assertEquals(interval.toString(), l.longValue(), Instant.parse("2018-05-15T13:23:00").getMillis());
assertEquals(interval.toString(), expected, start);
}
/*
private static Long makeWindowStartTime(Long timestamp, Duration interval) {
long today = getDayStart(timestamp);
long intoToday = timestamp - today;
// then modulo timestamp to find how far into the current window we'd be in
long intoPeriod = intoToday % interval.getMillis();
// then subtract modulo from current time to get window start
return timestamp - intoPeriod;
}
private static long getDayStart(Long timestamp) {
Calendar cal = Calendar.getInstance();
cal.setTimeInMillis(timestamp);
cal.set(Calendar.HOUR, 0);
cal.set(Calendar.MINUTE, 0);
cal.set(Calendar.SECOND, 0);
cal.set(Calendar.MILLISECOND, 0);
return cal.getTimeInMillis();
}
private static Long makeWindowEndTime(Long startTime, Duration interval) {
if (startTime == null || interval == null) return null;
return startTime + interval.getMillis();
}
private boolean inWindow(Long timestamp) {
if (timestamp == null) return false;
return timestamp >= windowStartTime && timestamp < windowEndTime;
}
*/
}

74
test.epl
View File

@ -1,6 +1,70 @@
-- in_trading_hours is set to true if the current time -- test.epl - sample setup
-- is inside the normal --
create variable bool in_trading_hours -- Statements must be separated by an empty line.
-- update in_trading_hours variable on each tick
on Tick as t set in_trading_hours = (EPLHelpers.getHour(t.time) >= 9 and EPLHelpers.getHour(t.time) < 17) -- The time the trading logic will begin to enter trades.
-- Exiting trades is 24/7.
create constant variable int StartTimeHour = 9
-- The time the trading logic will begin to enter trades.
-- Exiting trades is 24/7.
create constant variable int EndTimeHour = 17
-- The time frame or tickbar count of the data to be analyzed or
-- traded. Examples: 5s, 1m, 1h 30m, 2h, 12h, 1d.
create constant variable string Interval = '15s'
-- Amount to be traded, measured in units.
create constant variable int TradeSize = 100000
-- InTradingHours will be set to true if the current time is between
-- StartTime and EndTime.
create variable bool InTradingHours = false
-- Update on each tick
on TickEvent as t set InTradingHours =
(EPLHelpers.getHour(t.time) >= StartTimeHour and
EPLHelpers.getHour(t.time) < EndTimeHour)
-- A stream of OHLC values
create variant schema OHLCStream as OHLCValue
insert into OHLCStream
select * from TickEvent#OHLC(Interval, time, midDouble)
--
-- testing/debug/WIP below
--
-- create window LongEntryTimeWindow#time(60 min) as
-- select * from LongEntryEvent
--
-- create window TicksTimeWindow#time(30 sec) as
-- select * from TickEvent
-- on TickEvent insert into TicksTimeWindow select * from TickEvent
-- on TickEvent as t merge TicksTimeWindow as win
-- where win.time
-- insert into OrdersWindow select * from MerchandiseProductEvent
-- set InTradingHours =
-- (EPLHelpers.getHour(t.time) >= StartTimeHour and
-- EPLHelpers.getHour(t.time) < EndTimeHour)
-- LECount = (1,clock,60 min) == only allow one long entry per clock hour example: one LE between 1:00pm to 1:59pm
-- B1 = ValueWhen( (ref(sma(5,c),-1)>ref(sma(5,c),-2)<ref(sma(5,c),-3)), Ref(Low,-2), 1 )
-- something like "insert into CountPerSec select count(*) as cnt from XYZ.win:time_batch(1)" "select sum(cnt) from CountPerSec.win:time(1 hour)" for example, this has 1 second resolution of counts, define as you like
-- 1) String stmt = "insert into WebEventTotalCount select count(*) as cnt from " +
-- "WebEvent.win:firstlength(1) A where A.page.pageName is not null ";
-- 2) String stmt = "update istream WebEventTotalCount set cnt = cnt + ( select count(*) as cnt from " +
-- "WebEvent.win:time(1) A where A.page.pageName is not null ) " ;
-- 3) String stmt = " select cnt from WebEventTotalCount ";