bbgo_origin/pkg/bbgo/trader.go

465 lines
13 KiB
Go
Raw Permalink Normal View History

2020-07-10 13:34:39 +00:00
package bbgo
import (
"context"
2020-12-07 03:43:17 +00:00
"fmt"
"reflect"
2022-07-19 09:13:35 +00:00
"sync"
2020-07-13 04:57:18 +00:00
2021-02-16 07:58:21 +00:00
"github.com/pkg/errors"
2020-07-13 04:57:18 +00:00
log "github.com/sirupsen/logrus"
_ "github.com/go-sql-driver/mysql"
2022-01-14 18:52:46 +00:00
"github.com/c9s/bbgo/pkg/dynamic"
2022-01-14 18:52:46 +00:00
"github.com/c9s/bbgo/pkg/interact"
2020-07-10 13:34:39 +00:00
)
2022-07-19 09:13:35 +00:00
// Strategy method calls:
// -> Defaults() (optional method)
// -> Initialize() (optional method)
// -> Validate() (optional method)
// -> Run() (optional method)
// -> Shutdown(shutdownCtx context.Context, wg *sync.WaitGroup)
2022-05-10 05:25:03 +00:00
type StrategyID interface {
ID() string
}
// SingleExchangeStrategy represents the single Exchange strategy
type SingleExchangeStrategy interface {
2022-05-10 05:25:03 +00:00
StrategyID
Run(ctx context.Context, orderExecutor OrderExecutor, session *ExchangeSession) error
2020-07-14 06:54:23 +00:00
}
2022-07-19 08:59:56 +00:00
// StrategyInitializer's Initialize method is called before the Subscribe method call.
type StrategyInitializer interface {
Initialize() error
}
2022-07-19 08:59:56 +00:00
type StrategyDefaulter interface {
Defaults() error
}
2022-07-19 09:13:35 +00:00
type StrategyValidator interface {
Validate() error
}
type StrategyShutdown interface {
Shutdown(ctx context.Context, wg *sync.WaitGroup)
}
// ExchangeSessionSubscriber provides an interface for collecting subscriptions from different strategies
// Subscribe method will be called before the user data stream connection is created.
type ExchangeSessionSubscriber interface {
Subscribe(session *ExchangeSession)
}
type CrossExchangeSessionSubscriber interface {
CrossSubscribe(sessions map[string]*ExchangeSession)
}
type CrossExchangeStrategy interface {
2022-05-10 05:25:03 +00:00
StrategyID
CrossRun(ctx context.Context, orderExecutionRouter OrderExecutionRouter, sessions map[string]*ExchangeSession) error
}
2020-11-09 08:34:35 +00:00
type Logging interface {
EnableLogging()
DisableLogging()
}
type Logger interface {
Warnf(message string, args ...interface{})
Errorf(message string, args ...interface{})
Infof(message string, args ...interface{})
}
type SilentLogger struct{}
2021-05-02 15:47:57 +00:00
func (logger *SilentLogger) Infof(string, ...interface{}) {}
func (logger *SilentLogger) Warnf(string, ...interface{}) {}
func (logger *SilentLogger) Errorf(string, ...interface{}) {}
2020-11-09 08:34:35 +00:00
type Trader struct {
environment *Environment
2020-09-28 07:01:10 +00:00
riskControls *RiskControls
crossExchangeStrategies []CrossExchangeStrategy
exchangeStrategies map[string][]SingleExchangeStrategy
2020-11-09 08:34:35 +00:00
// gracefulShutdown is used for registering strategy's Shutdown calls
// when strategy implements Shutdown(ctx), the func ref will be stored in the callback.
2022-10-02 11:34:52 +00:00
gracefulShutdown GracefulShutdown
2020-11-09 08:34:35 +00:00
logger Logger
2020-07-13 05:25:48 +00:00
}
func NewTrader(environ *Environment) *Trader {
2020-09-07 06:20:03 +00:00
return &Trader{
environment: environ,
exchangeStrategies: make(map[string][]SingleExchangeStrategy),
2020-11-09 08:34:35 +00:00
logger: log.StandardLogger(),
2020-09-07 06:20:03 +00:00
}
}
2020-11-09 08:34:35 +00:00
func (trader *Trader) EnableLogging() {
trader.logger = log.StandardLogger()
}
func (trader *Trader) DisableLogging() {
trader.logger = &SilentLogger{}
}
func (trader *Trader) Configure(userConfig *Config) error {
if userConfig.RiskControls != nil {
trader.SetRiskControls(userConfig.RiskControls)
}
for _, entry := range userConfig.ExchangeStrategies {
for _, mount := range entry.Mounts {
log.Infof("attaching strategy %T on %s...", entry.Strategy, mount)
2021-03-15 10:51:29 +00:00
if err := trader.AttachStrategyOn(mount, entry.Strategy); err != nil {
return err
}
}
}
for _, strategy := range userConfig.CrossExchangeStrategies {
log.Infof("attaching cross exchange strategy %T", strategy)
trader.AttachCrossExchangeStrategy(strategy)
}
return nil
}
// AttachStrategyOn attaches the single exchange strategy on an exchange Session.
// Single exchange strategy is the default behavior.
2021-01-09 11:47:21 +00:00
func (trader *Trader) AttachStrategyOn(session string, strategies ...SingleExchangeStrategy) error {
if len(trader.environment.sessions) == 0 {
return fmt.Errorf("you don't have any session configured, please check your environment variable or config file")
}
if _, ok := trader.environment.sessions[session]; !ok {
2021-01-09 11:47:21 +00:00
var keys []string
for k := range trader.environment.sessions {
keys = append(keys, k)
}
return fmt.Errorf("session %s is not defined, valid sessions are: %v", session, keys)
}
2020-09-19 03:25:48 +00:00
2022-06-17 11:19:51 +00:00
trader.exchangeStrategies[session] = append(
trader.exchangeStrategies[session], strategies...)
2020-10-16 05:52:18 +00:00
2021-01-09 11:47:21 +00:00
return nil
}
2020-09-28 07:01:10 +00:00
// AttachCrossExchangeStrategy attaches the cross exchange strategy
2020-10-16 05:52:18 +00:00
func (trader *Trader) AttachCrossExchangeStrategy(strategy CrossExchangeStrategy) *Trader {
trader.crossExchangeStrategies = append(trader.crossExchangeStrategies, strategy)
2020-10-16 05:52:18 +00:00
return trader
}
2020-09-28 07:01:10 +00:00
2021-05-02 15:48:53 +00:00
// SetRiskControls sets the risk controller
// TODO: provide a more DSL way to configure risk controls
func (trader *Trader) SetRiskControls(riskControls *RiskControls) {
trader.riskControls = riskControls
}
2021-02-16 08:12:00 +00:00
func (trader *Trader) RunSingleExchangeStrategy(ctx context.Context, strategy SingleExchangeStrategy, session *ExchangeSession, orderExecutor OrderExecutor) error {
2022-07-19 09:13:35 +00:00
if v, ok := strategy.(StrategyValidator); ok {
if err := v.Validate(); err != nil {
return fmt.Errorf("failed to validate the config: %w", err)
}
}
2022-07-19 09:13:35 +00:00
if shutdown, ok := strategy.(StrategyShutdown); ok {
trader.gracefulShutdown.OnShutdown(shutdown.Shutdown)
2022-07-19 09:13:35 +00:00
}
2021-02-16 08:12:00 +00:00
return strategy.Run(ctx, orderExecutor, session)
}
2021-02-16 08:12:00 +00:00
func (trader *Trader) getSessionOrderExecutor(sessionName string) OrderExecutor {
var session = trader.environment.sessions[sessionName]
2021-02-16 08:12:00 +00:00
// default to base order executor
var orderExecutor OrderExecutor = session.OrderExecutor
2021-02-16 08:12:00 +00:00
// Since the risk controls are loaded from the config file
if trader.riskControls != nil && trader.riskControls.SessionBasedRiskControl != nil {
if control, ok := trader.riskControls.SessionBasedRiskControl[sessionName]; ok {
control.SetBaseOrderExecutor(session.OrderExecutor)
2021-02-16 08:12:00 +00:00
// pick the wrapped order executor
if control.OrderExecutor != nil {
return control.OrderExecutor
}
2021-02-16 08:12:00 +00:00
}
}
return orderExecutor
}
2021-02-16 08:12:00 +00:00
func (trader *Trader) RunAllSingleExchangeStrategy(ctx context.Context) error {
// load and run Session strategies
for sessionName, strategies := range trader.exchangeStrategies {
var session = trader.environment.sessions[sessionName]
var orderExecutor = trader.getSessionOrderExecutor(sessionName)
for _, strategy := range strategies {
if err := trader.RunSingleExchangeStrategy(ctx, strategy, session, orderExecutor); err != nil {
2020-09-28 07:01:10 +00:00
return err
}
}
}
2020-09-28 07:01:10 +00:00
2021-02-16 08:12:00 +00:00
return nil
}
func (trader *Trader) injectFieldsAndSubscribe(ctx context.Context) error {
// load and run Session strategies
for sessionName, strategies := range trader.exchangeStrategies {
var session = trader.environment.sessions[sessionName]
var orderExecutor = trader.getSessionOrderExecutor(sessionName)
for _, strategy := range strategies {
rs := reflect.ValueOf(strategy)
// get the struct element
rs = rs.Elem()
if rs.Kind() != reflect.Struct {
return errors.New("strategy object is not a struct")
}
if err := trader.injectCommonServices(strategy); err != nil {
return err
}
if err := dynamic.InjectField(rs, "OrderExecutor", orderExecutor, false); err != nil {
return errors.Wrapf(err, "failed to inject OrderExecutor on %T", strategy)
}
if defaulter, ok := strategy.(StrategyDefaulter); ok {
if err := defaulter.Defaults(); err != nil {
panic(err)
}
}
if initializer, ok := strategy.(StrategyInitializer); ok {
if err := initializer.Initialize(); err != nil {
panic(err)
}
}
if subscriber, ok := strategy.(ExchangeSessionSubscriber); ok {
subscriber.Subscribe(session)
} else {
log.Errorf("strategy %s does not implement ExchangeSessionSubscriber", strategy.ID())
}
if symbol, ok := dynamic.LookupSymbolField(rs); ok {
log.Infof("found symbol %s based strategy from %s", symbol, rs.Type())
if err := session.initSymbol(ctx, trader.environment, symbol); err != nil {
return errors.Wrapf(err, "failed to inject object into %T when initSymbol", strategy)
}
market, ok := session.Market(symbol)
if !ok {
return fmt.Errorf("market of symbol %s not found", symbol)
}
indicatorSet := session.StandardIndicatorSet(symbol)
2022-07-21 04:33:29 +00:00
if !ok {
return fmt.Errorf("standardIndicatorSet of symbol %s not found", symbol)
}
store, ok := session.MarketDataStore(symbol)
if !ok {
return fmt.Errorf("marketDataStore of symbol %s not found", symbol)
}
if err := dynamic.ParseStructAndInject(strategy,
market,
session,
session.OrderExecutor,
2022-07-21 04:33:29 +00:00
indicatorSet,
store,
); err != nil {
return errors.Wrapf(err, "failed to inject object into %T", strategy)
}
}
}
}
for _, strategy := range trader.crossExchangeStrategies {
rs := reflect.ValueOf(strategy)
// get the struct element from the struct pointer
rs = rs.Elem()
if rs.Kind() != reflect.Struct {
continue
}
if err := trader.injectCommonServices(strategy); err != nil {
return err
}
if defaulter, ok := strategy.(StrategyDefaulter); ok {
if err := defaulter.Defaults(); err != nil {
return err
}
}
if initializer, ok := strategy.(StrategyInitializer); ok {
if err := initializer.Initialize(); err != nil {
return err
}
}
if subscriber, ok := strategy.(CrossExchangeSessionSubscriber); ok {
subscriber.CrossSubscribe(trader.environment.sessions)
} else {
log.Errorf("strategy %s does not implement CrossExchangeSessionSubscriber", strategy.ID())
}
}
return nil
}
2021-02-16 08:12:00 +00:00
func (trader *Trader) Run(ctx context.Context) error {
2022-01-14 18:52:46 +00:00
// before we start the interaction,
// register the core interaction, because we can only get the strategies in this scope
// trader.environment.Connect will call interact.Start
interact.AddCustomInteraction(NewCoreInteraction(trader.environment, trader))
if err := trader.injectFieldsAndSubscribe(ctx); err != nil {
return err
}
2021-05-07 16:45:24 +00:00
if err := trader.environment.Start(ctx); err != nil {
2021-02-16 08:12:00 +00:00
return err
}
if err := trader.RunAllSingleExchangeStrategy(ctx); err != nil {
return err
}
router := &ExchangeOrderExecutionRouter{
sessions: trader.environment.sessions,
executors: make(map[string]OrderExecutor),
}
for sessionID := range trader.environment.sessions {
var orderExecutor = trader.getSessionOrderExecutor(sessionID)
router.executors[sessionID] = orderExecutor
}
for _, strategy := range trader.crossExchangeStrategies {
if err := strategy.CrossRun(ctx, router, trader.environment.sessions); err != nil {
return err
}
}
return trader.environment.Connect(ctx)
}
func (trader *Trader) LoadState() error {
if trader.environment.BacktestService != nil {
return nil
}
if persistenceServiceFacade == nil {
return nil
}
ps := persistenceServiceFacade.Get()
log.Infof("loading strategies states...")
2022-05-10 05:25:03 +00:00
return trader.IterateStrategies(func(strategy StrategyID) error {
2022-08-22 18:12:26 +00:00
id := dynamic.CallID(strategy)
return loadPersistenceFields(strategy, id, ps)
2022-05-10 05:25:03 +00:00
})
}
func (trader *Trader) IterateStrategies(f func(st StrategyID) error) error {
for _, strategies := range trader.exchangeStrategies {
for _, strategy := range strategies {
2022-05-10 05:25:03 +00:00
if err := f(strategy); err != nil {
return err
}
}
}
2022-05-10 05:25:03 +00:00
for _, strategy := range trader.crossExchangeStrategies {
2022-05-10 05:25:03 +00:00
if err := f(strategy); err != nil {
return err
}
}
return nil
}
// NOTICE: the ctx here is the trading context, which could already be canceled.
func (trader *Trader) SaveState(ctx context.Context) error {
if trader.environment.BacktestService != nil {
return nil
}
isolation := GetIsolationFromContext(ctx)
ps := isolation.persistenceServiceFacade.Get()
log.Debugf("saving strategy persistence states...")
2022-05-10 05:25:03 +00:00
return trader.IterateStrategies(func(strategy StrategyID) error {
2022-08-22 18:12:26 +00:00
id := dynamic.CallID(strategy)
if len(id) == 0 {
2022-05-10 05:25:03 +00:00
return nil
2020-07-13 05:25:48 +00:00
}
2022-05-10 05:25:03 +00:00
return storePersistenceFields(strategy, id, ps)
})
2020-07-10 13:34:39 +00:00
}
func (trader *Trader) Shutdown(ctx context.Context) {
trader.gracefulShutdown.Shutdown(ctx)
}
2022-03-06 06:06:19 +00:00
func (trader *Trader) injectCommonServices(s interface{}) error {
2022-03-14 13:21:04 +00:00
// a special injection for persistence selector:
// if user defined the selector, the facade pointer will be nil, hence we need to update the persistence facade pointer
sv := reflect.ValueOf(s).Elem()
if field, ok := dynamic.HasField(sv, "Persistence"); ok {
2022-03-14 13:21:04 +00:00
// the selector is set, but we need to update the facade pointer
if !field.IsNil() {
elem := field.Elem()
if elem.Kind() != reflect.Struct {
return fmt.Errorf("field Persistence is not a struct element, %s given", field)
}
if err := dynamic.InjectField(elem, "Facade", persistenceServiceFacade, true); err != nil {
2022-03-14 13:21:04 +00:00
return err
}
/*
if err := ParseStructAndInject(field.Interface(), persistenceFacade); err != nil {
2022-03-14 13:21:04 +00:00
return err
}
*/
}
}
return dynamic.ParseStructAndInject(s,
2022-03-05 04:39:39 +00:00
&trader.logger,
Notification,
2022-03-05 04:39:39 +00:00
trader.environment.TradeService,
2022-03-05 05:37:27 +00:00
trader.environment.OrderService,
trader.environment.DatabaseService,
2022-03-05 04:39:39 +00:00
trader.environment.AccountService,
2022-03-06 10:17:42 +00:00
trader.environment,
persistenceServiceFacade, // if the strategy use persistence facade separately
2022-03-05 04:39:39 +00:00
)
2021-02-16 07:58:21 +00:00
}