BarAggregator: use tick timestamps

src/ATrade/BarAggregator.hs

@@ -19,6 +19,7 @@ module ATrade.BarAggregator (
   mkAggregatorFromBars,
   handleTicks,
   handleTick,
+  updateTime,
   handleBar,
   hmsToDiffTime
 ) where
@@ -144,6 +145,42 @@ handleTick tick = runState $ do
           barClose = barClose bar,
           barVolume = 0 }
 
+updateTime :: Tick -> BarAggregator -> (Maybe Bar, BarAggregator)
+updateTime tick = runState $ do
+  lLastTicks %= M.insert (security tick, datatype tick) tick
+  tws <- gets tickTimeWindows
+  mybars <- gets bars
+  if (any (isInTimeInterval tick) tws)
+    then
+      case M.lookup (security tick) mybars of
+        Just series -> case bsBars series of
+          (b:bs) -> do
+            let currentBn = barNumber (barTimestamp b) (tfSeconds $ bsTimeframe series)
+            if currentBn == barNumber (timestamp tick) (tfSeconds $ bsTimeframe series)
+              then do
+                lBars %= M.insert (security tick) series { bsBars = updateBarTimestamp b tick : bs }
+                return Nothing
+              else do
+                lBars %= M.insert (security tick) series { bsBars = emptyBarFromTick tick : b : bs }
+                return $ Just b
+          _ -> return Nothing
+        _ -> return Nothing
+    else
+      return Nothing
+  where
+    isInTimeInterval t (a, b) = (utctDayTime . timestamp) t >= a && (utctDayTime . timestamp) t <= b
+    emptyBarFromTick !newtick = Bar { barSecurity = security newtick,
+      barTimestamp = timestamp newtick,
+      barOpen = value newtick,
+      barHigh = value newtick,
+      barLow = value newtick,
+      barClose = value newtick,
+      barVolume = 0 }
+
+    updateBarTimestamp !bar newtick = bar { barTimestamp = newTimestamp }
+      where
+        newTimestamp = timestamp newtick
+
 handleBar :: Bar -> BarAggregator -> (Maybe Bar, BarAggregator)
 handleBar bar = runState $ do
   tws <- gets tickTimeWindows

src/ATrade/Driver/Real/QuoteSourceThread.hs

@@ -37,11 +37,17 @@ startQuoteSourceThread ctx qsEp strategy eventChan agg tickFilter maybeSourceTim
       case qdata of
         QDTick tick -> when (goodTick tick) $ do
           writeChan eventChan (NewTick tick)
-          when (isNothing maybeSourceTimeframe) $ do
+          case maybeSourceTimeframe of
-            aggValue <- readIORef agg
+            Nothing -> do
-            case handleTick tick aggValue of
+              aggValue <- readIORef agg
-              (Just bar, !newAggValue) -> writeIORef agg newAggValue >> writeChan eventChan (NewBar bar)
+              case handleTick tick aggValue of
-              (Nothing, !newAggValue) -> writeIORef agg newAggValue
+                (Just bar, !newAggValue) -> writeIORef agg newAggValue >> writeChan eventChan (NewBar bar)
+                (Nothing, !newAggValue) -> writeIORef agg newAggValue
+            Just _ -> do
+              aggValue <- readIORef agg
+              case updateTime tick aggValue of
+                (Just bar, !newAggValue) -> writeIORef agg newAggValue >> writeChan eventChan (NewBar bar)
+                (Nothing, !newAggValue) -> writeIORef agg newAggValue
         QDBar (_, bar) -> do
           aggValue <- readIORef agg
           when (isJust maybeSourceTimeframe) $ do

test/Test/BarAggregator.hs

@@ -33,6 +33,7 @@ unitTests = testGroup "BarAggregator" [
   , testTwoBarsInSameBar
   , testTwoBarsInSameBarLastBar
   , testNextBarAfterBarClose
+  , testUpdateTime
   ]
 
 properties = testGroup "BarAggregator" [
@@ -192,6 +193,37 @@ testNextBarAfterBarClose = testCase "Three bars (smaller timeframe) - next bar a
             barClose = fromDouble c,
             barVolume = v }
 
+testUpdateTime :: TestTree
+testUpdateTime = testCase "updateTime - next bar - creates new bar with zero volume" $ do
+  let series = BarSeries "TEST_TICKER" (Timeframe 3600) []
+  let agg = mkAggregatorFromBars (M.fromList [("TEST_TICKER", series)]) [(0, 86400)]
+  let (_, newagg) = handleBar (bar testTimestamp1 12.00 13.00 10.00 11.00 1) agg
+  let (_, newagg') = handleBar (bar testTimestamp2 12.00 15.00 11.00 12.00 2) newagg
+  let (newBar, newagg'') = updateTime (tick testTimestamp4 13.00 100) newagg'
+  let expectedNewBar = Bar "TEST_TICKER" testTimestamp2 12.00 15.00 10.00 12.00 3
+  let expectedBar = Bar "TEST_TICKER" testTimestamp4 13.00 13.00 13.00 13.00 0
+  (head <$> bsBars <$> (M.lookup "TEST_TICKER" $ bars newagg'')) @?= Just expectedBar
+  newBar @?= Just expectedNewBar
+  where
+    testTimestamp1 = (UTCTime (fromGregorian 1970 1 1) 560)
+    testTimestamp2 = (UTCTime (fromGregorian 1970 1 1) 600)
+    testTimestamp3 = (UTCTime (fromGregorian 1970 1 1) 3600)
+    testTimestamp4 = (UTCTime (fromGregorian 1970 1 1) 3660)
+    tick ts v vol = Tick {
+              security = "TEST_TICKER"
+            , datatype = LastTradePrice
+            , timestamp = ts
+            , value = v
+            , volume = vol }
+    bar ts o h l c v = Bar {
+            barSecurity = "TEST_TICKER",
+            barTimestamp = ts,
+            barOpen = fromDouble o,
+            barHigh = fromDouble h,
+            barLow = fromDouble l,
+            barClose = fromDouble c,
+            barVolume = v }
+
 prop_allTicksInOneBar :: TestTree
 prop_allTicksInOneBar = QC.testProperty "All ticks in one bar" $ QC.forAll (QC.choose (1, 86400)) $ \timeframe ->
   QC.forAll (QC.listOf1 (genTick "TEST_TICKER" baseTime timeframe)) $ \ticks ->
@@ -215,4 +247,3 @@ prop_allTicksInOneBar = QC.testProperty "All ticks in one bar" $ QC.forAll (QC.c
     currentBar tickerId agg = headMay =<< (bsBars <$> M.lookup tickerId (bars agg))
     baseTime = UTCTime (fromGregorian 1970 1 1) 0
 
-