diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000..a4924cb Binary files /dev/null and b/.DS_Store differ diff --git a/.gitignore b/.gitignore index a85221e..56c69b8 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,6 @@ rlstock -.venv \ No newline at end of file +.venv +a.egg-info/ +__pycache__/ +*.egg*/ +.env \ No newline at end of file diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..23fd35f --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,3 @@ +{ + "editor.formatOnSave": true +} \ No newline at end of file diff --git a/README.md b/README.md index 7bd1a13..84bba11 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,6 @@ # RL Stock Portfolio Optimization -```pip install -r requirements.txt``` +```bash +pip install -r requirements.txt +echo TWELVE_DATA_API_KEY > .env +``` diff --git a/environment/__init__.py b/environment/__init__.py new file mode 100644 index 0000000..f01ae5e --- /dev/null +++ b/environment/__init__.py @@ -0,0 +1,2 @@ +from .trading_env import TradingEnv +from .raw_data import load_data diff --git a/environment/raw_data.py b/environment/raw_data.py new file mode 100644 index 0000000..4ed6f1b --- /dev/null +++ b/environment/raw_data.py @@ -0,0 +1,87 @@ +from dotenv import dotenv_values +from twelvedata import TDClient +import pandas_market_calendars as mcal +import pandas as pd +from datetime import timedelta +from itertools import batched +import time + +env_config = dotenv_values(".env") + +td = TDClient(env_config["TWELVE_DATA_API_KEY"]) +nyse_cal = mcal.get_calendar("NYSE") + + +def _load_data_limited(symbols: tuple[str], start_date, end_date): + """ + Get adjusted OHCLV data for symbols in open days across [start_date, end_date]. + + Uses 1 API token per symbol per call. Limited at 8 symbols per call (12data limit). + """ + + assert len(symbols) <= 8 + + # add 1 day to end_date to make the date range inclusive + if isinstance(end_date, str): + end_date = pd.to_datetime(end_date) + end_date += timedelta(days=1) + + open_days = nyse_cal.schedule(start_date=start_date, end_date=end_date).index + ts = ( + td.time_series( + symbol=symbols, + interval="1day", + outputsize=None, + start_date=start_date, + end_date=end_date, + ) + .as_pandas() + .reset_index() + .rename( + columns={ + "level_0": "Id", + "level_1": "Date", + "open": "Open", + "high": "High", + "low": "Low", + "close": "Close", + "volume": "Volume", + } + ) + ) + ts["Date"] = pd.to_datetime(ts["Date"]) + ts = ts[ts["Date"].isin(open_days)] + + return ts + + +def load_data(symbols: list[str], start_date, end_date, _batch_size=8): + """ + Get adjusted OHCLV data for symbols in open days across [start_date, end_date]. + + Uses len(symbols) // _batch_size API tokens per symbol per call. + + *Do not* pass in a different value for _batch_size unless you have a good reason to. + """ + _RATE_LIMIT = 60 # seconds between calls + + ret = pd.DataFrame() + for i, symbol_batch in enumerate(batched(symbols, n=_batch_size)): + if i != 0: + # don't wait for rate limit on first call + time.sleep(_RATE_LIMIT) + batch_df = _load_data_limited(symbol_batch, start_date, end_date) + ret = pd.concat([ret, batch_df]) + + ret.sort_values(by=["Date", "Id"], inplace=True) + ret.reset_index(drop=True, inplace=True) + return ret + + +if __name__ == "__main__": + data = load_data( + ["AAPL", "MSFT", "TSLA", "VOD", "NVDA", "AMZN", "BA", "DELL", "INTC", "CRM"], + start_date="2020-01-01", + end_date="2020-02-01", + ) + print(data) diff --git a/environment/trading_env.py b/environment/trading_env.py index 63edd51..2d0bd23 100644 --- a/environment/trading_env.py +++ b/environment/trading_env.py @@ -3,6 +3,9 @@ import pandas as pd import numpy as np from typing import Any, NewType, Tuple +from functools import cache + +# TODO: implement seeded rng import random import math @@ -12,25 +15,6 @@ # TODO: change this if a more specific ObsType is needed ObsType = pd.DataFrame -def reward_function( - yesterday_close: np.ndarray, - today_close: np.ndarray, - time: int, - num_risky_assets: int, - prev_positions: np.ndarray, - curr_positions: np.ndarray, - transaction_percentage: float, - ) -> float: - u_t = np.ones(num_risky_assets + 1) - u_t[1:] = today_close / yesterday_close - - momemtum_weights = (u_t * prev_positions)/(u_t.dot(prev_positions)) - transaction_cost = today_close.dot(np.abs(momemtum_weights[1:] - curr_positions[1:])) * transaction_percentage - - reward = math.log((u_t * transaction_cost).dot(prev_positions) - transaction_cost) - - return reward - class TradingEnv(gym.Env): """ @@ -53,6 +37,7 @@ class TradingEnv(gym.Env): _get_capital: Calculates the current capital based on positions and market changes. reset: Resets the environment to its initial state. """ + COL_ID = "Id" COL_OPEN = "Open" COL_CLOSE = "Close" @@ -70,12 +55,26 @@ def __init__( initial_capital: float, transaction_cost: float, reward_function: Callable, - batch_len: int, + episode_len: int, index_to_id: list[str], + seed: int, ) -> None: """ Initializes the trading environment. + The ohclv dataframe must contain a timeseries column for the following features: + - Date, Id, Open, High, Low, Close, Volume, + and optionally an AdjClose column. + + An example dataframe looks like: + Date Id Open High Low Close Volume + 0 2024-11-20 AAPL 228.059998 229.929993 225.889999 229.000000 35169600 + 1 2024-11-20 GOOG 178.627344 178.907025 175.131310 177.129044 15729800 + 2 2024-11-20 MSFT 416.037221 416.456396 409.759778 414.659973 19191700 + 3 2024-11-21 AAPL 228.880005 230.160004 225.710007 228.520004 42108300 + 4 2024-11-21 GOOG 175.256171 175.381029 165.122663 169.048218 38839400 + ... + Args: ohclv_data (pd.DataFrame): Input OHCLV data. num_risky_assets (int): Number of risky assets (e.g., stocks). @@ -85,27 +84,44 @@ def __init__( reward_function (Callable): Function to compute the reward. index_to_id (list[str]): List mapping indices to asset identifiers. """ + np.random.seed(seed) + random.seed(seed) + self._ohclv_data = ohclv_data + self._format_ohclv_data() self._num_risky_assets = num_risky_assets self._positions = np.zeros(num_risky_assets + 1) # +1 for cash self._positions[0] = 1 self._window_len = window_len self._initial_capital = initial_capital - self._capital = initial_capital + self._portfolio_value = initial_capital self._tcost = transaction_cost self._reward_function = reward_function self._index_to_id = index_to_id - self._batch_len = batch_len + self._episode_len = episode_len + self.reward = 0 self._data_len = self._ohclv_data[self.COL_TIME].max() # initially starts as the "start_time" - self._cur_end_time = random.randint(self._window_len - 1, self._data_len - self._batch_len + 1) + self._cur_end_time = random.randint( + self._window_len - 1, self._data_len - self._episode_len + 1 + ) + self._episode_end_time = self._cur_end_time + self._episode_len - # using the "start_time" to get the episode_end_timeself._episode_end_time = self._cur_end_time + self._batch_len + # using the "start_time" to get the episode_end_timeself._episode_end_time = self._cur_end_time + self._episode_len super().__init__() + def _format_ohclv_data(self): + for col in ("Date", "Id", "Open", "High", "Low", "Close", "Volume"): + assert ( + col in self._ohclv_data + ), f"Column={col} not found in columns of ohclv data" + + self._ohclv_data["Time"] = pd.factorize(self._ohclv_data["Date"])[0] + self._ohclv_data.drop(columns=["Date"], inplace=True) + def _get_observation(self) -> ObsType: """ Retrieves the current observation window of OHCLV data. @@ -114,11 +130,40 @@ def _get_observation(self) -> ObsType: ObsType: A subset of the OHCLV data representing the current window. """ observation = self._ohclv_data[ - (self._cur_end_time - self._window_len + 1 <= self._ohclv_data[self.COL_TIME]) + ( + self._cur_end_time - self._window_len + 1 + <= self._ohclv_data[self.COL_TIME] + ) & (self._ohclv_data[self.COL_TIME] <= self._cur_end_time) ] return observation + def reset( + self, *, seed: int | None = None, options: dict[str, Any] | None = None + ) -> tuple[ObsType, dict[str, Any]]: + """ + Resets the environment to its initial state. + + Args: + seed (int, optional): Random seed for reproducibility. Defaults to None. + options (dict, optional): Additional reset options. Defaults to None. + + Returns: + tuple: Initial observation and additional info. + """ + np.random.seed(seed) + random.seed(seed) + + self._cur_end_time = random.randint( + self._window_len - 1, self._data_len - self._episode_len + 1 + ) + self._episode_end_time = self._cur_end_time + self._episode_len + self._capital = self._initial_capital + self._positions = np.zeros(self._num_risky_assets) + self._positions[0] = 1 # First index represents cash value + + return self._get_observation(), {} + def step( self, action: list[float] ) -> tuple[ObsType, float, bool, bool, dict[str, Any]]: @@ -133,86 +178,108 @@ def step( """ assert len(action) == len(self._positions) - reward = self._reward_function() + action = np.array(action) + + # reward = self._reward_function(self._positions, action, self._trf) + reward = 0 + + ## Update positions and portfolio value self._positions = action - self._capital = self._get_capital(self._positions) + self._update_portfolio_value(action) + + ## Get observations of upcoming day to input to the model observation = self._get_observation() + self._cur_end_time += 1 terminated = self._cur_end_time > self._episode_end_time - truncated = self._capital <= 0 + truncated = self._portfolio_value <= 0 info = {} return observation, reward, terminated, truncated, info - def _update_positions(self, new_positions): + def get_relevant_close_prices(self) -> Tuple[np.ndarray, np.ndarray]: + """Returns the close prices for the current and previous days.""" + close_column = self.COL_ADJ_CLOSE + if self.COL_ADJ_CLOSE not in self._ohclv_data.columns: + close_column = self.COL_CLOSE + + ## Note: paper implementation states yesterday_close == today_open, not true in practice, need to resolve this + + yesterday_close = self._ohclv_data[ + (self._ohclv_data[self.COL_TIME] == self._cur_end_time - 1) + ][close_column].to_numpy() + today_close = self._ohclv_data[ + (self._ohclv_data[self.COL_TIME] == self._cur_end_time) + ][close_column].to_numpy() + + return yesterday_close, today_close + + def _get_relative_close_prices(self): """ - 1. p_(t-1) = self._positions - 2. p_t' = (today_close * p_(t-1)) / (today_close.dot(p_(t-1))) (L1 norm of today_close * self._positions) - 3. p_t = new_positions - 4. transaction_cost = today_close.dot(abs(p_t' - p_t)) * t_cost_percentage - 5. capital = today_close.dot(p_t) - transaction_cost + If an asset changes price from $100 to $120, this would be a 1.2 relative price change. + First element is the relative change in cash, which is generally 1. + y_t = p_t / p_{t-1} """ - yesterday_close, today_close = self.get_relevant_close_prices() - old_positions = self._positions - market_adjusted_positions = today_close * old_positions + rel = today_close / yesterday_close - def reset( - self, *, seed: int | None = None, options: dict[str, Any] | None = None - ) -> tuple[ObsType, dict[str, Any]]: - """ - Resets the environment to its initial state. + CASH_CHANGE = 0 + rel = np.append([1 + CASH_CHANGE], rel) - Args: - seed (int, optional): Random seed for reproducibility. Defaults to None. - options (dict, optional): Additional reset options. Defaults to None. + return rel - Returns: - tuple: Initial observation and additional info. + def _get_market_adjusted_positions(self) -> np.array: + """Gets the updated position weights after close due to market movements during the day.""" + rel_prices = self._get_relative_close_prices() + return self._positions * rel_prices / self._positions.dot(rel_prices) + + def _trf(self, action: np.array) -> float: + """ + Calculates the transaction remainder factor (TRF) for the given action. + """ + adj_positions = self._get_market_adjusted_positions() + + def f(mu): + a = 1 / (1 - self._tcost * action[0]) + b = ( + 1 + - self._tcost * adj_positions[0] + - (2 * self._tcost - self._tcost**2) + * np.maximum(adj_positions - mu * action, 0)[1:].sum() + ) + return a * b + + EPS = 1e-6 + mu = self._tcost * np.abs((adj_positions - action)[1:]).sum() + next_mu = f(mu) + while abs(mu - next_mu) > EPS: + mu, next_mu = next_mu, f(next_mu) + + return next_mu + + def _update_portfolio_value(self, action: np.array) -> float: + """ + Updates the portfolio value based on the action of the agent using the transaction remainder factor. """ - self._cur_end_time = random.randint(self._window_len - 1, self._data_len - self._batch_len + 1) - self._episode_end_time = self._cur_end_time = self._batch_len - self._capital = self._original_capital - self._positions = [0] * self._num_risky_assets - self._positions[0] = 1 # First index represents cash value - return self._get_observation(), {} + mu = self._trf(action) + self._portfolio_value = ( + self._portfolio_value + * mu + * (self._get_relative_close_prices().dot(self._positions)) + ) - def get_relevant_close_prices(self) -> Tuple[np.ndarray, np.ndarray]: - close_column = self.COL_ADJ_CLOSE - if self.COL_ADJ_CLOSE not in self._ohclv_data.columns: - close_column = self.COL_CLOSE + def reward_function( + self, + mu, + close_prices, + ): - yesterday_close = self._ohclv_data[(self._ohclv_data[self.COL_TIME] == self._cur_end_time - 1)][close_column].to_numpy() - today_close = self._ohclv_data[(self._ohclv_data[self.COL_TIME] == self._cur_end_time)][close_column].to_numpy() + mu = self._trf() # ERROR: _trf requires an action + + close_prices = self._get_relative_close_prices() + reward = np.log(close_prices.dot(self._positions)) + return reward - return yesterday_close, today_close -if __name__ == "__main__": - import yfinance as yf - - def get_random_positions(num_assets) -> np.array: - positions = np.random.random(num_assets + 1) - positions /= positions.sum() - return positions - - tickers = yf.Tickers("MSFT AAPL GOOG") - data = tickers.download() - data = data[["Open", "High", "Low", "Close", "Volume"]] - flattened_data = ( - data.stack(level=1) # Move tickers to rows - .reset_index() # Reset index to convert to a flat DataFrame - .rename(columns={"level_1": "Ticker"}) # Rename the column for tickers - ) - flattened_data["Time"] = pd.factorize(flattened_data["Date"])[0] - - my_env = TradingEnv( - flattened_data, 3, 5, 1000, 0.03, lambda: -999, ["MSFT", "AAPL", "GOOG"] - ) - - while True: - positions = list(get_random_positions(3)) - _, _, terminated, truncated, _ = my_env.step(positions) - if terminated: - break diff --git a/requirements.txt b/requirements.txt index 6e9a6f2..c50c2a5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,10 +1,29 @@ +-e git+https://github.com/WAT-ai/RL-Stock-Portfolio.git@7a178337a3459983259b479b3f9df5a69cfb1551#egg=a +beautifulsoup4==4.12.3 +certifi==2024.8.30 +charset-normalizer==3.4.0 cloudpickle==3.1.0 Farama-Notifications==0.0.4 +frozendict==2.4.6 gymnasium==1.0.0 +html5lib==1.1 +idna==3.10 +lxml==5.3.0 +multitasking==0.0.11 numpy==2.1.3 pandas==2.2.3 +peewee==3.17.8 +platformdirs==4.3.6 python-dateutil==2.9.0.post0 +python-dotenv==1.0.1 pytz==2024.2 +requests==2.32.3 +-e git+https://github.com/WAT-ai/RL-Stock-Portfolio.git@7a178337a3459983259b479b3f9df5a69cfb1551#egg=rl_stock_portfolio +setuptools==75.6.0 six==1.16.0 +soupsieve==2.6 typing_extensions==4.12.2 tzdata==2024.2 +urllib3==2.2.3 +webencodings==0.5.1 +yfinance==0.2.50 diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..b061c37 --- /dev/null +++ b/setup.py @@ -0,0 +1,3 @@ +from setuptools import setup, find_packages + +setup(name="rl-stock-portfolio", version="0.1a", packages=find_packages()) diff --git a/tests/envtest.py b/tests/envtest.py new file mode 100644 index 0000000..7d5ba8b --- /dev/null +++ b/tests/envtest.py @@ -0,0 +1,50 @@ +import unittest +from environment import TradingEnv, load_data + + +class TradingEnvTests(unittest.TestCase): + """ + Objectives for further test cases (more to be added later): + - do datatype checks on every column to ensure every column contains only data type + - must return float reward + - env.step must return tuple of not NaN values (tuple must be valid size >= 4) + - each value in tuple must not be None + - observation must be dataframe of floats + - trf must be float + - ensure no NaN values (can either preprocessed beforehand or keep this as an edge case - to be discussed further) + - reset must reset values to defaults (0, None, etc. depending on the value) + - ensure capital does not fall below min threshold + """ + environment = None + + def test_initialization(self): + # a demonstration of usage if anything + + symbols = ["AAPL", "VOD"] + ohclv = load_data( + symbols, + start_date="2020-10-01", + end_date="2020-11-05", + ) + TradingEnvTests.environment = TradingEnv( + ohclv, + num_risky_assets=len(symbols), + window_len=5, + initial_capital=1000, + transaction_cost=0, + reward_function=lambda x: 1, + episode_len=10, + index_to_id={x: i for i, x in enumerate(symbols)}, + seed=123, + ) + self.assertIsInstance(TradingEnvTests.environment, TradingEnv) + + def test_public_methods(self): + assert TradingEnvTests.environment.reset() + assert TradingEnvTests.environment.get_relevant_close_prices() + assert TradingEnvTests.environment.reward_function(0.1, [0.1]*50) + + + +if __name__ == "__main__": + unittest.main()