Merge pull request #392 from tensortrade-org/environment-random-start

Environment random start

tensortrade/env/default/__init__.py

@@ -20,6 +20,7 @@ def create(portfolio: 'Portfolio',
            feed: 'DataFeed',
            window_size: int = 1,
            min_periods: int = None,
+           random_start_pct: float = 0.00,
            **kwargs) -> TradingEnv:
     """Creates the default `TradingEnv` of the project to be used in training
     RL agents.
@@ -39,6 +40,9 @@ def create(portfolio: 'Portfolio',
         The size of the look back window to use for the observation space.
     min_periods : int, optional
         The minimum number of steps to warm up the `feed`.
+    random_start_pct : float, optional
+        Whether to randomize the starting point within the environment at each
+        observer reset, starting in the first X percentage of the sample
     **kwargs : keyword arguments
         Extra keyword arguments needed to build the environment.
 
@@ -86,5 +90,6 @@ def create(portfolio: 'Portfolio',
         informer=kwargs.get("informer", informers.TensorTradeInformer()),
         renderer=renderer,
         min_periods=min_periods,
+        random_start_pct=random_start_pct,
     )
     return env

tensortrade/env/default/observers.py

@@ -280,11 +280,11 @@ def has_next(self) -> bool:
         """
         return self.feed.has_next()
 
-    def reset(self) -> None:
+    def reset(self, random_start=0) -> None:
         """Resets the observer"""
         self.renderer_history = []
         self.history.reset()
-        self.feed.reset()
+        self.feed.reset(random_start)
         self.warmup()
 
 

tensortrade/env/generic/components/observer.py

@@ -54,6 +54,6 @@ def observe(self, env: 'TradingEnv') -> np.array:
         """
         raise NotImplementedError()
 
-    def reset(self):
+    def reset(self, random_start=0):
         """Resets the observer."""
         pass

tensortrade/env/generic/environment.py

@@ -16,6 +16,7 @@
 import logging
 
 from typing import Dict, Any, Tuple
+from random import randint
 
 import gym
 import numpy as np
@@ -65,6 +66,7 @@ def __init__(self,
                  informer: Informer,
                  renderer: Renderer,
                  min_periods: int = None,
+                 random_start_pct: float = 0.00,
                  **kwargs) -> None:
         super().__init__()
         self.clock = Clock()
@@ -76,6 +78,7 @@ def __init__(self,
         self.informer = informer
         self.renderer = renderer
         self.min_periods = min_periods
+        self.random_start_pct = random_start_pct
 
         for c in self.components.values():
             c.clock = self.clock
@@ -139,12 +142,21 @@ def reset(self) -> 'np.array':
         obs : `np.array`
             The first observation of the environment.
         """
+        if self.random_start_pct > 0.00:
+            size = len(self.observer.feed.process[-1].inputs[0].iterable)
+            random_start = randint(0, int(size * self.random_start_pct))
+        else:
+            random_start = 0
+
         self.episode_id = str(uuid.uuid4())
         self.clock.reset()
 
         for c in self.components.values():
             if hasattr(c, "reset"):
-                c.reset()
+                if isinstance(c, Observer):
+                    c.reset(random_start=random_start)
+                else:
+                    c.reset()
 
         obs = self.observer.observe(self)
 

tensortrade/feed/core/base.py

@@ -528,6 +528,8 @@ def __init__(self, source: "Iterable[T]", dtype: str = None):
         except StopIteration:
             self.stop = True
 
+        self._random_start = 0
+
     def forward(self) -> T:
         v = self.current
         try:
@@ -539,11 +541,14 @@ def forward(self) -> T:
     def has_next(self):
         return not self.stop
 
-    def reset(self):
+    def reset(self, random_start=0):
+        if random_start != 0:
+            self._random_start = random_start
+
         if self.is_gen:
             self.generator = self.gen_fn()
         else:
-            self.generator = iter(self.iterable)
+            self.generator = iter(self.iterable[self._random_start:])
         self.stop = False
 
         try:

tensortrade/feed/core/feed.py

@@ -2,7 +2,7 @@
 
 from typing import List
 
-from tensortrade.feed.core.base import Stream, T, Placeholder
+from tensortrade.feed.core.base import Stream, T, Placeholder, IterableStream
 
 
 class DataFeed(Stream[dict]):
@@ -54,9 +54,12 @@ def next(self) -> dict:
     def has_next(self) -> bool:
         return all(s.has_next() for s in self.process)
 
-    def reset(self) -> None:
+    def reset(self, random_start=0) -> None:
         for s in self.process:
-            s.reset()
+            if isinstance(s, IterableStream):
+                s.reset(random_start)
+            else:
+                s.reset()
 
 
 class PushFeed(DataFeed):

tensortrade/version.py

@@ -1 +1 @@
-__version__ = "1.0.4-dev0"
+__version__ = "1.0.4-dev1"

tests/tensortrade/unit/env/default/test_env.py

@@ -87,7 +87,58 @@ def test_runs_with_external_feed_only(portfolio):
         reward_scheme=reward_scheme,
         feed=feed,
         window_size=50,
-        enable_logger=False
+        enable_logger=False,
+    )
+
+    done = False
+    obs = env.reset()
+    while not done:
+        action = env.action_space.sample()
+        obs, reward, done, info = env.step(action)
+
+    assert obs.shape[0] == 50
+
+
+def test_runs_with_random_start(portfolio):
+
+    df = pd.read_csv("tests/data/input/bitfinex_(BTC,ETH)USD_d.csv").tail(100)
+    df = df.rename({"Unnamed: 0": "date"}, axis=1)
+    df = df.set_index("date")
+
+    bitfinex_btc = df.loc[:, [name.startswith("BTC") for name in df.columns]]
+    bitfinex_eth = df.loc[:, [name.startswith("ETH") for name in df.columns]]
+
+    ta.add_all_ta_features(
+        bitfinex_btc,
+        colprefix="BTC:",
+        **{k: "BTC:" + k for k in ['open', 'high', 'low', 'close', 'volume']}
+    )
+    ta.add_all_ta_features(
+        bitfinex_eth,
+        colprefix="ETH:",
+        **{k: "ETH:" + k for k in ['open', 'high', 'low', 'close', 'volume']}
+    )
+
+    streams = []
+    with NameSpace("bitfinex"):
+        for name in bitfinex_btc.columns:
+            streams += [Stream.source(list(bitfinex_btc[name]), dtype="float").rename(name)]
+        for name in bitfinex_eth.columns:
+            streams += [Stream.source(list(bitfinex_eth[name]), dtype="float").rename(name)]
+
+    feed = DataFeed(streams)
+
+    action_scheme = ManagedRiskOrders()
+    reward_scheme = SimpleProfit()
+
+    env = default.create(
+        portfolio=portfolio,
+        action_scheme=action_scheme,
+        reward_scheme=reward_scheme,
+        feed=feed,
+        window_size=50,
+        enable_logger=False,
+        random_start_pct=0.10,  # Randomly start within the first 10% of the sample
     )
 
     done = False