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 3d391810..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,17 +39,26 @@ ) from exc -def _market_hours_only(timestamps, market_timezone: str): +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. + 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 - 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, + ) ] @@ -291,7 +301,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, bars_subset + ) all_timestamps = _timestamps_in_window( all_timestamps, start_date, end_date, market_timezone ) @@ -584,7 +596,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, bars_subset + ) all_timestamps = _timestamps_in_window( all_timestamps, start_date, end_date, market_timezone ) diff --git a/dashboard/backend/domain/backtesting/bar_aggregation.py b/dashboard/backend/domain/backtesting/bar_aggregation.py index 73c4d4e7..50ca7e47 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, NamedTuple import numpy as np import pandas as pd @@ -24,7 +25,12 @@ 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 ( + FRAME_ATTR_OPEN_STAMPED_MINUTES, + is_session_close, + market_local, + session_windows, +) class BarAggregationError(ValueError): @@ -108,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) @@ -251,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, @@ -326,3 +346,66 @@ def summarize_aggregation_quality( summary["usable_decision_bars"] += usable summary["dropped_decision_bars"] += total - usable return summary + + +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, 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. + """ + by_day: Dict[date, List[Any]] = {} + for timestamp in source_timestamps: + 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_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. + 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) + ): + # 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 dbb60eac..aa49ffbb 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 @@ -69,7 +68,9 @@ ) 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, ) from dashboard.backend.domain.backtesting.metrics import ( @@ -120,7 +121,10 @@ 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, @@ -1623,33 +1627,62 @@ 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, 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. + 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 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 _plan_executions(self, decision_timestamps): + """``(decisions, valuation bars, {decision: ExecutionFill})`` 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, 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: ExecutionFill(timestamp, "close", timestamp) + for timestamp in decision_timestamps + }, + ) + raw_timestamps = self._market_hours_only( + 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, + fills, ) - return local.date().isoformat() def _run_daily_post_trade( self, @@ -1777,7 +1810,7 @@ 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, self.all_data) prior_market_dates = ( _prior_market_date_by_decision_date(all_timestamps) if self.runtime_type == AI_HEDGE_FUND_RUNTIME_TYPE @@ -1795,40 +1828,9 @@ 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 = self._plan_executions( + all_timestamps + ) print( f" Trading {len(all_timestamps)} hourly decision bars during " @@ -1970,7 +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. - execution_timestamp = execution_plan[timestamp] + # 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] @@ -1985,9 +1990,9 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: execution_timestamp, ) execution_prices = { - symbol: row["open"] + symbol: row[fill.price_field] for symbol, row in execution_market_data.items() - if "open" in row + if fill.price_field in row } # Execute trades (only if real data available) @@ -1995,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, ) diff --git a/dashboard/backend/domain/backtesting/external_run_service.py b/dashboard/backend/domain/backtesting/external_run_service.py index 12eda192..2d6f27cf 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,7 +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_fills: List[ExecutionFill] = [] self.data_quality: Dict[str, Any] = {} self.frequency_contract: Optional[Dict[str, str]] = None self.market_data_provenance: Dict[str, Any] = {} @@ -371,9 +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_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( @@ -437,10 +436,14 @@ 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 _effective_execution_fills(self) -> List[ExecutionFill]: + """One ``ExecutionFill`` per step. A dataset with no plan for these + 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, "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.""" @@ -774,19 +777,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_prices = { - symbol: row["open"] + symbol: row[fill.price_field] for symbol, row in execution_market_data.items() - if "open" 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() @@ -810,9 +814,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 5a5aa58b..f2c93f00 100644 --- a/dashboard/backend/domain/backtesting/market_data_store.py +++ b/dashboard/backend/domain/backtesting/market_data_store.py @@ -28,17 +28,17 @@ 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 ( + ExecutionFill, aggregate_bars_by_symbol, + plan_execution_fills, summarize_aggregation_quality, ) from dashboard.backend.infrastructure.market_data.alpaca_bars import AlpacaDataLoader @@ -52,7 +52,9 @@ verify_source_timeframe, ) 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,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_fills", + "source_timeframe", "decision_timeframe", "data_quality", "equity_metadata", ) @@ -89,7 +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_fills: Optional[List[ExecutionFill]] = None, source_timeframe: str = "60m", decision_timeframe: str = "60m", data_quality: Optional[Dict[str, Any]] = None, @@ -108,10 +111,15 @@ 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) + # One ExecutionFill per step. Without a plan -- a dataset whose + # decision bars ARE its source bars -- each step fills at its own bar's + # 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, "close", timestamp) for timestamp in timestamps] ) self.source_timeframe = source_timeframe self.decision_timeframe = decision_timeframe @@ -133,12 +141,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 +371,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, @@ -395,7 +404,9 @@ def _build_dataset( timezone=timezone, ) timestamps = _build_trading_timestamps( - all_data, market=market, timezone=timezone + all_data, + market=market, + timezone=timezone, ) if not timestamps: raise RuntimeError("No trading hours in the selected date range") @@ -407,23 +418,19 @@ def _build_dataset( timezone=timezone, ) source_price_cache = _build_price_cache(source_data, source_timestamps) - execution_timestamps = _build_execution_timestamps( - timestamps, - source_timestamps, - 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 - ] - price_cache = _build_price_cache(all_data, 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, @@ -432,7 +439,7 @@ def _build_dataset( source_data=source_data, source_timestamps=source_timestamps, source_price_cache=source_price_cache, - execution_timestamps=execution_timestamps, + execution_fills=execution_fills, source_timeframe=actual_source, decision_timeframe=requested_decision, data_quality=data_quality, @@ -444,39 +451,6 @@ 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], *, @@ -489,8 +463,11 @@ def _build_trading_timestamps( 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) @@ -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/strategies/_common.py b/dashboard/backend/domain/leaderboard/strategies/_common.py index ce2cfa70..4290f0f5 100644 --- a/dashboard/backend/domain/leaderboard/strategies/_common.py +++ b/dashboard/backend/domain/leaderboard/strategies/_common.py @@ -12,7 +12,10 @@ import pandas as pd import pytz -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") @@ -62,20 +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 only regular US market-hours timestamps (9:30–16:00 ET).""" +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) + if is_in_session( + ts, + market="US", + timezone=_ET.zone, + 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/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 32989356..6f330aaf 100644 --- a/dashboard/backend/infrastructure/market_data/frequency.py +++ b/dashboard/backend/infrastructure/market_data/frequency.py @@ -206,6 +206,12 @@ 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 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/sessions.py b/dashboard/backend/infrastructure/market_data/sessions.py index dd83b82a..41d0a8fe 100644 --- a/dashboard/backend/infrastructure/market_data/sessions.py +++ b/dashboard/backend/infrastructure/market_data/sessions.py @@ -15,7 +15,8 @@ from __future__ import annotations -from datetime import datetime, time +from datetime import datetime, time, timedelta +from typing import Any, Mapping import pytz @@ -65,8 +66,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 +79,86 @@ 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. +#: ``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" + - 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. +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: - 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 (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) + 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 <= open_minute and 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/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 47a365c7..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} @@ -115,9 +119,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 +134,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/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/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 ) 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_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 c4dcf4d7..dea54bf2 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,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 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 assert session.data_quality["usable_decision_bars"] == 7 assert session.data_quality["dropped_decision_bars"] == 0 @@ -110,6 +117,153 @@ 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 + # 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": []}) + 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"] == 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) + assert all( + pd.Timestamp(point["timestamp"]) != after_hours + for point in session.manager.equity_history + ) + + +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): @@ -131,11 +285,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( @@ -201,3 +355,85 @@ 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 = 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] == 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[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_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_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 af832d18..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 @@ -89,26 +90,183 @@ 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") + 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( + ts, + market="US", + timezone="US/Eastern", + open_stamped_minutes=open_minutes, + ) + ] + 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", {"X": frame} + ) == expected + assert mds._build_trading_timestamps( + {"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 + # 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: 10:00 through 15:00. + ("08:00", 60, False), + ("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_only_when_wholly_inside_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_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 = [ + ts.strftime("%H:%M") for ts in day + if sessions.is_in_session( + ts, market="US", timezone="US/Eastern", open_stamped_minutes=60 + ) ] - 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 + assert kept == ["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_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(): + from dashboard.backend.domain.backtesting.bar_aggregation import ( + 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")] + 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([_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(): + 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")], + source_minutes=5, + market="US", + timezone="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(): @@ -121,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) ] @@ -197,3 +356,35 @@ 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 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", + 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"), + ) + 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 + ) + clocks = [ + pd.Timestamp(point["timestamp"]).tz_convert("US/Eastern").strftime("%H:%M") + for point in curve + ] + 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 bdf665c5..49604f19 100644 --- a/dashboard/backend/tests/test_minute_data_frontend.py +++ b/dashboard/backend/tests/test_minute_data_frontend.py @@ -79,6 +79,35 @@ 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_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.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..9ebac174 100644 --- a/dashboard/frontend/app.js +++ b/dashboard/frontend/app.js @@ -10089,9 +10089,13 @@ 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' - ? `next ${execution} open fills` - : `${execution} execution`; + 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 25e424b4..222902b6 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 的开盘价，成交时间记为该 bar 收盘的 16:00，不早于决策本身；契约中以 +`session_close_fill: last_source_bar_close` 记录。回测详情 API 和外部 Agent 的完成结果会返回： - `frequency_contract`：5m 源数据、60m 决策 bar、1h 决策、5m 成交与估值； - `market_data_quality`：聚合后的可用、丢弃及异常计数。