deep-q-learning-for-cryptocurrency-trading.ipynb

{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"codemirror_mode":{"name":"ipython","version":3},"file_extension":".py","mimetype":"text/x-python","name":"python","nbconvert_exporter":"python","pygments_lexer":"ipython3","version":"3.7.12"},"papermill":{"default_parameters":{},"duration":18671.452533,"end_time":"2023-03-23T04:11:56.424729","environment_variables":{},"exception":null,"input_path":"__notebook__.ipynb","output_path":"__notebook__.ipynb","parameters":{},"start_time":"2023-03-22T23:00:44.972196","version":"2.3.4"}},"nbformat_minor":5,"nbformat":4,"cells":[{"cell_type":"code","source":"import pandas as pd\nimport numpy as np\nimport matplotlib.pyplot as plt\nimport random\nfrom tensorflow.keras.models import Sequential\nfrom tensorflow.keras.layers import Dense\nfrom tensorflow.keras.optimizers import Adam\nfrom tensorflow.keras.activations import relu, linear\nfrom keras.layers import Dense, Dropout, Conv1D, MaxPooling2D, Activation, Flatten, Embedding, Reshape,MaxPooling1D,LeakyReLU\n!pip install yfinance\nimport yfinance as yf","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"Crypto_name = [\"BTC-USD\"] # replace with other crypto currency e.g. \"ETH-USD\" 'XRP-USD' \"LTC-USD\"\nstart_date=\"2022-06-20\"\nend_date='2023-06-20'\nprices=pd.DataFrame()\nfor i in Crypto_name:\n    data= yf.Ticker(i)\n    data = data.history(start=start_date , end=end_date,interval=\"1h\")\n    colse=pd.DataFrame(data.Close)\n    prices[i] = colse","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Plot the closing changes ","metadata":{}},{"cell_type":"code","source":"# plot the closing price changes in the given period\nplt.xlabel(\"date\")\nplt.ylabel(\"closing price\")\nplt.title(f\"bitcoin closing prices from{start_date} to {end_date}\")\nplt.plot(prices['BTC-USD'])\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Generate the action space","metadata":{}},{"cell_type":"code","source":"import gym\nfrom gym import spaces\naction_choices = np.linspace(-20, 20, num=51) # using linespace to generate 25 actions to buy or sell in [0.5$,20$] interval\nprint(action_choices)\nplt.xlabel(\"action id\")\nplt.ylabel(\"action value\")\nplt.title(f\"generated discrete action space\")\nplt.scatter([act for act in range(len(action_choices))],action_choices)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Define the DQN agent (double DQN)","metadata":{}},{"cell_type":"code","source":"class DQNAgent :\n    def __init__(self, state_size, action_size,batch_size,update_target_interval=100):\n        self.state_size = state_size\n        self.action_size = action_size\n        self.memory = ReplayBuffer(1000000,state_size,action_size)\n        self.gamma = 0.99\n        self.epsilon = 1.0\n        self.epsilon_min =0.1\n        self.epsilon_decay = 0.995\n        self.batch_size = batch_size\n        self.learning_rate = 0.001\n        \n        self.model = self._build_model()\n        self.target_model = self._build_model()\n        self.target_model.set_weights(self.model.get_weights())\n        \n        self.update_target_interval =update_target_interval\n        self.update_target_counter=0\n        \n    def _build_model(self):\n        model = Sequential()\n        model.add(Conv1D(128,8, input_shape=(self.state_size,1), padding='same'))\n        model.add(LeakyReLU())\n        model.add(MaxPooling1D(2, padding='same'))\n        model.add(Conv1D(64,8, padding='same'))\n        model.add(LeakyReLU())\n        model.add(Flatten())\n        model.add(Dense(384))\n        model.add(Activation('relu'))\n        model.add(Dense(256))\n        model.add(Activation('relu'))\n        model.add(Dense(self.action_size, activation='linear')) \n        model.compile(loss=\"mse\", optimizer=Adam(lr=self.learning_rate), metrics=['accuracy'])\n        return model\n\n    def remember(self, state, action, reward, new_state, done):\n        self.memory.store_transition(state, action, reward, new_state, done)\n        \n    def act(self, state,test_mode=False):\n        if not test_mode:\n            if np.random.rand() <= self.epsilon :\n                return random.randrange(self.action_size)\n        act_values = self.model.predict(np.expand_dims(state,axis=0), verbose=0)\n        return np.argmax(act_values[0])\n    \n    def train(self, batch_size) :\n        if self.memory.mem_cntr  < batch_size:\n            return\n        state, action, reward, new_state, done = self.memory.sample_buffer(batch_size)\n        \n        qState=self.model.predict(state,verbose=0)\n        qNextState=self.model.predict(new_state,verbose=0)\n        qNextStateTarget=self.target_model.predict(new_state,verbose=0)\n        maxActions=np.argmax(qNextState,axis=1)\n        batchIndex = np.arange(batch_size, dtype=np.int32)\n        qState[batchIndex,action]=(reward+(self.gamma*qNextStateTarget[batchIndex,maxActions.astype(int)]*(1-done)))\n        _=self.model.fit(x=state,y=qState,verbose=0,epochs=65)\n\n        self.update_target_counter+=1\n        if self.update_target_counter % self.update_target_interval==0 :\n            self.target_model.set_weights(self.model.get_weights())\n        if self.epsilon > self.epsilon_min:\n            self.epsilon *= self.epsilon_decay","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"class ReplayBuffer():\n    def __init__(self, max_size, input_shape, n_actions):\n        self.mem_size = max_size\n        self.mem_cntr = 0\n        self.state_memory = np.zeros((self.mem_size, input_shape))\n        self.new_state_memory = np.zeros((self.mem_size, input_shape))\n        self.action_memory = np.zeros(self.mem_size,dtype=int)\n        self.reward_memory = np.zeros(self.mem_size)\n        self.terminal_memory = np.zeros(self.mem_size)\n\n    def store_transition(self, state, action, reward, state_, done):\n        index = self.mem_cntr % self.mem_size\n        self.state_memory[index] = state\n        self.action_memory[index] = action\n        self.reward_memory[index] = reward\n        self.new_state_memory[index] = state_\n        self.terminal_memory[index] = done\n\n        self.mem_cntr += 1\n\n    def sample_buffer(self, batch_size):\n        max_mem = min(self.mem_cntr, self.mem_size)\n\n        batch = np.random.choice(max_mem, batch_size)\n\n        states = self.state_memory[batch]\n        actions = self.action_memory[batch]\n        rewards = self.reward_memory[batch]\n        states_ = self.new_state_memory[batch]\n        dones = self.terminal_memory[batch]\n\n        return states, actions, rewards, states_, dones","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"class TradingEnv(gym.Env) :\n    def __init__(self,action_choices, init_capital=1000, stock_price_history=[], window_size=30):\n        self.init_capital = init_capital #amount of money we have at the initial step\n        self.stock = 0 # initial amount of stock we have (eg. 0 Bitcoin at start)\n        self.stock_price_history = stock_price_history # the full series of stock or currency values \n        self.window_size = window_size # amount of data we look at to predict the next price\n        self.current_step = 0 # the inital location to start\n        self.action_space = spaces.Discrete(len(action_choices))\n        self.observation_space = spaces.Box(low=0, high=np.inf, shape=(self.window_size,))\n        self.reset()\n        \n    def reset(self) :\n        self.current_step = 0\n        self.stock = 0\n        self.capital = self.init_capital #initial current capitaql to initial_capital\n        return self._next_observation() # return the first observation\n    \n    def _next_observation(self):\n        prices = self.stock_price_history[self.current_step:self.current_step+self.window_size] \n        #return the price seris according to current place and up to window_size eg. [23 to 23 + 30]\n        return np.array(prices)\n    \n    def step(self, action):\n        stock_price = self.stock_price_history[self.current_step+self.window_size]\n        portfolio_value = (self.capital + self.stock * stock_price) # total portfolio value including cash and stocks\n        if action > 0 and action <= self.capital :\n            self.capital -= action\n            self.stock += action/stock_price\n        elif action < 0 and (self.stock * stock_price)>(-action):\n            self.stock += action/stock_price                        \n            self.capital -= action\n        new_portfolie_value = self.capital + self.stock * self.stock_price_history[self.current_step+self.window_size+1]\n        reward = new_portfolie_value - portfolio_value  # reward = protfolio value after commiting action - portfolio before commiting action\n        self.current_step += 1\n        done = self.current_step+self.window_size + 2 >= len(self.stock_price_history) # if we will reach the end of price series in next step of the environment\n        return self._next_observation(), reward, done, new_portfolie_value # new_portfolie_value is returned due to track agent progress and its optional to log progress only","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Split data into training and testing sets\nclosing_price=prices[\"BTC-USD\"]\nsplit_index = int(0.8 * len(closing_price))\ntrain_prices = closing_price[:split_index]\ntest_prices = closing_price[split_index:]\n#train_prices,test_prices","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Initialize the trading environment and DQN agent\ntrain_env= TradingEnv(stock_price_history=train_prices,action_choices=action_choices)\ntest_env = TradingEnv(stock_price_history=test_prices,action_choices=action_choices)\nstate_size = train_env.observation_space.shape[0]\naction_size = train_env.action_space.n\nagent = DQNAgent(state_size, action_size,batch_size=50,update_target_interval=100)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#main loop\nagent_value = []\nfor e in range(10):\n    state = train_env.reset()\n    done = False\n    score = 0\n    steps=0\n    while not done:\n        action = agent.act(state)\n        next_state, reward, done ,value = train_env.step(action_choices[action])\n        agent.remember(state, action, reward, next_state, done)\n        state = next_state\n        score += reward\n        agent_value.append(value)\n        steps+=1\n        if (steps%10)==0:\n            print(f\"step{steps} value os far:{value}   cap:{train_env.capital} st:{train_env.stock} eps:{agent.epsilon}\")\n            plt.plot(agent_value)\n            plt.show()\n        agent.train(50)\n    print(f'Episode {e}, Score(total_reward): {score:.4f}')","metadata":{"trusted":true},"execution_count":null,"outputs":[]}]}