diff --git a/app/strategies/quant_algos/statistical_algos.py b/app/strategies/quant_algos/statistical_algos.py index ac5c4a3..dce1b8d 100644 --- a/app/strategies/quant_algos/statistical_algos.py +++ b/app/strategies/quant_algos/statistical_algos.py @@ -11,20 +11,23 @@ def linear_regression_channel( """Deviation from linear regression line. Trade reversion to line.""" if len(prices) < lookback: return 0.0, f"warming up ({len(prices)}/{lookback})" - x = list(range(lookback)) y = prices[-lookback:] - n = len(x) - sx, sy = sum(x), sum(y) - sxx = sum(xi * xi for xi in x) - sxy = sum(xi * yi for xi, yi in zip(x, y)) + n = lookback + sx = n * (n - 1) // 2 + sxx = n * (n - 1) * (2 * n - 1) // 6 + sy = sum(y) + sxy = sum(xi * yi for xi, yi in enumerate(y)) den = n * sxx - sx * sx if den == 0: return 0.0, "singular" slope = (n * sxy - sx * sy) / den intercept = (sy - slope * sx) / n pred = intercept + slope * (n - 1) - residuals = [yi - (intercept + slope * xi) for xi, yi in zip(x, y)] - var = sum(r * r for r in residuals) / len(residuals) + residual_sum_sq = 0.0 + for xi, yi in enumerate(y): + residual = yi - (intercept + slope * xi) + residual_sum_sq += residual * residual + var = residual_sum_sq / n std = math.sqrt(var) if var > 0 else 1e-10 p = prices[-1] dev = (p - pred) / std if std else 0 diff --git a/shared/tests/test_statistical_algos.py b/shared/tests/test_statistical_algos.py new file mode 100644 index 0000000..d1aea1d --- /dev/null +++ b/shared/tests/test_statistical_algos.py @@ -0,0 +1,54 @@ +import math + +from app.strategies.quant_algos.statistical_algos import linear_regression_channel + + +def _previous_linear_regression_channel(prices, lookback=20): + if len(prices) < lookback: + return 0.0, f"warming up ({len(prices)}/{lookback})" + x = list(range(lookback)) + y = prices[-lookback:] + n = len(x) + sx, sy = sum(x), sum(y) + sxx = sum(xi * xi for xi in x) + sxy = sum(xi * yi for xi, yi in zip(x, y)) + den = n * sxx - sx * sx + if den == 0: + return 0.0, "singular" + slope = (n * sxy - sx * sy) / den + intercept = (sy - slope * sx) / n + pred = intercept + slope * (n - 1) + residuals = [yi - (intercept + slope * xi) for xi, yi in zip(x, y)] + var = sum(r * r for r in residuals) / len(residuals) + std = math.sqrt(var) if var > 0 else 1e-10 + p = prices[-1] + dev = (p - pred) / std if std else 0 + sig = -dev + return max(-2, min(2, sig)), f"dev={dev:.2f} pred={pred:.2f}" + + +def test_linear_regression_channel_matches_previous_outputs(): + cases = [ + [100 + i * 0.2 for i in range(80)], + [100 + math.sin(i / 3) * 2 + i * 0.03 for i in range(80)], + [120 - i * 0.35 + math.cos(i / 4) for i in range(80)], + [100, 102, 101, 103, 99, 98, 100, 101, 103, 104, 102, 101], + ] + + for prices in cases: + expected_signal, expected_detail = _previous_linear_regression_channel( + prices, lookback=min(20, len(prices)) + ) + actual_signal, actual_detail = linear_regression_channel( + prices, lookback=min(20, len(prices)) + ) + + assert actual_signal == expected_signal + assert actual_detail == expected_detail + + +def test_linear_regression_channel_keeps_warmup_behavior(): + assert linear_regression_channel([1.0, 2.0, 3.0], lookback=5) == ( + 0.0, + "warming up (3/5)", + )