emit statistics to statsd

src/Config.hs

@@ -35,6 +35,8 @@ data TransaqConnectorConfig = TransaqConnectorConfig {
   tradesinkDashboard          :: T.Text,
   gotifyUri                   :: T.Text,
   gotifyToken                 :: T.Text,
+  statsdHost                  :: T.Text,
+  statsdPort                  :: Int,
   allTradesSubscriptions      :: [SubscriptionConfig],
   quotationsSubscriptions     :: [SubscriptionConfig],
   quotesSubscriptions         :: [SubscriptionConfig]

src/Main.hs

@@ -18,20 +18,27 @@ import           Colog                                    (LogAction, cfilter,
 import           Colog.Actions                            (logTextHandle)
 import           Config                                   (TransaqConnectorConfig (..),
                                                            loadConfig)
-import           Control.Concurrent                       (threadDelay)
+import           Control.Concurrent                       (killThread,
+                                                           threadDelay)
 import           Control.Concurrent.BoundedChan           (newBoundedChan)
 import           Control.Exception                        (bracket)
 import           Control.Monad                            (forever, void)
 import           Control.Monad.IO.Class                   (MonadIO)
 import qualified Data.Text                                as T
 import           Data.Version                             (showVersion)
-import           Debug.EventCounters                      (initEventCounters)
+import           Debug.EventCounters                      (emitEvent,
+                                                           initEventCounters)
 import           HistoryProviderServer                    (withHistoryProviderServer)
 import           Network.URI                              (parseURI)
 import           Prelude                                  hiding (log)
 import           System.IO                                (Handle,
                                                            IOMode (AppendMode),
                                                            withFile)
+import           System.Metrics                           (newStore)
+import           System.Remote.Monitoring.Statsd          (StatsdOptions (..),
+                                                           defaultStatsdOptions,
+                                                           forkStatsd,
+                                                           statsdThreadId)
 import           System.ZMQ4                              (withContext)
 import           TickerInfoServer                         (withTickerInfoServer)
 import qualified TXMLConnector                            as Connector
@@ -49,8 +56,11 @@ parseLoglevel _ = Trace
 
 main ::  IO ()
 main = do
-  initEventCounters
   cfg <- loadConfig "transaq-connector.dhall"
+  store <- newStore
+  initEventCounters store
+  statsdThread <- forkStatsd (defaultStatsdOptions { host = statsdHost cfg, port = statsdPort cfg, prefix = "transaq_connector"}) store
+
   let loglevel = parseLoglevel (logLevel cfg)
   withFile "transaq-connector.log" AppendMode $ \logH -> do
     let logger = mkLogger logH loglevel
@@ -86,7 +96,9 @@ main = do
                                     log Info "main" "Stopping TXMLConnector"
                                     Connector.stop txml) $ \_ -> do
                  withHistoryProviderServer ctx (historyProviderEndpoint cfg) txml tisH logger id $ \_ -> do
-                   forever $ threadDelay 1000000
+                   forever $ do
+                     threadDelay 200000
+                     emitEvent "main_loop"
     log Info "main" "Shutting down"
-
+    killThread $ statsdThreadId statsdThread
 

src/TXMLConnector/Internal.hs

@@ -114,7 +114,7 @@ import           GHC.Exts                       (IsList (..))
 import           Prelude                        hiding (log)
 import           TickerInfoServer               (TickerInfo (..),
                                                  TickerInfoServerHandle,
-                                                 putTickerInfo)
+                                                 getTickerInfo, putTickerInfo)
 import qualified Transaq
 import qualified TXML
 
@@ -312,20 +312,26 @@ handleTransaqData transaqData = do
         Just cb -> case BM.lookupR (ExchangeOrderId (tOrderNo transaqTrade)) trIdMap of
           Just oid -> case M.lookup oid orderMap of
             Just order -> do
-              let notif = BackendTradeNotification (fromTransaqTrade transaqTrade order)
+              tisH <- asks tisHandle
+              let tickerId' = tBoard transaqTrade <> "#" <> tSecCode transaqTrade
+              maybeTickerInfo <- liftIO $ getTickerInfo tickerId' tisH
+              let notif = BackendTradeNotification (fromTransaqTrade transaqTrade order maybeTickerInfo)
               log Debug "TXMLConnector.WorkThread" $ "Sending trade notification: " <> (T.pack . show) notif
               liftIO $ cb notif
             _ -> log Warning "TXMLConnector.WorkThread" $ "Unable to find order for trade: " <> (T.pack . show) transaqTrade
           _ -> log Warning "TXMLConnector.WorkThread" $ "Unable to find order in ordermap: " <> (T.pack . show) transaqTrade
         Nothing -> log Warning "TXMLConnector.WorkThread" "No callback for trade notification!"
 
-    fromTransaqTrade transaqTrade order =
+    fromTransaqTrade transaqTrade order maybeTickerInfo =
+      let vol = case maybeTickerInfo of
+            Just tickerInfo -> (tPrice transaqTrade / tiTickSize tickerInfo * tiTickPrice tickerInfo)
+            Nothing -> tPrice transaqTrade in
       Trade
       {
         tradeOrderId = orderId order
       , tradePrice = fromDouble (tPrice transaqTrade)
       , tradeQuantity = fromIntegral $ tQuantity transaqTrade
-      , tradeVolume = fromDouble $ tValue transaqTrade
+      , tradeVolume = fromDouble vol
       , tradeVolumeCurrency = ""
       , tradeOperation = fromDirection (tBuysell transaqTrade)
       , tradeAccount = tClient transaqTrade <> "#" <> tUnion transaqTrade
@@ -633,6 +639,7 @@ securityToTickerInfo sec =
     tiTicker = sBoard sec <> "#" <> sSeccode sec
   , tiLotSize = sLotSize sec
   , tiTickSize = sMinStep sec
+  , tiTickPrice = sPointCost sec
   }
 
 parseSecurityId :: TickerId -> Maybe SecurityId

src/TickerInfoServer.hs

@@ -26,7 +26,7 @@ import           Data.Aeson                  (FromJSON (parseJSON),
                                               ToJSON (toJSON), decode,
                                               eitherDecode, encode, object,
                                               withObject)
-import           Data.Aeson.Types            ((.:), (.=))
+import           Data.Aeson.Types            ((.!=), (.:), (.:?), (.=))
 import qualified Data.ByteString.Lazy        as BL
 import           Data.List.NonEmpty          (NonEmpty ((:|)))
 import qualified Data.Map.Strict             as M
@@ -44,6 +44,7 @@ data TickerInfo =
     tiTicker    :: TickerId
   , tiLotSize   :: Int
   , tiTickSize  :: Double
+  , tiTickPrice :: Double
   } deriving (Show, Eq, Ord)
 
 instance FromJSON TickerInfo where
@@ -51,12 +52,14 @@ instance FromJSON TickerInfo where
     TickerInfo <$>
       obj .: "ticker" <*>
       obj .: "lot_size" <*>
-      obj .: "tick_size")
+      obj .: "tick_size" <*>
+      obj .:? "tick_price" .!= 1)
 
 instance ToJSON TickerInfo where
   toJSON ti = object [ "ticker" .= tiTicker ti,
     "lot_size" .= tiLotSize ti,
-    "tick_size" .= tiTickSize ti ]
+    "tick_size" .= tiTickSize ti,
+    "tick_price" .= tiTickPrice ti]
 
 newtype TickerInfoRequest =
   TickerInfoRequest

stack.yaml

@@ -33,6 +33,7 @@ packages:
 - ../libatrade
 - ../zeromq4-haskell-zap
 - ../eventcounters
+- ../ekg-statsd
 
 # Dependency packages to be pulled from upstream that are not in the resolver.
 # These entries can reference officially published versions as well as

transaq-connector.cabal

@@ -57,6 +57,8 @@ executable transaq-connector
                      , bimap
                      , deque
                      , network-uri
+                     , ekg-statsd
+                     , ekg-core
   extra-lib-dirs:      lib
   ghc-options:         -Wall
                        -Wcompat
@@ -115,6 +117,8 @@ test-suite transaq-connector-test
                      , bimap
                      , deque
                      , network-uri
+                     , ekg-statsd
+                     , ekg-core
   default-extensions:  OverloadedStrings
                      , MultiWayIf
                      , MultiParamTypeClasses