perf: compute volume OBV trend in one pass

app/strategies/quant_algos/volume_algos.py

@@ -33,23 +33,22 @@ def obv_trend(
     """OBV trend: volume-confirmed momentum. Signal from OBV change over lookback."""
     if not candles or len(candles) < lookback + 1:
         return 0.0, f"warming up ({len(candles)}/{lookback+1})"
-    obv = 0
+    obv_prev = 0
+    obv_delta = 0
+    tail_start = len(candles) - lookback
     for i in range(1, len(candles)):
         v = candles[i].get("volume", 0)
         if candles[i]["close"] >= candles[i - 1]["close"]:
-            obv += v
+            signed_volume = v
         else:
-            obv -= v
-    obv_prev = 0
-    for i in range(1, len(candles) - lookback):
-        v = candles[i].get("volume", 0)
-        if candles[i]["close"] >= candles[i - 1]["close"]:
-            obv_prev += v
+            signed_volume = -v
+        if i < tail_start:
+            obv_prev += signed_volume
         else:
-            obv_prev -= v
+            obv_delta += signed_volume
     if obv_prev == 0:
         return 0.0, "obv_prev zero"
-    change_pct = (obv - obv_prev) / abs(obv_prev)
+    change_pct = obv_delta / abs(obv_prev)
     sig = max(-1, min(1, change_pct * 10))  # scale
     return sig, f"OBV change%={change_pct:.4f}"
 

tests/test_volume_algos.py

@@ -0,0 +1,59 @@
+from app.strategies.quant_algos.volume_algos import obv_trend
+
+
+def _reference_obv_trend(candles, lookback=10):
+    if not candles or len(candles) < lookback + 1:
+        return 0.0, f"warming up ({len(candles)}/{lookback+1})"
+    obv = 0
+    for i in range(1, len(candles)):
+        v = candles[i].get("volume", 0)
+        if candles[i]["close"] >= candles[i - 1]["close"]:
+            obv += v
+        else:
+            obv -= v
+    obv_prev = 0
+    for i in range(1, len(candles) - lookback):
+        v = candles[i].get("volume", 0)
+        if candles[i]["close"] >= candles[i - 1]["close"]:
+            obv_prev += v
+        else:
+            obv_prev -= v
+    if obv_prev == 0:
+        return 0.0, "obv_prev zero"
+    change_pct = (obv - obv_prev) / abs(obv_prev)
+    sig = max(-1, min(1, change_pct * 10))
+    return sig, f"OBV change%={change_pct:.4f}"
+
+
+def _candles(count):
+    candles = []
+    close = 100.0
+    for i in range(count):
+        close += ((i % 7) - 3) * 0.37
+        high = close + 1.5 + (i % 3) * 0.1
+        low = close - 1.2 - (i % 5) * 0.1
+        candles.append(
+            {
+                "open": close - 0.2,
+                "high": high,
+                "low": low,
+                "close": close,
+                "volume": 1000 + (i % 19) * 17,
+            }
+        )
+    return candles
+
+
+def test_obv_trend_matches_reference_two_pass_calculation():
+    candles = _candles(5000)
+
+    assert obv_trend(candles, lookback=25) == _reference_obv_trend(candles, lookback=25)
+
+
+def test_obv_trend_preserves_zero_prefix_behavior():
+    candles = [
+        {"high": 101, "low": 99, "close": 100, "volume": 100},
+        {"high": 102, "low": 100, "close": 101, "volume": 100},
+    ]
+
+    assert obv_trend(candles, lookback=1) == (0.0, "obv_prev zero")