remove datetime and only use pandas Timestamp class (#19)

Author: adrianhasse

.devcontainer/devcontainer.json

@@ -0,0 +1,45 @@
+{
+  "name": "kissbt dev",
+  "image": "mcr.microsoft.com/devcontainers/python:3.13",
+  "postCreateCommand": "pip install -e .[dev]",
+  "customizations": {
+    "vscode": {
+      "extensions": [
+        "ms-python.python",
+        "ms-python.vscode-pylance",
+        "ms-toolsai.jupyter",
+        "github.copilot",
+        "github.copilot-chat",
+        "charliermarsh.ruff",
+        "ms-python.mypy-type-checker"
+      ],
+      "settings": {
+        "python.defaultInterpreterPath": "/usr/local/bin/python",
+        "[python]": {
+          "editor.defaultFormatter": "charliermarsh.ruff",
+          "editor.formatOnSave": true,
+          "editor.codeActionsOnSave": {
+            "source.fixAll.ruff": "explicit",
+            "source.organizeImports.ruff": "explicit"
+          }
+        },
+        "ruff.enable": true,
+        "ruff.organizeImports": true,
+        "python.analysis.typeCheckingMode": "strict",
+        "python.analysis.autoImportCompletions": true,
+        "python.analysis.indexing": true,
+        "mypy.enabled": true,
+        "mypy.runUsingActiveInterpreter": true,
+        "python.testing.pytestEnabled": true,
+        "python.testing.unittestEnabled": false,
+        "python.testing.pytestArgs": ["tests"],
+        "files.exclude": {
+          "**/__pycache__": true,
+          "**/*.pyc": true,
+          "**/.mypy_cache": true,
+          "**/.ruff_cache": true
+        }
+      }
+    }
+  }
+}

examples/20241227_introduction.ipynb

@@ -23,13 +23,13 @@ def __init__(self, broker: Broker, bar_size: str = "1D") -> None:
         Parameters:
             broker (Broker): The broker instance containing the trading history.
             bar_size (str): The time interval of each bar in the data, supported units
-                are 'S' for seconds, 'M' for minutes, 'H' for hours and 'D' for days
+                are 'S' for seconds, 'T' for minutes, 'H' for hours and 'D' for days
                 (default is "1D").
         """
 
         value = int(bar_size[:-1])
         unit = bar_size[-1]
-        seconds_multiplier = {"S": 1, "M": 60, "H": 3600, "D": 3600 * 6.5}
+        seconds_multiplier = {"S": 1, "T": 60, "H": 3600, "D": 3600 * 6.5}
         if unit not in seconds_multiplier:
             raise ValueError(f"Unsupported bar size unit: {unit}")
         self.seconds_per_bar = value * seconds_multiplier[unit]
@@ -280,14 +280,14 @@ def plot_drawdowns(self, **kwargs: Dict[str, Any]) -> None:
             **kwargs(Dict[str, Any]): Additional keyword arguments to pass to the plot
                 function of pandas.
         """
-        columns_to_plot = ["date", "drawdown"]
+        columns_to_plot = ["timestamp", "drawdown"]
         if "benchmark_drawdown" in self.analysis_df.columns:
             columns_to_plot.append("benchmark_drawdown")
 
         self.analysis_df.loc[:, columns_to_plot].plot(
-            x="date",
+            x="timestamp",
             title="Portfolio Drawdown Over Time",
-            xlabel="Date",
+            xlabel="Timestamp",
             ylabel="Drawdown %",
             **kwargs,
         )
@@ -299,24 +299,24 @@ def plot_equity_curve(self, logy: bool = False, **kwargs: Dict[str, Any]) -> Non
         This method creates a line plot showing the portfolio's cash, total value, and
         benchmark over time for comparison. If the benchmark is not available, it will
         plot only the available columns. If logy is True, the cash column will be
-        excluded, to focus on the total value and benchmark on logaritmic scale.
+        excluded, to focus on the total value and benchmark on logarithmic scale.
 
         Parameters:
             logy (bool): If True, use a logarithmic scale for the y-axis and exclude the
                 cash column.
             **kwargs(Dict[str, Any]): Additional keyword arguments to pass to the plot
                 function of pandas.
         """
-        columns_to_plot = ["date", "total_value"]
+        columns_to_plot = ["timestamp", "total_value"]
         if "benchmark" in self.analysis_df.columns:
             columns_to_plot.append("benchmark")
         if not logy:
             columns_to_plot.append("cash")
 
         self.analysis_df.loc[:, columns_to_plot].plot(
-            x="date",
+            x="timestamp",
             title="Portfolio Equity Curve Over Time",
-            xlabel="Date",
+            xlabel="Timestamp",
             ylabel="Value",
             logy=logy,
             **kwargs,

kissbt/analyzer.py

@@ -1,5 +1,4 @@
-from datetime import datetime
-from typing import Dict, List, Optional, cast
+from typing import Dict, List, Optional
 
 import pandas as pd
 
@@ -24,7 +23,7 @@ class Broker:
     Typical usage:
         broker = Broker(start_capital=100000, fees=0.001)
         broker.place_order(Order("AAPL", 100, OrderType.OPEN))
-        broker.update(next_bar, next_datetime)
+        broker.update(next_bar, next_timestamp)
     """
 
     def __init__(
@@ -64,9 +63,9 @@ def __init__(
         self._open_orders: List[Order] = []
 
         self._current_bar: pd.DataFrame = pd.DataFrame()
-        self._current_datetime: Optional[datetime] = None
+        self._current_timestamp: Optional[pd.Timestamp] = None
         self._previous_bar: pd.DataFrame = pd.DataFrame()
-        self._previous_datetime: Optional[datetime] = None
+        self._previous_timestamp: Optional[pd.Timestamp] = None
 
         self._long_only = long_only
         self._short_fee_rate = short_fee_rate
@@ -75,8 +74,8 @@ def __init__(
         self._benchmark = benchmark
         self._benchmark_size = 0.0
 
-        self._history: Dict[str, List[float]] = {
-            "date": [],
+        self._history: Dict[str, List[float | int | pd.Timestamp]] = {
+            "timestamp": [],
             "cash": [],
             "long_position_value": [],
             "short_position_value": [],
@@ -90,7 +89,7 @@ def _update_history(self):
         """
         Updates the history dictionary with the current portfolio state.
         """
-        self._history["date"].append(self._current_datetime)
+        self._history["timestamp"].append(self._current_timestamp)
         self._history["cash"].append(self._cash)
         self._history["long_position_value"].append(self.long_position_value)
         self._history["short_position_value"].append(self.short_position_value)
@@ -164,7 +163,7 @@ def _get_price_for_order(self, order: Order, bar: pd.DataFrame) -> float | None:
             raise ValueError(f"Unknown order type {order.order_type}")
 
     def _update_closed_positions(
-        self, ticker: str, size: float, price: float, datetime: datetime
+        self, ticker: str, size: float, price: float, timestamp: pd.Timestamp
     ):
         """
         Updates the list of closed positions for a given trade.
@@ -178,7 +177,7 @@ def _update_closed_positions(
             ticker (str): The ticker symbol of the position
             size (float): Position size (positive for long, negative for short)
             price (float): The current closing/reduction price
-            datetime (datetime): Timestamp of the closing/reduction
+            timestamp (timestamp): Timestamp of the closing/reduction
         """
         if (
             ticker in self._open_positions
@@ -191,9 +190,9 @@ def _update_closed_positions(
                         self._open_positions[ticker].ticker,
                         min(self._open_positions[ticker].size, abs(size)),
                         self._open_positions[ticker].price,
-                        self._open_positions[ticker].datetime,
+                        self._open_positions[ticker].timestamp,
                         price,
-                        datetime,
+                        timestamp,
                     ),
                 )
             # if short position is closed/reduced
@@ -203,59 +202,60 @@ def _update_closed_positions(
                         self._open_positions[ticker].ticker,
                         max(self._open_positions[ticker].size, -size),
                         price,
-                        datetime,
+                        timestamp,
                         self._open_positions[ticker].price,
-                        self._open_positions[ticker].datetime,
+                        self._open_positions[ticker].timestamp,
                     ),
                 )
 
     def _update_open_positions(
-        self, ticker: str, size: float, price: float, datetime: datetime
+        self, ticker: str, size: float, price: float, timestamp: pd.Timestamp
     ):
         """
         Updates the open positions for a given ticker.
 
         If the ticker already exists in the open positions, it updates the size, price,
-        and datetime based on the new transaction. If the size of the position becomes
+        and timestamp based on the new transaction. If the size of the position becomes
         zero, the position is removed. If the ticker does not exist, a new open position
         is created.
 
         Args:
             ticker (str): The ticker symbol of the asset.
             size (float): The size of the position.
             price (float): The price at which the position was opened or updated.
-            datetime (datetime): The datetime when the position was opened or updated.
+            timestamp (Timestamp): The timestamp when the position was opened or
+                updated.
         """
         if ticker in self._open_positions:
             if size + self._open_positions[ticker].size == 0.0:
                 self._open_positions.pop(ticker)
             else:
                 open_position_size = self._open_positions[ticker].size + size
                 open_position_price = price
-                open_position_datetime = datetime
+                open_position_timestamp = timestamp
 
                 if size * self._open_positions[ticker].size > 0.0:
                     open_position_price = (
                         self._open_positions[ticker].size
                         * self._open_positions[ticker].price
                         + size * price
                     ) / (self._open_positions[ticker].size + size)
-                    open_position_datetime = self._open_positions[ticker].datetime
+                    open_position_timestamp = self._open_positions[ticker].timestamp
                 elif abs(self._open_positions[ticker].size) > abs(size):
-                    open_position_datetime = self._open_positions[ticker].datetime
+                    open_position_timestamp = self._open_positions[ticker].timestamp
                     open_position_price = self._open_positions[ticker].price
                 self._open_positions[ticker] = OpenPosition(
                     ticker,
                     open_position_size,
                     open_position_price,
-                    open_position_datetime,
+                    open_position_timestamp,
                 )
         else:
             self._open_positions[ticker] = OpenPosition(
                 ticker,
                 size,
                 price,
-                datetime,
+                timestamp,
             )
 
     def _update_cash(self, order: Order, price: float):
@@ -272,29 +272,29 @@ def _update_cash(self, order: Order, price: float):
         else:
             self._cash -= order.size * price * (1.0 - self._fees)
 
-    def _check_long_only_condition(self, order: Order, datetime: datetime):
+    def _check_long_only_condition(self, order: Order, timestamp: pd.Timestamp):
         size = order.size
         if order.ticker in self._open_positions:
             size += self._open_positions[order.ticker].size
 
         if size < 0.0:
             raise ValueError(
-                f"Short selling is not allowed for {order.ticker} on {datetime}."
+                f"Short selling is not allowed for {order.ticker} on {timestamp}."
             )
 
     def _execute_order(
         self,
         order: Order,
         bar: pd.DataFrame,
-        datetime: datetime,
+        timestamp: pd.Timestamp,
     ) -> bool:
         """
-        Executes an order based on the provided bar data and datetime.
+        Executes an order based on the provided bar data and timestamp.
 
         Args:
             order (Order): The order to be executed.
             bar (pd.DataFrame): The bar data containing price information.
-            datetime (datetime): The datetime at which the order is executed.
+            timestamp (Timestamp): The timestamp at which the order is executed.
 
         Returns:
             bool: True if the order was successfully executed, False otherwise.
@@ -308,7 +308,7 @@ def _execute_order(
             return False
 
         if self._long_only:
-            self._check_long_only_condition(order, datetime)
+            self._check_long_only_condition(order, timestamp)
 
         price = self._get_price_for_order(order, bar)
 
@@ -319,16 +319,16 @@ def _execute_order(
         # update cash for long and short positions
         self._update_cash(order, price)
 
-        self._update_closed_positions(ticker, order.size, price, datetime)
+        self._update_closed_positions(ticker, order.size, price, timestamp)
 
-        self._update_open_positions(ticker, order.size, price, datetime)
+        self._update_open_positions(ticker, order.size, price, timestamp)
 
         return True
 
     def update(
         self,
         next_bar: pd.DataFrame,
-        next_datetime: pd.Timestamp,
+        next_timestamp: pd.Timestamp,
     ):
         """
         Updates the broker's state with the next trading bar and executes pending
@@ -344,7 +344,7 @@ def update(
         Args:
             next_bar (pd.DataFrame): The next trading bar data containing at minimum
                 'close' prices for assets
-            next_datetime (pd.Timestamp): The timestamp for the next trading bar
+            next_timestamp (pd.Timestamp): The timestamp for the next trading bar
 
         Notes:
             - Short fees are calculated using the current bar's closing price
@@ -354,9 +354,9 @@ def update(
             - Good-till-cancel orders that aren't filled are retained for the next bar
         """
         self._previous_bar = self._current_bar
-        self._previous_datetime = self._current_datetime
+        self._previous_timestamp = self._current_timestamp
         self._current_bar = next_bar
-        self._current_datetime = next_datetime
+        self._current_timestamp = next_timestamp
 
         # consider short fees
         if not self._long_only:
@@ -384,7 +384,7 @@ def update(
                 self._execute_order(
                     Order(ticker, -self._open_positions[ticker].size, OrderType.CLOSE),
                     self._previous_bar,
-                    cast(datetime, self._previous_datetime),
+                    self._previous_timestamp,
                 )
 
         # buy and sell assets
@@ -396,18 +396,18 @@ def update(
             if open_order.ticker in ticker_not_available:
                 if open_order.size > 0:
                     print(
-                        f"{open_order.ticker} could not be bought on {self._current_datetime}."  # noqa: E501
+                        f"{open_order.ticker} could not be bought on {self._current_timestamp}."  # noqa: E501
                     )
                 else:
                     print(
-                        f"{open_order.ticker} could not be sold on {self._current_datetime}."  # noqa: E501
+                        f"{open_order.ticker} could not be sold on {self._current_timestamp}."  # noqa: E501
                     )
                 continue
             if (
                 not self._execute_order(
                     open_order,
                     self._current_bar,
-                    cast(datetime, self._current_datetime),
+                    self._current_timestamp,
                 )
                 and open_order.good_till_cancel
             ):
@@ -439,7 +439,7 @@ def liquidate_positions(self):
             self._execute_order(
                 Order(ticker, -self._open_positions[ticker].size, OrderType.CLOSE),
                 self._current_bar,
-                self._current_datetime,
+                self._current_timestamp,
             )
 
     def place_order(self, order: Order):
@@ -524,7 +524,7 @@ def open_positions(self) -> Dict[str, OpenPosition]:
         - ticker: Financial instrument identifier
         - size: Position size (positive=long, negative=short)
         - price: Average entry price
-        - datetime: Position opening timestamp
+        - timestamp: Position opening timestamp
 
         Returns:
             Dict[str, OpenPosition]: Dictionary mapping ticker symbols to positions.