fix(backtest): self-finance entries and include initial MDD high-water

src/crypto_strategies/backtest/live_pool_simulator.py

@@ -53,7 +53,8 @@ def _performance_metrics(
     cagr = float((equity.iloc[-1]) ** (1.0 / years) - 1.0) if equity.iloc[-1] > 0 else 0.0
     vol = float(clean.std(ddof=0) * np.sqrt(365.25))
     sharpe = float((clean.mean() * 365.25) / vol) if vol > 0 else 0.0
-    drawdown = float((equity / equity.cummax() - 1.0).min())
+    # High-water must include the unobserved initial equity of 1.0.
+    drawdown = float((equity / equity.cummax().clip(lower=1.0) - 1.0).min())
     win_rate = float((clean > 0).mean())
     return {
         "CAGR": cagr,
@@ -67,6 +68,39 @@ def _performance_metrics(
     }
 
 
+def _rebalance_holdings(
+    shares: pd.Series,
+    cash: float,
+    prices: pd.Series,
+    targets: pd.Series,
+    *,
+    cost_rate: float,
+) -> tuple[pd.Series, float, float, float, float]:
+    """Fill pre-fee targets, selling first and budgeting buy costs from cash.
+
+    Fractional buys scale together when cash is insufficient so entry fees cannot
+    self-finance. Returns shares, cash, total costs, sale notional, buy notional.
+    """
+    needed = (shares > 0.0) | (targets > 0.0)
+    if any(not math.isfinite(price) or price <= 0.0 for price in prices[needed]):
+        raise ValueError("required open prices must be finite and positive")
+    safe_prices = prices.where(needed, 1.0)
+    values = shares * safe_prices
+    equity = cash + float(values.sum())
+    delta = targets * equity - values
+    sells = -delta.clip(upper=0.0)
+    buys = delta.clip(lower=0.0)
+    sale_notional = float(sells.sum())
+    available_cash = cash + sale_notional * (1.0 - cost_rate)
+    desired_buys = float(buys.sum())
+    if desired_buys > 0.0:
+        buys *= min(1.0, available_cash / (desired_buys * (1.0 + cost_rate)))
+    purchase_notional = float(buys.sum())
+    costs = (sale_notional + purchase_notional) * cost_rate
+    cash = max(0.0, available_cash - purchase_notional * (1.0 + cost_rate))
+    return (values - sells + buys) / safe_prices, cash, costs, sale_notional, purchase_notional
+
+
 def run_live_pool_rotation_backtest(
     panel: pd.DataFrame,
     *,
@@ -83,7 +117,8 @@ def run_live_pool_rotation_backtest(
 
     A score observed on ``signal_date`` is tradable at ``effective_date`` after
     ``signal_lag`` rows; returns are measured from that effective open to the
-    next open. Costs are charged on half-L1 turnover at each rebalance.
+    next open. Rebalances use a cash/share ledger so entry costs constrain the
+    affordable notional instead of self-financing full target weights.
     Required execution and valuation opens must be finite and positive; missing
     prices on unexposed assets do not invalidate a cash or invested period.
     """
@@ -113,6 +148,12 @@ def run_live_pool_rotation_backtest(
     signal_lag = int(signal_lag_days)
     fee_bps = float(fee_bps)
     slippage_bps = float(slippage_bps)
+    fee_rate_value = fee_bps / 10_000.0
+    slippage_rate = slippage_bps / 10_000.0
+    cost_rate = fee_rate_value + slippage_rate
+    if not math.isfinite(cost_rate) or not 0.0 <= cost_rate < 1.0:
+        raise ValueError("transaction cost must be less than 1.0")
+
     dates = sorted(panel.index.get_level_values("date").unique())
     if dates and not pd.DatetimeIndex(dates).equals(pd.date_range(dates[0], dates[-1], freq="D")):
         raise ValueError("panel dates must be consecutive calendar days")
@@ -128,7 +169,9 @@ def run_live_pool_rotation_backtest(
         .astype(float)
     )
 
-    portfolio_weights = pd.Series(0.0, index=symbols, dtype=float)
+    shares = pd.Series(0.0, index=symbols, dtype=float)
+    cash = 1.0
+    equity = 1.0
     daily_returns: list[float] = []
     daily_turnover: list[float] = []
     daily_fees: list[float] = []
@@ -140,10 +183,11 @@ def run_live_pool_rotation_backtest(
     ):
         signal_idx = effective_idx - signal_lag
         signal_date = dates[signal_idx]
-        held = portfolio_weights.ne(0.0)
         turnover = 0.0
         fee = 0.0
         slippage = 0.0
+        current_prices = open_matrix.loc[effective_date]
+
         if signal_idx % rebalance_every == 0:
             snapshot = panel.xs(signal_date, level="date")
             ranked = (
@@ -156,15 +200,21 @@ def run_live_pool_rotation_backtest(
                 weight = 1.0 / len(ranked)
                 for symbol in ranked.index:
                     target_weights.loc[symbol] = weight
-            previous_cash_weight = 1.0 - float(portfolio_weights.sum())
-            target_cash_weight = 1.0 - float(target_weights.sum())
-            turnover = float(
-                ((target_weights - portfolio_weights).abs().sum() + abs(target_cash_weight - previous_cash_weight))
-                * 0.5
+            shares, cash, costs, sale_notional, purchase_notional = _rebalance_holdings(
+                shares,
+                cash,
+                current_prices,
+                target_weights,
+                cost_rate=cost_rate,
             )
-            fee = turnover * fee_bps / 10_000.0
-            slippage = turnover * slippage_bps / 10_000.0
-            portfolio_weights = target_weights
+            traded = sale_notional + purchase_notional
+            turnover = traded / (2.0 * equity) if equity > 0.0 else 0.0
+            if cost_rate > 0.0 and costs > 0.0:
+                fee = costs * (fee_rate_value / cost_rate)
+                slippage = costs * (slippage_rate / cost_rate)
+            else:
+                fee = 0.0
+                slippage = 0.0
             trade_records.append(
                 {
                     "signal_date": pd.Timestamp(signal_date),
@@ -176,36 +226,23 @@ def run_live_pool_rotation_backtest(
                 }
             )
 
-        exposed = portfolio_weights.ne(0.0)
-        # Exiting assets need this open, but only retained/new assets need the next.
-        current_prices = open_matrix.loc[effective_date, held | exposed]
-        next_prices = open_matrix.iloc[effective_idx + 1].loc[exposed]
-        required_prices = pd.concat([current_prices, next_prices])
-        if not (np.isfinite(required_prices) & required_prices.gt(0.0)).all():
-            raise ValueError("required open prices must be finite and positive")
-        open_returns = (
-            next_prices.div(open_matrix.loc[effective_date, exposed]).sub(1.0)
-            .reindex(symbols, fill_value=0.0)
-        )
-        gross_return = float((portfolio_weights * open_returns).sum())
-        cost = fee + slippage
-        if cost >= 1.0:
-            raise ValueError("transaction cost must be less than 1.0")
-        net_return = gross_return - cost
+        held = shares > 0.0
+        next_prices = open_matrix.iloc[effective_idx + 1]
+        if held.any():
+            required = next_prices.loc[held]
+            if not (np.isfinite(required) & required.gt(0.0)).all():
+                raise ValueError("required open prices must be finite and positive")
+        marked = cash + float((shares[held] * next_prices[held]).sum())
+        if equity <= 0.0:
+            raise ValueError("net return must be greater than -1.0")
+        net_return = marked / equity - 1.0
         if net_return <= -1.0:
             raise ValueError("net return must be greater than -1.0")
         daily_returns.append(net_return)
         daily_turnover.append(turnover)
         daily_fees.append(fee)
         daily_slippage.append(slippage)
-        gross_growth = 1.0 + gross_return
-        portfolio_weights = (
-            portfolio_weights.mul(1.0 + open_returns)
-            .div(gross_growth)
-            .fillna(0.0)
-            if gross_growth > 0.0
-            else pd.Series(0.0, index=symbols, dtype=float)
-        )
+        equity = marked
 
     returns = pd.Series(daily_returns, index=pd.DatetimeIndex(effective_dates))
     turnover = pd.Series(daily_turnover, index=returns.index, dtype=float)

tests/test_backtest_correctness.py

@@ -76,23 +76,49 @@ def test_weights_drift_and_cost_aggregates_are_mathematically_exact() -> None:
         slippage_bps=100,
     )
 
-    expected_returns = [0.48, 1.0 / 3.0, 0.495]
-    assert result.returns.tolist() == pytest.approx(expected_returns)
-
-    # Initial funding from cash is full turnover. After +100% then +50% in A,
-    # weights drift 1/2 -> 2/3 -> 3/4; the second rebalance is half-L1 1/4.
-    assert result.trade_log["turnover"].tolist() == pytest.approx([1.0, 0.25])
-    assert result.trade_log["fee"].tolist() == pytest.approx([0.01, 0.0025])
-    assert result.trade_log["slippage"].tolist() == pytest.approx([0.01, 0.0025])
-    assert result.trade_log["cost"].tolist() == pytest.approx([0.02, 0.005])
-
-    expected_total_return = (1.48 * (4.0 / 3.0) * 1.495) - 1.0
+    # Fee/slippage constrain entry notional (no self-financed full weight).
+    cost_rate = 0.02
+    entry_notional = 1.0 / (1.0 + cost_rate)
+    ret0 = 1.5 / (1.0 + cost_rate) / 1.0 - 1.0  # 50/50 with A +100%
+    ret1 = 1.0 / 3.0  # no trade; A +50% on drifted weights
+    # Second rebalance from 75/25 back toward 50/50 under fee drag.
+    equity_before_second = 2.0 / (1.0 + cost_rate)
+    sale = 0.25 * equity_before_second
+    purchase = sale * (1.0 - cost_rate) / (1.0 + cost_rate)
+    second_cost = (sale + purchase) * cost_rate
+    # After selling A back to 50%, keep prior B and add constrained buy; then B +100%.
+    a_after = 0.5 * equity_before_second
+    b_after = 0.25 * equity_before_second + purchase
+    marked = a_after + 2.0 * b_after
+    ret2 = marked / equity_before_second - 1.0
+
+    assert result.returns.tolist() == pytest.approx([ret0, ret1, ret2])
+    assert result.trade_log["turnover"].tolist() == pytest.approx([
+        entry_notional / 2.0,
+        (sale + purchase) / (2.0 * equity_before_second),
+    ])
+    assert result.trade_log["fee"].tolist() == pytest.approx([
+        entry_notional * 0.01,
+        second_cost * 0.5,
+    ])
+    assert result.trade_log["slippage"].tolist() == pytest.approx([
+        entry_notional * 0.01,
+        second_cost * 0.5,
+    ])
+    assert result.trade_log["cost"].tolist() == pytest.approx([
+        entry_notional * cost_rate,
+        second_cost,
+    ])
+
+    expected_total_return = (1.0 + ret0) * (1.0 + ret1) * (1.0 + ret2) - 1.0
     assert result.metrics["total_return"] == pytest.approx(expected_total_return)
     assert result.metrics["total_turnover"] == pytest.approx(result.trade_log["turnover"].sum())
     assert result.metrics["total_fees"] == pytest.approx(result.trade_log["fee"].sum())
     assert result.metrics["total_slippage"] == pytest.approx(result.trade_log["slippage"].sum())
     assert result.metrics["total_cost"] == pytest.approx(result.trade_log["cost"].sum())
-    assert result.metrics["Turnover"] == pytest.approx(1.25 / 3.0 * 365.25)
+    assert result.metrics["Turnover"] == pytest.approx(
+        result.trade_log["turnover"].sum() / 3.0 * 365.25
+    )
 
 
 def test_full_cash_entry_and_exit_charge_full_turnover() -> None:
@@ -112,8 +138,9 @@ def test_full_cash_entry_and_exit_charge_full_turnover() -> None:
         fee_bps=100,
     )
 
-    assert result.trade_log["turnover"].tolist() == pytest.approx([1.0, 1.0])
-    assert result.trade_log["fee"].tolist() == pytest.approx([0.01, 0.01])
+    entry = 1.0 / 1.01
+    assert result.trade_log["turnover"].tolist() == pytest.approx([entry / 2.0, 0.5])
+    assert result.trade_log["fee"].tolist() == pytest.approx([entry * 0.01, entry * 0.01])
 
 
 def test_full_asset_replacement_has_unit_turnover() -> None:
@@ -135,7 +162,7 @@ def test_full_asset_replacement_has_unit_turnover() -> None:
         signal_lag_days=0,
     )
 
-    assert result.trade_log["turnover"].tolist() == pytest.approx([1.0, 1.0])
+    assert result.trade_log["turnover"].tolist() == pytest.approx([0.5, 1.0])
 
 
 @pytest.mark.parametrize(("fee_bps", "fee_rate"), [(10, 0.01), (0, 0.001)])

tests/test_live_pool_simulator_prices.py

@@ -115,8 +115,9 @@ def test_cash_after_exit_does_not_require_future_asset_prices() -> None:
 
     result = run_live_pool_rotation_backtest(panel, top_n=1, rebalance_every=1, fee_bps=100)
 
-    assert result.returns.tolist() == pytest.approx([-0.01, -0.01, 0.0])
-    assert result.trade_log["turnover"].tolist() == pytest.approx([1.0, 1.0, 0.0])
+    # Self-financed buy: equity drops by fee/(1+fee); turnover uses traded/(2*equity).
+    assert result.returns.tolist() == pytest.approx([-0.01 / 1.01, -0.01, 0.0])
+    assert result.trade_log["turnover"].tolist() == pytest.approx([0.5 / 1.01, 0.5, 0.0])
 
 
 def test_cash_before_late_selection_does_not_require_asset_prices() -> None:
@@ -128,7 +129,8 @@ def test_cash_before_late_selection_does_not_require_asset_prices() -> None:
     result = run_live_pool_rotation_backtest(panel, top_n=1, rebalance_every=1)
 
     assert result.returns.tolist() == [0.0, 0.0, 0.0]
-    assert result.trade_log["turnover"].tolist() == [0.0, 0.0, 1.0]
+    # One-way cash->asset deployment is half-turn under traded/(2*equity).
+    assert result.trade_log["turnover"].tolist() == [0.0, 0.0, 0.5]
 
 
 @pytest.mark.parametrize("signal_lag", [0, 1, 5])

tests/test_orchestrator_runner.py

@@ -2,6 +2,10 @@
 
 from __future__ import annotations
 
+import math
+
+import pandas as pd
+
 import tempfile
 import unittest
 from datetime import date
@@ -119,5 +123,70 @@ def test_walk_forward_combo_profile(self) -> None:
             self.assertTrue(all(item.strategy_profile == CRYPTO_EQUITY_COMBO_PROFILE for item in results))
 
 
+
+class AccountingMetricsRegressionTests(unittest.TestCase):
+    """QSL-20260906-006 / 007: initial NAV drawdown + fee-constrained share ledger."""
+
+    def test_max_drawdown_includes_initial_nav(self) -> None:
+        from crypto_strategies.backtest.live_pool_simulator import _performance_metrics
+
+        cases = (
+            ([-0.1], -0.1),
+            ([-0.1, 0.0], -0.1),
+            ([-0.1, 0.1], -0.1),
+            ([0.1, -0.2], -0.2),
+        )
+        for returns, expected in cases:
+            with self.subTest(returns=returns):
+                metrics = _performance_metrics(pd.Series(returns, dtype=float))
+                self.assertAlmostEqual(metrics["Max Drawdown"], expected)
+
+    def test_cash_entry_fee_constrains_buyable_shares(self) -> None:
+        from crypto_strategies.backtest.live_pool_simulator import run_live_pool_rotation_backtest
+
+        panel = pd.DataFrame(
+            [
+                {"date": "2024-01-01", "symbol": "A", "in_universe": True, "open": 100.0, "final_score": 1.0},
+                {"date": "2024-01-02", "symbol": "A", "in_universe": True, "open": 100.0, "final_score": 1.0},
+                {"date": "2024-01-03", "symbol": "A", "in_universe": True, "open": 110.0, "final_score": 1.0},
+            ]
+        )
+        panel["date"] = pd.to_datetime(panel["date"])
+        panel = panel.set_index(["date", "symbol"])
+        result = run_live_pool_rotation_backtest(
+            panel, top_n=1, rebalance_every=1, fee_bps=100, slippage_bps=0.0
+        )
+        # Buyable notional is 1/1.01; +10% mark => 1.1/1.01 - 1.
+        self.assertAlmostEqual(float(result.returns.iloc[0]), 1.1 / 1.01 - 1.0)
+        self.assertAlmostEqual(float(result.trade_log.loc[0, "fee"]), 0.01 / 1.01)
+        self.assertAlmostEqual(float(result.trade_log.loc[0, "turnover"]), 0.5 / 1.01)
+
+    def test_no_trade_days_drift_without_self_financing(self) -> None:
+        from crypto_strategies.backtest.live_pool_simulator import run_live_pool_rotation_backtest
+
+        panel = pd.DataFrame(
+            [
+                {"date": "2024-01-01", "symbol": "A", "in_universe": True, "open": 100.0, "final_score": 1.0},
+                {"date": "2024-01-01", "symbol": "B", "in_universe": True, "open": 100.0, "final_score": 0.0},
+                {"date": "2024-01-02", "symbol": "A", "in_universe": True, "open": 100.0, "final_score": 1.0},
+                {"date": "2024-01-02", "symbol": "B", "in_universe": True, "open": 100.0, "final_score": 0.0},
+                {"date": "2024-01-03", "symbol": "A", "in_universe": True, "open": 200.0, "final_score": 1.0},
+                {"date": "2024-01-03", "symbol": "B", "in_universe": True, "open": 100.0, "final_score": 0.0},
+                {"date": "2024-01-04", "symbol": "A", "in_universe": True, "open": 100.0, "final_score": 1.0},
+                {"date": "2024-01-04", "symbol": "B", "in_universe": True, "open": 100.0, "final_score": 0.0},
+            ]
+        )
+        panel["date"] = pd.to_datetime(panel["date"])
+        panel = panel.set_index(["date", "symbol"])
+        result = run_live_pool_rotation_backtest(
+            panel, top_n=2, rebalance_every=7, fee_bps=0.0
+        )
+        # Equal-weight: +0.5 then -1/3, terminal equity returns to 1.
+        self.assertEqual(len(result.returns), 2)
+        self.assertAlmostEqual(float(result.returns.iloc[0]), 0.5)
+        self.assertAlmostEqual(float(result.returns.iloc[1]), -1.0 / 3.0)
+        self.assertAlmostEqual(float((1.0 + result.returns).prod()), 1.0)
+
+
 if __name__ == "__main__":
     unittest.main()