diff --git a/app/scheduler.py b/app/scheduler.py index 629c0bd..400b53f 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -16,6 +16,7 @@ from __future__ import annotations import json import logging import asyncio +import contextvars from datetime import date, datetime, timedelta, timezone from apscheduler.schedulers.asyncio import AsyncIOScheduler @@ -625,9 +626,16 @@ async def run_shadow_book() -> None: prices the discretionary book sees, leaving *selection* as the only difference between the two books. + When run as a pipeline step it only acts on a scan that completed *inside + this pipeline pass* (``require_scan_after``): if the pipeline's own scan was + disabled or failed, an earlier still-fresh manual scan must not stand in for + it. Triggered directly from Admin (no pipeline context) it falls back to the + scan-freshness window — an explicit operator action. + Opt-in (``shadow_book_enabled``) because it writes live trades. """ job_name = "shadow_book" + require_scan_after = _pipeline_started_at.get() _log_event(logging.INFO, "job_start", job=job_name) _runtime_start(job_name, total=1) @@ -646,7 +654,9 @@ async def run_shadow_book() -> None: activation_config = await get_activation_config(db) summary = await shadow_book_service.open_shadow_positions( - db, activation_config=activation_config + db, + activation_config=activation_config, + require_scan_after=require_scan_after, ) symbols = await shadow_book_service.symbols_for(db, summary["symbols"]) @@ -1319,6 +1329,15 @@ _INTRADAY_PIPELINE_STEPS = [ _NEAR_CLOSE_DURATION_WARN_SECONDS = 600 +# Wall-clock start of the pipeline invocation currently running, visible to its +# steps via the shared task context. The shadow book uses it to require that the +# scan it acts on completed *inside this pipeline pass* — a fresh but earlier +# manual scan must not stand in for a scan that failed or was disabled here. +_pipeline_started_at: contextvars.ContextVar[datetime | None] = contextvars.ContextVar( + "pipeline_started_at", default=None +) + + async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None: """Run an ordered list of (step_name, coroutine_name) steps. @@ -1337,6 +1356,7 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None: funcs = globals() done = 0 + token = _pipeline_started_at.set(datetime.now(timezone.utc)) try: for step_name, func_name in steps: _runtime_progress(job_name, processed=done, total=total, current_ticker=step_name) @@ -1350,6 +1370,8 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None: except Exception as exc: _runtime_finish(job_name, "error", processed=done, total=total, message=str(exc)) _log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc)) + finally: + _pipeline_started_at.reset(token) async def run_daily_pipeline() -> None: diff --git a/app/services/shadow_book_service.py b/app/services/shadow_book_service.py index 705b3a1..53b8a9f 100644 --- a/app/services/shadow_book_service.py +++ b/app/services/shadow_book_service.py @@ -169,14 +169,28 @@ async def _shadow_user_id(db: AsyncSession) -> int | None: return int(row[0]) if row else None -async def _last_scan_start(db: AsyncSession, *, now: datetime) -> datetime | None: - """Start of the last successful scan, if it ran in this pipeline pass. +async def _last_scan_start( + db: AsyncSession, + *, + now: datetime, + require_scan_after: datetime | None = None, +) -> datetime | None: + """Start of the last successful scan, if it is the one we may act on. - Returns None — meaning "no scan to act on" — unless the scanner's COMPLETED - marker is fresh. Pipeline steps fail independently, so a scan that was - disabled, errored, or produced nothing leaves a stale marker; trading on the - newest stored setups then would enter a previous session's picks at stale - prices. Freshness is proven by the marker, not by setup age. + Returns None — "no scan to act on" — unless the scanner's COMPLETED marker + passes the applicable check: + + * ``require_scan_after`` set (pipeline step): the scan must have completed + at/after the pipeline began. This is the airtight guarantee — a scan that + was disabled or failed in *this* pipeline leaves the marker at a previous + run, and a fresh but earlier manual scan completed before the pipeline + started, so neither can stand in for the pipeline's own scan. + * ``require_scan_after`` None (direct Admin trigger): fall back to the + freshness window. There is no pipeline scan to bind to, so acting on a + recent scan is the operator's explicit choice. + + Either way the returned value is the scan's start, the lower bound for the + setups belonging to that run. """ from app.services import rr_scanner_service as rr @@ -186,7 +200,10 @@ async def _last_scan_start(db: AsyncSession, *, now: datetime) -> datetime | Non ) if started is None or completed is None: return None - if now - completed > MAX_SCAN_AGE: + if require_scan_after is not None: + if completed < require_scan_after: + return None + elif now - completed > MAX_SCAN_AGE: return None return started @@ -201,7 +218,11 @@ def _parse_dt(raw: str | None) -> datetime | None: async def _todays_qualified_setups( - db: AsyncSession, config: dict, *, now: datetime + db: AsyncSession, + config: dict, + *, + now: datetime, + require_scan_after: datetime | None = None, ) -> list[TradeSetup]: """Long-only qualified setups from the scan that just ran, best rank first. @@ -219,7 +240,9 @@ async def _todays_qualified_setups( the reverse. 4. Qualify, then rank by ``strategy_rank`` (unranked sort last). """ - run_start = await _last_scan_start(db, now=now) + run_start = await _last_scan_start( + db, now=now, require_scan_after=require_scan_after + ) if run_start is None: return [] @@ -252,12 +275,17 @@ async def open_shadow_positions( *, activation_config: dict, opened_at: datetime | None = None, + require_scan_after: datetime | None = None, ) -> dict: """Fill free capacity with the top-ranked qualified setups. Mirrors the backtest: rank the qualified cross-section, walk it top-down, skip anything already held or locked out by post-stop gate-reset, and stop at capacity. Returns a summary for the job log. + + ``require_scan_after`` binds this run to a scan that completed at/after that + instant (the pipeline's start), so a stale or manual scan cannot substitute + for a scan that failed in this pipeline pass. See ``_last_scan_start``. """ summary = { "opened": 0, @@ -283,7 +311,10 @@ async def open_shadow_positions( equity, cash = await equity_and_cash(db, config["start_equity"], positions) timestamp = opened_at or datetime.now(timezone.utc) - for setup in await _todays_qualified_setups(db, activation_config, now=timestamp): + candidates = await _todays_qualified_setups( + db, activation_config, now=timestamp, require_scan_after=require_scan_after + ) + for setup in candidates: if free_slots <= 0: break if setup.ticker_id in held: diff --git a/tests/unit/test_shadow_book_service.py b/tests/unit/test_shadow_book_service.py index 33043d2..4e0f791 100644 --- a/tests/unit/test_shadow_book_service.py +++ b/tests/unit/test_shadow_book_service.py @@ -283,6 +283,44 @@ class TestScanFreshness: assert summary["opened"] == 0 +class TestPipelineScanBinding: + @pytest.mark.asyncio + async def test_fresh_manual_scan_before_pipeline_is_refused(self, session): + """A manual scan at 13:00 is still 'fresh' at 15:30, but the 15:30 + pipeline's own scan failed. Binding to the pipeline start rejects the + 13:00 batch — no successful scan happened in *this* pipeline pass.""" + ids = await _seed(session, ["AAA"]) + pipeline_start = datetime.now(timezone.utc) + manual_scan = pipeline_start - timedelta(hours=2, minutes=30) + session.add(_setup(ids["AAA"], rank=0.9, detected=manual_scan)) + await session.commit() + await _mark_scan(session, started=manual_scan - timedelta(minutes=5), + completed=manual_scan) + + summary = await shadow_book_service.open_shadow_positions( + session, activation_config=_CONFIG, require_scan_after=pipeline_start + ) + + assert summary["opened"] == 0 + + @pytest.mark.asyncio + async def test_pipeline_scan_after_start_is_accepted(self, session): + """The pipeline's own scan completes just after the pipeline began.""" + ids = await _seed(session, ["AAA"]) + pipeline_start = datetime.now(timezone.utc) + scan_completed = pipeline_start + timedelta(minutes=1) + session.add(_setup(ids["AAA"], rank=0.9, detected=scan_completed)) + await session.commit() + await _mark_scan(session, started=pipeline_start + timedelta(seconds=1), + completed=scan_completed) + + summary = await shadow_book_service.open_shadow_positions( + session, activation_config=_CONFIG, require_scan_after=pipeline_start + ) + + assert summary["opened"] == 1 + + class TestLongOnly: @pytest.mark.asyncio async def test_shorts_are_never_taken_even_with_gate_disabled(self, session):