ccxtExchange.go

package plugins

import (
	"fmt"
	"log"
	"sort"
	"strconv"
	"time"

	"github.com/stellar/kelp/api"
	"github.com/stellar/kelp/model"
	"github.com/stellar/kelp/support/sdk"
	"github.com/stellar/kelp/support/utils"
)

const ccxtBalancePrecision = 10

// ensure that ccxtExchange conforms to the Exchange interface
var _ api.Exchange = ccxtExchange{}

// ccxtExchangeSpecificParamFactory knows how to create the exchange-specific params for each exchange
type ccxtExchangeSpecificParamFactory interface {
	getParamsForGetOrderBook() map[string]interface{}
	getParamsForAddOrder(submitMode api.SubmitMode) interface{}
	getParamsForGetTradeHistory() interface{}
}

// ccxtExchange is the implementation for the CCXT REST library that supports many exchanges (https://github.com/franz-see/ccxt-rest, https://github.com/ccxt/ccxt/)
type ccxtExchange struct {
	assetConverter     model.AssetConverterInterface
	delimiter          string
	ocOverridesHandler *OrderConstraintsOverridesHandler
	api                *sdk.Ccxt
	simMode            bool
	esParamFactory     ccxtExchangeSpecificParamFactory
}

// makeCcxtExchange is a factory method to make an exchange using the CCXT interface
func makeCcxtExchange(
	exchangeName string,
	orderConstraintOverrides map[model.TradingPair]model.OrderConstraints,
	apiKeys []api.ExchangeAPIKey,
	exchangeParams []api.ExchangeParam,
	headers []api.ExchangeHeader,
	simMode bool,
	esParamFactory ccxtExchangeSpecificParamFactory,
) (api.Exchange, error) {
	if len(apiKeys) == 0 {
		return nil, fmt.Errorf("need at least 1 ExchangeAPIKey, even if it is an empty key")
	}

	if len(apiKeys) != 1 {
		return nil, fmt.Errorf("need exactly 1 ExchangeAPIKey")
	}

	c, e := sdk.MakeInitializedCcxtExchange(exchangeName, apiKeys[0], exchangeParams, headers)
	if e != nil {
		return nil, fmt.Errorf("error making a ccxt exchange: %s", e)
	}

	ocOverridesHandler := MakeEmptyOrderConstraintsOverridesHandler()
	if orderConstraintOverrides != nil {
		ocOverridesHandler = MakeOrderConstraintsOverridesHandler(orderConstraintOverrides)
	}

	return ccxtExchange{
		assetConverter:     model.CcxtAssetConverter,
		delimiter:          "/",
		ocOverridesHandler: ocOverridesHandler,
		api:                c,
		simMode:            simMode,
		esParamFactory:     esParamFactory,
	}, nil
}

// GetTickerPrice impl.
func (c ccxtExchange) GetTickerPrice(pairs []model.TradingPair) (map[model.TradingPair]api.Ticker, error) {
	pairsMap, e := model.TradingPairs2Strings(c.assetConverter, c.delimiter, pairs)
	if e != nil {
		return nil, e
	}

	priceResult := map[model.TradingPair]api.Ticker{}
	for _, p := range pairs {
		tickerMap, e := c.api.FetchTicker(pairsMap[p])
		if e != nil {
			return nil, fmt.Errorf("error while fetching ticker price for trading pair %s: %s", pairsMap[p], e)
		}

		askPrice, e := utils.CheckFetchFloat(tickerMap, "ask")
		if e != nil {
			return nil, fmt.Errorf("unable to correctly fetch 'ask' value from tickerMap: %s", e)
		}
		bidPrice, e := utils.CheckFetchFloat(tickerMap, "bid")
		if e != nil {
			return nil, fmt.Errorf("unable to correctly fetch 'bid' value from tickerMap: %s", e)
		}
		lastPrice, e := utils.CheckFetchFloat(tickerMap, "last")
		if e != nil {
			return nil, fmt.Errorf("unable to correctly fetch 'last' value from tickerMap: %s", e)
		}

		pricePrecision := c.GetOrderConstraints(&p).PricePrecision
		priceResult[p] = api.Ticker{
			AskPrice:  model.NumberFromFloat(askPrice, pricePrecision),
			BidPrice:  model.NumberFromFloat(bidPrice, pricePrecision),
			LastPrice: model.NumberFromFloat(lastPrice, pricePrecision),
		}
	}

	return priceResult, nil
}

// GetAssetConverter impl
func (c ccxtExchange) GetAssetConverter() model.AssetConverterInterface {
	return c.assetConverter
}

// GetOrderConstraints impl
func (c ccxtExchange) GetOrderConstraints(pair *model.TradingPair) *model.OrderConstraints {
	pairString, e := pair.ToString(c.assetConverter, c.delimiter)
	if e != nil {
		// this should never really panic because we would have converted this trading pair to a string previously
		panic(e)
	}

	// load from CCXT's cache
	ccxtMarket := c.api.GetMarket(pairString)
	if ccxtMarket == nil {
		panic(fmt.Errorf("CCXT does not have precision and limit data for the passed in market: %s", pairString))
	}
	volumePrecision := ccxtMarket.Precision.Amount
	if volumePrecision == 0 {
		volumePrecision = ccxtMarket.Precision.Price
	}
	oc := model.MakeOrderConstraintsWithCost(ccxtMarket.Precision.Price, volumePrecision, ccxtMarket.Limits.Amount.Min, ccxtMarket.Limits.Cost.Min)

	return c.ocOverridesHandler.Apply(pair, oc)
}

// OverrideOrderConstraints impl, can partially override values for specific pairs
func (c ccxtExchange) OverrideOrderConstraints(pair *model.TradingPair, override *model.OrderConstraintsOverride) {
	c.ocOverridesHandler.Upsert(pair, override)
}

// GetAccountBalances impl
func (c ccxtExchange) GetAccountBalances(assetList []interface{}) (map[interface{}]model.Number, error) {
	balanceResponse, e := c.api.FetchBalance()
	if e != nil {
		return nil, e
	}

	m := map[interface{}]model.Number{}
	for _, elem := range assetList {
		var asset model.Asset
		if v, ok := elem.(model.Asset); ok {
			asset = v
		} else {
			return nil, fmt.Errorf("invalid type of asset passed in, only model.Asset accepted")
		}

		ccxtAssetString, e := c.GetAssetConverter().ToString(asset)
		if e != nil {
			return nil, e
		}

		if ccxtBalance, ok := balanceResponse[ccxtAssetString]; ok {
			m[asset] = *model.NumberFromFloat(ccxtBalance.Total, ccxtBalancePrecision)
		} else {
			m[asset] = *model.NumberConstants.Zero
		}
	}
	return m, nil
}

// GetOrderBook impl
func (c ccxtExchange) GetOrderBook(pair *model.TradingPair, maxCount int32) (*model.OrderBook, error) {
	maxCountInt := int(maxCount)

	// if the exchange has limitations on how many orders we can fetch, these params will handle it and adjust the fetchLimit accordingly
	fetchLimit := maxCountInt
	if c.esParamFactory != nil {
		orderbookParamsMap := c.esParamFactory.getParamsForGetOrderBook()
		if orderbookParamsMap != nil {
			if transformLimitFnResult, ok := orderbookParamsMap["transform_limit"]; ok {
				transformLimitFn := transformLimitFnResult.(func(limit int) (int, error))
				newLimit, e := transformLimitFn(int(maxCount))
				if e != nil {
					return nil, fmt.Errorf("error while transforming maxCount limit: %s", e)
				}
				fetchLimit = newLimit
			}
		}
	}

	pairString, e := pair.ToString(c.assetConverter, c.delimiter)
	if e != nil {
		return nil, fmt.Errorf("error converting pair to string: %s", e)
	}

	ob, e := c.api.FetchOrderBook(pairString, &fetchLimit)
	if e != nil {
		return nil, fmt.Errorf("error while fetching orderbook for trading pair '%s': %s", pairString, e)
	}
	if _, ok := ob["asks"]; !ok {
		return nil, fmt.Errorf("orderbook did not contain the 'asks' field: %v", ob)
	}
	if _, ok := ob["bids"]; !ok {
		return nil, fmt.Errorf("orderbook did not contain the 'bids' field: %v", ob)
	}

	askCcxtOrders := ob["asks"]
	bidCcxtOrders := ob["bids"]
	if fetchLimit != maxCountInt {
		// we may not have fetched all the requested levels because the exchange may not have had that many levels in depth
		if len(askCcxtOrders) > maxCountInt {
			askCcxtOrders = askCcxtOrders[:maxCountInt]
		}
		if len(bidCcxtOrders) > maxCountInt {
			bidCcxtOrders = bidCcxtOrders[:maxCountInt]
		}
	}

	asks := c.readOrders(askCcxtOrders, pair, model.OrderActionSell)
	bids := c.readOrders(bidCcxtOrders, pair, model.OrderActionBuy)
	return model.MakeOrderBook(pair, asks, bids), nil
}

func (c ccxtExchange) readOrders(orders []sdk.CcxtOrder, pair *model.TradingPair, orderAction model.OrderAction) []model.Order {
	pricePrecision := c.GetOrderConstraints(pair).PricePrecision
	volumePrecision := c.GetOrderConstraints(pair).VolumePrecision

	result := []model.Order{}
	for _, o := range orders {
		result = append(result, model.Order{
			Pair:        pair,
			OrderAction: orderAction,
			OrderType:   model.OrderTypeLimit,
			Price:       model.NumberFromFloat(o.Price, pricePrecision),
			Volume:      model.NumberFromFloat(o.Amount, volumePrecision),
			Timestamp:   nil,
		})
	}
	return result
}

// GetTradeHistory impl
func (c ccxtExchange) GetTradeHistory(pair model.TradingPair, maybeCursorStart interface{}, maybeCursorEnd interface{}) (*api.TradeHistoryResult, error) {
	pairString, e := pair.ToString(c.assetConverter, c.delimiter)
	if e != nil {
		return nil, fmt.Errorf("error converting pair to string: %s", e)
	}

	// TODO fix limit logic to check result so we get full history instead of just 50 trades
	const limit = 50
	tradesRaw, e := c.api.FetchMyTrades(pairString, limit, maybeCursorStart)
	if e != nil {
		return nil, fmt.Errorf("error while fetching trade history for trading pair '%s': %s", pairString, e)
	}

	var maybeExchangeSpecificParams interface{}
	if c.esParamFactory != nil {
		maybeExchangeSpecificParams = c.esParamFactory.getParamsForGetTradeHistory()
	}

	trades := []model.Trade{}
	for _, raw := range tradesRaw {
		var t *model.Trade
		t, e = c.readTrade(&pair, pairString, raw)
		if e != nil {
			return nil, fmt.Errorf("error while reading trade: %s", e)
		}

		orderID := ""
		if maybeExchangeSpecificParams != nil {
			paramsMap := maybeExchangeSpecificParams.(map[string]interface{})
			if oidRes, ok := paramsMap["order_id"]; ok {
				oidFn := oidRes.(func(info interface{}) (string, error))
				orderID, e = oidFn(raw.Info)
				if e != nil {
					return nil, fmt.Errorf("error while reading 'order_id' from raw.Info for exchange with specific params: %s", e)
				}
			}
		}
		t.OrderID = orderID

		trades = append(trades, *t)
	}

	sort.Sort(model.TradesByTsID(trades))
	cursor := maybeCursorStart
	if len(trades) > 0 {
		lastCursor := trades[len(trades)-1].Order.Timestamp.AsInt64()
		// add 1 to lastCursor so we don't repeat the same cursor on the next run
		cursor = strconv.FormatInt(lastCursor+1, 10)
	}

	return &api.TradeHistoryResult{
		Cursor: cursor,
		Trades: trades,
	}, nil
}

// GetLatestTradeCursor impl.
func (c ccxtExchange) GetLatestTradeCursor() (interface{}, error) {
	timeNowMillis := time.Now().UnixNano() / int64(time.Millisecond)
	latestTradeCursor := fmt.Sprintf("%d", timeNowMillis)
	return latestTradeCursor, nil
}

// GetTrades impl
func (c ccxtExchange) GetTrades(pair *model.TradingPair, maybeCursor interface{}) (*api.TradesResult, error) {
	pairString, e := pair.ToString(c.assetConverter, c.delimiter)
	if e != nil {
		return nil, fmt.Errorf("error converting pair to string: %s", e)
	}

	// TODO use cursor when fetching trades
	tradesRaw, e := c.api.FetchTrades(pairString)
	if e != nil {
		return nil, fmt.Errorf("error while fetching trades for trading pair '%s': %s", pairString, e)
	}

	trades := []model.Trade{}
	for _, raw := range tradesRaw {
		var t *model.Trade
		t, e = c.readTrade(pair, pairString, raw)
		if e != nil {
			return nil, fmt.Errorf("error while reading trade: %s", e)
		}
		trades = append(trades, *t)
	}

	sort.Sort(model.TradesByTsID(trades))
	cursor := maybeCursor
	if len(trades) > 0 {
		lastCursor := trades[len(trades)-1].Order.Timestamp.AsInt64()
		// add 1 to lastCursor so we don't repeat the same cursor on the next run
		cursor = strconv.FormatInt(lastCursor+1, 10)
	}

	return &api.TradesResult{
		Cursor: cursor,
		Trades: trades,
	}, nil
}

func (c ccxtExchange) readTrade(pair *model.TradingPair, pairString string, rawTrade sdk.CcxtTrade) (*model.Trade, error) {
	if rawTrade.Symbol != pairString {
		return nil, fmt.Errorf("expected '%s' for 'symbol' field, got: %s", pairString, rawTrade.Symbol)
	}

	pricePrecision := c.GetOrderConstraints(pair).PricePrecision
	volumePrecision := c.GetOrderConstraints(pair).VolumePrecision
	// use bigger precision for fee and cost since they are logically derived from amount and price
	feecCostPrecision := pricePrecision
	if volumePrecision > pricePrecision {
		feecCostPrecision = volumePrecision
	}

	trade := model.Trade{
		Order: model.Order{
			Pair:      pair,
			Price:     model.NumberFromFloat(rawTrade.Price, pricePrecision),
			Volume:    model.NumberFromFloat(rawTrade.Amount, volumePrecision),
			OrderType: model.OrderTypeLimit,
			Timestamp: model.MakeTimestamp(rawTrade.Timestamp),
		},
		TransactionID: model.MakeTransactionID(rawTrade.ID),
		Cost:          model.NumberFromFloat(rawTrade.Cost, feecCostPrecision),
		Fee:           model.NumberFromFloat(rawTrade.Fee.Cost, feecCostPrecision),
		// OrderID read by calling function depending on override set for exchange params in "orderId" field of Info object
	}

	if rawTrade.Side == "sell" {
		trade.OrderAction = model.OrderActionSell
	} else if rawTrade.Side == "buy" {
		trade.OrderAction = model.OrderActionBuy
	} else {
		return nil, fmt.Errorf("unrecognized value for 'side' field: %s", rawTrade.Side)
	}

	if rawTrade.Cost == 0.0 {
		trade.Cost = model.NumberFromFloat(rawTrade.Price*rawTrade.Amount, feecCostPrecision)
	}

	return &trade, nil
}

// GetOpenOrders impl
func (c ccxtExchange) GetOpenOrders(pairs []*model.TradingPair) (map[model.TradingPair][]model.OpenOrder, error) {
	pairStrings := []string{}
	string2Pair := map[string]model.TradingPair{}
	for _, pair := range pairs {
		pairString, e := pair.ToString(c.assetConverter, c.delimiter)
		if e != nil {
			return nil, fmt.Errorf("error converting pairs to strings: %s", e)
		}
		pairStrings = append(pairStrings, pairString)
		string2Pair[pairString] = *pair
	}

	openOrdersMap, e := c.api.FetchOpenOrders(pairStrings)
	if e != nil {
		return nil, fmt.Errorf("error while fetching open orders for trading pairs '%v': %s", pairStrings, e)
	}

	result := map[model.TradingPair][]model.OpenOrder{}
	for asset, ccxtOrderList := range openOrdersMap {
		pair, ok := string2Pair[asset]
		if !ok {
			return nil, fmt.Errorf("symbol %s returned from FetchOpenOrders was not in the original list of trading pairs: %v", asset, pairStrings)
		}

		openOrderList := []model.OpenOrder{}
		for _, o := range ccxtOrderList {
			openOrder, e := c.convertOpenOrderFromCcxt(&pair, o)
			if e != nil {
				return nil, fmt.Errorf("cannot convertOpenOrderFromCcxt: %s", e)
			}
			openOrderList = append(openOrderList, *openOrder)
		}
		result[pair] = openOrderList
	}
	return result, nil
}

func (c ccxtExchange) convertOpenOrderFromCcxt(pair *model.TradingPair, o sdk.CcxtOpenOrder) (*model.OpenOrder, error) {
	if o.Type != "limit" {
		return nil, fmt.Errorf("we currently only support limit order types")
	}

	orderAction := model.OrderActionSell
	if o.Side == "buy" {
		orderAction = model.OrderActionBuy
	}
	ts := model.MakeTimestamp(o.Timestamp)

	return &model.OpenOrder{
		Order: model.Order{
			Pair:        pair,
			OrderAction: orderAction,
			OrderType:   model.OrderTypeLimit,
			Price:       model.NumberFromFloat(o.Price, c.GetOrderConstraints(pair).PricePrecision),
			Volume:      model.NumberFromFloat(o.Amount, c.GetOrderConstraints(pair).VolumePrecision),
			Timestamp:   ts,
		},
		ID:             o.ID,
		StartTime:      ts,
		ExpireTime:     nil,
		VolumeExecuted: model.NumberFromFloat(o.Filled, c.GetOrderConstraints(pair).VolumePrecision),
	}, nil
}

// AddOrder impl
func (c ccxtExchange) AddOrder(order *model.Order, submitMode api.SubmitMode) (*model.TransactionID, error) {
	pairString, e := order.Pair.ToString(c.assetConverter, c.delimiter)
	if e != nil {
		return nil, fmt.Errorf("error converting pair to string: %s", e)
	}

	side := "sell"
	if order.OrderAction.IsBuy() {
		side = "buy"
	}

	log.Printf("ccxt is submitting order: pair=%s, orderAction=%s, orderType=%s, volume=%s, price=%s, submitMode=%s\n",
		pairString, order.OrderAction.String(), order.OrderType.String(), order.Volume.AsString(), order.Price.AsString(), submitMode.String())

	var maybeExchangeSpecificParams interface{}
	if c.esParamFactory != nil {
		maybeExchangeSpecificParams = c.esParamFactory.getParamsForAddOrder(submitMode)
	}
	ccxtOpenOrder, e := c.api.CreateLimitOrder(pairString, side, order.Volume.AsFloat(), order.Price.AsFloat(), maybeExchangeSpecificParams)
	if e != nil {
		return nil, fmt.Errorf("error while creating limit order %s: %s", *order, e)
	}

	return model.MakeTransactionID(ccxtOpenOrder.ID), nil
}

// CancelOrder impl
func (c ccxtExchange) CancelOrder(txID *model.TransactionID, pair model.TradingPair) (model.CancelOrderResult, error) {
	log.Printf("ccxt is canceling order: ID=%s, tradingPair: %s\n", txID.String(), pair.String())

	resp, e := c.api.CancelOrder(txID.String(), pair.String())
	if e != nil {
		return model.CancelResultFailed, e
	}

	if resp == nil {
		return model.CancelResultFailed, fmt.Errorf("response from CancelOrder was nil")
	}
	return model.CancelResultCancelSuccessful, nil
}

// PrepareDeposit impl
func (c ccxtExchange) PrepareDeposit(asset model.Asset, amount *model.Number) (*api.PrepareDepositResult, error) {
	// TODO implement
	return nil, nil
}

// GetWithdrawInfo impl
func (c ccxtExchange) GetWithdrawInfo(asset model.Asset, amountToWithdraw *model.Number, address string) (*api.WithdrawInfo, error) {
	// TODO implement
	return nil, nil
}

// WithdrawFunds impl
func (c ccxtExchange) WithdrawFunds(
	asset model.Asset,
	amountToWithdraw *model.Number,
	address string,
) (*api.WithdrawFunds, error) {
	// TODO implement
	return nil, nil
}