From cf1019b21e3b48caa9de72f41a5effaccbc408a9 Mon Sep 17 00:00:00 2001 From: FlyM1ss Date: Wed, 23 Sep 2026 16:06:53 -0400 Subject: [PATCH 1/5] test(bar-cache): drop a stale claim that the dataset key is order-sensitive #520 keys market_data_store on sorted symbols; the docstring still said tuple(symbols). Co-Authored-By: Claude Opus 5.5 (1M context) --- .../tests/infrastructure/market_data/test_bar_cache.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/dashboard/backend/tests/infrastructure/market_data/test_bar_cache.py b/dashboard/backend/tests/infrastructure/market_data/test_bar_cache.py index 9710c76d..43185d80 100644 --- a/dashboard/backend/tests/infrastructure/market_data/test_bar_cache.py +++ b/dashboard/backend/tests/infrastructure/market_data/test_bar_cache.py @@ -215,8 +215,8 @@ def test_every_key_dimension_is_load_bearing(cache_dir, override): def test_symbols_are_keyed_individually_so_order_cannot_matter(cache_dir): - """Sidesteps market_data_store._dataset_key's order-sensitive - tuple(symbols): the same set in a different order is a hit here.""" + """One entry per symbol, so the order a caller lists them in never + reaches a key: the same set in a different order is a hit here.""" bar_cache.write_many( {"AAPL": _frame(), "MSFT": _frame(rows=4)}, last_fetch=LAST_FETCH, **KEY ) From 190707a27a0735ab8a3d03c566a635a29b2b3f69 Mon Sep 17 00:00:00 2001 From: FlyM1ss Date: Wed, 23 Sep 2026 16:08:34 -0400 Subject: [PATCH 2/5] fix(sessions): read Alpaca bars as open-stamped at the session close Alpaca stamps a bar at its open and serves extended hours, but every raw-bar session filter applied the close-stamp rule (inclusive 16:00). So: - minute mode kept the 16:00-16:05 after-hours source bar: the day's last equity mark was an after-hours price, and the 16:00 decision filled only if the tape happened to serve that bar (SIP does, carrying the closing auction; IEX often not); - the leaderboard's raw 1h path kept 10:00-16:00, i.e. the 16:00-17:00 after-hours bar in and the 09:00 bar holding the 09:30 open out. is_in_session takes open_stamped_minutes (an open-stamped bar is in session when it closes in one), profiles.bars_open_stamped says which sources need it (Alpaca; not iFinD or vnpy), and bar_aggregation.plan_execution_fills is the one fill planner for the engine and the protocol store: the final bucket fills at the last in-session source bar's close. Seven decisions a day either way; the hourly board becomes 09:00-15:00. Also imports DEFAULT_MARKET into market_data_store from sessions instead of restating it. Cached leaderboard rows keep the old rule until a force refresh. Co-Authored-By: Claude Opus 5.5 (1M context) --- dashboard/backend/baseline_generator.py | 29 +++- .../domain/backtesting/bar_aggregation.py | 54 +++++- .../backend/domain/backtesting/engine.py | 121 +++++++------ .../backtesting/external_run_service.py | 16 +- .../domain/backtesting/market_data_store.py | 102 +++++------ .../backend/domain/leaderboard/baselines.py | 5 +- .../domain/leaderboard/strategies/_common.py | 19 ++- .../domain/leaderboard/strategies/buy_hold.py | 5 +- .../strategies/equal_weight_buyhold.py | 5 +- .../strategies/equal_weight_index.py | 4 +- .../infrastructure/market_data/profiles.py | 12 ++ .../infrastructure/market_data/sessions.py | 62 +++++-- .../tests/test_external_minute_source.py | 106 +++++++++++- .../backend/tests/test_market_sessions.py | 160 ++++++++++++++++-- 14 files changed, 545 insertions(+), 155 deletions(-) diff --git a/dashboard/backend/baseline_generator.py b/dashboard/backend/baseline_generator.py index 3d391810..f80704b7 100644 --- a/dashboard/backend/baseline_generator.py +++ b/dashboard/backend/baseline_generator.py @@ -38,17 +38,27 @@ ) from exc -def _market_hours_only(timestamps, market_timezone: str): +def _market_hours_only( + timestamps, market_timezone: str, open_stamped_minutes: Optional[int] = None +): """Keep regular sessions in the timezone belonging to the market profile. This API takes only the zone, so the market is recovered from it through the profile registry rather than a zone-string comparison kept here. + ``open_stamped_minutes`` is set when the bars are raw Alpaca bars, stamped + at their open (see ``sessions.is_in_session``); aggregated decision bars + and iFinD bars are stamped at their close and leave it ``None``. """ market = market_for_timezone(market_timezone) return [ timestamp for timestamp in timestamps - if is_in_session(timestamp, market=market, timezone=market_timezone) + if is_in_session( + timestamp, + market=market, + timezone=market_timezone, + open_stamped_minutes=open_stamped_minutes, + ) ] @@ -245,6 +255,7 @@ def generate_buyhold_baseline( lot_size: int = 1, allocation_summary: Optional[Dict[str, Any]] = None, market_rule_calendar: MarketRuleCalendar | None = None, + open_stamped_minutes: Optional[int] = None, ) -> List[Dict]: """ Generate Buy & Hold baseline curve. @@ -291,7 +302,9 @@ def generate_buyhold_baseline( if not all_timestamps: return [] - all_timestamps = _market_hours_only(all_timestamps, market_timezone) + all_timestamps = _market_hours_only( + all_timestamps, market_timezone, open_stamped_minutes + ) all_timestamps = _timestamps_in_window( all_timestamps, start_date, end_date, market_timezone ) @@ -550,6 +563,7 @@ def generate_index_baseline( symbols_to_track: Optional[List[str]] = None, market_timezone: str = "US/Eastern", currency_context: CurrencyContext | None = None, + open_stamped_minutes: Optional[int] = None, ) -> List[Dict]: """ Generate Index baseline curve (equal-weight index). @@ -584,7 +598,9 @@ def generate_index_baseline( if not all_timestamps: return [] - all_timestamps = _market_hours_only(all_timestamps, market_timezone) + all_timestamps = _market_hours_only( + all_timestamps, market_timezone, open_stamped_minutes + ) all_timestamps = _timestamps_in_window( all_timestamps, start_date, end_date, market_timezone ) @@ -684,6 +700,7 @@ def generate_baselines( lot_size: int = 1, allocation_summary: Optional[Dict[str, Any]] = None, market_rule_calendar: MarketRuleCalendar | None = None, + open_stamped_minutes: Optional[int] = None, ) -> Tuple[List[Dict], List[Dict]]: """ Generate both baselines (Buy & Hold, Index). @@ -698,6 +715,8 @@ def generate_baselines( ``BaselineGenerator.generate_buyhold_baseline``. allocation_summary: Out-dict describing how much of the buy & hold sleeve actually filled. + open_stamped_minutes: Bar span when the bars are stamped at their + open (raw Alpaca bars); ``None`` for close-stamped bars. Returns: Tuple of (buyhold_curve, index_curve) @@ -717,6 +736,7 @@ def generate_baselines( lot_size=lot_size, allocation_summary=allocation_summary, market_rule_calendar=market_rule_calendar, + open_stamped_minutes=open_stamped_minutes, ) index_curve = generator.generate_index_baseline( @@ -727,6 +747,7 @@ def generate_baselines( symbols_list, market_timezone, currency_context, + open_stamped_minutes=open_stamped_minutes, ) return buyhold_curve, index_curve diff --git a/dashboard/backend/domain/backtesting/bar_aggregation.py b/dashboard/backend/domain/backtesting/bar_aggregation.py index 73c4d4e7..c8ce89ad 100644 --- a/dashboard/backend/domain/backtesting/bar_aggregation.py +++ b/dashboard/backend/domain/backtesting/bar_aggregation.py @@ -13,8 +13,9 @@ from __future__ import annotations -from datetime import time -from typing import Any, Dict, Iterable, Mapping +from bisect import bisect_left +from datetime import date, time +from typing import Any, Dict, Iterable, List, Mapping, Tuple import numpy as np import pandas as pd @@ -24,7 +25,10 @@ timeframe_minutes, ) # Re-exported: the session bounds have one owner, and it is not this module. -from dashboard.backend.infrastructure.market_data.sessions import session_windows +from dashboard.backend.infrastructure.market_data.sessions import ( + is_session_close, + session_windows, +) class BarAggregationError(ValueError): @@ -326,3 +330,47 @@ def summarize_aggregation_quality( summary["usable_decision_bars"] += usable summary["dropped_decision_bars"] += total - usable return summary + + +def _market_day(timestamp: Any, timezone: str) -> date: + stamp = pd.Timestamp(timestamp) + if stamp.tzinfo is None: + return stamp.tz_localize(timezone).date() + return stamp.tz_convert(timezone).date() + + +def plan_execution_fills( + decision_timestamps: Iterable[Any], + source_timestamps: List[Any], + *, + market: str, + timezone: str, +) -> Dict[Any, Tuple[Any, str]]: + """Map each decision bar to ``(source bar, price field)`` it fills on. + + A decision closes at its stamp, so it fills at the ``open`` of the source + bar opening at that instant -- the first fill without look-ahead. A + session's final bucket (16:00 ET) has no such bar once after-hours bars are + out of the source set, so it fills at the ``close`` of the last source bar + before it that day: the same instant, priced at the last regular-hours + trade. Filling it at the 16:00 bar instead made the day's closing decision + depend on whether the tape served an after-hours bar at all. + + One planner for the engine and the protocol path's dataset store, which + each used to carry their own copy of the exact-match rule. Decisions with + no fill are absent. ``source_timestamps`` must be sorted. + """ + by_day: Dict[date, List[Any]] = {} + for timestamp in source_timestamps: + by_day.setdefault(_market_day(timestamp, timezone), []).append(timestamp) + fills: Dict[Any, Tuple[Any, str]] = {} + for timestamp in decision_timestamps: + same_day = by_day.get(_market_day(timestamp, timezone), []) + index = bisect_left(same_day, timestamp) + if index < len(same_day) and same_day[index] == timestamp: + fills[timestamp] = (timestamp, "open") + elif index > 0 and is_session_close( + timestamp, market=market, timezone=timezone + ): + fills[timestamp] = (same_day[index - 1], "close") + return fills diff --git a/dashboard/backend/domain/backtesting/engine.py b/dashboard/backend/domain/backtesting/engine.py index dbb60eac..27751437 100644 --- a/dashboard/backend/domain/backtesting/engine.py +++ b/dashboard/backend/domain/backtesting/engine.py @@ -17,7 +17,6 @@ import inspect import json import uuid -from bisect import bisect_left from datetime import date, datetime from math import ceil # `from time import ...`, not `import time`. This module used to import @@ -70,6 +69,7 @@ from dashboard.backend.domain.backtesting.features import TechnicalIndicators from dashboard.backend.domain.backtesting.bar_aggregation import ( aggregate_bars_by_symbol, + plan_execution_fills, summarize_aggregation_quality, ) from dashboard.backend.domain.backtesting.metrics import ( @@ -126,6 +126,7 @@ LLM_DECISION_SOURCE, RULE_BASED_DECISION_SOURCE, MarketProfile, + bars_open_stamped, get_market_profile, resolve_decision_source, ) @@ -1623,33 +1624,73 @@ def _current_equity(self, manager: PortfolioManager, timestamp=None) -> float: return self._require_currency_context().to_reporting(manager.cash, timestamp) return float(manager.cash) - def _market_hours_only(self, timestamps): + def _market_hours_only(self, timestamps, *, open_stamped_minutes=None): """Filter timestamps using the selected market's local sessions. Through ``market_data.sessions``, the same owner the dataset store and the aggregation use, so the dashboard and protocol paths cannot count - different steps for one window. + different steps for one window. ``open_stamped_minutes`` is set for raw + provider bars stamped at their open; see :meth:`_raw_bar_open_minutes`. """ profile = self._effective_profile() return [ timestamp for timestamp in timestamps if is_in_session( - timestamp, market=profile.market, timezone=profile.timezone + timestamp, + market=profile.market, + timezone=profile.timezone, + open_stamped_minutes=open_stamped_minutes, ) ] - def _market_day_key(self, timestamp) -> str: - """Return a trading-day key in the market's local timezone.""" - import pytz + def _raw_bar_open_minutes(self, timeframe): + """The span of a raw provider bar when the provider stamps it at its + open, else ``None``. Aggregated decision bars are always stamped at + their close and must not go through this.""" + if not bars_open_stamped(getattr(self, "data_source", None)): + return None + return timeframe_minutes(timeframe) + + def _plan_executions(self, decision_timestamps): + """``(decisions, valuation bars, {decision: fill bar}, {decision: + price field})`` for the run loop. - market_tz = pytz.timezone(self._effective_profile().timezone) - local = ( - market_tz.localize(timestamp) - if timestamp.tzinfo is None - else timestamp.astimezone(market_tz) + Hourly mode fills and values on the decision bar itself. Minute mode + values on every in-session source bar and fills through + ``plan_execution_fills``; a decision it cannot fill is not a step. + """ + if not self.intraday_mode: + return ( + list(decision_timestamps), + list(decision_timestamps), + {timestamp: timestamp for timestamp in decision_timestamps}, + {}, + ) + raw_timestamps = self._market_hours_only( + self._timestamps_for_data(self.source_data), + open_stamped_minutes=self._raw_bar_open_minutes(self.source_timeframe), ) - return local.date().isoformat() + profile = self._effective_profile() + fills = plan_execution_fills( + decision_timestamps, + raw_timestamps, + market=profile.market, + timezone=profile.timezone, + ) + return ( + [timestamp for timestamp in decision_timestamps if timestamp in fills], + raw_timestamps, + {timestamp: source for timestamp, (source, _field) in fills.items()}, + {timestamp: field for timestamp, (_source, field) in fills.items()}, + ) + + def _decision_bar_open_minutes(self): + """:meth:`_raw_bar_open_minutes` for ``all_data``'s bars: ``None`` in + minute mode, where they are aggregated and stamped at their close.""" + if getattr(self, "intraday_mode", False): + return None + return self._raw_bar_open_minutes(self.source_timeframe) def _run_daily_post_trade( self, @@ -1777,7 +1818,10 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: all_timestamps = filtered - all_timestamps = self._market_hours_only(all_timestamps) + all_timestamps = self._market_hours_only( + all_timestamps, + open_stamped_minutes=self._decision_bar_open_minutes(), + ) prior_market_dates = ( _prior_market_date_by_decision_date(all_timestamps) if self.runtime_type == AI_HEDGE_FUND_RUNTIME_TYPE @@ -1795,40 +1839,12 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: f"failed step(s) before aborting\n" ) - raw_timestamps = all_timestamps - execution_plan = {timestamp: timestamp for timestamp in all_timestamps} - if self.intraday_mode: - raw_timestamps = self._market_hours_only( - self._timestamps_for_data(self.source_data) - ) - raw_by_market_day: Dict[str, List[Any]] = {} - for source_timestamp in raw_timestamps: - raw_by_market_day.setdefault( - self._market_day_key(source_timestamp), [] - ).append(source_timestamp) - execution_plan = {} - for timestamp in all_timestamps: - same_day_sources = raw_by_market_day.get( - self._market_day_key(timestamp), [] - ) - source_index = bisect_left(same_day_sources, timestamp) - next_source_timestamp = ( - same_day_sources[source_index] - if ( - source_index < len(same_day_sources) - and same_day_sources[source_index] == timestamp - ) - else None - ) - # The final partial session bucket (e.g. 15:30–16:00 ET) has - # no following source bar at 16:00 and cannot be executed. - if next_source_timestamp is not None: - execution_plan[timestamp] = next_source_timestamp - all_timestamps = [ - timestamp - for timestamp in all_timestamps - if timestamp in execution_plan - ] + ( + all_timestamps, + raw_timestamps, + execution_plan, + execution_fields, + ) = self._plan_executions(all_timestamps) print( f" Trading {len(all_timestamps)} hourly decision bars during " @@ -1970,7 +1986,10 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: # In minute mode, execute at the next source bar's open. The # decision bar closes at ``timestamp``; the source bar opening at # that same instant is the first non-look-ahead fill opportunity. + # A session's final bucket fills at the last source bar's close + # instead (``plan_execution_fills``). execution_timestamp = execution_plan[timestamp] + execution_field = execution_fields.get(timestamp, "open") execution_market_data = market_data execution_fallback_prices = { symbol: values[execution_timestamp] @@ -1985,9 +2004,9 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: execution_timestamp, ) execution_prices = { - symbol: row["open"] + symbol: row[execution_field] for symbol, row in execution_market_data.items() - if "open" in row + if execution_field in row } # Execute trades (only if real data available) @@ -2262,6 +2281,7 @@ def run_buyhold_baseline(self) -> Tuple[str, List[Dict]]: currency_context=self._require_currency_context(), transaction_cost_profile=profile.transaction_cost_profile, transaction_cost_totals=baseline_cost_totals, + open_stamped_minutes=self._decision_bar_open_minutes(), # The board lot is its own market rule. Deriving it from the cost # profile would floor buys to 100 in any future market that charges # fees but trades in single shares. @@ -2351,6 +2371,7 @@ def run_djia_baseline(self) -> Tuple[str, List[Dict]]: market_timezone=self._effective_profile().timezone, symbols_list=list(DJIA_30), currency_context=self._require_currency_context(), + open_stamped_minutes=self._decision_bar_open_minutes(), ) if not equity_history: diff --git a/dashboard/backend/domain/backtesting/external_run_service.py b/dashboard/backend/domain/backtesting/external_run_service.py index 12eda192..48ed40be 100644 --- a/dashboard/backend/domain/backtesting/external_run_service.py +++ b/dashboard/backend/domain/backtesting/external_run_service.py @@ -305,6 +305,7 @@ def __init__( self.source_timestamps: List[Any] = [] self.source_price_cache: Dict[str, Dict[Any, float]] = {} self.execution_timestamps: List[Any] = [] + self.execution_price_fields: List[str] = [] self.data_quality: Dict[str, Any] = {} self.frequency_contract: Optional[Dict[str, str]] = None self.market_data_provenance: Dict[str, Any] = {} @@ -374,6 +375,9 @@ def adopt_dataset(self, dataset: "market_data_store.MarketDataset") -> None: self.execution_timestamps = list( getattr(dataset, "execution_timestamps", self.timestamps) ) + self.execution_price_fields = list( + getattr(dataset, "execution_price_fields", None) or [] + ) self.data_quality = dict(getattr(dataset, "data_quality", {}) or {}) self.equity_metadata = dict(getattr(dataset, "equity_metadata", {}) or {}) self.source_timeframe = getattr( @@ -442,6 +446,13 @@ def _effective_execution_timestamps(self) -> List[Any]: return self.execution_timestamps return list(self.timestamps) + def _execution_price_field(self, step_index: int) -> str: + """The execution bar's "open", or its "close" for a session's final + bucket; see ``bar_aggregation.plan_execution_fills``.""" + if len(self.execution_price_fields) == self.total_steps: + return self.execution_price_fields[step_index] + return "open" + def _value_through(self, target_timestamp=None) -> None: """Mark the portfolio on each source bar through the given timestamp.""" source_timestamps = self._effective_source_timestamps() @@ -776,10 +787,11 @@ def _advance_step( timestamp = self.timestamps[self.step_index] execution_timestamp = self._effective_execution_timestamps()[self.step_index] execution_market_data = self._source_market_data_at(execution_timestamp) + execution_field = self._execution_price_field(self.step_index) execution_prices = { - symbol: row["open"] + symbol: row[execution_field] for symbol, row in execution_market_data.items() - if "open" in row + if execution_field in row } trades_before_execution = len(self.manager.trades) diff --git a/dashboard/backend/domain/backtesting/market_data_store.py b/dashboard/backend/domain/backtesting/market_data_store.py index 5a5aa58b..1ecc6503 100644 --- a/dashboard/backend/domain/backtesting/market_data_store.py +++ b/dashboard/backend/domain/backtesting/market_data_store.py @@ -28,17 +28,16 @@ import os import threading import time -from bisect import bisect_left from collections import OrderedDict from math import ceil from typing import Any, Callable, Dict, List, Optional, Tuple import pandas as pd -import pytz from dashboard.backend.domain.backtesting.features import TechnicalIndicators from dashboard.backend.domain.backtesting.bar_aggregation import ( aggregate_bars_by_symbol, + plan_execution_fills, summarize_aggregation_quality, ) from dashboard.backend.infrastructure.market_data.alpaca_bars import AlpacaDataLoader @@ -52,6 +51,7 @@ verify_source_timeframe, ) from dashboard.backend.infrastructure.market_data.sessions import ( + DEFAULT_MARKET, canonical_market, is_in_session, timezone_for_market, @@ -79,7 +79,8 @@ class MarketDataset: __slots__ = ( "key", "all_data", "timestamps", "price_cache", "total_steps", "source_data", "source_timestamps", "source_price_cache", - "execution_timestamps", "source_timeframe", "decision_timeframe", + "execution_timestamps", "execution_price_fields", + "source_timeframe", "decision_timeframe", "data_quality", "equity_metadata", ) @@ -90,6 +91,7 @@ def __init__(self, key: Tuple, all_data: Dict[str, pd.DataFrame], source_timestamps: Optional[List[Any]] = None, source_price_cache: Optional[Dict[str, Dict[Any, float]]] = None, execution_timestamps: Optional[List[Any]] = None, + execution_price_fields: Optional[List[str]] = None, source_timeframe: str = "60m", decision_timeframe: str = "60m", data_quality: Optional[Dict[str, Any]] = None, @@ -113,6 +115,13 @@ def __init__(self, key: Tuple, all_data: Dict[str, pd.DataFrame], if execution_timestamps is not None else list(timestamps) ) + # Which price of the execution bar fills each step: its "open", or the + # "close" for a session's final bucket (``plan_execution_fills``). + self.execution_price_fields = ( + execution_price_fields + if execution_price_fields is not None + else ["open"] * len(self.execution_timestamps) + ) self.source_timeframe = source_timeframe self.decision_timeframe = decision_timeframe self.data_quality = data_quality or {} @@ -133,12 +142,12 @@ def __init__(self): _cache: "OrderedDict[Tuple, _Entry]" = OrderedDict() -#: What this store assumed unconditionally before the market became a -#: parameter, and therefore what a caller that passes nothing still gets. Kept -#: as the default rather than made required so the in-process test doubles and -#: the legacy callers that predate the market dimension are unaffected: the -#: three shipped call sites all hold a `MarketProfile` and pass it. -DEFAULT_MARKET = "US" +# `DEFAULT_MARKET` (imported from `sessions`, which owns it) is what this store +# assumed unconditionally before the market became a parameter, and therefore +# what a caller that passes nothing still gets. Kept as the default rather than +# made required so the in-process test doubles and the legacy callers that +# predate the market dimension are unaffected: the three shipped call sites all +# hold a `MarketProfile` and pass it. DEFAULT_TIMEZONE = timezone_for_market(DEFAULT_MARKET) @@ -363,7 +372,8 @@ def _build_dataset( evidence="fetch", ) data_quality: Dict[str, Any] = {} - if timeframe_minutes(actual_source) < timeframe_minutes(requested_decision): + aggregated = timeframe_minutes(actual_source) < timeframe_minutes(requested_decision) + if aggregated: aggregated_data = aggregate_bars_by_symbol( source_data, source_timeframe=actual_source, @@ -394,8 +404,14 @@ def _build_dataset( all_data, timezone=timezone, ) + # This store's loaders are Alpaca's, which stamp a raw bar at its open; + # an aggregated decision bar is stamped at its close. + source_open_minutes = timeframe_minutes(actual_source) timestamps = _build_trading_timestamps( - all_data, market=market, timezone=timezone + all_data, + market=market, + timezone=timezone, + open_stamped_minutes=None if aggregated else source_open_minutes, ) if not timestamps: raise RuntimeError("No trading hours in the selected date range") @@ -405,25 +421,17 @@ def _build_dataset( min_symbol_coverage=0.0, market=market, timezone=timezone, + open_stamped_minutes=source_open_minutes, ) source_price_cache = _build_price_cache(source_data, source_timestamps) - execution_timestamps = _build_execution_timestamps( - timestamps, - source_timestamps, - timezone=timezone, + fills = plan_execution_fills( + timestamps, source_timestamps, market=market, timezone=timezone ) - if any(execution_timestamp is None for execution_timestamp in execution_timestamps): - timestamps = [ - timestamp - for timestamp, execution_timestamp in zip( - timestamps, execution_timestamps - ) - if execution_timestamp is not None - ] - execution_timestamps = [ - timestamp for timestamp in execution_timestamps if timestamp is not None - ] + if len(fills) < len(timestamps): + timestamps = [timestamp for timestamp in timestamps if timestamp in fills] price_cache = _build_price_cache(all_data, timestamps) + execution_timestamps = [fills[timestamp][0] for timestamp in timestamps] + execution_price_fields = [fills[timestamp][1] for timestamp in timestamps] dataset = MarketDataset( key, all_data, @@ -433,6 +441,7 @@ def _build_dataset( source_timestamps=source_timestamps, source_price_cache=source_price_cache, execution_timestamps=execution_timestamps, + execution_price_fields=execution_price_fields, source_timeframe=actual_source, decision_timeframe=requested_decision, data_quality=data_quality, @@ -444,45 +453,13 @@ def _build_dataset( return dataset -def _market_day_key(timestamp, timezone: str) -> str: - if timestamp.tzinfo is None: - local = pytz.timezone(timezone).localize(timestamp) - else: - local = timestamp.astimezone(pytz.timezone(timezone)) - return local.date().isoformat() - - -def _build_execution_timestamps( - decision_timestamps: List[Any], - source_timestamps: List[Any], - *, - timezone: str, -) -> List[Any]: - """Map a decision close to the source bar opening at that exact boundary.""" - source_by_day: Dict[str, List[Any]] = {} - for timestamp in source_timestamps: - source_by_day.setdefault(_market_day_key(timestamp, timezone), []).append( - timestamp - ) - result = [] - for timestamp in decision_timestamps: - same_day = source_by_day.get(_market_day_key(timestamp, timezone), []) - index = bisect_left(same_day, timestamp) - exact_match = ( - same_day[index] - if index < len(same_day) and same_day[index] == timestamp - else None - ) - result.append(exact_match) - return result - - def _build_trading_timestamps( all_data: Dict[str, pd.DataFrame], *, min_symbol_coverage: float = 0.8, market: str = DEFAULT_MARKET, timezone: str = DEFAULT_TIMEZONE, + open_stamped_minutes: Optional[int] = None, ) -> List[Any]: """Return in-session timestamps meeting the requested symbol coverage. @@ -506,7 +483,12 @@ def _build_trading_timestamps( return [ ts for ts in ordered - if is_in_session(ts, market=market, timezone=timezone) + if is_in_session( + ts, + market=market, + timezone=timezone, + open_stamped_minutes=open_stamped_minutes, + ) ] diff --git a/dashboard/backend/domain/leaderboard/baselines.py b/dashboard/backend/domain/leaderboard/baselines.py index a7bcae63..09d44f0d 100644 --- a/dashboard/backend/domain/leaderboard/baselines.py +++ b/dashboard/backend/domain/leaderboard/baselines.py @@ -19,6 +19,9 @@ import pandas as pd from dashboard.backend.domain.leaderboard.strategies import get_strategy +from dashboard.backend.domain.leaderboard.strategies._common import ( + LEADERBOARD_BAR_TIMEFRAME, +) from dashboard.backend.domain.backtesting.constants import INITIAL_CAPITAL from dashboard.backend.domain.backtesting.metrics import ( calculate_max_drawdown, @@ -42,7 +45,7 @@ def fetch_hourly_bars(symbols: List[str], start_date: str, end_date: str) -> Dic that persist a curve should read it with ``feed_provenance`` and store it, because the log line below outlives nothing. """ - loader = AlpacaDataLoader() + loader = AlpacaDataLoader(source_timeframe=LEADERBOARD_BAR_TIMEFRAME) end_inclusive = ( datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1) ).strftime("%Y-%m-%d") diff --git a/dashboard/backend/domain/leaderboard/strategies/_common.py b/dashboard/backend/domain/leaderboard/strategies/_common.py index ce2cfa70..1a91eb6d 100644 --- a/dashboard/backend/domain/leaderboard/strategies/_common.py +++ b/dashboard/backend/domain/leaderboard/strategies/_common.py @@ -12,10 +12,19 @@ import pandas as pd import pytz +from dashboard.backend.infrastructure.market_data.frequency import timeframe_minutes from dashboard.backend.infrastructure.market_data.sessions import is_in_session _ET = pytz.timezone("US/Eastern") +# Every leaderboard strategy runs on the raw Alpaca bars +# ``leaderboard/baselines.fetch_hourly_bars`` requests at this timeframe. +# Alpaca stamps a bar at its open, so the session filter needs the span: under +# the close-stamp rule the board kept 10:00-16:00, i.e. the 16:00-17:00 +# after-hours bar in and the 09:00 bar holding the 09:30 open out. +LEADERBOARD_BAR_TIMEFRAME = "60m" +LEADERBOARD_BAR_OPEN_MINUTES = timeframe_minutes(LEADERBOARD_BAR_TIMEFRAME) + def parse_config_date(date_str: str) -> dt.date: return dt.datetime.strptime(date_str, "%Y-%m-%d").date() @@ -63,10 +72,16 @@ def timestamps_in_reference( def filter_market_hours(timestamps: List[Any]) -> List[Any]: - """Keep only regular US market-hours timestamps (9:30–16:00 ET).""" + """Keep the board's open-stamped hourly bars that close in the 9:30–16:00 + ET session: 09:00 through 15:00.""" return [ ts for ts in timestamps - if is_in_session(ts, market="US", timezone=_ET.zone) + if is_in_session( + ts, + market="US", + timezone=_ET.zone, + open_stamped_minutes=LEADERBOARD_BAR_OPEN_MINUTES, + ) ] diff --git a/dashboard/backend/domain/leaderboard/strategies/buy_hold.py b/dashboard/backend/domain/leaderboard/strategies/buy_hold.py index b406bad4..5ba168c9 100644 --- a/dashboard/backend/domain/leaderboard/strategies/buy_hold.py +++ b/dashboard/backend/domain/leaderboard/strategies/buy_hold.py @@ -12,7 +12,7 @@ from dashboard.backend.baseline_generator import BaselineGenerator from .base import BaselineStrategy -from ._common import subset_bars +from ._common import LEADERBOARD_BAR_OPEN_MINUTES, subset_bars class BuyHoldStrategy(BaselineStrategy): @@ -33,7 +33,8 @@ def run( if not bars_subset: return [] return BaselineGenerator().generate_buyhold_baseline( - bars_subset, start_date, end_date, initial_capital, symbols + bars_subset, start_date, end_date, initial_capital, symbols, + open_stamped_minutes=LEADERBOARD_BAR_OPEN_MINUTES, ) def num_trades(self) -> int: diff --git a/dashboard/backend/domain/leaderboard/strategies/equal_weight_buyhold.py b/dashboard/backend/domain/leaderboard/strategies/equal_weight_buyhold.py index 3476ce98..4b443bf8 100644 --- a/dashboard/backend/domain/leaderboard/strategies/equal_weight_buyhold.py +++ b/dashboard/backend/domain/leaderboard/strategies/equal_weight_buyhold.py @@ -14,7 +14,7 @@ from dashboard.backend.infrastructure.llm.validator import DJIA_30 from .base import BaselineStrategy -from ._common import subset_bars +from ._common import LEADERBOARD_BAR_OPEN_MINUTES, subset_bars class EqualWeightBuyHoldStrategy(BaselineStrategy): @@ -36,7 +36,8 @@ def run( if not bars_subset: return [] return BaselineGenerator().generate_buyhold_baseline( - bars_subset, start_date, end_date, initial_capital, symbols + bars_subset, start_date, end_date, initial_capital, symbols, + open_stamped_minutes=LEADERBOARD_BAR_OPEN_MINUTES, ) def num_trades(self) -> int: diff --git a/dashboard/backend/domain/leaderboard/strategies/equal_weight_index.py b/dashboard/backend/domain/leaderboard/strategies/equal_weight_index.py index f892449f..7018b185 100644 --- a/dashboard/backend/domain/leaderboard/strategies/equal_weight_index.py +++ b/dashboard/backend/domain/leaderboard/strategies/equal_weight_index.py @@ -15,6 +15,7 @@ from dashboard.backend.infrastructure.llm.validator import DJIA_30 from .base import BaselineStrategy +from ._common import LEADERBOARD_BAR_OPEN_MINUTES class EqualWeightIndexStrategy(BaselineStrategy): @@ -33,5 +34,6 @@ def run( ) -> List[Dict[str, Any]]: symbols = self.required_symbols() return BaselineGenerator().generate_index_baseline( - bars_by_symbol, start_date, end_date, initial_capital, symbols + bars_by_symbol, start_date, end_date, initial_capital, symbols, + open_stamped_minutes=LEADERBOARD_BAR_OPEN_MINUTES, ) diff --git a/dashboard/backend/infrastructure/market_data/profiles.py b/dashboard/backend/infrastructure/market_data/profiles.py index 72d2c667..2f85f2dc 100644 --- a/dashboard/backend/infrastructure/market_data/profiles.py +++ b/dashboard/backend/infrastructure/market_data/profiles.py @@ -281,6 +281,18 @@ def llm_enabled(self) -> bool: } +def bars_open_stamped(data_source: object) -> bool: + """Whether this source stamps a raw bar at its OPEN. + + Alpaca does (10:00 is 10:00-11:00) and serves extended hours with it; + iFinD stamps at the close (10:30 is 09:30-10:30), and the vnpy simulation + emits in-session closes. The session filter must know which, since one + inclusive rule keeps Alpaca's after-hours 16:00 bar -- see + ``sessions.is_in_session(open_stamped_minutes=...)``. + """ + return str(data_source or ALPACA).strip().lower() == ALPACA + + def registered_market_timezones() -> dict[str, str]: """``{market: timezone}`` across the registered profiles. diff --git a/dashboard/backend/infrastructure/market_data/sessions.py b/dashboard/backend/infrastructure/market_data/sessions.py index dd83b82a..5dcde07a 100644 --- a/dashboard/backend/infrastructure/market_data/sessions.py +++ b/dashboard/backend/infrastructure/market_data/sessions.py @@ -15,7 +15,7 @@ from __future__ import annotations -from datetime import datetime, time +from datetime import datetime, time, timedelta import pytz @@ -65,8 +65,11 @@ def market_for_timezone(timezone: str) -> str: def time_in_session(local_time: time, market: object) -> bool: """Whether a wall-clock time in the market's own zone is in session. - Minute resolution, inclusive at both ends: a decision bar is stamped at the - END of its bucket, so the last US bucket is stamped 16:00 and must survive. + Minute resolution, inclusive at both ends. This is the rule for a bar + stamped at the END of its bucket -- an aggregated decision bar, or an + iFinD bar -- so the last US bucket is stamped 16:00 and must survive. A bar + stamped at its OPEN (every Alpaca bar) is a different question; pass + ``open_stamped_minutes`` to :func:`is_in_session` for it. Seconds are dropped because every copy this replaced that the live US path ran on (`hour == 16 and minute == 0`) admitted the whole 16:00 minute; bars are minute-aligned, so nothing real sits inside that minute either way. @@ -75,17 +78,50 @@ def time_in_session(local_time: time, market: object) -> bool: return any(start <= minute <= end for start, end in session_windows(market)) -def is_in_session(timestamp: datetime, *, market: object, timezone: str) -> bool: - """:func:`time_in_session` for a timestamp in any zone. - - A naive timestamp is read as market-local time, matching +def _market_local(timestamp: datetime, timezone: str) -> datetime: + """A naive timestamp is read as market-local time, matching ``bar_aggregation._as_local_index``; ``astimezone`` on a naive pandas ``Timestamp`` raises instead, which made the filter and the aggregation - disagree on exactly the data a local-time feed returns. - """ + disagree on exactly the data a local-time feed returns.""" zone = pytz.timezone(timezone) if timestamp.tzinfo is None: - local = zone.localize(timestamp) - else: - local = timestamp.astimezone(zone) - return time_in_session(local.time(), market) + return zone.localize(timestamp) + return timestamp.astimezone(zone) + + +def is_in_session( + timestamp: datetime, + *, + market: object, + timezone: str, + open_stamped_minutes: int | None = None, +) -> bool: + """:func:`time_in_session` for a timestamp in any zone. + + ``open_stamped_minutes`` says the timestamp is a bar's OPEN and the bar + spans that many minutes; the bar is then in session when it CLOSES inside + one (``start < close <= end``). Alpaca stamps every bar at its open and + returns extended hours, so the close-stamp rule kept its 16:00 bar -- the + 16:00-16:05 (or 16:00-17:00) after-hours bar -- and, for clock-aligned + hourly bars, dropped the 09:00 bar that holds the 09:30 open. + """ + local = _market_local(timestamp, timezone) + if open_stamped_minutes is None: + return time_in_session(local.time(), market) + close = local + timedelta(minutes=open_stamped_minutes) + if close.date() != local.date(): + return False + close_minute = close.time().replace(second=0, microsecond=0) + return any( + start < close_minute <= end for start, end in session_windows(market) + ) + + +def is_session_close(timestamp: datetime, *, market: object, timezone: str) -> bool: + """Whether ``timestamp`` falls in the minute a session ends (16:00 ET, + 11:30 and 15:00 CST): the stamp of a final bucket, which no in-session + source bar opens at.""" + minute = _market_local(timestamp, timezone).time().replace( + second=0, microsecond=0 + ) + return any(minute == end for _start, end in session_windows(market)) diff --git a/dashboard/backend/tests/test_external_minute_source.py b/dashboard/backend/tests/test_external_minute_source.py index c4dcf4d7..f125f45c 100644 --- a/dashboard/backend/tests/test_external_minute_source.py +++ b/dashboard/backend/tests/test_external_minute_source.py @@ -70,7 +70,9 @@ def test_external_session_serves_hourly_bars_but_fills_and_values_on_5m(): assert session.source_timeframe == "5m" assert session.intraday_mode is True - assert session.total_steps == 6 + # Seven: the 16:00 bucket fills at the 15:55 bar's close rather than + # needing an after-hours bar to open at 16:00. + assert session.total_steps == 7 assert session.data_quality["total_decision_bars"] == 7 assert session.data_quality["usable_decision_bars"] == 7 assert session.data_quality["dropped_decision_bars"] == 0 @@ -110,6 +112,73 @@ def test_external_session_serves_hourly_bars_but_fills_and_values_on_5m(): assert len(session.manager.equity_history) == 13 +def test_the_closing_decision_fills_at_the_last_regular_hours_close(monkeypatch): + """Alpaca stamps bars at their open and serves extended hours, so the bar + stamped 16:00 ET is 16:00-16:05 after hours. It must neither fill the + closing decision nor mark the curve; whether it exists is the tape's call. + """ + after_hours_open = 999.0 + + class _ExtendedHoursLoader(_MinuteLoader): + def fetch_bars(self, symbols, start, end): + bars = _minute_bars(symbols, start, end) + extra = pd.Timestamp("2026-04-15 20:00:00+00:00") + for frame in bars.values(): + frame.loc[extra] = { + "open": after_hours_open, + "high": after_hours_open, + "low": after_hours_open, + "close": after_hours_open, + "volume": 10, + } + return bars + + monkeypatch.setattr(ebs, "AlpacaDataLoader", _ExtendedHoursLoader) + session = ebs.ExternalBacktestSession( + backtest_id="bt-close", + session_id="sess-close", + agent_name="agent-close", + model_name="test-model", + start_date="2026-04-15", + end_date="2026-04-15", + symbols=["AAPL"], + ) + session.load_market_data() + + after_hours = pd.Timestamp("2026-04-15 20:00:00+00:00") + last_regular = pd.Timestamp("2026-04-15 19:55:00+00:00") + assert after_hours not in session.source_timestamps + assert session.total_steps == 7 + assert session.execution_timestamps[-1] == last_regular + assert session.execution_price_fields == ["open"] * 6 + ["close"] + + for _ in range(session.total_steps - 1): + session.submit_decisions({"actions": []}) + session.submit_decisions( + { + "actions": [ + { + "symbol": "AAPL", + "action": "buy", + "confidence": 1.0, + "reasoning": "closing buy", + "position_size": 1, + } + ] + } + ) + + trade = session.manager.trades[-1] + assert trade["timestamp"] == last_regular + last_close = session.source_data["AAPL"].loc[last_regular, "close"] + assert trade["price"] == pytest.approx(last_close) + assert trade["price"] != pytest.approx(after_hours_open) + assert all( + pd.Timestamp(point["timestamp"]) != after_hours + for point in session.manager.equity_history + ) + + def test_external_session_does_not_report_fill_without_next_symbol_bar(monkeypatch): class _MissingExecutionBarLoader(_MinuteLoader): def fetch_bars(self, symbols, start, end): @@ -201,3 +270,38 @@ def test_final_metrics_expose_minute_contract_without_symbol_quality_details(): assert metrics["market_data_feed"] == "iex" assert metrics["sip_fallback_to_iex"] is True assert metrics["end_clamped"] is False + + +def test_the_engine_plans_the_same_closing_fill(): + """The dashboard engine's twin of the protocol test above: the plan both + paths build through ``plan_execution_fills`` from the same source bars.""" + from pathlib import Path + + from dashboard.backend.domain.backtesting import engine as engine_module + from dashboard.backend.infrastructure.market_data import profiles + + bars = _minute_bars(["AAPL"], None, None) + after_hours = pd.Timestamp("2026-04-15 20:00:00+00:00") + bars["AAPL"].loc[after_hours] = [999.0] * 5 + backtester = engine_module.HourlyBacktester.__new__( + engine_module.HourlyBacktester + ) + backtester.data_source = profiles.ALPACA + backtester.profile = profiles.get_market_profile(profiles.ALPACA) + backtester.intraday_mode = True + backtester.source_timeframe = "5m" + backtester.source_data = bars + decisions = list( + pd.date_range("2026-04-15 14:30", "2026-04-15 19:30", freq="h", tz="UTC") + ) + [after_hours] # 10:30 ... 15:30 and the 16:00 ET closing bucket + + kept, valuation, plan, fields = backtester._plan_executions(decisions) + + last_regular = pd.Timestamp("2026-04-15 19:55:00+00:00") + assert kept == decisions + assert after_hours not in valuation and valuation[-1] == last_regular + assert plan[after_hours] == last_regular and fields[after_hours] == "close" + assert all(fields[ts] == "open" and plan[ts] == ts for ts in decisions[:-1]) + # ...and the run loop prices the fill by that field, not a literal "open". + source = Path(engine_module.__file__).read_text(encoding="utf-8") + assert "symbol: row[execution_field]" in source diff --git a/dashboard/backend/tests/test_market_sessions.py b/dashboard/backend/tests/test_market_sessions.py index af832d18..2abcb8a7 100644 --- a/dashboard/backend/tests/test_market_sessions.py +++ b/dashboard/backend/tests/test_market_sessions.py @@ -89,26 +89,127 @@ def _sample_timestamps(): def test_every_us_filter_agrees(): """The engine, the baseline generator, the leaderboard strategies and the dataset store were four copies; a bar the store dropped and the engine kept - was a step the protocol and dashboard paths counted differently.""" + was a step the protocol and dashboard paths counted differently. Each + stamp convention is one rule, whichever of them applies it.""" from dashboard.backend import baseline_generator from dashboard.backend.domain.backtesting import market_data_store as mds from dashboard.backend.domain.leaderboard.strategies import _common stamps = _sample_timestamps() - expected = [ - ts for ts in stamps - if sessions.is_in_session(ts, market="US", timezone="US/Eastern") - ] - assert expected, "the sample must straddle the session" - assert baseline_generator._market_hours_only(stamps, "US/Eastern") == expected - assert _common.filter_market_hours(stamps) == expected frame = pd.DataFrame({"close": 1.0}, index=pd.DatetimeIndex(stamps)) - assert mds._build_trading_timestamps( - {"X": frame}, market="US", timezone="US/Eastern" - ) == expected - # The 16:00:30 stamp is in: the store's own first rewrite excluded it - # while the engine kept it. - assert pd.Timestamp("2026-04-15 20:00:30", tz="UTC") in expected + for open_minutes in (None, 60): + expected = [ + ts for ts in stamps + if sessions.is_in_session( + ts, + market="US", + timezone="US/Eastern", + open_stamped_minutes=open_minutes, + ) + ] + assert expected, "the sample must straddle the session" + assert baseline_generator._market_hours_only( + stamps, "US/Eastern", open_minutes + ) == expected + assert mds._build_trading_timestamps( + {"X": frame}, + market="US", + timezone="US/Eastern", + open_stamped_minutes=open_minutes, + ) == expected + if open_minutes == _common.LEADERBOARD_BAR_OPEN_MINUTES: + assert _common.filter_market_hours(stamps) == expected + # The 16:00:30 stamp is in for a close-stamped bar: the store's own first + # rewrite excluded it while the engine kept it. + assert sessions.is_in_session( + pd.Timestamp("2026-04-15 20:00:30", tz="UTC"), + market="US", + timezone="US/Eastern", + ) + + +def _et(clock): + return pd.Timestamp(f"2026-04-15 {clock}", tz="US/Eastern") + + +@pytest.mark.parametrize( + ("clock", "minutes", "kept"), + [ + # Alpaca 5m, stamped at the open: 09:30 through 15:55. + ("09:25", 5, False), + ("09:30", 5, True), + ("15:55", 5, True), + ("16:00", 5, False), # 16:00-16:05 is after hours + # Alpaca 1h, clock-aligned: 09:00 (closes 10:00) through 15:00. + ("08:00", 60, False), + ("09:00", 60, True), + ("15:00", 60, True), + ("16:00", 60, False), # the bar the close-stamp rule used to keep + ], +) +def test_an_open_stamped_bar_is_in_session_when_it_closes_in_one( + clock, minutes, kept +): + assert sessions.is_in_session( + _et(clock), + market="US", + timezone="US/Eastern", + open_stamped_minutes=minutes, + ) is kept + + +def test_open_stamped_hourly_day_is_seven_bars(): + """The raw 1h path keeps seven bars a day, as the close-stamp rule did -- + but 09:00-15:00 rather than 10:00-16:00.""" + day = pd.date_range("2026-04-15 04:00", "2026-04-15 19:00", freq="h", + tz="US/Eastern") + kept = [ + ts.strftime("%H:%M") for ts in day + if sessions.is_in_session( + ts, market="US", timezone="US/Eastern", open_stamped_minutes=60 + ) + ] + assert kept == ["09:00", "10:00", "11:00", "12:00", "13:00", "14:00", "15:00"] + + +def test_session_close_is_each_window_end(): + assert sessions.is_session_close(_et("16:00"), market="US", timezone="US/Eastern") + assert not sessions.is_session_close( + _et("15:30"), market="US", timezone="US/Eastern" + ) + for clock, is_close in (("11:30", True), ("15:00", True), ("13:00", False)): + assert sessions.is_session_close( + pd.Timestamp(f"2026-04-15 {clock}", tz="Asia/Shanghai"), + market="CN", + timezone="Asia/Shanghai", + ) is is_close + + +def test_only_alpaca_stamps_bars_at_their_open(): + assert profiles.bars_open_stamped(profiles.ALPACA) + assert not profiles.bars_open_stamped(profiles.IFIND_ASHARE) + assert not profiles.bars_open_stamped(profiles.VNPY_SIMULATION) + + +def test_the_final_bucket_fills_at_the_last_in_session_close(): + from dashboard.backend.domain.backtesting.bar_aggregation import ( + plan_execution_fills, + ) + + source = [_et("15:25"), _et("15:30"), _et("15:55")] + decisions = [_et("15:30"), _et("16:00"), _et("15:45")] + fills = plan_execution_fills( + decisions, source, market="US", timezone="US/Eastern" + ) + assert fills == { + _et("15:30"): (_et("15:30"), "open"), + _et("16:00"): (_et("15:55"), "close"), + # Not a session close and no bar opens there: no fill. + } + # A day with no source bars has nothing to fill the close on. + assert plan_execution_fills( + [_et("16:00")], [], market="US", timezone="US/Eastern" + ) == {} def test_the_engine_filter_agrees_for_both_markets(): @@ -197,3 +298,34 @@ def test_no_module_restates_the_session_bounds(): "session bounds restated outside market_data/sessions.py: " + ", ".join(offenders) ) + + +@pytest.mark.parametrize( + "strategy_key", ["buy_hold", "equal_weight_buyhold", "equal_weight_index"] +) +def test_leaderboard_baselines_read_the_board_bars_as_open_stamped(strategy_key): + """These three hand the board's raw Alpaca hourly bars to the baseline + generator, which cannot tell the stamp convention from the frame.""" + from dashboard.backend.domain.leaderboard.strategies import get_strategy + + hours = pd.date_range("2026-04-15 08:00", "2026-04-15 18:00", freq="h", + tz="US/Eastern") + frame = pd.DataFrame( + { + "open": 100.0, + "high": 101.0, + "low": 99.0, + "close": [100.0 + i for i in range(len(hours))], + "volume": 1000, + }, + index=hours.tz_convert("UTC"), + ) + strategy = get_strategy({"strategy": strategy_key, "symbols": ["AAPL", "MSFT"]}) + curve = strategy.run( + {"AAPL": frame, "MSFT": frame.copy()}, "2026-04-15", "2026-04-15", 10_000.0 + ) + clocks = [ + pd.Timestamp(point["timestamp"]).tz_convert("US/Eastern").strftime("%H:%M") + for point in curve + ] + assert clocks[0] == "09:00" and clocks[-1] == "15:00" From 220156491959b315508f6f7fdfabb8c804735c3d Mon Sep 17 00:00:00 2001 From: FlyM1ss Date: Wed, 23 Sep 2026 16:25:11 -0400 Subject: [PATCH 3/5] fix(sessions): record the session-close fill in the frequency contract The 16:00 decision now fills at the 15:55 close instead of being dropped, so a 5m-source run executes 7 decisions a day, not 6. The verified contract gains `session_close_fill: last_source_bar_close` (additive, so stored runs keep their label) and the backtest details label names it. The planner also returned the decision stamp on an exact match instead of the equal source bar, which re-stamped every trade in UTC. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../domain/backtesting/bar_aggregation.py | 4 +++- .../infrastructure/market_data/frequency.py | 4 ++++ .../backtesting/test_engine_minute_source.py | 8 +++++--- .../infrastructure/market_data/test_frequency.py | 1 + .../tests/test_admin_analytics_frontend.py | 2 +- .../backend/tests/test_analytics_frontend.py | 2 +- .../tests/test_backtest_comparison_frontend.py | 2 +- .../backend/tests/test_frontend_fast_boot.py | 2 +- dashboard/backend/tests/test_market_sessions.py | 14 ++++++++++++++ .../backend/tests/test_minute_data_frontend.py | 16 ++++++++++++++++ dashboard/frontend/app.html | 2 +- dashboard/frontend/app.js | 5 ++++- docs/minute-data-hourly-trading.md | 5 ++++- 13 files changed, 56 insertions(+), 11 deletions(-) diff --git a/dashboard/backend/domain/backtesting/bar_aggregation.py b/dashboard/backend/domain/backtesting/bar_aggregation.py index c8ce89ad..23c4259a 100644 --- a/dashboard/backend/domain/backtesting/bar_aggregation.py +++ b/dashboard/backend/domain/backtesting/bar_aggregation.py @@ -368,7 +368,9 @@ def plan_execution_fills( same_day = by_day.get(_market_day(timestamp, timezone), []) index = bisect_left(same_day, timestamp) if index < len(same_day) and same_day[index] == timestamp: - fills[timestamp] = (timestamp, "open") + # The source's own object, not the equal decision stamp: they can + # differ in tz, and the fill bar is what a trade is stamped with. + fills[timestamp] = (same_day[index], "open") elif index > 0 and is_session_close( timestamp, market=market, timezone=timezone ): diff --git a/dashboard/backend/infrastructure/market_data/frequency.py b/dashboard/backend/infrastructure/market_data/frequency.py index 32989356..88380a16 100644 --- a/dashboard/backend/infrastructure/market_data/frequency.py +++ b/dashboard/backend/infrastructure/market_data/frequency.py @@ -206,6 +206,10 @@ def build_verified_intraday_contract( { "aggregation": "session_anchored_completed_bars", "fill_policy": "next_source_bar_open", + # The session's last decision has no next source bar inside the + # session, so it fills at the last one's close instead of on an + # after-hours bar (``bar_aggregation.plan_execution_fills``). + "session_close_fill": "last_source_bar_close", "verification_status": "verified", } ) diff --git a/dashboard/backend/tests/backtesting/test_engine_minute_source.py b/dashboard/backend/tests/backtesting/test_engine_minute_source.py index 47a365c7..8678c592 100644 --- a/dashboard/backend/tests/backtesting/test_engine_minute_source.py +++ b/dashboard/backend/tests/backtesting/test_engine_minute_source.py @@ -115,9 +115,10 @@ def buy_once(self, state): backtester.calculate_indicators() run_id, equity_curve = backtester.run_agent_backtest() - # Seven completed hourly buckets exist per day, but the 16:00 bucket has - # no next source bar and is intentionally not an executable decision. - assert len(decisions) == 60 + # Seven completed hourly buckets per day, all executable: the 16:00 + # bucket has no next in-session source bar, so it fills at the 15:55 + # bar's close rather than being dropped. + assert len(decisions) == 70 assert len(equity_curve) == 780 assert run_id.startswith("agent_") @@ -129,6 +130,7 @@ def buy_once(self, state): assert frequency["source_timeframe"] == "5m" assert frequency["decision_frequency"] == "1h" assert frequency["fill_policy"] == "next_source_bar_open" + assert frequency["session_close_fill"] == "last_source_bar_close" assert frequency["verification_status"] == "verified" quality = fake_db.runs[0]["metadata"]["market_data_quality"] assert quality["policy"] == "drop_incomplete_decision_bars" diff --git a/dashboard/backend/tests/infrastructure/market_data/test_frequency.py b/dashboard/backend/tests/infrastructure/market_data/test_frequency.py index f24b7e63..afc2a0f5 100644 --- a/dashboard/backend/tests/infrastructure/market_data/test_frequency.py +++ b/dashboard/backend/tests/infrastructure/market_data/test_frequency.py @@ -76,6 +76,7 @@ def test_runtime_contract_attests_fixed_intraday_execution_policy(): "valuation_frequency": "5m", "aggregation": "session_anchored_completed_bars", "fill_policy": "next_source_bar_open", + "session_close_fill": "last_source_bar_close", "verification_status": "verified", } diff --git a/dashboard/backend/tests/test_admin_analytics_frontend.py b/dashboard/backend/tests/test_admin_analytics_frontend.py index 3830a206..fabcb83e 100644 --- a/dashboard/backend/tests/test_admin_analytics_frontend.py +++ b/dashboard/backend/tests/test_admin_analytics_frontend.py @@ -162,7 +162,7 @@ def test_app_lifecycle_and_cache_versions_are_wired(): # Lockstep owner for the console's bumped tags and the /admin page's pins: # every bump edits this test in the same change (Global Constraints). assert 'styles.css?v=146' in APP_HTML - assert 'app.js?v=140' in APP_HTML + assert 'app.js?v=141' in APP_HTML assert 'js/admin-tabs.js?v=12' in APP_HTML for tag in ( 'href="admin.css?v=4"', diff --git a/dashboard/backend/tests/test_analytics_frontend.py b/dashboard/backend/tests/test_analytics_frontend.py index 008a64d4..b18d6c89 100644 --- a/dashboard/backend/tests/test_analytics_frontend.py +++ b/dashboard/backend/tests/test_analytics_frontend.py @@ -41,7 +41,7 @@ def _function_body(source: str, signature: str) -> str: def test_analytics_script_loads_between_app_and_page_scripts(): - app_at = APP_HTML.index('') + app_at = APP_HTML.index('') analytics_at = APP_HTML.index( '' ) diff --git a/dashboard/backend/tests/test_backtest_comparison_frontend.py b/dashboard/backend/tests/test_backtest_comparison_frontend.py index b4104090..0e363d17 100644 --- a/dashboard/backend/tests/test_backtest_comparison_frontend.py +++ b/dashboard/backend/tests/test_backtest_comparison_frontend.py @@ -191,7 +191,7 @@ def test_exact_raw_ties_mark_every_tied_series_best(): def test_comparison_script_and_semantic_table_ship_before_app(): helper = '' - app = '' + app = '' assert 'href="styles.css?v=146"' in APP_HTML assert APP_HTML.index(helper) < APP_HTML.index(app) for element_id in ( diff --git a/dashboard/backend/tests/test_frontend_fast_boot.py b/dashboard/backend/tests/test_frontend_fast_boot.py index fa6547f1..55aa99a5 100644 --- a/dashboard/backend/tests/test_frontend_fast_boot.py +++ b/dashboard/backend/tests/test_frontend_fast_boot.py @@ -191,7 +191,7 @@ def test_cache_busters_bumped(): # the next bump, so the exact one looks like the broken guard and gets # "fixed" by loosening it. That collision has already cost this repo one # round of follow-ups (#347/#348). - assert "app.js?v=140" in APP_HTML + assert "app.js?v=141" in APP_HTML assert "js/agent-editor.js?v=31" in APP_HTML assert "styles.css?v=146" in APP_HTML assert "js/leaderboard.js?v=33" in APP_HTML diff --git a/dashboard/backend/tests/test_market_sessions.py b/dashboard/backend/tests/test_market_sessions.py index 2abcb8a7..ed14669c 100644 --- a/dashboard/backend/tests/test_market_sessions.py +++ b/dashboard/backend/tests/test_market_sessions.py @@ -212,6 +212,20 @@ def test_the_final_bucket_fills_at_the_last_in_session_close(): ) == {} +def test_an_exact_fill_is_the_source_bar_not_the_equal_decision_stamp(): + from dashboard.backend.domain.backtesting.bar_aggregation import ( + plan_execution_fills, + ) + + # Aggregated decisions arrive in UTC, source bars in ET: equal instants, + # but the trade is stamped with the fill bar, so its tz must survive. + decision = _et("15:30").tz_convert("UTC") + fills = plan_execution_fills( + [decision], [_et("15:30")], market="US", timezone="US/Eastern" + ) + assert str(fills[decision][0].tz) == "US/Eastern" + + def test_the_engine_filter_agrees_for_both_markets(): from dashboard.backend.domain.backtesting.engine import HourlyBacktester diff --git a/dashboard/backend/tests/test_minute_data_frontend.py b/dashboard/backend/tests/test_minute_data_frontend.py index bdf665c5..4ace1c74 100644 --- a/dashboard/backend/tests/test_minute_data_frontend.py +++ b/dashboard/backend/tests/test_minute_data_frontend.py @@ -79,6 +79,22 @@ def test_minute_frequency_formatter_states_fixed_execution_policy(): assert value == "5m source · 1h decisions · next 5m open fills · 5m valuation" +def test_minute_frequency_formatter_names_the_session_close_fill(): + value = _run_formatters( + "formatBacktestFrequencyContract({" + "source_timeframe:'5m',decision_timeframe:'60m'," + "decision_frequency:'1h',execution_timeframe:'5m'," + "valuation_frequency:'5m',fill_policy:'next_source_bar_open'," + "session_close_fill:'last_source_bar_close'" + "})" + ) + + assert value == ( + "5m source · 1h decisions · next 5m open fills " + "(last close at session end) · 5m valuation" + ) + + def test_quality_formatter_reports_dropped_and_problem_counts(): value = _run_formatters( "formatBacktestMarketDataQuality({" diff --git a/dashboard/frontend/app.html b/dashboard/frontend/app.html index 9f404f33..f62508de 100644 --- a/dashboard/frontend/app.html +++ b/dashboard/frontend/app.html @@ -2393,7 +2393,7 @@

Refund Credits purchase

- + diff --git a/dashboard/frontend/app.js b/dashboard/frontend/app.js index db657001..1fee9656 100644 --- a/dashboard/frontend/app.js +++ b/dashboard/frontend/app.js @@ -10089,9 +10089,12 @@ function formatBacktestFrequencyContract(contract) { const execution = contract.execution_timeframe; const valuation = contract.valuation_frequency; if (!source || !decision || !execution || !valuation) return null; - const fill = contract.fill_policy === 'next_source_bar_open' + let fill = contract.fill_policy === 'next_source_bar_open' ? `next ${execution} open fills` : `${execution} execution`; + if (contract.session_close_fill === 'last_source_bar_close') { + fill += ' (last close at session end)'; + } const verification = contract.verification_status === 'verified' ? ' · verified' : ''; diff --git a/docs/minute-data-hourly-trading.md b/docs/minute-data-hourly-trading.md index 25e424b4..3b769229 100644 --- a/docs/minute-data-hourly-trading.md +++ b/docs/minute-data-hourly-trading.md @@ -44,7 +44,10 @@ Phase 3 将分钟链路的数据质量和无未来数据约束固化为可测试 ## Phase 4：API 可观测性 Phase 4 保持唯一、固定的成交规则 `next_source_bar_open`，不增加成交策略 -配置项。回测详情 API 和外部 Agent 的完成结果会返回： +配置项。唯一例外是每个交易日收盘时点的决策（如 16:00 ET）：会话内已没有 +下一根源 bar，因此以最后一根会话内源 bar（15:55）的收盘价成交，而不是 +盘后 bar 的开盘价；契约中以 `session_close_fill: last_source_bar_close` +记录。回测详情 API 和外部 Agent 的完成结果会返回： - `frequency_contract`：5m 源数据、60m 决策 bar、1h 决策、5m 成交与估值； - `market_data_quality`：聚合后的可用、丢弃及异常计数。 From 0a61438621fede73e435b8ae438156fa6cbac7ee Mon Sep 17 00:00:00 2001 From: FlyM1ss Date: Wed, 23 Sep 2026 17:04:49 -0400 Subject: [PATCH 4/5] fix(sessions): carry the bar-stamp convention on the frame Address PR #529 review findings: the loader stamps open-stamped frames, session filters read that attr instead of a per-caller parameter (the baseline worker path lost every 16:00 close), aggregation refuses unlabelled sources, and one ExecutionFill record replaces the parallel execution lists. The session-close fill is stamped 16:00, and session_close_fill now passes the API allow-list. Co-Authored-By: Claude Opus 5.5 (1M context) --- dashboard/backend/api/routers/backtests.py | 1 + dashboard/backend/baseline_generator.py | 25 ++-- .../domain/backtesting/bar_aggregation.py | 83 +++++++++---- .../backend/domain/backtesting/engine.py | 84 ++++++------- .../backtesting/external_run_service.py | 45 +++---- .../domain/backtesting/market_data_store.py | 60 +++++----- .../backend/domain/leaderboard/baselines.py | 5 +- .../domain/leaderboard/strategies/_common.py | 36 +++--- .../domain/leaderboard/strategies/buy_hold.py | 5 +- .../strategies/equal_weight_buyhold.py | 5 +- .../strategies/equal_weight_index.py | 4 +- .../infrastructure/market_data/alpaca_bars.py | 20 ++++ .../infrastructure/market_data/frequency.py | 6 +- .../infrastructure/market_data/profiles.py | 12 -- .../infrastructure/market_data/sessions.py | 63 ++++++++-- .../tests/backtesting/test_bar_aggregation.py | 46 +++++++- .../backtesting/test_engine_minute_source.py | 4 + .../tests/backtesting/test_engine_move.py | 4 + .../backend/tests/test_backtests_router.py | 21 ++++ .../tests/test_external_minute_source.py | 84 +++++++++++-- .../backend/tests/test_market_data_store.py | 4 + .../backend/tests/test_market_sessions.py | 111 ++++++++++++------ .../tests/test_minute_data_frontend.py | 13 ++ dashboard/frontend/app.js | 11 +- docs/minute-data-hourly-trading.md | 4 +- 25 files changed, 499 insertions(+), 257 deletions(-) diff --git a/dashboard/backend/api/routers/backtests.py b/dashboard/backend/api/routers/backtests.py index b90b7605..32d32bd6 100644 --- a/dashboard/backend/api/routers/backtests.py +++ b/dashboard/backend/api/routers/backtests.py @@ -464,6 +464,7 @@ def _run_metadata_response(run: Dict[str, Any]) -> RunMetadata: "valuation_frequency", "aggregation", "fill_policy", + "session_close_fill", "verification_status", ) if name in metadata[field] diff --git a/dashboard/backend/baseline_generator.py b/dashboard/backend/baseline_generator.py index f80704b7..e7017f2e 100644 --- a/dashboard/backend/baseline_generator.py +++ b/dashboard/backend/baseline_generator.py @@ -21,6 +21,7 @@ from dashboard.backend.domain.backtesting.market_rules import MarketRuleCalendar from dashboard.backend.domain.trading.execution import calculate_transaction_costs from dashboard.backend.infrastructure.market_data.sessions import ( + frames_open_stamped_minutes, is_in_session, market_for_timezone, ) @@ -38,18 +39,17 @@ ) from exc -def _market_hours_only( - timestamps, market_timezone: str, open_stamped_minutes: Optional[int] = None -): +def _market_hours_only(timestamps, market_timezone: str, bars): """Keep regular sessions in the timezone belonging to the market profile. This API takes only the zone, so the market is recovered from it through - the profile registry rather than a zone-string comparison kept here. - ``open_stamped_minutes`` is set when the bars are raw Alpaca bars, stamped - at their open (see ``sessions.is_in_session``); aggregated decision bars - and iFinD bars are stamped at their close and leave it ``None``. + the profile registry rather than a zone-string comparison kept here. The + stamp convention is read off ``bars``, the frames the timestamps came from + (``sessions.frames_open_stamped_minutes``): raw Alpaca bars are stamped at + their open, aggregated decision bars and iFinD bars at their close. """ market = market_for_timezone(market_timezone) + open_stamped_minutes = frames_open_stamped_minutes(bars) return [ timestamp for timestamp in timestamps @@ -255,7 +255,6 @@ def generate_buyhold_baseline( lot_size: int = 1, allocation_summary: Optional[Dict[str, Any]] = None, market_rule_calendar: MarketRuleCalendar | None = None, - open_stamped_minutes: Optional[int] = None, ) -> List[Dict]: """ Generate Buy & Hold baseline curve. @@ -303,7 +302,7 @@ def generate_buyhold_baseline( return [] all_timestamps = _market_hours_only( - all_timestamps, market_timezone, open_stamped_minutes + all_timestamps, market_timezone, bars_subset ) all_timestamps = _timestamps_in_window( all_timestamps, start_date, end_date, market_timezone @@ -563,7 +562,6 @@ def generate_index_baseline( symbols_to_track: Optional[List[str]] = None, market_timezone: str = "US/Eastern", currency_context: CurrencyContext | None = None, - open_stamped_minutes: Optional[int] = None, ) -> List[Dict]: """ Generate Index baseline curve (equal-weight index). @@ -599,7 +597,7 @@ def generate_index_baseline( return [] all_timestamps = _market_hours_only( - all_timestamps, market_timezone, open_stamped_minutes + all_timestamps, market_timezone, bars_subset ) all_timestamps = _timestamps_in_window( all_timestamps, start_date, end_date, market_timezone @@ -700,7 +698,6 @@ def generate_baselines( lot_size: int = 1, allocation_summary: Optional[Dict[str, Any]] = None, market_rule_calendar: MarketRuleCalendar | None = None, - open_stamped_minutes: Optional[int] = None, ) -> Tuple[List[Dict], List[Dict]]: """ Generate both baselines (Buy & Hold, Index). @@ -715,8 +712,6 @@ def generate_baselines( ``BaselineGenerator.generate_buyhold_baseline``. allocation_summary: Out-dict describing how much of the buy & hold sleeve actually filled. - open_stamped_minutes: Bar span when the bars are stamped at their - open (raw Alpaca bars); ``None`` for close-stamped bars. Returns: Tuple of (buyhold_curve, index_curve) @@ -736,7 +731,6 @@ def generate_baselines( lot_size=lot_size, allocation_summary=allocation_summary, market_rule_calendar=market_rule_calendar, - open_stamped_minutes=open_stamped_minutes, ) index_curve = generator.generate_index_baseline( @@ -747,7 +741,6 @@ def generate_baselines( symbols_list, market_timezone, currency_context, - open_stamped_minutes=open_stamped_minutes, ) return buyhold_curve, index_curve diff --git a/dashboard/backend/domain/backtesting/bar_aggregation.py b/dashboard/backend/domain/backtesting/bar_aggregation.py index 23c4259a..50ca7e47 100644 --- a/dashboard/backend/domain/backtesting/bar_aggregation.py +++ b/dashboard/backend/domain/backtesting/bar_aggregation.py @@ -15,7 +15,7 @@ from bisect import bisect_left from datetime import date, time -from typing import Any, Dict, Iterable, List, Mapping, Tuple +from typing import Any, Dict, Iterable, List, Mapping, NamedTuple import numpy as np import pandas as pd @@ -26,7 +26,9 @@ ) # Re-exported: the session bounds have one owner, and it is not this module. from dashboard.backend.infrastructure.market_data.sessions import ( + FRAME_ATTR_OPEN_STAMPED_MINUTES, is_session_close, + market_local, session_windows, ) @@ -112,7 +114,19 @@ def aggregate_bars( f"source bars are missing required columns: {', '.join(missing)}" ) if frame.empty: - return frame.copy() + result = frame.copy() + result.attrs.pop(FRAME_ATTR_OPEN_STAMPED_MINUTES, None) + return result + # Bucketing below reads a source stamp as its bar's OPEN (09:30-10:25 -> + # the 10:30 bar), and ``plan_execution_fills`` then fills at the bar opening + # on a decision's close. A close-stamped source would land one bar late in + # every bucket with nothing to show for it, so it is refused, not guessed. + stamped = frame.attrs.get(FRAME_ATTR_OPEN_STAMPED_MINUTES) + if stamped != source_minutes: + raise BarAggregationError( + f"aggregation requires {source}-bars stamped at their open; the " + f"frame is stamped {'at the close' if stamped is None else f'{stamped}m at the open'}" + ) local = _as_local_index(frame, timezone) windows = session_windows(market) @@ -255,6 +269,8 @@ def aggregate_bars( result = pd.DataFrame.from_records(records).set_index("timestamp").sort_index() result.attrs.update(dict(getattr(frame, "attrs", {}) or {})) + # A decision bar is stamped at its close, whatever its source was. + result.attrs.pop(FRAME_ATTR_OPEN_STAMPED_MINUTES, None) result.attrs.update( { "aggregation_source_timeframe": source, @@ -332,47 +348,64 @@ def summarize_aggregation_quality( return summary -def _market_day(timestamp: Any, timezone: str) -> date: - stamp = pd.Timestamp(timestamp) - if stamp.tzinfo is None: - return stamp.tz_localize(timezone).date() - return stamp.tz_convert(timezone).date() +class ExecutionFill(NamedTuple): + """How one decision fills: the source ``bar`` it is priced from, which of + that bar's prices (``price_field``), and the instant it fills + (``filled_at``). One record rather than parallel lists, so a bar can never + be paired with another step's price field.""" + + bar: Any + price_field: str + filled_at: Any def plan_execution_fills( decision_timestamps: Iterable[Any], source_timestamps: List[Any], *, + source_minutes: int, market: str, timezone: str, -) -> Dict[Any, Tuple[Any, str]]: - """Map each decision bar to ``(source bar, price field)`` it fills on. - - A decision closes at its stamp, so it fills at the ``open`` of the source - bar opening at that instant -- the first fill without look-ahead. A - session's final bucket (16:00 ET) has no such bar once after-hours bars are - out of the source set, so it fills at the ``close`` of the last source bar - before it that day: the same instant, priced at the last regular-hours - trade. Filling it at the 16:00 bar instead made the day's closing decision - depend on whether the tape served an after-hours bar at all. +) -> Dict[Any, ExecutionFill]: + """Map each decision bar to the :class:`ExecutionFill` it fills on. + + ``source_timestamps`` are open-stamped bars of ``source_minutes`` (the only + kind :func:`aggregate_bars` accepts), sorted. A decision closes at its + stamp, so it fills at the ``open`` of the source bar opening at that + instant -- the first fill without look-ahead. A session's final bucket + (16:00 ET) has no such bar once after-hours bars are out of the source set, + so it fills at the ``close`` of the source bar ending at that instant, + priced at the last regular-hours trade and stamped at that bar's close + rather than its open, so the trade never predates the decision. + Filling it at the 16:00 bar instead made the day's closing decision depend + on whether the tape served an after-hours bar at all. One planner for the engine and the protocol path's dataset store, which each used to carry their own copy of the exact-match rule. Decisions with - no fill are absent. ``source_timestamps`` must be sorted. + no fill are absent. """ by_day: Dict[date, List[Any]] = {} for timestamp in source_timestamps: - by_day.setdefault(_market_day(timestamp, timezone), []).append(timestamp) - fills: Dict[Any, Tuple[Any, str]] = {} + by_day.setdefault(market_local(timestamp, timezone).date(), []).append( + timestamp + ) + span = pd.Timedelta(minutes=source_minutes) + fills: Dict[Any, ExecutionFill] = {} for timestamp in decision_timestamps: - same_day = by_day.get(_market_day(timestamp, timezone), []) + same_day = by_day.get(market_local(timestamp, timezone).date(), []) index = bisect_left(same_day, timestamp) if index < len(same_day) and same_day[index] == timestamp: # The source's own object, not the equal decision stamp: they can # differ in tz, and the fill bar is what a trade is stamped with. - fills[timestamp] = (same_day[index], "open") - elif index > 0 and is_session_close( - timestamp, market=market, timezone=timezone + bar = same_day[index] + fills[timestamp] = ExecutionFill(bar, "open", bar) + elif ( + index > 0 + and same_day[index - 1] + span == timestamp + and is_session_close(timestamp, market=market, timezone=timezone) ): - fills[timestamp] = (same_day[index - 1], "close") + # Only the bar closing AT the session close: an earlier one's close + # predates the decision it would fill. + bar = same_day[index - 1] + fills[timestamp] = ExecutionFill(bar, "close", bar + span) return fills diff --git a/dashboard/backend/domain/backtesting/engine.py b/dashboard/backend/domain/backtesting/engine.py index 27751437..aa49ffbb 100644 --- a/dashboard/backend/domain/backtesting/engine.py +++ b/dashboard/backend/domain/backtesting/engine.py @@ -68,6 +68,7 @@ ) from dashboard.backend.domain.backtesting.features import TechnicalIndicators from dashboard.backend.domain.backtesting.bar_aggregation import ( + ExecutionFill, aggregate_bars_by_symbol, plan_execution_fills, summarize_aggregation_quality, @@ -120,13 +121,15 @@ timeframe_minutes, verify_source_timeframe, ) -from dashboard.backend.infrastructure.market_data.sessions import is_in_session +from dashboard.backend.infrastructure.market_data.sessions import ( + frames_open_stamped_minutes, + is_in_session, +) from dashboard.backend.infrastructure.market_data.profiles import ( IFIND_ASHARE, LLM_DECISION_SOURCE, RULE_BASED_DECISION_SOURCE, MarketProfile, - bars_open_stamped, get_market_profile, resolve_decision_source, ) @@ -1624,15 +1627,17 @@ def _current_equity(self, manager: PortfolioManager, timestamp=None) -> float: return self._require_currency_context().to_reporting(manager.cash, timestamp) return float(manager.cash) - def _market_hours_only(self, timestamps, *, open_stamped_minutes=None): + def _market_hours_only(self, timestamps, bars): """Filter timestamps using the selected market's local sessions. Through ``market_data.sessions``, the same owner the dataset store and the aggregation use, so the dashboard and protocol paths cannot count - different steps for one window. ``open_stamped_minutes`` is set for raw - provider bars stamped at their open; see :meth:`_raw_bar_open_minutes`. + different steps for one window. The stamp convention is read off + ``bars``, the frames the timestamps came from: raw Alpaca bars are + stamped at their open, aggregated decision bars at their close. """ profile = self._effective_profile() + open_stamped_minutes = frames_open_stamped_minutes(bars) return [ timestamp for timestamp in timestamps @@ -1644,54 +1649,41 @@ def _market_hours_only(self, timestamps, *, open_stamped_minutes=None): ) ] - def _raw_bar_open_minutes(self, timeframe): - """The span of a raw provider bar when the provider stamps it at its - open, else ``None``. Aggregated decision bars are always stamped at - their close and must not go through this.""" - if not bars_open_stamped(getattr(self, "data_source", None)): - return None - return timeframe_minutes(timeframe) - def _plan_executions(self, decision_timestamps): - """``(decisions, valuation bars, {decision: fill bar}, {decision: - price field})`` for the run loop. + """``(decisions, valuation bars, {decision: ExecutionFill})`` for the + run loop. - Hourly mode fills and values on the decision bar itself. Minute mode - values on every in-session source bar and fills through - ``plan_execution_fills``; a decision it cannot fill is not a step. + Hourly mode fills and values on the decision bar itself, priced off its + market data (the bar's close). Minute mode values on every in-session + source bar and fills through ``plan_execution_fills``; a decision it + cannot fill is not a step. """ if not self.intraday_mode: return ( list(decision_timestamps), list(decision_timestamps), - {timestamp: timestamp for timestamp in decision_timestamps}, - {}, + { + timestamp: ExecutionFill(timestamp, "close", timestamp) + for timestamp in decision_timestamps + }, ) raw_timestamps = self._market_hours_only( - self._timestamps_for_data(self.source_data), - open_stamped_minutes=self._raw_bar_open_minutes(self.source_timeframe), + self._timestamps_for_data(self.source_data), self.source_data ) profile = self._effective_profile() fills = plan_execution_fills( decision_timestamps, raw_timestamps, + source_minutes=timeframe_minutes(self.source_timeframe), market=profile.market, timezone=profile.timezone, ) return ( [timestamp for timestamp in decision_timestamps if timestamp in fills], raw_timestamps, - {timestamp: source for timestamp, (source, _field) in fills.items()}, - {timestamp: field for timestamp, (_source, field) in fills.items()}, + fills, ) - def _decision_bar_open_minutes(self): - """:meth:`_raw_bar_open_minutes` for ``all_data``'s bars: ``None`` in - minute mode, where they are aggregated and stamped at their close.""" - if getattr(self, "intraday_mode", False): - return None - return self._raw_bar_open_minutes(self.source_timeframe) - def _run_daily_post_trade( self, *, @@ -1818,10 +1810,7 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: all_timestamps = filtered - all_timestamps = self._market_hours_only( - all_timestamps, - open_stamped_minutes=self._decision_bar_open_minutes(), - ) + all_timestamps = self._market_hours_only(all_timestamps, self.all_data) prior_market_dates = ( _prior_market_date_by_decision_date(all_timestamps) if self.runtime_type == AI_HEDGE_FUND_RUNTIME_TYPE @@ -1839,12 +1828,9 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: f"failed step(s) before aborting\n" ) - ( - all_timestamps, - raw_timestamps, - execution_plan, - execution_fields, - ) = self._plan_executions(all_timestamps) + all_timestamps, raw_timestamps, execution_plan = self._plan_executions( + all_timestamps + ) print( f" Trading {len(all_timestamps)} hourly decision bars during " @@ -1986,10 +1972,10 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: # In minute mode, execute at the next source bar's open. The # decision bar closes at ``timestamp``; the source bar opening at # that same instant is the first non-look-ahead fill opportunity. - # A session's final bucket fills at the last source bar's close - # instead (``plan_execution_fills``). - execution_timestamp = execution_plan[timestamp] - execution_field = execution_fields.get(timestamp, "open") + # A session's final bucket fills at the close of the source bar + # ending then instead (``plan_execution_fills``). + fill = execution_plan[timestamp] + execution_timestamp = fill.bar execution_market_data = market_data execution_fallback_prices = { symbol: values[execution_timestamp] @@ -2004,9 +1990,9 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: execution_timestamp, ) execution_prices = { - symbol: row[execution_field] + symbol: row[fill.price_field] for symbol, row in execution_market_data.items() - if execution_field in row + if fill.price_field in row } # Execute trades (only if real data available) @@ -2014,7 +2000,7 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: manager.execute_actions( decision["actions"], execution_market_data, - execution_timestamp, + fill.filled_at, fallback_prices=execution_fallback_prices, execution_prices=execution_prices, ) @@ -2281,7 +2267,6 @@ def run_buyhold_baseline(self) -> Tuple[str, List[Dict]]: currency_context=self._require_currency_context(), transaction_cost_profile=profile.transaction_cost_profile, transaction_cost_totals=baseline_cost_totals, - open_stamped_minutes=self._decision_bar_open_minutes(), # The board lot is its own market rule. Deriving it from the cost # profile would floor buys to 100 in any future market that charges # fees but trades in single shares. @@ -2371,7 +2356,6 @@ def run_djia_baseline(self) -> Tuple[str, List[Dict]]: market_timezone=self._effective_profile().timezone, symbols_list=list(DJIA_30), currency_context=self._require_currency_context(), - open_stamped_minutes=self._decision_bar_open_minutes(), ) if not equity_history: diff --git a/dashboard/backend/domain/backtesting/external_run_service.py b/dashboard/backend/domain/backtesting/external_run_service.py index 48ed40be..d384f685 100644 --- a/dashboard/backend/domain/backtesting/external_run_service.py +++ b/dashboard/backend/domain/backtesting/external_run_service.py @@ -33,6 +33,7 @@ actions_to_executable, parse_actions_payload, ) +from dashboard.backend.domain.backtesting.bar_aggregation import ExecutionFill from dashboard.backend.domain.backtesting.constants import ( fractional_return, resolve_initial_capital, @@ -304,8 +305,7 @@ def __init__( self.source_data: Dict[str, pd.DataFrame] = {} self.source_timestamps: List[Any] = [] self.source_price_cache: Dict[str, Dict[Any, float]] = {} - self.execution_timestamps: List[Any] = [] - self.execution_price_fields: List[str] = [] + self.execution_fills: List[ExecutionFill] = [] self.data_quality: Dict[str, Any] = {} self.frequency_contract: Optional[Dict[str, str]] = None self.market_data_provenance: Dict[str, Any] = {} @@ -372,12 +372,7 @@ def adopt_dataset(self, dataset: "market_data_store.MarketDataset") -> None: self.source_price_cache = getattr( dataset, "source_price_cache", self.price_cache ) - self.execution_timestamps = list( - getattr(dataset, "execution_timestamps", self.timestamps) - ) - self.execution_price_fields = list( - getattr(dataset, "execution_price_fields", None) or [] - ) + self.execution_fills = list(getattr(dataset, "execution_fills", None) or []) self.data_quality = dict(getattr(dataset, "data_quality", {}) or {}) self.equity_metadata = dict(getattr(dataset, "equity_metadata", {}) or {}) self.source_timeframe = getattr( @@ -441,17 +436,13 @@ def _effective_source_timestamps(self) -> List[Any]: def _effective_source_price_cache(self) -> Dict[str, Dict[Any, float]]: return self.source_price_cache or self.price_cache - def _effective_execution_timestamps(self) -> List[Any]: - if len(self.execution_timestamps) == self.total_steps: - return self.execution_timestamps - return list(self.timestamps) - - def _execution_price_field(self, step_index: int) -> str: - """The execution bar's "open", or its "close" for a session's final - bucket; see ``bar_aggregation.plan_execution_fills``.""" - if len(self.execution_price_fields) == self.total_steps: - return self.execution_price_fields[step_index] - return "open" + def _effective_execution_fills(self) -> List[ExecutionFill]: + """One ``ExecutionFill`` per step. A dataset with no plan for these + steps fills each on its own bar's open; the bar and its price field + travel together, so a remapped bar can never borrow a default field.""" + if len(self.execution_fills) == self.total_steps: + return self.execution_fills + return [ExecutionFill(timestamp, "open", timestamp) for timestamp in self.timestamps] def _value_through(self, target_timestamp=None) -> None: """Mark the portfolio on each source bar through the given timestamp.""" @@ -785,20 +776,20 @@ def _advance_step( # agent_runs.metadata). self.timeout_holds += 1 timestamp = self.timestamps[self.step_index] - execution_timestamp = self._effective_execution_timestamps()[self.step_index] + fill = self._effective_execution_fills()[self.step_index] + execution_timestamp = fill.bar execution_market_data = self._source_market_data_at(execution_timestamp) - execution_field = self._execution_price_field(self.step_index) execution_prices = { - symbol: row[execution_field] + symbol: row[fill.price_field] for symbol, row in execution_market_data.items() - if execution_field in row + if fill.price_field in row } trades_before_execution = len(self.manager.trades) self.manager.execute_actions( executable, execution_market_data, - execution_timestamp, + fill.filled_at, fallback_prices={ symbol: values[execution_timestamp] for symbol, values in self._effective_source_price_cache().items() @@ -822,9 +813,9 @@ def _advance_step( "timestamp": timestamp.isoformat() if hasattr(timestamp, "isoformat") else str(timestamp), - "execution_timestamp": execution_timestamp.isoformat() - if hasattr(execution_timestamp, "isoformat") - else str(execution_timestamp), + "execution_timestamp": fill.filled_at.isoformat() + if hasattr(fill.filled_at, "isoformat") + else str(fill.filled_at), "decision_source": decision_source, "actions_submitted": raw_actions or [], "actions_executed": len(self.last_executed), diff --git a/dashboard/backend/domain/backtesting/market_data_store.py b/dashboard/backend/domain/backtesting/market_data_store.py index 1ecc6503..c4d81ce6 100644 --- a/dashboard/backend/domain/backtesting/market_data_store.py +++ b/dashboard/backend/domain/backtesting/market_data_store.py @@ -36,6 +36,7 @@ from dashboard.backend.domain.backtesting.features import TechnicalIndicators from dashboard.backend.domain.backtesting.bar_aggregation import ( + ExecutionFill, aggregate_bars_by_symbol, plan_execution_fills, summarize_aggregation_quality, @@ -53,6 +54,7 @@ from dashboard.backend.infrastructure.market_data.sessions import ( DEFAULT_MARKET, canonical_market, + frames_open_stamped_minutes, is_in_session, timezone_for_market, ) @@ -79,7 +81,7 @@ class MarketDataset: __slots__ = ( "key", "all_data", "timestamps", "price_cache", "total_steps", "source_data", "source_timestamps", "source_price_cache", - "execution_timestamps", "execution_price_fields", + "execution_fills", "source_timeframe", "decision_timeframe", "data_quality", "equity_metadata", @@ -90,8 +92,7 @@ def __init__(self, key: Tuple, all_data: Dict[str, pd.DataFrame], *, source_data: Optional[Dict[str, pd.DataFrame]] = None, source_timestamps: Optional[List[Any]] = None, source_price_cache: Optional[Dict[str, Dict[Any, float]]] = None, - execution_timestamps: Optional[List[Any]] = None, - execution_price_fields: Optional[List[str]] = None, + execution_fills: Optional[List[ExecutionFill]] = None, source_timeframe: str = "60m", decision_timeframe: str = "60m", data_quality: Optional[Dict[str, Any]] = None, @@ -110,17 +111,13 @@ def __init__(self, key: Tuple, all_data: Dict[str, pd.DataFrame], if source_price_cache is not None else price_cache ) - self.execution_timestamps = ( - execution_timestamps - if execution_timestamps is not None - else list(timestamps) - ) - # Which price of the execution bar fills each step: its "open", or the - # "close" for a session's final bucket (``plan_execution_fills``). - self.execution_price_fields = ( - execution_price_fields - if execution_price_fields is not None - else ["open"] * len(self.execution_timestamps) + # One ExecutionFill per step. Without a plan -- a dataset whose + # decision bars ARE its source bars -- each step fills at its own bar's + # open, as the protocol path always has. + self.execution_fills = ( + execution_fills + if execution_fills is not None + else [ExecutionFill(timestamp, "open", timestamp) for timestamp in timestamps] ) self.source_timeframe = source_timeframe self.decision_timeframe = decision_timeframe @@ -404,14 +401,10 @@ def _build_dataset( all_data, timezone=timezone, ) - # This store's loaders are Alpaca's, which stamp a raw bar at its open; - # an aggregated decision bar is stamped at its close. - source_open_minutes = timeframe_minutes(actual_source) timestamps = _build_trading_timestamps( all_data, market=market, timezone=timezone, - open_stamped_minutes=None if aggregated else source_open_minutes, ) if not timestamps: raise RuntimeError("No trading hours in the selected date range") @@ -421,17 +414,21 @@ def _build_dataset( min_symbol_coverage=0.0, market=market, timezone=timezone, - open_stamped_minutes=source_open_minutes, ) source_price_cache = _build_price_cache(source_data, source_timestamps) - fills = plan_execution_fills( - timestamps, source_timestamps, market=market, timezone=timezone - ) - if len(fills) < len(timestamps): - timestamps = [timestamp for timestamp in timestamps if timestamp in fills] - price_cache = _build_price_cache(all_data, timestamps) - execution_timestamps = [fills[timestamp][0] for timestamp in timestamps] - execution_price_fields = [fills[timestamp][1] for timestamp in timestamps] + execution_fills = None + if aggregated: + fills = plan_execution_fills( + timestamps, + source_timestamps, + source_minutes=timeframe_minutes(actual_source), + market=market, + timezone=timezone, + ) + if len(fills) < len(timestamps): + timestamps = [timestamp for timestamp in timestamps if timestamp in fills] + price_cache = _build_price_cache(all_data, timestamps) + execution_fills = [fills[timestamp] for timestamp in timestamps] dataset = MarketDataset( key, all_data, @@ -440,8 +437,7 @@ def _build_dataset( source_data=source_data, source_timestamps=source_timestamps, source_price_cache=source_price_cache, - execution_timestamps=execution_timestamps, - execution_price_fields=execution_price_fields, + execution_fills=execution_fills, source_timeframe=actual_source, decision_timeframe=requested_decision, data_quality=data_quality, @@ -459,15 +455,17 @@ def _build_trading_timestamps( min_symbol_coverage: float = 0.8, market: str = DEFAULT_MARKET, timezone: str = DEFAULT_TIMEZONE, - open_stamped_minutes: Optional[int] = None, ) -> List[Any]: """Return in-session timestamps meeting the requested symbol coverage. Session membership comes from ``market_data.sessions`` rather than a literal here, so this filter and the aggregation that produced the bars agree about when the market is open -- including for tz-naive bars, which - both read as market-local time. + both read as market-local time. The stamp convention is the frames' own + (``sessions.frames_open_stamped_minutes``): a raw Alpaca bar is stamped at + its open, an aggregated decision bar or a legacy loader's at its close. """ + open_stamped_minutes = frames_open_stamped_minutes(all_data) all_timestamps: set = set() for df in all_data.values(): all_timestamps.update(df.index) diff --git a/dashboard/backend/domain/leaderboard/baselines.py b/dashboard/backend/domain/leaderboard/baselines.py index 09d44f0d..a7bcae63 100644 --- a/dashboard/backend/domain/leaderboard/baselines.py +++ b/dashboard/backend/domain/leaderboard/baselines.py @@ -19,9 +19,6 @@ import pandas as pd from dashboard.backend.domain.leaderboard.strategies import get_strategy -from dashboard.backend.domain.leaderboard.strategies._common import ( - LEADERBOARD_BAR_TIMEFRAME, -) from dashboard.backend.domain.backtesting.constants import INITIAL_CAPITAL from dashboard.backend.domain.backtesting.metrics import ( calculate_max_drawdown, @@ -45,7 +42,7 @@ def fetch_hourly_bars(symbols: List[str], start_date: str, end_date: str) -> Dic that persist a curve should read it with ``feed_provenance`` and store it, because the log line below outlives nothing. """ - loader = AlpacaDataLoader(source_timeframe=LEADERBOARD_BAR_TIMEFRAME) + loader = AlpacaDataLoader() end_inclusive = ( datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1) ).strftime("%Y-%m-%d") diff --git a/dashboard/backend/domain/leaderboard/strategies/_common.py b/dashboard/backend/domain/leaderboard/strategies/_common.py index 1a91eb6d..4290f0f5 100644 --- a/dashboard/backend/domain/leaderboard/strategies/_common.py +++ b/dashboard/backend/domain/leaderboard/strategies/_common.py @@ -12,19 +12,13 @@ import pandas as pd import pytz -from dashboard.backend.infrastructure.market_data.frequency import timeframe_minutes -from dashboard.backend.infrastructure.market_data.sessions import is_in_session +from dashboard.backend.infrastructure.market_data.sessions import ( + frames_open_stamped_minutes, + is_in_session, +) _ET = pytz.timezone("US/Eastern") -# Every leaderboard strategy runs on the raw Alpaca bars -# ``leaderboard/baselines.fetch_hourly_bars`` requests at this timeframe. -# Alpaca stamps a bar at its open, so the session filter needs the span: under -# the close-stamp rule the board kept 10:00-16:00, i.e. the 16:00-17:00 -# after-hours bar in and the 09:00 bar holding the 09:30 open out. -LEADERBOARD_BAR_TIMEFRAME = "60m" -LEADERBOARD_BAR_OPEN_MINUTES = timeframe_minutes(LEADERBOARD_BAR_TIMEFRAME) - def parse_config_date(date_str: str) -> dt.date: return dt.datetime.strptime(date_str, "%Y-%m-%d").date() @@ -71,26 +65,36 @@ def timestamps_in_reference( return [ts for ts in timestamps if ref_start <= timestamp_date(ts) < contest] -def filter_market_hours(timestamps: List[Any]) -> List[Any]: - """Keep the board's open-stamped hourly bars that close in the 9:30–16:00 - ET session: 09:00 through 15:00.""" +def filter_market_hours( + timestamps: List[Any], *, open_stamped_minutes: Optional[int] = None +) -> List[Any]: + """Keep only regular US market-hours timestamps (9:30–16:00 ET). + + ``open_stamped_minutes`` is the bars' span when they are stamped at their + open, as the board's raw Alpaca hourly bars are: those keep 10:00 through + 15:00, the bars lying wholly inside the session. + """ return [ ts for ts in timestamps if is_in_session( ts, market="US", timezone=_ET.zone, - open_stamped_minutes=LEADERBOARD_BAR_OPEN_MINUTES, + open_stamped_minutes=open_stamped_minutes, ) ] def market_timestamps(bars_subset: Dict[str, pd.DataFrame]) -> List[Any]: - """Sorted, market-hours-only union of timestamps across the given symbols.""" + """Sorted, market-hours-only union of timestamps across the given symbols, + filtered under the stamp convention the bars' loader recorded.""" all_ts = set() for df in bars_subset.values(): all_ts.update(df.index) - return filter_market_hours(sorted(all_ts)) + return filter_market_hours( + sorted(all_ts), + open_stamped_minutes=frames_open_stamped_minutes(bars_subset), + ) def build_price_cache( diff --git a/dashboard/backend/domain/leaderboard/strategies/buy_hold.py b/dashboard/backend/domain/leaderboard/strategies/buy_hold.py index 5ba168c9..b406bad4 100644 --- a/dashboard/backend/domain/leaderboard/strategies/buy_hold.py +++ b/dashboard/backend/domain/leaderboard/strategies/buy_hold.py @@ -12,7 +12,7 @@ from dashboard.backend.baseline_generator import BaselineGenerator from .base import BaselineStrategy -from ._common import LEADERBOARD_BAR_OPEN_MINUTES, subset_bars +from ._common import subset_bars class BuyHoldStrategy(BaselineStrategy): @@ -33,8 +33,7 @@ def run( if not bars_subset: return [] return BaselineGenerator().generate_buyhold_baseline( - bars_subset, start_date, end_date, initial_capital, symbols, - open_stamped_minutes=LEADERBOARD_BAR_OPEN_MINUTES, + bars_subset, start_date, end_date, initial_capital, symbols ) def num_trades(self) -> int: diff --git a/dashboard/backend/domain/leaderboard/strategies/equal_weight_buyhold.py b/dashboard/backend/domain/leaderboard/strategies/equal_weight_buyhold.py index 4b443bf8..3476ce98 100644 --- a/dashboard/backend/domain/leaderboard/strategies/equal_weight_buyhold.py +++ b/dashboard/backend/domain/leaderboard/strategies/equal_weight_buyhold.py @@ -14,7 +14,7 @@ from dashboard.backend.infrastructure.llm.validator import DJIA_30 from .base import BaselineStrategy -from ._common import LEADERBOARD_BAR_OPEN_MINUTES, subset_bars +from ._common import subset_bars class EqualWeightBuyHoldStrategy(BaselineStrategy): @@ -36,8 +36,7 @@ def run( if not bars_subset: return [] return BaselineGenerator().generate_buyhold_baseline( - bars_subset, start_date, end_date, initial_capital, symbols, - open_stamped_minutes=LEADERBOARD_BAR_OPEN_MINUTES, + bars_subset, start_date, end_date, initial_capital, symbols ) def num_trades(self) -> int: diff --git a/dashboard/backend/domain/leaderboard/strategies/equal_weight_index.py b/dashboard/backend/domain/leaderboard/strategies/equal_weight_index.py index 7018b185..f892449f 100644 --- a/dashboard/backend/domain/leaderboard/strategies/equal_weight_index.py +++ b/dashboard/backend/domain/leaderboard/strategies/equal_weight_index.py @@ -15,7 +15,6 @@ from dashboard.backend.infrastructure.llm.validator import DJIA_30 from .base import BaselineStrategy -from ._common import LEADERBOARD_BAR_OPEN_MINUTES class EqualWeightIndexStrategy(BaselineStrategy): @@ -34,6 +33,5 @@ def run( ) -> List[Dict[str, Any]]: symbols = self.required_symbols() return BaselineGenerator().generate_index_baseline( - bars_by_symbol, start_date, end_date, initial_capital, symbols, - open_stamped_minutes=LEADERBOARD_BAR_OPEN_MINUTES, + bars_by_symbol, start_date, end_date, initial_capital, symbols ) diff --git a/dashboard/backend/infrastructure/market_data/alpaca_bars.py b/dashboard/backend/infrastructure/market_data/alpaca_bars.py index fc879647..68a9bcfb 100644 --- a/dashboard/backend/infrastructure/market_data/alpaca_bars.py +++ b/dashboard/backend/infrastructure/market_data/alpaca_bars.py @@ -27,8 +27,12 @@ from dashboard.backend.paths import CREDENTIALS_DIR from dashboard.backend.infrastructure.market_data.frequency import ( normalize_bar_timeframe, + timeframe_minutes, ) from dashboard.backend.infrastructure.market_data import bar_cache +from dashboard.backend.infrastructure.market_data.sessions import ( + FRAME_ATTR_OPEN_STAMPED_MINUTES, +) # Basic plan may query SIP historical bars, but not the most recent window. # Docs: https://docs.alpaca.markets/docs/market-data-faq @@ -532,7 +536,23 @@ def fetch_bars( timeframe, same feed) finds five of its thirty names on disk. A per-request key would miss that entirely, because the symbol lists differ. + + Every returned frame is stamped with + ``sessions.FRAME_ATTR_OPEN_STAMPED_MINUTES``: Alpaca stamps a bar at its + OPEN, and the session filters downstream read that off the frame rather + than off whoever holds it. Stamped here, above the cache, so a hit + carries it exactly as a live fetch does. """ + frames = self._fetch_bars_resolved(symbols, start, end) + span = timeframe_minutes(self.source_timeframe) + for frame in frames.values(): + frame.attrs[FRAME_ATTR_OPEN_STAMPED_MINUTES] = span + return frames + + def _fetch_bars_resolved( + self, symbols: List[str], start: str, end: str + ) -> Dict[str, pd.DataFrame]: + """:meth:`fetch_bars` before stamping: cache hits plus a live fetch.""" symbols = list(symbols) # Hoisted above the batch recursion. With no client every chunk # returned {} anyway, so the result is identical; the warning now diff --git a/dashboard/backend/infrastructure/market_data/frequency.py b/dashboard/backend/infrastructure/market_data/frequency.py index 88380a16..6f330aaf 100644 --- a/dashboard/backend/infrastructure/market_data/frequency.py +++ b/dashboard/backend/infrastructure/market_data/frequency.py @@ -207,8 +207,10 @@ def build_verified_intraday_contract( "aggregation": "session_anchored_completed_bars", "fill_policy": "next_source_bar_open", # The session's last decision has no next source bar inside the - # session, so it fills at the last one's close instead of on an - # after-hours bar (``bar_aggregation.plan_execution_fills``). + # session, so it fills at the close of the bar ending then instead + # of on an after-hours bar (``bar_aggregation.plan_execution_fills``). + # True of every contract built here: aggregation accepts only + # open-stamped source bars, the one kind that planner serves. "session_close_fill": "last_source_bar_close", "verification_status": "verified", } diff --git a/dashboard/backend/infrastructure/market_data/profiles.py b/dashboard/backend/infrastructure/market_data/profiles.py index 2f85f2dc..72d2c667 100644 --- a/dashboard/backend/infrastructure/market_data/profiles.py +++ b/dashboard/backend/infrastructure/market_data/profiles.py @@ -281,18 +281,6 @@ def llm_enabled(self) -> bool: } -def bars_open_stamped(data_source: object) -> bool: - """Whether this source stamps a raw bar at its OPEN. - - Alpaca does (10:00 is 10:00-11:00) and serves extended hours with it; - iFinD stamps at the close (10:30 is 09:30-10:30), and the vnpy simulation - emits in-session closes. The session filter must know which, since one - inclusive rule keeps Alpaca's after-hours 16:00 bar -- see - ``sessions.is_in_session(open_stamped_minutes=...)``. - """ - return str(data_source or ALPACA).strip().lower() == ALPACA - - def registered_market_timezones() -> dict[str, str]: """``{market: timezone}`` across the registered profiles. diff --git a/dashboard/backend/infrastructure/market_data/sessions.py b/dashboard/backend/infrastructure/market_data/sessions.py index 5dcde07a..41d0a8fe 100644 --- a/dashboard/backend/infrastructure/market_data/sessions.py +++ b/dashboard/backend/infrastructure/market_data/sessions.py @@ -16,6 +16,7 @@ from __future__ import annotations from datetime import datetime, time, timedelta +from typing import Any, Mapping import pytz @@ -78,11 +79,41 @@ def time_in_session(local_time: time, market: object) -> bool: return any(start <= minute <= end for start, end in session_windows(market)) -def _market_local(timestamp: datetime, timezone: str) -> datetime: - """A naive timestamp is read as market-local time, matching - ``bar_aggregation._as_local_index``; ``astimezone`` on a naive pandas - ``Timestamp`` raises instead, which made the filter and the aggregation - disagree on exactly the data a local-time feed returns.""" +#: ``DataFrame.attrs`` key a loader sets on every frame whose bars it stamps at +#: their OPEN, holding the bar span in minutes. Absent means stamped at the +#: close -- an aggregated decision bar, an iFinD bar, a legacy double. It lives +#: on the frame because the convention is a fact about the data, not about +#: whoever is filtering it: passing it by hand let the protocol baseline worker, +#: which builds its backtester without loading anything, filter aggregated +#: close-stamped bars under the raw-bar rule and drop every 16:00 close. +FRAME_ATTR_OPEN_STAMPED_MINUTES = "bar_open_stamped_minutes" + + +def frames_open_stamped_minutes(frames: Mapping[str, Any]) -> int | None: + """The span the loader stamped on ``frames``, or ``None`` if their bars are + stamped at the close. Empty frames carry no bars and are ignored. + + Raises ``ValueError`` on a mix: no one rule filters both conventions, and + choosing one would silently misfilter the other half of the universe. + """ + spans = { + frame.attrs.get(FRAME_ATTR_OPEN_STAMPED_MINUTES) + for frame in frames.values() + if len(frame) + } + if len(spans) > 1: + raise ValueError( + f"bars mix stamp conventions (open-stamped spans {sorted(spans, key=str)})" + ) + return spans.pop() if spans else None + + +def market_local(timestamp: datetime, timezone: str) -> datetime: + """``timestamp`` in the market's zone. A naive timestamp is read as + market-local time, matching ``bar_aggregation._as_local_index``; + ``astimezone`` on a naive pandas ``Timestamp`` raises instead, which made + the filter and the aggregation disagree on exactly the data a local-time + feed returns.""" zone = pytz.timezone(timezone) if timestamp.tzinfo is None: return zone.localize(timestamp) @@ -99,21 +130,27 @@ def is_in_session( """:func:`time_in_session` for a timestamp in any zone. ``open_stamped_minutes`` says the timestamp is a bar's OPEN and the bar - spans that many minutes; the bar is then in session when it CLOSES inside - one (``start < close <= end``). Alpaca stamps every bar at its open and - returns extended hours, so the close-stamp rule kept its 16:00 bar -- the - 16:00-16:05 (or 16:00-17:00) after-hours bar -- and, for clock-aligned - hourly bars, dropped the 09:00 bar that holds the 09:30 open. + spans that many minutes (read it off the frames with + :func:`frames_open_stamped_minutes`). Such a bar is in session only when it + lies wholly inside one: it opens at or after the session starts and closes + at or before it ends. Alpaca stamps bars at their open and serves extended + hours, so the close-stamp rule kept its 16:00 bar -- 16:00-16:05, or + 16:00-17:00, after hours. Clock-aligned hourly bars straddle the 09:30 + open, and that 09:00 bar is out too: its open, high, low and volume include + half an hour of pre-market trading, so admitting it would price the day's + first fill off a pre-market print. """ - local = _market_local(timestamp, timezone) + local = market_local(timestamp, timezone) if open_stamped_minutes is None: return time_in_session(local.time(), market) close = local + timedelta(minutes=open_stamped_minutes) if close.date() != local.date(): return False + open_minute = local.time().replace(second=0, microsecond=0) close_minute = close.time().replace(second=0, microsecond=0) return any( - start < close_minute <= end for start, end in session_windows(market) + start <= open_minute and close_minute <= end + for start, end in session_windows(market) ) @@ -121,7 +158,7 @@ def is_session_close(timestamp: datetime, *, market: object, timezone: str) -> b """Whether ``timestamp`` falls in the minute a session ends (16:00 ET, 11:30 and 15:00 CST): the stamp of a final bucket, which no in-session source bar opens at.""" - minute = _market_local(timestamp, timezone).time().replace( + minute = market_local(timestamp, timezone).time().replace( second=0, microsecond=0 ) return any(minute == end for _start, end in session_windows(market)) diff --git a/dashboard/backend/tests/backtesting/test_bar_aggregation.py b/dashboard/backend/tests/backtesting/test_bar_aggregation.py index 2f7412f2..55257074 100644 --- a/dashboard/backend/tests/backtesting/test_bar_aggregation.py +++ b/dashboard/backend/tests/backtesting/test_bar_aggregation.py @@ -1,17 +1,23 @@ from datetime import datetime import pandas as pd +import pytest import pytz from dashboard.backend.domain.backtesting.bar_aggregation import ( + BarAggregationError, aggregate_bars, summarize_aggregation_quality, ) +from dashboard.backend.infrastructure.market_data.sessions import ( + FRAME_ATTR_OPEN_STAMPED_MINUTES, +) def _bars(timestamps): + """5m bars stamped at their open, as Alpaca's loader returns them.""" prices = list(range(100, 100 + len(timestamps))) - return pd.DataFrame( + frame = pd.DataFrame( { "open": prices, "high": [price + 1 for price in prices], @@ -22,6 +28,8 @@ def _bars(timestamps): }, index=pd.DatetimeIndex(timestamps), ) + frame.attrs[FRAME_ATTR_OPEN_STAMPED_MINUTES] = 5 + return frame def test_us_bars_are_anchored_to_0930_and_labeled_at_bucket_end(): @@ -207,3 +215,39 @@ def test_quality_summary_counts_usable_and_rejected_buckets(): assert summary["dropped_decision_bars"] == 1 assert summary["missing_source_bars"] == 1 assert summary["symbols"]["AAPL"]["dropped_decision_bars"] == 1 + + +def test_aggregation_refuses_bars_not_stamped_at_their_open(): + """Bucketing reads a source stamp as its bar's open; a close-stamped (or + unlabelled) frame would land every bar one bucket late, silently.""" + eastern = pytz.timezone("US/Eastern") + timestamps = pd.date_range( + eastern.localize(datetime(2026, 3, 2, 9, 30)), + eastern.localize(datetime(2026, 3, 2, 10, 25)), + freq="5min", + ) + unlabelled = _bars(timestamps) + unlabelled.attrs.pop(FRAME_ATTR_OPEN_STAMPED_MINUTES) + wrong_span = _bars(timestamps) + wrong_span.attrs[FRAME_ATTR_OPEN_STAMPED_MINUTES] = 1 + for frame in (unlabelled, wrong_span): + with pytest.raises(BarAggregationError, match="stamped at their open"): + aggregate_bars(frame, source_timeframe="5m", decision_timeframe="60m") + + +def test_aggregated_bars_are_stamped_at_their_close(): + """The source's open-stamp label must not ride along into the decision + bars: every session filter downstream would read 10:30 as 10:30-11:30.""" + eastern = pytz.timezone("US/Eastern") + timestamps = pd.date_range( + eastern.localize(datetime(2026, 3, 2, 9, 30)), + eastern.localize(datetime(2026, 3, 2, 10, 25)), + freq="5min", + ) + source = _bars(timestamps) + source.attrs["alpaca_feed"] = "sip" + + result = aggregate_bars(source, source_timeframe="5m", decision_timeframe="60m") + + assert FRAME_ATTR_OPEN_STAMPED_MINUTES not in result.attrs + assert result.attrs["alpaca_feed"] == "sip" diff --git a/dashboard/backend/tests/backtesting/test_engine_minute_source.py b/dashboard/backend/tests/backtesting/test_engine_minute_source.py index 8678c592..ff740efd 100644 --- a/dashboard/backend/tests/backtesting/test_engine_minute_source.py +++ b/dashboard/backend/tests/backtesting/test_engine_minute_source.py @@ -12,6 +12,9 @@ FRAME_ATTR_SIP_FALLBACK, MarketDataUnavailableError, ) +from dashboard.backend.infrastructure.market_data.sessions import ( + FRAME_ATTR_OPEN_STAMPED_MINUTES, +) class _MinuteLoader: @@ -72,6 +75,7 @@ def _make_minute_bars(): index=pd.DatetimeIndex(timestamps), ) frame.attrs[FRAME_ATTR_FEED] = "sip" + frame.attrs[FRAME_ATTR_OPEN_STAMPED_MINUTES] = 5 frame.attrs[FRAME_ATTR_SIP_FALLBACK] = False frame.attrs[FRAME_ATTR_END_CLAMPED] = True return {"AAPL": frame} diff --git a/dashboard/backend/tests/backtesting/test_engine_move.py b/dashboard/backend/tests/backtesting/test_engine_move.py index 5f8a04aa..753e546a 100644 --- a/dashboard/backend/tests/backtesting/test_engine_move.py +++ b/dashboard/backend/tests/backtesting/test_engine_move.py @@ -29,6 +29,9 @@ ) from dashboard.backend.infrastructure.ai_hedge_fund.adapter import AiHedgeFundRuntime from dashboard.scripts import backtest_hourly_agent as bha +from dashboard.backend.infrastructure.market_data.sessions import ( + FRAME_ATTR_OPEN_STAMPED_MINUTES, +) _REPO_ROOT = Path(__file__).resolve().parents[4] ENGINE_MODULE = "dashboard.backend.domain.backtesting.engine" @@ -443,6 +446,7 @@ def test_djia_baseline_drops_incomplete_decision_bars(monkeypatch): }, index=timestamps, ) + source_frame.attrs[FRAME_ATTR_OPEN_STAMPED_MINUTES] = 5 _FakeLoader.bars = {symbol: source_frame for symbol in engine_mod.DJIA_30} captured = {} diff --git a/dashboard/backend/tests/test_backtests_router.py b/dashboard/backend/tests/test_backtests_router.py index 139c6001..4970a958 100644 --- a/dashboard/backend/tests/test_backtests_router.py +++ b/dashboard/backend/tests/test_backtests_router.py @@ -625,6 +625,27 @@ def test_run_metadata_response_exposes_minute_data_contract_and_quality(): assert response.end_clamped is True +def test_run_metadata_response_passes_every_field_the_contract_builder_writes(): + """Pinned against the producer, not a hand-copied dict: the allow-list + once dropped ``session_close_fill``, and the details label that reads it + never rendered while a formatter-only test stayed green.""" + from dashboard.backend.infrastructure.market_data.frequency import ( + build_verified_intraday_contract, + ) + + contract = build_verified_intraday_contract( + source_timeframe="5m", + decision_timeframe="60m", + decision_frequency="1h", + ) + response = bt._run_metadata_response( + _run_record({"data_source": "alpaca", "frequency_contract": contract}) + ) + + assert response.frequency_contract == contract + assert response.frequency_contract["session_close_fill"] == "last_source_bar_close" + + def test_run_metadata_response_exposes_sanitized_llm_execution_evidence(): response = bt._run_metadata_response( _run_record({ diff --git a/dashboard/backend/tests/test_external_minute_source.py b/dashboard/backend/tests/test_external_minute_source.py index f125f45c..8cdf61a2 100644 --- a/dashboard/backend/tests/test_external_minute_source.py +++ b/dashboard/backend/tests/test_external_minute_source.py @@ -5,11 +5,15 @@ import dashboard.backend.domain.backtesting.external_run_service as ebs from dashboard.backend.domain.backtesting import market_data_store as mds +from dashboard.backend.domain.backtesting.bar_aggregation import ExecutionFill from dashboard.backend.infrastructure.market_data.alpaca_bars import ( FRAME_ATTR_END_CLAMPED, FRAME_ATTR_FEED, FRAME_ATTR_SIP_FALLBACK, ) +from dashboard.backend.infrastructure.market_data.sessions import ( + FRAME_ATTR_OPEN_STAMPED_MINUTES, +) def _minute_bars(symbols, start, end): @@ -34,6 +38,7 @@ def _minute_bars(symbols, start, end): symbol_frame.attrs[FRAME_ATTR_FEED] = "iex" symbol_frame.attrs[FRAME_ATTR_SIP_FALLBACK] = True symbol_frame.attrs[FRAME_ATTR_END_CLAMPED] = False + symbol_frame.attrs[FRAME_ATTR_OPEN_STAMPED_MINUTES] = 5 return frames @@ -70,7 +75,7 @@ def test_external_session_serves_hourly_bars_but_fills_and_values_on_5m(): assert session.source_timeframe == "5m" assert session.intraday_mode is True - # Seven: the 16:00 bucket fills at the 15:55 bar's close rather than + # Seven: the 16:00 bucket fills at the close of the 15:55 bar rather than # needing an after-hours bar to open at 16:00. assert session.total_steps == 7 assert session.data_quality["total_decision_bars"] == 7 @@ -149,8 +154,14 @@ def fetch_bars(self, symbols, start, end): last_regular = pd.Timestamp("2026-04-15 19:55:00+00:00") assert after_hours not in session.source_timestamps assert session.total_steps == 7 - assert session.execution_timestamps[-1] == last_regular - assert session.execution_price_fields == ["open"] * 6 + ["close"] + # Priced off the 15:55 bar's close, and stamped at that close (16:00), so + # the fill never predates the decision it answers. + assert session.execution_fills[-1] == ExecutionFill( + last_regular, "close", after_hours + ) + assert [fill.price_field for fill in session.execution_fills] == ( + ["open"] * 6 + ["close"] + ) for _ in range(session.total_steps - 1): session.submit_decisions({"actions": []}) @@ -169,7 +180,11 @@ def fetch_bars(self, symbols, start, end): ) trade = session.manager.trades[-1] - assert trade["timestamp"] == last_regular + assert trade["timestamp"] == after_hours + assert trade["timestamp"] >= session.timestamps[-1] + assert session.get_decisions()[-1]["execution_timestamp"] == ( + "2026-04-15T20:00:00+00:00" + ) last_close = session.source_data["AAPL"].loc[last_regular, "close"] assert trade["price"] == pytest.approx(last_close) assert trade["price"] != pytest.approx(after_hours_open) @@ -200,11 +215,11 @@ def fetch_bars(self, symbols, start, end): ) session.load_market_data() assert session.timestamps[0] == pd.Timestamp("2026-04-15 14:30:00+00:00") - assert session.execution_timestamps[0] == pd.Timestamp( + assert session.execution_fills[0].bar == pd.Timestamp( "2026-04-15 14:30:00+00:00" ) assert "AAPL" not in session._source_market_data_at( - session.execution_timestamps[0] + session.execution_fills[0].bar ) result = session.submit_decisions( @@ -295,13 +310,60 @@ def test_the_engine_plans_the_same_closing_fill(): pd.date_range("2026-04-15 14:30", "2026-04-15 19:30", freq="h", tz="UTC") ) + [after_hours] # 10:30 ... 15:30 and the 16:00 ET closing bucket - kept, valuation, plan, fields = backtester._plan_executions(decisions) + kept, valuation, plan = backtester._plan_executions(decisions) last_regular = pd.Timestamp("2026-04-15 19:55:00+00:00") assert kept == decisions assert after_hours not in valuation and valuation[-1] == last_regular - assert plan[after_hours] == last_regular and fields[after_hours] == "close" - assert all(fields[ts] == "open" and plan[ts] == ts for ts in decisions[:-1]) - # ...and the run loop prices the fill by that field, not a literal "open". + assert plan[after_hours] == ExecutionFill(last_regular, "close", after_hours) + assert all(plan[ts] == ExecutionFill(ts, "open", ts) for ts in decisions[:-1]) + # ...and the run loop prices by the fill's field and stamps the trade at + # its fill instant, not a literal "open" on the bar's own stamp. source = Path(engine_module.__file__).read_text(encoding="utf-8") - assert "symbol: row[execution_field]" in source + assert "symbol: row[fill.price_field]" in source + assert " fill.filled_at,\n" in source + + +def test_protocol_baselines_keep_each_session_close(monkeypatch): + """``baseline_worker._run_job`` builds an ``HourlyBacktester`` it never + loads -- ``intraday_mode`` stays False -- and hands it the dataset's + aggregated bars. Those are stamped at their close, so the 16:00 bar is the + day's closing mark. Deciding the convention from ``intraday_mode`` read + them as raw 5m bars and dropped it; the frames now say which they are.""" + from dashboard.backend.domain.backtesting import engine as engine_module + + session = ebs.ExternalBacktestSession( + backtest_id="bt-baseline", + session_id="sess-baseline", + agent_name="agent-baseline", + model_name="test-model", + start_date="2026-04-15", + end_date="2026-04-15", + symbols=["AAPL"], + ) + session.load_market_data() + + class _DB: + def insert_run(self, **kwargs): + pass + + def insert_equity_points(self, run_id, points): + pass + + monkeypatch.setattr(engine_module, "db", _DB()) + monkeypatch.setattr( + engine_module, + "create_market_data_provider", + lambda *args, **kwargs: _MinuteLoader(), + ) + # Exactly what `_run_job` does with a finalized run's dataset. + backtester = engine_module.HourlyBacktester( + session.start_date, session.end_date, session.session_id, use_llm=False + ) + backtester.all_data = session.all_data + assert backtester.intraday_mode is False + _run_id, history = backtester.run_buyhold_baseline() + + assert history, "the baseline must produce a curve" + closing = pd.Timestamp("2026-04-15 20:00:00+00:00") # 16:00 ET + assert pd.Timestamp(history[-1]["timestamp"]) == closing diff --git a/dashboard/backend/tests/test_market_data_store.py b/dashboard/backend/tests/test_market_data_store.py index 5b8b397d..122f95ed 100644 --- a/dashboard/backend/tests/test_market_data_store.py +++ b/dashboard/backend/tests/test_market_data_store.py @@ -11,6 +11,9 @@ from dashboard.backend.domain.backtesting import market_data_store as mds from dashboard.backend.infrastructure.market_data.frequency import FrequencyConfigError +from dashboard.backend.infrastructure.market_data.sessions import ( + FRAME_ATTR_OPEN_STAMPED_MINUTES, +) def _synth_bars(symbols=("AAPL", "MSFT"), start="2026-04-15", end="2026-04-16"): @@ -480,6 +483,7 @@ def test_minute_dataset_exposes_dropped_bucket_quality(): index=timestamps, ) } + bars["AAPL"].attrs[FRAME_ATTR_OPEN_STAMPED_MINUTES] = 5 class _MinuteLoader: source_timeframe = "60m" diff --git a/dashboard/backend/tests/test_market_sessions.py b/dashboard/backend/tests/test_market_sessions.py index ed14669c..09c12e3b 100644 --- a/dashboard/backend/tests/test_market_sessions.py +++ b/dashboard/backend/tests/test_market_sessions.py @@ -7,6 +7,7 @@ """ import ast +import functools from datetime import time as clock_time from pathlib import Path @@ -96,8 +97,10 @@ def test_every_us_filter_agrees(): from dashboard.backend.domain.leaderboard.strategies import _common stamps = _sample_timestamps() - frame = pd.DataFrame({"close": 1.0}, index=pd.DatetimeIndex(stamps)) for open_minutes in (None, 60): + frame = pd.DataFrame({"close": 1.0}, index=pd.DatetimeIndex(stamps)) + if open_minutes is not None: + frame.attrs[sessions.FRAME_ATTR_OPEN_STAMPED_MINUTES] = open_minutes expected = [ ts for ts in stamps if sessions.is_in_session( @@ -108,17 +111,17 @@ def test_every_us_filter_agrees(): ) ] assert expected, "the sample must straddle the session" + # Each reads the convention off the frame, as its callers hand it one. assert baseline_generator._market_hours_only( - stamps, "US/Eastern", open_minutes + stamps, "US/Eastern", {"X": frame} ) == expected assert mds._build_trading_timestamps( - {"X": frame}, - market="US", - timezone="US/Eastern", - open_stamped_minutes=open_minutes, + {"X": frame}, market="US", timezone="US/Eastern" + ) == expected + assert _common.market_timestamps({"X": frame}) == expected + assert _common.filter_market_hours( + stamps, open_stamped_minutes=open_minutes ) == expected - if open_minutes == _common.LEADERBOARD_BAR_OPEN_MINUTES: - assert _common.filter_market_hours(stamps) == expected # The 16:00:30 stamp is in for a close-stamped bar: the store's own first # rewrite excluded it while the engine kept it. assert sessions.is_in_session( @@ -140,14 +143,15 @@ def _et(clock): ("09:30", 5, True), ("15:55", 5, True), ("16:00", 5, False), # 16:00-16:05 is after hours - # Alpaca 1h, clock-aligned: 09:00 (closes 10:00) through 15:00. + # Alpaca 1h, clock-aligned: 10:00 through 15:00. ("08:00", 60, False), - ("09:00", 60, True), + ("09:00", 60, False), # 09:00-10:00 holds 30 minutes of pre-market + ("10:00", 60, True), ("15:00", 60, True), ("16:00", 60, False), # the bar the close-stamp rule used to keep ], ) -def test_an_open_stamped_bar_is_in_session_when_it_closes_in_one( +def test_an_open_stamped_bar_is_in_session_only_when_wholly_inside_one( clock, minutes, kept ): assert sessions.is_in_session( @@ -158,9 +162,10 @@ def test_an_open_stamped_bar_is_in_session_when_it_closes_in_one( ) is kept -def test_open_stamped_hourly_day_is_seven_bars(): - """The raw 1h path keeps seven bars a day, as the close-stamp rule did -- - but 09:00-15:00 rather than 10:00-16:00.""" +def test_open_stamped_hourly_day_is_six_bars(): + """The raw 1h path keeps 10:00-15:00. The close-stamp rule kept 10:00-16:00, + the last of which is after hours; 09:00 straddles the open, so its prices + are part pre-market and it cannot stand in for the 09:30-10:00 half hour.""" day = pd.date_range("2026-04-15 04:00", "2026-04-15 19:00", freq="h", tz="US/Eastern") kept = [ @@ -169,7 +174,7 @@ def test_open_stamped_hourly_day_is_seven_bars(): ts, market="US", timezone="US/Eastern", open_stamped_minutes=60 ) ] - assert kept == ["09:00", "10:00", "11:00", "12:00", "13:00", "14:00", "15:00"] + assert kept == ["10:00", "11:00", "12:00", "13:00", "14:00", "15:00"] def test_session_close_is_each_window_end(): @@ -185,10 +190,41 @@ def test_session_close_is_each_window_end(): ) is is_close -def test_only_alpaca_stamps_bars_at_their_open(): - assert profiles.bars_open_stamped(profiles.ALPACA) - assert not profiles.bars_open_stamped(profiles.IFIND_ASHARE) - assert not profiles.bars_open_stamped(profiles.VNPY_SIMULATION) +def test_the_alpaca_loader_stamps_every_frame_it_returns(monkeypatch): + """Stamped above the bar cache, so a hit carries the convention too.""" + from dashboard.backend.infrastructure.market_data.alpaca_bars import ( + AlpacaDataLoader, + ) + + loader = AlpacaDataLoader.__new__(AlpacaDataLoader) + for timeframe, span in (("5m", 5), ("60m", 60)): + loader.source_timeframe = timeframe + monkeypatch.setattr( + loader, + "_fetch_bars_resolved", + lambda symbols, start, end: { + symbol: pd.DataFrame({"close": [1.0]}) for symbol in symbols + }, + ) + frames = loader.fetch_bars(["AAPL", "MSFT"], "2026-04-15", "2026-04-16") + assert sessions.frames_open_stamped_minutes(frames) == span + + +def test_frames_open_stamped_minutes_refuses_a_mix(): + def frame(span=None, rows=1): + result = pd.DataFrame({"close": [1.0] * rows}) + if span is not None: + result.attrs[sessions.FRAME_ATTR_OPEN_STAMPED_MINUTES] = span + return result + + assert sessions.frames_open_stamped_minutes({}) is None + assert sessions.frames_open_stamped_minutes({"A": frame()}) is None + # An empty frame holds no bars to misfilter, whatever it carries. + assert sessions.frames_open_stamped_minutes( + {"A": frame(5), "B": frame(rows=0)} + ) == 5 + with pytest.raises(ValueError, match="mix stamp conventions"): + sessions.frames_open_stamped_minutes({"A": frame(5), "B": frame()}) def test_the_final_bucket_fills_at_the_last_in_session_close(): @@ -196,20 +232,22 @@ def test_the_final_bucket_fills_at_the_last_in_session_close(): plan_execution_fills, ) + plan = functools.partial( + plan_execution_fills, source_minutes=5, market="US", timezone="US/Eastern" + ) source = [_et("15:25"), _et("15:30"), _et("15:55")] decisions = [_et("15:30"), _et("16:00"), _et("15:45")] - fills = plan_execution_fills( - decisions, source, market="US", timezone="US/Eastern" - ) - assert fills == { - _et("15:30"): (_et("15:30"), "open"), - _et("16:00"): (_et("15:55"), "close"), + assert plan(decisions, source) == { + _et("15:30"): (_et("15:30"), "open", _et("15:30")), + # Priced at the 15:55 bar's close and stamped then: 16:00, never before + # the decision it fills. + _et("16:00"): (_et("15:55"), "close", _et("16:00")), # Not a session close and no bar opens there: no fill. } # A day with no source bars has nothing to fill the close on. - assert plan_execution_fills( - [_et("16:00")], [], market="US", timezone="US/Eastern" - ) == {} + assert plan([_et("16:00")], []) == {} + # Nor does one whose last bar ends before the close: 15:50's close is 15:55. + assert plan([_et("16:00")], [_et("15:45"), _et("15:50")]) == {} def test_an_exact_fill_is_the_source_bar_not_the_equal_decision_stamp(): @@ -221,9 +259,14 @@ def test_an_exact_fill_is_the_source_bar_not_the_equal_decision_stamp(): # but the trade is stamped with the fill bar, so its tz must survive. decision = _et("15:30").tz_convert("UTC") fills = plan_execution_fills( - [decision], [_et("15:30")], market="US", timezone="US/Eastern" + [decision], + [_et("15:30")], + source_minutes=5, + market="US", + timezone="US/Eastern", ) - assert str(fills[decision][0].tz) == "US/Eastern" + assert str(fills[decision].bar.tz) == "US/Eastern" + assert str(fills[decision].filled_at.tz) == "US/Eastern" def test_the_engine_filter_agrees_for_both_markets(): @@ -236,7 +279,8 @@ def test_the_engine_filter_agrees_for_both_markets(): ): engine = HourlyBacktester.__new__(HourlyBacktester) engine.profile = profiles.get_market_profile(data_source) - assert engine._market_hours_only(stamps) == [ + close_stamped = {"X": pd.DataFrame({"close": 1.0}, index=pd.DatetimeIndex(stamps))} + assert engine._market_hours_only(stamps, close_stamped) == [ ts for ts in stamps if sessions.is_in_session(ts, market=market, timezone=zone) ] @@ -319,7 +363,7 @@ def test_no_module_restates_the_session_bounds(): ) def test_leaderboard_baselines_read_the_board_bars_as_open_stamped(strategy_key): """These three hand the board's raw Alpaca hourly bars to the baseline - generator, which cannot tell the stamp convention from the frame.""" + generator, which reads the stamp convention the loader left on them.""" from dashboard.backend.domain.leaderboard.strategies import get_strategy hours = pd.date_range("2026-04-15 08:00", "2026-04-15 18:00", freq="h", @@ -334,6 +378,7 @@ def test_leaderboard_baselines_read_the_board_bars_as_open_stamped(strategy_key) }, index=hours.tz_convert("UTC"), ) + frame.attrs[sessions.FRAME_ATTR_OPEN_STAMPED_MINUTES] = 60 strategy = get_strategy({"strategy": strategy_key, "symbols": ["AAPL", "MSFT"]}) curve = strategy.run( {"AAPL": frame, "MSFT": frame.copy()}, "2026-04-15", "2026-04-15", 10_000.0 @@ -342,4 +387,4 @@ def test_leaderboard_baselines_read_the_board_bars_as_open_stamped(strategy_key) pd.Timestamp(point["timestamp"]).tz_convert("US/Eastern").strftime("%H:%M") for point in curve ] - assert clocks[0] == "09:00" and clocks[-1] == "15:00" + assert clocks[0] == "10:00" and clocks[-1] == "15:00" diff --git a/dashboard/backend/tests/test_minute_data_frontend.py b/dashboard/backend/tests/test_minute_data_frontend.py index 4ace1c74..49604f19 100644 --- a/dashboard/backend/tests/test_minute_data_frontend.py +++ b/dashboard/backend/tests/test_minute_data_frontend.py @@ -95,6 +95,19 @@ def test_minute_frequency_formatter_names_the_session_close_fill(): ) +def test_session_close_suffix_only_qualifies_the_next_open_policy(): + value = _run_formatters( + "formatBacktestFrequencyContract({" + "source_timeframe:'5m',decision_timeframe:'60m'," + "decision_frequency:'1h',execution_timeframe:'5m'," + "valuation_frequency:'5m',fill_policy:'decision_bar_close'," + "session_close_fill:'last_source_bar_close'" + "})" + ) + + assert value == "5m source · 1h decisions · 5m execution · 5m valuation" + + def test_quality_formatter_reports_dropped_and_problem_counts(): value = _run_formatters( "formatBacktestMarketDataQuality({" diff --git a/dashboard/frontend/app.js b/dashboard/frontend/app.js index 1fee9656..9ebac174 100644 --- a/dashboard/frontend/app.js +++ b/dashboard/frontend/app.js @@ -10089,11 +10089,12 @@ function formatBacktestFrequencyContract(contract) { const execution = contract.execution_timeframe; const valuation = contract.valuation_frequency; if (!source || !decision || !execution || !valuation) return null; - let fill = contract.fill_policy === 'next_source_bar_open' - ? `next ${execution} open fills` - : `${execution} execution`; - if (contract.session_close_fill === 'last_source_bar_close') { - fill += ' (last close at session end)'; + let fill = `${execution} execution`; + if (contract.fill_policy === 'next_source_bar_open') { + fill = `next ${execution} open fills`; + if (contract.session_close_fill === 'last_source_bar_close') { + fill += ' (last close at session end)'; + } } const verification = contract.verification_status === 'verified' ? ' · verified' diff --git a/docs/minute-data-hourly-trading.md b/docs/minute-data-hourly-trading.md index 3b769229..222902b6 100644 --- a/docs/minute-data-hourly-trading.md +++ b/docs/minute-data-hourly-trading.md @@ -46,8 +46,8 @@ Phase 3 将分钟链路的数据质量和无未来数据约束固化为可测试 Phase 4 保持唯一、固定的成交规则 `next_source_bar_open`，不增加成交策略 配置项。唯一例外是每个交易日收盘时点的决策（如 16:00 ET）：会话内已没有 下一根源 bar，因此以最后一根会话内源 bar（15:55）的收盘价成交，而不是 -盘后 bar 的开盘价；契约中以 `session_close_fill: last_source_bar_close` -记录。回测详情 API 和外部 Agent 的完成结果会返回： +盘后 bar 的开盘价，成交时间记为该 bar 收盘的 16:00，不早于决策本身；契约中以 +`session_close_fill: last_source_bar_close` 记录。回测详情 API 和外部 Agent 的完成结果会返回： - `frequency_contract`：5m 源数据、60m 决策 bar、1h 决策、5m 成交与估值； - `market_data_quality`：聚合后的可用、丢弃及异常计数。 From f120ece324cafc5c465e5bd913ed9d1e0d464a4b Mon Sep 17 00:00:00 2001 From: FlyM1ss Date: Wed, 23 Sep 2026 18:14:33 -0400 Subject: [PATCH 5/5] fix(protocol): record unplanned fills at the decision bar's close A dataset whose decision bars are its source bars has no execution plan, and its default ExecutionFill named each bar's open. The executor ignored that field outside intraday mode and filled at the close, matching the decision_bar_close contract, so no run was mispriced. But the record contradicted both the contract and engine._plan_executions, and one unconditional execution_prices would have turned it into look-ahead. Default to close in MarketDataset and _effective_execution_fills, and pin the unaggregated fill price with a regression test. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../backtesting/external_run_service.py | 7 +- .../domain/backtesting/market_data_store.py | 6 +- .../tests/test_external_minute_source.py | 70 +++++++++++++++++++ 3 files changed, 78 insertions(+), 5 deletions(-) diff --git a/dashboard/backend/domain/backtesting/external_run_service.py b/dashboard/backend/domain/backtesting/external_run_service.py index d384f685..2d6f27cf 100644 --- a/dashboard/backend/domain/backtesting/external_run_service.py +++ b/dashboard/backend/domain/backtesting/external_run_service.py @@ -438,11 +438,12 @@ def _effective_source_price_cache(self) -> Dict[str, Dict[Any, float]]: def _effective_execution_fills(self) -> List[ExecutionFill]: """One ``ExecutionFill`` per step. A dataset with no plan for these - steps fills each on its own bar's open; the bar and its price field - travel together, so a remapped bar can never borrow a default field.""" + steps fills each at its own bar's close (``decision_bar_close``); the + bar and its price field travel together, so a remapped bar can never + borrow a default field.""" if len(self.execution_fills) == self.total_steps: return self.execution_fills - return [ExecutionFill(timestamp, "open", timestamp) for timestamp in self.timestamps] + return [ExecutionFill(timestamp, "close", timestamp) for timestamp in self.timestamps] def _value_through(self, target_timestamp=None) -> None: """Mark the portfolio on each source bar through the given timestamp.""" diff --git a/dashboard/backend/domain/backtesting/market_data_store.py b/dashboard/backend/domain/backtesting/market_data_store.py index c4d81ce6..f2c93f00 100644 --- a/dashboard/backend/domain/backtesting/market_data_store.py +++ b/dashboard/backend/domain/backtesting/market_data_store.py @@ -113,11 +113,13 @@ def __init__(self, key: Tuple, all_data: Dict[str, pd.DataFrame], ) # One ExecutionFill per step. Without a plan -- a dataset whose # decision bars ARE its source bars -- each step fills at its own bar's - # open, as the protocol path always has. + # close (``decision_bar_close``), as ``engine._plan_executions`` does. + # Its open is an hour before the decision, and a field left naming it + # is one unconditional ``execution_prices`` away from look-ahead. self.execution_fills = ( execution_fills if execution_fills is not None - else [ExecutionFill(timestamp, "open", timestamp) for timestamp in timestamps] + else [ExecutionFill(timestamp, "close", timestamp) for timestamp in timestamps] ) self.source_timeframe = source_timeframe self.decision_timeframe = decision_timeframe diff --git a/dashboard/backend/tests/test_external_minute_source.py b/dashboard/backend/tests/test_external_minute_source.py index 8cdf61a2..dea54bf2 100644 --- a/dashboard/backend/tests/test_external_minute_source.py +++ b/dashboard/backend/tests/test_external_minute_source.py @@ -194,6 +194,76 @@ def fetch_bars(self, symbols, start, end): ) +def test_an_unaggregated_run_fills_at_the_decision_bar_close(monkeypatch): + """A run whose source bars already are its decision bars has no execution + plan. Each bar is stamped at its close, so its open is an hour before the + decision and filling there is look-ahead: the contract says + ``decision_bar_close``, and both the recorded fill and the price agree.""" + + class _LegacyHourlyLoader: + # No ``source_timeframe``: the dataset takes it as 60m, unaggregated. + def fetch_bars(self, symbols, start, end): + timestamps = pd.date_range( + "2026-04-15 14:00:00+00:00", + "2026-04-15 20:00:00+00:00", + freq="60min", + ) + closes = [200.0 + index for index in range(len(timestamps))] + frame = pd.DataFrame( + { + "open": [close - 50.0 for close in closes], + "high": [close + 1.0 for close in closes], + "low": [close - 51.0 for close in closes], + "close": closes, + "volume": [1000] * len(closes), + }, + index=timestamps, + ) + return {symbol: frame.copy() for symbol in symbols} + + monkeypatch.setattr(ebs, "AlpacaDataLoader", _LegacyHourlyLoader) + session = ebs.ExternalBacktestSession( + backtest_id="bt-hourly", + session_id="sess-hourly", + agent_name="agent-hourly", + model_name="test-model", + start_date="2026-04-15", + end_date="2026-04-15", + symbols=["AAPL"], + ) + session.load_market_data() + + assert session.intraday_mode is False + assert session.execution_fills == [ + ExecutionFill(timestamp, "close", timestamp) for timestamp in session.timestamps + ] + assert session.frequency_contract["fill_policy"] == "decision_bar_close" + + session.submit_decisions( + { + "actions": [ + { + "symbol": "AAPL", + "action": "buy", + "confidence": 1.0, + "reasoning": "first bar buy", + "position_size": 1, + } + ] + } + ) + + decision_bar = session.timestamps[0] + trade = session.manager.trades[0] + assert trade["timestamp"] == decision_bar + assert trade["price"] == pytest.approx( + session.all_data["AAPL"].loc[decision_bar, "close"] + ) + assert trade["price"] != pytest.approx( + session.all_data["AAPL"].loc[decision_bar, "open"] + ) + + def test_external_session_does_not_report_fill_without_next_symbol_bar(monkeypatch): class _MissingExecutionBarLoader(_MinuteLoader): def fetch_bars(self, symbols, start, end):