diff --git a/pkg/bbgo/environment.go b/pkg/bbgo/environment.go
index c1f83f7aa..60fe162bc 100644
--- a/pkg/bbgo/environment.go
+++ b/pkg/bbgo/environment.go
@@ -490,8 +490,6 @@ func (environ *Environment) Connect(ctx context.Context) error {
 		if err := session.Stream.Connect(ctx); err != nil {
 			return err
 		}
-
-		session.Stream.EmitStart()
 	}
 
 	return nil
diff --git a/pkg/exchange/binance/stream.go b/pkg/exchange/binance/stream.go
index 1d32ce6e8..b7bb15b5a 100644
--- a/pkg/exchange/binance/stream.go
+++ b/pkg/exchange/binance/stream.go
@@ -310,6 +310,8 @@ func (s *Stream) Connect(ctx context.Context) error {
 	}
 
 	go s.read(ctx)
+
+	s.EmitStart()
 	return nil
 }
 
diff --git a/pkg/exchange/max/stream.go b/pkg/exchange/max/stream.go
index c179f37c7..212c71c29 100644
--- a/pkg/exchange/max/stream.go
+++ b/pkg/exchange/max/stream.go
@@ -171,7 +171,13 @@ func (s *Stream) Subscribe(channel types.Channel, symbol string, options types.S
 }
 
 func (s *Stream) Connect(ctx context.Context) error {
-	return s.websocketService.Connect(ctx)
+	err := s.websocketService.Connect(ctx)
+	if err != nil {
+		return err
+	}
+
+	s.EmitStart()
+	return nil
 }
 
 func (s *Stream) Close() error {