Add cross-sectional momentum strategy

README.md

@@ -36,6 +36,23 @@ Run the SPY example (this step downloads market data):
 python run.py
 ```
 
+Run the cross-sectional sector-ETF example:
+
+```bash
+python run_cross_sectional.py
+```
+
+That experiment ranks nine US sector ETFs using a fixed 12-1 momentum signal,
+holds the top third, and rebalances approximately every 21 trading days. It
+compares the strategy with an equal-weight buy-and-hold basket through the same
+event engine and cost model. Results at 0, 5, 10, and 25 bps are written to
+`data/cross_sectional/`.
+
+The implementation is intentionally long-only. Adding a negative target to a
+long-only portfolio would not constitute a valid short simulation: borrow,
+margin, short proceeds, financing, and forced-liquidation rules must be modeled
+explicitly first.
+
 Outputs are saved under `data/`, including every ledger, the cost-sensitivity table, and an equity chart.
 
 ## Minimal network-free use

docs/METHODOLOGY.md

@@ -12,6 +12,22 @@ Market orders receive adverse slippage: buys pay above the open and sells receiv
 
 The example compares the one included moving-average rule with buy-and-hold. Both start with identical capital and use the same entry commission/slippage assumptions. This is a baseline, not evidence of a profitable strategy.
 
+## Cross-sectional momentum example
+
+`run_cross_sectional.py` uses nine sector ETFs to demonstrate multi-asset target
+weights without individual-stock survivorship selection. At each scheduled
+rebalance close, the strategy ranks trailing returns from approximately
+`t-252` through `t-21`, assigns 90% gross capital equally to the top third, and
+fills changes at the following open. Reductions are submitted before increases;
+conservative expected sale proceeds may reserve replacement buys, while actual
+next-open affordability still determines final fills.
+
+The equal-weight comparison runs through the same portfolio, execution, fee,
+and marking components. The cost table varies adverse slippage but holds the
+commission schedule fixed. This is an integration example rather than a new
+independent momentum discovery test; repeated inspection of the same interval
+must not be described as untouched evaluation.
+
 ## Research discipline
 
 For strategy research beyond the example, split data chronologically into training, validation, and untouched test periods. Choose rules only with training/validation, report the untouched result once, and examine sensitivity across parameters, costs, and market regimes. Walk-forward evaluation should retrain only on information available at each historical date.

engine/portfolio.py

@@ -28,16 +28,28 @@ def __init__(self, data, events, initial_capital=100_000.0, reserve_buffer=0.02)
 
     def _equity(self) -> float:
         market_value = sum(
-            self.positions[symbol] * self.data.get_latest_close(symbol)
-            for symbol in self.symbols
+            self.positions[symbol] * self.data.get_latest_close(symbol) for symbol in self.symbols
         )
         return self.cash + market_value
 
+    def _pending_sell_proceeds(self) -> float:
+        """Conservative proceeds from sells scheduled before pending buys."""
+        return sum(
+            max(
+                0.0,
+                order["quantity"] * order["reference_price"] * (1.0 - self.reserve_buffer) - 1.0,
+            )
+            for order in self.orders
+            if order["status"] == "pending" and order["direction"] == "SELL"
+        )
+
     def update_signal(self, event) -> None:
         symbol = event.symbol
         price = self.data.get_latest_close(symbol)
         equity = self._equity()
-        target_quantity = int((equity * event.target_weight) // price)
+        if not 0.0 <= event.target_weight <= 1.0:
+            raise ValueError("this portfolio supports long-only target weights in [0, 1]")
+        target_quantity = int(equity * event.target_weight / price)
         quantity_delta = target_quantity - self.positions[symbol]
         if quantity_delta == 0:
             return
@@ -48,7 +60,10 @@ def update_signal(self, event) -> None:
         if direction == "SELL":
             quantity = min(quantity, self.positions[symbol])
         else:
-            available = max(0.0, self.cash - self.reserved_cash)
+            available = max(
+                0.0,
+                self.cash + self._pending_sell_proceeds() - self.reserved_cash,
+            )
             estimated_unit_cost = price * (1.0 + self.reserve_buffer)
             quantity = min(quantity, int(max(0.0, available - 1.0) // estimated_unit_cost))
 

engine/strategy.py

@@ -1,7 +1,11 @@
-"""Strategy interface and one deliberately simple moving-average example."""
+"""Strategy interfaces and deliberately small reference strategies."""
 
 from __future__ import annotations
 
+import math
+
+import numpy as np
+
 from .events import SignalEvent
 
 
@@ -38,3 +42,106 @@ def calculate_signals(self, event) -> None:
             if target != self.targets[symbol]:
                 self.events.put(SignalEvent(symbol, event.dt, target))
                 self.targets[symbol] = target
+
+
+class EqualWeightBuyAndHold(Strategy):
+    """Invest once in an equal-weight basket and then leave it untouched."""
+
+    def __init__(self, data, events, gross_allocation=0.90):
+        if not 0 < gross_allocation <= 1:
+            raise ValueError("gross_allocation must be in (0, 1]")
+        self.data = data
+        self.events = events
+        self.weight = gross_allocation / len(data.symbols)
+        self.invested = False
+
+    def calculate_signals(self, event) -> None:
+        if event.type != "MARKET" or self.invested:
+            return
+        for symbol in self.data.symbols:
+            self.events.put(SignalEvent(symbol, event.dt, self.weight))
+        self.invested = True
+
+
+class CrossSectionalMomentum(Strategy):
+    """Periodically hold the strongest assets by trailing return.
+
+    A score observed at close[t] uses prices from t-lookback through t-skip.
+    Target-weight orders fill no earlier than open[t+1].
+    """
+
+    def __init__(
+        self,
+        data,
+        events,
+        lookback=252,
+        skip=21,
+        rebalance_every=21,
+        top_fraction=1 / 3,
+        gross_allocation=0.90,
+    ):
+        if lookback < 1:
+            raise ValueError("lookback must be positive")
+        if not 0 <= skip < lookback:
+            raise ValueError("skip must satisfy 0 <= skip < lookback")
+        if rebalance_every < 1:
+            raise ValueError("rebalance_every must be positive")
+        if not 0 < top_fraction <= 1:
+            raise ValueError("top_fraction must be in (0, 1]")
+        if not 0 < gross_allocation <= 1:
+            raise ValueError("gross_allocation must be in (0, 1]")
+
+        self.data = data
+        self.events = events
+        self.lookback = int(lookback)
+        self.skip = int(skip)
+        self.rebalance_every = int(rebalance_every)
+        self.top_fraction = float(top_fraction)
+        self.gross_allocation = float(gross_allocation)
+        self.targets = {symbol: 0.0 for symbol in data.symbols}
+        self.rebalance_log = []
+
+    def _score(self, symbol):
+        closes = self.data.get_latest_closes(symbol, self.lookback + 1)
+        if len(closes) < self.lookback + 1:
+            return None
+        start = float(closes[0])
+        end = float(closes[-self.skip - 1]) if self.skip else float(closes[-1])
+        score = end / start - 1.0
+        return score if np.isfinite(score) else None
+
+    def calculate_signals(self, event) -> None:
+        if event.type != "MARKET" or self.data.i < self.lookback:
+            return
+        if (self.data.i - self.lookback) % self.rebalance_every:
+            return
+
+        scores = {symbol: score for symbol in self.data.symbols if (score := self._score(symbol)) is not None}
+        if not scores:
+            return
+
+        count = max(1, math.ceil(len(scores) * self.top_fraction))
+        winners = set(sorted(scores, key=lambda symbol: (-scores[symbol], symbol))[:count])
+        weight = self.gross_allocation / len(winners)
+        new_targets = {symbol: weight if symbol in winners else 0.0 for symbol in self.data.symbols}
+
+        # Submit reductions first so their conservative expected proceeds can
+        # fund purchases. The broker preserves this order at the next open.
+        changed = [
+            symbol
+            for symbol in self.data.symbols
+            if not math.isclose(new_targets[symbol], self.targets[symbol], abs_tol=1e-12)
+        ]
+        changed.sort(key=lambda symbol: new_targets[symbol] - self.targets[symbol])
+        for symbol in changed:
+            self.events.put(SignalEvent(symbol, event.dt, new_targets[symbol]))
+
+        self.rebalance_log.append(
+            {
+                "dt": event.dt,
+                "winners": tuple(sorted(winners)),
+                "scores": scores.copy(),
+                "targets": new_targets.copy(),
+            }
+        )
+        self.targets = new_targets

run_cross_sectional.py

@@ -0,0 +1,86 @@
+"""Run a transparent sector-ETF cross-sectional momentum experiment."""
+
+from pathlib import Path
+
+import pandas as pd
+
+from engine.backtest import Backtest
+from engine.data import download_price_data
+from engine.metrics import performance
+from engine.strategy import CrossSectionalMomentum, EqualWeightBuyAndHold
+
+
+SYMBOLS = ["XLB", "XLE", "XLF", "XLI", "XLK", "XLP", "XLU", "XLV", "XLY"]
+CAPITAL = 100_000.0
+
+
+def run_strategy(price_data, strategy_cls, bps, strategy_kwargs):
+    return Backtest(
+        SYMBOLS,
+        price_data=price_data,
+        initial_capital=CAPITAL,
+        strategy_cls=strategy_cls,
+        strategy_kwargs=strategy_kwargs,
+        commission_per_share=0.005,
+        minimum_commission=1.0,
+        slippage_bps=bps,
+    ).run()
+
+
+def main():
+    output = Path("data/cross_sectional")
+    output.mkdir(parents=True, exist_ok=True)
+    price_data = download_price_data(SYMBOLS, "2010-01-01", "2025-01-01")
+
+    strategy_kwargs = {
+        "lookback": 252,
+        "skip": 21,
+        "rebalance_every": 21,
+        "top_fraction": 1 / 3,
+        "gross_allocation": 0.90,
+    }
+    benchmark_kwargs = {"gross_allocation": 0.90}
+    rows = []
+    plotted = {}
+
+    for bps in (0, 5, 10, 25):
+        momentum = run_strategy(
+            price_data,
+            CrossSectionalMomentum,
+            bps,
+            strategy_kwargs,
+        )
+        equal_weight = run_strategy(
+            price_data,
+            EqualWeightBuyAndHold,
+            bps,
+            benchmark_kwargs,
+        )
+        for label, result in (("momentum", momentum), ("equal_weight", equal_weight)):
+            stats, curve = performance(result["equity"], CAPITAL)
+            rows.append({"strategy": label, "cost_bps": bps, **stats, "fills": len(result["fills"])})
+            if bps == 5:
+                plotted[label] = curve["equity"]
+
+        if bps == 5:
+            for name, ledger in momentum.items():
+                ledger.to_csv(output / f"momentum_{name}.csv", index=False)
+            for name, ledger in equal_weight.items():
+                ledger.to_csv(output / f"equal_weight_{name}.csv", index=False)
+
+    sensitivity = pd.DataFrame(rows)
+    sensitivity.to_csv(output / "cost_sensitivity.csv", index=False)
+    ax = pd.DataFrame(plotted).plot(
+        figsize=(11, 6),
+        title="Sector ETFs: 12-1 momentum vs equal-weight buy-and-hold (5 bps)",
+    )
+    ax.set_ylabel("Portfolio value ($)")
+    ax.figure.tight_layout()
+    ax.figure.savefig(output / "equity_comparison.png", dpi=150)
+
+    print(sensitivity.to_string(index=False))
+    print(f"\nSaved results to {output}/")
+
+
+if __name__ == "__main__":
+    main()

tests/test_cross_sectional.py

@@ -0,0 +1,96 @@
+import unittest
+
+import pandas as pd
+
+from engine.backtest import Backtest
+from engine.strategy import CrossSectionalMomentum
+
+
+def bars(closes, start="2024-01-01"):
+    index = pd.date_range(start, periods=len(closes), freq="D")
+    return pd.DataFrame({"Open": closes, "Close": closes}, index=index)
+
+
+class CrossSectionalMomentumTests(unittest.TestCase):
+    def run_bt(self, price_data, **strategy_kwargs):
+        return Backtest(
+            symbols=list(price_data),
+            price_data=price_data,
+            initial_capital=10_000,
+            strategy_cls=CrossSectionalMomentum,
+            strategy_kwargs=strategy_kwargs,
+            commission_per_share=0,
+            minimum_commission=0,
+            slippage_bps=0,
+        ).run()
+
+    def test_selects_top_assets_and_fills_next_bar(self):
+        data = {
+            "A": bars([10, 11, 13, 14, 15, 16]),
+            "B": bars([10, 11, 12, 12, 12, 12]),
+            "C": bars([10, 9, 8, 8, 8, 8]),
+            "D": bars([10, 10, 10, 10, 10, 10]),
+        }
+        result = self.run_bt(
+            data,
+            lookback=3,
+            skip=1,
+            rebalance_every=99,
+            top_fraction=0.5,
+            gross_allocation=0.8,
+        )
+
+        fills = result["fills"]
+        self.assertEqual(set(fills["symbol"]), {"A", "B"})
+        self.assertTrue((fills["submitted_dt"] < fills["fill_dt"]).all())
+        self.assertEqual(set(fills["fill_dt"]), {data["A"].index[4]})
+
+    def test_rotation_sells_before_buy_and_preserves_nonnegative_cash(self):
+        data = {
+            "A": bars([10, 12, 14, 14, 14, 14]),
+            "B": bars([10, 10, 10, 20, 30, 40]),
+        }
+        result = self.run_bt(
+            data,
+            lookback=2,
+            skip=0,
+            rebalance_every=1,
+            top_fraction=0.5,
+            gross_allocation=0.9,
+        )
+
+        fills = result["fills"].reset_index(drop=True)
+        rotation = fills[fills["fill_dt"] == data["A"].index[4]]
+        self.assertEqual(list(rotation["direction"]), ["SELL", "BUY"])
+        self.assertEqual(list(rotation["symbol"]), ["A", "B"])
+        self.assertGreaterEqual(result["cash"]["cash"].min(), 0)
+        final_b = result["positions"].query("symbol == 'B'").iloc[-1]
+        self.assertGreater(final_b["quantity"], 0)
+
+    def test_skip_excludes_most_recent_return_from_rank(self):
+        data = {
+            "STEADY": bars([10, 12, 14, 14, 14]),
+            "JUMP": bars([10, 10, 10, 100, 100]),
+        }
+        result = self.run_bt(
+            data,
+            lookback=3,
+            skip=1,
+            rebalance_every=99,
+            top_fraction=0.5,
+        )
+        self.assertEqual(set(result["fills"]["symbol"]), {"STEADY"})
+
+    def test_rejects_invalid_configuration(self):
+        data = {"A": bars([10, 11]), "B": bars([10, 11])}
+        with self.assertRaises(ValueError):
+            Backtest(
+                symbols=list(data),
+                price_data=data,
+                strategy_cls=CrossSectionalMomentum,
+                strategy_kwargs={"lookback": 2, "skip": 2},
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()