src/environment/portfolio.py
def _step(self, action):
        """ Step the environment by a vector of actions

        Args:
            action: (num_models, num_stocks + 1)

        Returns:

        """
        assert action.ndim == 2, 'Action must be a two dimensional array with shape (num_models, num_stocks + 1)'
        assert action.shape[1] == len(self.sim[0].asset_names) + 1
        assert action.shape[0] == len(self.model_names)
        # normalise just in case
        action = np.clip(action, 0, 1)
        weights = action  # np.array([cash_bias] + list(action))  # [w0, w1...]
        weights /= (np.sum(weights, axis=1, keepdims=True) + eps)
        # so if weights are all zeros we normalise to [1,0...]
        weights[:, 0] += np.clip(1 - np.sum(weights, axis=1), 0, 1)
        assert ((action >= 0) * (action <= 1)).all(), 'all action values should be between 0 and 1. Not %s' % action
        np.testing.assert_almost_equal(np.sum(weights, axis=1), np.ones(shape=(weights.shape[0])), 3,
                                       err_msg='weights should sum to 1. action="%s"' % weights)
        observation, done1, ground_truth_obs = self.src._step()

        # concatenate observation with ones
        cash_observation = np.ones((1, self.window_length, observation.shape[2]))
        observation = np.concatenate((cash_observation, observation), axis=0)

        cash_ground_truth = np.ones((1, 1, ground_truth_obs.shape[2]))
        ground_truth_obs = np.concatenate((cash_ground_truth, ground_truth_obs), axis=0)

        # relative price vector of last observation day (close/open)
        close_price_vector = observation[:, -1, 3]
        open_price_vector = observation[:, -1, 0]
        y1 = close_price_vector / open_price_vector

        rewards = np.empty(shape=(weights.shape[0]))
        info = {}
        dones = np.empty(shape=(weights.shape[0]), dtype=bool)
        for i in range(weights.shape[0]):
            reward, current_info, done2 = self.sim[i]._step(weights[i], y1)
            rewards[i] = reward
            info[self.model_names[i]] = current_info['portfolio_value']
            info['return'] = current_info['return']
            dones[i] = done2

        # calculate return for buy and hold a bit of each asset
        info['market_value'] = np.cumprod([inf["return"] for inf in self.infos + [info]])[-1]
        # add dates
        info['date'] = index_to_date(self.start_idx + self.src.idx + self.src.step)
        info['steps'] = self.src.step
        info['next_obs'] = ground_truth_obs

        self.infos.append(info)

        return observation, rewards, np.all(dones) or done1, info
 Step the environment by a vector of actions

        Args:
            action: (num_models, num_stocks + 1)

        Returns:

        
Step the environment by a vector of actions
Step the environment by a vector of actions
def _step(self, action):
        assert action.ndim == 2, 'Action must be a two dimensional array with shape (num_models, num_stocks + 1)'
        assert action.shape[1] == len(self.sim[0].asset_names) + 1
        assert action.shape[0] == len(self.model_names)
        action = np.clip(action, 0, 1)
        weights = action  
        weights /= (np.sum(weights, axis=1, keepdims=True) + eps)
        weights[:, 0] += np.clip(1 - np.sum(weights, axis=1), 0, 1)
        assert ((action >= 0) * (action <= 1)).all(), 'all action values should be between 0 and 1. Not %s' % action
        np.testing.assert_almost_equal(np.sum(weights, axis=1), np.ones(shape=(weights.shape[0])), 3,
                                       err_msg='weights should sum to 1. action="%s"' % weights)
        observation, done1, ground_truth_obs = self.src._step()
        cash_observation = np.ones((1, self.window_length, observation.shape[2]))
        observation = np.concatenate((cash_observation, observation), axis=0)
        cash_ground_truth = np.ones((1, 1, ground_truth_obs.shape[2]))
        ground_truth_obs = np.concatenate((cash_ground_truth, ground_truth_obs), axis=0)
        close_price_vector = observation[:, -1, 3]
        open_price_vector = observation[:, -1, 0]
        y1 = close_price_vector / open_price_vector
        rewards = np.empty(shape=(weights.shape[0]))
        info = {}
        dones = np.empty(shape=(weights.shape[0]), dtype=bool)
        for i in range(weights.shape[0]):
            reward, current_info, done2 = self.sim[i]._step(weights[i], y1)
            rewards[i] = reward
            info[self.model_names[i]] = current_info['portfolio_value']
            info['return'] = current_info['return']
            dones[i] = done2
        info['market_value'] = np.cumprod([inf["return"] for inf in self.infos + [info]])[-1]
        info['date'] = index_to_date(self.start_idx + self.src.idx + self.src.step)
        info['steps'] = self.src.step
        info['next_obs'] = ground_truth_obs
        self.infos.append(info)
        return observation, rewards, np.all(dones) or done1, info
def _step ( self , action ) : assert action . ndim == 2 , 'Action must be a two dimensional array with shape (num_models, num_stocks + 1)' assert action . shape [ 1 ] == len ( self . sim [ 0 ] . asset_names ) + 1 assert action . shape [ 0 ] == len ( self . model_names ) action = np . clip ( action , 0 , 1 ) weights = action weights /= ( np . sum ( weights , axis = 1 , keepdims = True ) + eps ) weights [ : , 0 ] += np . clip ( 1 - np . sum ( weights , axis = 1 ) , 0 , 1 ) assert ( ( action >= 0 ) * ( action <= 1 ) ) . all ( ) , 'all action values should be between 0 and 1. Not %s' % action np . testing . assert_almost_equal ( np . sum ( weights , axis = 1 ) , np . ones ( shape = ( weights . shape [ 0 ] ) ) , 3 , err_msg = 'weights should sum to 1. action="%s"' % weights ) observation , done1 , ground_truth_obs = self . src . _step ( ) cash_observation = np . ones ( ( 1 , self . window_length , observation . shape [ 2 ] ) ) observation = np . concatenate ( ( cash_observation , observation ) , axis = 0 ) cash_ground_truth = np . ones ( ( 1 , 1 , ground_truth_obs . shape [ 2 ] ) ) ground_truth_obs = np . concatenate ( ( cash_ground_truth , ground_truth_obs ) , axis = 0 ) close_price_vector = observation [ : , - 1 , 3 ] open_price_vector = observation [ : , - 1 , 0 ] y1 = close_price_vector / open_price_vector rewards = np . empty ( shape = ( weights . shape [ 0 ] ) ) info = { } dones = np . empty ( shape = ( weights . shape [ 0 ] ) , dtype = bool ) for i in range ( weights . shape [ 0 ] ) : reward , current_info , done2 = self . sim [ i ] . _step ( weights [ i ] , y1 ) rewards [ i ] = reward info [ self . model_names [ i ] ] = current_info [ 'portfolio_value' ] info [ 'return' ] = current_info [ 'return' ] dones [ i ] = done2 info [ 'market_value' ] = np . cumprod ( [ inf [ "return" ] for inf in self . infos + [ info ] ] ) [ - 1 ] info [ 'date' ] = index_to_date ( self . start_idx + self . src . idx + self . src . step ) info [ 'steps' ] = self . src . step info [ 'next_obs' ] = ground_truth_obs self . infos . append ( info ) return observation , rewards , np . all ( dones ) or done1 , info
Step the environment by a vector of actions
Step the environment by a vector of actions
""" Step the environment by a vector of actions

        Args:
            action: (num_models, num_stocks + 1)

        Returns:

        """ # normalise just in case # np.array([cash_bias] + list(action))  # [w0, w1...] # so if weights are all zeros we normalise to [1,0...] # concatenate observation with ones # relative price vector of last observation day (close/open) # calculate return for buy and hold a bit of each asset # add dates
None
self
action
(num_models, num_stocks + 1)
( num_models num_stocks + 1 )