12 changed files with 64 additions and 452 deletions
@ -1,136 +0,0 @@
@@ -1,136 +0,0 @@
|
||||
{-# LANGUAGE OverloadedStrings #-} |
||||
|
||||
module Broker ( |
||||
SignalId(..), |
||||
OrderPrice(..), |
||||
Operation(..), |
||||
OrderId(..), |
||||
OrderState(..), |
||||
Order(..), |
||||
Trade(..), |
||||
Broker(..) |
||||
) where |
||||
|
||||
import Data.Decimal |
||||
import Data.Time.Clock |
||||
import Data.Aeson |
||||
import Data.Aeson.Types |
||||
import Control.Monad |
||||
|
||||
data SignalId = SignalId { |
||||
strategyId :: String, |
||||
signalName :: String, |
||||
comment :: String } |
||||
deriving (Show, Eq) |
||||
|
||||
instance FromJSON SignalId where |
||||
parseJSON (Object o) = SignalId <$> |
||||
o .: "strategy-id" .!= "" <*> |
||||
o .: "signal-name" .!= "" <*> |
||||
o .: "commen" .!= "" |
||||
parseJSON _ = fail "Should be object" |
||||
|
||||
data OrderPrice = Market | Limit Decimal | Stop Decimal Decimal | StopMarket Decimal |
||||
deriving (Show, Eq) |
||||
|
||||
decimal :: (RealFrac r) => r -> Decimal |
||||
decimal = realFracToDecimal 10 |
||||
|
||||
instance FromJSON OrderPrice where |
||||
parseJSON (String s) = when (s /= "market") (fail "If string, then should be 'market'") >> |
||||
return Market |
||||
|
||||
parseJSON (Number n) = return $ Limit $ decimal n |
||||
parseJSON (Object v) = do |
||||
triggerPrice <- v .: "trigger" :: Parser Double |
||||
execPrice <- v .: "execution" |
||||
case execPrice of |
||||
(String s) -> when (s /= "market") (fail "If string, then should be 'market'") >> return $ StopMarket (decimal triggerPrice) |
||||
(Number n) -> return $ Stop (decimal triggerPrice) (decimal n) |
||||
_ -> fail "Should be either number or 'market'" |
||||
|
||||
parseJSON _ = fail "OrderPrice" |
||||
|
||||
data Operation = Buy | Sell |
||||
deriving (Show, Eq) |
||||
|
||||
instance FromJSON Operation where |
||||
parseJSON (String s) |
||||
| s == "buy" = return Buy |
||||
| s == "sell" = return Sell |
||||
| otherwise = fail "Should be either 'buy' or 'sell'" |
||||
parseJSON _ = fail "Should be string" |
||||
|
||||
type OrderId = Integer |
||||
|
||||
data OrderState = Unsubmitted |
||||
| Submitted |
||||
| PartiallyExecuted |
||||
| Executed |
||||
| Cancelled |
||||
| Rejected String |
||||
| Error String |
||||
deriving (Show, Eq) |
||||
|
||||
instance FromJSON OrderState where |
||||
parseJSON (String s) |
||||
| s == "unsubmitted" = return Unsubmitted |
||||
| s == "submitted" = return Submitted |
||||
| s == "partially-executed" = return PartiallyExecuted |
||||
| s == "executed" = return Executed |
||||
| s == "cancelled" = return Cancelled |
||||
| s == "rejected" = return $ Rejected "" |
||||
| s == "error" = return $ Broker.Error "" |
||||
| otherwise = fail "Invlaid state" |
||||
|
||||
parseJSON _ = fail "Should be string" |
||||
|
||||
data Order = Order { |
||||
orderId :: OrderId, |
||||
orderAccountId :: String, |
||||
orderSecurity :: String, |
||||
orderPrice :: OrderPrice, |
||||
orderQuantity :: Integer, |
||||
orderExecutedQuantity :: Integer, |
||||
orderOperation :: Operation, |
||||
orderState :: OrderState, |
||||
orderSignalId :: SignalId } |
||||
deriving (Show, Eq) |
||||
|
||||
instance FromJSON Order where |
||||
parseJSON (Object v) = Order <$> |
||||
v .:? "order-id" .!= 0 <*> |
||||
v .: "account" <*> |
||||
v .: "security" <*> |
||||
v .: "price" <*> |
||||
v .: "quantity" <*> |
||||
v .:? "executed-quantity" .!= 0 <*> |
||||
v .: "operation" <*> |
||||
v .: "state" .!= Unsubmitted <*> |
||||
v .: "signal-id" |
||||
|
||||
parseJSON _ = fail "Should be string" |
||||
|
||||
|
||||
data Trade = Trade { |
||||
tradeOrderId :: OrderId, |
||||
tradePrice :: Decimal, |
||||
tradeQuantity :: Integer, |
||||
tradeVolume :: Decimal, |
||||
tradeVolumeCurrency :: String, |
||||
tradeAccount :: String, |
||||
tradeSecurity :: String, |
||||
tradeTimestamp :: UTCTime, |
||||
tradeSignalId :: SignalId } |
||||
deriving (Show, Eq) |
||||
|
||||
data Broker = Broker { |
||||
accounts :: [String], |
||||
setTradeCallback :: Maybe (Trade -> IO ()) -> IO(), |
||||
setOrderCallback :: Maybe (Order -> IO ()) -> IO(), |
||||
submitOrder :: Order -> IO OrderId, |
||||
cancelOrder :: OrderId -> IO (), |
||||
getOrder :: OrderId -> IO (Maybe Order), |
||||
destroyBroker :: IO () |
||||
} |
||||
|
||||
@ -1,117 +0,0 @@
@@ -1,117 +0,0 @@
|
||||
{-# LANGUAGE OverloadedStrings #-} |
||||
|
||||
module Broker.Server ( |
||||
) where |
||||
|
||||
import System.ZMQ4 |
||||
import qualified Data.Map as M |
||||
import qualified Data.ByteString as B |
||||
import qualified Data.ByteString.Lazy as BL |
||||
import Data.ATrade |
||||
import Data.IORef |
||||
import qualified Data.HashMap.Strict as HM |
||||
import Broker |
||||
import Control.Concurrent |
||||
import Control.Exception |
||||
import Data.Aeson |
||||
import Data.Aeson.Types |
||||
import Data.Int |
||||
import Data.Time.Clock |
||||
import Data.List as L |
||||
import qualified Data.List.NonEmpty as LN |
||||
import System.Log.Logger |
||||
|
||||
type RequestSqnum = Int64 |
||||
type PeerId = B.ByteString |
||||
|
||||
data BrokerServerState = BrokerServerState { |
||||
bsSocket :: Socket Router, |
||||
orderMap :: M.Map OrderId B.ByteString, -- Matches 0mq client identities with corresponding orders |
||||
lastPacket :: M.Map B.ByteString (RequestSqnum, B.ByteString), |
||||
pendingNotifications :: [(Order, UTCTime)], -- List of tuples (Order with new state, Time when notification enqueued) |
||||
brokers :: [Broker] |
||||
} |
||||
|
||||
newtype BrokerServerHandle = BrokerServerHandle ThreadId |
||||
|
||||
mkBrokerServer :: [Broker] -> Context -> String -> IO BrokerServerHandle |
||||
mkBrokerServer brokers c ep = do |
||||
sock <- socket c Router |
||||
bind sock ep |
||||
tid <- myThreadId |
||||
state <- newIORef BrokerServerState { |
||||
bsSocket = sock, |
||||
orderMap = M.empty, |
||||
lastPacket = M.empty, |
||||
pendingNotifications = [], |
||||
brokers = brokers |
||||
} |
||||
BrokerServerHandle <$> forkIO (brokerServerThread state) |
||||
|
||||
data BrokerServerMessage = SubmitOrder RequestSqnum Order | CancelOrder RequestSqnum OrderId |
||||
data BrokerServerResponse = OrderIdResponse OrderId |
||||
instance ToJSON BrokerServerResponse where |
||||
toJSON (OrderIdResponse oid) = object ["order-id" .= oid] |
||||
|
||||
parseMessage :: Value -> Parser BrokerServerMessage |
||||
parseMessage (Object obj) = do |
||||
rqsqnum <- obj .: "request-sqnum" :: Parser Int64 |
||||
case HM.lookup "order" obj of |
||||
Just (Object orderJson) -> do |
||||
order <- obj .: "order" |
||||
return $ SubmitOrder rqsqnum order |
||||
_ -> case HM.lookup "cancel-order" obj of |
||||
Just orderIdJson -> do |
||||
order <- obj .: "cancel-order" |
||||
return $ CancelOrder rqsqnum order |
||||
Nothing -> fail "Either 'order' or 'cancel-order' field should be present" |
||||
where |
||||
|
||||
parseMessage _ = fail "Should be object" |
||||
|
||||
brokerServerThread :: IORef BrokerServerState -> IO () |
||||
brokerServerThread state = finally brokerServerThread' cleanup |
||||
where |
||||
cleanup = do |
||||
s <- bsSocket <$> readIORef state |
||||
close s |
||||
|
||||
brokerServerThread' = do |
||||
s <- bsSocket <$> readIORef state |
||||
msg <- receiveMulti s |
||||
tryDeliverPendingNotifications |
||||
handleMessage msg |
||||
|
||||
tryDeliverPendingNotifications = return () |
||||
|
||||
handleMessage :: [B.ByteString] -> IO () |
||||
handleMessage (peerId:_:json:_) = maybe (return ()) (handleMessage' peerId) (decode (BL.fromStrict json) >>= parseMaybe parseMessage) |
||||
handleMessage _ = warningM "BrokerServer" "Invalid packet received, should be at least 3 parts" |
||||
|
||||
handleMessage' :: PeerId -> BrokerServerMessage -> IO () |
||||
handleMessage' peerId (SubmitOrder sqnum order) = do |
||||
s <- bsSocket <$> readIORef state |
||||
lastPack <- M.lookup peerId . lastPacket <$> readIORef state |
||||
case shouldResend lastPack sqnum of |
||||
Just packet -> sendMulti s $ LN.fromList [peerId, B.empty, packet] |
||||
Nothing -> do |
||||
brs <- brokers <$> readIORef state |
||||
case findBroker brs (orderAccountId order) of |
||||
Just broker -> do |
||||
orderId <- submitOrder broker order |
||||
let packet = BL.toStrict . encode $ OrderIdResponse orderId |
||||
atomicModifyIORef' state (\s -> (s { lastPacket = M.insert peerId (sqnum, packet) $ lastPacket s }, ())) |
||||
sendMulti s $ LN.fromList [peerId, B.empty, packet] |
||||
|
||||
Nothing -> warningM "BrokerServer" $ "Invalid account requested: " ++ orderAccountId order |
||||
where |
||||
shouldResend lastPack sqnum = case lastPack of |
||||
Nothing -> Nothing |
||||
Just (oldSqnum, packet) -> if oldSqnum == sqnum |
||||
then Just packet |
||||
else Nothing |
||||
findBroker brokers account = L.find (L.elem account . accounts) brokers |
||||
|
||||
handleMessage' peerId (CancelOrder sqnum orderId) = undefined |
||||
|
||||
|
||||
@ -1,77 +0,0 @@
@@ -1,77 +0,0 @@
|
||||
|
||||
module Data.ATrade ( |
||||
Tick(..), |
||||
DataType(..), |
||||
serializeTick |
||||
) where |
||||
|
||||
import Data.Decimal |
||||
import Data.Time.Clock |
||||
import Data.DateTime |
||||
import Data.ByteString.Lazy as B |
||||
import Data.Text as T |
||||
import Data.Text.Encoding as E |
||||
import Data.List as L |
||||
import Data.Binary.Builder |
||||
|
||||
data DataType = Unknown |
||||
| Price |
||||
| OpenInterest |
||||
| BestBid |
||||
| BestOffer |
||||
| Depth |
||||
| TheoryPrice |
||||
| Volatility |
||||
| TotalSupply |
||||
| TotalDemand |
||||
deriving (Show, Eq, Ord) |
||||
|
||||
instance Enum DataType where |
||||
fromEnum x |
||||
| x == Price = 1 |
||||
| x == OpenInterest = 3 |
||||
| x == BestBid = 4 |
||||
| x == BestOffer = 5 |
||||
| x == Depth = 6 |
||||
| x == TheoryPrice = 7 |
||||
| x == Volatility = 8 |
||||
| x == TotalSupply = 9 |
||||
| x == TotalDemand = 10 |
||||
| x == Unknown = -1 |
||||
| otherwise = -1 |
||||
|
||||
toEnum x |
||||
| x == 1 = Price |
||||
| x == 3 = OpenInterest |
||||
| x == 4 = BestBid |
||||
| x == 5 = BestOffer |
||||
| x == 6 = Depth |
||||
| x == 7 = TheoryPrice |
||||
| x == 8 = Volatility |
||||
| x == 9 = TotalSupply |
||||
| x == 10 = TotalDemand |
||||
| otherwise = Unknown |
||||
|
||||
data Tick = Tick { |
||||
security :: String, |
||||
datatype :: DataType, |
||||
timestamp :: UTCTime, |
||||
value :: Decimal, |
||||
volume :: Integer |
||||
} deriving (Show, Eq) |
||||
|
||||
serializeTick :: Tick -> [ByteString] |
||||
serializeTick tick = header : [rawdata] |
||||
where |
||||
header = B.fromChunks [ E.encodeUtf8 . T.pack $ security tick ] |
||||
rawdata = toLazyByteString $ mconcat [ |
||||
putWord32le 1, |
||||
putWord64le $ fromIntegral . toSeconds . timestamp $ tick, |
||||
putWord32le $ fromIntegral . truncate . (* 1000000) . fractionalPart . utctDayTime . timestamp $ tick, |
||||
putWord32le $ fromIntegral . fromEnum . datatype $ tick, |
||||
putWord64le $ truncate . value $ tick, |
||||
putWord32le $ truncate . (* 1000000000) . fractionalPart $ value tick, |
||||
putWord32le $ fromIntegral $ volume tick ] |
||||
fractionalPart :: (RealFrac a) => a -> a |
||||
fractionalPart x = x - fromIntegral (floor x) |
||||
|
||||
@ -1,6 +0,0 @@
@@ -1,6 +0,0 @@
|
||||
module Lib |
||||
( someFunc |
||||
) where |
||||
|
||||
someFunc :: IO () |
||||
someFunc = putStrLn "someFunc" |
||||
@ -1,51 +0,0 @@
@@ -1,51 +0,0 @@
|
||||
|
||||
module QuoteSource.Server ( |
||||
startQuoteSourceServer, |
||||
stopQuoteSourceServer |
||||
) where |
||||
|
||||
import System.ZMQ4 |
||||
import Control.Concurrent.BoundedChan |
||||
import Data.ATrade |
||||
import Control.Concurrent hiding (readChan) |
||||
import Control.Monad |
||||
import Control.Exception |
||||
import qualified Data.ByteString.Lazy as BL |
||||
import Data.List.NonEmpty hiding (map) |
||||
import System.Log.Logger |
||||
|
||||
data QuoteSourceServer = QuoteSourceServerState { |
||||
ctx :: Context, |
||||
outSocket :: Socket Pub, |
||||
tickChannel :: BoundedChan Tick, |
||||
serverThreadId :: ThreadId |
||||
} |
||||
|
||||
serverThread :: QuoteSourceServer -> IO () |
||||
serverThread state = do |
||||
finally serverThread' cleanup |
||||
debugM "QuoteSource" "server thread done" |
||||
where |
||||
cleanup = close $ outSocket state |
||||
|
||||
serverThread' = forever $ do |
||||
tick <- readChan $ tickChannel state |
||||
sendMulti (outSocket state) $ fromList . map BL.toStrict $ serializeTick tick |
||||
|
||||
startQuoteSourceServer :: BoundedChan Tick -> Context -> String -> IO QuoteSourceServer |
||||
startQuoteSourceServer chan c ep = do |
||||
sock <- socket c Pub |
||||
bind sock ep |
||||
tid <- myThreadId |
||||
let state = QuoteSourceServerState { |
||||
ctx = c, |
||||
outSocket = sock, |
||||
tickChannel = chan, |
||||
serverThreadId = tid |
||||
} |
||||
stid <- forkIO $ serverThread state |
||||
return $ state { serverThreadId = stid } |
||||
|
||||
stopQuoteSourceServer :: QuoteSourceServer -> IO () |
||||
stopQuoteSourceServer server = killThread $ serverThreadId server |
||||
|
||||
Loading…
Reference in new issue