trade2copy.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Setting up the Trading Environment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# %pip install nsepy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define state space and action space\n",
    "n_features = 5  # Number of features in the state space\n",
    "action_space = [0, 1, 2]  # 0: Hold, 1: Buy Straddle, 2: Sell Straddle\n",
    "\n",
    "# Define reward function\n",
    "def calculate_reward_buy_straddle(curr_data, position, portfolio_value):\n",
    "    underlying_price = curr_data['Close']\n",
    "    call_price = curr_data['Call Option Price']\n",
    "    put_price = curr_data['Put Option Price']\n",
    "\n",
    "    if position == 0:\n",
    "        cost = call_price + put_price\n",
    "        if portfolio_value >= cost:\n",
    "            reward = 0  # No reward for opening a position\n",
    "        else:\n",
    "            reward = -100  # Penalty for insufficient funds\n",
    "    else:\n",
    "        reward = -10  # Penalty for invalid action\n",
    "\n",
    "    return reward\n",
    "\n",
    "def calculate_reward_sell_straddle(curr_data, position):\n",
    "    underlying_price = curr_data['Close']\n",
    "    call_price = curr_data['Call Option Price']\n",
    "    put_price = curr_data['Put Option Price']\n",
    "\n",
    "    if position == 0:\n",
    "        premium = call_price + put_price\n",
    "        reward = premium  # Reward for selling the straddle\n",
    "    else:\n",
    "        reward = -10  # Penalty for invalid action\n",
    "\n",
    "    return reward"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setting up the Trading Environment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import gym\n",
    "from gym import spaces\n",
    "import numpy as np\n",
    "\n",
    "class StraddleEnvironment(gym.Env):\n",
    "    def __init__(self, n_features, n_steps, initial_capital=10000):\n",
    "        super(StraddleEnvironment, self).__init__()\n",
    "        # Define action and observation spaces\n",
    "        self.action_space = spaces.Discrete(3)  # 0: hold, 1: buy straddle, 2: sell straddle\n",
    "        self.observation_space = spaces.Box(low=0, high=np.inf, shape=(n_features,), dtype=np.float32)\n",
    "\n",
    "        # Initialize other variables\n",
    "        self.initial_capital = initial_capital\n",
    "        self.current_capital = initial_capital\n",
    "        self.current_step = 0\n",
    "        self.max_steps = n_steps  # Define the number of steps in each episode\n",
    "\n",
    "    def reset(self):\n",
    "        # Reset the environment to the initial state\n",
    "        self.current_capital = self.initial_capital\n",
    "        self.current_step = 0\n",
    "        self.state = self.get_observation()\n",
    "        return self.state\n",
    "\n",
    "    def step(self, action):\n",
    "        # Execute one step within the environment\n",
    "        reward = 0\n",
    "        done = False\n",
    "        info = {}\n",
    "\n",
    "        # Implement action logic\n",
    "        if action == 1:\n",
    "            # Buy straddle\n",
    "            # Execute buy logic\n",
    "            reward = calculate_reward_buy_straddle()\n",
    "        elif action == 2:\n",
    "            # Sell straddle\n",
    "            # Execute sell logic\n",
    "            reward = calculate_reward_sell_straddle()\n",
    "\n",
    "        # Update state\n",
    "        self.current_step += 1\n",
    "        self.state = self.get_observation()\n",
    "\n",
    "        # Check if the episode is done\n",
    "        if self.current_step >= self.max_steps:\n",
    "            done = True\n",
    "\n",
    "        return self.state, reward, done, info\n",
    "\n",
    "    def get_observation(self):\n",
    "        # Get the current observation/state\n",
    "        # You need to define the structure of the observation data\n",
    "        # based on the features you want to include\n",
    "        observation_data = np.random.uniform(low=0, high=1, size=(self.observation_space.shape))\n",
    "        return observation_data\n",
    "\n",
    "    def render(self, mode='human'):\n",
    "        # Render the environment (optional)\n",
    "        pass\n",
    "\n",
    "# Instantiate the environment with the required parameters\n",
    "n_features = 5  # Number of features in the state space\n",
    "n_steps = 100  # Number of steps in each episode\n",
    "env = StraddleEnvironment(n_features, n_steps)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data Gathering and Preprocessing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from nsepy import get_history\n",
    "from datetime import date\n",
    "import pandas as pd\n",
    "\n",
    "# Collect historical data\n",
    "start_date = date(2023, 1, 1)\n",
    "end_date = date(2023, 12, 31)\n",
    "underlying_data = get_history(symbol='SBIN', start=start_date, end=end_date, index=True)\n",
    "\n",
    "# Preprocessing (e.g., handling missing values, outliers, feature engineering)\n",
    "def preprocess_data(data):\n",
    "    # Example preprocessing steps\n",
    "    data = data.fillna(method='ffill')  # Fill missing values with forward-fill\n",
    "    data['SMA'] = data['Close'].rolling(window=20).mean()  # Calculate simple moving average\n",
    "    data['RSI'] = calculate_rsi(data['Close'], 14)  # Calculate RSI\n",
    "    return data\n",
    "\n",
    "# Function for calculating RSI\n",
    "def calculate_rsi(prices, window):\n",
    "    deltas = np.diff(prices)\n",
    "    seed = deltas[:window]\n",
    "    up = seed[seed >= 0].sum() / window\n",
    "    down = -seed[seed < 0].sum() / window\n",
    "    rs = up / down\n",
    "    rsi = np.zeros_like(prices)\n",
    "    rsi[:window] = 100 - 100 / (1 + rs)\n",
    "\n",
    "    for i in range(window, len(prices)):\n",
    "        delta = deltas[i - 1]\n",
    "        if delta > 0:\n",
    "            upval = delta\n",
    "            downval = 0\n",
    "        else:\n",
    "            upval = 0\n",
    "            downval = -delta\n",
    "\n",
    "        up = (up * (window - 1) + upval) / window\n",
    "        down = (down * (window - 1) + downval) / window\n",
    "        rs = up / down\n",
    "        rsi[i] = 100 - 100 / (1 + rs)\n",
    "\n",
    "    return rsi\n",
    "\n",
    "# Preprocess the data\n",
    "preprocessed_data = preprocess_data(underlying_data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Reinforcement Learning Model Design"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "from collections import deque\n",
    "\n",
    "\n",
    "# Define Q-learning algorithm\n",
    "class QLearningAgent:\n",
    "    def __init__(self, state_size, action_size, learning_rate=0.1, discount_factor=0.99, epsilon=1.0, epsilon_decay=0.995, epsilon_min=0.01):\n",
    "        self.state_size = state_size\n",
    "        self.action_size = action_size\n",
    "        self.learning_rate = learning_rate\n",
    "        self.discount_factor = discount_factor\n",
    "        self.epsilon = epsilon\n",
    "        self.epsilon_decay = epsilon_decay\n",
    "        self.epsilon_min = epsilon_min\n",
    "        self.q_table = np.zeros((state_size, action_size))\n",
    "        self.memory = deque(maxlen=2000)\n",
    "\n",
    "    def act(self, state):\n",
    "        if np.random.rand() <= self.epsilon:\n",
    "            return np.random.choice(self.action_size)\n",
    "        else:\n",
    "            return np.argmax(self.q_table[state])\n",
    "\n",
    "    def learn(self, state, action, reward, next_state, done):\n",
    "        q_value = self.q_table[state, action]\n",
    "        next_max_q = np.max(self.q_table[next_state])\n",
    "        new_q = reward + self.discount_factor * next_max_q * (1 - done)\n",
    "        self.q_table[state, action] += self.learning_rate * (new_q - q_value)\n",
    "        self.epsilon = max(self.epsilon_decay * self.epsilon, self.epsilon_min)\n",
    "\n",
    "    def train(self, env, episodes):\n",
    "        for episode in range(episodes):\n",
    "            state = env.reset()\n",
    "            done = False\n",
    "            while not done:\n",
    "                action = self.act(state)\n",
    "                next_state, reward, done, _ = env.step(action)\n",
    "                self.learn(state, action, reward, next_state, done)\n",
    "                state = next_state\n",
    "\n",
    "# Create the Q-learning agent\n",
    "state_size = env.observation_space.shape[0]\n",
    "action_size = len(action_space)\n",
    "agent = QLearningAgent(state_size, action_size)\n",
    "\n",
    "# Train the agent\n",
    "episodes = 1000\n",
    "agent.train(env, episodes)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Training the Agent"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Training the Q-learning agent\n",
    "def train_agent():\n",
    "    # Train the agent using historical data\n",
    "    for episode in range(num_episodes):\n",
    "        state = env.reset()\n",
    "        done = False\n",
    "        while not done:\n",
    "            action = agent.act(state)\n",
    "            next_state, reward, done, info = env.step(action)\n",
    "            \n",
    "            # Calculate the reward based on the action\n",
    "            if action == 1:  # Buy Straddle\n",
    "                reward = calculate_reward_buy_straddle(env.data.iloc[env.current_step], env.position, env.portfolio_value)\n",
    "            elif action == 2:  # Sell Straddle\n",
    "                reward = calculate_reward_sell_straddle(env.data.iloc[env.current_step], env.position)\n",
    "            \n",
    "            agent.learn(state, action, reward, next_state, done)\n",
    "            state = next_state\n",
    "\n",
    "# Set the number of episodes\n",
    "num_episodes = 1000\n",
    "\n",
    "# Train the agent\n",
    "train_agent()"
   ]
  }
 ],
 "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.11.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}