fix(sessions): read Alpaca bars as open-stamped

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]

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
         )

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

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,17 +1990,17 @@ 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)
             trades_before_execution = len(manager.trades)
             manager.execute_actions(
                 decision["actions"],
                 execution_market_data,
-                execution_timestamp,
+                fill.filled_at,
                 fallback_prices=execution_fallback_prices,
                 execution_prices=execution_prices,
             )