diff --git a/app/providers/alpaca.py b/app/providers/alpaca.py index 86fed45..f79814e 100644 --- a/app/providers/alpaca.py +++ b/app/providers/alpaca.py @@ -4,7 +4,7 @@ from __future__ import annotations import asyncio import logging -from datetime import date +from datetime import date, datetime, time, timedelta, timezone from alpaca.data.historical import StockHistoricalDataClient from alpaca.data.requests import StockBarsRequest @@ -16,6 +16,11 @@ from app.providers.protocol import OHLCVData logger = logging.getLogger(__name__) +# Free plans may not query data from the most recent ~15 minutes, and a window +# reaching into it fails the *entire* request — which would silently leave the +# near-close scan on yesterday's close. Margin over the documented boundary. +_RECENT_DATA_CUTOFF = timedelta(minutes=20) + class AlpacaOHLCVProvider: """Fetches daily OHLCV bars from Alpaca Markets Data API.""" @@ -25,6 +30,26 @@ class AlpacaOHLCVProvider: raise ProviderError("Alpaca API key and secret are required") self._client = StockHistoricalDataClient(api_key, api_secret) + @staticmethod + def _resolve_window(start_date: date, end_date: date) -> tuple[datetime, datetime]: + """Return the instants covering ``start_date``..``end_date`` inclusive. + + Two boundaries have to be right or today's bar disappears: + + * Daily bars are stamped at the session start in UTC (04:00Z under EDT), + so an ``end`` of midnight on ``end_date`` lands *before* that day's bar + and silently drops it — extend to the following midnight instead. + * The window must stay out of the delayed-data period, otherwise the + request is rejected outright with "subscription does not permit + querying recent SIP data". Clamping keeps today's in-progress bar + available, roughly 20 minutes behind live. + """ + start = datetime.combine(start_date, time.min, tzinfo=timezone.utc) + end = datetime.combine( + end_date + timedelta(days=1), time.min, tzinfo=timezone.utc + ) + return start, min(end, datetime.now(timezone.utc) - _RECENT_DATA_CUTOFF) + @staticmethod def _to_alpaca_symbol(symbol: str) -> str: """Convert internal symbol format (BRK-B) to Alpaca format (BRK.B).""" @@ -40,12 +65,15 @@ class AlpacaOHLCVProvider: ) -> list[OHLCVData]: """Fetch daily OHLCV bars for *ticker* between *start_date* and *end_date*.""" alpaca_symbol = self._to_alpaca_symbol(ticker) + start, end = self._resolve_window(start_date, end_date) + if end <= start: + return [] try: request = StockBarsRequest( symbol_or_symbols=alpaca_symbol, timeframe=TimeFrame.Day, - start=start_date, - end=end_date, + start=start, + end=end, adjustment=Adjustment.SPLIT, ) diff --git a/app/scheduler.py b/app/scheduler.py index 94aeb27..680a326 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -487,7 +487,12 @@ def _chunked(symbols: list[str], chunk_size: int) -> list[list[str]]: # --------------------------------------------------------------------------- -async def collect_ohlcv(full_backfill: bool = False, job_name: str = "data_collector") -> None: +async def collect_ohlcv( + full_backfill: bool = False, + job_name: str = "data_collector", + *, + refetch_days: int = 0, +) -> None: """Fetch latest daily OHLCV for all tracked tickers. Uses AlpacaOHLCVProvider. Processes each ticker independently. @@ -500,6 +505,10 @@ async def collect_ohlcv(full_backfill: bool = False, job_name: str = "data_colle ``settings.ohlcv_history_days`` window (ignoring incremental resume) — used by the manual data_backfill job to deepen shallow histories. ``job_name`` lets the backfill report its own runtime/resume state separate from data_collector. + + ``refetch_days`` re-pulls the last N days regardless of ingestion progress — + the after-close run uses it to overwrite the day's partial intraday bar, which + resume logic would otherwise skip as "already up to date". """ _log_event(logging.INFO, "job_start", job=job_name) _runtime_start(job_name) @@ -536,11 +545,14 @@ async def collect_ohlcv(full_backfill: bool = False, job_name: str = "data_colle return end_date = date.today() - # Full backfill: pass an explicit start_date so fetch_and_ingest re-pulls - # the whole window instead of resuming from the last stored bar. - backfill_start = ( - end_date - timedelta(days=settings.ohlcv_history_days) if full_backfill else None - ) + # An explicit start_date makes fetch_and_ingest re-pull that window instead + # of resuming from the last stored bar (upsert overwrites, so this is safe). + if full_backfill: + backfill_start = end_date - timedelta(days=settings.ohlcv_history_days) + elif refetch_days: + backfill_start = end_date - timedelta(days=refetch_days) + else: + backfill_start = None for symbol in symbols: _runtime_progress(job_name, processed=processed, total=total, current_ticker=symbol) @@ -598,6 +610,18 @@ async def backfill_ohlcv() -> None: await collect_ohlcv(full_backfill=True, job_name="data_backfill") +async def collect_ohlcv_final() -> None: + """After-close OHLCV refresh that replaces the day's partial bar. + + Intraday runs store today's bar while the session is still open, so ingestion + progress already reads "today" and incremental resume would skip the day + entirely — leaving a partial bar as the permanent record. ``refetch_days`` + forces the last few sessions to be re-pulled so outcome evaluation and + fill-quality checks grade against the real close. + """ + await collect_ohlcv(refetch_days=_FINAL_REFETCH_DAYS) + + # --------------------------------------------------------------------------- # Job: Sentiment Collector # --------------------------------------------------------------------------- @@ -1183,6 +1207,10 @@ async def sync_ticker_universe() -> None: # — 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). +# Sessions re-pulled by the after-close fetch so the consolidated bar overwrites +# the intraday partial one (covers a long weekend / holiday gap). +_FINAL_REFETCH_DAYS = 5 + _DAILY_PIPELINE_STEPS = [ ("data_collector", "collect_ohlcv"), ("benchmark_collector", "collect_benchmark"), @@ -1206,6 +1234,8 @@ _DAILY_PIPELINE_STEPS = [ # entries behave like stale_close (still acceptable per execution-recovery matrix). # No exchange calendar dependency. _NEAR_CLOSE_PIPELINE_STEPS = [ + # Must land today's in-progress bar (~20 min behind live), or the scan falls + # back to the previous close and execution degrades to the stale_close floor. ("data_collector", "collect_ohlcv"), ("rr_scanner", "scan_rr"), ("alerts", "dispatch_alerts_job"), @@ -1214,7 +1244,7 @@ _NEAR_CLOSE_PIPELINE_STEPS = [ # 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"), + ("data_collector", "collect_ohlcv_final"), ("outcome_evaluator", "evaluate_outcomes"), ] @@ -1338,18 +1368,22 @@ def _parse_frequency(freq: str) -> dict[str, int]: # 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. +# DAY-OF-WEEK MUST BE NAMES, NEVER NUMBERS. APScheduler's from_crontab() passes +# field 5 straight to its own day_of_week, where 0=Monday — so "1-5" resolves to +# Tue–Sat, silently skipping every Monday and scanning on Saturdays. Names are +# unambiguous in both dialects. SCHEDULE_DEFAULTS: dict[str, str] = { "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", + "schedule_near_close_pipeline_cron": "30 15 * * mon-fri", # Fetch final bars → outcome eval (must not run on the partial near-close bar). - "schedule_after_close_pipeline_cron": "45 16 * * 1-5", + "schedule_after_close_pipeline_cron": "45 16 * * mon-fri", # Hourly mid-session price + outcome (10:00–15:00 ET Mon–Fri). - "schedule_intraday_pipeline_cron": "0 10-15 * * 1-5", + "schedule_intraday_pipeline_cron": "0 10-15 * * mon-fri", # Weekly fundamentals early Monday NY. - "schedule_fundamentals_cron": "0 1 * * 1", + "schedule_fundamentals_cron": "0 1 * * mon", } # job id -> schedule setting key diff --git a/tests/unit/test_alpaca_provider_window.py b/tests/unit/test_alpaca_provider_window.py new file mode 100644 index 0000000..5d82ab5 --- /dev/null +++ b/tests/unit/test_alpaca_provider_window.py @@ -0,0 +1,88 @@ +"""Alpaca fetch window / feed selection. + +Regression cover for the 2026-07-20 outage: the near-close scan silently ran on +the previous session's close because ``end`` resolved to midnight on end_date, +which is *before* that day's bar timestamp (04:00Z under EDT). Widening the +window also has to stay clear of the delayed-data period, which rejects the whole +request. +""" + +from __future__ import annotations + +from datetime import date, datetime, timedelta, timezone + +import pytest + +from app.providers.alpaca import AlpacaOHLCVProvider + + +class _CapturingClient: + """Stands in for StockHistoricalDataClient, recording the request.""" + + def __init__(self) -> None: + self.request = None + + def get_stock_bars(self, request): + self.request = request + return {"AAPL": []} + + +def _provider() -> tuple[AlpacaOHLCVProvider, _CapturingClient]: + provider = AlpacaOHLCVProvider("key", "secret") + client = _CapturingClient() + provider._client = client + return provider, client + + +def _midnight(day: date) -> datetime: + """Naive-UTC midnight — the SDK strips tzinfo from request datetimes.""" + return datetime.combine(day, datetime.min.time()) + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc).replace(tzinfo=None) + + +@pytest.mark.asyncio +async def test_todays_in_progress_bar_is_inside_the_window(): + """The whole near-close design depends on today's bar being fetchable.""" + provider, client = _provider() + today = date.today() + + await provider.fetch_ohlcv("AAPL", today - timedelta(days=5), today) + + # Daily bars are stamped at session start (04:00Z); a midnight end drops them. + assert client.request.end > _midnight(today) + + +@pytest.mark.asyncio +async def test_window_stays_out_of_the_delayed_data_period(): + """A window reaching the last ~15 minutes fails the entire request.""" + provider, client = _provider() + + await provider.fetch_ohlcv("AAPL", date.today() - timedelta(days=5), date.today()) + + assert client.request.end <= _utcnow() - timedelta(minutes=15) + + +@pytest.mark.asyncio +async def test_completed_past_day_is_fully_covered(): + """Clamping must not swallow the last day of a historical window.""" + provider, client = _provider() + end_date = date.today() - timedelta(days=3) + + await provider.fetch_ohlcv("AAPL", end_date - timedelta(days=5), end_date) + + assert client.request.end > _midnight(end_date) + + +@pytest.mark.asyncio +async def test_window_collapsing_to_nothing_skips_the_call(): + """A start inside the delayed period yields no request at all, not an error.""" + provider, client = _provider() + tomorrow = date.today() + timedelta(days=1) + + records = await provider.fetch_ohlcv("AAPL", tomorrow, tomorrow) + + assert records == [] + assert client.request is None diff --git a/tests/unit/test_schedule_config.py b/tests/unit/test_schedule_config.py index 04b9a4f..cb2dde3 100644 --- a/tests/unit/test_schedule_config.py +++ b/tests/unit/test_schedule_config.py @@ -32,6 +32,55 @@ class TestValidateCron: validate_cron("0 7 * * *", "Mars/Phobos") +class TestTradingDayCrons: + """APScheduler's from_crontab() uses 0=Monday, so numeric "1-5" means + Tue–Sat: it skips every Monday and fires on Saturdays. Weekday schedules + must therefore be spelled with day *names*. + """ + + _WEEKDAY_KEYS = ( + "schedule_near_close_pipeline_cron", + "schedule_after_close_pipeline_cron", + "schedule_intraday_pipeline_cron", + ) + + @pytest.mark.parametrize("key", _WEEKDAY_KEYS) + def test_fires_monday_and_never_saturday(self, key: str): + from datetime import datetime, timedelta + + from apscheduler.triggers.cron import CronTrigger + + trigger = CronTrigger.from_crontab( + SCHEDULE_DEFAULTS[key], timezone=SCHEDULE_DEFAULTS["schedule_timezone"] + ) + # Walk a full week of fire times from a known Sunday. + cursor = datetime(2026, 7, 19, tzinfo=trigger.timezone) + weekdays = set() + previous = None + for _ in range(12): + fire = trigger.get_next_fire_time(previous, cursor) + weekdays.add(fire.strftime("%a")) + previous = fire + cursor = fire + timedelta(seconds=1) + + assert "Mon" in weekdays, f"{key} skips Mondays — numeric day-of-week?" + assert {"Sat", "Sun"}.isdisjoint(weekdays), f"{key} fires on a weekend" + + def test_fundamentals_runs_on_monday(self): + from datetime import datetime + + from apscheduler.triggers.cron import CronTrigger + + trigger = CronTrigger.from_crontab( + SCHEDULE_DEFAULTS["schedule_fundamentals_cron"], + timezone=SCHEDULE_DEFAULTS["schedule_timezone"], + ) + fire = trigger.get_next_fire_time( + None, datetime(2026, 7, 19, tzinfo=trigger.timezone) + ) + assert fire.strftime("%a") == "Mon" + + class TestScheduleConfig: async def test_defaults_when_unset(self, session: AsyncSession): config = await get_schedule_config(session)