bbgo_origin/pkg/risk/circuitbreaker/basic.go

package circuitbreaker

import (
	"fmt"
	"sync"
	"time"

	"github.com/prometheus/client_golang/prometheus"

	"github.com/c9s/bbgo/pkg/fixedpoint"
	"github.com/c9s/bbgo/pkg/types"

	log "github.com/sirupsen/logrus"
)

var metricsLabels = []string{"strategy", "strategyInstance", "symbol"}

var consecutiveTotalLossMetrics = prometheus.NewGaugeVec(
	prometheus.GaugeOpts{
		Name: "bbgo_circuit_breaker_consecutive_total_loss",
		Help: "",
	}, metricsLabels)

var consecutiveLossTimesCounterMetrics = prometheus.NewGaugeVec(
	prometheus.GaugeOpts{
		Name: "bbgo_circuit_breaker_consecutive_loss_times",
		Help: "",
	}, metricsLabels)

var haltCounterMetrics = prometheus.NewGaugeVec(
	prometheus.GaugeOpts{
		Name: "bbgo_circuit_breaker_halt_counter",
		Help: "",
	}, metricsLabels)

var haltMetrics = prometheus.NewGaugeVec(
	prometheus.GaugeOpts{
		Name: "bbgo_circuit_breaker_halt",
		Help: "",
	}, metricsLabels)

var totalProfitMetrics = prometheus.NewGaugeVec(
	prometheus.GaugeOpts{
		Name: "bbgo_circuit_breaker_total_profit",
		Help: "",
	}, metricsLabels)

var profitWinCounterMetrics = prometheus.NewCounterVec(
	prometheus.CounterOpts{
		Name: "bbgo_circuit_breaker_profit_win_counter",
		Help: "profit winning counter",
	}, metricsLabels)

var profitLossCounterMetrics = prometheus.NewCounterVec(
	prometheus.CounterOpts{
		Name: "bbgo_circuit_breaker_profit_loss_counter",
		Help: "profit loss counter",
	}, metricsLabels)

var winningRatioMetrics = prometheus.NewGaugeVec(
	prometheus.GaugeOpts{
		Name: "bbgo_circuit_breaker_winning_ratio",
		Help: "winning ratio",
	}, metricsLabels)

func init() {
	prometheus.MustRegister(
		consecutiveTotalLossMetrics,
		consecutiveLossTimesCounterMetrics,
		haltCounterMetrics,
		totalProfitMetrics,
		haltMetrics,
		profitWinCounterMetrics,
		profitLossCounterMetrics,
		winningRatioMetrics,
	)
}

//go:generate callbackgen -type BasicCircuitBreaker
type BasicCircuitBreaker struct {
	Enabled bool `json:"enabled"`

	MaximumConsecutiveTotalLoss fixedpoint.Value `json:"maximumConsecutiveTotalLoss"`

	MaximumConsecutiveLossTimes int `json:"maximumConsecutiveLossTimes"`

	IgnoreDustLoss               bool             `json:"ignoreConsecutiveDustLoss"`
	ConsecutiveDustLossThreshold fixedpoint.Value `json:"consecutiveDustLossThreshold"`

	MaximumLossPerRound fixedpoint.Value `json:"maximumLossPerRound"`

	MaximumTotalLoss fixedpoint.Value `json:"maximumTotalLoss"`

	MaximumHaltTimes int `json:"maximumHaltTimes"`

	MaximumHaltTimesExceededPanic bool `json:"maximumHaltTimesExceededPanic"`

	HaltDuration types.Duration `json:"haltDuration"`

	strategyID, strategyInstance, symbol string

	panicCallbacks []func()

	haltCounter int
	haltReason  string
	halted      bool

	haltedAt, haltTo time.Time

	// totalProfit is the total PnL, could be negative or positive
	totalProfit          fixedpoint.Value
	consecutiveLossTimes int

	winTimes, lossTimes int
	winRatio            float64

	// consecutiveLoss is a negative number, presents the consecutive loss
	consecutiveLoss fixedpoint.Value

	mu sync.Mutex

	metricsLabels prometheus.Labels
}

func NewBasicCircuitBreaker(strategyID, strategyInstance, symbol string) *BasicCircuitBreaker {
	b := &BasicCircuitBreaker{
		Enabled:                       true,
		MaximumConsecutiveLossTimes:   8,
		MaximumHaltTimes:              3,
		MaximumHaltTimesExceededPanic: false,

		HaltDuration:     types.Duration(1 * time.Hour),
		strategyID:       strategyID,
		strategyInstance: strategyInstance,
		symbol:           symbol,
		metricsLabels: prometheus.Labels{
			"strategy":         strategyID,
			"strategyInstance": strategyInstance,
			"symbol":           symbol,
		},
	}

	b.updateMetrics()
	return b
}

func (b *BasicCircuitBreaker) SetMetricsInfo(strategyID, strategyInstance, symbol string) {
	b.strategyID = strategyID
	b.strategyInstance = strategyInstance
	b.symbol = symbol
	b.metricsLabels = prometheus.Labels{
		"strategy":         b.strategyID,
		"strategyInstance": b.strategyInstance,
		"symbol":           b.symbol,
	}
}

func (b *BasicCircuitBreaker) getMetricsLabels() prometheus.Labels {
	if b.metricsLabels != nil {
		return b.metricsLabels
	}

	return prometheus.Labels{
		"strategy":         b.strategyID,
		"strategyInstance": b.strategyInstance,
		"symbol":           b.symbol,
	}
}

func (b *BasicCircuitBreaker) updateMetrics() {
	labels := b.getMetricsLabels()
	consecutiveTotalLossMetrics.With(labels).Set(b.consecutiveLoss.Float64())
	consecutiveLossTimesCounterMetrics.With(labels).Set(float64(b.consecutiveLossTimes))
	totalProfitMetrics.With(labels).Set(b.totalProfit.Float64())

	if b.halted {
		haltMetrics.With(labels).Set(1.0)
	} else {
		haltMetrics.With(labels).Set(0.0)
	}

	haltCounterMetrics.With(labels).Set(float64(b.haltCounter))
	winningRatioMetrics.With(labels).Set(b.winRatio)
}

func (b *BasicCircuitBreaker) RecordProfit(profit fixedpoint.Value, now time.Time) {
	b.mu.Lock()
	defer b.mu.Unlock()

	b.totalProfit = b.totalProfit.Add(profit)

	if profit.Sign() < 0 {
		if b.IgnoreDustLoss && profit.Abs().Compare(b.ConsecutiveDustLossThreshold) <= 0 {
			// ignore dust loss
			log.Infof("ignore dust loss (threshold %f): %f", b.ConsecutiveDustLossThreshold.Float64(), profit.Float64())
		} else {
			b.lossTimes++
			b.consecutiveLossTimes++
			b.consecutiveLoss = b.consecutiveLoss.Add(profit)
			profitLossCounterMetrics.With(b.getMetricsLabels()).Inc()
		}
	} else {
		b.winTimes++
		b.consecutiveLossTimes = 0
		b.consecutiveLoss = fixedpoint.Zero
		profitWinCounterMetrics.With(b.getMetricsLabels()).Inc()
	}

	if b.lossTimes == 0 {
		b.winRatio = 999.0
	} else {
		b.winRatio = float64(b.winTimes) / float64(b.lossTimes)
	}

	defer b.updateMetrics()

	if b.MaximumConsecutiveLossTimes > 0 && b.consecutiveLossTimes >= b.MaximumConsecutiveLossTimes {
		b.halt(now, "exceeded MaximumConsecutiveLossTimes")
		return
	}

	if b.MaximumConsecutiveTotalLoss.Sign() > 0 && b.consecutiveLoss.Neg().Compare(b.MaximumConsecutiveTotalLoss) >= 0 {
		b.halt(now, "exceeded MaximumConsecutiveTotalLoss")
		return
	}

	if b.MaximumLossPerRound.Sign() > 0 && profit.Sign() < 0 && profit.Neg().Compare(b.MaximumLossPerRound) >= 0 {
		b.halt(now, "exceeded MaximumLossPerRound")
		return
	}

	// - (-120 [profit]) > 100 [maximum total loss]
	if b.MaximumTotalLoss.Sign() > 0 && b.totalProfit.Neg().Compare(b.MaximumTotalLoss) >= 0 {
		b.halt(now, "exceeded MaximumTotalLoss")
		return
	}
}

func (b *BasicCircuitBreaker) Reset() {
	b.mu.Lock()
	b.reset()
	b.mu.Unlock()
}

func (b *BasicCircuitBreaker) reset() {
	b.halted = false
	b.haltedAt = time.Time{}
	b.haltTo = time.Time{}

	b.totalProfit = fixedpoint.Zero
	b.consecutiveLossTimes = 0
	b.consecutiveLoss = fixedpoint.Zero
	b.updateMetrics()
}

func (b *BasicCircuitBreaker) IsHalted(now time.Time) (string, bool) {
	if !b.Enabled {
		return "disabled", false
	}

	b.mu.Lock()
	defer b.mu.Unlock()

	if !b.halted {
		return "", false
	}

	// check if it's an expired halt
	if now.After(b.haltTo) {
		b.reset()
		return "", false
	}

	return b.haltReason, true
}

func (b *BasicCircuitBreaker) halt(now time.Time, reason string) {
	b.halted = true
	b.haltCounter++
	b.haltReason = reason
	b.haltedAt = now
	b.haltTo = now.Add(b.HaltDuration.Duration())

	labels := b.getMetricsLabels()
	haltCounterMetrics.With(labels).Set(float64(b.haltCounter))
	haltMetrics.With(labels).Set(1.0)

	defer b.updateMetrics()

	if b.MaximumHaltTimesExceededPanic && b.haltCounter > b.MaximumHaltTimes {
		b.EmitPanic()

		panic(fmt.Errorf("total %d halt times > maximumHaltTimesExceededPanic %d", b.haltCounter, b.MaximumHaltTimes))
	}
}