Building a Momentum Strategy with AlphaFlow

Momentum strategies are among the most popular quantitative trading approaches, based on the principle that assets that have performed well recently tend to continue performing well in the near term. In this tutorial, we'll build a momentum strategy from scratch using AlphaFlow.

Strategy Overview

Our momentum strategy will:

1. Track the performance of multiple stocks over a lookback period
2. Identify the top performers (highest returns)
3. Rebalance the portfolio monthly to hold the best momentum stocks
4. Equal-weight the positions across our top picks

Setting Up the Backtest

First, let's import the necessary components and create our flow:

import alphaflow as af
from datetime import datetime

# Create the flow
flow = af.AlphaFlow()

# Set initial capital
flow.set_cash(100000)

# Define our investment universe (S&P 500 technology stocks)
universe = ["AAPL", "MSFT", "GOOGL", "NVDA", "META", "TSLA", "AVGO", "ORCL", "ADBE", "CRM"]
for symbol in universe:
    flow.add_equity(symbol)

# Set SPY as benchmark for comparison
flow.set_benchmark("SPY")

# Set backtest timeframe
flow.set_backtest_start_timestamp(datetime(2020, 1, 1))
flow.set_backtest_end_timestamp(datetime(2024, 12, 31))

Creating a Custom Momentum Strategy

While AlphaFlow provides built-in strategies like BuyAndHoldStrategy, we'll create a custom strategy to implement our momentum logic. Custom strategies in AlphaFlow extend the Strategy class and subscribe to events:

from alphaflow import Strategy
from alphaflow.enums import Topic, Side, OrderType
from alphaflow.events import MarketDataEvent, OrderEvent
from alphaflow.events.event import Event
from collections import defaultdict
from datetime import timedelta

class MomentumStrategy(Strategy):
    def __init__(self, universe, lookback_days=60, rebalance_days=30, top_n=3):
        self.universe = universe
        self.lookback_days = lookback_days
        self.rebalance_days = rebalance_days
        self.top_n = top_n
        self.price_history = defaultdict(list)
        self.last_rebalance = None
        
    def topic_subscriptions(self) -> list[Topic]:
        """Subscribe to market data events"""
        return [Topic.MARKET_DATA]
        
    def read_event(self, event: Event) -> None:
        """Called when new market data arrives"""
        if not isinstance(event, MarketDataEvent):
            return
            
        symbol = event.symbol
        close_price = event.close
        timestamp = event.timestamp
        
        # Store price history for all symbols in our universe
        if symbol in self.universe:
            self.price_history[symbol].append({
                'timestamp': timestamp,
                'price': close_price
            })
        
        # Check if it's time to rebalance (only on events for first symbol to avoid duplicate triggers)
        if symbol == self.universe[0] and self.should_rebalance(timestamp):
            self.rebalance_portfolio(timestamp)
            self.last_rebalance = timestamp
    
    def should_rebalance(self, current_time):
        """Determine if we should rebalance"""
        if self.last_rebalance is None:
            # Need enough data first
            min_data = all(
                len(self.price_history[sym]) >= self.lookback_days 
                for sym in self.universe
            )
            return min_data
        
        days_since_rebalance = (current_time - self.last_rebalance).days
        return days_since_rebalance >= self.rebalance_days
    
    def calculate_momentum(self, symbol, current_time):
        """Calculate momentum score for a symbol"""
        prices = self.price_history[symbol]
        
        if len(prices) < self.lookback_days:
            return None
        
        # Use prices from lookback period ago and current
        lookback_price = prices[-self.lookback_days]['price']
        current_price = prices[-1]['price']
        
        # Calculate return over lookback period
        momentum = (current_price - lookback_price) / lookback_price
        
        return momentum
    
    def rebalance_portfolio(self, current_time):
        """Rebalance to top momentum stocks"""
        # Calculate momentum for all stocks
        momentum_scores = {}
        for symbol in self.universe:
            score = self.calculate_momentum(symbol, current_time)
            if score is not None:
                momentum_scores[symbol] = score
        
        if not momentum_scores:
            return
        
        # Select top N stocks by momentum
        top_stocks = sorted(
            momentum_scores.items(), 
            key=lambda x: x[1], 
            reverse=True
        )[:self.top_n]
        
        # Get current portfolio state
        portfolio = self._alpha_flow.portfolio
        total_value = portfolio.get_portfolio_value(current_time)
        
        # Calculate target position size for each top stock
        target_value_per_stock = total_value / self.top_n
        
        # Generate buy orders for top stocks
        for symbol, score in top_stocks:
            current_position = portfolio.get_position(symbol)
            current_price = self.price_history[symbol][-1]['price']
            target_qty = int(target_value_per_stock / current_price)
            qty_diff = target_qty - current_position
            
            if qty_diff > 0:
                # Need to buy more
                self._alpha_flow.event_bus.publish(
                    Topic.ORDER,
                    OrderEvent(
                        timestamp=current_time,
                        symbol=symbol,
                        side=Side.BUY,
                        qty=qty_diff,
                        order_type=OrderType.MARKET
                    )
                )
            elif qty_diff < 0:
                # Need to sell some
                self._alpha_flow.event_bus.publish(
                    Topic.ORDER,
                    OrderEvent(
                        timestamp=current_time,
                        symbol=symbol,
                        side=Side.SELL,
                        qty=abs(qty_diff),
                        order_type=OrderType.MARKET
                    )
                )
        
        # Close positions not in top stocks
        selected_symbols = [s for s, _ in top_stocks]
        for symbol in self.universe:
            if symbol not in selected_symbols:
                position = portfolio.get_position(symbol)
                if position > 0:
                    self._alpha_flow.event_bus.publish(
                        Topic.ORDER,
                        OrderEvent(
                            timestamp=current_time,
                            symbol=symbol,
                            side=Side.SELL,
                            qty=position,
                            order_type=OrderType.MARKET
                        )
                    )

Adding the Strategy to the Flow

Now we can add our custom momentum strategy to the backtest:

# Create and add the momentum strategy
momentum_strategy = MomentumStrategy(
    universe=universe,
    lookback_days=60,    # 60-day momentum window
    rebalance_days=30,   # Rebalance monthly
    top_n=3              # Hold top 3 momentum stocks
)
flow.add_strategy(momentum_strategy)

Configuring Execution and Analysis

Let's set up the broker and analyzer:

from alphaflow.brokers import SimpleBroker
from alphaflow.analyzers import DefaultAnalyzer

# Set up the broker
flow.set_broker(SimpleBroker())

# Add performance analyzer
flow.add_analyzer(
    DefaultAnalyzer(
        plot_path="momentum_strategy_results.png",
        plot_title="Momentum Strategy Performance"
    )
)

Running the Backtest

Finally, set up the data feed and run the backtest:

from alphaflow.data_feeds import FMPDataFeed

# Set up data feed with FinancialModelingPrep
# API key can be provided directly or via FMP_API_KEY environment variable
flow.set_data_feed(FMPDataFeed(api_key="YOUR_API_KEY"))

# Run the backtest
flow.run()

Understanding the Results

After running the backtest, AlphaFlow will generate a performance report including:

• Total Return: Overall profit/loss percentage
• Sharpe Ratio: Risk-adjusted returns
• Sortino Ratio: Downside risk-adjusted returns
• Maximum Drawdown: Largest peak-to-trough decline
• Comparison vs SPY: How the strategy performed against the benchmark

Key Implementation Details

The momentum strategy demonstrates several important AlphaFlow patterns:

1. Event Subscription: topic_subscriptions() tells AlphaFlow which events to receive
2. Event Processing: read_event() handles incoming market data
3. Event Bus: self._alpha_flow.event_bus.publish() sends order events
4. Portfolio Access: self._alpha_flow.portfolio provides current positions and values
5. Chronological Safety: Events are processed in order, preventing look-ahead bias

Strategy Variations to Try

Once you have the basic momentum strategy working, try these variations:

1. Dual Momentum: Combine relative momentum (vs other stocks) with absolute momentum (vs cash)
2. Volatility Filtering: Exclude stocks with excessive volatility
3. Sector Rotation: Apply momentum within sectors rather than across all stocks
4. Risk Parity: Weight positions by inverse volatility instead of equal weighting

Best Practices

When developing momentum strategies:

• Avoid Overfitting: Don't optimize parameters on the same data you test on
• Account for Costs: Momentum strategies can trade frequently - commissions matter
• Handle Survivorship Bias: Use complete historical data including delisted stocks
• Consider Timing: Monthly rebalancing often works better than daily for momentum
• Test Multiple Periods: Verify the strategy works across different market regimes

Conclusion

AlphaFlow's event-driven architecture makes it straightforward to implement sophisticated strategies like momentum. By subscribing to market data events and generating orders through the event bus, your strategy naturally avoids look-ahead bias and maintains realistic timing.

Next Steps

Ready to build your own strategies? Check out the AlphaFlow documentation for more examples and API details.

Happy trading!