junction: correct order submission

robocom-zero.cabal

@@ -34,6 +34,7 @@ library
                      , ATrade.Driver.Junction.QuoteStream
                      , ATrade.Driver.Junction.RobotDriverThread
                      , ATrade.Driver.Junction.ProgramConfiguration
+                     , ATrade.Driver.Junction.BrokerService
                      , ATrade.BarAggregator
                      , ATrade.RoboCom
                      , ATrade.Quotes.HistoryProvider

src/ATrade/Driver/Junction.hs

@@ -17,7 +17,9 @@ import           ATrade.Broker.Client                        (BrokerClientHandle
 import           ATrade.Broker.Protocol                      (Notification (OrderNotification, TradeNotification),
                                                               NotificationSqnum,
                                                               getNotificationSqnum)
-import           ATrade.Driver.Junction.ProgramConfiguration (ProgramConfiguration (brokerClientCert, brokerEndpoint, brokerNotificationEndpoint, brokerServerCert, instances, qhpEndpoint, qtisEndpoint, redisSocket, robotsConfigsPath),
+import           ATrade.Driver.Junction.BrokerService        (BrokerService,
+                                                              mkBrokerService)
+import           ATrade.Driver.Junction.ProgramConfiguration (ProgramConfiguration (ProgramConfiguration, brokerClientCert, brokerEndpoint, brokerNotificationEndpoint, brokerServerCert, instances, qhpEndpoint, qtisEndpoint, redisSocket, robotsConfigsPath),
                                                               ProgramOptions (ProgramOptions, configPath))
 import           ATrade.Driver.Junction.QuoteStream          (QuoteStream (addSubscription, removeSubscription),
                                                               QuoteSubscription (QuoteSubscription),
@@ -38,16 +40,17 @@ import           ATrade.Driver.Junction.Types                (StrategyDescriptor
                                                               confStrategy,
                                                               strategyState,
                                                               strategyTimers)
-import           ATrade.Logging                              (Message,
-                                                              Severity (Info),
+import           ATrade.Logging                              (Message, Severity (Debug, Error, Info, Trace, Warning),
                                                               fmtMessage,
                                                               logWarning,
                                                               logWith)
 import           ATrade.Quotes.QHP                           (mkQHPHandle)
 import           ATrade.RoboCom.ConfigStorage                (ConfigStorage (loadConfig))
+import           ATrade.RoboCom.Monad                        (StrategyEnvironment (..))
 import           ATrade.RoboCom.Persistence                  (MonadPersistence (loadState, saveState))
+import           ATrade.RoboCom.Types                        (Bars)
 import           ATrade.Types                                (ClientSecurityParams (ClientSecurityParams),
-                                                              OrderId,
+                                                              Order, OrderId,
                                                               Trade (tradeOrderId))
 import           Colog                                       (HasLog (getLogAction, setLogAction),
                                                               LogAction,
@@ -76,6 +79,7 @@ import qualified Data.Set                                    as S
 import qualified Data.Text                                   as T
 import           Data.Text.Encoding                          (encodeUtf8)
 import           Data.Text.IO                                (readFile)
+import           Data.Time                                   (getCurrentTime)
 import           Data.Time.Clock.POSIX                       (getPOSIXTime)
 import           Database.Redis                              (ConnectInfo (..),
                                                               Connection,
@@ -162,8 +166,8 @@ junctionMain :: M.Map T.Text StrategyDescriptorE -> IO ()
 junctionMain descriptors = do
   opts <- parseOptions
 
-  let bootstrapLogAction = fmtMessage >$< logTextStdout
-  let log = logWith bootstrapLogAction
+  let logger = fmtMessage >$< logTextStdout
+  let log = logWith logger
 
   log Info "Junction" $ "Reading config from: " <> (T.pack . show) (configPath opts)
 
@@ -171,15 +175,19 @@ junctionMain descriptors = do
 
   barsMap <- newIORef M.empty
 
+  log Info "Junction" $ "Connecting to redis: " <> redisSocket cfg
   redis <- checkedConnect (defaultConnectInfo { connectPort = UnixSocket (T.unpack $ redisSocket cfg) })
+  log Info "Junction" "redis: connected"
   withContext $ \ctx -> do
+    log Debug "Junction" "0mq context created"
     let downloaderLogAction = fmtMessage >$< logTextStdout
     let downloaderEnv = DownloaderEnv (mkQHPHandle ctx (qhpEndpoint cfg)) ctx (qtisEndpoint cfg) downloaderLogAction
     robotsMap <- newIORef M.empty
     ordersMap <- newIORef M.empty
     handledNotifications <- newIORef S.empty
-    withBroker cfg ctx robotsMap ordersMap handledNotifications $ \bro ->
+    withBroker cfg ctx robotsMap ordersMap handledNotifications logger $ \bro ->
       withQThread downloaderEnv barsMap cfg ctx $ \qt -> do
+        broService <- mkBrokerService bro ordersMap
         let junctionLogAction = fmtMessage >$< logTextStdout
         let env =
               JunctionEnv
@@ -192,7 +200,7 @@ junctionMain descriptors = do
                 peLogAction = junctionLogAction
               }
         withJunction env $ do
-          startRobots cfg bro barsMap
+          startRobots cfg barsMap broService
           forever $ do
             saveRobots
             liftIO $ threadDelay 5000000
@@ -209,7 +217,9 @@ junctionMain descriptors = do
       currentTimers <- liftIO $ readIORef (strategyTimers inst)
       saveState currentTimers (strategyInstanceId inst <> ":timers")
 
-    startRobots cfg bro barsMap = forM_ (instances cfg) $ \inst ->
+    startRobots :: ProgramConfiguration -> IORef Bars -> BrokerService -> JunctionM ()
+    startRobots cfg barsMap broService = forM_ (instances cfg) $ \inst -> do
+      now <- liftIO getCurrentTime
       case M.lookup (strategyBaseName inst) descriptors of
         Just (StrategyDescriptorE desc) -> do
           bigConf <- loadConfig (configKey inst)
@@ -217,7 +227,14 @@ junctionMain descriptors = do
           rState <- loadState (stateKey inst) >>= liftIO . newIORef
           rTimers <- loadState (stateKey inst <> ":timers") >>= liftIO . newIORef
           let robotLogAction = fmtMessage >$< logTextStdout
-          let robotEnv = RobotEnv rState rConf rTimers bro barsMap robotLogAction
+          stratEnv <- liftIO $ newIORef StrategyEnvironment
+                         {
+                           _seInstanceId = strategyId inst,
+                           _seAccount = "test", -- TODO configure
+                           _seVolume = 1,
+                           _seLastTimestamp = now
+                         }
+          let robotEnv = RobotEnv rState rConf rTimers barsMap stratEnv robotLogAction broService
           robot <- createRobotDriverThread inst desc (flip runReaderT robotEnv . unRobotM) bigConf rConf rState rTimers
           robotsMap' <- asks peRobots
           liftIO $ atomicModifyIORef' robotsMap' (\s -> (M.insert (strategyId inst) robot s, ()))
@@ -229,16 +246,20 @@ junctionMain descriptors = do
     handleBrokerNotification :: IORef (M.Map T.Text RobotDriverHandle) ->
                                 IORef (M.Map OrderId T.Text) ->
                                 IORef (S.Set NotificationSqnum) ->
+                                LogAction IO Message ->
                                 Notification ->
                                 IO ()
-    handleBrokerNotification robotsRef ordersMapRef handled notification =
+    handleBrokerNotification robotsRef ordersMapRef handled logger notification= do
+      logWith logger Trace "Junction" $ "Incoming notification: " <> (T.pack . show) notification
       whenM (notMember (getNotificationSqnum notification) <$> readIORef handled) $ do
         robotsMap <- readIORef robotsRef
         ordersMap <- readIORef ordersMapRef
 
         case getNotificationTarget robotsMap ordersMap notification of
           Just robot -> postNotificationEvent robot notification
-          Nothing    -> return () --logWarning "Junction" "Unknown order" -- TODO log
+          Nothing    -> do
+            logWith logger Warning "Junction" $ "Unknown order: " <> (T.pack . show) (notificationOrderId notification)
+            logWith logger Debug "Junction" $ "Ordermap: " <> (T.pack . show) (M.toList ordersMap)
 
         atomicModifyIORef' handled (\s -> (S.insert (getNotificationSqnum notification) s, ()))
 
@@ -250,7 +271,7 @@ junctionMain descriptors = do
     notificationOrderId (OrderNotification _ oid _) = oid
     notificationOrderId (TradeNotification _ trade) = tradeOrderId trade
 
-    withBroker cfg ctx robotsMap ordersMap handled f = do
+    withBroker cfg ctx robotsMap ordersMap handled logger f = do
       securityParameters <- loadBrokerSecurityParameters cfg
       bracket
         (startBrokerClient
@@ -258,8 +279,9 @@ junctionMain descriptors = do
           ctx
           (brokerEndpoint cfg)
           (brokerNotificationEndpoint cfg)
-          [handleBrokerNotification robotsMap ordersMap handled]
-          securityParameters)
+          [handleBrokerNotification robotsMap ordersMap handled logger]
+          securityParameters
+          logger)
         stopBrokerClient f
 
     loadBrokerSecurityParameters cfg =

src/ATrade/Driver/Junction/BrokerService.hs

@@ -0,0 +1,56 @@
+{-# LANGUAGE FlexibleContexts  #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module ATrade.Driver.Junction.BrokerService
+  (
+    BrokerService,
+    mkBrokerService,
+    submitOrder,
+    cancelOrder,
+    getNotifications
+  ) where
+
+import qualified ATrade.Broker.Client       as Bro
+import           ATrade.Broker.Protocol     (Notification (..))
+import           ATrade.Logging             (Message, logDebug)
+import           ATrade.Types               (Order (..), OrderId)
+import           Colog                      (WithLog)
+import           Control.Monad.IO.Class     (MonadIO (liftIO))
+import           Control.Monad.Reader.Class (MonadReader)
+import           Data.IORef                 (IORef, atomicModifyIORef',
+                                             newIORef)
+import qualified Data.Map.Strict            as M
+import qualified Data.Text                  as T
+
+data BrokerService =
+  BrokerService
+  {
+    broker         :: Bro.BrokerClientHandle,
+    orderMap       :: IORef (M.Map OrderId T.Text),
+    orderIdCounter :: IORef OrderId
+  }
+
+mkBrokerService :: Bro.BrokerClientHandle -> IORef (M.Map OrderId T.Text) -> IO BrokerService
+mkBrokerService h om = BrokerService h om <$> newIORef 1
+
+submitOrder :: (MonadIO m, WithLog env Message m, MonadReader env m) => BrokerService -> T.Text -> Order -> m OrderId
+submitOrder service identity order = do
+  oid <- nextOrderId service
+  logDebug "BrokerService" $ "New order, id: " <> (T.pack . show) oid
+  liftIO $ atomicModifyIORef' (orderMap service) (\s -> (M.insert oid identity s, ()))
+  _ <- liftIO $ Bro.submitOrder (broker service) order { orderId = oid }
+  return oid
+  where
+    nextOrderId srv = liftIO $ atomicModifyIORef' (orderIdCounter srv) (\s -> (s + 1, s))
+
+cancelOrder :: BrokerService -> OrderId -> IO ()
+cancelOrder service oid = do
+  _ <- Bro.cancelOrder (broker service) oid
+  return ()
+
+getNotifications :: BrokerService -> IO [Notification]
+getNotifications service = do
+  v <- Bro.getNotifications (broker service)
+  case v of
+    Left _  -> return []
+    Right n -> return n

src/ATrade/Driver/Junction/RobotDriverThread.hs

@@ -15,28 +15,31 @@ module ATrade.Driver.Junction.RobotDriverThread
   onStrategyInstance,
   postNotificationEvent) where
 
-import           ATrade.Broker.Client               (BrokerClientHandle)
-import qualified ATrade.Broker.Client               as Bro
 import           ATrade.Broker.Protocol               (Notification (OrderNotification, TradeNotification))
+import qualified ATrade.Driver.Junction.BrokerService as Bro
 import           ATrade.Driver.Junction.QuoteStream   (QuoteStream (addSubscription),
                                                        QuoteSubscription (QuoteSubscription))
 import           ATrade.Driver.Junction.Types         (BigConfig,
                                                        StrategyDescriptor,
                                                        StrategyInstance (StrategyInstance, strategyEventCallback),
                                                        StrategyInstanceDescriptor (configKey),
-                                                     confStrategy, confTickers,
+                                                       confStrategy,
+                                                       confTickers,
                                                        eventCallback, stateKey,
                                                        strategyId, tickerId,
                                                        timeframe)
-import           ATrade.Logging                     (Message, logInfo)
+import           ATrade.Logging                       (Message, logDebug,
+                                                       logInfo, logWarning)
 import           ATrade.QuoteSource.Client            (QuoteData (..))
 import           ATrade.RoboCom.ConfigStorage         (ConfigStorage)
-import           ATrade.RoboCom.Monad               (Event (NewBar, NewTick, NewTrade, OrderUpdate),
-                                                     MonadRobot (..))
+import           ATrade.RoboCom.Monad                 (Event (NewBar, NewTick, NewTrade, OrderSubmitted, OrderUpdate),
+                                                       MonadRobot (..),
+                                                       StrategyEnvironment (StrategyEnvironment, _seInstanceId, _seLastTimestamp))
 import           ATrade.RoboCom.Persistence           (MonadPersistence)
 import           ATrade.RoboCom.Types                 (BarSeriesId (BarSeriesId),
                                                        Bars)
-import           ATrade.Types                       (OrderId, OrderState, Trade)
+import           ATrade.Types                         (Order (orderId), OrderId,
+                                                       OrderState, Trade)
 import           Colog                                (HasLog (getLogAction, setLogAction),
                                                        LogAction)
 import           Control.Concurrent                   (ThreadId, forkIO)
@@ -46,14 +49,16 @@ import           Control.Concurrent.BoundedChan     (BoundedChan,
 import           Control.Exception.Safe               (MonadThrow)
 import           Control.Monad                        (forM_, forever, void)
 import           Control.Monad.IO.Class               (MonadIO, liftIO)
-import           Control.Monad.Reader               (MonadReader, ReaderT, asks)
+import           Control.Monad.Reader                 (MonadReader (local),
+                                                       ReaderT, asks)
 import           Data.Aeson                           (FromJSON, ToJSON)
 import           Data.Default
-import           Data.IORef                         (IORef, atomicModifyIORef',
+import           Data.IORef                           (IORef,
+                                                       atomicModifyIORef',
                                                        readIORef, writeIORef)
 import qualified Data.Map.Strict                      as M
 import qualified Data.Text.Lazy                       as TL
-import           Data.Time                          (UTCTime)
+import           Data.Time                            (UTCTime, getCurrentTime)
 import           Dhall                                (FromDhall)
 
 data RobotDriverHandle = forall c s. (FromDhall c, Default s, FromJSON s, ToJSON s) =>
@@ -94,6 +99,7 @@ createRobotDriverThread :: (MonadIO m1,
                             ToJSON s,
                             FromDhall c,
                             MonadIO m,
+                            MonadReader (RobotEnv c s) m,
                             MonadRobot m c s) =>
      StrategyInstanceDescriptor
   -> StrategyDescriptor c s
@@ -130,9 +136,10 @@ data RobotEnv c s =
     stateRef      :: IORef s,
     configRef     :: IORef c,
     timersRef     :: IORef [UTCTime],
-    broker    :: BrokerClientHandle,
     bars          :: IORef Bars,
-    logAction :: LogAction (RobotM c s) Message
+    env           :: IORef StrategyEnvironment,
+    logAction     :: LogAction (RobotM c s) Message,
+    brokerService :: Bro.BrokerService
   }
 
 newtype RobotM c s a = RobotM { unRobotM :: ReaderT (RobotEnv c s) IO a }
@@ -144,12 +151,13 @@ instance HasLog (RobotEnv c s) Message (RobotM c s) where
 
 instance MonadRobot (RobotM c s) c s where
   submitOrder order = do
-    bro <- asks broker
-    liftIO $ void $ Bro.submitOrder bro order
+    instId <- _seInstanceId <$> (asks env >>= liftIO . readIORef)
+    bro <- asks brokerService
+    Bro.submitOrder bro instId order
 
   cancelOrder oid = do
-    bro <- asks broker
-    liftIO $ void $ Bro.cancelOrder bro oid
+    bro <- asks brokerService
+    liftIO . void $ Bro.cancelOrder bro oid
 
   appendToLog = logInfo "RobotM" . TL.toStrict -- TODO get instance id from environment and better use it instead of generic 'RobotM'
 
@@ -161,7 +169,11 @@ instance MonadRobot (RobotM c s) c s where
   getConfig = asks configRef >>= liftIO . readIORef
   getState = asks stateRef >>= liftIO . readIORef
   setState newState = asks stateRef >>= liftIO . flip writeIORef newState
-  getEnvironment = undefined
+  getEnvironment = do
+    ref <- asks env
+    now <- liftIO getCurrentTime
+    liftIO $ atomicModifyIORef' ref (\e -> (e { _seLastTimestamp = now }, e { _seLastTimestamp = now}))
+
   getTicker tid tf = do
     b <- asks bars >>= liftIO . readIORef
     return $ M.lookup (BarSeriesId tid tf) b

src/ATrade/RoboCom/Monad.hs

@@ -34,7 +34,7 @@ import           Language.Haskell.Printf
 import           Language.Haskell.TH.Quote (QuasiQuoter)
 
 class (Monad m) => MonadRobot m c s | m -> c, m -> s where
-  submitOrder :: Order -> m ()
+  submitOrder :: Order -> m OrderId
   cancelOrder :: OrderId -> m ()
   appendToLog :: TL.Text -> m ()
   setupTimer :: UTCTime -> m ()

src/ATrade/RoboCom/Positions.hs

@@ -65,7 +65,8 @@ module ATrade.RoboCom.Positions
   setStopLoss,
   setLimitStopLoss,
   setTakeProfit,
-  setStopLossAndTakeProfit
+  setStopLossAndTakeProfit,
+  handlePositions
 ) where
 
 import           GHC.Generics
@@ -191,9 +192,9 @@ dispatchPosition event pos = case posState pos of
       if orderDeadline (posSubmissionDeadline pos) lastTs
         then return $ pos { posState = PositionCancelled } -- TODO call TimeoutHandler if present
         else case event of
-          OrderSubmitted order ->
-            return $ if order `orderCorrespondsTo` pendingOrder
-              then pos { posCurrentOrder = Just order,
+          OrderUpdate oid Submitted -> do
+            return $ if orderId pendingOrder == oid
+              then pos { posCurrentOrder = Just pendingOrder,
                 posState = PositionWaitingOpen,
                 posSubmissionDeadline = Nothing }
               else pos
@@ -206,7 +207,6 @@ dispatchPosition event pos = case posState pos of
           then
             if posBalance pos == 0
               then do
-                appendToLog $ [t|"In PositionWaitingOpen: execution timeout: %?/%?"|] (posExecutionDeadline pos) lastTs
                 cancelOrder $ orderId order
                 return $ pos { posState = PositionWaitingPendingCancellation, posNextState = Just PositionCancelled }
               else do
@@ -271,8 +271,11 @@ dispatchPosition event pos = case posState pos of
             (OrderUpdate _ newstate, Just _, Just (PositionWaitingCloseSubmission nextOrder)) ->
               if newstate == Cancelled
                 then do
-                  submitOrder nextOrder
-                  return pos { posState = PositionWaitingCloseSubmission nextOrder, posSubmissionDeadline = Just (10 `addUTCTime` lastTs), posExecutionDeadline = Nothing }
+                  oid <- submitOrder nextOrder
+                  return pos
+                           { posState = PositionWaitingCloseSubmission nextOrder { orderId = oid },
+                             posSubmissionDeadline = Just (10 `addUTCTime` lastTs),
+                             posExecutionDeadline = Nothing }
                 else return pos
             (OrderUpdate _ newstate, Just _, Just PositionCancelled) ->
               if newstate == Cancelled
@@ -292,9 +295,9 @@ dispatchPosition event pos = case posState pos of
             Nothing    -> doNothing
           return $ pos { posCurrentOrder = Nothing, posState = PositionOpen, posSubmissionDeadline = Nothing } -- TODO call TimeoutHandler if present
         else case event of
-          OrderSubmitted order ->
-            return $ if order `orderCorrespondsTo` pendingOrder
-              then pos { posCurrentOrder = Just order,
+          OrderUpdate oid Submitted ->
+            return $ if orderId pendingOrder == oid
+              then pos { posCurrentOrder = Just pendingOrder,
                 posState = PositionWaitingClose,
                 posSubmissionDeadline = Nothing }
               else pos
@@ -464,8 +467,8 @@ enterAtMarket operationSignalName operation = do
 enterAtMarketWithParams :: (StateHasPositions s, ParamsHasMainTicker c, MonadRobot m c s) => T.Text -> Int -> SignalId -> Operation -> m Position
 enterAtMarketWithParams account quantity signalId operation = do
   tickerId <- snd . mainTicker <$> getConfig
-  submitOrder $ order tickerId
-  newPosition (order tickerId) account tickerId operation quantity 20
+  oid <- submitOrder $ order tickerId
+  newPosition ((order tickerId) { orderId = oid }) account tickerId operation quantity 20
   where
    order tickerId = mkOrder {
     orderAccountId = account,
@@ -508,9 +511,9 @@ enterAtLimitForTicker tickerId timeToCancel operationSignalName price operation
 enterAtLimitForTickerWithParams :: (StateHasPositions s, MonadRobot m c s) => TickerId -> NominalDiffTime -> T.Text -> Int -> SignalId -> Price -> Operation -> m Position
 enterAtLimitForTickerWithParams tickerId timeToCancel account quantity signalId price operation = do
   lastTs <- view seLastTimestamp <$> getEnvironment
-  submitOrder order
+  oid <- submitOrder order
   appendToLog $ [t|enterAtLimit: %?, deadline: %?|] tickerId (timeToCancel `addUTCTime` lastTs)
-  newPosition order account tickerId operation quantity 20 >>=
+  newPosition (order {orderId = oid}) account tickerId operation quantity 20 >>=
     modifyPosition (\p -> p { posExecutionDeadline = Just $ timeToCancel `addUTCTime` lastTs })
   where
     order = mkOrder {
@@ -554,10 +557,10 @@ exitAtMarket position operationSignalName = do
           posExecutionDeadline = Nothing }) position
 
     Nothing -> do
-      submitOrder (closeOrder inst)
+      oid <- submitOrder (closeOrder inst)
       modifyPosition (\pos ->
         pos { posCurrentOrder = Nothing,
-          posState = PositionWaitingCloseSubmission (closeOrder inst),
+          posState = PositionWaitingCloseSubmission (closeOrder inst) { orderId = oid },
           posNextState = Just PositionClosed,
           posSubmissionDeadline = Just $ 10 `addUTCTime` lastTs,
           posExecutionDeadline = Nothing }) position
@@ -578,11 +581,11 @@ exitAtLimit timeToCancel price position operationSignalName = do
   case posCurrentOrder position of
     Just order -> cancelOrder (orderId order)
     Nothing    -> doNothing
-  submitOrder (closeOrder inst)
+  oid <- submitOrder (closeOrder inst)
   appendToLog $ [t|exitAtLimit: %?, deadline: %?|] (posTicker position) (timeToCancel `addUTCTime` lastTs)
   modifyPosition (\pos ->
     pos { posCurrentOrder = Nothing,
-      posState = PositionWaitingCloseSubmission (closeOrder inst),
+      posState = PositionWaitingCloseSubmission (closeOrder inst) { orderId = oid },
       posNextState = Just PositionClosed,
       posSubmissionDeadline = Just $ 10 `addUTCTime` lastTs,
       posExecutionDeadline = Just $ timeToCancel `addUTCTime` lastTs }) position