Skip to content

Strategies

BuyAndHoldStrategy

BuyAndHoldStrategy

Bases: Strategy

A buy-and-hold rebalancing strategy that maintains a target portfolio weight.

This strategy monitors portfolio value and rebalances positions to maintain the target weight, with optional thresholds to avoid excessive trading.

Source code in alphaflow/strategies/buy_and_hold_strategy.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class BuyAndHoldStrategy(Strategy):
    """A buy-and-hold rebalancing strategy that maintains a target portfolio weight.

    This strategy monitors portfolio value and rebalances positions to maintain
    the target weight, with optional thresholds to avoid excessive trading.
    """

    def __init__(
        self,
        symbol: str,
        target_weight: float,
        min_dollar_delta: float = 0,
        min_share_delta: float = 0,
        share_quantization: float | None = None,
    ) -> None:
        """Initialize the buy-and-hold strategy.

        Args:
            symbol: The ticker symbol to trade.
            target_weight: Target portfolio weight (0.0 to 1.0).
            min_dollar_delta: Minimum dollar difference to trigger rebalancing.
            min_share_delta: Minimum share difference to trigger rebalancing.
            share_quantization: Round shares to multiples of this value (e.g., 1 for whole shares).

        """
        self.symbol = symbol
        self.target_weight = target_weight
        self.min_dollar_delta = min_dollar_delta
        self.min_share_delta = min_share_delta
        self.share_quantization = share_quantization

    def topic_subscriptions(self) -> list[Topic]:
        """Return the topics this strategy subscribes to.

        Returns:
            List of topics this strategy listens to.

        """
        return [Topic.MARKET_DATA]

    def read_event(self, event: Event) -> None:
        """Process market data events and generate rebalancing orders.

        Args:
            event: The market data event containing price information.

        """
        # Type narrowing - we only get MarketDataEvent from Topic.MARKET_DATA
        if not isinstance(event, MarketDataEvent):
            return

        if event.symbol != self.symbol:
            return
        if self._alpha_flow.backtest_start_timestamp and event.timestamp < self._alpha_flow.backtest_start_timestamp:
            return
        if self._alpha_flow.backtest_end_timestamp and event.timestamp > self._alpha_flow.backtest_end_timestamp:
            return

        portfolio_value = self._alpha_flow.portfolio.get_portfolio_value(event.timestamp)
        position_value = self._alpha_flow.portfolio.get_position_value(self.symbol, event.timestamp)
        target_value = portfolio_value * self.target_weight
        purchase_value = target_value - position_value
        if abs(purchase_value) < self.min_dollar_delta:
            # Too small of a value difference
            return
        shares_needed = purchase_value / event.close

        if self.share_quantization:
            shares_needed = math.floor(shares_needed / self.share_quantization) * self.share_quantization

        if abs(shares_needed) < self.min_share_delta:
            # Too small of a share purchase
            return
        side = Side.BUY if shares_needed > 0 else Side.SELL
        self._alpha_flow.event_bus.publish(
            Topic.ORDER,
            OrderEvent(
                timestamp=event.timestamp,
                symbol=self.symbol,
                side=side,
                qty=abs(shares_needed),
                order_type=OrderType.MARKET,
            ),
        )

__init__(symbol, target_weight, min_dollar_delta=0, min_share_delta=0, share_quantization=None)

Initialize the buy-and-hold strategy.

Parameters:

Name Type Description Default
symbol str

The ticker symbol to trade.

 required
target_weight float

Target portfolio weight (0.0 to 1.0).

 required
min_dollar_delta float

Minimum dollar difference to trigger rebalancing.

 0
min_share_delta float

Minimum share difference to trigger rebalancing.

 0
share_quantization float | None

Round shares to multiples of this value (e.g., 1 for whole shares).

 None
Source code in alphaflow/strategies/buy_and_hold_strategy.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def __init__(
    self,
    symbol: str,
    target_weight: float,
    min_dollar_delta: float = 0,
    min_share_delta: float = 0,
    share_quantization: float | None = None,
) -> None:
    """Initialize the buy-and-hold strategy.

    Args:
        symbol: The ticker symbol to trade.
        target_weight: Target portfolio weight (0.0 to 1.0).
        min_dollar_delta: Minimum dollar difference to trigger rebalancing.
        min_share_delta: Minimum share difference to trigger rebalancing.
        share_quantization: Round shares to multiples of this value (e.g., 1 for whole shares).

    """
    self.symbol = symbol
    self.target_weight = target_weight
    self.min_dollar_delta = min_dollar_delta
    self.min_share_delta = min_share_delta
    self.share_quantization = share_quantization

topic_subscriptions()

Return the topics this strategy subscribes to.

Returns:

Type Description
list[Topic]

List of topics this strategy listens to.
Source code in alphaflow/strategies/buy_and_hold_strategy.py 
45
46
47
48
49
50
51
52
def topic_subscriptions(self) -> list[Topic]:
    """Return the topics this strategy subscribes to.

    Returns:
        List of topics this strategy listens to.

    """
    return [Topic.MARKET_DATA]

read_event(event)

Process market data events and generate rebalancing orders.

Parameters:

Name Type Description Default
event Event

The market data event containing price information.

 required
Source code in alphaflow/strategies/buy_and_hold_strategy.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def read_event(self, event: Event) -> None:
    """Process market data events and generate rebalancing orders.

    Args:
        event: The market data event containing price information.

    """
    # Type narrowing - we only get MarketDataEvent from Topic.MARKET_DATA
    if not isinstance(event, MarketDataEvent):
        return

    if event.symbol != self.symbol:
        return
    if self._alpha_flow.backtest_start_timestamp and event.timestamp < self._alpha_flow.backtest_start_timestamp:
        return
    if self._alpha_flow.backtest_end_timestamp and event.timestamp > self._alpha_flow.backtest_end_timestamp:
        return

    portfolio_value = self._alpha_flow.portfolio.get_portfolio_value(event.timestamp)
    position_value = self._alpha_flow.portfolio.get_position_value(self.symbol, event.timestamp)
    target_value = portfolio_value * self.target_weight
    purchase_value = target_value - position_value
    if abs(purchase_value) < self.min_dollar_delta:
        # Too small of a value difference
        return
    shares_needed = purchase_value / event.close

    if self.share_quantization:
        shares_needed = math.floor(shares_needed / self.share_quantization) * self.share_quantization

    if abs(shares_needed) < self.min_share_delta:
        # Too small of a share purchase
        return
    side = Side.BUY if shares_needed > 0 else Side.SELL
    self._alpha_flow.event_bus.publish(
        Topic.ORDER,
        OrderEvent(
            timestamp=event.timestamp,
            symbol=self.symbol,
            side=side,
            qty=abs(shares_needed),
            order_type=OrderType.MARKET,
        ),
    )