ferro-ta/python/ferro_ta/analysis/multitf.py

"""
Multi-timeframe signal utilities.

MultiTimeframeEngine wraps BacktestEngine with a higher-timeframe signal computation step.

Usage:
    from ferro_ta.analysis.multitf import MultiTimeframeEngine

    result = (
        MultiTimeframeEngine(factor=4)          # 4 fine bars per coarse bar
        .with_htf_strategy("rsi_30_70")         # strategy runs on coarse bars
        .with_ohlcv(high=h, low=l, open_=o)
        .with_stop_loss(0.02)
        .run(close_fine)
    )
"""

from __future__ import annotations

import numpy as np
from numpy.typing import ArrayLike

from ferro_ta.analysis.backtest import AdvancedBacktestResult, BacktestEngine
from ferro_ta.analysis.resample import align_to_coarse, resample_ohlcv

__all__ = ["MultiTimeframeEngine"]


class MultiTimeframeEngine:
    """Backtests using signals computed on a higher timeframe (coarser bars).

    Parameters
    ----------
    factor : int
        Number of fine-resolution bars per coarse bar.
    """

    def __init__(self, factor: int) -> None:
        if factor < 1:
            raise ValueError(f"factor must be >= 1, got {factor}")
        self._factor = factor
        self._htf_strategy = "rsi_30_70"
        self._inner = BacktestEngine()

        # Store OHLCV separately so we can resample them
        self._high: np.ndarray | None = None
        self._low: np.ndarray | None = None
        self._open: np.ndarray | None = None

    def with_htf_strategy(self, strategy) -> MultiTimeframeEngine:
        """Set the strategy function or name used on coarse bars."""
        self._htf_strategy = strategy
        return self

    def with_ohlcv(self, *, high, low, open_) -> MultiTimeframeEngine:
        """Store OHLCV data for resampling and pass to inner engine after resampling."""
        self._high = np.asarray(high, dtype=np.float64)
        self._low = np.asarray(low, dtype=np.float64)
        self._open = np.asarray(open_, dtype=np.float64)
        return self

    def with_stop_loss(self, pct: float) -> MultiTimeframeEngine:
        self._inner.with_stop_loss(pct)
        return self

    def with_take_profit(self, pct: float) -> MultiTimeframeEngine:
        self._inner.with_take_profit(pct)
        return self

    def with_trailing_stop(self, pct: float) -> MultiTimeframeEngine:
        self._inner.with_trailing_stop(pct)
        return self

    def with_commission(self, rate: float) -> MultiTimeframeEngine:
        self._inner.with_commission(rate)
        return self

    def with_commission_model(self, model) -> MultiTimeframeEngine:
        self._inner.with_commission_model(model)
        return self

    def with_slippage(self, bps: float) -> MultiTimeframeEngine:
        self._inner.with_slippage(bps)
        return self

    def with_initial_capital(self, capital: float) -> MultiTimeframeEngine:
        self._inner.with_initial_capital(capital)
        return self

    def with_fill_mode(self, mode: str) -> MultiTimeframeEngine:
        self._inner.with_fill_mode(mode)
        return self

    def with_leverage(
        self, margin_ratio: float, margin_call_pct: float = 0.5
    ) -> MultiTimeframeEngine:
        self._inner.with_leverage(margin_ratio, margin_call_pct)
        return self

    def with_loss_limits(
        self, daily: float = 0.0, total: float = 0.0
    ) -> MultiTimeframeEngine:
        self._inner.with_loss_limits(daily, total)
        return self

    def run(
        self, close_fine: ArrayLike, **htf_strategy_kwargs
    ) -> AdvancedBacktestResult:
        """Run multi-timeframe backtest.

        1. Resample close_fine (and stored OHLCV) to coarse bars
        2. Run htf_strategy on coarse close to get coarse signals
        3. Align coarse signals back to fine resolution (repeat each coarse signal `factor` times)
        4. Run BacktestEngine on fine bars with aligned signals

        Parameters
        ----------
        close_fine : array-like
            Fine-resolution close prices.
        **htf_strategy_kwargs
            Extra keyword arguments passed to the HTF strategy.

        Returns
        -------
        AdvancedBacktestResult
        """
        c_fine = np.asarray(close_fine, dtype=np.float64)
        n_fine = len(c_fine)
        factor = self._factor

        # ------------------------------------------------------------------
        # 1. Resample close to coarse resolution
        # ------------------------------------------------------------------
        # Build dummy OHLCV if OHLCV not provided
        if self._high is not None and self._low is not None and self._open is not None:
            coarse_o, coarse_h, coarse_l, coarse_c, _ = resample_ohlcv(
                self._open,
                self._high,
                self._low,
                c_fine,
                np.ones(n_fine),  # volume placeholder
                factor,
            )
        else:
            coarse_o, coarse_h, coarse_l, coarse_c, _ = resample_ohlcv(
                c_fine,
                c_fine,
                c_fine,
                c_fine,
                np.ones(n_fine),
                factor,
            )

        # ------------------------------------------------------------------
        # 2. Compute coarse-bar signals via htf_strategy
        # ------------------------------------------------------------------
        from ferro_ta.analysis.backtest import _resolve_strategy

        strategy_fn = _resolve_strategy(self._htf_strategy)
        # Ensure the coarse close array is C-contiguous (required by Rust kernels)
        coarse_c = np.ascontiguousarray(coarse_c, dtype=np.float64)
        coarse_signals = np.asarray(
            strategy_fn(coarse_c, **htf_strategy_kwargs), dtype=np.float64
        )

        # ------------------------------------------------------------------
        # 3. Align coarse signals back to fine resolution
        # ------------------------------------------------------------------
        aligned_signals = align_to_coarse(coarse_signals, factor, n_fine)

        # ------------------------------------------------------------------
        # 4. Set up OHLCV on inner engine if provided and run
        # ------------------------------------------------------------------
        if self._high is not None and self._low is not None and self._open is not None:
            self._inner.with_ohlcv(
                high=self._high,
                low=self._low,
                open_=self._open,
            )

        # Use a passthrough lambda so the already-computed aligned_signals are used
        return self._inner.run(
            c_fine,
            strategy=lambda c, **kw: aligned_signals,
        )