fix: harden Structural S/R after OHLCV writes and surface cleanup failures

Honor custom S/R tolerance as a transient detect, refresh levels after OHLCV
mutations without failing committed price writes, report per-ticker S/R
rebuild failures from admin cleanup, and warn in the admin UI when refresh is partial.

README.md

@@ -204,7 +204,7 @@ Use this as the historical ranking/exit regression guardrail, not as a return pr
 | Item | Historical weekly baseline |
 |---|---|
 | Strategy version | `residual_highvol_80_20_atr_trail3_v1` |
-| Production gate | Long-only, residual 12-1 momentum percentile >= 80, headline gate-target R:R >= 2.0 (live `activation_min_rr`; the code default is 1.2), primary-target reach-probability >= 20%, NEUTRAL excluded, confidence floor off (0) |
+| Production gate | Long-only, residual 12-1 momentum percentile >= 80, headline gate-target R:R >= 2.0 (live `activation_min_rr`; code default 2.0), primary-target reach-probability >= 20%, NEUTRAL excluded, confidence floor off (0) |
 | Production rank | 80% residual momentum percentile + 20% 6-month realized-volatility percentile |
 | Exit | Initial ATR stop plus 3x ATR trailing stop, max 30 trading days |
 | Portfolio CAGR | +50.4% |

app/routers/ingestion.py

@@ -19,6 +19,7 @@ from app.dependencies import get_db, require_access
 from app.exceptions import ProviderError
 from app.models.ohlcv import OHLCVRecord
 from app.models.settings import IngestionProgress
+from app.models.sr_level import SRLevel
 from app.models.ticker import Ticker
 from app.models.user import User
 from app.providers.alpaca import AlpacaOHLCVProvider
@@ -105,8 +106,13 @@ async def fetch_symbol(
                 await db.execute(
                     delete(IngestionProgress).where(IngestionProgress.ticker_id == ticker_obj.id)
                 )
+                # Drop Structural S/R with the bars; a failed re-fetch must not
+                # leave zones computed from deleted history.
+                await db.execute(
+                    delete(SRLevel).where(SRLevel.ticker_id == ticker_obj.id)
+                )
                 await db.commit()
-                logger.info("force_refetch: cleared OHLCV and progress for %s", symbol_upper)
+                logger.info("force_refetch: cleared OHLCV, S/R, and progress for %s", symbol_upper)
         except Exception as exc:
             logger.error("force_refetch cleanup failed for %s: %s", symbol_upper, exc)
 

app/routers/sr_levels.py

@@ -74,7 +74,11 @@ async def read_sr_levels(
         None,
         ge=0,
         le=0.1,
-        description="Merge tolerance as fraction of price; omit for ATR-adaptive default",
+        description=(
+            "Merge tolerance as fraction of price. Omit to return persisted levels "
+            "(ATR-adaptive at last recalculation). When set, returns a transient "
+            "detect with this tolerance (not written to the DB)."
+        ),
     ),
     max_zones: int = Query(6, ge=0, description="Max S/R zones to return (default 6)"),
     _user=Depends(require_access),

app/services/admin_service.py

@@ -318,14 +318,18 @@ async def update_ticker_universe_default(db: AsyncSession, universe: str) -> dic
 # Data cleanup
 # ---------------------------------------------------------------------------
 
-async def cleanup_data(db: AsyncSession, older_than_days: int) -> dict[str, int]:
+async def cleanup_data(db: AsyncSession, older_than_days: int) -> dict:
     """Delete OHLCV, sentiment, and fundamental records older than N days.
 
-    Preserves tickers, users, and latest scores.
+    Preserves tickers, users, and latest scores. After OHLCV pruning, rebuilds
-    Returns a dict with counts of deleted records per table.
+    Structural S/R for every ticker so chart levels match the remaining history.
+
+    Returns deleted-row counts plus S/R refresh outcomes. A per-ticker S/R
+    failure rolls the session back (so later tickers still run) and is listed
+    in ``sr_refresh_failures`` rather than aborting the whole cleanup.
     """
     cutoff = datetime.now(timezone.utc) - timedelta(days=older_than_days)
-    counts: dict[str, int] = {}
+    counts: dict = {}
 
     # OHLCV — date column is a date, compare with cutoff date
     result = await db.execute(
@@ -346,6 +350,36 @@ async def cleanup_data(db: AsyncSession, older_than_days: int) -> dict[str, int]
     counts["fundamentals"] = result.rowcount  # type: ignore[assignment]
 
     await db.commit()
+
+    counts["sr_refresh_ok"] = 0
+    counts["sr_refresh_failed"] = 0
+    counts["sr_refresh_failures"] = []
+
+    # Structural S/R is derived from OHLCV; recompute after history shrinks.
+    if counts["ohlcv"]:
+        from app.services.sr_service import recalculate_sr_levels
+
+        symbols = list(
+            (await db.execute(select(Ticker.symbol).order_by(Ticker.symbol))).scalars().all()
+        )
+        for symbol in symbols:
+            try:
+                await recalculate_sr_levels(db, symbol)
+                counts["sr_refresh_ok"] += 1
+            except Exception as exc:
+                logger.exception("S/R refresh after cleanup failed for %s", symbol)
+                try:
+                    await db.rollback()
+                except Exception:
+                    logger.exception(
+                        "Session rollback after S/R cleanup failure also failed for %s",
+                        symbol,
+                    )
+                counts["sr_refresh_failed"] += 1
+                counts["sr_refresh_failures"].append(
+                    {"symbol": symbol, "error": f"{type(exc).__name__}: {exc}"}
+                )
+
     return counts
 
 

app/services/ingestion_service.py

@@ -23,6 +23,15 @@ from app.services import price_service
 logger = logging.getLogger(__name__)
 
 
+async def _refresh_structural_sr(db: AsyncSession, symbol: str) -> None:
+    """Rebuild Structural S/R after batch OHLCV writes (best-effort).
+
+    Price bars are already committed; an S/R failure must not discard the
+    ingestion result. Shared with single-bar upsert via price_service.
+    """
+    await price_service._refresh_structural_sr_best_effort(db, symbol)
+
+
 @dataclass
 class IngestionResult:
     """Result of an ingestion run."""
@@ -213,6 +222,8 @@ async def fetch_and_ingest(
                 low=record.low,
                 close=record.close,
                 volume=record.volume,
+                # One S/R rebuild at the end of the batch, not per bar.
+                refresh_sr=False,
             )
             ingested_count += 1
             last_ingested = record.date
@@ -221,12 +232,15 @@ async def fetch_and_ingest(
             await _update_progress(db, ticker.id, record.date)
 
         except RateLimitError:
-            # Mid-ingestion rate limit — return partial progress
+            # Mid-ingestion rate limit — return partial progress after
+            # refreshing S/R from whatever bars we already wrote.
             logger.warning(
                 "Rate limited during ingestion for %s after %d records",
                 ticker.symbol,
                 ingested_count,
             )
+            if ingested_count > 0:
+                await _refresh_structural_sr(db, ticker.symbol)
             return IngestionResult(
                 symbol=ticker.symbol,
                 records_ingested=ingested_count,
@@ -235,6 +249,9 @@ async def fetch_and_ingest(
                 message=f"Rate limited. Ingested {ingested_count} records. Resume available.",
             )
 
+    if ingested_count > 0:
+        await _refresh_structural_sr(db, ticker.symbol)
+
     return IngestionResult(
         symbol=ticker.symbol,
         records_ingested=ingested_count,

app/services/price_service.py

@@ -1,5 +1,8 @@
 """Price Store service: upsert and query OHLCV records."""
 
+from __future__ import annotations
+
+import logging
 from datetime import date, datetime
 
 from sqlalchemy import select
@@ -10,6 +13,8 @@ from app.exceptions import NotFoundError, ValidationError
 from app.models.ohlcv import OHLCVRecord
 from app.models.ticker import Ticker
 
+logger = logging.getLogger(__name__)
+
 
 async def _get_ticker(db: AsyncSession, symbol: str) -> Ticker:
     """Look up a ticker by symbol. Raises NotFoundError if missing."""
@@ -44,11 +49,22 @@ async def upsert_ohlcv(
     low: float,
     close: float,
     volume: int,
+    *,
+    refresh_sr: bool = True,
 ) -> OHLCVRecord:
     """Insert or update an OHLCV record for (ticker, date).
 
     Validates business rules, resolves ticker, then uses
     ON CONFLICT DO UPDATE on the (ticker_id, date) unique constraint.
+
+    ``refresh_sr`` (default True) recalculates persisted Structural S/R after
+    the write so chart levels stay current. Batch ingestion passes
+    ``refresh_sr=False`` and refreshes once at the end of the ticker batch.
+
+    The OHLCV commit is authoritative: if S/R rebuild fails after a successful
+    price write, the error is logged, the session is rolled back to clear
+    poison, and the upsert still returns the persisted bar (caller can retry
+    S/R via the scanner/ingestion pipeline).
     """
     _validate_ohlcv(high, low, open_, close, volume, record_date)
     ticker = await _get_ticker(db, symbol)
@@ -84,9 +100,32 @@ async def upsert_ohlcv(
 
     indicator_cache.invalidate_ticker(ticker.symbol)
 
+    if refresh_sr:
+        await _refresh_structural_sr_best_effort(db, ticker.symbol)
+
     return record
 
 
+async def _refresh_structural_sr_best_effort(db: AsyncSession, symbol: str) -> bool:
+    """Rebuild Structural S/R; never fail a successful OHLCV write.
+
+    Returns True on success. On failure rolls the session back so a later
+    operation on the same session is not poisoned by the failed unit of work.
+    """
+    from app.services.sr_service import recalculate_sr_levels
+
+    try:
+        await recalculate_sr_levels(db, symbol)
+        return True
+    except Exception:
+        logger.exception("Structural S/R refresh failed for %s after OHLCV write", symbol)
+        try:
+            await db.rollback()
+        except Exception:
+            logger.exception("Session rollback after S/R failure also failed for %s", symbol)
+        return False
+
+
 async def query_ohlcv(
     db: AsyncSession,
     symbol: str,

app/services/sr_service.py

@@ -857,17 +857,42 @@ async def get_sr_levels(
     symbol: str,
     tolerance: float | None = None,
 ) -> list[SRLevel]:
-    """Return persisted Structural S/R levels, strength descending.
+    """Return Structural S/R for a ticker, strength descending.
 
-    Read-only: does not recompute or rewrite. Pipeline/ingestion call
+    Default (``tolerance is None``): read persisted levels only — no rewrite.
-    ``recalculate_sr_levels`` to refresh. ``tolerance`` is kept for API
+    Pipeline/ingestion call ``recalculate_sr_levels`` after OHLCV changes.
-    compatibility and ignored on read (it only applies at recalculation).
+
+    When ``tolerance`` is set: build a **transient** detect view with that merge
+    tolerance and do not persist it (custom merge for API clients). Transient
+    rows use negative ids so they cannot be confused with stored levels.
     """
-    del tolerance  # API compat only; levels were stored at last recalculation
+    if tolerance is None:
+        ticker = await _get_ticker(db, symbol)
+        result = await db.execute(
+            select(SRLevel)
+            .where(SRLevel.ticker_id == ticker.id)
+            .order_by(SRLevel.strength.desc())
+        )
+        return list(result.scalars().all())
+
+    from types import SimpleNamespace
+
     ticker = await _get_ticker(db, symbol)
-    result = await db.execute(
+    records = await query_ohlcv(db, symbol)
-        select(SRLevel)
+    if not records:
-        .where(SRLevel.ticker_id == ticker.id)
+        return []
-        .order_by(SRLevel.strength.desc())
+    _, highs, lows, closes, volumes = _extract_ohlcv(records)
-    )
+    detected = detect_sr_levels(highs, lows, closes, volumes, tolerance)
-    return list(result.scalars().all())
+    now = datetime.utcnow()
+    # Ephemeral objects with the SRLevel attribute shape the router expects.
+    return [
+        SimpleNamespace(  # type: ignore[return-value]
+            id=-(i + 1),
+            price_level=lvl["price_level"],
+            type=lvl["type"],
+            strength=lvl["strength"],
+            detection_method=lvl["detection_method"],
+            created_at=now,
+        )
+        for i, lvl in enumerate(detected)
+    ]

docs/research/README.md

@@ -80,7 +80,7 @@ Reports: `backtest-20260712-min-rr-sweep.json` (in-sample), `-oos.json` (test wi
 | min_rr | qualified | In-sample Sharpe / CAGR | **OOS** Sharpe / CAGR (entries ≥ 2024-07) |
 |---|---|---|---|
 | 0.0 (floor off) | 6636 | 1.98 / 58.5% | 2.02 / 66.2% |
-| 1.2 (code default) | 3897 | 1.34 / 33.9% | 1.12 / 28.8% |
+| 1.2 (old code default) | 3897 | 1.34 / 33.9% | 1.12 / 28.8% |
 | 1.5 | 3127 | 1.20 / 29.6% | 1.12 / 28.8% |
 | 1.75 | 1974 | 1.64 / 44.5% | 1.15 / 27.4% |
 | **2.0 (live)** | 1089 | **2.04 / 50.4%** | **2.78 / 73.3%** |

frontend/src/api/admin.ts

@@ -269,9 +269,18 @@ export function acknowledgeSystemEvents(days = 7) {
 }
 
 // Data cleanup
+export interface CleanupResult {
+  ohlcv: number;
+  sentiment: number;
+  fundamentals: number;
+  sr_refresh_ok: number;
+  sr_refresh_failed: number;
+  sr_refresh_failures: { symbol: string; error: string }[];
+}
+
 export function cleanupData(olderThanDays: number) {
   return apiClient
-    .post<{ message: string }>('admin/data/cleanup', {
+    .post<CleanupResult>('admin/data/cleanup', {
       older_than_days: olderThanDays,
     })
     .then((r) => r.data);

frontend/src/components/ui/Toast.tsx

@@ -1,7 +1,7 @@
 import { createContext, useCallback, useContext, useEffect, useRef, useState } from 'react';
 import { createPortal } from 'react-dom';
 
-type ToastType = 'success' | 'error' | 'info';
+type ToastType = 'success' | 'error' | 'info' | 'warning';
 
 interface Toast {
   id: string;
@@ -22,6 +22,7 @@ const typeStyles: Record<ToastType, string> = {
   error: 'border-red-500/30 bg-red-500/10 text-red-300',
   success: 'border-emerald-500/30 bg-emerald-500/10 text-emerald-300',
   info: 'border-blue-500/30 bg-blue-500/10 text-blue-300',
+  warning: 'border-amber-500/30 bg-amber-500/10 text-amber-300',
 };
 
 export function ToastProvider({ children }: { children: React.ReactNode }) {

frontend/src/hooks/useAdmin.ts

@@ -404,7 +404,30 @@ export function useCleanupData() {
   return useMutation({
     mutationFn: (olderThanDays: number) => adminApi.cleanupData(olderThanDays),
     onSuccess: (data) => {
-      addToast('success', (data as { message: string }).message || 'Cleanup completed');
+      const deleted =
+        (data.ohlcv ?? 0) + (data.sentiment ?? 0) + (data.fundamentals ?? 0);
+      const failed = data.sr_refresh_failed ?? 0;
+      if (failed > 0) {
+        const symbols = (data.sr_refresh_failures ?? [])
+          .map((f) => f.symbol)
+          .filter(Boolean);
+        const sample = symbols.slice(0, 8).join(', ');
+        const more = symbols.length > 8 ? ` (+${symbols.length - 8} more)` : '';
+        addToast(
+          'warning',
+          `Cleanup deleted ${deleted} row(s), but Structural S/R refresh failed for ${failed} ticker(s)` +
+            (sample ? `: ${sample}${more}` : '') +
+            '. Re-run the R:R scan or a per-ticker refresh to rebuild levels.',
+        );
+        return;
+      }
+      addToast(
+        'success',
+        `Cleanup completed — deleted ${deleted} row(s)` +
+          (data.sr_refresh_ok
+            ? `; rebuilt S/R for ${data.sr_refresh_ok} ticker(s)`
+            : ''),
+      );
     },
     onError: (error: Error) => {
       addToast('error', error.message || 'Failed to cleanup data');