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.

README.md

@@ -60,11 +60,11 @@ flowchart TD
 
 **What happens after an initial stop.** The stop always closes the trade and realizes its costs. The ticker is then locked until a successful daily full-universe scan first observes it outside the production gate and a later scan observes a fresh qualification. A continuously qualified ticker therefore cannot generate an immediate duplicate entry. Other exit reasons do not start this reset. See the [daily post-stop re-entry study](docs/research/post-stop-reentry.md).
 
-**Live timing matters.** The full daily pipeline runs the R:R scan before Outcome Eval. A stop closed by that Outcome Eval—or by an intraday evaluation after the day's full scan—therefore cannot use its stop-day gate state. The earliest failure observation is the next successful full scan, and requalification needs a subsequent full scan. The research `gate_reset` arm evaluated the stop before its same-session gate check; the live boundary is consequently analogous to the study's stricter `strict_gate_reset` arm. This known event-ordering difference is quantified below.
+**Live timing matters.** The **only** full-universe R:R scan runs near the US close (~15:30 ET), then Telegram alerts fire immediately so manual fills can still hit MOC. Outcome eval runs later (~16:45 ET) after a fresh OHLCV fetch of the final bar. Morning jobs refresh data/sentiment/regime without scanning. Stops closed by earlier same-day intraday outcome evals can get a **same-day** fail observation at the near-close scan — closer to the promoted research `gate_reset` arm than the old morning-scan `strict_gate_reset` analogue. Stops after the bell still need a later day. Same-day fail+qualify cannot unlock: `trade_policy` requires the failure to fall on an earlier America/New_York trading date.
 
 ## How It Works
 
-Scheduled pipelines turn raw prices into a ranked, gated list of tradeable setups. Everything downstream of OHLCV is recomputed from stored data, so each refresh is cheap and idempotent. Job timing is cron-based and configurable in **Admin → Jobs** (default timezone Europe/Berlin).
+Scheduled pipelines turn raw prices into a ranked, gated list of tradeable setups. Everything downstream of OHLCV is recomputed from stored data, so each refresh is cheap and idempotent. Job timing is cron-based and configurable in **Admin → Jobs** (default timezone **America/New_York** so the near-close scan tracks the cash close through DST).
 
 ### Price-level architecture: two different jobs
 
@@ -127,26 +127,34 @@ percentile rails show each input. Only momentum carries the live activation-
 gate marker. These are cross-sectional scan percentiles, not historical chart
 indicators.
 
-### Daily Load — the full refresh
+### Pipelines (America/New_York)
 
-Once a day (default 07:00). Steps run **in dependency order**, each consuming the previous step's fresh output:
+**Morning** (~02:00 ET) — data and display only, **no** qualifying R:R scan:
 
-1. **OHLCV** — fetch the latest daily bars for every tracked ticker (Alpaca); new tickers backfill ~5 years.
+1. **OHLCV** — latest daily bars (Alpaca); new tickers backfill ~5 years.
-2. **Sentiment** — fetch sentiment for the names that matter and are stale (> 5 days): top-pick feeders (residual-momentum leaders with a tradeable long setup), the watchlist, and open paper trades, plus a top-N-by-composite discovery net. Runs *before* the scan so the scan sees fresh sentiment.
+2. **Sentiment** — stale names that matter (top-pick feeders, watchlist, open paper, discovery net). Display context only; the activation gate is price-only.
-3. **R:R Scan** — persist clean Structural S/R for charts/alerts, recompute the 5-dimension scores, and build long/short setups from a transient Gate Target Ladder (ATR stops and nominal gate targets) for every ticker. Attach each ticker's residual 12‑1 momentum activation percentile plus the promoted 80/20 production rank. The completed full-universe scan also advances post-stop locks from gate failure to later requalification; failed scans never count as a transition.
+3. **Market Regime** + **Regime Monitor** — breadth/trend and the v2 risk thermometer; feed no trades.
-4. **Outcome Eval** — resolve setups that hit target/stop or expired (default 30 trading days) and auto-close paper trades per the exit policy (default: 3x ATR trail with a 30-trading-day max hold).
-5. **Market Regime** — recompute the regime index (breadth/trend).
-6. **Regime Monitor** — separate v2 State/Warning risk thermometer with fixed-basket breadth, VIX, credit, and point-in-time fundamentals; feeds no trades.
 
-A failing step is logged; the pipeline continues with the next.
+**Near-close** (~15:30 ET Mon–Fri) — the only full-universe qualifying observation:
+
+1. **OHLCV fetch** — refresh the in-progress day-t bar (same path as intraday).
+2. **R:R Scan** — Structural S/R, scores, Gate Target Ladder setups, residual 12‑1 + 80/20 rank. Advances post-stop gate-reset transitions; failed scans never count.
+3. **Telegram alerts** — chained immediately so manual MOC fills can still hit ~15:50/15:55.
+
+**After close** (~16:45 ET Mon–Fri):
+
+1. **OHLCV fetch** — final bar (not the partial near-close bar).
+2. **Outcome Eval** — resolve setups and auto-close paper trades (default 3× ATR trail, 30-day max hold).
+
+A failing step is logged; the pipeline continues with the next. Near-close duration is logged; warn if > 10 minutes.
 
 ### Intraday — light refresh
 
-Hourly across the US session (Mon–Fri): only **OHLCV → Outcome Eval**, to keep prices current and close paper trades intraday. No scan/sentiment — the dashboard recomputes live R:R from the latest price, so fresh prices are enough.
+Hourly mid-session (Mon–Fri ~10:00–15:00 ET): only **OHLCV → Outcome Eval**, to keep prices current and close paper trades intraday. No scan/sentiment — the dashboard recomputes live R:R from the latest price.
 
 ### Other jobs
 
-Fundamentals (weekly, early Monday) · Alerts (hourly, Telegram) · Backtest (weekly) · Ticker-universe sync (daily). Deep history backfill and event study are manual-only (Admin → Jobs).
+Fundamentals (weekly, early Monday ET) · Backtest (weekly) · Ticker-universe sync (daily). Alerts auto-fire only via the near-close pipeline (still manually triggerable). Deep history backfill and event study are manual-only (Admin → Jobs).
 
 ### From score to "top pick"
 
@@ -193,7 +201,7 @@ The production policy is **normal gate reset**, evaluated with daily setup oppor
 
 In the disjoint 2025+ book, gate reset also beat immediate re-entry (Sharpe 1.66 vs 1.55; CAGR 41.8% vs 39.3%) and the fixed five-session rule (Sharpe 1.43; CAGR 32.7%). Its lead over both survived costs of 0.2% and 0.3% per side. The result is capacity-specific: cooldown 5 won at capacity 5, while immediate had slightly higher return and Sharpe at capacity 15. Production uses capacity 10, so that is the portfolio for which this decision is valid.
 
-Those promotion numbers belong to the selected normal-reset study arm. Under the live scheduler's stricter first-observation timing, the full-period analogue was Sharpe 1.68 / CAGR 44.8% / DD 23.4%; in the disjoint 2025+ book it was Sharpe 1.38 / CAGR 32.9% / DD 21.0%. The matrix therefore validates the state-machine choice but is not exact scheduler-order parity. Closing this timing gap would require a separately reviewed pipeline-order change, not a documentation reinterpretation.
+Those promotion numbers belong to the selected normal-reset study arm. Under the **pre-cutover** morning-scan scheduler (scan always before any outcome eval), live first-observation timing matched the stricter `strict_gate_reset` analogue (full-period Sharpe 1.68 / CAGR 44.8% / DD 23.4%). After the **near-close cutover** (2026-07), stops closed by earlier same-day intraday evals can receive a same-day fail observation at ~15:30 ET — moving live behavior **toward** the promoted `gate_reset` arm. Requalification still requires a later America/New_York trading date than the failure (`trade_policy` distinct-day guard). Full definitions and all nine policy arms: [docs/research/post-stop-reentry.md](docs/research/post-stop-reentry.md); execution evidence: [docs/research/execution-recovery.md](docs/research/execution-recovery.md).
 
 `gate_reset` and a simple `next_session` block happened to produce the same executed live-universe portfolio in this sample. Their rules are still different: this establishes that same-day re-entry was harmful here, but does not isolate a separate historical return premium from the reset condition. Gate reset was promoted because it represents a genuinely new signal episode and did not sacrifice results in the production book. Full definitions, all nine policy arms, cost/capacity sensitivity, and legacy-rank results are in [docs/research/post-stop-reentry.md](docs/research/post-stop-reentry.md); source report: [`reports/daily_reentry_matrix.json`](reports/daily_reentry_matrix.json).
 

alembic/versions/023_near_close_schedule_and_fill_mode.py

@@ -0,0 +1,73 @@
+"""near-close schedule cutover + paper trade fill_mode era tag
+
+Revision ID: 023
+Revises: 022
+Create Date: 2026-07-18 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "023"
+down_revision: Union[str, None] = "022"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+# Deliberate schedule rewrite (not a soft defaults refresh). Old stored values
+# are logged then replaced so prod does not keep scanning at 07:00 Berlin.
+_SCHEDULE_REWRITE: dict[str, str] = {
+    "schedule_timezone": "America/New_York",
+    "schedule_daily_pipeline_cron": "0 2 * * *",
+    "schedule_near_close_pipeline_cron": "30 15 * * 1-5",
+    "schedule_after_close_pipeline_cron": "45 16 * * 1-5",
+    "schedule_intraday_pipeline_cron": "0 10-15 * * 1-5",
+    "schedule_fundamentals_cron": "0 1 * * 1",
+}
+
+
+def upgrade() -> None:
+    op.add_column(
+        "paper_trades",
+        sa.Column("fill_mode", sa.String(length=20), nullable=True),
+    )
+
+    conn = op.get_bind()
+    settings = sa.table(
+        "system_settings",
+        sa.column("id", sa.Integer),
+        sa.column("key", sa.String),
+        sa.column("value", sa.Text),
+        sa.column("updated_at", sa.DateTime(timezone=True)),
+    )
+    now = sa.func.now()
+
+    for key, new_value in _SCHEDULE_REWRITE.items():
+        row = conn.execute(
+            sa.select(settings.c.value).where(settings.c.key == key)
+        ).fetchone()
+        old_value = row[0] if row is not None else None
+        # Always log so ops can recover the pre-cutover schedule from migration output.
+        print(
+            f"schedule_cutover {key}: {old_value!r} -> {new_value!r}",
+            flush=True,
+        )
+        if row is None:
+            conn.execute(
+                sa.insert(settings).values(
+                    key=key, value=new_value, updated_at=now
+                )
+            )
+        else:
+            conn.execute(
+                sa.update(settings)
+                .where(settings.c.key == key)
+                .values(value=new_value, updated_at=now)
+            )
+
+
+def downgrade() -> None:
+    op.drop_column("paper_trades", "fill_mode")
+    # Do not restore old crons — unknown prior values; leave stored schedule as-is.

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
+    """Morning flow: OHLCV → benchmark → sentiment → market regime (no scan)."""
-    (+paper close) → market regime."""
     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_timezone": "America/New_York",
-    "schedule_daily_pipeline_cron": "0 7 * * *",        # full refresh, ready by ~8am
+    # Morning data/display refresh (no qualifying R:R scan).
-    "schedule_intraday_pipeline_cron": "0 14-22 * * 1-5",  # hourly across the US session
+    "schedule_daily_pipeline_cron": "0 2 * * *",
-    "schedule_fundamentals_cron": "0 4 * * 1",          # weekly, early Monday (slow job)
+    # 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,
+        dispatch_alerts_job, "interval", weeks=520,
-        id="alerts", name="Alerts Dispatcher", replace_existing=True,
+        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={
+    _log_event(
-                "cron": cfg["schedule_daily_pipeline_cron"],
+        logging.INFO,
-                "steps": [name for name, _ in _DAILY_PIPELINE_STEPS],
+        "scheduler_configured",
-            }, intraday_pipeline={
+        timezone=tz,
-                "cron": cfg["schedule_intraday_pipeline_cron"],
+        daily_pipeline={
-                "steps": [name for name, _ in _INTRADAY_PIPELINE_STEPS],
+            "cron": cfg["schedule_daily_pipeline_cron"],
-            }, fundamental_collector={"cron": cfg["schedule_fundamentals_cron"]}, independent=["ticker_universe_sync", "alerts", "backtest"])
+            "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
+            # Study semantics: requalify only on a *subsequent* daily observation.
-            updated.add(ticker_id)
+            # 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()

docs/research/execution-recovery.md

@@ -99,40 +99,79 @@ tail-trimming until shown otherwise.
 
 ---
 
-## Ops design — settle before touching the scheduler
+## Ops design — implementation plan (code-checked)
 
-### 1. One canonical qualifying scan per day (non-negotiable)
+Assumptions verified against current code before ship:
 
-Post-stop gate-reset was validated on **one full-universe observation per day**.
+- Intraday pipeline already fetches/upserts the **in-progress day-t bar** all
-Adding a near-close scan *alongside* the 07:00 scan would let fail→qualify
+  session (`fetch_ohlcv` end_date defaults to today). Near-close job =
-transitions complete twice as fast and **silently change** the validated re-entry
+  **OHLCV fetch → R:R scan** (no new snapshot synthesizer).
-policy.
+- One global `schedule_timezone` (default `Europe/Berlin`); stored
+  `SystemSetting` values shadow code defaults — **defaults alone do not
+  migrate prod**.
+- `observe_reentry_gate_transitions` stamps timestamps with **no same-day
+  guard** today — dual scans would accelerate fail→requalify unless fixed in
+  `trade_policy`.
 
-**Move** the R:R scan to near-close. **Leave** sentiment / fundamentals /
+### Semantic guard (ship step 1 — precondition)
-OHLCV-backfill at 07:00 (or existing early slots). Preserve **scan before
-Outcome Eval** (scan late session, eval after close) so the documented
-strict-gate-reset live analogue is unchanged.
 
-### 2. Cron in `America/New_York`, not `Europe/Berlin`
+In `trade_policy` (not the scheduler):
 
-DST offsets shift on different dates. A Berlin-fixed wall time drifts ~1 hour
+> `reentry_gate_requalified_at` may only be set when `reentry_gate_failed_at`
-off the US close for a week or two twice a year.
+> falls on an **earlier America/New_York trading date** than the current
+> observation.
 
-### 3. Fill mechanics and feed honesty
+Manual mid-day scans stay allowed; same-day fail+qualify cannot unlock.
+Unit test: fail 10:00 / qualify 15:35 same day → still locked; qualify next day → unlocked.
 
-- NYSE MOC cutoff **15:50 ET**; Nasdaq **15:55 ET**.  
+### Schedule split
-- Scan ~**15:30–15:40 ET** on a latest-price snapshot; place entries by 15:50.  
-- For now: paper-trade entries marked at the **actual close**.  
-- Document Alpaca entitlement: if only 15-minute-delayed SIP, a 15:35 scan sees
-  ~15:20 prices — immaterial for a 12-month signal, but write it down so nobody
-  treats it as a bug.
 
-### 4. Partial-bar plumbing
+| Slot (America/New_York) | Jobs |
+|---|---|
+| Morning (~02:00) | OHLCV backfill, benchmark, sentiment, fundamentals — **no** qualifying R:R scan |
+| Near-close (~15:30 Mon–Fri) | OHLCV fetch (refresh day-t bar) → **R:R scan** (only daily qualifying observation) |
+| After close (~16:30–17:00) | **Outcome eval** on its own slot (not chained to the partial-bar scan) |
+| Intraday hourly | Unchanged in NY terms; last ~16:00 still mid-session under 15m feed |
 
-- Scan synthesizes day-t’s in-progress bar from the snapshot.  
+- Near-close scan **1–5 only**; US-holiday no-ops are fine (stale identical data
-- Nightly OHLCV job overwrites with the final bar via existing idempotent upsert.  
+  can’t flip gates) — comment only, no exchange calendar.
-- Forward paper record marks entries at the actual near-close fill so the live
+- **Do not** run morning + near-close qualifying scans; move the scan, don’t add a second.
-  track measures the new execution honestly.
+
+### Behavior change to document (not an accident)
+
+With scan at ~15:35 ET, stops closed by **earlier same-day** intraday outcome
+evals can get a **same-day fail observation** — closer to the **promoted**
+`gate_reset` arm (stop-day close may establish failure) than today’s
+`strict_gate_reset` analogue (scan always before any eval). Stops after the
+bell still wait a day. Rewrite README “Live timing matters” / post-stop sections
+and a line here when shipping.
+
+### Feed / paper honesty
+
+- Document 15-minute delayed SIP: 15:35 scan may see ~15:20 prices; OK for 12-1.
+- Paper entry price ≈ scan entry (near close) is nearly automatic; add
+  **`fill_mode=near_close` era tag** so Track Record can separate morning-scan /
+  near-close / future broker-routed eras.
+- Morning sentiment staleness is display-only; gate is price-only (GTL parity /
+  neutral-sentiment backtest). One doc line closes that.
+
+### Stored settings migration (ship step 4)
+
+Flip global default TZ to `America/New_York` and re-express crons in NY time.
+**Also** migration (or documented Admin rewrite) of stored `schedule_*` keys so
+prod does not keep 07:00 Berlin silently.
+
+### Ship order
+
+1. `trade_policy` distinct-day requalify guard + unit test  
+2. Near-close job = existing fetch → scan; outcome eval own after-close slot  
+3. Paper `fill_mode=near_close` era tag; verify entry marking  
+4. Defaults + **stored settings migration** + README/research timing rewrite  
+
+### Out of scope
+
+Broker MOC routing, more fill-timing sim, nasdaq_all / fip / sector (grade later
+under the fill mode you trade).
 
 ---
 
@@ -141,6 +180,13 @@ off the US close for a week or two twice a year.
 | Item | Status |
 |---|---|
 | Research evidence | **Closed** — this doc + matrix report |
-| Scheduler move (R:R scan → NY near-close) | **Not started** — blocked on ops design above |
+| Distinct-day gate-reset guard | **Shipped** — `trade_policy` + unit test |
-| Partial-bar scan path | **Not started** |
+| Schedule split + near-close scan→alert | **Shipped** — morning / near-close / after-close (fetch→outcome) |
-| Paper fill-at-close marking | **Not started** (may already mark at close; verify when shipping) |
+| Paper era tag | **Shipped** — `fill_mode=near_close` on new paper trades |
+| Settings migration + docs | **Shipped** — alembic 023 rewrites schedule_*; README updated |
+
+### Shipped behavior change (not accidental)
+
+Near-close scan at ~15:30 ET lets same-day fail observations after earlier
+intraday stop closes — closer to promoted `gate_reset` than the old
+morning-scan `strict_gate_reset` analogue. Documented in README.

frontend/src/components/admin/ScheduleSettings.tsx

@@ -4,34 +4,48 @@ import { useScheduleSettings, useUpdateScheduleSettings } from '../../hooks/useA
 import { SkeletonTable } from '../ui/Skeleton';
 
 const DEFAULTS: ScheduleConfig = {
-  schedule_timezone: 'Europe/Berlin',
+  schedule_timezone: 'America/New_York',
-  schedule_daily_pipeline_cron: '0 7 * * *',
+  schedule_daily_pipeline_cron: '0 2 * * *',
-  schedule_intraday_pipeline_cron: '0 14-22 * * 1-5',
+  schedule_near_close_pipeline_cron: '30 15 * * 1-5',
-  schedule_fundamentals_cron: '0 4 * * 1',
+  schedule_after_close_pipeline_cron: '45 16 * * 1-5',
+  schedule_intraday_pipeline_cron: '0 10-15 * * 1-5',
+  schedule_fundamentals_cron: '0 1 * * 1',
 };
 
 const FIELDS: { key: keyof ScheduleConfig; label: string; hint: string; mono?: boolean }[] = [
   {
     key: 'schedule_timezone',
     label: 'Timezone',
-    hint: 'IANA name, e.g. Europe/Berlin. All times below are in this zone.',
+    hint: 'IANA name. Prefer America/New_York so the near-close scan tracks the US cash close through DST.',
   },
   {
     key: 'schedule_daily_pipeline_cron',
-    label: 'Daily pipeline (full)',
+    label: 'Morning pipeline',
-    hint: 'OHLCV → sentiment → R:R scan → outcomes → regime. Default 07:00 so data is ready by 8.',
+    hint: 'OHLCV → benchmark → sentiment → regime (no R:R scan). Default 02:00 ET.',
+    mono: true,
+  },
+  {
+    key: 'schedule_near_close_pipeline_cron',
+    label: 'Near-close pipeline (scan + alert)',
+    hint: 'OHLCV fetch → R:R scan → Telegram. Default 15:30 ET Mon–Fri so manual MOC fills can still hit ~15:50/15:55.',
+    mono: true,
+  },
+  {
+    key: 'schedule_after_close_pipeline_cron',
+    label: 'After-close pipeline (outcome)',
+    hint: 'OHLCV fetch (final bar) → outcome eval. Default 16:45 ET Mon–Fri — not chained to the partial near-close bar.',
     mono: true,
   },
   {
     key: 'schedule_intraday_pipeline_cron',
     label: 'Intraday pipeline (light)',
-    hint: 'Refresh prices + resolve outcomes. Default hourly across the US session, weekdays.',
+    hint: 'Refresh prices + resolve outcomes mid-session. Default hourly 10:00–15:00 ET weekdays.',
     mono: true,
   },
   {
     key: 'schedule_fundamentals_cron',
     label: 'Fundamentals (weekly)',
-    hint: 'Slow, rate-limited. Default early Monday so it finishes well before the day starts.',
+    hint: 'Slow, rate-limited. Default early Monday ET.',
     mono: true,
   },
 ];
@@ -43,7 +57,7 @@ export function ScheduleSettings() {
   const [form, setForm] = useState<ScheduleConfig>(DEFAULTS);
 
   useEffect(() => {
-    if (data) setForm(data);
+    if (data) setForm({ ...DEFAULTS, ...data });
   }, [data]);
 
   if (isLoading) return <SkeletonTable rows={2} cols={2} />;
@@ -55,8 +69,8 @@ export function ScheduleSettings() {
         <h3 className="text-sm font-semibold text-gray-200">Pipeline Schedule</h3>
         <p className="mt-1 text-xs text-gray-500">
           When the jobs run, as 5-field cron (<span className="num">min hour day month weekday</span>).
-          Saved changes apply to the running scheduler immediately — no redeploy. The big nightly run
+          Saved changes apply to the running scheduler immediately — no redeploy.
-          does the full refresh; the light intraday run just keeps prices current.
+          One qualifying R:R scan per day is the near-close job; alerts fire immediately after that scan.
         </p>
       </div>
 
@@ -66,7 +80,7 @@ export function ScheduleSettings() {
             <span className="text-xs text-gray-400">{f.label}</span>
             <input
               type="text"
-              value={form[f.key]}
+              value={form[f.key] ?? ''}
               spellCheck={false}
               onChange={(e) => setForm((prev) => ({ ...prev, [f.key]: e.target.value }))}
               className={`w-full input-glass px-3 py-2 text-sm ${f.mono ? 'num' : ''}`}

frontend/src/lib/types.ts

@@ -187,10 +187,12 @@ export interface ActivationConfig {
   exclude_neutral: boolean;
 }
 
-// Cron schedule for the daily/intraday pipelines + fundamentals
+// Cron schedule for morning / near-close / after-close / intraday + fundamentals
 export interface ScheduleConfig {
   schedule_timezone: string;
   schedule_daily_pipeline_cron: string;
+  schedule_near_close_pipeline_cron: string;
+  schedule_after_close_pipeline_cron: string;
   schedule_intraday_pipeline_cron: string;
   schedule_fundamentals_cron: string;
 }

tests/unit/test_scheduler.py

@@ -115,6 +115,8 @@ class TestConfigureScheduler:
             "event_study",
             "backtest",
             "daily_pipeline",
+            "near_close_pipeline",
+            "after_close_pipeline",
             "intraday_pipeline",
         }
 
@@ -125,6 +127,7 @@ class TestConfigureScheduler:
         job_ids = [j.id for j in scheduler.get_jobs()]
         # Each ID should appear exactly once
         assert sorted(job_ids) == sorted([
+            "after_close_pipeline",
             "alerts",
             "backtest",
             "benchmark_collector",
@@ -134,6 +137,7 @@ class TestConfigureScheduler:
             "data_backfill",
             "fundamental_collector",
             "market_regime",
+            "near_close_pipeline",
             "regime_monitor",
             "event_study",
             "outcome_evaluator",

tests/unit/test_trade_policy.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from datetime import datetime, timedelta, timezone
 
 import pytest
+from sqlalchemy import select
 
 from app.models.paper_trade import PaperTrade
 from app.models.ticker import Ticker
@@ -129,6 +130,54 @@ async def test_latest_stop_starts_a_new_gate_reset_episode(session):
     assert ticker.id in await get_reentry_gate_locks(session)
 
 
+async def test_same_day_fail_then_qualify_stays_locked(session):
+    """Fail at 10:00 NY and qualify at 15:35 NY same day must not unlock."""
+    session.add(User(id=1, username="u", password_hash="x", role="user", has_access=True))
+    ticker = Ticker(symbol="SAMEDAY")
+    session.add(ticker)
+    await session.flush()
+
+    stopped_at = datetime(2026, 7, 15, 14, 0, tzinfo=timezone.utc)  # 10:00 ET
+    session.add(_stopped_trade(ticker.id, closed_at=stopped_at))
+    await session.commit()
+
+    fail_at = datetime(2026, 7, 15, 14, 5, tzinfo=timezone.utc)  # ~10:05 ET
+    updated = await observe_reentry_gate_transitions(
+        session,
+        evaluated_ticker_ids={ticker.id},
+        qualified_ticker_ids=set(),
+        observed_at=fail_at,
+    )
+    assert updated == {ticker.id}
+
+    qualify_same_day = datetime(2026, 7, 15, 19, 35, tzinfo=timezone.utc)  # 15:35 ET
+    updated = await observe_reentry_gate_transitions(
+        session,
+        evaluated_ticker_ids={ticker.id},
+        qualified_ticker_ids={ticker.id},
+        observed_at=qualify_same_day,
+    )
+    assert updated == set()
+    assert ticker.id in await get_reentry_gate_locks(session)
+    trade = (
+        await session.execute(select(PaperTrade).where(PaperTrade.ticker_id == ticker.id))
+    ).scalar_one()
+    assert trade.reentry_gate_failed_at is not None
+    assert trade.reentry_gate_requalified_at is None
+
+    qualify_next_day = datetime(2026, 7, 16, 19, 35, tzinfo=timezone.utc)  # next day 15:35 ET
+    updated = await observe_reentry_gate_transitions(
+        session,
+        evaluated_ticker_ids={ticker.id},
+        qualified_ticker_ids={ticker.id},
+        observed_at=qualify_next_day,
+    )
+    assert updated == {ticker.id}
+    await session.refresh(trade)
+    assert trade.reentry_gate_requalified_at is not None
+    assert ticker.id not in await get_reentry_gate_locks(session)
+
+
 async def test_newer_non_stop_exit_supersedes_historical_stop(session):
     session.add(User(id=1, username="u", password_hash="x", role="user", has_access=True))
     ticker = Ticker(symbol="LATEREXIT")