pythonlessons

def act_continuous(self, state):
    # Use the network to predict the next action to take, using the model
    pred = self.Actor.predict(state)

    low, high = -1.0, 1.0 # -1 and 1 are boundaries of tanh
    action = pred + np.random.uniform(low, high, size=pred.shape) * self.std
    action = np.clip(action, low, high)

    logp_t = self.gaussian_likelihood(action, pred, self.log_std)

pythonlessons

def replay(self, states, actions, rewards, dones, next_states, logp_ts):
    # reshape memory to appropriate shape for training
    states = np.vstack(states)
    next_states = np.vstack(next_states)
    actions = np.vstack(actions)
    logp_ts = np.vstack(logp_ts)

    # Get Critic network predictions 
    values = self.Critic.predict(states)
    next_values = self.Critic.predict(next_states)

pythonlessons

class CustomEnv:
    def __init__(self, arg1, arg2, ...):
        # Define action space and state size
        
        # Example when using discrete actions of 0,1,2:
        self.action_space = np.array([0, 1, 2])
        
        # Example for using image as input for custom environment:
        self.state_size = np.empty((HEIGHT, WIDTH, CHANNELS)), dtype=np.uint8)
    

pythonlessons

import pandas as pd
import numpy as np
import random
from collections import deque

class CustomEnv:
    # A custom Bitcoin trading environment
    def __init__(self, df, initial_balance=1000, lookback_window_size=50):
        # Define action space and state size and other custom parameters
        self.df = df.dropna().reset_index()

pythonlessons

# Reset the state of the environment to an initial state
def reset(self, env_steps_size = 0):
    self.balance = self.initial_balance
    self.net_worth = self.initial_balance
    self.prev_net_worth = self.initial_balance
    self.crypto_held = 0
    self.crypto_sold = 0
    self.crypto_bought = 0
    if env_steps_size > 0: # used for training dataset
        self.start_step = random.randint(self.lookback_window_size, self.df_total_steps - env_steps_size)

pythonlessons

# Execute one time step within the environment
def step(self, action):
    self.crypto_bought = 0
    self.crypto_sold = 0
    self.current_step += 1

    # Set the current price to a random price between open and close
    current_price = random.uniform(
        self.df.loc[self.current_step, 'Open'],
        self.df.loc[self.current_step, 'Close'])

pythonlessons

# Get the data points for the given current_step
def _next_observation(self):
    self.market_history.append([self.df.loc[self.current_step, 'Open'],
                                self.df.loc[self.current_step, 'High'],
                                self.df.loc[self.current_step, 'Low'],
                                self.df.loc[self.current_step, 'Close'],
                                self.df.loc[self.current_step, 'Volume']
                                ])
    obs = np.concatenate((self.market_history, self.orders_history), axis=1)
    return obs

pythonlessons

# render environment
def render(self):
    print(f'Step: {self.current_step}, Net Worth: {self.net_worth}')

pythonlessons

def Random_games(env, train_episodes = 50, training_batch_size=500):
    average_net_worth = 0
    for episode in range(train_episodes):
        state = env.reset(env_steps_size = training_batch_size)

        while True:
            env.render()

            action = np.random.randint(3, size=1)[0]

pythonlessons

df = pd.read_csv('./pricedata.csv')
df = df.sort_values('Date')

lookback_window_size = 50
train_df = df[:-720-lookback_window_size]
test_df = df[-720-lookback_window_size:] # 30 days

train_env = CustomEnv(train_df, lookback_window_size=lookback_window_size)
test_env = CustomEnv(test_df, lookback_window_size=lookback_window_size)