ATS_Esper/epl/trading_system_1.epl

-- trading_system_1.epl
--
-- Statements must be separated by an empty line.


-- 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 for OHLC calculation.
-- Example values: '5s', '1m', '1h 30m', '2h', '12h', '1d'.
create constant variable string OHLCInterval = '10s'

-- Amount to be traded, measured in units.
create constant variable int TradeSize = 100000

-- How many ticks to use for simple moving average calculation
create constant variable int SMASize = 5

-- How many ticks to store for Ref() access
create constant variable int RefSize = 5


--
-- A named window that contains the current tick
--

create window CurrentTickWindow#length(1) as TickEvent

insert into CurrentTickWindow select * from TickEvent


--
-- Trading window
--

-- InTradingHours will be set to true when the current time is between
-- StartTime and EndTime.
create variable bool InTradingHours = false

-- Update on each tick
-- NOTE: see "timer:within" pattern for possible alternate formulation
on TickEvent as t set InTradingHours =
  (EPLHelpers.getHour(t.time) >= StartTimeHour and
   EPLHelpers.getHour(t.time) < EndTimeHour)


--
-- A stream of OHLC values calculated from TickEvents
--

create variant schema OHLCStream as OHLCEvent

insert into OHLCStream
  select * from TickEvent#OHLC(OHLCInterval, time, midDouble)


--
-- Simple moving average streams
--

-- Average the most recent OHLC close values and create events that
-- contain open, high, low, close and SMA(close).  The number of
-- OHLC events used in the SMA calc is set by SMASize.
create schema SMACloseStream as ats.plugin.OHLCValueEvent

insert into SMACloseStream
  select new ats.plugin.OHLCValueEvent(time, open, high, low, close, Avg(close))
    from OHLCStream#length(SMASize)


--
-- ValueWhen calculations
--

-- A stream that feeds B1 and B2
create schema BStream as (low double, time org.joda.time.DateTime)

-- Find a bar with a lower close than its neighbors.
-- Add that low value to the stream.
insert into BStream
  select prev(1, low) as low, prev(1, time) as time from SMACloseStream#length(RefSize)
    where prev(0, value) > prev(1, value)
      and prev(1, value) < prev(2, value)

create schema B1 (low double, time org.joda.time.DateTime)

insert into B1 select prev(0, low) as low, prev(0, time) as time
  from BStream#length(RefSize)

create schema B2 (low double, time org.joda.time.DateTime)

insert into B2 select prev(1, low) as low, prev(1, time) as time
  from BStream#length(RefSize)


-- A stream that feeds P1 and P2
create schema PStream as (low double, time org.joda.time.DateTime)

-- Find a bar with a higher close than its neighbors.
-- Add that low value to the stream.
insert into PStream
  select prev(1, low) as low, prev(1, time) as time from SMACloseStream#length(RefSize)
    where prev(0, value) < prev(1, value)
      and prev(1, value) > prev(2, value)

create schema P1 (low double, time org.joda.time.DateTime)

insert into P1 select prev(0, low) as low, prev(0, time) as time
  from PStream#length(RefSize)

create schema P2 (low double, time org.joda.time.DateTime)

insert into P2 select prev(1, low) as low, prev(1, time) as time
  from PStream#length(RefSize)


--
-- Long entry
--

-- keep track of the highest OHLC high value from the most recent three
create window MaxHigh3Window#length(1) as (high double)

insert into MaxHigh3Window
  select max(high) as high from OHLCStream#length(3)

-- LE calc below is translated from:
--
-- LE = C > HHV(High,3)
--      and B1 < B2
--      and P1 < P2
--      and BT1 > PT1
--      and BT2 > PT2
--      and PT1 > BT2

create schema LongEntryStream as (current BigDecimal, time org.joda.time.DateTime)

insert into LongEntryStream
  select C.mid as current, C.time as time
    from CurrentTickWindow as C,
         MaxHigh3Window as T,
         B1#lastevent, B2#lastevent,
         P1#lastevent, P2#lastevent
    where C.mid > T.high
      and B1.low < B2.low
      and P1.low < P2.low
      and EPLHelpers.laterThan(B1.time, P1.time)
      and EPLHelpers.laterThan(B2.time, P2.time)
      and EPLHelpers.laterThan(P1.time, B2.time)

create schema LongEntryDistinct as (current BigDecimal, time org.joda.time.DateTime)

insert into LongEntryDistinct
  select le.current as current, le.time as time
    from pattern [every-distinct(le.time) le=LongEntryStream]



--
-- Event logging
--

-- TODO: look into LogSink http://esper.espertech.com/release-7.1.0/esper-reference/html/dataflow.html#dataflow-reference-logsink

create schema LogStream as (stream string, event string)

-- Enable logging specific stream events by uncommenting individual
-- lines below:

-- insert into LogStream select 'TickEvent' as stream, EPLHelpers.str(*) as event from TickEvent

-- insert into LogStream select 'OHLCStream' as stream, EPLHelpers.str(*) as event from OHLCStream

-- insert into LogStream select 'BStream' as stream, EPLHelpers.str(*) as event from BStream

-- insert into LogStream select 'PStream' as stream, EPLHelpers.str(*) as event from PStream

insert into LogStream select 'B1' as stream, EPLHelpers.str(*) as event from B1

insert into LogStream select 'B2' as stream, EPLHelpers.str(*) as event from B2

insert into LogStream select 'P1' as stream, EPLHelpers.str(*) as event from P1

insert into LogStream select 'P2' as stream, EPLHelpers.str(*) as event from P2

-- insert into LogStream select 'MaxHigh3Window' as stream, EPLHelpers.str(*) as event from MaxHigh3Window

-- insert into LogStream select 'LongEntryStream' as stream, EPLHelpers.str(*) as event from LongEntryStream

-- insert into LogStream select 'LongEntryDistinct' as stream, EPLHelpers.str(*) as event from LongEntryDistinct