qbtrade/pkg/grpc/server.go
2024-06-27 22:42:38 +08:00

356 lines
8.9 KiB
Go

package grpc
import (
"context"
"fmt"
"net"
"strconv"
"time"
"github.com/pkg/errors"
log "github.com/sirupsen/logrus"
"google.golang.org/grpc"
"google.golang.org/grpc/reflection"
"git.qtrade.icu/lychiyu/qbtrade/pkg/pb"
"git.qtrade.icu/lychiyu/qbtrade/pkg/qbtrade"
"git.qtrade.icu/lychiyu/qbtrade/pkg/types"
)
type TradingService struct {
Config *qbtrade.Config
Environ *qbtrade.Environment
Trader *qbtrade.Trader
pb.UnimplementedTradingServiceServer
}
func (s *TradingService) SubmitOrder(ctx context.Context, request *pb.SubmitOrderRequest) (*pb.SubmitOrderResponse, error) {
sessionName := request.Session
if len(sessionName) == 0 {
return nil, fmt.Errorf("session name can not be empty")
}
session, ok := s.Environ.Session(sessionName)
if !ok {
return nil, fmt.Errorf("session %s not found", sessionName)
}
submitOrders := toSubmitOrders(request.SubmitOrders)
for i := range submitOrders {
if market, ok := session.Market(submitOrders[i].Symbol); ok {
submitOrders[i].Market = market
} else {
log.Warnf("session %s market %s not found", sessionName, submitOrders[i].Symbol)
}
}
// we will return this error later because some orders could be succeeded
createdOrders, _, err := qbtrade.BatchRetryPlaceOrder(ctx, session.Exchange, nil, nil, log.StandardLogger(), submitOrders...)
// convert response
resp := &pb.SubmitOrderResponse{
Session: sessionName,
Orders: nil,
}
for _, createdOrder := range createdOrders {
resp.Orders = append(resp.Orders, transOrder(session, createdOrder))
}
return resp, err
}
func (s *TradingService) CancelOrder(ctx context.Context, request *pb.CancelOrderRequest) (*pb.CancelOrderResponse, error) {
sessionName := request.Session
if len(sessionName) == 0 {
return nil, fmt.Errorf("session name can not be empty")
}
session, ok := s.Environ.Session(sessionName)
if !ok {
return nil, fmt.Errorf("session %s not found", sessionName)
}
uuidOrderID := ""
orderID, err := strconv.ParseUint(request.OrderId, 10, 64)
if err != nil {
// TODO: validate uuid
uuidOrderID = request.OrderId
}
session.Exchange.CancelOrders(ctx, types.Order{
SubmitOrder: types.SubmitOrder{
ClientOrderID: request.ClientOrderId,
},
OrderID: orderID,
UUID: uuidOrderID,
})
resp := &pb.CancelOrderResponse{}
return resp, nil
}
func (s *TradingService) QueryOrder(ctx context.Context, request *pb.QueryOrderRequest) (*pb.QueryOrderResponse, error) {
panic("implement me")
}
func (s *TradingService) QueryOrders(ctx context.Context, request *pb.QueryOrdersRequest) (*pb.QueryOrdersResponse, error) {
panic("implement me")
}
func (s *TradingService) QueryTrades(ctx context.Context, request *pb.QueryTradesRequest) (*pb.QueryTradesResponse, error) {
panic("implement me")
}
type UserDataService struct {
Config *qbtrade.Config
Environ *qbtrade.Environment
Trader *qbtrade.Trader
pb.UnimplementedUserDataServiceServer
}
func (s *UserDataService) Subscribe(request *pb.UserDataRequest, server pb.UserDataService_SubscribeServer) error {
sessionName := request.Session
if len(sessionName) == 0 {
return fmt.Errorf("session name can not be empty")
}
session, ok := s.Environ.Session(sessionName)
if !ok {
return fmt.Errorf("session %s not found", sessionName)
}
userDataStream := session.Exchange.NewStream()
userDataStream.OnOrderUpdate(func(order types.Order) {
err := server.Send(&pb.UserData{
Channel: pb.Channel_ORDER,
Event: pb.Event_UPDATE,
Orders: []*pb.Order{transOrder(session, order)},
})
if err != nil {
log.WithError(err).Errorf("grpc: can not send user data")
}
})
userDataStream.OnTradeUpdate(func(trade types.Trade) {
err := server.Send(&pb.UserData{
Channel: pb.Channel_TRADE,
Event: pb.Event_UPDATE,
Trades: []*pb.Trade{transTrade(session, trade)},
})
if err != nil {
log.WithError(err).Errorf("grpc: can not send user data")
}
})
balanceHandler := func(balances types.BalanceMap) {
err := server.Send(&pb.UserData{
Channel: pb.Channel_BALANCE,
Event: pb.Event_UPDATE,
Balances: transBalances(session, balances),
})
if err != nil {
log.WithError(err).Errorf("grpc: can not send user data")
}
}
userDataStream.OnBalanceUpdate(balanceHandler)
userDataStream.OnBalanceSnapshot(balanceHandler)
ctx := server.Context()
balances, err := session.Exchange.QueryAccountBalances(ctx)
if err != nil {
return err
}
err = server.Send(&pb.UserData{
Channel: pb.Channel_BALANCE,
Event: pb.Event_SNAPSHOT,
Balances: transBalances(session, balances),
})
if err != nil {
log.WithError(err).Errorf("grpc: can not send user data")
}
go userDataStream.Connect(ctx)
defer func() {
if err := userDataStream.Close(); err != nil {
log.WithError(err).Errorf("user data stream close error")
}
}()
<-ctx.Done()
return nil
}
type MarketDataService struct {
Config *qbtrade.Config
Environ *qbtrade.Environment
Trader *qbtrade.Trader
pb.UnimplementedMarketDataServiceServer
}
func (s *MarketDataService) Subscribe(request *pb.SubscribeRequest, server pb.MarketDataService_SubscribeServer) error {
exchangeSubscriptions := map[string][]types.Subscription{}
for _, sub := range request.Subscriptions {
session, ok := s.Environ.Session(sub.Exchange)
if !ok {
return fmt.Errorf("exchange %s not found", sub.Exchange)
}
ss, err := toSubscriptions(sub)
if err != nil {
return err
}
exchangeSubscriptions[session.Name] = append(exchangeSubscriptions[session.Name], ss)
}
streamPool := map[string]types.Stream{}
for sessionName, subs := range exchangeSubscriptions {
session, ok := s.Environ.Session(sessionName)
if !ok {
log.Errorf("session %s not found", sessionName)
continue
}
stream := session.Exchange.NewStream()
stream.SetPublicOnly()
for _, sub := range subs {
log.Infof("%s subscribe %s %s %+v", sessionName, sub.Channel, sub.Symbol, sub.Options)
stream.Subscribe(sub.Channel, sub.Symbol, sub.Options)
}
stream.OnMarketTrade(func(trade types.Trade) {
if err := server.Send(transMarketTrade(session, trade)); err != nil {
log.WithError(err).Error("grpc stream send error")
}
})
stream.OnBookSnapshot(func(book types.SliceOrderBook) {
if err := server.Send(transBook(session, book, pb.Event_SNAPSHOT)); err != nil {
log.WithError(err).Error("grpc stream send error")
}
})
stream.OnBookUpdate(func(book types.SliceOrderBook) {
if err := server.Send(transBook(session, book, pb.Event_UPDATE)); err != nil {
log.WithError(err).Error("grpc stream send error")
}
})
stream.OnKLineClosed(func(kline types.KLine) {
err := server.Send(transKLineResponse(session, kline))
if err != nil {
log.WithError(err).Error("grpc stream send error")
}
})
streamPool[sessionName] = stream
}
for _, stream := range streamPool {
go stream.Connect(server.Context())
}
defer func() {
for _, stream := range streamPool {
if err := stream.Close(); err != nil {
log.WithError(err).Errorf("market data stream close error")
}
}
}()
ctx := server.Context()
<-ctx.Done()
return ctx.Err()
}
func (s *MarketDataService) QueryKLines(ctx context.Context, request *pb.QueryKLinesRequest) (*pb.QueryKLinesResponse, error) {
exchangeName, err := types.ValidExchangeName(request.Exchange)
if err != nil {
return nil, err
}
for _, session := range s.Environ.Sessions() {
if session.ExchangeName == exchangeName {
response := &pb.QueryKLinesResponse{
Klines: nil,
Error: nil,
}
options := types.KLineQueryOptions{
Limit: int(request.Limit),
}
endTime := time.Now()
if request.EndTime != 0 {
endTime = time.Unix(request.EndTime, 0)
}
options.EndTime = &endTime
if request.StartTime != 0 {
startTime := time.Unix(request.StartTime, 0)
options.StartTime = &startTime
}
klines, err := session.Exchange.QueryKLines(ctx, request.Symbol, types.Interval(request.Interval), options)
if err != nil {
return nil, err
}
for _, kline := range klines {
response.Klines = append(response.Klines, transKLine(session, kline))
}
return response, nil
}
}
return nil, nil
}
type Server struct {
Config *qbtrade.Config
Environ *qbtrade.Environment
Trader *qbtrade.Trader
}
func (s *Server) ListenAndServe(bind string) error {
conn, err := net.Listen("tcp", bind)
if err != nil {
return errors.Wrapf(err, "failed to bind network at %s", bind)
}
var grpcServer = grpc.NewServer()
pb.RegisterMarketDataServiceServer(grpcServer, &MarketDataService{
Config: s.Config,
Environ: s.Environ,
Trader: s.Trader,
})
pb.RegisterTradingServiceServer(grpcServer, &TradingService{
Config: s.Config,
Environ: s.Environ,
Trader: s.Trader,
})
pb.RegisterUserDataServiceServer(grpcServer, &UserDataService{
Config: s.Config,
Environ: s.Environ,
Trader: s.Trader,
})
reflection.Register(grpcServer)
if err := grpcServer.Serve(conn); err != nil {
return errors.Wrap(err, "failed to serve grpc connections")
}
return nil
}