Update

Author: ClementPerroud

.gitignore

@@ -6,4 +6,5 @@ render_logs
 my_example.py
 pyproject.toml
 .pypirc
-*.egg-info
+*.egg-info
+*.pkl

README.md

@@ -48,6 +48,23 @@ df.sort_index(inplace= True)
 df.dropna(inplace= True)
 df.drop_duplicates(inplace=True)
 ```
+**1.1 (Optional) Download data** : The package provide a easy way to download data (works with CCTX ans use asyncio to get FAST) :
+```python
+from gym_trading_env.downloader import download
+import datetime
+
+download(
+    exchange_names = ["binance", "bitfinex2", "huobi"],
+    symbols= ["BTC/USDT", "ETH/USDT"],
+    timeframe= "30m",
+    dir = "test/data",
+    since= datetime.datetime(year= 2019, month= 1, day=1),
+    until = datetime.datetime(year= 2023, month= 1, day=1),
+)
+```
+This function use pickle format to save the OHLCV data. You will need to import the dataset with ```pd.read_pickle('... .pkl', ...)```. The function supports exchange_names ```binance```, ```biftfinex2``` (API v2) and ```huobi```.
+
+
 **2 - Create your feature**. Your RL-agent will need some good, preprocessed features. It is your job to make sure it has everything it needs.
 **The feature column names need to contain the keyword 'feature'. The environment will automatically detect them !**
 
@@ -72,16 +89,16 @@ The history object is similar to a DataFrame. It uses timestep and/or columns to
 
 
 Accessible columns of history object :
-- ```step```: ...,# Step = t
-- ```date```: ...,# Date at step t, datetime
-- ```reward```: ..., # Reward at step t
-- ```position_index```: ..., # Index of the position at step t amoung your position argument
-- ```position``` : ..., # Portfolio position at step t
+- ```step``` : Step = t
+- ```date``` : Date at step t, datetime
+- ```reward``` : Reward at step t
+- ```position_index``` : Index of the position at step t amoung your position argument
+- ```position``` : Portfolio position at step t
 
     *It gathers every data (not used as features) from your DataFrame and labels them with 'data_{column}'. For example :*
-- ```data_close```: Close price,
-- ```data_open```: Open price,
-- ```data_high```:Hight price,
+- ```data_close``` : Close price,
+- ```data_open``` : Open price,
+- ```data_high``` : High price,
 
     *......*
 
@@ -91,13 +108,13 @@ Accessible columns of history object :
     *It stores the distribution of the portfolio :*
 - ```portfolio_distribution_asset``` : The amount of owned asset (stock),
 - ```portfolio_distribution_fiat``` : The amount of owned fiat currency,
-- ```portfolio_distribution_borrowed_asset```: The amount of borrowed asset,
-- ```portfolio_distribution_borrowed_fiat```: The amount of borrowed fiat currency,
-- ```portfolio_distribution_interest_asset```: The total of cumalated interest generated by the borrowed asset.
-- ```portfolio_distribution_interest_fiat```: The total of cumalated interest generated by the borrowed fiat currency.
+- ```portfolio_distribution_borrowed_asset``` : The amount of borrowed asset,
+- ```portfolio_distribution_borrowed_fiat``` : The amount of borrowed fiat currency,
+- ```portfolio_distribution_interest_asset``` : The total of cumalated interest generated by the borrowed asset.
+- ```portfolio_distribution_interest_fiat``` : The total of cumalated interest generated by the borrowed fiat currency.
 
 
-**4 - Create the environment**
+**4 - Initiate the environment**
 
 ```python
 env = TradingEnv(...)
@@ -140,60 +157,52 @@ First, you need to save your results at the end of every episode you want to ren
 
 ```python
 ...
-# At the end of episode you want to render
+# At the end of the episode you want to render
 env.save_for_render(dir = "render_logs")
 ```
 
 Then in the separated render script. You can import and initiate a render object, and run the render in a localhost web app :
 ```python
 from gym_trading_env.renderer import Renderer
-renderer = Renderer(render_dir="render_logs")
+renderer = Renderer(render_logs_dir="render_logs")
 renderer.run()
 ```
 
 You can add **metrics** and plot **lines** with :
 ```python
-# Add lines
-# - Simple Moving Average - 10
-renderer.add_scatter(
-        name = "sma10",
-        function = lambda df : df["close"].rolling(10).mean(),
-        scatter_args = {
-            "line": {"color":'blue'}
-        })
-# - Simple Moving Average - 40
-renderer.add_scatter(
-        name = "sma40",
-        function = lambda df : df["close"].rolling(40).mean(),
-        scatter_args = {
-            "line": {"color": "purple"}
-        })
-
-# Add metrics 
-def max_drawdown(df):
-    current_max = df["portfolio_valuation"].iloc[0]
-    max_drawdown = 0
-    for i in range(len(df)):
-        current_max = max(df["portfolio_valuation"].iloc[i], current_max)
-        max_drawdown = min(max_drawdown, (df["portfolio_valuation"].iloc[i] - current_max)/current_max)
-    return f"{max_drawdown*100:0.2f}%"
-
-renderer.add_metric("Max drawdown", max_drawdown)
+renderer = Renderer(render_logs_dir="render_logs")
+
+# Add Custom Lines (Simple Moving Average)
+renderer.add_line( name= "sma10", function= lambda df : df["close"].rolling(10).mean(), line_options ={"width" : 1, "color": "purple"})
+renderer.add_line( name= "sma20", function= lambda df : df["close"].rolling(20).mean(), line_options ={"width" : 1, "color": "blue"})
+
+# Add Custom Metrics (Annualized metrics)
+renderer.add_metric(
+    name = "Annual Market Return",
+    function = lambda df : f"{ ((df['close'].iloc[-1] / df['close'].iloc[0])**(pd.Timedelta(days=365)/(df.index.values[-1] - df.index.values[0]))-1)*100:0.2f}%"
+)
+
+renderer.add_metric(
+        name = "Annual Portfolio Return",
+        function = lambda df : f"{((df['portfolio_valuation'].iloc[-1] / df['portfolio_valuation'].iloc[0])**(pd.Timedelta(days=365)/(df.index.values[-1] - df.index.values[0]))-1)*100:0.2f}%"
+)
+
+renderer.run()
 ```
 
 <img alt="Render example" src ="https://github.com/ClementPerroud/Gym-Trading-Env/blob/main/readme_images/render_customization.gif?raw=true" width = "800"/>
 
 
 
-```.add_scatter``` takes arguments :
-- ```name``` : The name of the scatter
-- ```function``` : The function used to compute the line. The function must take an argument ```df``` which is a DateFrame and return a Series, 1D-Array or list.
-- ```scatter_args``` : Paramaters added to the go.Scatter object during the process. It can be used to customize your plots. The [documentation of the go.Scatter object](https://plotly.com/python-api-reference/generated/plotly.graph_objects.Scatter.html) might help you.
+```.add_line``` takes arguments :
+- ```name``` (*required*): The name of the scatter
+- ```function``` (*required*): The function used to compute the line. The function must take an argument ```df``` which is a DateFrame and return a Series, 1D-Array or list.
+- ```line_options``` : Can contain a dict with keys ```color``` and ```width```
 
 
 ```.add_metric``` takes arguments :
 - ```name``` : The name of the metric
-- ```function``` : The function used to compute the line. The function must take an argument ```df``` which is a DateFrame and return a scalar.
+- ```function``` : The function used to compute the line. The function must take an argument ```df``` which is a DateFrame and return a **string** !
 
 
 Enjoy :)

readme_images/render.PNG

@@ -0,0 +1,96 @@
+import asyncio
+import ccxt.async_support as ccxt
+import pandas as pd
+import datetime
+from tqdm import tqdm
+
+
+EXCHANGE_LIMIT_RATES = {
+    "bitfinex2": {
+        "limit":10_000,
+        "pause_every": 1,
+        "pause" : 3, #seconds
+    },
+    "binance": {
+        "limit":1_000,
+        "pause_every": 10,
+        "pause" : 1, #seconds
+    },
+    "huobi": {
+        "limit":1_000,
+        "pause_every": 10,
+        "pause" : 1, #seconds
+    }
+}
+
+async def _ohlcv(exchange, symbol, timeframe, limit, step_since, timedelta):
+    result = await exchange.fetch_ohlcv(symbol = symbol, timeframe= timeframe, limit= limit, since=step_since)
+    result_df = pd.DataFrame(result, columns=["timestamp_open", "open", "high", "low", "close", "volume"])
+    result_df["date_open"] = pd.to_datetime(result_df["timestamp_open"], unit= "ms")
+    result_df["date_close"] = pd.to_datetime(result_df["timestamp_open"] + timedelta, unit= "ms")
+    return result_df
+
+async def _download_symbol(exchange, symbol, timeframe = '5m', since = int(datetime.datetime(year=2020, month= 1, day= 1).timestamp()*1E3), until = int(datetime.datetime.now().timestamp()*1E3), limit = 1000, pause_every = 10, pause = 1):
+    timedelta = int(pd.Timedelta(timeframe).to_timedelta64()/1E6)
+    tasks = []
+    results = []
+    for step_since in range(since, until, limit * timedelta):
+        tasks.append(
+            asyncio.create_task(_ohlcv(exchange, symbol, timeframe, limit, step_since, timedelta))
+        )
+        if len(tasks) >= pause_every:
+            results.extend(await asyncio.gather(*tasks))
+            await asyncio.sleep(pause)
+            tasks = []
+    if len(tasks) > 0 :
+        results.extend(await asyncio.gather(*tasks))
+    final_df = pd.concat(results, ignore_index= True)
+    final_df = final_df.loc[(since < final_df["timestamp_open"]) & (final_df["timestamp_open"] < until), :]
+    final_df.set_index('date_open', drop=True, inplace=True)
+    final_df.sort_index(inplace= True)
+    final_df.dropna(inplace=True)
+    final_df.drop_duplicates(inplace=True)
+    return final_df
+
+async def _download_symbols(exchange_name, symbols, dir, timeframe,  **kwargs):
+    exchange = getattr(ccxt, exchange_name)({ 'enableRateLimit': True })
+    for symbol in symbols:
+        df = await _download_symbol(exchange = exchange, symbol = symbol, timeframe= timeframe, **kwargs)
+        df.to_pickle(f"{dir}/{exchange_name}-{symbol.replace('/', '')}-{timeframe}.pkl")
+    await exchange.close()
+
+async def _download(exchange_names, symbols, timeframe, dir, since : datetime.datetime, until : datetime.datetime = datetime.datetime.now()):
+    tasks = []
+    for exchange_name in exchange_names:
+        
+        limit = EXCHANGE_LIMIT_RATES[exchange_name]["limit"]
+        pause_every = EXCHANGE_LIMIT_RATES[exchange_name]["pause_every"]
+        pause = EXCHANGE_LIMIT_RATES[exchange_name]["pause"]
+        tasks.append(
+            _download_symbols(
+                exchange_name = exchange_name, symbols= symbols, timeframe= timeframe, dir = dir,
+                limit = limit, pause_every = pause_every, pause = pause,
+                since = int(since.timestamp()*1E3), until = int(until.timestamp()*1E3)
+            )
+        )
+    await asyncio.gather(*tasks)
+def download(*args, **kwargs):
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(
+        _download(*args, **kwargs)
+    )
+
+async def main():
+    await _download(
+        ["binance", "bitfinex2", "huobi"],
+        symbols= ["BTC/USDT", "ETH/USDT"],
+        timeframe= "30m",
+        dir = "test/data",
+        since= datetime.datetime(year= 2019, month= 1, day=1),
+    )
+
+
+
+if __name__ == "__main__":
+    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+    asyncio.run(main())

readme_images/render.gif

@@ -4,6 +4,7 @@
 import numpy as np
 import datetime
 import glob
+from pathlib import Path    
 
 from collections import Counter
 from .utils.history import History
@@ -81,9 +82,8 @@ def _get_obs(self):
             "position" : self._position
             }
 
-    def reset(self, seed = None, df = None):
+    def reset(self, seed = None):
         super().reset(seed = seed)
-        if df is not None: self._set_df(df)
         self._step = 0
         self._limit_orders = {}
         if self.windows is not None: self._step = self.windows
@@ -178,19 +178,28 @@ def save_for_render(self, dir = "render_logs"):
         render_df.to_pickle(f"{dir}/{self.name}_{datetime.datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}.pkl")
 
 class MultiDatasetTradingEnv(TradingEnv):
-    def __init(self, dataset_dir, *args, **kwargs):
+    def __init__(self, dataset_dir, preprocess, *args, **kwargs):
         self.dataset_dir = dataset_dir
+        self.preprocess = preprocess
         self.dataset_pathes = glob.glob(self.dataset_dir)
         self.dataset_nb_uses = np.zeros(shape=(len(self.dataset_pathes), ))
-        df = self.pick_dataset()
-        super().__init__(df, *args, **kwargs)
+        super().__init__(self.next_dataset(), *args, **kwargs)
 
-    def pick_dataset(self):
+    def next_dataset(self):
         # Find the indexes of the less explored dataset
         potential_dataset_pathes = np.where(self.dataset_nb_uses == self.dataset_nb_uses.min())[0]
         # Pick one of them
         random_int = np.random.randint(potential_dataset_pathes.size)
         dataset_path = self.dataset_pathes[random_int]
         self.dataset_nb_uses[random_int] += 1 # Update nb use counts
-        return pd.read_pickle(dataset_path)
+
+        self.name = Path(dataset_path).name
+        return self.preprocess(pd.read_pickle(dataset_path))
+
+    def reset(self, seed=None):
+        self._set_df(
+            self.next_dataset()
+        )
+        print(f"Selected dataset {self.name} ...")
+        return super().reset(seed)
 

render.html

@@ -27,6 +27,9 @@
     padding: 10px;
     border: 1px solid #ededed;
     border-radius: 5%;
+    max-width: 100%;
+    white-space:nowrap; 
+    text-overflow:ellipsis;
 }
 #metrics{
     display: flex;

render_logs/BTCUSD_2023-04-02_18-58-48.pkl

@@ -0,0 +1,15 @@
+
+import sys  
+sys.path.append("./src")
+
+
+from gym_trading_env.downloader import download
+import datetime
+
+download(
+    exchange_names = ["binance", "bitfinex2", "huobi"],
+    symbols= ["BTC/USDT", "ETH/USDT"],
+    timeframe= "30m",
+    dir = "test/data",
+    since= datetime.datetime(year= 2019, month= 1, day=1),
+)

render_logs/BTCUSD_2023-04-02_19-39-56.pkl

@@ -39,9 +39,9 @@ def reward_function(history):
     )
 
 # Run the simulation
-truncated = False
+truncated, done = False, False
 observation, info = env.reset()
-while not truncated:
+while not truncated and not done:
     action = 5 #OR manually : action = int(input("Action : ")) 
     observation, reward, done, truncated, info = env.step(action)
 

src/gym_trading_env/downloader.py

@@ -0,0 +1,45 @@
+import sys  
+sys.path.append("./src")
+
+import pandas as pd
+import numpy as np
+import time
+from gym_trading_env.environments import MultiDatasetTradingEnv
+
+# Generating features
+# WARNING : the column names need to contain keyword 'feature' !
+def preprocess(df):
+    df["feature_close"] = df["close"].pct_change()
+    df["feature_open"] = df["open"]/df["close"]
+    df["feature_high"] = df["high"]/df["close"]
+    df["feature_low"] = df["low"]/df["close"]
+    df["feature_volume"] = df["volume"] / df["volume"].rolling(7*24).max()
+    df.dropna(inplace= True)
+    return df
+
+
+# Create your own reward function with the history object
+def reward_function(history):
+    return np.log(history["portfolio_valuation", -1] / history["portfolio_valuation", -2]) #log (p_t / p_t-1 )
+
+env = MultiDatasetTradingEnv(
+        dataset_dir= 'test/data/*.pkl',
+        preprocess= preprocess,
+        windows= 5,
+        positions = [ -1, -0.5, 0, 0.5, 1, 1.5, 2], # From -1 (=full SHORT), to +1 (=full LONG) with 0 = no position
+        initial_position = 0, #Initial position
+        trading_fees = 0.01/100, # 0.01% per stock buy / sell
+        borrow_interest_rate= 0.0003/100, #per timestep (= 1h here)
+        reward_function = reward_function,
+        portfolio_initial_value = 1000, # in FIAT (here, USD)
+    )
+
+# Run the simulation
+truncated = False
+observation, info = env.reset()
+while not truncated:
+    action = env.action_space.sample() #OR manually : action = int(input("Action : ")) 
+    observation, reward, done, truncated, info = env.step(action)
+
+# Render
+env.save_for_render()