Precompute ATR series for paper-trade trailing exits

Both _atr_trailing_close (scheduled) and _atr_trailing_level (dashboard
read path) recomputed ATR from scratch on every post-entry bar via
compute_atr(rows[:idx+1]) — O(n*k) per trade. Replace with a single O(n)
Wilder pass, _atr_series_from_rows, that stores round(running, 4) at each
index. compute_atr keeps its running ATR unrounded through the recurrence
and rounds only at return, so this reproduces its per-prefix value exactly
(no behavior change; live-vs-backtest atr_trail3 parity still byte-identical).

Remove the now-unused _atr_from_rows and its compute_atr import. Add a
per-index parity test against compute_atr; existing ATR tests now mock
_atr_series_from_rows (same effect as the old fixed-ATR mock).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app/services/paper_trade_service.py

@@ -12,7 +12,6 @@ from app.models.ohlcv import OHLCVRecord
 from app.models.paper_trade import PaperTrade
 from app.models.ticker import Ticker
 from app.services import benchmark_service, settings_store
-from app.services.indicator_service import compute_atr
 from app.services.outcome_service import (
     OUTCOME_AMBIGUOUS,
     OUTCOME_STOP_HIT,
@@ -181,17 +180,33 @@ def _trailing_close(
     return None
 
 
-def _atr_from_rows(rows: list[tuple], idx: int) -> float | None:
-    try:
-        result = compute_atr(
-            [float(r[2]) for r in rows[: idx + 1]],
-            [float(r[3]) for r in rows[: idx + 1]],
-            [float(r[4]) for r in rows[: idx + 1]],
-        )
-    except Exception:
-        return None
-    atr = result.get("atr")
-    return float(atr) if atr and atr > 0 else None
+def _atr_series_from_rows(rows: list[tuple], period: int = 14) -> list[float | None]:
+    """ATR at each index i, equal to ``compute_atr(rows[: i + 1])["atr"]`` but
+    computed in a single O(n) Wilder pass instead of re-smoothing the whole
+    prefix per bar. None where there are fewer than ``period + 1`` bars or the
+    rounded ATR is non-positive. ``period`` mirrors ``compute_atr``'s default;
+    keep them in sync.
+
+    Exactness: ``compute_atr`` keeps its running ATR unrounded through the
+    recurrence and rounds only at return, so storing ``round(running, 4)`` at
+    each index reproduces its per-prefix value bit-for-bit.
+    """
+    n = len(rows)
+    out: list[float | None] = [None] * n
+    if n < period + 1:
+        return out
+    tr = [0.0] * n
+    for i in range(1, n):
+        high, low, prev_close = float(rows[i][2]), float(rows[i][3]), float(rows[i - 1][4])
+        tr[i] = max(high - low, abs(high - prev_close), abs(low - prev_close))
+    running = sum(tr[1 : period + 1]) / period
+    rounded = round(running, 4)
+    out[period] = rounded if rounded > 0 else None
+    for j in range(period + 1, n):
+        running = (running * (period - 1) + tr[j]) / period
+        rounded = round(running, 4)
+        out[j] = rounded if rounded > 0 else None
+    return out
 
 
 def _atr_trailing_level(
@@ -206,11 +221,12 @@ def _atr_trailing_level(
     long = direction == "long"
     stop = float(init_stop)
     anchor = float(entry)
+    atr_by_idx = _atr_series_from_rows(rows)
     for idx, (d, _, _, _, close) in enumerate(rows):
         if d <= opened_on:
             continue
         close = float(close)
-        atr = _atr_from_rows(rows, idx)
+        atr = atr_by_idx[idx]
         if long:
             anchor = max(anchor, close)
             if atr is not None:
@@ -244,6 +260,7 @@ def _atr_trailing_close(
     stop = float(init_stop)
     anchor = float(entry)
     bars_held = 0
+    atr_by_idx = _atr_series_from_rows(rows)
     for idx, (d, open_, high, low, close) in enumerate(rows):
         if d <= opened_on:
             continue
@@ -265,7 +282,7 @@ def _atr_trailing_close(
         if bars_held >= hold_days:
             return close, d, "time"
 
-        atr = _atr_from_rows(rows, idx)
+        atr = atr_by_idx[idx]
         if long:
             anchor = max(anchor, close)
             if atr is not None:

tests/unit/test_paper_trade_service.py

@@ -206,7 +206,7 @@ class TestTrailingClose:
 
 class TestAtrTrailingClose:
     def test_long_uses_ratchet_on_next_bar(self, monkeypatch):
-        monkeypatch.setattr(svc, "compute_atr", lambda *_args, **_kwargs: {"atr": 5.0})
+        monkeypatch.setattr(svc, "_atr_series_from_rows", lambda rows, period=14: [5.0] * len(rows))
         rows = [
             _r(date(2026, 1, 1), 100, 100, 100, 100),
             _r(date(2026, 1, 2), 115, 121, 114, 120),
@@ -222,7 +222,7 @@ class TestAtrTrailingClose:
         assert reason == "trailing"
 
     def test_max_hold_still_closes(self, monkeypatch):
-        monkeypatch.setattr(svc, "compute_atr", lambda *_args, **_kwargs: {"atr": 50.0})
+        monkeypatch.setattr(svc, "_atr_series_from_rows", lambda rows, period=14: [50.0] * len(rows))
         rows = [
             _r(date(2026, 1, 1), 100, 100, 100, 100),
             _r(date(2026, 1, 2), 101, 102, 100, 101),
@@ -264,6 +264,36 @@ def _r(d: date, open_: float, hi: float, lo: float, close: float) -> tuple:
     return (d, open_, hi, lo, close)
 
 
+def test_atr_series_matches_compute_atr_per_prefix():
+    """_atr_series_from_rows[i] must equal compute_atr(rows[: i + 1])['atr'] (with
+    the same >0 / insufficient-history guards) at every index. The O(n) rewrite is
+    only valid because it reproduces the per-prefix value exactly."""
+    from app.services.indicator_service import compute_atr
+
+    price = 100.0
+    closes = []
+    for i in range(200):
+        price = max(1.0, price + (3.0 if i % 3 else -2.0) + (i % 7) * 0.25)
+        closes.append(price)
+    rows = [
+        _r(date(2024, 1, 1) + timedelta(days=i), c, c + 1.5, c - 1.2, c)
+        for i, c in enumerate(closes)
+    ]
+
+    series = svc._atr_series_from_rows(rows)
+    highs = [r[2] for r in rows]
+    lows = [r[3] for r in rows]
+    closes_col = [r[4] for r in rows]
+    assert len(series) == len(rows)
+    for i in range(len(rows)):
+        if i < 14:
+            assert series[i] is None
+        else:
+            raw = compute_atr(highs[: i + 1], lows[: i + 1], closes_col[: i + 1])["atr"]
+            expected = float(raw) if raw and raw > 0 else None
+            assert series[i] == expected, f"index {i}: {series[i]} != {expected}"
+
+
 class TestTimeClose:
     def test_closes_at_hold_days_close(self):
         rows = [
@@ -322,7 +352,7 @@ async def test_resolve_trailing_closes_with_reason(session):
 
 
 async def test_resolve_atr_trailing_closes_with_reason(session, monkeypatch):
-    monkeypatch.setattr(svc, "compute_atr", lambda *_args, **_kwargs: {"atr": 5.0})
+    monkeypatch.setattr(svc, "_atr_series_from_rows", lambda rows, period=14: [5.0] * len(rows))
     await svc.set_exit_policy(session, mode="atr_trailing", atr_multiplier=3.0)
     tid = await _seed(session, "AAA", close=100.0)
     await _add_open_trade(session, tid, "long", entry=100.0, shares=10, days_ago=10)
@@ -352,7 +382,7 @@ async def test_list_open_exposes_trailing_stop(session):
 
 
 async def test_list_open_exposes_atr_trailing_stop(session, monkeypatch):
-    monkeypatch.setattr(svc, "compute_atr", lambda *_args, **_kwargs: {"atr": 5.0})
+    monkeypatch.setattr(svc, "_atr_series_from_rows", lambda rows, period=14: [5.0] * len(rows))
     await svc.set_exit_policy(session, mode="atr_trailing", atr_multiplier=3.0)
     tid = await _seed(session, "AAA", close=120.0)
     await _add_open_trade(session, tid, "long", entry=100.0, shares=10, days_ago=10)