feat: add context.record() for tracking custom variables during backtests

- Add record(**kwargs) method to Context for event-driven backtests
- Add generate_recorded_values() method to TradingStrategy for vectorized backtests
- Add recorded_values field to BacktestRun with serialization support
- Wire recorded values into EventBacktestService and VectorBacktestService
- Add 8 unit tests for record functionality

Closes #455

Author: MDUYN

investing_algorithm_framework/app/context.py

@@ -53,6 +53,7 @@ def __init__(
         self._blotter = None
         self._fx_rate_provider = None
         self._base_currency = None
+        self._recorded_values = {}  # key -> list of (datetime, value)
 
     def _validate_target_symbol(self, target_symbol, market=None):
         """
@@ -2334,3 +2335,64 @@ def get_transactions(self):
             list[Transaction]: Recorded transactions.
         """
         return self._blotter.get_transactions()
+
+    def record(self, **kwargs):
+        """
+        Record arbitrary key-value pairs at the current backtest timestamp.
+
+        This method allows you to store any custom indicator, metric, or
+        variable during a backtest. Each key creates a time series of
+        values that can be retrieved after the backtest completes via
+        ``BacktestRun.recorded_values``.
+
+        The values are stored as a list of ``(datetime, value)`` tuples
+        per key, allowing you to track any indicator over time.
+
+        This method only records during backtesting. In live mode it is
+        a no-op.
+
+        Args:
+            **kwargs: Arbitrary key-value pairs to record. Keys are
+                strings, values can be any type (float, int, str,
+                dict, list, etc.).
+
+        Example::
+
+            def on_run(self, context, data):
+                context.record(
+                    rsi=compute_rsi(data),
+                    sma_20=compute_sma(data, 20),
+                    signal_strength=0.85,
+                )
+        """
+        is_backtest = self.configuration_service.config.get(
+            BACKTESTING_FLAG, False
+        )
+
+        if not is_backtest:
+            return
+
+        current_datetime = self.configuration_service.config.get(
+            INDEX_DATETIME
+        )
+
+        for key, value in kwargs.items():
+            if key not in self._recorded_values:
+                self._recorded_values[key] = []
+            self._recorded_values[key].append((current_datetime, value))
+
+    def get_recorded_values(self):
+        """
+        Get all recorded values from the context.
+
+        Returns:
+            dict: A dictionary mapping keys to lists of
+                ``(datetime, value)`` tuples.
+        """
+        return self._recorded_values
+
+    def clear_recorded_values(self):
+        """
+        Clear all recorded values from the context.
+        """
+        self._recorded_values = {}

investing_algorithm_framework/app/strategy.py

@@ -850,6 +850,32 @@ def generate_scale_out_signals(
         """
         return None
 
+    def generate_recorded_values(
+        self, data: Dict[str, Any]
+    ) -> Union[Dict[str, pd.Series], None]:
+        """
+        Optional method to generate recorded values for vectorized
+        backtesting. Override this to record arbitrary indicators,
+        metrics, or variables as time series during a vectorized
+        backtest.
+
+        Each key in the returned dict becomes a recorded variable
+        with the Series index as timestamps and the Series values
+        as the recorded data.
+
+        This is the vectorized equivalent of calling
+        ``context.record()`` in event-driven backtests.
+
+        Args:
+            data (Dict[str, Any]): The market data for the strategy.
+
+        Returns:
+            Dict[str, Series] | None: A dictionary where keys are
+                variable names and values are pandas Series with the
+                recorded values. Return None to not record anything.
+        """
+        return None
+
     def on_trade_closed(self, context: Context, trade: Trade):
         pass
 

investing_algorithm_framework/domain/backtesting/backtest_run.py

@@ -112,6 +112,7 @@ class BacktestRun:
     metadata: Dict[str, str] = field(default_factory=dict)
     signals: Dict[str, Dict[str, Any]] = field(default_factory=dict)
     signal_events: List[Dict[str, Any]] = field(default_factory=list)
+    recorded_values: Dict[str, List] = field(default_factory=dict)
 
     def to_dict(self) -> dict:
         """
@@ -167,6 +168,16 @@ def ensure_iso(value):
                     "date": ensure_iso(evt["date"])
                 } for evt in self.signal_events
             ],
+            "recorded_values": {
+                key: [
+                    {
+                        "datetime": ensure_iso(entry[0]),
+                        "value": entry[1]
+                    }
+                    for entry in entries
+                ]
+                for key, entries in self.recorded_values.items()
+            },
         }
 
     @staticmethod
@@ -330,10 +341,26 @@ def open(directory_path: Union[str, Path]) -> 'BacktestRun':
                     pass
             signal_events.append(parsed)
 
+        # Parse recorded_values
+        raw_recorded = data.pop("recorded_values", {})
+        recorded_values = {}
+        for key, entries in raw_recorded.items():
+            parsed_entries = []
+            for entry in entries:
+                dt = entry.get("datetime")
+                if isinstance(dt, str):
+                    try:
+                        dt = datetime.fromisoformat(dt)
+                    except (ValueError, TypeError):
+                        pass
+                parsed_entries.append((dt, entry.get("value")))
+            recorded_values[key] = parsed_entries
+
         return BacktestRun(
             backtest_metrics=backtest_metrics,
             signals=signals,
             signal_events=signal_events,
+            recorded_values=recorded_values,
             **data
         )
 

investing_algorithm_framework/infrastructure/services/backtesting/event_backtest_service.py

@@ -112,6 +112,8 @@ def run(
             backtest_date_range=backtest_date_range,
             number_of_runs=event_loop_service.total_number_of_runs,
             risk_free_rate=risk_free_rate,
+            recorded_values=event_loop_service.context
+            .get_recorded_values(),
         )
 
     def generate_schedule(
@@ -175,6 +177,7 @@ def _create_backtest_run(
         backtest_date_range: BacktestDateRange,
         number_of_runs: int,
         risk_free_rate: float,
+        recorded_values: dict = None,
     ) -> BacktestRun:
         """
         Create a BacktestRun from the current state after event loop execution.
@@ -184,6 +187,7 @@ def _create_backtest_run(
             backtest_date_range: The date range of the backtest.
             number_of_runs: Total number of strategy executions.
             risk_free_rate: Risk-free rate for metrics calculation.
+            recorded_values: Optional dict of recorded values from context.
 
         Returns:
             BacktestRun: The completed backtest run with metrics.
@@ -215,6 +219,7 @@ def _create_backtest_run(
             positions=self._position_repository.get_all(
                 {"portfolio": portfolio.id}
             ),
+            recorded_values=recorded_values or {},
         )
 
         # Calculate and add metrics

investing_algorithm_framework/infrastructure/services/backtesting/vector_backtest_service.py

@@ -75,6 +75,9 @@ def run(
         scale_in_signals = strategy.generate_scale_in_signals(data)
         scale_out_signals = strategy.generate_scale_out_signals(data)
 
+        # Generate optional recorded values
+        raw_recorded = strategy.generate_recorded_values(data)
+
         if scale_in_signals is None:
             scale_in_signals = buy_signals
 
@@ -798,6 +801,7 @@ def _partial_close(sym, sym_data, price, date, sell_pct):
             symbols=list(buy_signals.keys()),
             signals=raw_signals,
             signal_events=signal_events,
+            recorded_values=self._convert_recorded_values(raw_recorded),
         )
 
         # Create backtest metrics
@@ -806,6 +810,34 @@ def _partial_close(sym, sym_data, price, date, sell_pct):
         )
         return run
 
+    @staticmethod
+    def _convert_recorded_values(raw_recorded):
+        """
+        Convert recorded values from pandas Series to list-of-tuples format.
+
+        Args:
+            raw_recorded: Dict[str, pd.Series] or None from
+                strategy.generate_recorded_values().
+
+        Returns:
+            Dict[str, List[Tuple[datetime, Any]]]: Converted values.
+        """
+        if raw_recorded is None:
+            return {}
+
+        recorded_values = {}
+        for key, series in raw_recorded.items():
+            entries = []
+            for ts, val in series.items():
+                dt = ts
+                if isinstance(dt, pd.Timestamp):
+                    dt = dt.to_pydatetime()
+                if hasattr(dt, 'tzinfo') and dt.tzinfo is None:
+                    dt = dt.replace(tzinfo=timezone.utc)
+                entries.append((dt, val))
+            recorded_values[key] = entries
+        return recorded_values
+
     @staticmethod
     def get_most_granular_ohlcv_data_source(data_sources):
         """