From 736451e26fab9ccdc050ec312f3b41a4617141dd Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Sat, 18 Jul 2026 17:55:39 +0200 Subject: [PATCH] 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 | 36 ++-- .../023_near_close_schedule_and_fill_mode.py | 73 ++++++++ app/models/paper_trade.py | 3 + app/scheduler.py | 158 +++++++++++++++--- app/schemas/admin.py | 4 +- app/schemas/paper_trade.py | 2 + app/services/admin_service.py | 9 +- app/services/paper_trade_service.py | 4 + app/services/trade_policy.py | 23 ++- docs/research/execution-recovery.md | 102 +++++++---- .../src/components/admin/ScheduleSettings.tsx | 40 +++-- frontend/src/lib/types.ts | 4 +- tests/unit/test_scheduler.py | 4 + tests/unit/test_trade_policy.py | 49 ++++++ 14 files changed, 428 insertions(+), 83 deletions(-) create mode 100644 alembic/versions/023_near_close_schedule_and_fill_mode.py diff --git a/README.md b/README.md index ae8c405..c681d06 100644 --- a/README.md +++ b/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. -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. -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. -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. +1. **OHLCV** — latest daily bars (Alpaca); new tickers backfill ~5 years. +2. **Sentiment** — stale names that matter (top-pick feeders, watchlist, open paper, discovery net). Display context only; the activation gate is price-only. +3. **Market Regime** + **Regime Monitor** — breadth/trend and the v2 risk thermometer; feed 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). diff --git a/alembic/versions/023_near_close_schedule_and_fill_mode.py b/alembic/versions/023_near_close_schedule_and_fill_mode.py new file mode 100644 index 0000000..6021212 --- /dev/null +++ b/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. diff --git a/app/models/paper_trade.py b/app/models/paper_trade.py index f41ac4a..96f612d 100644 --- a/app/models/paper_trade.py +++ b/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) diff --git a/app/scheduler.py b/app/scheduler.py index 246634c..9500bed 100644 --- a/app/scheduler.py +++ b/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"], + ) diff --git a/app/schemas/admin.py b/app/schemas/admin.py index b5a0966..1eb5f9e 100644 --- a/app/schemas/admin.py +++ b/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) diff --git a/app/schemas/paper_trade.py b/app/schemas/paper_trade.py index 31e0bfe..ae71568 100644 --- a/app/schemas/paper_trade.py +++ b/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 diff --git a/app/services/admin_service.py b/app/services/admin_service.py index ee31f9d..02e3a90 100644 --- a/app/services/admin_service.py +++ b/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", } diff --git a/app/services/paper_trade_service.py b/app/services/paper_trade_service.py index 1e4b368..2da72cd 100644 --- a/app/services/paper_trade_service.py +++ b/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, } diff --git a/app/services/trade_policy.py b/app/services/trade_policy.py index deb39a5..b4419a2 100644 --- a/app/services/trade_policy.py +++ b/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() diff --git a/docs/research/execution-recovery.md b/docs/research/execution-recovery.md index be959ee..23be6f3 100644 --- a/docs/research/execution-recovery.md +++ b/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**. -Adding a near-close scan *alongside* the 07:00 scan would let fail→qualify -transitions complete twice as fast and **silently change** the validated re-entry -policy. +- Intraday pipeline already fetches/upserts the **in-progress day-t bar** all + session (`fetch_ohlcv` end_date defaults to today). Near-close job = + **OHLCV fetch → R:R scan** (no new snapshot synthesizer). +- 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 / -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. +### Semantic guard (ship step 1 — precondition) -### 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 -off the US close for a week or two twice a year. +> `reentry_gate_requalified_at` may only be set when `reentry_gate_failed_at` +> 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**. -- 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. +### Schedule split -### 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. -- Nightly OHLCV job overwrites with the final bar via existing idempotent upsert. -- Forward paper record marks entries at the actual near-close fill so the live - track measures the new execution honestly. +- Near-close scan **1–5 only**; US-holiday no-ops are fine (stale identical data + can’t flip gates) — comment only, no exchange calendar. +- **Do not** run morning + near-close qualifying scans; move the scan, don’t add a second. + +### 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 | -| Partial-bar scan path | **Not started** | -| Paper fill-at-close marking | **Not started** (may already mark at close; verify when shipping) | +| Distinct-day gate-reset guard | **Shipped** — `trade_policy` + unit test | +| Schedule split + near-close scan→alert | **Shipped** — morning / near-close / after-close (fetch→outcome) | +| 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. diff --git a/frontend/src/components/admin/ScheduleSettings.tsx b/frontend/src/components/admin/ScheduleSettings.tsx index 1b8b503..eef445b 100644 --- a/frontend/src/components/admin/ScheduleSettings.tsx +++ b/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_daily_pipeline_cron: '0 7 * * *', - schedule_intraday_pipeline_cron: '0 14-22 * * 1-5', - schedule_fundamentals_cron: '0 4 * * 1', + 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', }; 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)', - hint: 'OHLCV → sentiment → R:R scan → outcomes → regime. Default 07:00 so data is ready by 8.', + label: 'Morning pipeline', + 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(DEFAULTS); useEffect(() => { - if (data) setForm(data); + if (data) setForm({ ...DEFAULTS, ...data }); }, [data]); if (isLoading) return ; @@ -55,8 +69,8 @@ export function ScheduleSettings() {

Pipeline Schedule

When the jobs run, as 5-field cron (min hour day month weekday). - Saved changes apply to the running scheduler immediately — no redeploy. The big nightly run - does the full refresh; the light intraday run just keeps prices current. + Saved changes apply to the running scheduler immediately — no redeploy. + One qualifying R:R scan per day is the near-close job; alerts fire immediately after that scan.

@@ -66,7 +80,7 @@ export function ScheduleSettings() { {f.label} setForm((prev) => ({ ...prev, [f.key]: e.target.value }))} className={`w-full input-glass px-3 py-2 text-sm ${f.mono ? 'num' : ''}`} diff --git a/frontend/src/lib/types.ts b/frontend/src/lib/types.ts index 515bc78..e562220 100644 --- a/frontend/src/lib/types.ts +++ b/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; } diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py index 494feb6..01253a3 100644 --- a/tests/unit/test_scheduler.py +++ b/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", diff --git a/tests/unit/test_trade_policy.py b/tests/unit/test_trade_policy.py index 8691672..5353646 100644 --- a/tests/unit/test_trade_policy.py +++ b/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")