Momentum and Trend Following at Scale

1import numpy as np
2import pandas as pd
3import matplotlib.pyplot as plt
4
5class TimeSeriesMomentum:
6    """Time-series momentum strategy."""
7    
8    def __init__(self, lookback=252, holding_period=21):
9        self.lookback = lookback
10        self.holding_period = holding_period
11    
12    def compute_signals(self, prices):
13        """
14        Generate momentum signals.
15        
16        Signal = sign(return over lookback period)
17        """
18        returns = prices.pct_change(self.lookback)
19        signals = np.sign(returns)
20        
21        # Replace NaN with 0
22        signals = signals.fillna(0)
23        
24        return signals
25    
26    def backtest(self, prices, transaction_cost=0.001):
27        """
28        Backtest strategy.
29        
30        Args:
31            prices: DataFrame of prices
32            transaction_cost: Cost per trade (bps)
33        """
34        signals = self.compute_signals(prices)
35        returns = prices.pct_change()
36        
37        # Strategy returns
38        strategy_returns = signals.shift(1) * returns
39        
40        # Transaction costs
41        position_changes = signals.diff().abs()
42        costs = position_changes * transaction_cost
43        
44        net_returns = strategy_returns - costs
45        
46        # Cumulative returns
47        cum_returns = (1 + net_returns).cumprod()
48        
49        return {
50            'signals': signals,
51            'returns': net_returns,
52            'cumulative': cum_returns,
53            'sharpe': net_returns.mean() / net_returns.std() * np.sqrt(252)
54        }
55
56# Example: Single asset
57np.random.seed(42)
58n_days = 1000
59
60# Generate trending price series
61trend = np.linspace(0, 0.5, n_days)
62noise = np.cumsum(np.random.normal(0, 0.01, n_days))
63log_prices = trend + noise
64prices = pd.Series(100 * np.exp(log_prices))
65
66# Backtest
67ts_mom = TimeSeriesMomentum(lookback=60, holding_period=21)
68results = ts_mom.backtest(prices)
69
70print("Sharpe ratio:", results['sharpe'])
71
72# Plot
73fig, axes = plt.subplots(2, 1, figsize=(14, 10))
74
75axes[0].plot(prices.index, prices.values, label='Price')
76axes[0].set_ylabel('Price')
77axes[0].set_title('Asset Price')
78axes[0].legend()
79axes[0].grid(True, alpha=0.3)
80
81axes[1].plot(results['cumulative'].index, results['cumulative'].values)
82axes[1].set_xlabel('Time')
83axes[1].set_ylabel('Cumulative Return')
84axes[1].set_title('Time-Series Momentum Strategy')
85axes[1].grid(True, alpha=0.3)
86
87plt.tight_layout()
88plt.show()
89

1class MultiAssetTSMomentum:
2    """Time-series momentum across multiple assets."""
3    
4    def __init__(self, lookback=252):
5        self.lookback = lookback
6    
7    def compute_portfolio_weights(self, prices, vol_target=0.10):
8        """
9        Compute equal-risk portfolio weights.
10        
11        Each asset contributes equally to portfolio volatility.
12        """
13        # Momentum signals
14        returns_lookback = prices.pct_change(self.lookback)
15        signals = np.sign(returns_lookback)
16        
17        # Volatility estimation (exponentially weighted)
18        returns = prices.pct_change()
19        vol = returns.ewm(span=60).std() * np.sqrt(252)
20        
21        # Inverse volatility weights
22        inv_vol = 1 / vol
23        weights = signals * inv_vol
24        
25        # Normalize to target volatility
26        portfolio_vol = np.sqrt((weights**2 * vol**2).sum(axis=1))
27        scaling = vol_target / portfolio_vol
28        
29        weights = weights.multiply(scaling, axis=0)
30        
31        return weights.fillna(0)
32    
33    def backtest(self, prices, vol_target=0.10, transaction_cost=0.001):
34        """Backtest multi-asset strategy."""
35        weights = self.compute_portfolio_weights(prices, vol_target)
36        returns = prices.pct_change()
37        
38        # Portfolio returns
39        portfolio_returns = (weights.shift(1) * returns).sum(axis=1)
40        
41        # Transaction costs
42        weight_changes = weights.diff().abs().sum(axis=1)
43        costs = weight_changes * transaction_cost
44        
45        net_returns = portfolio_returns - costs
46        cum_returns = (1 + net_returns).cumprod()
47        
48        return {
49            'weights': weights,
50            'returns': net_returns,
51            'cumulative': cum_returns,
52            'sharpe': net_returns.mean() / net_returns.std() * np.sqrt(252),
53            'turnover': weight_changes.mean() * 252
54        }
55
56# Example: 5 assets
57n_assets = 5
58n_days = 1000
59
60# Generate correlated trending assets
61correlation = 0.3
62cov_matrix = np.eye(n_assets) * (1 - correlation) + correlation
63
64asset_returns = np.random.multivariate_normal(
65    np.ones(n_assets) * 0.0005,
66    cov_matrix * 0.01**2,
67    n_days
68)
69
70prices_multi = pd.DataFrame(
71    100 * np.exp(np.cumsum(asset_returns, axis=0)),
72    columns=['Asset' + str(i) for i in range(n_assets)]
73)
74
75# Backtest
76multi_ts_mom = MultiAssetTSMomentum(lookback=126)
77multi_results = multi_ts_mom.backtest(prices_multi, vol_target=0.15)
78
79print("Multi-asset Sharpe:", multi_results['sharpe'])
80print("Annual turnover:", multi_results['turnover'])
81
82# Plot
83plt.figure(figsize=(14, 6))
84plt.plot(multi_results['cumulative'])
85plt.xlabel('Time')
86plt.ylabel('Cumulative Return')
87plt.title('Multi-Asset Time-Series Momentum')
88plt.grid(True, alpha=0.3)
89plt.show()
90

1class CrossSectionalMomentum:
2    """Cross-sectional momentum (relative strength)."""
3    
4    def __init__(self, lookback=126, n_long=5, n_short=5):
5        self.lookback = lookback
6        self.n_long = n_long
7        self.n_short = n_short
8    
9    def compute_ranks(self, prices):
10        """Rank assets by past returns."""
11        returns_lookback = prices.pct_change(self.lookback)
12        ranks = returns_lookback.rank(axis=1, ascending=False)
13        return ranks
14    
15    def compute_weights(self, prices):
16        """
17        Long-short portfolio: long winners, short losers.
18        """
19        ranks = self.compute_ranks(prices)
20        n_assets = prices.shape[1]
21        
22        weights = pd.DataFrame(0.0, index=prices.index, columns=prices.columns)
23        
24        # Long top N
25        for idx in ranks.index:
26            top_n = ranks.loc[idx].nsmallest(self.n_long).index
27            weights.loc[idx, top_n] = 1.0 / self.n_long
28            
29            # Short bottom N
30            bottom_n = ranks.loc[idx].nlargest(self.n_short).index
31            weights.loc[idx, bottom_n] = -1.0 / self.n_short
32        
33        return weights
34    
35    def backtest(self, prices, transaction_cost=0.001):
36        """Backtest cross-sectional strategy."""
37        weights = self.compute_weights(prices)
38        returns = prices.pct_change()
39        
40        # Portfolio returns
41        portfolio_returns = (weights.shift(1) * returns).sum(axis=1)
42        
43        # Transaction costs
44        weight_changes = weights.diff().abs().sum(axis=1)
45        costs = weight_changes * transaction_cost
46        
47        net_returns = portfolio_returns - costs
48        cum_returns = (1 + net_returns).cumprod()
49        
50        return {
51            'weights': weights,
52            'returns': net_returns,
53            'cumulative': cum_returns,
54            'sharpe': net_returns.mean() / net_returns.std() * np.sqrt(252)
55        }
56
57# Backtest
58cs_mom = CrossSectionalMomentum(lookback=63, n_long=2, n_short=2)
59cs_results = cs_mom.backtest(prices_multi)
60
61print("Cross-sectional Sharpe:", cs_results['sharpe'])
62
63# Compare strategies
64plt.figure(figsize=(14, 6))
65plt.plot(multi_results['cumulative'], label='Time-Series Momentum')
66plt.plot(cs_results['cumulative'], label='Cross-Sectional Momentum')
67plt.xlabel('Time')
68plt.ylabel('Cumulative Return')
69plt.title('Momentum Strategy Comparison')
70plt.legend()
71plt.grid(True, alpha=0.3)
72plt.show()
73

1class RiskParityMomentum:
2    """Momentum with risk parity weighting."""
3    
4    def __init__(self, lookback=126):
5        self.lookback = lookback
6    
7    def compute_risk_parity_weights(self, returns, signals):
8        """
9        Compute weights such that each asset contributes equally to risk.
10        
11        Uses iterative algorithm to solve for weights.
12        """
13        from scipy.optimize import minimize
14        
15        # Covariance matrix
16        cov = returns.cov() * 252
17        
18        n_assets = len(returns.columns)
19        
20        def risk_budget_objective(weights):
21            """Minimize difference in risk contributions."""
22            portfolio_var = weights @ cov @ weights
23            marginal_contrib = cov @ weights
24            risk_contrib = weights * marginal_contrib
25            
26            # Target: equal risk contribution
27            target = portfolio_var / n_assets
28            return np.sum((risk_contrib - target)**2)
29        
30        # Constraints: weights sum to 1, respect signals
31        constraints = [
32            {'type': 'eq', 'fun': lambda w: np.sum(w) - 1}
33        ]
34        
35        # Bounds: long-only or long-short based on signals
36        bounds = [(0, 1) if s > 0 else (-1, 0) if s < 0 else (0, 0) 
37                  for s in signals]
38        
39        # Initial guess
40        x0 = np.ones(n_assets) / n_assets
41        
42        # Optimize
43        result = minimize(
44            risk_budget_objective,
45            x0,
46            method='SLSQP',
47            bounds=bounds,
48            constraints=constraints
49        )
50        
51        return result.x if result.success else x0
52    
53    def backtest(self, prices):
54        """Backtest risk parity momentum."""
55        returns = prices.pct_change()
56        
57        # Momentum signals
58        returns_lookback = prices.pct_change(self.lookback)
59        signals = np.sign(returns_lookback)
60        
61        # Compute weights
62        weights_list = []
63        
64        for i in range(self.lookback, len(prices)):
65            recent_returns = returns.iloc[i-60:i]  # Use last 60 days for cov
66            current_signals = signals.iloc[i]
67            
68            if recent_returns.shape[0] < 20:
69                weights_list.append(np.zeros(len(prices.columns)))
70                continue
71            
72            w = self.compute_risk_parity_weights(recent_returns, current_signals)
73            weights_list.append(w)
74        
75        # Pad with zeros
76        weights = pd.DataFrame(
77            np.vstack([np.zeros((self.lookback, len(prices.columns))), 
78                      np.array(weights_list)]),
79            index=prices.index,
80            columns=prices.columns
81        )
82        
83        # Portfolio returns
84        portfolio_returns = (weights.shift(1) * returns).sum(axis=1)
85        cum_returns = (1 + portfolio_returns).cumprod()
86        
87        return {
88            'weights': weights,
89            'returns': portfolio_returns,
90            'cumulative': cum_returns,
91            'sharpe': portfolio_returns.mean() / portfolio_returns.std() * np.sqrt(252)
92        }
93
94# Backtest (simplified for speed)
95# rp_mom = RiskParityMomentum(lookback=63)
96# rp_results = rp_mom.backtest(prices_multi)
97# print("Risk parity Sharpe:", rp_results['sharpe'])
98

1class VolatilityTargetedMomentum:
2    """Scale positions to target constant volatility."""
3    
4    def __init__(self, lookback=126, vol_target=0.15, vol_lookback=60):
5        self.lookback = lookback
6        self.vol_target = vol_target
7        self.vol_lookback = vol_lookback
8    
9    def backtest(self, prices):
10        """Backtest with volatility targeting."""
11        returns = prices.pct_change()
12        
13        # Momentum signals
14        returns_lookback = prices.pct_change(self.lookback)
15        signals = np.sign(returns_lookback)
16        
17        # Rolling volatility
18        vol = returns.rolling(self.vol_lookback).std() * np.sqrt(252)
19        
20        # Position sizing: target_vol / realized_vol
21        position_size = self.vol_target / vol
22        position_size = position_size.clip(0, 2)  # Cap at 2x leverage
23        
24        # Combine signals and sizing
25        weights = signals * position_size
26        
27        # Portfolio returns
28        portfolio_returns = (weights.shift(1) * returns).sum(axis=1)
29        cum_returns = (1 + portfolio_returns).cumprod()
30        
31        # Realized volatility
32        realized_vol = portfolio_returns.rolling(252).std() * np.sqrt(252)
33        
34        return {
35            'weights': weights,
36            'returns': portfolio_returns,
37            'cumulative': cum_returns,
38            'realized_vol': realized_vol,
39            'sharpe': portfolio_returns.mean() / portfolio_returns.std() * np.sqrt(252)
40        }
41
42# Backtest
43vol_target_mom = VolatilityTargetedMomentum(lookback=126, vol_target=0.15)
44vol_results = vol_target_mom.backtest(prices_multi)
45
46print("Vol-targeted Sharpe:", vol_results['sharpe'])
47
48# Plot volatility
49plt.figure(figsize=(14, 6))
50plt.plot(vol_results['realized_vol'])
51plt.axhline(y=0.15, color='red', linestyle='--', label='Target vol')
52plt.xlabel('Time')
53plt.ylabel('Realized Volatility')
54plt.title('Portfolio Volatility (Annualized)')
55plt.legend()
56plt.grid(True, alpha=0.3)
57plt.show()
58

1from multiprocessing import Pool
2import itertools
3
4def backtest_single_asset(args):
5    """Backtest single asset (for parallel processing)."""
6    prices, lookback, vol_target = args
7    
8    strategy = VolatilityTargetedMomentum(lookback=lookback, vol_target=vol_target)
9    results = strategy.backtest(pd.DataFrame(prices))
10    
11    return {
12        'sharpe': results['sharpe'],
13        'returns': results['returns']
14    }
15
16def distributed_backtest(prices_dict, lookbacks, vol_targets, n_processes=4):
17    """
18    Backtest multiple assets in parallel.
19    
20    Args:
21        prices_dict: Dict of {asset_name: prices}
22        lookbacks: List of lookback periods to test
23        vol_targets: List of volatility targets
24        n_processes: Number of parallel processes
25    """
26    # Create parameter combinations
27    tasks = []
28    for asset_name, prices in prices_dict.items():
29        for lookback in lookbacks:
30            for vol_target in vol_targets:
31                tasks.append((prices, lookback, vol_target))
32    
33    # Parallel processing
34    with Pool(n_processes) as pool:
35        results = pool.map(backtest_single_asset, tasks)
36    
37    return results
38
39# Example (simplified)
40prices_dict = {
41    'Asset' + str(i): prices_multi.iloc[:, i] 
42    for i in range(prices_multi.shape[1])
43}
44
45lookbacks = [63, 126, 252]
46vol_targets = [0.10, 0.15, 0.20]
47
48# results = distributed_backtest(prices_dict, lookbacks, vol_targets, n_processes=2)
49# print("Completed {} backtests".format(len(results)))
50

1class TransactionCostModel:
2    """Model transaction costs for momentum strategies."""
3    
4    def __init__(self, fixed_cost=0.0001, impact_coef=0.0005):
5        self.fixed_cost = fixed_cost  # Fixed cost per trade
6        self.impact_coef = impact_coef  # Market impact coefficient
7    
8    def compute_costs(self, weights, prices, volumes):
9        """
10        Compute transaction costs.
11        
12        Cost = fixed_cost + impact_coef * sqrt(trade_size / volume)
13        """
14        weight_changes = weights.diff().abs()
15        
16        # Trade sizes in dollars
17        portfolio_value = 1000000  # $1M portfolio
18        trade_sizes = weight_changes * portfolio_value
19        
20        # Market impact (square root model)
21        impact = self.impact_coef * np.sqrt(trade_sizes / volumes)
22        
23        # Total costs
24        total_costs = self.fixed_cost * (weight_changes > 0) + impact
25        
26        return total_costs.sum(axis=1)
27
28# Example usage
29volumes = pd.DataFrame(
30    np.random.uniform(1e6, 10e6, prices_multi.shape),
31    index=prices_multi.index,
32    columns=prices_multi.columns
33)
34
35cost_model = TransactionCostModel(fixed_cost=0.0001, impact_coef=0.0005)
36costs = cost_model.compute_costs(multi_results['weights'], prices_multi, volumes)
37
38print("Average daily cost:", costs.mean())
39print("Annual cost:", costs.sum())
40