From cad4b49e7c3d7dff9611dfe8f7b6369d0d219556 Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Sat, 18 Jul 2026 13:44:34 +0200 Subject: [PATCH] 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 | 2 +- app/routers/ingestion.py | 8 ++++- app/routers/sr_levels.py | 6 +++- app/services/admin_service.py | 42 ++++++++++++++++++++++--- app/services/ingestion_service.py | 19 ++++++++++- app/services/price_service.py | 39 +++++++++++++++++++++++ app/services/sr_service.py | 47 +++++++++++++++++++++------- docs/research/README.md | 2 +- frontend/src/api/admin.ts | 11 ++++++- frontend/src/components/ui/Toast.tsx | 3 +- frontend/src/hooks/useAdmin.ts | 25 ++++++++++++++- 11 files changed, 181 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index 397a365..ae8c405 100644 --- a/README.md +++ b/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% | diff --git a/app/routers/ingestion.py b/app/routers/ingestion.py index c2c391d..32c29e9 100644 --- a/app/routers/ingestion.py +++ b/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) diff --git a/app/routers/sr_levels.py b/app/routers/sr_levels.py index beaf381..4d60763 100644 --- a/app/routers/sr_levels.py +++ b/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), diff --git a/app/services/admin_service.py b/app/services/admin_service.py index 8353b2a..ee31f9d 100644 --- a/app/services/admin_service.py +++ b/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. - Returns a dict with counts of deleted records per table. + Preserves tickers, users, and latest scores. After OHLCV pruning, rebuilds + 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 diff --git a/app/services/ingestion_service.py b/app/services/ingestion_service.py index b8a5e6e..444aa6b 100644 --- a/app/services/ingestion_service.py +++ b/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, diff --git a/app/services/price_service.py b/app/services/price_service.py index e655bcd..dbc6769 100644 --- a/app/services/price_service.py +++ b/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, diff --git a/app/services/sr_service.py b/app/services/sr_service.py index 440af02..422f10e 100644 --- a/app/services/sr_service.py +++ b/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 - ``recalculate_sr_levels`` to refresh. ``tolerance`` is kept for API - compatibility and ignored on read (it only applies at recalculation). + Default (``tolerance is None``): read persisted levels only — no rewrite. + Pipeline/ingestion call ``recalculate_sr_levels`` after OHLCV changes. + + 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( - select(SRLevel) - .where(SRLevel.ticker_id == ticker.id) - .order_by(SRLevel.strength.desc()) - ) - return list(result.scalars().all()) + records = await query_ohlcv(db, symbol) + if not records: + return [] + _, highs, lows, closes, volumes = _extract_ohlcv(records) + detected = detect_sr_levels(highs, lows, closes, volumes, tolerance) + 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) + ] diff --git a/docs/research/README.md b/docs/research/README.md index f223296..6502978 100644 --- a/docs/research/README.md +++ b/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%** | diff --git a/frontend/src/api/admin.ts b/frontend/src/api/admin.ts index 117bdd6..d941603 100644 --- a/frontend/src/api/admin.ts +++ b/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('admin/data/cleanup', { older_than_days: olderThanDays, }) .then((r) => r.data); diff --git a/frontend/src/components/ui/Toast.tsx b/frontend/src/components/ui/Toast.tsx index 68f603b..b9a7095 100644 --- a/frontend/src/components/ui/Toast.tsx +++ b/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 = { 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 }) { diff --git a/frontend/src/hooks/useAdmin.ts b/frontend/src/hooks/useAdmin.ts index d59c6ff..7d2d6b2 100644 --- a/frontend/src/hooks/useAdmin.ts +++ b/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');