Project the trendline with fixed weights, not a polyfit per row (0.14.4)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "cotmetrics"
-version = "0.14.3"
+version = "0.14.4"
 description = "Turn raw COT data into positioning metrics and trading signals (positioning index, concentration/clustering/position-size, reversal signals)."
 readme = "README.md"
 license = { text = "MIT" }

src/cotmetrics/conditions.py

@@ -151,11 +151,33 @@ def project_trendline(price_col):
     return projected_y
 
 
+def rolling_trendline_projection(series, window):
+    """``project_trendline`` over every trailing window, in one vectorised pass.
+
+    For a fixed window the least-squares line through (0..n-1, y) projected to x=n
+    is a fixed linear combination of the y values: w_i = 1/n + (x_i - x̄)(n - x̄)/Sxx.
+    So the rolling fit is a dot product with those weights rather than one polyfit
+    per row, which was the largest single cost in append_trading_signals (~1/3 of
+    it: a polyfit per row per market per basis). Same NaN rule as
+    ``rolling(window).apply``: any NaN in the window gives NaN. The values agree with
+    polyfit to rounding (~1e-15 relative), not bit for bit, because lstsq sums in a
+    different order; pinned against it in tests.
+    """
+    values = series.to_numpy(dtype=float)
+    out = np.full(len(values), np.nan)
+    if window > 0 and len(values) >= window:
+        x = np.arange(window, dtype=float)
+        x_bar = x.mean()
+        weights = 1.0 / window + (x - x_bar) * (window - x_bar) / ((x - x_bar) ** 2).sum()
+        out[window - 1:] = np.lib.stride_tricks.sliding_window_view(values, window) @ weights
+    return pd.Series(out, index=series.index)
+
+
 def is_down_trend_line_break(df, price_col=const.CLOSING_PRICE, last_n_highs=5):
     is_downtrend = (df[price_col].shift(1) < df[price_col].shift(2)) & (df[price_col].shift(2) < df[price_col].shift(3))
 
     # Shift the data FIRST, then apply the rolling window.
-    projected_resistance = df[price_col].shift(1).rolling(window=last_n_highs).apply(project_trendline, raw=True)
+    projected_resistance = rolling_trendline_projection(df[price_col].shift(1), last_n_highs)
 
     # The Breakout Logic
     # Compare today's Close to projected resistance line.

tests/test_conditions.py

@@ -55,6 +55,25 @@ def test_project_trendline_flat_line():
     assert conditions.project_trendline(flat) == pytest.approx(42.0)
 
 
+@pytest.mark.parametrize("window", [2, 3, 5, 10])
+def test_rolling_trendline_projection_matches_rolling_polyfit(window):
+    # The vectorised form must agree with a polyfit per window, NaN for NaN, since
+    # is_down_trend_line_break's crossover compares a close against it.
+    rng = np.random.default_rng(7)
+    s = pd.Series(100 + rng.normal(0, 2, 400).cumsum())
+    s.iloc[[0, 37, 38, 200]] = np.nan  # leading NaN and gaps inside windows
+    expected = s.rolling(window).apply(conditions.project_trendline, raw=True)
+    got = conditions.rolling_trendline_projection(s, window)
+    assert got.index.equals(s.index)
+    assert (got.isna() == expected.isna()).all()
+    np.testing.assert_allclose(got, expected, rtol=1e-10, equal_nan=True)
+
+
+def test_rolling_trendline_projection_shorter_than_window_is_all_nan():
+    got = conditions.rolling_trendline_projection(pd.Series([1.0, 2.0, 3.0]), 5)
+    assert got.isna().all() and len(got) == 3
+
+
 # ── calculate_cot_macd ──────────────────────────────────────────────────────
 def test_cot_macd_constant_series_is_all_zero():
     net = pd.Series([100.0] * 60)