9 changed files with 290 additions and 94 deletions
@ -0,0 +1,75 @@
@@ -0,0 +1,75 @@
|
||||
{-# LANGUAGE OverloadedStrings #-} |
||||
|
||||
module ATrade.Broker.TradeSinks.TelegramTradeSink ( |
||||
withTelegramTradeSink |
||||
) where |
||||
|
||||
import Control.Exception |
||||
import Control.Concurrent |
||||
import qualified Control.Concurrent.BoundedChan as BC |
||||
import Data.Aeson |
||||
import Data.Aeson.Types |
||||
import Data.IORef |
||||
import Data.Maybe |
||||
import Data.List.NonEmpty |
||||
import qualified Data.List as L |
||||
import qualified Data.ByteString as B hiding (putStrLn) |
||||
import qualified Data.ByteString.Lazy as BL hiding (putStrLn) |
||||
import System.Log.Logger |
||||
import Control.Monad.Loops |
||||
import Control.Monad.Extra |
||||
|
||||
import ATrade.Types |
||||
import ATrade.Broker.Protocol |
||||
import Network.Connection |
||||
import Network.HTTP.Client |
||||
import Network.HTTP.Client.TLS |
||||
|
||||
import qualified Data.Text as T |
||||
import qualified Data.Text.Lazy as TL |
||||
import Data.Text.Format |
||||
import qualified Data.ByteString.UTF8 as BU8 |
||||
|
||||
withTelegramTradeSink apitoken chatId f = do |
||||
killMv <- newEmptyMVar |
||||
chan <- BC.newBoundedChan 1000 |
||||
bracket (forkIO $ sinkThread apitoken chatId killMv chan) (stopSinkThread killMv) (\_ -> f $ sink chan) |
||||
where |
||||
sink = BC.writeChan |
||||
|
||||
sinkThread apitoken chatId killMv chan = do |
||||
man <- newManager $ mkManagerSettings tlsSettings Nothing |
||||
whileM_ (not <$> wasKilled) $ do |
||||
maybeTrade <- BC.tryReadChan chan |
||||
whenJust maybeTrade (\trade -> sendMessage man apitoken chatId $ format "Trade: {} {} of {} at {} for {} ({}/{})" |
||||
(show (tradeOperation trade), |
||||
show (tradeQuantity trade), |
||||
tradeSecurity trade, |
||||
show (tradePrice trade), |
||||
tradeAccount trade, |
||||
(strategyId . tradeSignalId) trade, |
||||
(signalName . tradeSignalId) trade)) |
||||
where |
||||
tlsSettings = TLSSettingsSimple { settingDisableCertificateValidation = True, settingDisableSession = False, settingUseServerName = False } |
||||
wasKilled = isJust <$> tryReadMVar killMv |
||||
encodeTrade :: Trade -> B.ByteString |
||||
encodeTrade = BL.toStrict . encode . convertTrade |
||||
convertTrade trade = TradeSinkTrade { |
||||
tsAccountId = tradeAccount trade, |
||||
tsSecurity = tradeSecurity trade, |
||||
tsPrice = fromRational . toRational . tradePrice $ trade, |
||||
tsQuantity = fromInteger $ tradeQuantity trade, |
||||
tsVolume = fromRational . toRational . tradeVolume $ trade, |
||||
tsCurrency = tradeVolumeCurrency trade, |
||||
tsOperation = tradeOperation trade, |
||||
tsExecutionTime = tradeTimestamp trade, |
||||
tsSignalId = tradeSignalId trade |
||||
} |
||||
|
||||
sendMessage httpManager apitoken chatId text = do |
||||
req <- parseUrl $ "https://api.telegram.org/bot" ++ (T.unpack apitoken) ++ "/sendMessage" |
||||
void $ withResponse (req { method = "POST", requestHeaders = [("Content-Type", BU8.fromString "application/json")], requestBody = (RequestBodyLBS . encode) (object ["chat_id" .= chatId, "text" .= text]) }) httpManager (\resp -> brConsume (responseBody resp)) |
||||
|
||||
|
||||
stopSinkThread killMv threadId = putMVar killMv () >> killThread threadId |
||||
|
||||
@ -0,0 +1,75 @@
@@ -0,0 +1,75 @@
|
||||
|
||||
module ATrade.Broker.TradeSinks.ZMQTradeSink ( |
||||
withZMQTradeSink |
||||
) where |
||||
|
||||
import Control.Exception |
||||
import Control.Concurrent |
||||
import qualified Control.Concurrent.BoundedChan as BC |
||||
import Data.Aeson |
||||
import Data.IORef |
||||
import Data.Maybe |
||||
import qualified Data.Text as T |
||||
import Data.List.NonEmpty |
||||
import qualified Data.List as L |
||||
import qualified Data.ByteString as B hiding (putStrLn) |
||||
import qualified Data.ByteString.Lazy as BL hiding (putStrLn) |
||||
import Control.Monad.Loops |
||||
import Control.Monad.Extra |
||||
import System.Log.Logger |
||||
import System.Timeout |
||||
import System.ZMQ4 |
||||
|
||||
import ATrade.Types |
||||
import ATrade.Broker.Protocol |
||||
|
||||
withZMQTradeSink ctx tradeSinkEp f = do |
||||
killMv <- newEmptyMVar |
||||
chan <- BC.newBoundedChan 1000 |
||||
bracket (forkIO $ sinkThread ctx tradeSinkEp killMv chan) (stopSinkThread killMv) (\_ -> f $ sink chan) |
||||
where |
||||
sink = BC.writeChan |
||||
|
||||
sinkThread ctx tradeSinkEp killMv chan = whileM_ (not <$> wasKilled) $ do |
||||
handle (\e -> do |
||||
warningM "Broker.Server" $ "Trade sink: exception: " ++ show (e :: SomeException) ++ "; isZMQ: " ++ show (isZMQError e) |
||||
when (isZMQError e) $ do |
||||
debugM "Broker.Server" "Rethrowing exception" |
||||
throwIO e) sinkThread' |
||||
where |
||||
sinkThread' = withSocket ctx Dealer (\sock -> do |
||||
connect sock $ T.unpack tradeSinkEp |
||||
whenM (not <$> wasKilled) $ do |
||||
maybeTrade <- BC.tryReadChan chan |
||||
case maybeTrade of |
||||
Just trade -> do |
||||
sendMulti sock $ B.empty :| [encodeTrade trade] |
||||
void $ receiveMulti sock |
||||
Nothing -> do |
||||
sendMulti sock $ B.empty :| [BL.toStrict $ encode TradeSinkHeartBeat] |
||||
events <- poll 1000 [Sock sock [In] Nothing] |
||||
if L.null . L.head $ events |
||||
then warningM "Broker.Server" "Trade sink timeout" |
||||
else do |
||||
void . receive $ sock -- anything will do |
||||
sinkThread') |
||||
|
||||
|
||||
isZMQError e = "ZMQError" `L.isPrefixOf` show e |
||||
wasKilled = isJust <$> tryReadMVar killMv |
||||
encodeTrade :: Trade -> B.ByteString |
||||
encodeTrade = BL.toStrict . encode . convertTrade |
||||
convertTrade trade = TradeSinkTrade { |
||||
tsAccountId = tradeAccount trade, |
||||
tsSecurity = tradeSecurity trade, |
||||
tsPrice = fromRational . toRational . tradePrice $ trade, |
||||
tsQuantity = fromInteger $ tradeQuantity trade, |
||||
tsVolume = fromRational . toRational . tradeVolume $ trade, |
||||
tsCurrency = tradeVolumeCurrency trade, |
||||
tsOperation = tradeOperation trade, |
||||
tsExecutionTime = tradeTimestamp trade, |
||||
tsSignalId = tradeSignalId trade |
||||
} |
||||
|
||||
stopSinkThread killMv threadId = putMVar killMv () >> killThread threadId |
||||
|
||||
@ -0,0 +1,47 @@
@@ -0,0 +1,47 @@
|
||||
{-# LANGUAGE OverloadedStrings #-} |
||||
|
||||
module TestZMQTradeSink ( |
||||
unitTests |
||||
) where |
||||
|
||||
import Test.Tasty |
||||
import Test.Tasty.SmallCheck as SC |
||||
import Test.Tasty.QuickCheck as QC |
||||
import Test.Tasty.HUnit |
||||
|
||||
import ATrade.Types |
||||
import ATrade.Broker.Protocol |
||||
import ATrade.Broker.TradeSinks.ZMQTradeSink |
||||
import Control.Concurrent |
||||
import System.ZMQ4 |
||||
import Data.Aeson |
||||
import Data.Time.Calendar |
||||
import Data.Time.Clock |
||||
import qualified Data.ByteString as B |
||||
import qualified Data.ByteString.Lazy as BL |
||||
|
||||
unitTests = testGroup "Broker.Server.TradeSinks.ZMQTradeSink" [ testZMQTradeSink ] |
||||
|
||||
testZMQTradeSink = testCase "Test ZMQTradeSink trade serialization" $ |
||||
withContext (\ctx -> withSocket ctx Rep (\insock -> do |
||||
bind insock "inproc://test-sink" |
||||
withZMQTradeSink ctx "inproc://test-sink" (\f -> do |
||||
f trade |
||||
raw <- receive insock |
||||
send insock [] B.empty |
||||
case decode $ BL.fromStrict raw of |
||||
Just t -> mkTradeMessage trade @?= t |
||||
Nothing -> assertFailure "Unable to decode incoming message"))) |
||||
where |
||||
trade = Trade { |
||||
tradeOrderId = 0, |
||||
tradePrice = 10, |
||||
tradeQuantity = 20, |
||||
tradeVolume = 30, |
||||
tradeVolumeCurrency = "TEST", |
||||
tradeOperation = Buy, |
||||
tradeAccount = "FOO", |
||||
tradeSecurity = "BAR", |
||||
tradeTimestamp = UTCTime (fromGregorian 1970 1 1) 0, |
||||
tradeSignalId = SignalId "foo" "bar" "" } |
||||
|
||||
Loading…
Reference in new issue