Merge CHIDI-FEB10-EVB-OPTIMIZATION into main

module_test/raw_code/DataManagers/DataManagers.py

@@ -347,10 +347,14 @@ def query_thetadata(self,
                     data = pd.DataFrame(columns=THETA_DATA_COLUMNS)
                     return data
                 data = data[~data.index.duplicated(keep='first')]
-                open_interest = retrieve_openInterest(symbol=self.symbol, end_date=end, exp=exp, right=right, start_date=start, strike=strike, print_url=print_url).set_index('Datetime')
-                open_interest.drop_duplicates(inplace = True)
-                data['Open_interest'] = open_interest['Open_interest']
-                data.index = default_timestamp(data.index)
+                try:
+                    open_interest = retrieve_openInterest(symbol=self.symbol, end_date=end, exp=exp, right=right, start_date=start, strike=strike, print_url=print_url).set_index('Datetime')
+                    open_interest.drop_duplicates(inplace = True)
+                    data['Open_interest'] = open_interest['Open_interest']
+                    data.index = default_timestamp(data.index)
+                except Exception as e:
+                    logger.error(f"Error retrieving open interest data for {self.symbol} from {start} to {end}: {e}. Filling Open_interest with NaN.")
+                    data['Open_interest'] = np.nan
                 return data
 
             else:

trade/backtester_/_helper.py

@@ -1,3 +1,4 @@
+
 from typing import Any, Callable, Dict, Optional, TYPE_CHECKING
 import pandas as pd
 from .data import PTDataset
@@ -86,7 +87,16 @@ def _next(self):
             if verbose:
                 print(f"Opening position on {date} at price {self.data.Close[-1]}")
                 print(f"Info: {self.brain.info_on_date(date=date)}")
-            self.buy()
+            if open_decision.side == 1:
+                if verbose:
+                    print("Going LONG")
+                self.buy()
+            elif open_decision.side == -1:
+                if verbose:
+                    print("Going SHORT")
+                self.sell()
+            else:
+                raise ValueError(f"Invalid side in open_decision: {open_decision.side}")
             self.brain.open_action(
                 date=date,
                 signal_id=open_decision.signal_id,

trade/backtester_/_strategy.py

@@ -14,6 +14,8 @@
 from ._types import Side, SideInt  # noqa
 from trade.backtester_.indicators import (
     compute_atr_loss,
+    update_atr_trail_long,
+    update_atr_trail_short,
 )
 
 
@@ -270,7 +272,7 @@ def __init__(
 
         self.position_info: Optional[PositionInfo] = PositionInfo()
         self.stop: Optional[float] = None
-        self.indicators: Dict[str, Any] = {}
+        self.indicators: Dict[str, Indicator] = {}
 
         # Cache index + numpy views for speed and consistent date handling
         self._df = self.data.data.copy()  # expects a DataFrame-like
@@ -1179,7 +1181,7 @@ def plot_strategy_indicators(self, log_scale: bool = True, add_signal_marker: bo
                     )
         ## Indicators
         for ind_name, indicator in self.indicators.items():
-            ind_values = indicator.values.loc[df.index]
+            ind_values = indicator.series[df.index]
             if indicator.overlay:
                 fig.add_trace(
                     go.Scatter(
@@ -1281,13 +1283,17 @@ def __init__(
         average_type: str = "w",
         start_trading_date: Optional[str] = None,
         ticker: Optional[str] = None,
+        is_long: bool = True,
+        is_short: bool = False,
         **kwargs,
     ):
         super().__init__(data=data, **kwargs)
         self.atr_period = atr_period
         self.atr_factor = atr_factor
         self.trail_type = trail_type
         self.average_type = average_type
+        self.is_long = is_long
+        self.is_short = is_short
         self.loss_series: Optional[pd.Series] = None
 
     def setup(self) -> None:
@@ -1298,4 +1304,22 @@ def setup(self) -> None:
             atr_factor=self.atr_factor,
             trail_type=self.trail_type,
             average_type=self.average_type,
-        )
+        )
+
+    def open_action(self, *, signal_id = None, entry_price = None, side = None, date = None, index = None):
+        idx, _ = self._resolve(date=date, index=index)
+
+        if self.is_long:
+            self.stop = update_atr_trail_long(
+                close=float(self.close[idx]),
+                loss=float(self.loss_series[idx]),
+                prev_trail=self.stop,
+                reset=False,
+            )
+        elif self.is_short:
+            self.stop = update_atr_trail_short(
+                close=float(self.close[idx]),
+                loss=float(self.loss_series[idx]),
+                prev_trail=self.stop,
+                reset=False,
+            )

trade/backtester_/backtester_.py

@@ -11,7 +11,7 @@
 from .data import PTDataset # noqa: F401
 from ._helper import make_bt_wrapper # noqa: F401
 from ._strategy import StrategyBase
-logger = setup_logger('trade.backtester_.backtester_', stream_log_level="DEBUG")
+logger = setup_logger('trade.backtester_.backtester_', stream_log_level="WARNING")
 
 ## TODO: Include Benchmark DD in Portfolio Plot
 ## FIX: After optimization, reset strategy settings to default. Currently, it is not resetting
@@ -89,6 +89,11 @@ def __init__(self,
             This is useful for when you have a strategy that starts at a certain date, which is greater than the earliest date in the dataset.
             Typically to allow buffer to calculate indicators.
         **kwargs: Additional keyword arguments to be passed to the Backtest class in backtesting.py
+            which include: `trade_on_close` (bool): Whether to trade on close or not. Currently, PTBacktester only supports trade_on_close=True. trade_on_close=False is not supported.
+                            `finalize_trades` (bool): Whether to finalize trades or not. This is useful for when you want to keep trades open after the backtest is done. Defaults to True.
+                            `verbose` (bool): Whether to print verbose logs during strategy setup. Defaults to False.
+                            `plot_indicators` (bool): Whether to plot indicators in the backtest plot. Defaults to True.
+                            `commission` (float): Commission to be passed to the Backtest class in backtesting.py. Defaults to 0.0.
 
         Returns:
         None
@@ -98,17 +103,20 @@ def __init__(self,
         - _runIndex: Index of the ticker in the dataset list
 
         """
+        if kwargs.pop("reset", True):
+            for dataset in datalist:
+                dataset.reset()
         trade_on_close = kwargs.pop("trade_on_close", True)
         finalize_trades = kwargs.pop("finalize_trades", True)
         if not trade_on_close:
             raise ValueError("PTBacktester currently only supports trade_on_close=True. trade_on_close=False is not supported.")
         else:
             logger.info(f"PTBacktester initialized with trade_on_close={trade_on_close}, finalize_trades={finalize_trades}")
-        self.datasets = []
+        self.datasets: List[PTDataset] = []
         self.__strategy = deepcopy((_setup_strategy(strategy, 
                                                     start_date=start_overwrite, 
-                                                    verbose=kwargs.get('verbose', False), 
-                                                    plot_indicators=kwargs.get('plot_indicators', True))))
+                                                    verbose=kwargs.pop('verbose', False), 
+                                                    plot_indicators=kwargs.pop('plot_indicators', True))))
         self.__port_stats = None
         self._trades = None
         self._equity = None
@@ -161,7 +169,7 @@ def update_settings(self, datalist) -> None:
                     raise ValueError(f'For datasets settings, please assign a dictionary containing parameters as key and values, got {type(param_setting)}')    
                 dataset_obj = [x for x in datalist if x.name == name][0]
                 dataset_obj.param_settings = param_setting
-            except:
+            except: # noqa
                 ## Use default settings as settings in not given names
                 dataset_obj.param_settings = default_setting
                 no_settings_names.append(name)
@@ -190,7 +198,7 @@ def run(self) -> pd.DataFrame:
         """
         results = []
         for i, d in enumerate(self.datasets):
-            d.backtest._strategy._name = d.name
+            d.backtest._strategy._name = d.name            
             d.backtest._strategy._runIndex = i
             if d.param_settings:
                 # if d.param_settings:
@@ -205,11 +213,11 @@ def run(self) -> pd.DataFrame:
             self.reset_settings() if d.param_settings else None
             try:
                 del d.backtest._strategy._name
-            except: 
+            except: # noqa
                 pass
             try:
                 del d.backtest._strategy._runIndex
-            except:
+            except: # noqa
                 pass
             results.append(stats)
         self.__port_stats = {d.name: results[i] for i, d in enumerate(self.datasets)}
@@ -228,6 +236,30 @@ def pf_value_ts(self) -> pd.DataFrame:
         """
         Returns Timeseries of periodic portfolio value
         """
+
+        ## Initialize empty dataframe to hold equity curves for each ticker.
+        eq = pd.DataFrame()
+
+        ## If cash is a single value, we create a dictionary with the same cash for each ticker to fill in missing values in the equity curve
+        if isinstance(self.cash, (float, int)):
+            cash_per_asset = {tick: self.cash for tick in self.get_port_stats().keys()}
+
+        ## Loop through each ticker's stats, extract the equity curve, and fill in missing values with the initial cash
+        for tick, stats in self.get_port_stats().items():
+            eq[tick] = stats["_equity_curve"]["Equity"]
+
+            ## If cash is a dict, we use the specific cash for that ticker to fill in missing values. If not, we use the single cash value for all tickers.
+            if tick not in cash_per_asset:
+                raise ValueError(f"Cash value for ticker {tick} not found in cash_per_asset. Please provide a cash value for this ticker.")
+            eq[tick].fillna(cash_per_asset[tick], inplace=True)
+        eq["Total"] = eq.sum(axis=1)
+        eq.index = pd.to_datetime(eq.index)
+        if self.start_overwrite:
+            eq = eq[eq.index.date >= pd.to_datetime(self.start_overwrite).date()]
+
+        return eq
+
+
         PortStats = self.__port_stats
         if self.start_overwrite:
             start = pd.to_datetime(self.start_overwrite).date()
@@ -399,12 +431,21 @@ def position_optimize(self,
         default_params = {}
         for param in param_kwargs.keys():
             default_params[param] = getattr(self.strategy, param)
+
 
         ## Loop through each datasets backtest, optimize & append to optimized dataframe
         for dataset in self.datasets:
             name = dataset.name
+
+            # If the dataset has specific settings for the strategy, we set the strategy settings.
+            if dataset.param_settings:
+                print(f"Optimizing for dataset: {name} with settings: {dataset.param_settings}")
+                for setting, value in dataset.param_settings.items():
+                    setattr(self.strategy, setting, value)
             ## Make sure strategy name is set for each optimize run
             dataset.backtest._strategy._name = name
+
+
 
             if return_heatmap:
                 opt, hm = dataset.backtest.optimize(**param_kwargs, **kwargs)

trade/backtester_/data.py

@@ -1,5 +1,6 @@
 import pandas as pd
-
+from backtesting import Backtest
+from typing import Optional
 class PTDataset:
     """
     Custom dataset holding ticker name, ticker timeseries & backtest object from backtesting.py
@@ -14,13 +15,20 @@ def __init__(self, name: str, data: pd.DataFrame, param_settings: dict = None):
         self.__param_settings = param_settings  ## Making param_settings private
         self.name = name
         self.data = data
-        self.backtest = None
+        self.backtest: Optional[Backtest] = None
+        self.data.columns = self.data.columns.str.capitalize()  # Ensure columns are capitalized
+
 
     def __repr__(self):
         return f"PTDataset({self.name})"
 
     def __str__(self):
         return f"PTDataset({self.name})"
+
+    def reset(self):
+        """Reset the dataset's backtest to None"""
+        self.backtest = None
+        self.param_settings = None
 
     @property
     def param_settings(self):
@@ -30,4 +38,6 @@ def param_settings(self):
     @param_settings.setter
     def param_settings(self, value: dict):
         """Setter for param_settings with type checking"""
+        if not isinstance(value, (dict, type(None))):
+            raise TypeError("param_settings must be a dictionary or None")
         self.__param_settings = value