feat: near-close scan schedule and distinct-day gate reset

Move the only qualifying R:R scan to 15:30 ET with chained Telegram alerts,
put outcome eval after a final-bar OHLCV fetch, enforce NY trading-day
requalify semantics, stamp paper trades fill_mode=near_close, and migrate
stored schedule_* keys to America/New_York.

app/models/paper_trade.py

@@ -46,3 +46,6 @@ class PaperTrade(Base):
     reentry_gate_requalified_at: Mapped[datetime | None] = mapped_column(
         DateTime(timezone=True), nullable=True
     )
+    # Execution era for forward vs backtest comparison:
+    # null/legacy = pre-cutover morning-scan, "near_close" = post near-close cutover.
+    fill_mode: Mapped[str | None] = mapped_column(String(20), nullable=True)

app/scheduler.py

@@ -93,7 +93,9 @@ _JOB_NAMES = [
     "regime_monitor",
     "event_study",
     "backtest",
-    "daily_pipeline",
+    "daily_pipeline",  # morning: OHLCV/sentiment/regime — no qualifying scan
+    "near_close_pipeline",  # OHLCV fetch → R:R scan → Telegram alerts
+    "after_close_pipeline",  # OHLCV fetch → outcome eval (final bar)
     "intraday_pipeline",
 ]
 
@@ -1177,17 +1179,41 @@ async def sync_ticker_universe() -> None:
 # updates its own runtime status while the pipeline runs.
 #
 # Daily (full): the complete data→signal refresh, once a day.
+# Morning (America/New_York ~02:00): refresh data + display context. No R:R scan
+# — the qualifying full-universe scan runs once near the US close so post-stop
+# gate-reset sees one observation per trading day (plus the trade_policy
+# distinct-day guard for manual re-scans).
 _DAILY_PIPELINE_STEPS = [
     ("data_collector", "collect_ohlcv"),
     ("benchmark_collector", "collect_benchmark"),
     ("sentiment_collector", "collect_sentiment"),
-    ("rr_scanner", "scan_rr"),
-    ("outcome_evaluator", "evaluate_outcomes"),
     ("market_regime", "compute_market_regime"),
-    # Observational only — runs here for scheduling; its output feeds nothing else.
+    # Observational only — display/alerts; not trade selection.
     ("regime_monitor", "compute_regime_monitor"),
 ]
 
+# Near-close (~15:30 ET Mon–Fri): refresh in-progress day-t bars (already how
+# the intraday pipeline keeps the dashboard live), then the only daily
+# qualifying R:R scan, then Telegram immediately so manual fills can still hit
+# MOC cutoffs (~15:50/15:55). Under a 15-minute delayed SIP feed a 15:30 scan
+# may see ~15:15 prices — immaterial for a 12-1 momentum signal.
+#
+# US early-close days (~3/year, 13:00 ET close): this job runs post-close and
+# entries behave like stale_close (still acceptable per execution-recovery matrix).
+# No exchange calendar dependency.
+_NEAR_CLOSE_PIPELINE_STEPS = [
+    ("data_collector", "collect_ohlcv"),
+    ("rr_scanner", "scan_rr"),
+    ("alerts", "dispatch_alerts_job"),
+]
+
+# After close (~16:45 ET Mon–Fri): fresh OHLCV fetch so outcomes resolve on the
+# final bar, not the near-close partial bar, then outcome/paper close.
+_AFTER_CLOSE_PIPELINE_STEPS = [
+    ("data_collector", "collect_ohlcv"),
+    ("outcome_evaluator", "evaluate_outcomes"),
+]
+
 # Intraday (light): keep prices current and resolve outcomes through the day,
 # without the expensive scan/sentiment. The dashboard recomputes live R:R from
 # the latest price, so refreshing OHLCV is enough to stop prices lagging; the
@@ -1197,6 +1223,10 @@ _INTRADAY_PIPELINE_STEPS = [
     ("outcome_evaluator", "evaluate_outcomes"),
 ]
 
+# Warn if near-close fetch+scan+alert drifts past this — entries leave the close
+# and the stale_close floor quietly becomes the ceiling.
+_NEAR_CLOSE_DURATION_WARN_SECONDS = 600
+
 
 async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None:
     """Run an ordered list of (step_name, coroutine_name) steps.
@@ -1232,11 +1262,44 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None:
 
 
 async def run_daily_pipeline() -> None:
-    """Full daily flow: OHLCV → benchmark → sentiment → R:R scan → outcome eval
-    (+paper close) → market regime."""
+    """Morning flow: OHLCV → benchmark → sentiment → market regime (no scan)."""
     await _run_pipeline("daily_pipeline", _DAILY_PIPELINE_STEPS)
 
 
+async def run_near_close_pipeline() -> None:
+    """Near-close flow: OHLCV fetch → R:R scan → Telegram alerts.
+
+    Logs wall duration; warn if past 10 minutes so operators notice close drift.
+    """
+    import time
+
+    started = time.monotonic()
+    await _run_pipeline("near_close_pipeline", _NEAR_CLOSE_PIPELINE_STEPS)
+    elapsed = time.monotonic() - started
+    payload = {
+        "job": "near_close_pipeline",
+        "duration_seconds": round(elapsed, 1),
+    }
+    if elapsed > _NEAR_CLOSE_DURATION_WARN_SECONDS:
+        _log_event(
+            logging.WARNING,
+            "near_close_pipeline_slow",
+            **payload,
+            threshold_seconds=_NEAR_CLOSE_DURATION_WARN_SECONDS,
+            message=(
+                "Near-close pipeline exceeded 10 minutes — entries may drift from "
+                "the close toward the stale_close research floor"
+            ),
+        )
+    else:
+        _log_event(logging.INFO, "near_close_pipeline_duration", **payload)
+
+
+async def run_after_close_pipeline() -> None:
+    """After-close flow: OHLCV fetch (final bar) → outcome eval (+paper close)."""
+    await _run_pipeline("after_close_pipeline", _AFTER_CLOSE_PIPELINE_STEPS)
+
+
 async def run_intraday_pipeline() -> None:
     """Light intraday flow: refresh OHLCV → evaluate outcomes (+paper close)."""
     await _run_pipeline("intraday_pipeline", _INTRADAY_PIPELINE_STEPS)
@@ -1268,16 +1331,28 @@ def _parse_frequency(freq: str) -> dict[str, int]:
 # every process restart, so on a box that's redeployed often it can keep being
 # deferred and never fire. Cron fires at a fixed local time regardless.
 
+# All wall times are America/New_York after the near-close execution cutover.
+# Stored SystemSetting values shadow these defaults — deploy migration 023
+# rewrites schedule_* keys so prod does not keep scanning at 07:00 Berlin.
 SCHEDULE_DEFAULTS: dict[str, str] = {
-    "schedule_timezone": "Europe/Berlin",
-    "schedule_daily_pipeline_cron": "0 7 * * *",        # full refresh, ready by ~8am
-    "schedule_intraday_pipeline_cron": "0 14-22 * * 1-5",  # hourly across the US session
-    "schedule_fundamentals_cron": "0 4 * * 1",          # weekly, early Monday (slow job)
+    "schedule_timezone": "America/New_York",
+    # Morning data/display refresh (no qualifying R:R scan).
+    "schedule_daily_pipeline_cron": "0 2 * * *",
+    # Fetch in-progress bars → scan → Telegram (manual MOC window).
+    "schedule_near_close_pipeline_cron": "30 15 * * 1-5",
+    # Fetch final bars → outcome eval (must not run on the partial near-close bar).
+    "schedule_after_close_pipeline_cron": "45 16 * * 1-5",
+    # Hourly mid-session price + outcome (10:00–15:00 ET Mon–Fri).
+    "schedule_intraday_pipeline_cron": "0 10-15 * * 1-5",
+    # Weekly fundamentals early Monday NY.
+    "schedule_fundamentals_cron": "0 1 * * 1",
 }
 
 # job id -> schedule setting key
 _CRON_JOBS: dict[str, str] = {
     "daily_pipeline": "schedule_daily_pipeline_cron",
+    "near_close_pipeline": "schedule_near_close_pipeline_cron",
+    "after_close_pipeline": "schedule_after_close_pipeline_cron",
     "intraday_pipeline": "schedule_intraday_pipeline_cron",
     "fundamental_collector": "schedule_fundamentals_cron",
 }
@@ -1357,7 +1432,29 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
     scheduler.add_job(
         run_daily_pipeline,
         _cron_trigger(cfg["schedule_daily_pipeline_cron"], tz, "schedule_daily_pipeline_cron"),
-        id="daily_pipeline", name="Daily Pipeline", replace_existing=True,
+        id="daily_pipeline", name="Morning Pipeline", replace_existing=True,
+    )
+    scheduler.add_job(
+        run_near_close_pipeline,
+        _cron_trigger(
+            cfg["schedule_near_close_pipeline_cron"],
+            tz,
+            "schedule_near_close_pipeline_cron",
+        ),
+        id="near_close_pipeline",
+        name="Near-Close Pipeline (scan+alert)",
+        replace_existing=True,
+    )
+    scheduler.add_job(
+        run_after_close_pipeline,
+        _cron_trigger(
+            cfg["schedule_after_close_pipeline_cron"],
+            tz,
+            "schedule_after_close_pipeline_cron",
+        ),
+        id="after_close_pipeline",
+        name="After-Close Pipeline (outcome)",
+        replace_existing=True,
     )
     scheduler.add_job(
         run_intraday_pipeline,
@@ -1377,10 +1474,12 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
         sync_ticker_universe, "interval", hours=24,
         id="ticker_universe_sync", name="Ticker Universe Sync", replace_existing=True,
     )
-    alerts_interval = _parse_frequency(settings.alerts_frequency)
+    # Alerts auto-fire only via near_close_pipeline (scan → alert before MOC).
+    # Keep the job registered for Admin manual trigger; no independent interval.
     scheduler.add_job(
-        dispatch_alerts_job, "interval", **alerts_interval,
-        id="alerts", name="Alerts Dispatcher", replace_existing=True,
+        dispatch_alerts_job, "interval", weeks=520,
+        id="alerts", name="Alerts Dispatcher",
+        replace_existing=True, next_run_time=None,
     )
     scheduler.add_job(
         run_backtest_job, "interval", hours=168,
@@ -1400,10 +1499,27 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
         replace_existing=True, next_run_time=None,
     )
 
-    _log_event(logging.INFO, "scheduler_configured", timezone=tz, daily_pipeline={
-                "cron": cfg["schedule_daily_pipeline_cron"],
-                "steps": [name for name, _ in _DAILY_PIPELINE_STEPS],
-            }, intraday_pipeline={
-                "cron": cfg["schedule_intraday_pipeline_cron"],
-                "steps": [name for name, _ in _INTRADAY_PIPELINE_STEPS],
-            }, fundamental_collector={"cron": cfg["schedule_fundamentals_cron"]}, independent=["ticker_universe_sync", "alerts", "backtest"])
+    _log_event(
+        logging.INFO,
+        "scheduler_configured",
+        timezone=tz,
+        daily_pipeline={
+            "cron": cfg["schedule_daily_pipeline_cron"],
+            "steps": [name for name, _ in _DAILY_PIPELINE_STEPS],
+        },
+        near_close_pipeline={
+            "cron": cfg["schedule_near_close_pipeline_cron"],
+            "steps": [name for name, _ in _NEAR_CLOSE_PIPELINE_STEPS],
+        },
+        after_close_pipeline={
+            "cron": cfg["schedule_after_close_pipeline_cron"],
+            "steps": [name for name, _ in _AFTER_CLOSE_PIPELINE_STEPS],
+        },
+        intraday_pipeline={
+            "cron": cfg["schedule_intraday_pipeline_cron"],
+            "steps": [name for name, _ in _INTRADAY_PIPELINE_STEPS],
+        },
+        fundamental_collector={"cron": cfg["schedule_fundamentals_cron"]},
+        independent=["ticker_universe_sync", "backtest"],
+        manual_only=["alerts", "data_backfill", "event_study"],
+    )

app/schemas/admin.py

@@ -75,9 +75,11 @@ class ActivationConfigUpdate(BaseModel):
 
 class ScheduleConfigUpdate(BaseModel):
     """Cron schedule for the pipelines + fundamentals. Crons are 5-field
-    (min hour dom month dow); timezone is an IANA name (e.g. Europe/Berlin)."""
+    (min hour dom month dow); timezone is an IANA name (e.g. America/New_York)."""
     schedule_timezone: str | None = Field(default=None, max_length=64)
     schedule_daily_pipeline_cron: str | None = Field(default=None, max_length=120)
+    schedule_near_close_pipeline_cron: str | None = Field(default=None, max_length=120)
+    schedule_after_close_pipeline_cron: str | None = Field(default=None, max_length=120)
     schedule_intraday_pipeline_cron: str | None = Field(default=None, max_length=120)
     schedule_fundamentals_cron: str | None = Field(default=None, max_length=120)
 

app/schemas/paper_trade.py

@@ -47,6 +47,8 @@ class PaperTradeResponse(BaseModel):
     alpha_pct: float | None = None
     alpha_usd: float | None = None
     close_reason: str | None = None
+    # Execution era: null = pre-cutover / unknown; "near_close" = post schedule cutover.
+    fill_mode: str | None = None
     # Live trailing-stop level + how far price sits above it (% ), for open trades
     # when the trailing exit policy is active.
     trailing_stop: float | None = None

app/services/admin_service.py

@@ -566,6 +566,8 @@ VALID_JOB_NAMES = {
     "event_study",
     "backtest",
     "daily_pipeline",
+    "near_close_pipeline",
+    "after_close_pipeline",
     "intraday_pipeline",
 }
 
@@ -583,17 +585,20 @@ JOB_LABELS = {
     "regime_monitor": "Regime Monitor",
     "event_study": "Event Study",
     "backtest": "Backtest",
-    "daily_pipeline": "Daily Pipeline",
+    "daily_pipeline": "Morning Pipeline",
+    "near_close_pipeline": "Near-Close Pipeline (scan+alert)",
+    "after_close_pipeline": "After-Close Pipeline (outcome)",
     "intraday_pipeline": "Intraday Pipeline",
 }
 
-# Jobs driven by the daily_pipeline (in order) rather than their own timer.
+# Jobs driven by a pipeline (in order) rather than their own auto timer.
 PIPELINE_MEMBERS = {
     "data_collector",
     "benchmark_collector",
     "sentiment_collector",
     "rr_scanner",
     "outcome_evaluator",
+    "alerts",
     "market_regime",
     "regime_monitor",
 }

app/services/paper_trade_service.py

@@ -333,6 +333,9 @@ async def create_trade(
         target=target,
         status="open",
         opened_at=datetime.now(timezone.utc),
+        # Near-close cutover era — Track Record must not mix with morning-scan
+        # fills or future broker-routed fills when comparing to backtests.
+        fill_mode="near_close",
     )
     db.add(trade)
     await db.commit()
@@ -386,6 +389,7 @@ def _to_dict(
         "alpha_pct": alpha_pct,
         "alpha_usd": alpha_usd,
         "close_reason": trade.close_reason,
+        "fill_mode": trade.fill_mode,
         "trailing_stop": trailing[0] if trailing else None,
         "trailing_distance_pct": trailing[1] if trailing else None,
     }

app/services/trade_policy.py

@@ -3,13 +3,24 @@
 from __future__ import annotations
 
 from collections.abc import Iterable
-from datetime import datetime, timezone
+from datetime import date, datetime, timezone
+from zoneinfo import ZoneInfo
 
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models.paper_trade import PaperTrade
 
+# Gate-reset "day" boundary matches US cash equities session calendar, not UTC.
+_REENTRY_DAY_TZ = ZoneInfo("America/New_York")
+
+
+def _ny_trading_date(moment: datetime) -> date:
+    """Calendar date in America/New_York for a gate-reset observation."""
+    if moment.tzinfo is None:
+        moment = moment.replace(tzinfo=timezone.utc)
+    return moment.astimezone(_REENTRY_DAY_TZ).date()
+
 
 async def _latest_initial_stop_trades(
     db: AsyncSession,
@@ -93,8 +104,14 @@ async def observe_reentry_gate_transitions(
                 trade.reentry_gate_failed_at = timestamp
                 updated.add(ticker_id)
         elif ticker_id in qualified:
-            trade.reentry_gate_requalified_at = timestamp
-            updated.add(ticker_id)
+            # Study semantics: requalify only on a *subsequent* daily observation.
+            # Same America/New_York calendar day as the failure does not unlock,
+            # even if multiple full-universe scans run (manual + near-close).
+            if _ny_trading_date(trade.reentry_gate_failed_at) < _ny_trading_date(
+                timestamp
+            ):
+                trade.reentry_gate_requalified_at = timestamp
+                updated.add(ticker_id)
 
     if updated:
         await db.flush()