qbtrade/pkg/strategy/ccinr/strategy.go

package ccinr

import (
	"context"
	"fmt"
	"git.qtrade.icu/lychiyu/qbtrade/pkg/exchange/binance"
	"git.qtrade.icu/lychiyu/qbtrade/pkg/fixedpoint"
	indicatorv2 "git.qtrade.icu/lychiyu/qbtrade/pkg/indicator/v2"
	"git.qtrade.icu/lychiyu/qbtrade/pkg/qbtrade"
	"git.qtrade.icu/lychiyu/qbtrade/pkg/strategy/common"
	"git.qtrade.icu/lychiyu/qbtrade/pkg/types"
	log "github.com/sirupsen/logrus"
	"strings"
	"sync"
)

const ID = "ccinr"

func init() {
	qbtrade.RegisterStrategy(ID, &Strategy{})
}

type Strategy struct {
	*common.Strategy

	Market      types.Market
	Environment *qbtrade.Environment
	markets     map[string]types.Market
	//Symbol      string `json:"symbol"`

	Symbols        []string         `json:"symbols"`
	Interval       types.Interval   `json:"interval"`
	NrCount        int              `json:"nrCount"`
	StrictMode     bool             `json:"strictMode"`
	PlacePriceType int              `json:"placePriceType"`
	DryRun         bool             `json:"dryRun"`
	CCIWindow      int              `json:"cciWindow"`
	LongCCI        fixedpoint.Value `json:"longCCI"`
	ShortCCI       fixedpoint.Value `json:"shortCCI"`
	Leverage       fixedpoint.Value `json:"leverage"`
	ProfitRange    fixedpoint.Value `json:"profitRange"`
	LossRange      fixedpoint.Value `json:"lossRange"`

	qbtrade.QuantityOrAmount

	//Position        *types.Position               `persistence:"position"`
	Positions       map[string]*types.Position    `persistence:"position"`
	ProfitStats     map[string]*types.ProfitStats `persistence:"profit_stats"`
	ExchangeSession *qbtrade.ExchangeSession
	orderExecutor   *qbtrade.GeneralOrderExecutor
	orderExecutors  map[string]*qbtrade.GeneralOrderExecutor

	qbtrade.StrategyController

	Traded     map[string]bool
	TradeType  map[string]string
	TradeRetry map[string]int

	// orders
	LongOrder        map[string]types.SubmitOrder
	LongProfitOrder  map[string]types.SubmitOrder
	LongLossOrder    map[string]types.SubmitOrder
	ShortOrder       map[string]types.SubmitOrder
	ShortProfitOrder map[string]types.SubmitOrder
	ShortLossOrder   map[string]types.SubmitOrder

	// 开仓
	OpenTrade map[string][]types.Trade
	// 清仓
	EndTrade     map[string][]types.Trade
	OpenQuantity map[string]fixedpoint.Value
	EndQuantity  map[string]fixedpoint.Value

	nr  map[string]*indicatorv2.NRStrean
	cci map[string]*indicatorv2.CCIStream

	TotalProfit     fixedpoint.Value
	TotalFree       fixedpoint.Value
	TotalOrderCount int
}

func (s *Strategy) ID() string {
	return ID
}

func (s *Strategy) Subscribe(session *qbtrade.ExchangeSession) {
	for _, symbol := range s.Symbols {
		session.Subscribe(types.KLineChannel, symbol, types.SubscribeOptions{Interval: s.Interval})
		session.Subscribe(types.MarketTradeChannel, symbol, types.SubscribeOptions{})
	}
}

func (s *Strategy) Initialize() error {
	if s.Strategy == nil {
		s.Strategy = &common.Strategy{}
	}

	return nil
}

func (s *Strategy) InstanceID() string {
	return fmt.Sprintf("%s:%s:%s", ID, strings.Join(s.Symbols, "-"), s.Interval)
}

func (s *Strategy) ClosePosition(ctx context.Context, percentage fixedpoint.Value) error {
	return s.orderExecutor.ClosePosition(ctx, percentage)
}

func (s *Strategy) CurrentPosition() *types.Position {
	return s.Position
}

func (s *Strategy) cancelOrders(ctx context.Context, symbol string) {
	if len(s.orderExecutors[symbol].ActiveMakerOrders().Orders()) <= 0 {
		return
	}
	log.Infof(fmt.Sprintf("[%s] the order is not filled, will cancel all orders", symbol))
	if err := s.orderExecutors[symbol].GracefulCancel(ctx); err != nil {
		log.WithError(err).Errorf("failed to cancel orders")
	}

	s.Traded[symbol] = false
	s.TradeType[symbol] = ""
	s.TradeRetry[symbol] = 0
}

func (s *Strategy) placeOrders(ctx context.Context, kline types.KLine) {
	symbol := kline.Symbol
	orders, err := s.generateOrders(ctx, kline)
	if err != nil {
		log.WithError(err).Error(fmt.Sprintf("failed to generate orders (%s)", symbol))
		return
	}
	log.Infof("orders: %+v", orders)

	if s.DryRun {
		log.Infof("dry run, not submitting orders (%s)", symbol)
		return
	}

	createdOrders, err := s.orderExecutors[symbol].SubmitOrders(ctx, orders...)
	if err != nil {
		log.WithError(err).Error(fmt.Sprintf("failed to submit orders (%s)", symbol))
		return
	}
	log.Infof("created orders (%s): %+v", symbol, createdOrders)
	return
}

func (s *Strategy) getPlacePrice(ctx context.Context, kline types.KLine) fixedpoint.Value {
	symbol := kline.Symbol

	placePrice := fixedpoint.Value(0)
	midPrice := (kline.High.Add(kline.Low)).Div(fixedpoint.Value(2.0))
	switch s.PlacePriceType {
	case 0:
		if s.TradeType[symbol] == "long" {
			placePrice = kline.High
		} else if s.TradeType[symbol] == "short" {
			placePrice = kline.Low
		}
	case 1:
		if s.TradeType[symbol] == "long" {
			placePrice = kline.Low
		} else if s.TradeType[symbol] == "short" {
			placePrice = kline.High
		}
	case 2:
		if s.TradeType[symbol] == "long" {
			placePrice = midPrice
		} else if s.TradeType[symbol] == "short" {
			placePrice = midPrice
		}
	}
	return placePrice
}

func (s *Strategy) generateOrders(ctx context.Context, kline types.KLine) ([]types.SubmitOrder, error) {
	var orders []types.SubmitOrder
	symbol := kline.Symbol

	log.Infof(fmt.Sprintf("place order keline info: symbol %s, high %v, low %v, open %v, close %v", symbol,
		kline.High.Float64(), kline.Low.Float64(), kline.Open.Float64(), kline.Close.Float64()))

	if s.TradeType[symbol] == "" {
		return orders, nil
	}

	// 获取下单价格
	placePrice := s.getPlacePrice(ctx, kline)

	// 下单数量
	placeQuantity := s.QuantityOrAmount.CalculateQuantity(placePrice).Mul(s.Leverage)
	log.Infof(fmt.Sprintf("will place order, price %v, quantity %v", placePrice.Float64(),
		placeQuantity.Float64()))

	s.ShortOrder[symbol] = types.SubmitOrder{
		Symbol:       symbol,
		Side:         types.SideTypeSell,
		Type:         types.OrderTypeLimit,
		Price:        placePrice,
		PositionSide: types.PositionSideTypeShort,
		Quantity:     placeQuantity,
		TimeInForce:  types.TimeInForceGTC,
		Market:       s.Market,
	}

	s.ShortProfitOrder[symbol] = types.SubmitOrder{
		Symbol:        symbol,
		Side:          types.SideTypeBuy,
		Type:          types.OrderTypeTakeProfitMarket,
		PositionSide:  types.PositionSideTypeShort,
		StopPrice:     placePrice.Sub(placePrice.Mul(s.ProfitRange)),
		TimeInForce:   types.TimeInForceGTC,
		Market:        s.Market,
		ClosePosition: true,
	}

	s.ShortLossOrder[symbol] = types.SubmitOrder{
		Symbol:        symbol,
		Side:          types.SideTypeBuy,
		Type:          types.OrderTypeStopMarket,
		PositionSide:  types.PositionSideTypeShort,
		StopPrice:     placePrice.Add(placePrice.Mul(s.LossRange)),
		TimeInForce:   types.TimeInForceGTC,
		Market:        s.Market,
		ClosePosition: true,
	}

	s.LongOrder[symbol] = types.SubmitOrder{
		Symbol:       symbol,
		Side:         types.SideTypeBuy,
		Type:         types.OrderTypeLimit,
		Price:        placePrice,
		PositionSide: types.PositionSideTypeLong,
		Quantity:     placeQuantity,
		TimeInForce:  types.TimeInForceGTC,
		Market:       s.Market,
	}

	s.LongProfitOrder[symbol] = types.SubmitOrder{
		Symbol:        symbol,
		Side:          types.SideTypeSell,
		Type:          types.OrderTypeTakeProfitMarket,
		PositionSide:  types.PositionSideTypeLong,
		StopPrice:     placePrice.Add(placePrice.Mul(s.ProfitRange)),
		TimeInForce:   types.TimeInForceGTC,
		Market:        s.Market,
		ClosePosition: true,
	}

	s.LongLossOrder[symbol] = types.SubmitOrder{
		Symbol:        symbol,
		Side:          types.SideTypeSell,
		Type:          types.OrderTypeStopMarket,
		PositionSide:  types.PositionSideTypeLong,
		StopPrice:     placePrice.Sub(placePrice.Mul(s.LossRange)),
		TimeInForce:   types.TimeInForceGTC,
		Market:        s.Market,
		ClosePosition: true,
	}

	if s.TradeType[symbol] == "short" {
		// 挂空单
		orders = append(orders, s.ShortOrder[symbol])
		// 空单止盈
		orders = append(orders, s.ShortProfitOrder[symbol])
		// 空单止损
		orders = append(orders, s.ShortLossOrder[symbol])
	}

	if s.TradeType[symbol] == "long" {
		// 挂多单
		orders = append(orders, s.LongOrder[symbol])
		// 多单止盈
		orders = append(orders, s.LongProfitOrder[symbol])
		// 多单止损
		orders = append(orders, s.LongLossOrder[symbol])
	}

	return orders, nil
}

func (s *Strategy) notifyProfit(ctx context.Context, symbol string) {
	if s.EndQuantity[symbol] != s.OpenQuantity[symbol] {
		return
	}
	profit := fixedpoint.Value(0)
	openProfit := fixedpoint.Value(0)
	endProfit := fixedpoint.Value(0)
	free := fixedpoint.Value(0)

	var openMsgs []string
	var endMsgs []string
	// 开仓成本
	for _, trade := range s.OpenTrade[symbol] {
		openProfit = openProfit.Add(trade.Price.Mul(trade.Quantity))
		free = free.Add(trade.Fee)
		openMsgs = append(openMsgs, fmt.Sprintf("价格：%v, 数量：%v, 手续费：%v；",
			trade.Price.Float64(), trade.Quantity.Float64(), trade.Fee.Float64()))
	}

	// 清仓资产
	for _, trade := range s.EndTrade[symbol] {
		endProfit = endProfit.Add(trade.Price.Mul(trade.Quantity))
		free = free.Add(trade.Fee)
		endMsgs = append(endMsgs, fmt.Sprintf("价格：%v, 数量：%v, 手续费：%v；",
			trade.Price.Float64(), trade.Quantity.Float64(), trade.Fee.Float64()))
	}

	side := s.OpenTrade[symbol][0].Side
	// 做多
	if side == types.SideTypeBuy {
		profit = endProfit.Sub(openProfit).Sub(free)
	}

	// 做空
	if side == types.SideTypeSell {
		profit = openProfit.Sub(endProfit).Sub(free)
	}

	msg := fmt.Sprintf("交易完成：\n 币种: %s, 方向：%v, 收益：%v, 手续费：%v \n Trade详情：\n  开仓Trade：\n   %s\n  清仓Trade：\n   %s",
		symbol, s.TradeType[symbol], profit.Float64(), free.Float64(), strings.Join(openMsgs, "\n"), strings.Join(endMsgs, "\n"))

	s.TotalProfit = s.TotalProfit.Add(profit)
	s.TotalFree = s.TotalFree.Add(free)
	s.TotalOrderCount += 1

	log.Infof(msg)
	qbtrade.Notify(msg)

	// 重置
	s.OpenTrade[symbol] = []types.Trade{}
	s.EndTrade[symbol] = []types.Trade{}
	s.OpenQuantity[symbol] = fixedpoint.Value(0)
	s.EndQuantity[symbol] = fixedpoint.Value(0)

	// 记得取消订单
	s.cancelOrders(ctx, symbol)

	qbtrade.Notify(fmt.Sprintf("总开仓：%v, 总收益：%v, 总手续费：%v", s.TotalOrderCount, s.TotalProfit.Float64(), s.TotalFree.Float64()))
}
func (s *Strategy) Run(ctx context.Context, orderExecutor qbtrade.OrderExecutor, session *qbtrade.ExchangeSession) error {

	s.ExchangeSession = session
	s.markets = s.ExchangeSession.Markets()
	s.Positions = make(map[string]*types.Position)
	s.ProfitStats = make(map[string]*types.ProfitStats)
	s.orderExecutors = make(map[string]*qbtrade.GeneralOrderExecutor)

	s.Traded = make(map[string]bool)
	s.TradeType = make(map[string]string)
	s.TradeRetry = make(map[string]int)

	s.ShortOrder = make(map[string]types.SubmitOrder)
	s.ShortProfitOrder = make(map[string]types.SubmitOrder)
	s.ShortLossOrder = make(map[string]types.SubmitOrder)
	s.LongOrder = make(map[string]types.SubmitOrder)
	s.LongProfitOrder = make(map[string]types.SubmitOrder)
	s.LongLossOrder = make(map[string]types.SubmitOrder)

	s.OpenTrade = make(map[string][]types.Trade)
	s.EndTrade = make(map[string][]types.Trade)
	s.OpenQuantity = make(map[string]fixedpoint.Value)
	s.EndQuantity = make(map[string]fixedpoint.Value)

	s.nr = make(map[string]*indicatorv2.NRStrean)
	s.cci = make(map[string]*indicatorv2.CCIStream)

	s.TotalProfit = fixedpoint.Value(0)
	s.TotalFree = fixedpoint.Value(0)
	s.TotalOrderCount = 0

	for _, symbol := range s.Symbols {
		s.Positions[symbol] = types.NewPositionFromMarket(s.markets[symbol])
		s.ProfitStats[symbol] = types.NewProfitStats(s.markets[symbol])

		s.orderExecutors[symbol] = qbtrade.NewGeneralOrderExecutor(session, symbol, ID, s.InstanceID(), s.Positions[symbol])
		s.orderExecutors[symbol].BindEnvironment(s.Environment)
		_ = s.orderExecutors[symbol].GracefulCancel(ctx)
		//s.orderExecutors[symbol].BindProfitStats(s.ProfitStats[symbol])
		//s.orderExecutors[symbol].TradeCollector().OnPositionUpdate(func(position *types.Position) {
		//	qbtrade.Sync(ctx, s)
		//})
		s.orderExecutors[symbol].Bind()

		// 初始化
		s.Traded[symbol] = false
		s.TradeType[symbol] = ""
	}

	qbtrade.Notify("CCINR策略开始执行...")

	for _, symbol := range s.Symbols {
		s.nr[symbol] = session.Indicators(symbol).NR(s.Interval, s.NrCount, s.StrictMode)
		s.cci[symbol] = session.Indicators(symbol).CCI(s.Interval, s.CCIWindow)
		s.TradeRetry[symbol] = 0
	}

	session.MarketDataStream.OnKLineClosed(func(k types.KLine) {
		for _, symbol := range s.Symbols {
			if k.Symbol != symbol || k.Interval != s.Interval {
				continue
			}

			if !s.Traded[symbol] {
				// 如若在下一根k线未成交 则取消订单
				if s.TradeType[symbol] != "" && s.TradeRetry[symbol] > 1 {
					qbtrade.Notify(fmt.Sprintf("交易信号未成交，取消订单: %s", symbol))
					s.cancelOrders(ctx, symbol)
				}

				if s.TradeType[symbol] != "" && s.TradeRetry[symbol] <= 1 {
					s.TradeRetry[symbol] = s.TradeRetry[symbol] + 1
				}
			}
		}
	})

	for _, symbol := range s.Symbols {
		sym := symbol
		s.nr[sym].OnUpdate(func(v float64) {
			if s.Traded[sym] {
				return
			}
			cciV := s.cci[sym].Last(0)
			if cciV <= s.LongCCI.Float64() {
				s.TradeType[sym] = "long"
			} else if cciV >= s.ShortCCI.Float64() {
				s.TradeType[sym] = "short"
			} else {
				return
			}
			msg := fmt.Sprintf("交易信号：币种：%s, 方向 %s, 时间: %s, 最高价：%f，最低价:%f, CCI: %v",
				sym, s.TradeType[sym], s.nr[sym].NrKLine.GetStartTime(), s.nr[sym].NrKLine.High.Float64(),
				s.nr[sym].NrKLine.Low.Float64(), cciV)
			qbtrade.Notify(msg)
			tk := s.nr[sym].NrKLine
			s.placeOrders(ctx, tk)
		})
	}

	//
	//session.MarketDataStream.OnMarketTrade(func(trade types.Trade) {
	//	// handle market trade event here
	//	fmt.Println(trade)
	//})

	b, ok := s.getBalance(ctx)
	fmt.Println(b, ok)
	session.UserDataStream.OnOrderUpdate(func(order types.Order) {
		orderSymbol := order.Symbol
		if order.Status == types.OrderStatusFilled {
			if order.Type == types.OrderTypeLimit && order.Side == types.SideTypeBuy {
				log.Infof("the long order is filled: %+v,id is %d, symbol is %s, type is %s, status is %s",
					order, order.OrderID, orderSymbol, order.Type, order.Status)
				s.Traded[orderSymbol] = true
				s.TradeRetry[orderSymbol] = 0
				qbtrade.Notify("订单成交通知：\n 币种：%s, 方向：%s, 价格：%s, 数量：%s", order.Symbol, s.TradeType[orderSymbol],
					order.Price, order.Quantity)
			}
			if order.Type == types.OrderTypeLimit && order.Side == types.SideTypeSell {
				log.Infof("the short order is filled: %+v,id is %d, symbol is %s, type is %s, status is %s",
					order, order.OrderID, orderSymbol, order.Type, order.Status)
				s.Traded[orderSymbol] = true
				s.TradeRetry[orderSymbol] = 0
				qbtrade.Notify("订单成交通知：\n 币种：%s, 方向：%s, 价格：%s, 数量：%s", order.Symbol, s.TradeType[orderSymbol],
					order.Price, order.Quantity)
			}

			if order.Type == types.OrderTypeMarket {
				log.Infof("the loss or profit order is filled: %+v,id is %d, symbol is %s, type is %s, "+
					"status is %s", order, order.OrderID, orderSymbol, order.Type, order.Status)
				qbtrade.Notify("订单止盈或止损通知：\n %s", order.Symbol)
				s.Traded[orderSymbol] = false
				s.TradeRetry[orderSymbol] = 0
				s.TradeType[orderSymbol] = ""
			} else {
				log.Infof("the order is: %+v,id is %d, symbol is %s, type is %s, status is %s",
					order, order.OrderID, orderSymbol, order.Type, order.Status)
			}
		}
	})

	session.UserDataStream.OnTradeUpdate(func(trade types.Trade) {
		symbol := trade.Symbol

		if (trade.Side == types.SideTypeBuy && s.TradeType[symbol] == "long") || (trade.Side == types.SideTypeSell && s.TradeType[symbol] == "short") {
			s.OpenTrade[symbol] = append(s.OpenTrade[symbol], trade)
			s.OpenQuantity[symbol] += trade.Quantity
		}
		if (trade.Side == types.SideTypeSell && s.TradeType[symbol] == "long") || (trade.Side == types.SideTypeBuy && s.TradeType[symbol] == "short") {
			s.EndTrade[symbol] = append(s.EndTrade[symbol], trade)
			s.EndQuantity[symbol] += trade.Quantity
			s.notifyProfit(ctx, symbol)
		}
		log.Infof("trade: symbol %s, side %s, price %f, fee %f, quantity %f, buyer %v, maker %v",
			symbol, trade.Side, trade.Price.Float64(), trade.Fee.Float64(), trade.Quantity.Float64(),
			trade.IsBuyer, trade.IsMaker)
	})

	s.OnSuspend(func() {
		// Cancel active orders
		for _, symbol := range s.Symbols {
			_ = s.orderExecutors[symbol].GracefulCancel(ctx)
		}
	})

	s.OnEmergencyStop(func() {
		// Cancel active orders
		for _, symbol := range s.Symbols {
			_ = s.orderExecutors[symbol].GracefulCancel(ctx)
		}
		// Close 100% position
		//_ = s.ClosePosition(ctx, fixedpoint.One)
	})

	qbtrade.OnShutdown(ctx, func(ctx context.Context, wg *sync.WaitGroup) {
		defer wg.Done()

		if err := s.Strategy.OrderExecutor.GracefulCancel(ctx); err != nil {
			log.WithError(err).Error("unable to cancel open orders...")
		}

		qbtrade.Sync(ctx, s)
	})

	return nil
}

func (s *Strategy) handleBalanceUpdate(balances types.BalanceMap) {
	for _, b := range balances {
		if b.Available.IsZero() && b.Borrowed.IsZero() {
			continue
		}
	}
}

func (s *Strategy) handleBinanceBalanceUpdateEvent(event *binance.BalanceUpdateEvent) {
	account := s.ExchangeSession.GetAccount()

	fmt.Println(account)
	delta := event.Delta

	// ignore outflow
	if delta.Sign() < 0 {
		return
	}
}

// getBalance 获取账户余额
func (s *Strategy) getBalance(ctx context.Context) (balance types.Balance, ok bool) {
	// 更新并获取account信息
	account, err := s.ExchangeSession.UpdateAccount(ctx)
	if err != nil {
		log.WithError(err).Error("unable to update account")
		return
	}

	// 获取balance信息
	return account.Balance("USDT")
}