diff --git a/pyproject.toml b/pyproject.toml index 9a86a09..045041e 100644 --- a/pyproject.toml +++ b/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" } diff --git a/src/cotmetrics/conditions.py b/src/cotmetrics/conditions.py index e250558..af51c6a 100644 --- a/src/cotmetrics/conditions.py +++ b/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. diff --git a/tests/test_conditions.py b/tests/test_conditions.py index cdb6152..2105226 100644 --- a/tests/test_conditions.py +++ b/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)