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


--
-- Setup variables
--

-- The time the trading logic will begin to enter trades.
-- Exiting trades is 24/7.
create constant variable DateTime TradingStartTime = EPLHelpers.parseTime("00:00")

-- The time the trading logic will no longer enter trades.
-- Exiting trades is 24/7.
create constant variable DateTime TradingEndTime = EPLHelpers.parseTime("23:59")

-- The time frame for OHLC calculation.
-- Example values: '5s', '1m', '1h 30m', '2h', '12h', '1d'.
create constant variable string OHLCInterval = '10s'

-- The number of ticks for OHLC TB calculation.
create constant variable int OHLCTicks = 100

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

-- How large is a pip?
create constant variable BigDecimal PipSize = new BigDecimal(0.0001)

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

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

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

-- Define the window as length 1, using the structure of the TickEvent
-- java class to describe what the window contains.
-- create window CurrentTick#length(1) as TickEvent

-- Describe how events get added to the window.  This runs every time
-- a new TickEvent is posted.
-- insert into CurrentTick select * from TickEvent


--
-- Trading window
--

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

-- Update on each tick
on TickEvent as t
  set InTradingHours = EPLHelpers.inTimeRange(t.time, TradingStartTime, TradingEndTime)


--
-- In long position.  Set later.
--

create variable bool InLongEntry = false

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

-- Send every TickEvent to the OHLC plugin.  The plugin will post an
-- OHLCEvent to OHLCStream every OHLCInterval amount of time.  It uses
-- TickEvent.time ("time") as the source of the timestamp, and uses
-- TickEvent.midDouble() as the value to use in the OHLC calculation.

create variant schema OHLCStream as OHLCEvent

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

-- Send every TickEvent to the OHLC plugin.  The plugin will post an
-- OHLCEvent using Heikin-Ashi calculations to HKStream every
-- OHLCInterval amount of time.  It uses TickEvent.midDouble() as the
-- value to use in the OHLC calculation.

create variant schema HKStream as OHLCEvent

insert into HKStream
  select * from TickEvent#OHLC("HeikinAshi", "time", OHLCInterval, time, midDouble)

-- Send every TickEvent to the OHLC plugin.  The plugin will post an
-- OHLCEvent to TBStream every OHLCTicks ticks.  It uses
-- TickEvent.time ("time") as the source of the timestamp, and uses
-- TickEvent.midDouble() as the value to use in the OHLC calculation.

create variant schema TBStream as OHLCEvent

insert into TBStream
  select * from TickEvent#OHLC("HeikinAshi", "ticks", OHLCTicks, time, midDouble)

-- Send every TickEvent to the OHLC plugin.  The plugin will post an
-- OHLCEvent using Heikin-Ashi calculations to TBHKStream every
-- OHLCTicks ticks.  It uses TickEvent.midDouble() as the value to use
-- in the OHLC calculation.

create variant schema TBHKStream as OHLCEvent

insert into TBHKStream
  select * from TickEvent#OHLC("HeikinAshi", "ticks", OHLCTicks, time, midDouble)


--
-- Simple moving average streams
--

-- SMACloseStream bundles a simple moving average of the close values
-- with the values of the OHLCStream.
create schema SMACloseStream as (time DateTime,
                                 open double,
                                 high double,
                                 low double,
                                 close double,
                                 averageClose double)

-- Average the most recent OHLC close values from OHLCStream and
-- post an event that contains open, high, low, close, and
-- SMA(close).  The number of OHLC events used in the SMA calc is set
-- by the SMASize variable.
insert into SMACloseStream
  select time, open, high, low, close, Avg(close) as averageClose
    from OHLCStream#length(SMASize)


--
-- ValueWhen calculations
--

-- A stream that feeds B1 and B2.  Each event contains a double
-- precision floating point value "low", and a timestamp called
-- "time".
create schema BStream as (low double, time DateTime)

-- Listen to the last "RefSize" number of SMACloseStream events. Look
-- for an OHLC bar with a lower average close than its neighbors.  Add
-- that bar's low value and its timestamp to the stream.
insert into BStream
  select prev(1, low) as low, prev(1, time) as time from SMACloseStream#length(RefSize)
    where prev(0, averageClose) > prev(1, averageClose)
      and prev(1, averageClose) < prev(2, averageClose)


-- Define B1 to contain the same fields as BStream
create schema B1 (low double, time DateTime)

-- B1 contains the most recent low value and time from BStream.
-- This is the last time an average close was lower
-- than the ones before and after.
-- Since the time is included in the event, no separate BT1 is needed.
insert into B1 select prev(0, low) as low, prev(0, time) as time
  from BStream#length(RefSize)

-- B2 contains the *second* most recent occurrence in BStream, but is
-- otherwise the same as B1.
create schema B2 (low double, 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 DateTime)

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

-- P1 contains the most recent low value and time from PStream.
-- This is the last time an average close was higher
-- than the ones before and after.
-- Since the time is included in the event, no separate PT1 is needed.
create schema P1 (low double, time DateTime)

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

-- P2 contains the second most recent occurrence in PStream.
create schema P2 (low double, time DateTime)

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


--
-- Long entry
--

-- A helper for LE calc.  Keep track of the highest OHLC high value.
create window MaxHigh3Window#length(1) as (high double)

-- Post the largest high value on OHLCStream from the most recent
-- three bars.
insert into MaxHigh3Window
  select max(high) as high from OHLCStream#length(3)


-- Long entry events contain the current tick's midpoint value,
-- timestamp, and instrument name.
create schema LongEntryStream as (current BigDecimal, time DateTime, instrument String)

-- The long entry calc below is translated from this entry in the
-- spreadsheet:
--
-- LE = C > HHV(High,3)
--      and B1 < B2
--      and P1 < P2
--      and BT1 > PT1
--      and BT2 > PT2
--      and PT1 > BT2

insert into LongEntryStream
  select C.mid as current, C.time as time, C.instrument as instrument
    from TickEvent#lastevent 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)
      and InTradingHours
      and not InLongEntry

-- Because multiple streams feed LongEntryStream (CurrentTick,
-- MaxHigh3Window, B1, B2...), an event on any of those streams causes
-- the LongEntryStream logic above to be triggered.  This often causes
-- multiple LongEntryStream events to be generated for a single tick
-- when several of the feeder streams are updated at the same time.
--
-- LongEntryDistinct filters out duplicate LongEntryStream events,
-- leaving a maximum of one event per tick.
create schema LongEntryDistinct
 as (current BigDecimal,
     time DateTime,
     instrument String,
     units int)

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


--
-- Long entry derived values
--

-- Register if We're in a long entry when we get the LongEntryDistinct
-- event.  (InLongEntry is declared near top of file)
on LongEntryDistinct as le set InLongEntry = true

-- LongEntryTime contains the timestamp of the last long entry,
-- or null if none has happened yet.
create variable DateTime LongEntryTime = null

on LongEntryDistinct as le set LongEntryTime = le.time

-- LongEntryStopBarCount contains the number of OHLCStream bars that
-- have occurred since the most recent long entry.
create variable int LongEntryStopBarCount = 0

-- Reset LongEntryStopBarCount when we get a new LE.
on LongEntryDistinct as le set LongEntryStopBarCount = 0

-- Increment LongEntryStopBarCount on every bar.
on OHLCStream as ohlc set LongEntryStopBarCount = LongEntryStopBarCount + 1

-- LongEntryPrice contains the price of the last long entry,
-- or null if none has happened yet.
create variable BigDecimal LongEntryPrice = null

on LongEntryDistinct as le set LongEntryPrice = le.current

-- LongEntryPreEntryBar is a stream that keeps bars just prior to the
-- last long entry
create schema LongEntryPreEntryBar
  as (time DateTime,
      open double,
      high double,
      low double,
      close double)

-- Add bars so long as we're not in a long position
insert into LongEntryPreEntryBar
  select time, open, high, low, close from OHLCStream
    where InLongEntry is false

-- Contains the second-most-recent LongEntryPreEntryBar's "low" value
create schema LongEntryPreEntryBarPrevLow (low double)

insert into LongEntryPreEntryBarPrevLow select prev(1, low) as low
  from LongEntryPreEntryBar#length(2)


--
-- Long exit
--

-- Long exit event definition
create schema LongExitStream
  as (current BigDecimal,
      time DateTime,
      instrument String,
      units int)

-- The long exit calc below is translated from this entry in the
-- spreadsheet:
--
-- LX = (LESBC <= 60 and C < MIN(PreEntrybar(Low,1), (LongEntryPrice - 10 pips)))
--      or
--      (LESBC>60 and C<(EntryPrice + 2 pips)

-- insert into LongExitStream
--   select C.mid as current,
--          C.time as time,
--          C.instrument as instrument,
--          0 - TradeSize as units -- negative for "sell"
--     from CurrentTick as C
--     where LongEntryStopBarCount = 2
--       and InLongEntry

    -- where ((LongEntryStopBarCount <= 60 and C.mid < min(L.low, LongEntryPrice - (10 * PipSize))) or
    --        (LongEntryStopBarCount > 60 and C.mid < LongEntryPrice + (2 * PipSize)))

-- Register the long position as closed
-- on LongExitStream as lx set InLongEntry = false

-- Long exit event definition
create schema LongExitStream
  as (id string)

-- Send an event whenever we should close an order.
-- Received and acted upon in EsperProcessor.
on TickEvent as C
insert into LongExitStream
  select ot.id as id
    from OrderTable ot
    where ot.stopBarCount > 1 and ot.open is true
      and ot.id is not 'dummy'



create schema OpenOrderStream
  as (id string,
      instrument String,
      price BigDecimal,
      units BigDecimal)

-- Add an order to the order table.
-- OpenOrderStream events are sent from Java TradingManagers.
on OpenOrderStream as oos
  insert into OrderTable(id, time, instrument, price, units, open, stopBarCount)
  select oos.id, EPLHelpers.makeTime(current_timestamp()), oos.instrument, oos.price, oos.units, true, 0

create schema CloseOrderStream
  as (id string)

-- Add an order to the order table.
-- CloseOrderStream events are sent from Java TradingManagers.
on CloseOrderStream
  update OrderTable
  set OrderTable.open = false where OrderTable.id = CloseOrderStream.id


--
-- Event logging
--

-- Log events consist of the stream name and an event description.
create schema LogStream as (stream string, event string)

-- Enable logging the events on specific streams by uncommenting
-- individual lines below.  Depending on ongoing debugging needs, some
-- of these can be either helpful or too noisy.  Comment/uncomment as
-- you see fit.

-- 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 'HKStream' as stream, EPLHelpers.str(*) as event from HKStream

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

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

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

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

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

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

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

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

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

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

-- insert into LogStream select 'LXPrice' as stream, EPLHelpers.str(mid, LongEntryPrice, LongEntryPrice - (10 * PipSize)) as event from TickEvent
-- where InLongEntry and mid < LongEntryPrice

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

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

-- on LongEntryDistinct
-- insert into LogStream select 'OrderTable' as stream,
--   EPLHelpers.str(OrderTable.time,
--                  OrderTable.instrument,
--                  OrderTable.id,
--                  OrderTable.units,
--                  OrderTable.open,
--                  OrderTable.stopBarCount,
--                  OrderTable.price) as event
--   from OrderTable order by OrderTable.time

-- @Priority(-9999)
-- @Name("LogOrderTableClose")
-- on LongExitStream
-- insert into LogStream select 'OrderTable_CLOSE' as stream,
--   EPLHelpers.str(OrderTable.time,
--                  OrderTable.instrument,
--                  OrderTable.id,
--                  OrderTable.units,
--                  OrderTable.open,
--                  OrderTable.stopBarCount,
--                  OrderTable.price) as event
--   from OrderTable order by OrderTable.time

-- on LongExitStream
-- insert into LogStream select 'OrderTable' as stream, EPLHelpers.str(count(*)) as event from OrderTable

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

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

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


--
-- Visualization data logging
--

create schema VizDataStream as (event Map)

-- Enable visualizing the events on specific streams by uncommenting
-- individual lines below.

insert into VizDataStream select ats.viz.VizLog.eventString('tick', 'C', *) as event from TickEvent

insert into VizDataStream select ats.viz.VizLog.eventString('bar', 'ohlc', *) as event from OHLCStream

-- insert into VizDataStream select 'le' as tag, VizLog.eventString(*) as event from LongEntryDistinct

-- insert into VizDataStream select 'lx' as tag, VizLog.eventString(*) as event from LongExitStream

insert into VizDataStream select ats.viz.VizLog.eventString('event', 'open', *) as event from OpenOrderStream

insert into VizDataStream select ats.viz.VizLog.eventString('event', 'close', *) as event from CloseOrderStream