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/qbtrade"
	"git.qtrade.icu/lychiyu/qbtrade/pkg/strategy/common"
	"git.qtrade.icu/lychiyu/qbtrade/pkg/types"
	log "github.com/sirupsen/logrus"
	"sync"
)

const ID = "ccinr"

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

type Strategy struct {
	*common.Strategy

	Symbol   string         `json:"symbol"`
	Interval types.Interval `json:"interval"`

	ExchangeSession *qbtrade.ExchangeSession
}

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

func (s *Strategy) Subscribe(session *qbtrade.ExchangeSession) {
	session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{Interval: s.Interval})
	if !qbtrade.IsBackTesting {
		session.Subscribe(types.MarketTradeChannel, s.Symbol, types.SubscribeOptions{})
	}
}

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

	return nil
}

func (s *Strategy) Run(ctx context.Context, orderExecutor qbtrade.OrderExecutor, session *qbtrade.ExchangeSession) error {
	s.ExchangeSession = session

	nr := session.Indicators(s.Symbol).NR(s.Interval, 4, true)
	nr.OnUpdate(func(v float64) {
		msg := fmt.Sprintf("交易信号：时间: %s, 最高价：%f，最低价:%f", nr.NrKLine.GetStartTime(), nr.NrKLine.High.Float64(), nr.NrKLine.Low.Float64())
		qbtrade.Notify(msg)
		fmt.Println(v)
	})

	//session.MarketDataStream.OnKLineClosed(func(k types.KLine) {
	//	if k.Symbol != s.Symbol || k.Interval != s.Interval {
	//		return
	//	}
	//	fmt.Println(k)
	//})
	//
	//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) {
		if order.Status == types.OrderStatusFilled {
			log.Infof("your order is filled: %+v", order)
		}
	})

	session.UserDataStream.OnTradeUpdate(func(trade types.Trade) {
		log.Infof("trade price %f, fee %f %s", trade.Price.Float64(), trade.Fee.Float64(), trade.FeeCurrency)
	})

	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) {
	qbtrade.Notify(event)

	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")
}