diff --git a/dashboard/backend/api/routers/backtests.py b/dashboard/backend/api/routers/backtests.py index bef6abd6..ef1aa677 100644 --- a/dashboard/backend/api/routers/backtests.py +++ b/dashboard/backend/api/routers/backtests.py @@ -302,6 +302,11 @@ class RunMetadata(BaseModel): # same reason as above — the records live on the detail endpoint. t1_deferred_events: Optional[int] = None t1_deferred_shares: Optional[float] = None + frequency_contract: Optional[Dict[str, Any]] = None + market_data_quality: Optional[Dict[str, Any]] = None + market_data_feed: Optional[str] = None + sip_fallback_to_iex: Optional[bool] = None + end_clamped: Optional[bool] = None class EquityCurve(BaseModel): @@ -374,6 +379,11 @@ def _run_metadata_response(run: Dict[str, Any]) -> RunMetadata: "t1_deferred_events", "t1_deferred_shares", "llm_execution", + "frequency_contract", + "market_data_quality", + "market_data_feed", + "sip_fallback_to_iex", + "end_clamped", ): if field in metadata: if field == "llm_execution" and isinstance(metadata[field], dict): @@ -388,6 +398,43 @@ def _run_metadata_response(run: Dict[str, Any]) -> RunMetadata: ).model_dump(mode="json") except Exception: # noqa: BLE001 - legacy/malformed metadata continue + elif field == "frequency_contract" and isinstance( + metadata[field], dict + ): + payload[field] = { + name: metadata[field][name] + for name in ( + "source_timeframe", + "decision_timeframe", + "decision_frequency", + "execution_timeframe", + "valuation_frequency", + "aggregation", + "fill_policy", + "verification_status", + ) + if name in metadata[field] + } + elif field == "market_data_quality" and isinstance( + metadata[field], dict + ): + # Per-symbol detail remains in the owned run record. List + # routes only need bounded aggregate counts for the UI. + payload[field] = { + name: metadata[field][name] + for name in ( + "policy", + "decision_timestamp_min_symbol_coverage", + "total_decision_bars", + "usable_decision_bars", + "dropped_decision_bars", + "missing_source_bars", + "duplicate_source_bars", + "off_grid_source_bars", + "invalid_source_bars", + ) + if name in metadata[field] + } else: payload[field] = metadata[field] return RunMetadata(**payload) diff --git a/dashboard/backend/domain/backtesting/bar_aggregation.py b/dashboard/backend/domain/backtesting/bar_aggregation.py new file mode 100644 index 00000000..12e5a460 --- /dev/null +++ b/dashboard/backend/domain/backtesting/bar_aggregation.py @@ -0,0 +1,279 @@ +"""Session-aware aggregation of source bars into decision bars. + +The market-data provider returns bars at the configured source resolution. A +strategy must only see a completed decision bar, so this module labels each +bucket at its *right* edge. For example, US 5-minute bars from 09:30 through +10:25 become the 10:30 decision bar. The next source bar, opening at 10:30, +can therefore be used as the execution bar without look-ahead. + +This is intentionally independent of any provider SDK. It also avoids a +plain pandas ``resample`` because exchange sessions do not begin at midnight +and some markets have a lunch break. +""" + +from __future__ import annotations + +from datetime import time +from typing import Any, Dict, Iterable, Mapping + +import numpy as np +import pandas as pd + +from dashboard.backend.infrastructure.market_data.frequency import ( + normalize_bar_timeframe, + timeframe_minutes, +) + + +class BarAggregationError(ValueError): + """Raised when source bars cannot be safely aggregated.""" + + +_QUALITY_COUNT_COLUMNS = ( + "missing_source_bars", + "duplicate_source_bars", + "off_grid_source_bars", + "invalid_source_bars", +) + + +def _session_windows(market: str) -> tuple[tuple[time, time], ...]: + canonical = str(market or "US").strip().upper() + if canonical == "CN": + return ((time(9, 30), time(11, 30)), (time(13, 0), time(15, 0))) + return ((time(9, 30), time(16, 0)),) + + +def _as_local_index(frame: pd.DataFrame, timezone: str) -> pd.DataFrame: + if not isinstance(frame.index, pd.DatetimeIndex): + raise BarAggregationError("source bars must use a DatetimeIndex") + result = frame.copy() + if result.index.tz is None: + result.index = result.index.tz_localize(timezone) + else: + result.index = result.index.tz_convert(timezone) + return result.sort_index() + + +def _session_for_timestamp( + timestamp: pd.Timestamp, + windows: Iterable[tuple[time, time]], +) -> tuple[pd.Timestamp, pd.Timestamp] | None: + local_date = timestamp.normalize() + for start_time, end_time in windows: + start = local_date + pd.Timedelta( + hours=start_time.hour, minutes=start_time.minute + ) + end = local_date + pd.Timedelta( + hours=end_time.hour, minutes=end_time.minute + ) + if start <= timestamp < end: + return start, end + return None + + +def _weighted_vwap(group: pd.DataFrame, close: float) -> float: + if "vwap" not in group.columns: + return close + values = pd.to_numeric(group["vwap"], errors="coerce") + volumes = pd.to_numeric(group["volume"], errors="coerce").fillna(0.0) + valid = values.notna() & volumes.gt(0) + if valid.any() and float(volumes[valid].sum()) > 0: + return float((values[valid] * volumes[valid]).sum() / volumes[valid].sum()) + return close + + +def aggregate_bars( + frame: pd.DataFrame, + *, + source_timeframe: str, + decision_timeframe: str = "60m", + market: str = "US", + timezone: str = "US/Eastern", +) -> pd.DataFrame: + """Aggregate one symbol's source bars into completed session bars. + + The returned index is timezone-aware UTC, matching the canonical provider + boundary. Incomplete or missing source bars are not synthesized; quality + columns make the gap visible to callers. + """ + source = normalize_bar_timeframe(source_timeframe) + decision = normalize_bar_timeframe(decision_timeframe) + source_minutes = timeframe_minutes(source) + decision_minutes = timeframe_minutes(decision) + if source_minutes >= decision_minutes: + raise BarAggregationError( + "aggregation requires source_timeframe to be finer than " + "decision_timeframe" + ) + required = ("open", "high", "low", "close", "volume") + missing = sorted(set(required).difference(frame.columns)) + if missing: + raise BarAggregationError( + f"source bars are missing required columns: {', '.join(missing)}" + ) + if frame.empty: + return frame.copy() + + local = _as_local_index(frame, timezone) + windows = _session_windows(market) + buckets: dict[pd.Timestamp, list[pd.Series]] = {} + bucket_ends: dict[pd.Timestamp, pd.Timestamp] = {} + for timestamp, row in local.iterrows(): + session = _session_for_timestamp(timestamp, windows) + if session is None: + continue + session_start, session_end = session + elapsed_minutes = int((timestamp - session_start).total_seconds() // 60) + offset_minutes = (elapsed_minutes // decision_minutes) * decision_minutes + bucket_start = session_start + pd.Timedelta(minutes=offset_minutes) + bucket_end = min( + bucket_start + pd.Timedelta(minutes=decision_minutes), session_end + ) + # A source bar can only belong to a decision bucket that has not ended. + if bucket_start >= bucket_end: + continue + buckets.setdefault(bucket_start, []).append(row) + bucket_ends[bucket_start] = bucket_end + + records: list[dict] = [] + for bucket_start in sorted(buckets): + group = pd.DataFrame(buckets[bucket_start]) + group = group.sort_index() + bucket_end = bucket_ends[bucket_start] + expected = int( + (bucket_end - bucket_start).total_seconds() // (source_minutes * 60) + ) + expected_index = pd.date_range( + bucket_start, + periods=expected, + freq=f"{source_minutes}min", + ) + actual_index = pd.DatetimeIndex(group.index) + unique_actual_index = actual_index.unique() + missing_source_bars = len(expected_index.difference(unique_actual_index)) + duplicate_source_bars = len(actual_index) - len(unique_actual_index) + off_grid_source_bars = len(unique_actual_index.difference(expected_index)) + required_values = group.loc[:, list(required)].apply( + pd.to_numeric, errors="coerce" + ) + invalid_source_bars = int( + (~np.isfinite(required_values.to_numpy(dtype=float)).all(axis=1)).sum() + ) + is_complete = not any( + ( + missing_source_bars, + duplicate_source_bars, + off_grid_source_bars, + invalid_source_bars, + ) + ) + volume = float(pd.to_numeric(group["volume"], errors="coerce").fillna(0).sum()) + close = float(group["close"].iloc[-1]) + record = { + "timestamp": bucket_end.tz_convert("UTC"), + "open": float(group["open"].iloc[0]), + "high": float(pd.to_numeric(group["high"], errors="coerce").max()), + "low": float(pd.to_numeric(group["low"], errors="coerce").min()), + "close": close, + "volume": volume, + "source_bar_count": int(len(group)), + "expected_source_bars": expected, + "missing_source_bars": int(missing_source_bars), + "duplicate_source_bars": int(duplicate_source_bars), + "off_grid_source_bars": int(off_grid_source_bars), + "invalid_source_bars": invalid_source_bars, + "is_complete": is_complete, + "has_gap": not is_complete, + } + if "trade_count" in group.columns: + record["trade_count"] = float( + pd.to_numeric(group["trade_count"], errors="coerce") + .fillna(0) + .sum() + ) + if "vwap" in group.columns: + record["vwap"] = _weighted_vwap(group, close) + records.append(record) + + if not records: + columns = ["open", "high", "low", "close", "volume"] + return pd.DataFrame(columns=columns, index=pd.DatetimeIndex([], tz="UTC")) + + result = pd.DataFrame.from_records(records).set_index("timestamp").sort_index() + result.attrs.update(dict(getattr(frame, "attrs", {}) or {})) + result.attrs.update( + { + "aggregation_source_timeframe": source, + "aggregation_decision_timeframe": decision, + "aggregation_market": str(market or "US").strip().upper(), + "aggregation_timezone": timezone, + } + ) + return result + + +def aggregate_bars_by_symbol( + bars_by_symbol: Mapping[str, pd.DataFrame], + *, + source_timeframe: str, + decision_timeframe: str = "60m", + market: str = "US", + timezone: str = "US/Eastern", +) -> Dict[str, pd.DataFrame]: + """Aggregate each symbol independently, preserving the symbol mapping.""" + return { + symbol: aggregate_bars( + frame, + source_timeframe=source_timeframe, + decision_timeframe=decision_timeframe, + market=market, + timezone=timezone, + ) + for symbol, frame in bars_by_symbol.items() + } + + +def summarize_aggregation_quality( + bars_by_symbol: Mapping[str, pd.DataFrame], +) -> Dict[str, Any]: + """Return a JSON-safe audit summary before incomplete bars are dropped. + + Counts are symbol-bar counts: the same decision timestamp contributes once + for each symbol. Keeping this summary before filtering makes a completed + run distinguishable from one that silently lost source observations. + """ + + summary: Dict[str, Any] = { + "policy": "drop_incomplete_decision_bars", + "decision_timestamp_min_symbol_coverage": 0.8, + "total_decision_bars": 0, + "usable_decision_bars": 0, + "dropped_decision_bars": 0, + **{column: 0 for column in _QUALITY_COUNT_COLUMNS}, + "symbols": {}, + } + for symbol, frame in bars_by_symbol.items(): + total = int(len(frame)) + if "is_complete" in frame.columns: + usable = int(frame["is_complete"].fillna(False).astype(bool).sum()) + else: + usable = total + symbol_summary: Dict[str, Any] = { + "total_decision_bars": total, + "usable_decision_bars": usable, + "dropped_decision_bars": total - usable, + } + for column in _QUALITY_COUNT_COLUMNS: + value = ( + int(pd.to_numeric(frame[column], errors="coerce").fillna(0).sum()) + if column in frame.columns + else 0 + ) + symbol_summary[column] = value + summary[column] += value + summary["symbols"][symbol] = symbol_summary + summary["total_decision_bars"] += total + summary["usable_decision_bars"] += usable + summary["dropped_decision_bars"] += total - usable + return summary diff --git a/dashboard/backend/domain/backtesting/engine.py b/dashboard/backend/domain/backtesting/engine.py index 4b74ec86..17e01fbb 100644 --- a/dashboard/backend/domain/backtesting/engine.py +++ b/dashboard/backend/domain/backtesting/engine.py @@ -14,9 +14,12 @@ be extracted in a later phase. """ +import inspect import json import uuid +from bisect import bisect_left from datetime import date, datetime, time +from math import ceil from typing import Any, Dict, List, Optional, Tuple from dashboard.backend.database import db @@ -32,6 +35,10 @@ CurrencyContextError, ) from dashboard.backend.domain.backtesting.features import TechnicalIndicators +from dashboard.backend.domain.backtesting.bar_aggregation import ( + aggregate_bars_by_symbol, + summarize_aggregation_quality, +) from dashboard.backend.domain.backtesting.metrics import ( calculate_sharpe, calculate_max_drawdown, @@ -49,7 +56,10 @@ normalize_runtime_type, ) from dashboard.backend.infrastructure.ai_hedge_fund.adapter import AiHedgeFundRuntime -from dashboard.backend.infrastructure.market_data.alpaca_bars import MarketDataUnavailableError +from dashboard.backend.infrastructure.market_data.alpaca_bars import ( + MarketDataUnavailableError, + feed_provenance, +) from dashboard.backend.infrastructure.market_data.ifind_client import IFindClientError from dashboard.backend.infrastructure.market_data.ifind_fx import ( IFindFxError, @@ -60,6 +70,13 @@ ALPACA, create_market_data_provider, ) +from dashboard.backend.infrastructure.market_data.frequency import ( + FrequencyConfigError, + build_verified_intraday_contract, + normalize_bar_timeframe, + timeframe_minutes, + verify_source_timeframe, +) from dashboard.backend.infrastructure.market_data.profiles import ( IFIND_ASHARE, LLM_DECISION_SOURCE, @@ -170,6 +187,7 @@ def __init__( stock_pool: Optional[str] = None, pool_mode: Optional[str] = None, universe_selection: Optional[Dict] = None, + source_timeframe: Optional[str] = None, ): # Validate and swap dates if they're in the wrong order from datetime import datetime as dt_parser @@ -227,6 +245,26 @@ def __init__( # run metadata so a partially-degraded run is legible after the fact. self.runtime_step_failures: List[str] = [] self.profile = get_market_profile(data_source, universe) + self.requested_source_timeframe = normalize_bar_timeframe( + source_timeframe + if source_timeframe is not None + else getattr(self.profile, "source_timeframe", self.profile.timeframe) + ) + self.source_timeframe = self.requested_source_timeframe + self.decision_timeframe = getattr( + self.profile, "decision_timeframe", self.profile.timeframe + ) + self.execution_timeframe = getattr( + self.profile, "execution_timeframe", self.profile.timeframe + ) + self.valuation_frequency = getattr( + self.profile, "valuation_frequency", self.profile.timeframe + ) + self.intraday_mode = False + self.source_data = {} + self.data_quality = {} + self.frequency_contract = None + self.market_data_provenance = {} self.currency_context: CurrencyContext | None = None self.native_initial_capital = self.initial_capital if self.profile.native_currency == self.profile.reporting_currency: @@ -322,10 +360,27 @@ def __init__( else: print(f"✅ LLM initialized (model={self.model})") - self.data_loader = create_market_data_provider( - data_source, - self.profile.universe, - ) + self.data_loader = self._create_market_data_provider() + + def _create_market_data_provider(self): + """Create the selected provider without breaking legacy test doubles.""" + factory = create_market_data_provider + try: + parameters = inspect.signature(factory).parameters.values() + accepts_source_timeframe = any( + parameter.name == "source_timeframe" + or parameter.kind == inspect.Parameter.VAR_KEYWORD + for parameter in parameters + ) + except (TypeError, ValueError): + accepts_source_timeframe = False + if accepts_source_timeframe: + return factory( + self.data_source, + self.profile.universe, + source_timeframe=self.requested_source_timeframe, + ) + return factory(self.data_source, self.profile.universe) def _serialize_trades(self, trades: List[Dict]) -> List[Dict]: serialized = [] @@ -561,7 +616,7 @@ def _publish_live_progress(self, step: int, total_steps: int, manager) -> None: print(f" ⚠️ Could not write live progress: {exc}") def load_data(self): - """Fetch hourly data from the selected normalized provider.""" + """Fetch source bars and build the strategy's decision-bar dataset.""" # Keep the error path usable for legacy callers that construct an # instance with ``__new__`` (or inject a loader) before initialization. symbols = getattr(self, "symbols", ()) @@ -569,10 +624,10 @@ def load_data(self): f" Universe: {len(symbols)} symbols ({', '.join(symbols[:8])}" f"{'…' if len(symbols) > 8 else ''})" ) - self.all_data = self.data_loader.fetch_bars( + self.source_data = self.data_loader.fetch_bars( symbols, self.start_date, self.end_date ) - if not self.all_data: + if not self.source_data: # Raise, don't sys.exit(1): this runs inside server threads # (external runs, algo service) where SystemExit evades # `except Exception` and strands the run (the B0 class). @@ -581,6 +636,87 @@ def load_data(self): f"No {self.data_source} market data available for " f"{self.start_date}..{self.end_date}" ) + + configured_source_timeframe = getattr( + self.data_loader, "source_timeframe", None + ) + if configured_source_timeframe is None: + # A provider replacement that predates the frequency contract is + # assumed to return its historical profile resolution. This keeps + # injected hourly loaders from being accidentally aggregated as if + # they had honoured the new optional factory argument. + actual_source_timeframe = normalize_bar_timeframe( + self.profile.timeframe + ) + else: + try: + actual_source_timeframe = verify_source_timeframe( + self.requested_source_timeframe, + configured_source_timeframe, + evidence="configured", + ) + except FrequencyConfigError as exc: + raise MarketDataUnavailableError( + f"Market data frequency contract failed: {exc}" + ) from exc + fetch_evidence = getattr(self.data_loader, "last_fetch", None) + if isinstance(fetch_evidence, dict) and fetch_evidence.get( + "source_timeframe" + ): + try: + actual_source_timeframe = verify_source_timeframe( + self.requested_source_timeframe, + fetch_evidence["source_timeframe"], + evidence="fetch", + ) + except FrequencyConfigError as exc: + raise MarketDataUnavailableError( + f"Market data frequency contract failed: {exc}" + ) from exc + self.source_timeframe = actual_source_timeframe + self.market_data_provenance = feed_provenance(self.source_data) or {} + self.intraday_mode = timeframe_minutes(actual_source_timeframe) < timeframe_minutes( + self.decision_timeframe + ) + if self.intraday_mode: + # This Phase 2 engine uses the fetched source bar for both fill and + # valuation. If a caller explicitly selects 1m instead of the + # profile's 5m target, metadata must describe the effective clocks. + self.execution_timeframe = actual_source_timeframe + self.valuation_frequency = actual_source_timeframe + self.frequency_contract = build_verified_intraday_contract( + source_timeframe=actual_source_timeframe, + decision_timeframe=self.decision_timeframe, + decision_frequency=self.profile.decision_frequency, + ) + print( + f" Aggregating {actual_source_timeframe} source bars into " + f"{self.decision_timeframe} decision bars..." + ) + aggregated_data = aggregate_bars_by_symbol( + self.source_data, + source_timeframe=actual_source_timeframe, + decision_timeframe=self.decision_timeframe, + market=self.profile.market, + timezone=self.profile.timezone, + ) + self.data_quality = summarize_aggregation_quality(aggregated_data) + self.all_data = { + symbol: ( + frame.loc[frame["is_complete"]].copy() + if "is_complete" in frame.columns + else frame + ) + for symbol, frame in aggregated_data.items() + if not frame.empty + } + if not self.all_data: + raise MarketDataUnavailableError( + "Source bars were fetched, but no completed decision bars " + "could be built" + ) + else: + self.all_data = self.source_data if self.data_source == IFIND_ASHARE: self._ifind_common_start = self._validate_ifind_loaded_data() self._initialize_ifind_market_rules() @@ -737,7 +873,21 @@ def _run_metadata( "native_currency": profile.native_currency, "reporting_currency": profile.reporting_currency, "lot_size": profile.lot_size, + **dict(getattr(self, "market_data_provenance", {}) or {}), } + if getattr(self, "intraday_mode", False): + frequency_contract = getattr(self, "frequency_contract", None) + metadata["frequency_contract"] = dict( + frequency_contract + or build_verified_intraday_contract( + source_timeframe=self.source_timeframe, + decision_timeframe=self.decision_timeframe, + decision_frequency=profile.decision_frequency, + ) + ) + metadata["market_data_quality"] = dict( + getattr(self, "data_quality", {}) + ) if getattr(self, "universe_selection", None) is not None: metadata["universe_selection"] = dict(self.universe_selection) if profile.transaction_cost_profile is not None: @@ -961,7 +1111,11 @@ def _market_hours_only(self, timestamps): market_tz = pytz.timezone(profile.timezone) kept = [] for timestamp in timestamps: - local = timestamp.astimezone(market_tz) + local = ( + market_tz.localize(timestamp) + if timestamp.tzinfo is None + else timestamp.astimezone(market_tz) + ) local_time = local.time() if profile.market == "CN": is_market_hours = ( @@ -978,6 +1132,18 @@ def _market_hours_only(self, timestamps): kept.append(timestamp) return kept + def _market_day_key(self, timestamp) -> str: + """Return a trading-day key in the market's local timezone.""" + import pytz + + market_tz = pytz.timezone(self._effective_profile().timezone) + local = ( + market_tz.localize(timestamp) + if timestamp.tzinfo is None + else timestamp.astimezone(market_tz) + ) + return local.date().isoformat() + def _run_daily_post_trade( self, *, @@ -1022,8 +1188,53 @@ def _run_daily_post_trade( self.prompt_adaptations.append(record) self.pipeline = recombine_pipeline(patched, post_trade_steps) + @staticmethod + def _timestamps_for_data(data: Dict[str, Any]) -> List[datetime]: + timestamps = set() + for frame in data.values(): + timestamps.update(frame.index) + return sorted(timestamps) + + @staticmethod + def _market_data_at( + data: Dict[str, Any], symbols: List[str], timestamp: datetime + ) -> Dict[str, Any]: + return { + symbol: data[symbol].loc[timestamp] + for symbol in symbols + if symbol in data and timestamp in data[symbol].index + } + + @staticmethod + def _forward_filled_price_cache( + data: Dict[str, Any], timestamps: List[datetime] + ) -> Dict[str, Dict[datetime, Any]]: + cache: Dict[str, Dict[datetime, Any]] = {} + for symbol, frame in data.items(): + prices: Dict[datetime, Any] = {} + last_price = None + for timestamp in timestamps: + if timestamp in frame.index: + last_price = frame.loc[timestamp, "close"] + if last_price is not None: + prices[timestamp] = last_price + cache[symbol] = prices + return cache + + def _annualization_periods(self) -> float | None: + """Return the Sharpe sampling factor for the curve this run emits.""" + if not getattr(self, "intraday_mode", False): + return None + minutes = timeframe_minutes(self.source_timeframe) + return 252 * 6.5 * (60 / minutes) + def run_agent_backtest(self) -> Tuple[str, List[Dict]]: - """Run backtest with agent making hourly decisions.""" + """Run a backtest with hourly strategy decisions. + + When a finer source dataset is configured, the strategy still receives + one completed hourly bar per step while fills and mark-to-market use + the finer source timeline. + """ print("🤖 Running Agent backtest (hourly decisions)...\n") # Track LLM usage for results metadata @@ -1046,21 +1257,18 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: "once per trading day\n" ) - # Get all timestamps - all_timestamps = set() - for df in self.all_data.values(): - all_timestamps.update(df.index) - all_timestamps = sorted(all_timestamps) + # Decision timestamps come from the completed decision-bar dataset. + all_timestamps = self._timestamps_for_data(self.all_data) # Filter: only keep hours with real data for 80%+ of symbols - min_required = int(len(self.all_data) * 0.8) + min_required = max(1, ceil(len(self.all_data) * 0.8)) filtered = [] for ts in all_timestamps: real_data_count = sum(1 for df in self.all_data.values() if ts in df.index) if real_data_count >= min_required: filtered.append(ts) - all_timestamps = filtered if filtered else all_timestamps + all_timestamps = filtered all_timestamps = self._market_hours_only(all_timestamps) prior_market_dates = ( @@ -1080,9 +1288,50 @@ 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 + ] + print( - f" Trading {len(all_timestamps)} bars during " - f"{self.profile.market} {self.profile.timeframe} sessions...\n" + f" Trading {len(all_timestamps)} hourly decision bars during " + f"{self.profile.market} {self.profile.timeframe} sessions" + + ( + f"; executing/valuing on {self.source_timeframe} bars" + if self.intraday_mode + else "" + ) + + "...\n" ) total_steps = len(all_timestamps) # Declare the run length so a strict-LLM run can absorb a small number @@ -1095,22 +1344,20 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: f"{manager.strict_llm_fallback_budget()} unusable response(s)\n" ) - # Build forward-filled price cache to handle missing hourly data + # Build separate decision and valuation caches. In minute mode the + # strategy sees hourly prices while portfolio valuation sees every + # source bar. print(" Pre-computing forward-filled price cache...") - price_cache = {} - for symbol, df in self.all_data.items(): - price_cache[symbol] = {} - last_price = None - - for timestamp in all_timestamps: - if timestamp in df.index: - last_price = df.loc[timestamp, "close"] - price_cache[symbol][timestamp] = last_price - else: - # Fallback (shouldn't happen with daily data) - if last_price is not None: - price_cache[symbol][timestamp] = last_price - + price_cache = self._forward_filled_price_cache( + self.all_data, all_timestamps + ) + valuation_price_cache = ( + self._forward_filled_price_cache(self.source_data, raw_timestamps) + if self.intraday_mode + else price_cache + ) + valuation_cursor = 0 + print(" ✅ Cache ready\n") day_episode: Dict = { @@ -1131,15 +1378,10 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: "latest_step_outputs": [], } - # Get market data for this hour (real data when available) - market_data = {} - for symbol in self.symbols: - if symbol not in self.all_data: - continue - df = self.all_data[symbol] - if timestamp not in df.index: - continue - market_data[symbol] = df.loc[timestamp] + # Decision signals always use the completed hourly bar. + market_data = self._market_data_at( + self.all_data, self.symbols, timestamp + ) # Get portfolio state (uses real data for signals, forward-fill for valuation) state = manager.get_portfolio_state(market_data, price_cache, timestamp) @@ -1217,31 +1459,78 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: decision = {"actions": []} runtime_invoked = self.runtime_dispatcher.calls > runtime_calls_before + # 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] + execution_market_data = market_data + execution_fallback_prices = { + symbol: values[execution_timestamp] + for symbol, values in valuation_price_cache.items() + if execution_timestamp in values + } + execution_prices = None + if self.intraday_mode: + execution_market_data = self._market_data_at( + self.source_data, + self.symbols, + execution_timestamp, + ) + execution_prices = { + symbol: row["open"] + for symbol, row in execution_market_data.items() + if "open" in row + } + # Execute trades (only if real data available) trades_before_execution = len(manager.trades) - fallback_prices = { - symbol: values[timestamp] - for symbol, values in price_cache.items() - if timestamp in values - } manager.execute_actions( decision["actions"], - market_data, - timestamp, - fallback_prices=fallback_prices, + execution_market_data, + execution_timestamp, + fallback_prices=execution_fallback_prices, + execution_prices=execution_prices, ) if runtime_invoked: self.runtime_dispatcher.record_latest_execution( len(manager.trades) - trades_before_execution ) - # Update equity (uses forward-filled prices for smooth valuation) - manager.update_equity(market_data, price_cache, timestamp) - manager.equity_history[-1] = ( - self._require_currency_context().reporting_equity_record( - manager.equity_history[-1] + # Update equity. The minute path emits one mark for every source + # bar through the fill, while the legacy path emits one hourly mark. + if self.intraday_mode: + while ( + valuation_cursor < len(raw_timestamps) + and raw_timestamps[valuation_cursor] <= execution_timestamp + ): + valuation_timestamp = raw_timestamps[valuation_cursor] + valuation_market_data = self._market_data_at( + self.source_data, + self.symbols, + valuation_timestamp, + ) + manager.update_equity( + valuation_market_data, + valuation_price_cache, + valuation_timestamp, + ) + manager.equity_history[-1] = ( + self._require_currency_context().reporting_equity_record( + manager.equity_history[-1] + ) + ) + valuation_cursor += 1 + else: + manager.update_equity( + execution_market_data, + price_cache, + execution_timestamp, + ) + manager.equity_history[-1] = ( + self._require_currency_context().reporting_equity_record( + manager.equity_history[-1] + ) ) - ) self._publish_live_progress(i + 1, total_steps, manager) if post_trade_steps and is_last_bar_of_trading_day(all_timestamps, i): @@ -1255,7 +1544,29 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: if (i + 1) % 100 == 0: equity = manager.equity_history[-1]["equity"] pct_return = ((equity - self.initial_capital) / self.initial_capital) * 100 - print(f" Hour {i+1}/{len(all_timestamps)}: Equity ${equity:,.0f} ({pct_return:+.1f}%)") + print(f" Decision {i+1}/{len(all_timestamps)}: Equity ${equity:,.0f} ({pct_return:+.1f}%)") + + if self.intraday_mode: + # Mark any remaining source bars after the final decision/fill so + # the curve closes at the end of the requested market window. + while valuation_cursor < len(raw_timestamps): + valuation_timestamp = raw_timestamps[valuation_cursor] + valuation_market_data = self._market_data_at( + self.source_data, + self.symbols, + valuation_timestamp, + ) + manager.update_equity( + valuation_market_data, + valuation_price_cache, + valuation_timestamp, + ) + manager.equity_history[-1] = ( + self._require_currency_context().reporting_equity_record( + manager.equity_history[-1] + ) + ) + valuation_cursor += 1 equity_curve = manager.get_equity_curve() @@ -1327,7 +1638,9 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]: initial_equity=initial_eq, final_equity=final_eq, total_return=total_return, - sharpe_ratio=self._calc_sharpe(equity_curve), + sharpe_ratio=self._calc_sharpe( + equity_curve, periods_per_year=self._annualization_periods() + ), max_drawdown=self._calc_max_dd(equity_curve), num_trades=len(manager.trades), llm_model=llm_model, # Track which model was used @@ -1464,6 +1777,14 @@ def run_djia_baseline(self) -> Tuple[str, List[Dict]]: if not bars: print(" ⚠️ No DJIA bars available; skipping index baseline") return None, [] + if self.intraday_mode: + bars = aggregate_bars_by_symbol( + bars, + source_timeframe=self.source_timeframe, + decision_timeframe=self.decision_timeframe, + market=self.profile.market, + timezone=self.profile.timezone, + ) _, equity_history = generate_baselines( bars_by_symbol=bars, @@ -1510,14 +1831,16 @@ def run_djia_baseline(self) -> Tuple[str, List[Dict]]: return run_id, equity_history @staticmethod - def _calc_sharpe(equity_curve: List[Dict]) -> float: - """Annualized hourly Sharpe ratio. + def _calc_sharpe( + equity_curve: List[Dict], periods_per_year: Optional[float] = None + ) -> float: + """Annualized Sharpe ratio for the curve's sampling frequency. Delegates to dashboard.backend.domain.backtesting.metrics.calculate_sharpe; - inputs, outputs, edge cases, and the hourly annualization factor are - unchanged. + Omitting ``periods_per_year`` preserves the historical hourly factor; + minute-valued runs pass the finer sampling factor explicitly. """ - return calculate_sharpe(equity_curve) + return calculate_sharpe(equity_curve, periods_per_year=periods_per_year) @staticmethod def _calc_max_dd(equity_curve: List[Dict]) -> float: diff --git a/dashboard/backend/domain/backtesting/external_run_service.py b/dashboard/backend/domain/backtesting/external_run_service.py index 9fe8721b..4857360f 100644 --- a/dashboard/backend/domain/backtesting/external_run_service.py +++ b/dashboard/backend/domain/backtesting/external_run_service.py @@ -38,8 +38,15 @@ calculate_max_drawdown, calculate_sharpe, ) +from dashboard.backend.infrastructure.market_data.frequency import ( + build_verified_intraday_contract, + timeframe_minutes, +) from dashboard.backend.domain.backtesting.portfolio_manager import PortfolioManager -from dashboard.backend.infrastructure.market_data.alpaca_bars import AlpacaDataLoader +from dashboard.backend.infrastructure.market_data.alpaca_bars import ( + AlpacaDataLoader, + feed_provenance, +) from dashboard.backend.infrastructure.market_data.profiles import ( ALPACA, get_market_profile, @@ -196,6 +203,23 @@ def build_final_metrics(run: Optional[Dict[str, Any]]) -> Dict[str, Any]: {} for a missing row.""" if not run: return {} + metadata = run.get("metadata") + if not isinstance(metadata, dict): + metadata = {} + frequency_contract = metadata.get("frequency_contract") + if isinstance(frequency_contract, dict): + frequency_contract = dict(frequency_contract) + else: + frequency_contract = None + market_data_quality = metadata.get("market_data_quality") + if isinstance(market_data_quality, dict): + market_data_quality = { + key: value + for key, value in market_data_quality.items() + if key != "symbols" + } + else: + market_data_quality = None return { "total_return": run.get("total_return"), "sharpe_ratio": run.get("sharpe_ratio"), @@ -206,8 +230,12 @@ def build_final_metrics(run: Optional[Dict[str, Any]]) -> Dict[str, Any]: "input_tokens": run.get("input_tokens"), "output_tokens": run.get("output_tokens"), "est_cost_usd": run.get("est_cost_usd"), - "timeout_holds": (run.get("metadata") or {}).get("timeout_holds") - if isinstance(run.get("metadata"), dict) else None, + "timeout_holds": metadata.get("timeout_holds"), + "frequency_contract": frequency_contract, + "market_data_quality": market_data_quality, + "market_data_feed": metadata.get("market_data_feed"), + "sip_fallback_to_iex": metadata.get("sip_fallback_to_iex"), + "end_clamped": metadata.get("end_clamped"), } @@ -260,6 +288,9 @@ def __init__( # A-share run silently execute with US same-day settlement — no error, # no log, just wrong fills. self.profile = get_market_profile(ALPACA) + self.source_timeframe = self.profile.source_timeframe + self.decision_timeframe = self.profile.decision_timeframe + self.intraday_mode = False self.manager = PortfolioManager( initial_capital=self.initial_capital, t_plus_one_enabled=self.profile.t_plus_one_enabled, @@ -267,6 +298,14 @@ def __init__( self.all_data: Dict[str, pd.DataFrame] = {} self.timestamps: List[Any] = [] self.price_cache: Dict[str, Dict[Any, float]] = {} + 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.data_quality: Dict[str, Any] = {} + self.frequency_contract: Optional[Dict[str, str]] = None + self.market_data_provenance: Dict[str, Any] = {} + self._valuation_cursor = 0 self.step_opened_at: Optional[datetime] = None # Stamped by sweep_terminal_sessions() the first time it sees this @@ -299,6 +338,8 @@ def load_market_data(self) -> None: dataset = market_data_store.get_dataset( DJIA_30, self.start_date, self.end_date, loader_factory=AlpacaDataLoader, + source_timeframe=self.profile.source_timeframe, + decision_timeframe=self.profile.decision_timeframe, ) self.adopt_dataset(dataset) @@ -317,6 +358,44 @@ def adopt_dataset(self, dataset: "market_data_store.MarketDataset") -> None: self.timestamps = dataset.timestamps self.price_cache = dataset.price_cache self.total_steps = dataset.total_steps + self.source_data = getattr(dataset, "source_data", self.all_data) + self.source_timestamps = list( + getattr(dataset, "source_timestamps", self.timestamps) + ) + self.source_price_cache = getattr( + dataset, "source_price_cache", self.price_cache + ) + self.execution_timestamps = list( + getattr(dataset, "execution_timestamps", self.timestamps) + ) + self.data_quality = dict(getattr(dataset, "data_quality", {}) or {}) + self.source_timeframe = getattr( + dataset, "source_timeframe", self.profile.timeframe + ) + self.decision_timeframe = getattr( + dataset, "decision_timeframe", self.profile.timeframe + ) + self.intraday_mode = timeframe_minutes(self.source_timeframe) < timeframe_minutes( + self.decision_timeframe + ) + self.market_data_provenance = feed_provenance(self.source_data) or {} + if self.intraday_mode: + self.frequency_contract = build_verified_intraday_contract( + source_timeframe=self.source_timeframe, + decision_timeframe=self.decision_timeframe, + decision_frequency=self.profile.decision_frequency, + ) + else: + self.frequency_contract = { + "source_timeframe": self.source_timeframe, + "decision_timeframe": self.decision_timeframe, + "decision_frequency": self.profile.decision_frequency, + "execution_timeframe": self.decision_timeframe, + "valuation_frequency": self.decision_timeframe, + "aggregation": "none", + "fill_policy": "decision_bar_close", + } + self._valuation_cursor = 0 with self._step_lock: if self.status in TERMINAL_STATUSES: @@ -326,7 +405,8 @@ def adopt_dataset(self, dataset: "market_data_store.MarketDataset") -> None: def _market_data_at(self, timestamp) -> Dict[str, pd.Series]: market_data = {} - for symbol in DJIA_30: + symbols = self.symbols or list(self.all_data) + for symbol in symbols: if symbol not in self.all_data: continue df = self.all_data[symbol] @@ -334,6 +414,42 @@ def _market_data_at(self, timestamp) -> Dict[str, pd.Series]: market_data[symbol] = df.loc[timestamp] return market_data + def _source_market_data_at(self, timestamp) -> Dict[str, pd.Series]: + """Return source-resolution bars used for fills and valuation.""" + source_data = self.source_data or self.all_data + symbols = self.symbols or list(source_data) + return { + symbol: source_data[symbol].loc[timestamp] + for symbol in symbols + if symbol in source_data and timestamp in source_data[symbol].index + } + + def _effective_source_timestamps(self) -> List[Any]: + return list(self.source_timestamps or self.timestamps) + + 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 _value_through(self, target_timestamp=None) -> None: + """Mark the portfolio on each source bar through the given timestamp.""" + source_timestamps = self._effective_source_timestamps() + source_price_cache = self._effective_source_price_cache() + while self._valuation_cursor < len(source_timestamps): + timestamp = source_timestamps[self._valuation_cursor] + if target_timestamp is not None and timestamp > target_timestamp: + break + self.manager.update_equity( + self._source_market_data_at(timestamp), + source_price_cache, + timestamp, + ) + self._valuation_cursor += 1 + def _open_current_step(self) -> None: self.step_opened_at = _utcnow() self.last_decision_source = None @@ -640,34 +756,55 @@ def _advance_step( # agent_runs.metadata). self.timeout_holds += 1 timestamp = self.timestamps[self.step_index] - market_data = self._market_data_at(timestamp) - - self.last_executed = [] - for action in executable: - self.last_executed.append({ - "symbol": action.get("symbol"), - "action": action.get("action"), - "shares": action.get("shares"), - "reason": action.get("reason"), - }) + execution_timestamp = self._effective_execution_timestamps()[self.step_index] + execution_market_data = self._source_market_data_at(execution_timestamp) + execution_prices = { + symbol: row["open"] + for symbol, row in execution_market_data.items() + if "open" in row + } - self.manager.execute_actions(executable, market_data, timestamp) - self.manager.update_equity(market_data, self.price_cache, timestamp) + trades_before_execution = len(self.manager.trades) + self.manager.execute_actions( + executable, + execution_market_data, + execution_timestamp, + fallback_prices={ + symbol: values[execution_timestamp] + for symbol, values in self._effective_source_price_cache().items() + if execution_timestamp in values + }, + execution_prices=execution_prices if self.intraday_mode else None, + ) + self.last_executed = [ + { + "symbol": trade.get("symbol"), + "action": str(trade.get("side", "")).lower(), + "shares": trade.get("shares"), + "reason": trade.get("reason"), + } + for trade in self.manager.trades[trades_before_execution:] + ] + self._value_through(execution_timestamp) self.decision_log.append({ "step_index": self.step_index, "timestamp": timestamp.isoformat() if hasattr(timestamp, "isoformat") else str(timestamp), + "execution_timestamp": execution_timestamp.isoformat() + if hasattr(execution_timestamp, "isoformat") + else str(execution_timestamp), "decision_source": decision_source, "actions_submitted": raw_actions or [], - "actions_executed": len(executable), + "actions_executed": len(self.last_executed), "context_ref": self.context_ref_by_step.get(self.step_index), }) self.last_decision_source = decision_source self.step_index += 1 if self.step_index >= self.total_steps: + self._value_through() self._finalize() else: self.status = "waiting_decision" @@ -701,7 +838,14 @@ def _finalize(self) -> None: initial_equity=initial_eq, final_equity=final_eq, total_return=total_return, - sharpe_ratio=calculate_sharpe(equity_curve), + sharpe_ratio=calculate_sharpe( + equity_curve, + periods_per_year=( + 252 * 6.5 * (60 / timeframe_minutes(self.source_timeframe)) + if self.intraday_mode + else None + ), + ), max_drawdown=calculate_max_drawdown(equity_curve), num_trades=len(self.manager.trades), llm_model=self.model_name, @@ -712,6 +856,13 @@ def _finalize(self) -> None: metadata={ "decision_timeout_seconds": DECISION_TIMEOUT_SECONDS, "timeout_holds": self.timeout_holds, + "frequency_contract": dict(self.frequency_contract or {}), + **( + {"market_data_quality": self.data_quality} + if self.intraday_mode + else {} + ), + **self.market_data_provenance, }, ) db.insert_equity_points(self.run_id, equity_curve) @@ -1105,7 +1256,13 @@ def start_backtest( # thread at all: attach it and open step 0 synchronously. Miss or # build-in-flight falls through to the loader thread exactly as before # (get_dataset inside the thread blocks/dedupes there). - dataset = market_data_store.peek(DJIA_30, start_date, end_date) + dataset = market_data_store.peek( + DJIA_30, + start_date, + end_date, + source_timeframe=session.profile.source_timeframe, + decision_timeframe=session.profile.decision_timeframe, + ) if dataset is not None: session.adopt_dataset(dataset) else: diff --git a/dashboard/backend/domain/backtesting/market_data_store.py b/dashboard/backend/domain/backtesting/market_data_store.py index 6c4dbd25..024a38ab 100644 --- a/dashboard/backend/domain/backtesting/market_data_store.py +++ b/dashboard/backend/domain/backtesting/market_data_store.py @@ -1,10 +1,11 @@ """Shared, immutable market-data datasets for backtest sessions (T1). -One dataset (indicator-enriched bars + trading timestamps + price cache) per -``(symbols, start_date, end_date)`` key, shared by every session with that -config. READ-ONLY CONTRACT: every consumer treats ``all_data`` frames, -``timestamps`` and ``price_cache`` as immutable — verified convention across -the engine, baselines, and PortfolioManager. Never mutate a dataset. +One dataset (indicator-enriched decision bars + source bars + trading +timestamps + price caches) per ``(symbols, start_date, end_date, +source_timeframe, decision_timeframe)`` key, shared by every session with that +config. READ-ONLY CONTRACT: every consumer treats the dataset frames, +timestamps and caches as immutable — verified convention across the engine, +baselines, and PortfolioManager. Never mutate a dataset. Concurrency model (deliberately NOT cache.py's coordinator, whose followers never block): the first requester for a key builds; concurrent requesters @@ -22,14 +23,25 @@ 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, + summarize_aggregation_quality, +) from dashboard.backend.infrastructure.market_data.alpaca_bars import AlpacaDataLoader +from dashboard.backend.infrastructure.market_data.frequency import ( + normalize_bar_timeframe, + timeframe_minutes, + verify_source_timeframe, +) # Read once at import (tests monkeypatch the module constant). Entry count, not # bytes: measured ~1.7 MB for a month-long dataset (was cited as ~50 MB), but @@ -50,15 +62,44 @@ class MarketDataset: """Immutable bundle of everything a session needs from market data.""" - __slots__ = ("key", "all_data", "timestamps", "price_cache", "total_steps") + __slots__ = ( + "key", "all_data", "timestamps", "price_cache", "total_steps", + "source_data", "source_timestamps", "source_price_cache", + "execution_timestamps", "source_timeframe", "decision_timeframe", + "data_quality", + ) def __init__(self, key: Tuple, all_data: Dict[str, pd.DataFrame], - timestamps: List[Any], price_cache: Dict[str, Dict[Any, float]]): + timestamps: List[Any], price_cache: Dict[str, Dict[Any, float]], + *, 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, + source_timeframe: str = "60m", + decision_timeframe: str = "60m", + data_quality: Optional[Dict[str, Any]] = None): self.key = key self.all_data = all_data self.timestamps = timestamps self.price_cache = price_cache self.total_steps = len(timestamps) + self.source_data = source_data if source_data is not None else all_data + self.source_timestamps = ( + source_timestamps if source_timestamps is not None else timestamps + ) + self.source_price_cache = ( + source_price_cache + if source_price_cache is not None + else price_cache + ) + self.execution_timestamps = ( + execution_timestamps + if execution_timestamps is not None + else list(timestamps) + ) + self.source_timeframe = source_timeframe + self.decision_timeframe = decision_timeframe + self.data_quality = data_quality or {} class _Entry: @@ -75,15 +116,42 @@ def __init__(self): _cache: "OrderedDict[Tuple, _Entry]" = OrderedDict() -def _dataset_key(symbols, start_date, end_date) -> Tuple: - return (tuple(symbols), str(start_date), str(end_date)) - - -def peek(symbols, start_date, end_date) -> Optional[MarketDataset]: +def _dataset_key( + symbols, + start_date, + end_date, + source_timeframe: str = "60m", + decision_timeframe: str = "60m", +) -> Tuple: + return ( + tuple(symbols), + str(start_date), + str(end_date), + normalize_bar_timeframe(source_timeframe), + normalize_bar_timeframe(decision_timeframe), + ) + + +def peek( + symbols, + start_date, + end_date, + *, + source_timeframe: str = "60m", + decision_timeframe: str = "60m", +) -> Optional[MarketDataset]: """Non-blocking: the resident dataset, or None (miss / build in flight / negative-cached failure). The only store call allowed under _create_lock.""" with _cache_lock: - entry = _cache.get(_dataset_key(symbols, start_date, end_date)) + entry = _cache.get( + _dataset_key( + symbols, + start_date, + end_date, + source_timeframe, + decision_timeframe, + ) + ) if entry is None or entry.dataset is None: return None _cache.move_to_end(entry.dataset.key) @@ -91,9 +159,17 @@ def peek(symbols, start_date, end_date) -> Optional[MarketDataset]: def get_dataset(symbols, start_date, end_date, - loader_factory: Optional[Callable[[], Any]] = None) -> MarketDataset: + loader_factory: Optional[Callable[[], Any]] = None, + *, source_timeframe: str = "60m", + decision_timeframe: str = "60m") -> MarketDataset: """Blocking single-flight build-or-wait. NEVER call under _create_lock.""" - key = _dataset_key(symbols, start_date, end_date) + key = _dataset_key( + symbols, + start_date, + end_date, + source_timeframe, + decision_timeframe, + ) factory = loader_factory or AlpacaDataLoader while True: with _cache_lock: @@ -112,7 +188,15 @@ def get_dataset(symbols, start_date, end_date, if is_leader: try: - dataset = _build_dataset(key, symbols, start_date, end_date, factory) + dataset = _build_dataset( + key, + symbols, + start_date, + end_date, + factory, + source_timeframe=source_timeframe, + decision_timeframe=decision_timeframe, + ) except BaseException as exc: with _cache_lock: entry.error = exc @@ -146,38 +230,160 @@ def get_dataset(symbols, start_date, end_date, # Entry was reset underneath us (tests); retry from scratch. -def _build_dataset(key, symbols, start_date, end_date, factory) -> MarketDataset: +def _build_dataset( + key, + symbols, + start_date, + end_date, + factory, + *, + source_timeframe: str, + decision_timeframe: str, +) -> MarketDataset: loader = factory() - all_data = loader.fetch_bars(list(symbols), start_date, end_date) - if not all_data: + requested_source = normalize_bar_timeframe(source_timeframe) + requested_decision = normalize_bar_timeframe(decision_timeframe) + configure = getattr(loader, "configure_source_timeframe", None) + if callable(configure): + configure(requested_source) + configured_source = getattr(loader, "source_timeframe", None) + if configured_source is None: + # Legacy test doubles and old hourly loaders have no runtime evidence; + # preserve their historical 60m behavior without attesting it as 5m. + actual_source = "60m" + else: + actual_source = verify_source_timeframe( + requested_source, + configured_source, + evidence="configured", + ) + source_data = loader.fetch_bars(list(symbols), start_date, end_date) + if not source_data: raise RuntimeError("No market data returned from Alpaca") + fetch_evidence = getattr(loader, "last_fetch", None) + if isinstance(fetch_evidence, dict) and fetch_evidence.get("source_timeframe"): + actual_source = verify_source_timeframe( + requested_source, + fetch_evidence["source_timeframe"], + evidence="fetch", + ) + data_quality: Dict[str, Any] = {} + if timeframe_minutes(actual_source) < timeframe_minutes(requested_decision): + aggregated_data = aggregate_bars_by_symbol( + source_data, + source_timeframe=actual_source, + decision_timeframe=requested_decision, + market="US", + timezone="US/Eastern", + ) + data_quality = summarize_aggregation_quality(aggregated_data) + all_data = { + symbol: frame.loc[frame["is_complete"]].copy() + for symbol, frame in aggregated_data.items() + if not frame.empty + } + else: + all_data = source_data + if not all_data: + raise RuntimeError("No completed decision bars returned from Alpaca") for symbol, df in all_data.items(): all_data[symbol] = TechnicalIndicators.calculate_indicators(df) timestamps = _build_trading_timestamps(all_data) if not timestamps: raise RuntimeError("No trading hours in the selected date range") price_cache = _build_price_cache(all_data, timestamps) - dataset = MarketDataset(key, all_data, timestamps, price_cache) + source_timestamps = _build_trading_timestamps( + source_data, + min_symbol_coverage=0.0, + ) + source_price_cache = _build_price_cache(source_data, source_timestamps) + execution_timestamps = _build_execution_timestamps( + timestamps, + source_timestamps, + timezone="US/Eastern", + ) + 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) + dataset = MarketDataset( + key, + all_data, + timestamps, + price_cache, + source_data=source_data, + source_timestamps=source_timestamps, + source_price_cache=source_price_cache, + execution_timestamps=execution_timestamps, + source_timeframe=actual_source, + decision_timeframe=requested_decision, + data_quality=data_quality, + ) mb = sum(float(df.memory_usage(deep=True).sum()) for df in all_data.values()) / 1e6 print(f"📊 market-data dataset built: {key[1]}→{key[2]} " f"({len(key[0])} syms, {dataset.total_steps} steps, ~{mb:.1f} MB)") return dataset -def _build_trading_timestamps(all_data: Dict[str, pd.DataFrame]) -> List[Any]: - """Moved verbatim from ExternalBacktestSession._build_trading_timestamps.""" +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, +) -> List[Any]: + """Return in-session timestamps meeting the requested symbol coverage.""" all_timestamps: set = set() for df in all_data.values(): all_timestamps.update(df.index) ordered = sorted(all_timestamps) - min_required = int(len(all_data) * 0.8) + min_required = max(1, ceil(len(all_data) * min_symbol_coverage)) filtered = [] for ts in ordered: real_count = sum(1 for df in all_data.values() if ts in df.index) if real_count >= min_required: filtered.append(ts) - ordered = filtered if filtered else ordered + ordered = filtered market_hours = [] for ts in ordered: diff --git a/dashboard/backend/domain/backtesting/metrics.py b/dashboard/backend/domain/backtesting/metrics.py index 19b298d1..688cf8e3 100644 --- a/dashboard/backend/domain/backtesting/metrics.py +++ b/dashboard/backend/domain/backtesting/metrics.py @@ -9,22 +9,23 @@ legacy methods now delegate here. """ -from typing import Dict, List +from typing import Dict, List, Optional import numpy as np -def calculate_sharpe(equity_curve: List[Dict]) -> float: +def calculate_sharpe( + equity_curve: List[Dict], periods_per_year: Optional[float] = None +) -> float: """ Calculate Sharpe ratio from hourly equity curve. Formula: sharpe = (mean(returns) / std(returns)) * sqrt(periods_per_year) - Data is HOURLY, so annualization factor = sqrt(252 * 6.5): - - 252 = trading days per year - - 6.5 = trading hours per day (9:30 AM - 4:00 PM ET) - - Total: sqrt(1638) ≈ 40.47 + ``periods_per_year`` defaults to the legacy hourly assumption. A 5-minute + valuation curve can pass ``252 * 6.5 * 12`` without changing the historical + hourly behavior. Returns: float Annualized Sharpe ratio. Returns 0 if insufficient data or zero volatility. @@ -38,8 +39,10 @@ def calculate_sharpe(equity_curve: List[Dict]) -> float: if len(returns) == 0 or np.std(returns) == 0: return 0 - # Annualize for hourly data: sqrt(252 trading days * 6.5 hours/day) - annualization_factor = np.sqrt(252 * 6.5) + periods_per_year = 252 * 6.5 if periods_per_year is None else periods_per_year + if periods_per_year <= 0: + raise ValueError("periods_per_year must be positive") + annualization_factor = np.sqrt(periods_per_year) return (np.mean(returns) / np.std(returns)) * annualization_factor diff --git a/dashboard/backend/domain/backtesting/portfolio_manager.py b/dashboard/backend/domain/backtesting/portfolio_manager.py index c2f3d84e..81ec0347 100644 --- a/dashboard/backend/domain/backtesting/portfolio_manager.py +++ b/dashboard/backend/domain/backtesting/portfolio_manager.py @@ -826,6 +826,7 @@ def execute_actions( market_data: Dict, timestamp: datetime, fallback_prices: Optional[Dict] = None, + execution_prices: Optional[Dict] = None, ): """Execute trading decisions.""" trades_before = len(self.trades) @@ -857,6 +858,7 @@ def execute_actions( transaction_cost_profile=self.transaction_cost_profile, market_rules=market_rules, fallback_prices=fallback_prices, + execution_prices=execution_prices, ) for trade in self.trades[trades_before:]: for field in self.transaction_cost_totals: diff --git a/dashboard/backend/domain/trading/execution.py b/dashboard/backend/domain/trading/execution.py index 75062236..69aac024 100644 --- a/dashboard/backend/domain/trading/execution.py +++ b/dashboard/backend/domain/trading/execution.py @@ -387,6 +387,7 @@ def execute_actions( transaction_cost_profile: Any = None, market_rules: Optional[Dict[str, Any]] = None, fallback_prices: Optional[Dict[str, Any]] = None, + execution_prices: Optional[Dict[str, Any]] = None, ) -> float: """Apply ``actions`` to the given portfolio state in place. @@ -450,7 +451,9 @@ def _record_rejection(**fields) -> None: market_rule = market_rules.get(symbol) reference_price = None - if symbol in market_data: + if execution_prices is not None and symbol in execution_prices: + reference_price = execution_prices[symbol] + elif symbol in market_data: reference_price = market_data[symbol]["close"] elif fallback_prices is not None: reference_price = fallback_prices.get(symbol) @@ -523,7 +526,11 @@ def _record_rejection(**fields) -> None: if symbol not in market_data: continue - price = market_data[symbol]["close"] + price = ( + execution_prices[symbol] + if execution_prices is not None and symbol in execution_prices + else market_data[symbol]["close"] + ) if market_rule is not None: price_tick = getattr(transaction_cost_profile, "price_tick", 0.01) diff --git a/dashboard/backend/execution/backtest_backend.py b/dashboard/backend/execution/backtest_backend.py index 50fb8f1d..d8c56b6f 100644 --- a/dashboard/backend/execution/backtest_backend.py +++ b/dashboard/backend/execution/backtest_backend.py @@ -86,8 +86,27 @@ def load_blocking(self) -> None: def start_background_load(self) -> None: # Fast path (runs under the shared create lock — peek only, never # get_dataset): a resident dataset skips the loader thread entirely. + # Prefer the session's effective clocks. Compatibility fakes and older + # adapters may expose neither these attributes nor ``profile``; their + # historical contract is the all-hourly default. + profile = getattr(self.session, "profile", None) + source_timeframe = getattr( + self.session, + "source_timeframe", + getattr(profile, "source_timeframe", "60m"), + ) + decision_timeframe = getattr( + self.session, + "decision_timeframe", + getattr(profile, "decision_timeframe", "60m"), + ) dataset = ext.market_data_store.peek( - DJIA_30, self.session.start_date, self.session.end_date) + DJIA_30, + self.session.start_date, + self.session.end_date, + source_timeframe=source_timeframe, + decision_timeframe=decision_timeframe, + ) if dataset is not None: self.session.adopt_dataset(dataset) # Mirror the background loader's post-load row transition, with the diff --git a/dashboard/backend/infrastructure/market_data/alpaca_bars.py b/dashboard/backend/infrastructure/market_data/alpaca_bars.py index be1d354d..5eae6cab 100644 --- a/dashboard/backend/infrastructure/market_data/alpaca_bars.py +++ b/dashboard/backend/infrastructure/market_data/alpaca_bars.py @@ -25,6 +25,9 @@ import pandas as pd from dashboard.backend.paths import CREDENTIALS_DIR +from dashboard.backend.infrastructure.market_data.frequency import ( + normalize_bar_timeframe, +) # Basic plan may query SIP historical bars, but not the most recent window. # Docs: https://docs.alpaca.markets/docs/market-data-faq @@ -236,15 +239,26 @@ def feed_provenance(bars: Dict[str, pd.DataFrame]) -> Optional[Dict[str, Any]]: priced from Yahoo, which never touches Alpaca) so callers can tell "not applicable" from "IEX fallback". """ + feeds = set() + sip_fallback_to_iex = False + end_clamped = False for frame in bars.values(): attrs = getattr(frame, "attrs", None) or {} if FRAME_ATTR_FEED in attrs: - return { - "market_data_feed": attrs.get(FRAME_ATTR_FEED), - "sip_fallback_to_iex": bool(attrs.get(FRAME_ATTR_SIP_FALLBACK)), - "end_clamped": bool(attrs.get(FRAME_ATTR_END_CLAMPED)), - } - return None + feed = str(attrs.get(FRAME_ATTR_FEED) or "").strip().lower() + if feed: + feeds.add(feed) + sip_fallback_to_iex = sip_fallback_to_iex or bool( + attrs.get(FRAME_ATTR_SIP_FALLBACK) + ) + end_clamped = end_clamped or bool(attrs.get(FRAME_ATTR_END_CLAMPED)) + if not feeds: + return None + return { + "market_data_feed": next(iter(feeds)) if len(feeds) == 1 else "mixed", + "sip_fallback_to_iex": sip_fallback_to_iex, + "end_clamped": end_clamped, + } def _apply_default_timeout(client: Any) -> None: @@ -282,10 +296,22 @@ def _request_with_default_timeout(*args, **kwargs): class AlpacaDataLoader: - """Fetches historical hourly bars from Alpaca API.""" + """Fetches historical bars from Alpaca API at a configured resolution. - def __init__(self, api_key: Optional[str] = None, secret_key: Optional[str] = None): - """Initialize with Alpaca credentials.""" + ``60m`` remains the constructor default for backward compatibility with + the existing hourly backtest and baseline callers. Minute-data callers + can pass ``source_timeframe="5m"`` or call + :meth:`configure_source_timeframe` before fetching. + """ + + def __init__( + self, + api_key: Optional[str] = None, + secret_key: Optional[str] = None, + source_timeframe: str = "60m", + ): + """Initialize with Alpaca credentials and a source bar timeframe.""" + self.configure_source_timeframe(source_timeframe) if not api_key or not secret_key: creds = self._load_credentials() api_key = creds.get("api_key") @@ -299,12 +325,13 @@ def __init__(self, api_key: Optional[str] = None, secret_key: Optional[str] = No from alpaca.data.enums import DataFeed from alpaca.data.historical import StockHistoricalDataClient from alpaca.data.requests import StockBarsRequest - from alpaca.data.timeframe import TimeFrame + from alpaca.data.timeframe import TimeFrame, TimeFrameUnit self.client = StockHistoricalDataClient(self.api_key, self.secret_key) _apply_default_timeout(self.client) self.StockBarsRequest = StockBarsRequest self.TimeFrame = TimeFrame + self.TimeFrameUnit = TimeFrameUnit self.DataFeed = DataFeed self.last_fetch: Optional[Dict[str, Any]] = None print("✅ Alpaca credentials loaded") @@ -315,6 +342,25 @@ def __init__(self, api_key: Optional[str] = None, secret_key: Optional[str] = No "alpaca-py is not installed (pip install alpaca-py)" ) from e + def configure_source_timeframe(self, source_timeframe: str) -> None: + """Set the source resolution used by the next ``fetch_bars`` call.""" + self.source_timeframe = normalize_bar_timeframe(source_timeframe) + + def _alpaca_timeframe(self): + """Translate the canonical application timeframe to alpaca-py.""" + if self.source_timeframe == "1m": + return self.TimeFrame.Minute + if self.source_timeframe == "5m": + return self.TimeFrame(5, self.TimeFrameUnit.Minute) + if self.source_timeframe == "60m": + return self.TimeFrame.Hour + # ``configure_source_timeframe`` validates this field. Keep the + # guard explicit so a malformed test double cannot silently request a + # different resolution from the provider. + raise ValueError( + f"Unsupported Alpaca source timeframe: {self.source_timeframe!r}" + ) + def _resolve_data_feed(self): return resolve_alpaca_data_feed(self.DataFeed) @@ -366,6 +412,7 @@ def _record_fetch( ) -> None: self.last_fetch = { "feed": getattr(feed, "value", str(feed)), + "source_timeframe": self.source_timeframe, "requested_end": requested_end, "effective_end": effective_end, "sip_fallback_to_iex": sip_fallback_to_iex, @@ -416,18 +463,37 @@ def _bars_to_frames(self, bars, symbols: List[str]) -> Dict[str, pd.DataFrame]: for symbol in symbols: if symbol in bars.df.index.get_level_values(0): df = bars.df.xs(symbol).reset_index() - df = df[["timestamp", "open", "high", "low", "close", "volume"]].copy() + columns = [ + "timestamp", + "open", + "high", + "low", + "close", + "volume", + ] + # Alpaca includes these fields for stock bars. Keep them when + # present so the domain aggregator can calculate volume-aware + # VWAP, while retaining the historical OHLCV shape for test + # doubles and older SDK responses. + optional_columns = [ + column + for column in ("trade_count", "vwap") + if column in df.columns + ] + df = df[columns + optional_columns].copy() df["timestamp"] = pd.to_datetime(df["timestamp"]) df.set_index("timestamp", inplace=True) data[symbol] = df.sort_index() - print(f" ✅ {symbol}: {len(df)} hourly bars") + print( + f" ✅ {symbol}: {len(df)} {self.source_timeframe} bars" + ) else: print(f" ⚠️ {symbol}: No data available") return data def fetch_bars(self, symbols: List[str], start: str, end: str) -> Dict[str, pd.DataFrame]: """ - Fetch hourly OHLCV data from Alpaca API. + Fetch OHLCV data from Alpaca API at ``source_timeframe``. Args: symbols: List of stock symbols @@ -451,15 +517,16 @@ def fetch_bars(self, symbols: List[str], start: str, end: str) -> Dict[str, pd.D print(f"\n📊 Fetching {len(symbols)} symbols from {start} to {end}...") feed = self._resolve_data_feed() + alpaca_timeframe = self._alpaca_timeframe() effective_end, end_clamped = self._effective_end(end, feed, start) print( - f" Timeframe: Hourly (1h) feed={feed.value} " + f" Timeframe: {self.source_timeframe} feed={feed.value} " f"end={effective_end} with forward-filled price cache\n" ) request = self.StockBarsRequest( symbol_or_symbols=symbols, - timeframe=self.TimeFrame.Hour, + timeframe=alpaca_timeframe, start=start, end=effective_end, feed=feed, @@ -501,7 +568,7 @@ def fetch_bars(self, symbols: List[str], start: str, end: str) -> Dict[str, pd.D try: retry = self.StockBarsRequest( symbol_or_symbols=symbols, - timeframe=self.TimeFrame.Hour, + timeframe=alpaca_timeframe, start=start, end=end, feed=self.DataFeed.IEX, diff --git a/dashboard/backend/infrastructure/market_data/frequency.py b/dashboard/backend/infrastructure/market_data/frequency.py new file mode 100644 index 00000000..32989356 --- /dev/null +++ b/dashboard/backend/infrastructure/market_data/frequency.py @@ -0,0 +1,212 @@ +"""Explicit time-frequency contracts for market-data and trading loops. + +The backtest historically used one ``timeframe`` value for several different +concepts. Minute-data support needs those concepts to be explicit: + +* ``source_timeframe``: the resolution fetched from a provider; +* ``decision_timeframe`` / ``decision_frequency``: the completed bar and + cadence supplied to a strategy; +* ``execution_timeframe``: the resolution used to model a fill; and +* ``valuation_frequency``: the resolution used for mark-to-market updates. + +This module is deliberately provider-neutral. It validates configuration but +does not aggregate bars; aggregation belongs to the backtesting domain layer. +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +class FrequencyConfigError(ValueError): + """Raised when a frequency value is unknown or internally inconsistent.""" + + +SUPPORTED_BAR_TIMEFRAMES = ("1m", "5m", "60m") + +_BAR_TIMEFRAME_ALIASES = { + "1m": "1m", + "1min": "1m", + "1minute": "1m", + "minute": "1m", + "min": "1m", + "5m": "5m", + "5min": "5m", + "5minute": "5m", + "60m": "60m", + "60min": "60m", + "60minute": "60m", + "1h": "60m", + "1hour": "60m", + "hour": "60m", + "hourly": "60m", +} + +_DECISION_FREQUENCY_ALIASES = { + "1h": "1h", + "60m": "1h", + "60min": "1h", + "1hour": "1h", + "hour": "1h", + "hourly": "1h", +} + + +def _clean(value: str, field_name: str) -> str: + if value is None: + raise FrequencyConfigError(f"{field_name} must be non-empty") + cleaned = str(value).strip().lower() + if not cleaned: + raise FrequencyConfigError(f"{field_name} must be non-empty") + return cleaned + + +def normalize_bar_timeframe(value: str) -> str: + """Return the canonical bar timeframe used by the application.""" + cleaned = _clean(value, "bar timeframe") + try: + return _BAR_TIMEFRAME_ALIASES[cleaned] + except KeyError as exc: + allowed = ", ".join(SUPPORTED_BAR_TIMEFRAMES) + raise FrequencyConfigError( + f"Unsupported bar timeframe {value!r}; expected one of {allowed}" + ) from exc + + +def normalize_decision_frequency(value: str) -> str: + """Return the canonical strategy decision cadence.""" + cleaned = _clean(value, "decision frequency") + try: + return _DECISION_FREQUENCY_ALIASES[cleaned] + except KeyError as exc: + raise FrequencyConfigError( + f"Unsupported decision frequency {value!r}; expected '1h'" + ) from exc + + +def timeframe_minutes(value: str) -> int: + """Return the duration, in minutes, of a canonical bar timeframe.""" + canonical = normalize_bar_timeframe(value) + return {"1m": 1, "5m": 5, "60m": 60}[canonical] + + +def decision_frequency_minutes(value: str) -> int: + """Return the duration, in minutes, of a canonical decision cadence.""" + return {"1h": 60}[normalize_decision_frequency(value)] + + +def verify_source_timeframe( + requested: str, + reported: str, + *, + evidence: str, +) -> str: + """Return the reported canonical timeframe or fail on provider drift. + + A run must not be labelled as minute-sourced when a provider ignored the + requested resolution. ``evidence`` names the runtime signal in the error + (for example ``configured`` or ``fetch``); it is not user configuration. + """ + expected = normalize_bar_timeframe(requested) + actual = normalize_bar_timeframe(reported) + if actual != expected: + raise FrequencyConfigError( + f"{evidence} source timeframe mismatch: requested {expected}, " + f"reported {actual}" + ) + return actual + + +@dataclass(frozen=True) +class TradingFrequency: + """Validated frequency contract for one market profile. + + The default remains the legacy all-hourly configuration. Use + :meth:`minute_source_hourly_decisions` for the Phase 0/1 target contract. + """ + + source_timeframe: str = "60m" + decision_timeframe: str = "60m" + decision_frequency: str = "1h" + execution_timeframe: str = "60m" + valuation_frequency: str = "60m" + + def __post_init__(self) -> None: + source = normalize_bar_timeframe(self.source_timeframe) + decision = normalize_bar_timeframe(self.decision_timeframe) + decision_frequency = normalize_decision_frequency(self.decision_frequency) + execution = normalize_bar_timeframe(self.execution_timeframe) + valuation = normalize_bar_timeframe(self.valuation_frequency) + + if timeframe_minutes(source) > timeframe_minutes(decision): + raise FrequencyConfigError( + "source_timeframe cannot be coarser than decision_timeframe" + ) + if decision_frequency_minutes(decision_frequency) != timeframe_minutes(decision): + raise FrequencyConfigError( + "decision_frequency must match decision_timeframe" + ) + + object.__setattr__(self, "source_timeframe", source) + object.__setattr__(self, "decision_timeframe", decision) + object.__setattr__(self, "decision_frequency", decision_frequency) + object.__setattr__(self, "execution_timeframe", execution) + object.__setattr__(self, "valuation_frequency", valuation) + + @classmethod + def minute_source_hourly_decisions(cls) -> "TradingFrequency": + """Return the target 5m-source / 1h-decision Phase 0/1 contract.""" + return cls( + source_timeframe="5m", + decision_timeframe="60m", + decision_frequency="1h", + execution_timeframe="5m", + valuation_frequency="5m", + ) + + def to_metadata(self) -> dict[str, str]: + """Return a stable JSON-safe representation for run metadata.""" + return { + "source_timeframe": self.source_timeframe, + "decision_timeframe": self.decision_timeframe, + "decision_frequency": self.decision_frequency, + "execution_timeframe": self.execution_timeframe, + "valuation_frequency": self.valuation_frequency, + } + + +def build_verified_intraday_contract( + *, + source_timeframe: str, + decision_timeframe: str, + decision_frequency: str, +) -> dict[str, str]: + """Build the fixed runtime contract for finer-source hourly backtests. + + Execution and valuation deliberately inherit the source resolution. The + aggregation and fill policy are constants, not strategy options. Returning + ``verification_status=verified`` attests that these invariants were checked + from the runtime's effective timeframes before result persistence. + """ + source = normalize_bar_timeframe(source_timeframe) + decision = normalize_bar_timeframe(decision_timeframe) + if timeframe_minutes(source) >= timeframe_minutes(decision): + raise FrequencyConfigError( + "Verified intraday contracts require source_timeframe to be finer " + "than decision_timeframe" + ) + contract = TradingFrequency( + source_timeframe=source, + decision_timeframe=decision, + decision_frequency=decision_frequency, + execution_timeframe=source, + valuation_frequency=source, + ).to_metadata() + contract.update( + { + "aggregation": "session_anchored_completed_bars", + "fill_policy": "next_source_bar_open", + "verification_status": "verified", + } + ) + return contract diff --git a/dashboard/backend/infrastructure/market_data/profiles.py b/dashboard/backend/infrastructure/market_data/profiles.py index 668f9d9c..f8d6cf9b 100644 --- a/dashboard/backend/infrastructure/market_data/profiles.py +++ b/dashboard/backend/infrastructure/market_data/profiles.py @@ -6,6 +6,11 @@ import math from dashboard.backend.infrastructure.llm.validator import DJIA_30 +from dashboard.backend.infrastructure.market_data.frequency import ( + TradingFrequency, + normalize_bar_timeframe, + normalize_decision_frequency, +) ALPACA = "alpaca" @@ -137,6 +142,44 @@ class MarketProfile: t_plus_one_enabled: bool lot_size: int = 1 transaction_cost_profile: TransactionCostProfile | None = None + # ``timeframe`` remains the backward-compatible decision-bar field. The + # fields below make the source, decision, execution and valuation clocks + # explicit for minute-data work without breaking existing callers. + source_timeframe: str = "60m" + decision_frequency: str = "1h" + execution_timeframe: str = "60m" + valuation_frequency: str = "60m" + + def __post_init__(self) -> None: + decision_timeframe = normalize_bar_timeframe(self.timeframe) + contract = TradingFrequency( + source_timeframe=self.source_timeframe, + decision_timeframe=decision_timeframe, + decision_frequency=self.decision_frequency, + execution_timeframe=self.execution_timeframe, + valuation_frequency=self.valuation_frequency, + ) + object.__setattr__(self, "timeframe", decision_timeframe) + object.__setattr__(self, "source_timeframe", contract.source_timeframe) + object.__setattr__(self, "decision_frequency", normalize_decision_frequency(contract.decision_frequency)) + object.__setattr__(self, "execution_timeframe", contract.execution_timeframe) + object.__setattr__(self, "valuation_frequency", contract.valuation_frequency) + + @property + def decision_timeframe(self) -> str: + """Canonical name for the legacy ``timeframe`` decision-bar field.""" + return self.timeframe + + @property + def frequency_contract(self) -> TradingFrequency: + """Return the validated frequency contract for this profile.""" + return TradingFrequency( + source_timeframe=self.source_timeframe, + decision_timeframe=self.decision_timeframe, + decision_frequency=self.decision_frequency, + execution_timeframe=self.execution_timeframe, + valuation_frequency=self.valuation_frequency, + ) @property def decision_source(self) -> str: @@ -169,6 +212,9 @@ def llm_enabled(self) -> bool: native_currency="USD", reporting_currency="USD", t_plus_one_enabled=False, + source_timeframe="5m", + execution_timeframe="5m", + valuation_frequency="5m", ), (VNPY_SIMULATION, "djia_30"): MarketProfile( data_source=VNPY_SIMULATION, diff --git a/dashboard/backend/infrastructure/market_data/provider.py b/dashboard/backend/infrastructure/market_data/provider.py index dcfe5310..25ef8868 100644 --- a/dashboard/backend/infrastructure/market_data/provider.py +++ b/dashboard/backend/infrastructure/market_data/provider.py @@ -9,6 +9,7 @@ import pandas as pd from .alpaca_bars import AlpacaDataLoader +from .frequency import normalize_bar_timeframe from .profiles import ALPACA, IFIND_ASHARE, VNPY_SIMULATION @@ -18,7 +19,7 @@ class MarketDataProvider(Protocol): - """Normalized hourly market-data input consumed by backtests.""" + """Normalized market-data input consumed by backtests.""" def fetch_bars( self, @@ -97,21 +98,49 @@ def ensure_market_data_source_available(data_source: str) -> None: def create_market_data_provider( data_source: str = ALPACA, universe: str | None = None, + *, + source_timeframe: str | None = None, ) -> MarketDataProvider: - """Create a provider while keeping optional vn.py imports isolated.""" + """Create a provider while keeping optional imports isolated. + + ``source_timeframe`` is intentionally explicit. Omitting it preserves + the legacy behavior of requesting the profile's decision timeframe, so an + existing hourly backtest cannot accidentally start making decisions on + every minute bar. Phase 2 will pass ``profile.source_timeframe`` after + the minute-to-decision-bar aggregation path is connected. + """ from .profiles import get_market_profile profile = get_market_profile(data_source, universe) ensure_market_data_source_available(data_source) + requested_timeframe = normalize_bar_timeframe( + profile.timeframe if source_timeframe is None else source_timeframe + ) if data_source == ALPACA: - return AlpacaDataLoader() + loader = AlpacaDataLoader() + # Configure after construction to keep compatibility with lightweight + # test doubles and legacy integrations that replace AlpacaDataLoader + # with a zero-argument class. + configure_market_data_provider(loader, requested_timeframe) + return loader if data_source == IFIND_ASHARE: + if requested_timeframe != profile.timeframe: + raise ValueError( + "iFinD A-share provider currently supports only its profile " + f"timeframe {profile.timeframe!r}" + ) from .ifind_ashare import IFindAshareProvider return IFindAshareProvider(profile=profile) + if requested_timeframe != profile.timeframe: + raise ValueError( + "vn.py simulation provider currently supports only its profile " + f"timeframe {profile.timeframe!r}" + ) + try: from .vnpy_simulation import VnpySimulationProvider except ModuleNotFoundError as exc: @@ -123,3 +152,28 @@ def create_market_data_provider( raise return VnpySimulationProvider() + + +def configure_market_data_provider( + provider: MarketDataProvider, + source_timeframe: str, +) -> MarketDataProvider: + """Configure a provider's source timeframe when it supports the feature. + + This small compatibility boundary lets the factory configure the real + Alpaca loader without requiring every existing injected provider or test + double to change its constructor signature. + """ + canonical = normalize_bar_timeframe(source_timeframe) + configure = getattr(provider, "configure_source_timeframe", None) + if callable(configure): + configure(canonical) + else: + # A replacement provider may not expose the optional capability. Keep + # the attribute visible for diagnostics while leaving its own fetch + # implementation untouched. + try: + setattr(provider, "source_timeframe", canonical) + except (AttributeError, TypeError): + pass + return provider diff --git a/dashboard/backend/tests/backtesting/test_bar_aggregation.py b/dashboard/backend/tests/backtesting/test_bar_aggregation.py new file mode 100644 index 00000000..2037f4b0 --- /dev/null +++ b/dashboard/backend/tests/backtesting/test_bar_aggregation.py @@ -0,0 +1,208 @@ +from datetime import datetime + +import pandas as pd +import pytz + +from dashboard.backend.domain.backtesting.bar_aggregation import ( + aggregate_bars, + summarize_aggregation_quality, +) + + +def _bars(timestamps): + prices = list(range(100, 100 + len(timestamps))) + return pd.DataFrame( + { + "open": prices, + "high": [price + 1 for price in prices], + "low": [price - 1 for price in prices], + "close": prices, + "volume": [10] * len(prices), + "vwap": [price + 0.5 for price in prices], + }, + index=pd.DatetimeIndex(timestamps), + ) + + +def test_us_bars_are_anchored_to_0930_and_labeled_at_bucket_end(): + eastern = pytz.timezone("US/Eastern") + timestamps = pd.date_range( + eastern.localize(datetime(2026, 3, 2, 9, 30)), + eastern.localize(datetime(2026, 3, 2, 15, 55)), + freq="5min", + ) + + result = aggregate_bars( + _bars(timestamps), + source_timeframe="5m", + decision_timeframe="60m", + market="US", + timezone="US/Eastern", + ) + + assert list(result.index[:2]) == [ + pd.Timestamp("2026-03-02 15:30:00", tz="UTC"), + pd.Timestamp("2026-03-02 16:30:00", tz="UTC"), + ] + assert result.iloc[0]["open"] == 100 + assert result.iloc[0]["close"] == 111 + assert result.iloc[0]["source_bar_count"] == 12 + assert result.iloc[0]["expected_source_bars"] == 12 + assert bool(result.iloc[0]["is_complete"]) is True + assert result.iloc[0]["vwap"] == 106.0 + # The final 15:30-16:00 bucket is complete, but its 16:00 label has no + # following source bar and is filtered from the execution plan by the engine. + assert result.index[-1] == pd.Timestamp("2026-03-02 21:00:00", tz="UTC") + + +def test_cn_lunch_break_does_not_create_a_cross_session_bucket(): + shanghai = pytz.timezone("Asia/Shanghai") + morning = pd.date_range( + shanghai.localize(datetime(2026, 3, 2, 9, 30)), + shanghai.localize(datetime(2026, 3, 2, 11, 25)), + freq="5min", + ) + afternoon = pd.date_range( + shanghai.localize(datetime(2026, 3, 2, 13, 0)), + shanghai.localize(datetime(2026, 3, 2, 14, 55)), + freq="5min", + ) + + result = aggregate_bars( + _bars(morning.append(afternoon)), + source_timeframe="5m", + decision_timeframe="60m", + market="CN", + timezone="Asia/Shanghai", + ) + + assert [timestamp.tz_convert("Asia/Shanghai").strftime("%H:%M") for timestamp in result.index] == [ + "10:30", + "11:30", + "14:00", + "15:00", + ] + assert all(result["source_bar_count"] == 12) + + +def test_missing_source_bar_is_visible_in_quality_columns(): + 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", + ).delete(3) + + result = aggregate_bars( + _bars(timestamps), + source_timeframe="5m", + decision_timeframe="60m", + market="US", + timezone="US/Eastern", + ) + + assert result.iloc[0]["source_bar_count"] == 11 + assert result.iloc[0]["expected_source_bars"] == 12 + assert bool(result.iloc[0]["is_complete"]) is False + assert bool(result.iloc[0]["has_gap"]) is True + + +def test_duplicate_cannot_hide_a_missing_source_slot(): + eastern = pytz.timezone("US/Eastern") + timestamps = list( + pd.date_range( + eastern.localize(datetime(2026, 3, 2, 9, 30)), + eastern.localize(datetime(2026, 3, 2, 10, 25)), + freq="5min", + ) + ) + timestamps.remove(eastern.localize(datetime(2026, 3, 2, 9, 45))) + timestamps.append(eastern.localize(datetime(2026, 3, 2, 9, 40))) + + result = aggregate_bars( + _bars(timestamps), + source_timeframe="5m", + decision_timeframe="60m", + market="US", + timezone="US/Eastern", + ) + + assert result.iloc[0]["source_bar_count"] == 12 + assert result.iloc[0]["missing_source_bars"] == 1 + assert result.iloc[0]["duplicate_source_bars"] == 1 + assert bool(result.iloc[0]["is_complete"]) is False + + +def test_off_grid_or_invalid_source_bar_makes_bucket_incomplete(): + eastern = pytz.timezone("US/Eastern") + timestamps = list( + pd.date_range( + eastern.localize(datetime(2026, 3, 2, 9, 30)), + eastern.localize(datetime(2026, 3, 2, 10, 25)), + freq="5min", + ) + ) + timestamps[3] = eastern.localize(datetime(2026, 3, 2, 9, 47)) + bars = _bars(timestamps) + bars.loc[timestamps[5], "close"] = float("nan") + + result = aggregate_bars( + bars, + source_timeframe="5m", + decision_timeframe="60m", + market="US", + timezone="US/Eastern", + ) + + assert result.iloc[0]["missing_source_bars"] == 1 + assert result.iloc[0]["off_grid_source_bars"] == 1 + assert result.iloc[0]["invalid_source_bars"] == 1 + assert bool(result.iloc[0]["is_complete"]) is False + + +def test_decision_bar_does_not_include_source_bar_at_its_right_edge(): + eastern = pytz.timezone("US/Eastern") + timestamps = pd.date_range( + eastern.localize(datetime(2026, 3, 2, 9, 30)), + eastern.localize(datetime(2026, 3, 2, 10, 30)), + freq="5min", + ) + bars = _bars(timestamps) + bars.loc[timestamps[-1], ["open", "high", "low", "close"]] = 10_000 + + result = aggregate_bars( + bars, + source_timeframe="5m", + decision_timeframe="60m", + market="US", + timezone="US/Eastern", + ) + + first_decision = result.loc[pd.Timestamp("2026-03-02 15:30:00", tz="UTC")] + assert first_decision["close"] == 111 + assert first_decision["high"] == 112 + + +def test_quality_summary_counts_usable_and_rejected_buckets(): + eastern = pytz.timezone("US/Eastern") + timestamps = pd.date_range( + eastern.localize(datetime(2026, 3, 2, 9, 30)), + eastern.localize(datetime(2026, 3, 2, 11, 25)), + freq="5min", + ).delete(15) + aggregated = aggregate_bars( + _bars(timestamps), + source_timeframe="5m", + decision_timeframe="60m", + market="US", + timezone="US/Eastern", + ) + + summary = summarize_aggregation_quality({"AAPL": aggregated}) + + assert summary["policy"] == "drop_incomplete_decision_bars" + assert summary["total_decision_bars"] == 2 + assert summary["usable_decision_bars"] == 1 + assert summary["dropped_decision_bars"] == 1 + assert summary["missing_source_bars"] == 1 + assert summary["symbols"]["AAPL"]["dropped_decision_bars"] == 1 diff --git a/dashboard/backend/tests/backtesting/test_engine_minute_source.py b/dashboard/backend/tests/backtesting/test_engine_minute_source.py new file mode 100644 index 00000000..47a365c7 --- /dev/null +++ b/dashboard/backend/tests/backtesting/test_engine_minute_source.py @@ -0,0 +1,170 @@ +from datetime import datetime, timedelta + +import pandas as pd +import pytz +import pytest + +from dashboard.backend.domain.backtesting.engine import HourlyBacktester +from dashboard.backend.domain.backtesting import engine as engine_mod +from dashboard.backend.infrastructure.market_data.alpaca_bars import ( + FRAME_ATTR_END_CLAMPED, + FRAME_ATTR_FEED, + FRAME_ATTR_SIP_FALLBACK, + MarketDataUnavailableError, +) + + +class _MinuteLoader: + def __init__(self, bars): + self.bars = bars + self.source_timeframe = None + + def configure_source_timeframe(self, value): + self.source_timeframe = value + + def fetch_bars(self, symbols, start_date, end_date): + return {symbol: self.bars[symbol] for symbol in symbols} + + +class _DB: + def __init__(self): + self.runs = [] + self.equity_points = [] + self.trades = [] + + def insert_run(self, **kwargs): + self.runs.append(kwargs) + + def insert_equity_points(self, run_id, points): + self.equity_points.append((run_id, list(points))) + + def insert_trades(self, run_id, trades): + self.trades.append((run_id, list(trades))) + + def insert_decisions(self, run_id, decisions): + pass + + +def _make_minute_bars(): + eastern = pytz.timezone("US/Eastern") + timestamps = [] + day = datetime(2026, 3, 2) + while len(timestamps) < 10 * 78: + if day.weekday() < 5: + timestamps.extend( + pd.date_range( + eastern.localize(datetime(day.year, day.month, day.day, 9, 30)), + eastern.localize(datetime(day.year, day.month, day.day, 15, 55)), + freq="5min", + ) + ) + day += timedelta(days=1) + timestamps = timestamps[: 10 * 78] + prices = [100 + index * 0.01 for index in range(len(timestamps))] + frame = pd.DataFrame( + { + "open": [price + 0.25 for price in prices], + "high": [price + 0.5 for price in prices], + "low": [price - 0.5 for price in prices], + "close": prices, + "volume": [1000] * len(prices), + }, + index=pd.DatetimeIndex(timestamps), + ) + frame.attrs[FRAME_ATTR_FEED] = "sip" + frame.attrs[FRAME_ATTR_SIP_FALLBACK] = False + frame.attrs[FRAME_ATTR_END_CLAMPED] = True + return {"AAPL": frame} + + +def test_minute_source_keeps_hourly_decisions_and_5m_execution(monkeypatch): + loader = _MinuteLoader(_make_minute_bars()) + + def factory(data_source="alpaca", universe=None, *, source_timeframe=None): + loader.configure_source_timeframe(source_timeframe) + return loader + + fake_db = _DB() + monkeypatch.setattr(engine_mod, "create_market_data_provider", factory) + monkeypatch.setattr(engine_mod, "db", fake_db) + decisions = [] + + def buy_once(self, state): + decisions.append(state["timestamp"]) + if not self.positions: + return {"actions": [{"symbol": "AAPL", "action": "buy", "shares": 1}]} + return {"actions": []} + + monkeypatch.setattr( + "dashboard.backend.domain.backtesting.portfolio_manager.PortfolioManager.make_trading_decision", + buy_once, + ) + + backtester = HourlyBacktester( + "2026-03-02", + "2026-03-13", + use_llm=False, + symbols=["AAPL"], + ) + backtester.load_data() + assert backtester.source_timeframe == "5m" + assert backtester.intraday_mode is True + assert len(backtester.source_data["AAPL"]) == 780 + assert len(backtester.all_data["AAPL"]) == 70 + + 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 + assert len(equity_curve) == 780 + assert run_id.startswith("agent_") + + trade = fake_db.trades[0][1][0] + assert trade["timestamp"] == "2026-03-02T10:30:00-05:00" + assert trade["price"] == 100.37 + + frequency = fake_db.runs[0]["metadata"]["frequency_contract"] + assert frequency["source_timeframe"] == "5m" + assert frequency["decision_frequency"] == "1h" + assert frequency["fill_policy"] == "next_source_bar_open" + assert frequency["verification_status"] == "verified" + quality = fake_db.runs[0]["metadata"]["market_data_quality"] + assert quality["policy"] == "drop_incomplete_decision_bars" + assert quality["total_decision_bars"] == 70 + assert quality["usable_decision_bars"] == 70 + assert quality["dropped_decision_bars"] == 0 + metadata = fake_db.runs[0]["metadata"] + assert metadata["market_data_feed"] == "sip" + assert metadata["sip_fallback_to_iex"] is False + assert metadata["end_clamped"] is True + + +def test_direct_engine_rejects_provider_frequency_drift(monkeypatch): + class _IgnoringLoader(_MinuteLoader): + def __init__(self, bars): + super().__init__(bars) + self.source_timeframe = "60m" + + def configure_source_timeframe(self, value): + pass + + loader = _IgnoringLoader(_make_minute_bars()) + + def factory(data_source="alpaca", universe=None, *, source_timeframe=None): + return loader + + monkeypatch.setattr(engine_mod, "create_market_data_provider", factory) + backtester = HourlyBacktester( + "2026-03-02", + "2026-03-13", + use_llm=False, + symbols=["AAPL"], + ) + + with pytest.raises( + MarketDataUnavailableError, + match="configured source timeframe mismatch: requested 5m, reported 60m", + ): + backtester.load_data() diff --git a/dashboard/backend/tests/infrastructure/market_data/test_alpaca_bars.py b/dashboard/backend/tests/infrastructure/market_data/test_alpaca_bars.py index 33533d50..d29a0359 100644 --- a/dashboard/backend/tests/infrastructure/market_data/test_alpaca_bars.py +++ b/dashboard/backend/tests/infrastructure/market_data/test_alpaca_bars.py @@ -12,6 +12,7 @@ from alpaca.data.timeframe import TimeFrame from dashboard.backend.infrastructure.market_data.alpaca_bars import ( + FRAME_ATTR_END_CLAMPED, FRAME_ATTR_FEED, FRAME_ATTR_SIP_FALLBACK, AlpacaDataLoader, @@ -333,6 +334,23 @@ def test_feed_provenance_reads_frame_stamps(fake_alpaca): assert feed_provenance({"AAPL": pd.DataFrame()}) is None +def test_feed_provenance_marks_mixed_batched_results_conservatively(): + sip = pd.DataFrame() + sip.attrs[FRAME_ATTR_FEED] = "sip" + sip.attrs[FRAME_ATTR_SIP_FALLBACK] = False + sip.attrs[FRAME_ATTR_END_CLAMPED] = True + iex = pd.DataFrame() + iex.attrs[FRAME_ATTR_FEED] = "iex" + iex.attrs[FRAME_ATTR_SIP_FALLBACK] = True + iex.attrs[FRAME_ATTR_END_CLAMPED] = False + + assert feed_provenance({"AAPL": sip, "MSFT": iex}) == { + "market_data_feed": "mixed", + "sip_fallback_to_iex": True, + "end_clamped": True, + } + + def test_clamped_fetch_is_marked_in_provenance(fake_alpaca, monkeypatch): from datetime import datetime, timezone diff --git a/dashboard/backend/tests/infrastructure/market_data/test_alpaca_frequency.py b/dashboard/backend/tests/infrastructure/market_data/test_alpaca_frequency.py new file mode 100644 index 00000000..72aba3db --- /dev/null +++ b/dashboard/backend/tests/infrastructure/market_data/test_alpaca_frequency.py @@ -0,0 +1,131 @@ +"""Tests for configurable Alpaca source bar timeframes.""" + +from __future__ import annotations + +import pandas as pd + +from alpaca.data.timeframe import TimeFrame, TimeFrameUnit + +from dashboard.backend.infrastructure.market_data.alpaca_bars import AlpacaDataLoader +from dashboard.backend.infrastructure.market_data import provider + + +def _bars_df(symbol: str = "AAPL"): + index = pd.MultiIndex.from_tuples( + [(symbol, pd.Timestamp("2026-01-02 14:30:00Z"))], + names=["symbol", "timestamp"], + ) + return pd.DataFrame( + {"open": [1.0], "high": [1.1], "low": [0.9], "close": [1.05], "volume": [100]}, + index=index, + ) + + +def test_alpaca_loader_maps_5m_to_sdk_timeframe(monkeypatch): + state = {"requests": []} + + class FakeBars: + df = _bars_df() + + class FakeSession: + def request(self, *args, **kwargs): + raise AssertionError("HTTP session should not be used") + + class FakeClient: + def __init__(self, api_key, secret_key): + self._session = FakeSession() + + def get_stock_bars(self, request): + state["requests"].append(request) + return FakeBars() + + monkeypatch.setattr( + "alpaca.data.historical.StockHistoricalDataClient", + FakeClient, + ) + + loader = AlpacaDataLoader( + api_key="key", + secret_key="secret", + source_timeframe="5m", + ) + result = loader.fetch_bars(["AAPL"], "2026-01-01", "2026-01-03") + + request = state["requests"][0] + assert request.timeframe.value == TimeFrame(5, TimeFrameUnit.Minute).value + assert result["AAPL"].attrs + assert loader.last_fetch["source_timeframe"] == "5m" + + +def test_alpaca_loader_maps_1m_to_sdk_timeframe(monkeypatch): + state = {"requests": []} + + class FakeBars: + df = _bars_df() + + class FakeSession: + def request(self, *args, **kwargs): + raise AssertionError("HTTP session should not be used") + + class FakeClient: + def __init__(self, api_key, secret_key): + self._session = FakeSession() + + def get_stock_bars(self, request): + state["requests"].append(request) + return FakeBars() + + monkeypatch.setattr( + "alpaca.data.historical.StockHistoricalDataClient", + FakeClient, + ) + + loader = AlpacaDataLoader( + api_key="key", + secret_key="secret", + source_timeframe="1min", + ) + loader.fetch_bars(["AAPL"], "2026-01-01", "2026-01-03") + + assert state["requests"][0].timeframe.value == TimeFrame.Minute.value + assert loader.last_fetch["source_timeframe"] == "1m" + + +def test_provider_factory_configures_explicit_source_timeframe(monkeypatch): + created = [] + + class FakeAlpacaLoader: + def __init__(self): + self.configured = None + created.append(self) + + def configure_source_timeframe(self, value): + self.configured = value + + monkeypatch.setattr(provider, "AlpacaDataLoader", FakeAlpacaLoader) + + loader = provider.create_market_data_provider( + provider.ALPACA, + source_timeframe="5Min", + ) + + assert loader is created[0] + assert loader.configured == "5m" + + +def test_provider_factory_default_remains_profile_decision_timeframe(monkeypatch): + created = [] + + class FakeAlpacaLoader: + def __init__(self): + self.configured = None + created.append(self) + + def configure_source_timeframe(self, value): + self.configured = value + + monkeypatch.setattr(provider, "AlpacaDataLoader", FakeAlpacaLoader) + + loader = provider.create_market_data_provider(provider.ALPACA) + + assert loader.configured == "60m" diff --git a/dashboard/backend/tests/infrastructure/market_data/test_frequency.py b/dashboard/backend/tests/infrastructure/market_data/test_frequency.py new file mode 100644 index 00000000..f24b7e63 --- /dev/null +++ b/dashboard/backend/tests/infrastructure/market_data/test_frequency.py @@ -0,0 +1,106 @@ +"""Tests for the Phase 0/1 market-data and trading frequency contract.""" + +from __future__ import annotations + +import pytest + +from dashboard.backend.infrastructure.market_data.frequency import ( + FrequencyConfigError, + TradingFrequency, + build_verified_intraday_contract, + normalize_bar_timeframe, + normalize_decision_frequency, + verify_source_timeframe, +) +from dashboard.backend.infrastructure.market_data.profiles import ( + ALPACA, + get_market_profile, +) + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("1m", "1m"), + ("1Min", "1m"), + ("5MIN", "5m"), + ("hourly", "60m"), + ("1h", "60m"), + ], +) +def test_bar_timeframe_aliases_are_canonical(value, expected): + assert normalize_bar_timeframe(value) == expected + + +def test_decision_frequency_is_canonical(): + assert normalize_decision_frequency(" hourly ") == "1h" + assert normalize_decision_frequency("60m") == "1h" + + +def test_minute_source_hourly_decision_contract(): + contract = TradingFrequency.minute_source_hourly_decisions() + + assert contract.to_metadata() == { + "source_timeframe": "5m", + "decision_timeframe": "60m", + "decision_frequency": "1h", + "execution_timeframe": "5m", + "valuation_frequency": "5m", + } + + +def test_frequency_contract_rejects_coarser_source_than_decision(): + with pytest.raises(FrequencyConfigError, match="coarser"): + TradingFrequency(source_timeframe="60m", decision_timeframe="5m") + + +def test_frequency_contract_rejects_mismatched_decision_cadence(): + with pytest.raises(FrequencyConfigError, match="match"): + TradingFrequency( + source_timeframe="5m", + decision_timeframe="5m", + decision_frequency="1h", + ) + + +def test_runtime_contract_attests_fixed_intraday_execution_policy(): + assert build_verified_intraday_contract( + source_timeframe="5m", + decision_timeframe="60m", + decision_frequency="1h", + ) == { + "source_timeframe": "5m", + "decision_timeframe": "60m", + "decision_frequency": "1h", + "execution_timeframe": "5m", + "valuation_frequency": "5m", + "aggregation": "session_anchored_completed_bars", + "fill_policy": "next_source_bar_open", + "verification_status": "verified", + } + + +def test_runtime_contract_rejects_non_intraday_source(): + with pytest.raises(FrequencyConfigError, match="finer"): + build_verified_intraday_contract( + source_timeframe="60m", + decision_timeframe="60m", + decision_frequency="1h", + ) + + +def test_source_timeframe_verification_rejects_provider_drift(): + with pytest.raises(FrequencyConfigError, match="requested 5m.*reported 60m"): + verify_source_timeframe("5m", "60m", evidence="provider") + + +def test_alpaca_profile_records_minute_source_and_hourly_decisions(): + profile = get_market_profile(ALPACA) + + assert profile.timeframe == "60m" + assert profile.decision_timeframe == "60m" + assert profile.source_timeframe == "5m" + assert profile.decision_frequency == "1h" + assert profile.execution_timeframe == "5m" + assert profile.valuation_frequency == "5m" + assert profile.frequency_contract == TradingFrequency.minute_source_hourly_decisions() diff --git a/dashboard/backend/tests/test_analytics_integration.py b/dashboard/backend/tests/test_analytics_integration.py index 7d117a5c..c2d57d68 100644 --- a/dashboard/backend/tests/test_analytics_integration.py +++ b/dashboard/backend/tests/test_analytics_integration.py @@ -281,7 +281,9 @@ def test_synthetic_acceptance_scenario_has_no_real_credentials(monkeypatch): event_name="backtest_completed", user_id=subject["id"], run_id="synthetic-platform-run", - occurred_at=now - timedelta(hours=1), + # Current-day metrics read raw events before daily rollups exist. + # Keeping this flow on ``now`` avoids a UTC-midnight-only failure. + occurred_at=now, ) instrumentation.emit_resource_event( event_name="model_usage_recorded", @@ -298,7 +300,7 @@ def test_synthetic_acceptance_scenario_has_no_real_credentials(monkeypatch): "output_tokens": 40, "cost_micro_usd": 420_000, }, - occurred_at=now - timedelta(minutes=59), + occurred_at=now, ) instrumentation.emit_resource_event( event_name="credits_settled", @@ -308,7 +310,7 @@ def test_synthetic_acceptance_scenario_has_no_real_credentials(monkeypatch): correlation_id="synthetic-platform-run", billing_mode="platform_credits", properties={"amount_micro": 420, "bucket": "grant"}, - occurred_at=now - timedelta(minutes=58), + occurred_at=now, ) token = users.create_session(admin["id"]) diff --git a/dashboard/backend/tests/test_backtests_router.py b/dashboard/backend/tests/test_backtests_router.py index 135f6d2d..334f166b 100644 --- a/dashboard/backend/tests/test_backtests_router.py +++ b/dashboard/backend/tests/test_backtests_router.py @@ -569,6 +569,57 @@ def test_run_metadata_response_keeps_new_fields_optional_for_legacy_runs(): assert response.order_events_count is None assert response.order_events_truncated is None assert response.llm_execution is None + assert response.frequency_contract is None + assert response.market_data_quality is None + assert response.market_data_feed is None + assert response.sip_fallback_to_iex is None + assert response.end_clamped is None + + +def test_run_metadata_response_exposes_minute_data_contract_and_quality(): + response = bt._run_metadata_response( + _run_record( + { + "data_source": "alpaca", + "frequency_contract": { + "source_timeframe": "5m", + "decision_timeframe": "60m", + "decision_frequency": "1h", + "execution_timeframe": "5m", + "valuation_frequency": "5m", + "aggregation": "session_anchored_completed_bars", + "fill_policy": "next_source_bar_open", + "verification_status": "verified", + }, + "market_data_quality": { + "policy": "drop_incomplete_decision_bars", + "decision_timestamp_min_symbol_coverage": 0.8, + "total_decision_bars": 210, + "usable_decision_bars": 207, + "dropped_decision_bars": 3, + "missing_source_bars": 2, + "duplicate_source_bars": 1, + "off_grid_source_bars": 0, + "invalid_source_bars": 0, + "symbols": {"AAPL": {"dropped_decision_bars": 3}}, + }, + "market_data_feed": "iex", + "sip_fallback_to_iex": True, + "end_clamped": True, + } + ) + ) + + assert response.frequency_contract["source_timeframe"] == "5m" + assert response.frequency_contract["decision_frequency"] == "1h" + assert response.frequency_contract["fill_policy"] == "next_source_bar_open" + assert response.frequency_contract["verification_status"] == "verified" + assert response.market_data_quality["usable_decision_bars"] == 207 + assert response.market_data_quality["dropped_decision_bars"] == 3 + assert "symbols" not in response.market_data_quality + assert response.market_data_feed == "iex" + assert response.sip_fallback_to_iex is True + assert response.end_clamped is True def test_run_metadata_response_exposes_sanitized_llm_execution_evidence(): diff --git a/dashboard/backend/tests/test_external_minute_source.py b/dashboard/backend/tests/test_external_minute_source.py new file mode 100644 index 00000000..c4dcf4d7 --- /dev/null +++ b/dashboard/backend/tests/test_external_minute_source.py @@ -0,0 +1,203 @@ +from datetime import datetime + +import pandas as pd +import pytest + +import dashboard.backend.domain.backtesting.external_run_service as ebs +from dashboard.backend.domain.backtesting import market_data_store as mds +from dashboard.backend.infrastructure.market_data.alpaca_bars import ( + FRAME_ATTR_END_CLAMPED, + FRAME_ATTR_FEED, + FRAME_ATTR_SIP_FALLBACK, +) + + +def _minute_bars(symbols, start, end): + timestamps = pd.date_range( + "2026-04-15 13:30:00+00:00", + "2026-04-15 19:55:00+00:00", + freq="5min", + ) + prices = [100 + index * 0.01 for index in range(len(timestamps))] + frame = pd.DataFrame( + { + "open": [price + 0.25 for price in prices], + "high": [price + 0.5 for price in prices], + "low": [price - 0.5 for price in prices], + "close": prices, + "volume": [1000] * len(prices), + }, + index=timestamps, + ) + frames = {symbol: frame.copy() for symbol in symbols} + for symbol_frame in frames.values(): + symbol_frame.attrs[FRAME_ATTR_FEED] = "iex" + symbol_frame.attrs[FRAME_ATTR_SIP_FALLBACK] = True + symbol_frame.attrs[FRAME_ATTR_END_CLAMPED] = False + return frames + + +class _MinuteLoader: + source_timeframe = "60m" + + def configure_source_timeframe(self, value): + self.source_timeframe = value + + def fetch_bars(self, symbols, start, end): + return _minute_bars(symbols, start, end) + + +@pytest.fixture(autouse=True) +def _isolate_store(monkeypatch): + mds._reset_for_tests() + monkeypatch.setattr(ebs, "AlpacaDataLoader", _MinuteLoader) + monkeypatch.setattr(ebs, "DJIA_30", ["AAPL"]) + yield + mds._reset_for_tests() + + +def test_external_session_serves_hourly_bars_but_fills_and_values_on_5m(): + session = ebs.ExternalBacktestSession( + backtest_id="bt-minute", + session_id="sess-minute", + agent_name="agent-minute", + model_name="test-model", + start_date="2026-04-15", + end_date="2026-04-15", + symbols=["AAPL"], + ) + session.load_market_data() + + assert session.source_timeframe == "5m" + assert session.intraday_mode is True + assert session.total_steps == 6 + assert session.data_quality["total_decision_bars"] == 7 + assert session.data_quality["usable_decision_bars"] == 7 + assert session.data_quality["dropped_decision_bars"] == 0 + assert session.frequency_contract["verification_status"] == "verified" + assert session.market_data_provenance == { + "market_data_feed": "iex", + "sip_fallback_to_iex": True, + "end_clamped": False, + } + assert session.get_current_step()["timestamp"] == "2026-04-15T14:30:00+00:00" + assert session.protocol_bars(session.timestamps[0])["AAPL"]["close"] == pytest.approx(100.11) + + result = session.submit_decisions( + { + "actions": [ + { + "symbol": "AAPL", + "action": "buy", + "confidence": 1.0, + "reasoning": "test buy", + "position_size": 1, + } + ] + } + ) + + assert result["accepted"] is True + assert session.manager.trades[0]["timestamp"] == pd.Timestamp( + datetime(2026, 4, 15, 14, 30), tz="UTC" + ) + assert session.manager.trades[0]["price"] == pytest.approx(100.37) + decision_audit = session.get_decisions()[0] + assert decision_audit["timestamp"] == "2026-04-15T14:30:00+00:00" + assert decision_audit["execution_timestamp"] == "2026-04-15T14:30:00+00:00" + assert decision_audit["actions_executed"] == 1 + # 09:30 through 10:30 ET inclusive: 13 five-minute valuation points. + assert len(session.manager.equity_history) == 13 + + +def test_external_session_does_not_report_fill_without_next_symbol_bar(monkeypatch): + class _MissingExecutionBarLoader(_MinuteLoader): + def fetch_bars(self, symbols, start, end): + bars = _minute_bars(symbols, start, end) + execution_timestamp = pd.Timestamp("2026-04-15 14:30:00+00:00") + bars["AAPL"] = bars["AAPL"].drop(execution_timestamp) + return bars + + monkeypatch.setattr(ebs, "AlpacaDataLoader", _MissingExecutionBarLoader) + monkeypatch.setattr(ebs, "DJIA_30", ["AAPL", "MSFT"]) + session = ebs.ExternalBacktestSession( + backtest_id="bt-missing-fill", + session_id="sess-missing-fill", + agent_name="agent-missing-fill", + model_name="test-model", + start_date="2026-04-15", + end_date="2026-04-15", + symbols=["AAPL", "MSFT"], + ) + 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( + "2026-04-15 14:30:00+00:00" + ) + assert "AAPL" not in session._source_market_data_at( + session.execution_timestamps[0] + ) + + result = session.submit_decisions( + { + "actions": [ + { + "symbol": "AAPL", + "action": "buy", + "confidence": 1.0, + "reasoning": "missing execution bar", + "position_size": 1, + } + ] + } + ) + + assert result["accepted"] is True + assert result["executed_count"] == 0 + assert result["executed"] == [] + assert session.manager.trades == [] + audit = session.get_decisions()[0] + assert audit["actions_executed"] == 0 + assert audit["execution_timestamp"] == "2026-04-15T14:30:00+00:00" + + +def test_final_metrics_expose_minute_contract_without_symbol_quality_details(): + metrics = ebs.build_final_metrics( + { + "total_return": 0.1, + "metadata": { + "frequency_contract": { + "source_timeframe": "5m", + "decision_timeframe": "60m", + "decision_frequency": "1h", + "execution_timeframe": "5m", + "valuation_frequency": "5m", + "aggregation": "session_anchored_completed_bars", + "fill_policy": "next_source_bar_open", + "verification_status": "verified", + }, + "market_data_quality": { + "policy": "drop_incomplete_decision_bars", + "total_decision_bars": 70, + "usable_decision_bars": 69, + "dropped_decision_bars": 1, + "missing_source_bars": 1, + "duplicate_source_bars": 0, + "off_grid_source_bars": 0, + "invalid_source_bars": 0, + "symbols": {"AAPL": {"dropped_decision_bars": 1}}, + }, + "market_data_feed": "iex", + "sip_fallback_to_iex": True, + "end_clamped": False, + }, + } + ) + + assert metrics["frequency_contract"]["source_timeframe"] == "5m" + assert metrics["frequency_contract"]["verification_status"] == "verified" + assert metrics["market_data_quality"]["dropped_decision_bars"] == 1 + assert "symbols" not in metrics["market_data_quality"] + assert metrics["market_data_feed"] == "iex" + assert metrics["sip_fallback_to_iex"] is True + assert metrics["end_clamped"] is False diff --git a/dashboard/backend/tests/test_finalize_split.py b/dashboard/backend/tests/test_finalize_split.py index c21892e5..fd0dd011 100644 --- a/dashboard/backend/tests/test_finalize_split.py +++ b/dashboard/backend/tests/test_finalize_split.py @@ -10,6 +10,11 @@ import dashboard.backend.database as db_module import dashboard.backend.domain.backtesting.external_run_service as ebs from dashboard.backend.domain.backtesting import baseline_worker as bw +from dashboard.backend.infrastructure.market_data.alpaca_bars import ( + FRAME_ATTR_END_CLAMPED, + FRAME_ATTR_FEED, + FRAME_ATTR_SIP_FALLBACK, +) def _synth_bars(symbols, start, end): @@ -23,9 +28,13 @@ def _synth_bars(symbols, start, end): for si, sym in enumerate(sorted(symbols)): n = len(idx) close = 100.0 + si + np.linspace(0, 1.0, n) - data[sym] = pd.DataFrame( + frame = pd.DataFrame( {"open": close, "high": close + 0.5, "low": close - 0.5, "close": close, "volume": 1000.0}, index=idx) + frame.attrs[FRAME_ATTR_FEED] = "iex" + frame.attrs[FRAME_ATTR_SIP_FALLBACK] = True + frame.attrs[FRAME_ATTR_END_CLAMPED] = True + data[sym] = frame return data @@ -88,6 +97,10 @@ def test_final_submit_completes_before_baselines(_isolate): assert last["compare_url"] == f"/compare?run_ids={s.run_id}" # no baseline ids assert s.baseline_run_ids == {} assert _isolate.get_run(s.run_id) is not None + metadata = _isolate.get_run(s.run_id)["metadata"] + assert metadata["market_data_feed"] == "iex" + assert metadata["sip_fallback_to_iex"] is True + assert metadata["end_clamped"] is True _FakeBacktester.gate.set() assert bw.wait_idle(10) # Polled surfaces self-heal once the worker lands. diff --git a/dashboard/backend/tests/test_market_data_store.py b/dashboard/backend/tests/test_market_data_store.py index 9961ecf4..11f4f13f 100644 --- a/dashboard/backend/tests/test_market_data_store.py +++ b/dashboard/backend/tests/test_market_data_store.py @@ -3,12 +3,14 @@ import threading import time +from datetime import datetime import numpy as np import pandas as pd import pytest from dashboard.backend.domain.backtesting import market_data_store as mds +from dashboard.backend.infrastructure.market_data.frequency import FrequencyConfigError def _synth_bars(symbols=("AAPL", "MSFT"), start="2026-04-15", end="2026-04-16"): @@ -221,3 +223,113 @@ def fetch_bars(self, symbols, start, end): # entry it had just built. assert mds.peek(SYMS, *slow) is not None assert mds.peek(SYMS, *fast) is None + + +def test_timestamp_quorum_rounds_up_to_a_real_eighty_percent(): + timestamp = pd.Timestamp("2026-04-15 14:30:00+00:00") + frame = pd.DataFrame( + {"close": [100.0]}, + index=pd.DatetimeIndex([timestamp]), + ) + bars = { + "A": frame.copy(), + "B": frame.copy(), + "C": frame.iloc[0:0].copy(), + } + + assert mds._build_trading_timestamps(bars) == [] + + +def test_minute_dataset_exposes_dropped_bucket_quality(): + eastern = "US/Eastern" + timestamps = pd.date_range( + pd.Timestamp(datetime(2026, 4, 15, 9, 30), tz=eastern), + pd.Timestamp(datetime(2026, 4, 15, 11, 25), tz=eastern), + freq="5min", + ).delete(15) + prices = np.arange(len(timestamps), dtype=float) + 100 + bars = { + "AAPL": pd.DataFrame( + { + "open": prices, + "high": prices + 1, + "low": prices - 1, + "close": prices, + "volume": 1000.0, + }, + index=timestamps, + ) + } + + class _MinuteLoader: + source_timeframe = "60m" + + def configure_source_timeframe(self, value): + self.source_timeframe = value + + def fetch_bars(self, symbols, start, end): + return bars + + dataset = mds.get_dataset( + ["AAPL"], + "2026-04-15", + "2026-04-15", + loader_factory=_MinuteLoader, + source_timeframe="5m", + decision_timeframe="60m", + ) + + assert dataset.total_steps == 1 + assert dataset.data_quality["total_decision_bars"] == 2 + assert dataset.data_quality["usable_decision_bars"] == 1 + assert dataset.data_quality["dropped_decision_bars"] == 1 + assert dataset.data_quality["missing_source_bars"] == 1 + + +def test_minute_dataset_rejects_loader_that_ignores_requested_timeframe(): + class _IgnoringLoader: + source_timeframe = "60m" + + def configure_source_timeframe(self, value): + pass + + def fetch_bars(self, symbols, start, end): + pytest.fail("frequency drift must fail before fetching") + + with pytest.raises( + FrequencyConfigError, + match="configured source timeframe mismatch: requested 5m, reported 60m", + ): + mds.get_dataset( + ["AAPL"], + "2026-04-15", + "2026-04-15", + loader_factory=_IgnoringLoader, + source_timeframe="5m", + decision_timeframe="60m", + ) + + +def test_minute_dataset_rejects_fetch_evidence_with_wrong_timeframe(): + class _MisreportingLoader: + source_timeframe = "60m" + + def configure_source_timeframe(self, value): + self.source_timeframe = value + + def fetch_bars(self, symbols, start, end): + self.last_fetch = {"source_timeframe": "60m"} + return _synth_bars(symbols, start, end) + + with pytest.raises( + FrequencyConfigError, + match="fetch source timeframe mismatch: requested 5m, reported 60m", + ): + mds.get_dataset( + ["AAPL"], + "2026-04-15", + "2026-04-15", + loader_factory=_MisreportingLoader, + source_timeframe="5m", + decision_timeframe="60m", + ) diff --git a/dashboard/backend/tests/test_minute_data_frontend.py b/dashboard/backend/tests/test_minute_data_frontend.py new file mode 100644 index 00000000..09416916 --- /dev/null +++ b/dashboard/backend/tests/test_minute_data_frontend.py @@ -0,0 +1,129 @@ +"""Frontend contracts for five-minute source data and hourly decisions.""" + +from __future__ import annotations + +import json +import shutil +import subprocess + +import pytest + +from dashboard.backend.tests._frontend_source import APP_HTML, FRONTEND, fn_body + + +def _run_formatters(expression: str): + node = shutil.which("node") + if not node: + pytest.skip("node is not installed") + script = "\n".join( + [ + fn_body("function formatBacktestFrequencyContract("), + fn_body("function formatBacktestMarketDataQuality("), + fn_body("function formatBacktestMarketDataProvenance("), + f"console.log(JSON.stringify({expression}));", + ] + ) + result = subprocess.run( + [node, "-e", script], + capture_output=True, + text=True, + check=True, + timeout=30, + ) + return json.loads(result.stdout) + + +def _render_data_source_badge(run: dict): + node = shutil.which("node") + if not node: + pytest.skip("node is not installed") + script = "\n".join( + [ + "const badge = {};", + "const document = { getElementById: () => badge };", + fn_body("function renderBacktestDataSourceBadge("), + f"renderBacktestDataSourceBadge({json.dumps(run)});", + "console.log(JSON.stringify(badge));", + ] + ) + result = subprocess.run( + [node, "-e", script], + capture_output=True, + text=True, + check=True, + timeout=30, + ) + return json.loads(result.stdout) + + +def test_backtest_details_have_frequency_and_quality_rows(): + assert 'id="backtestConfigFrequencyRow"' in APP_HTML + assert 'id="backtestConfigFrequency"' in APP_HTML + assert 'id="backtestConfigDataQualityRow"' in APP_HTML + assert 'id="backtestConfigDataQuality"' in APP_HTML + assert 'id="backtestConfigProvenanceRow"' in APP_HTML + assert 'id="backtestConfigProvenance"' in APP_HTML + + +def test_minute_frequency_formatter_states_fixed_execution_policy(): + 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'" + "})" + ) + + assert value == "5m source · 1h decisions · next 5m open fills · 5m valuation" + + +def test_quality_formatter_reports_dropped_and_problem_counts(): + value = _run_formatters( + "formatBacktestMarketDataQuality({" + "total_decision_bars:210,usable_decision_bars:207," + "dropped_decision_bars:3,missing_source_bars:2," + "duplicate_source_bars:1,off_grid_source_bars:0,invalid_source_bars:0" + "})" + ) + + assert value == "207/210 usable · 3 dropped · 2 missing · 1 duplicate" + + +def test_verified_contract_and_iex_fallback_are_visible(): + frequency = _run_formatters( + "formatBacktestFrequencyContract({" + "source_timeframe:'5m',decision_timeframe:'60m'," + "decision_frequency:'1h',execution_timeframe:'5m'," + "valuation_frequency:'5m',fill_policy:'next_source_bar_open'," + "verification_status:'verified'" + "})" + ) + provenance = _run_formatters( + "formatBacktestMarketDataProvenance({" + "market_data_feed:'iex',sip_fallback_to_iex:true,end_clamped:true" + "})" + ) + + assert frequency.endswith(" · verified") + assert provenance == "Alpaca IEX · SIP fallback · end clamped" + + +def test_alpaca_badge_and_strategy_page_describe_minute_source_hourly_decisions(): + badge = _render_data_source_badge( + { + "data_source": "alpaca", + "frequency_contract": { + "source_timeframe": "5m", + "decision_frequency": "1h", + }, + } + ) + render = fn_body("function renderBacktestRunConfig(") + assert badge["textContent"] == "Alpaca · 5m source · hourly decisions" + assert badge["hidden"] is False + assert "run?.frequency_contract" in render + assert "run?.market_data_quality" in render + assert "run?.market_data_feed" in render + + strategy_source = (FRONTEND / "strategy.html").read_text(encoding="utf-8") + assert "Alpaca 5-minute data + hourly decisions + hosted model" in strategy_source diff --git a/dashboard/frontend/app.html b/dashboard/frontend/app.html index f87b17ef..309dd102 100644 --- a/dashboard/frontend/app.html +++ b/dashboard/frontend/app.html @@ -1233,9 +1233,12 @@

Run config

Show advanced details
Symbols—
-
Timeframe—
+
Decision bars—
Decision method—
Status—
+ + + diff --git a/dashboard/frontend/app.js b/dashboard/frontend/app.js index 8c0c317c..14e35f99 100644 --- a/dashboard/frontend/app.js +++ b/dashboard/frontend/app.js @@ -3008,9 +3008,18 @@ function renderBacktestDataSourceBadge(run) { const isSimulation = run.data_source === 'vnpy_simulation'; const isIFind = run.data_source === 'ifind_ashare'; + const frequency = run.frequency_contract; + const sourceTimeframe = frequency?.source_timeframe; + const decisionCadence = frequency?.decision_frequency === '1h' + ? 'hourly' + : frequency?.decision_frequency; badge.textContent = isIFind ? 'iFinD China A-Shares · 60m' - : (isSimulation ? 'vn.py simulated data' : 'Alpaca data'); + : (isSimulation + ? 'vn.py simulated data' + : (sourceTimeframe && decisionCadence + ? `Alpaca · ${sourceTimeframe} source · ${decisionCadence} decisions` + : 'Alpaca data')); badge.className = `data-source-badge ${isIFind ? 'is-ifind' : (isSimulation ? 'is-simulated' : 'is-alpaca')}`; badge.hidden = false; } @@ -8227,6 +8236,52 @@ function formatTransactionCostTotals(totals) { ].filter(Boolean).join(' · '); } +function formatBacktestFrequencyContract(contract) { + if (!contract || typeof contract !== 'object') return null; + const source = contract.source_timeframe; + const decision = contract.decision_frequency || contract.decision_timeframe; + 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`; + const verification = contract.verification_status === 'verified' + ? ' · verified' + : ''; + return `${source} source · ${decision} decisions · ${fill} · ${valuation} valuation${verification}`; +} + +function formatBacktestMarketDataQuality(quality) { + if (!quality || typeof quality !== 'object') return null; + const total = Number(quality.total_decision_bars); + const usable = Number(quality.usable_decision_bars); + if (!Number.isFinite(total) || !Number.isFinite(usable)) return null; + const dropped = Number(quality.dropped_decision_bars || 0); + const parts = [`${usable}/${total} usable`]; + parts.push(dropped > 0 ? `${dropped} dropped` : 'no drops'); + for (const [field, label] of [ + ['missing_source_bars', 'missing'], + ['duplicate_source_bars', 'duplicate'], + ['off_grid_source_bars', 'off-grid'], + ['invalid_source_bars', 'invalid'], + ]) { + const count = Number(quality[field] || 0); + if (count > 0) parts.push(`${count} ${label}`); + } + return parts.join(' · '); +} + +function formatBacktestMarketDataProvenance(provenance) { + if (!provenance || typeof provenance !== 'object') return null; + const feed = String(provenance.market_data_feed || '').trim().toUpperCase(); + if (!feed) return null; + const parts = [`Alpaca ${feed}`]; + if (provenance.sip_fallback_to_iex) parts.push('SIP fallback'); + if (provenance.end_clamped) parts.push('end clamped'); + return parts.join(' · '); +} + function renderBacktestRunConfig( run, { @@ -8253,6 +8308,20 @@ function renderBacktestRunConfig( const metadata = run?.metadata && typeof run.metadata === 'object' ? run.metadata : {}; + const frequencyContract = run?.frequency_contract + || metadata.frequency_contract + || cfg?.frequencyContract + || null; + const marketDataQuality = run?.market_data_quality + || metadata.market_data_quality + || null; + const frequencyLabel = formatBacktestFrequencyContract(frequencyContract); + const dataQualityLabel = formatBacktestMarketDataQuality(marketDataQuality); + const provenanceLabel = formatBacktestMarketDataProvenance({ + market_data_feed: run?.market_data_feed ?? metadata.market_data_feed, + sip_fallback_to_iex: run?.sip_fallback_to_iex ?? metadata.sip_fallback_to_iex, + end_clamped: run?.end_clamped ?? metadata.end_clamped, + }); const llmExecution = run?.llm_execution && typeof run.llm_execution === 'object' ? run.llm_execution : (metadata.llm_execution && typeof metadata.llm_execution === 'object' @@ -8319,7 +8388,11 @@ function renderBacktestRunConfig( || '—'; const symbolCount = cfg?.symbolCount ?? (Array.isArray(runSymbols) ? runSymbols.length : null); - const timeframe = cfg?.timeframe || metadata.timeframe || run?.timeframe || '60m'; + const timeframe = frequencyContract?.decision_timeframe + || cfg?.timeframe + || metadata.timeframe + || run?.timeframe + || '60m'; const decisionSource = cfg?.decisionSource || metadata.decision_source || run?.decision_source @@ -8435,6 +8508,21 @@ function renderBacktestRunConfig( Number.isFinite(Number(symbolCount)) ? String(symbolCount) : '—', ); setBacktestConfigText('backtestConfigTimeframe', timeframe); + const frequencyRow = document.getElementById('backtestConfigFrequencyRow'); + const dataQualityRow = document.getElementById('backtestConfigDataQualityRow'); + const provenanceRow = document.getElementById('backtestConfigProvenanceRow'); + if (frequencyRow) frequencyRow.hidden = !frequencyLabel; + if (dataQualityRow) dataQualityRow.hidden = !dataQualityLabel; + if (provenanceRow) provenanceRow.hidden = !provenanceLabel; + if (frequencyLabel) { + setBacktestConfigText('backtestConfigFrequency', frequencyLabel); + } + if (dataQualityLabel) { + setBacktestConfigText('backtestConfigDataQuality', dataQualityLabel); + } + if (provenanceLabel) { + setBacktestConfigText('backtestConfigProvenance', provenanceLabel); + } setBacktestConfigText( 'backtestConfigDecisionSource', decisionSourceLabel, @@ -8744,6 +8832,17 @@ async function runBacktest() { universeKey: selectedIFindUniverse, symbolCount: assets.length, timeframe: isIFind ? IFIND_ASHARE_TIMEFRAME : '60m', + frequencyContract: isIFind || isSimulation + ? null + : { + source_timeframe: '5m', + decision_timeframe: '60m', + decision_frequency: '1h', + execution_timeframe: '5m', + valuation_frequency: '5m', + aggregation: 'session_anchored_completed_bars', + fill_policy: 'next_source_bar_open', + }, decisionSource, billingMode: isRuleBasedDecision ? null : selectedBillingMode, providerId: isRuleBasedDecision || selectedBillingMode === 'platform_credits' @@ -8751,7 +8850,7 @@ async function runBacktest() { : selectedProviderId, marketDataLabel: isIFind ? 'iFinD A-Share' - : (isSimulation ? 'vn.py Simulation' : 'Alpaca'), + : (isSimulation ? 'vn.py Simulation' : 'Alpaca · 5m source'), dataSource, startedAt: new Date().toISOString(), }; @@ -8760,6 +8859,7 @@ async function runBacktest() { renderBacktestDataSourceBadge({ data_source: dataSource, timeframe: isIFind ? IFIND_ASHARE_TIMEFRAME : null, + frequency_contract: launchConfigBase.frequencyContract, }); // Pin live view BEFORE navigateToPage → showPlaygroundPanel → loadData(), diff --git a/dashboard/frontend/strategy.html b/dashboard/frontend/strategy.html index addd094d..08428b19 100644 --- a/dashboard/frontend/strategy.html +++ b/dashboard/frontend/strategy.html @@ -151,7 +151,7 @@

Trading Strategy

runBtn.disabled = true; metricsEl.hidden = true; statusEl.className = "status"; - statusEl.textContent = "Submitting backtest (Alpaca hourly bars + hosted model)…"; + statusEl.textContent = "Submitting backtest (Alpaca 5-minute data + hourly decisions + hosted model)…"; const headers = { "Content-Type": "application/json", "X-Session-Id": sessionId() }; try { diff --git a/dashboard/scripts/backtest_hourly_agent.py b/dashboard/scripts/backtest_hourly_agent.py index 7b6f177b..bdf7045a 100644 --- a/dashboard/scripts/backtest_hourly_agent.py +++ b/dashboard/scripts/backtest_hourly_agent.py @@ -486,7 +486,10 @@ def main(): print("⚙️ Using rule-based logic for trading decisions\n") # Step 1: Load data - print(f"1️⃣ Loading historical hourly data from {args.data_source}...") + print( + f"1️⃣ Loading historical source data from {args.data_source} " + f"(decisions remain hourly)..." + ) backtester.load_data() # Step 2: Calculate indicators diff --git a/docs/minute-data-hourly-trading.md b/docs/minute-data-hourly-trading.md new file mode 100644 index 00000000..d29c7478 --- /dev/null +++ b/docs/minute-data-hourly-trading.md @@ -0,0 +1,121 @@ +# 分钟源数据与小时级交易契约 + +Phase 0/1 固定四个时间概念，避免把数据频率误当成策略频率： + +| 字段 | Phase 0/1 目标 | 含义 | +| --- | --- | --- | +| `source_timeframe` | `5m` | 数据供应商返回的原始 bar 周期 | +| `decision_timeframe` | `60m` | 策略观察的已完成 bar 周期 | +| `decision_frequency` | `1h` | agent/策略的调用频率 | +| `execution_timeframe` | `5m` | 后续成交模拟使用的 bar 周期 | +| `valuation_frequency` | `5m` | 后续组合盯市使用的周期 | + +Phase 2 已接入 `HourlyBacktester` 和 API 的 `ExternalBacktestSession`：Alpaca +回测会显式请求 profile 的 `5m` +源数据，按交易时段聚合成已完成的 `60m` 决策 bar。策略/LLM 仍每小时调用 +一次；成交使用对应下一根源 bar 的开盘价，组合估值按每根 `5m` bar 更新。 +共享数据集缓存也按源/决策频率隔离，避免同一日期窗口的旧小时数据和分钟数据 +相互复用。非 Alpaca provider 和旧的小时级注入 loader 仍保持原有路径。 + +聚合器位于 `dashboard/backend/domain/backtesting/bar_aggregation.py`，使用 +美股 09:30 和 A 股早午盘边界切桶，不跨越午休；聚合结果保留 +`source_bar_count`、`expected_source_bars`、`is_complete`、`has_gap` 等质量字段。 +最后一个没有下一根源 bar 的收盘桶不进入可执行决策。 + +## Phase 3：正确性硬化与审计 + +Phase 3 将分钟链路的数据质量和无未来数据约束固化为可测试契约： + +- 小时 bar 使用左闭右开区间。例如 10:30 决策只包含 09:30–10:25 的 + 5 分钟 bar；10:30 的源 bar 只可作为该决策的成交 bar。 +- 完整性按预期的 5 分钟时间槽逐一校验，而非只比较行数。缺失、重复、 + 非 5 分钟网格时间戳和无效 OHLCV 都会使该小时 bar 失效。 +- 失效小时 bar 不进入 agent 决策集。多标的决策时点必须覆盖至少 80% 的 + 标的，门槛向上取整；没有时点达标时不再回退到稀疏时间轴。 +- 成交必须精确命中小时决策边界处的源 bar。该标的缺少这根 bar 时保持 + 未成交，不会延迟到后续 5 分钟 bar 补成交。 +- `market_data_quality` 随回测结果写入元数据，记录可用/丢弃小时 bar 数， + 以及缺失、重复、错位和无效源 bar 数；网页版决策审计同时记录 + `timestamp` 与 `execution_timestamp`。 + +质量统计以“标的 × 决策 bar”为计数单位。同一小时若 30 个标的均有数据， +会计为 30 个 decision bars。 + +## Phase 4：API 可观测性 + +Phase 4 保持唯一、固定的成交规则 `next_source_bar_open`，不增加成交策略 +配置项。回测详情 API 和外部 Agent 的完成结果会返回： + +- `frequency_contract`：5m 源数据、60m 决策 bar、1h 决策、5m 成交与估值； +- `market_data_quality`：聚合后的可用、丢弃及异常计数。 + +数据库仍保留逐标的质量明细；列表 API 仅返回有界的汇总字段，避免运行列表 +随股票池大小无限膨胀。 + +## Phase 5：网页展示与产品契约 + +网页版回测在运行前和完成后均明确显示 Alpaca 5m 源数据与小时级决策。 +高级详情展示固定成交时点、估值频率和数据质量摘要；历史旧回测缺少新元数据时 +保持原有显示，不伪造分钟级来源。独立策略页面的旧 “Alpaca hourly bars” +文案同步更新。 + +## Phase 6：运行时认证与数据来源 + +Phase 6 在实际数据完成加载后认证频率契约。若 provider 的配置或 fetch +证据报告了不同于请求值的源周期，回测立即失败并返回明确的 frequency +contract 错误，不会把小时数据标成分钟数据继续执行。通过认证的分钟回测在 +`frequency_contract.verification_status` 中记录 `verified`；固定成交规则仍为 +`next_source_bar_open`，不存在用户可配置入口。 + +Alpaca frame 上的来源标记也随运行持久化并经 API 展示：实际 feed（SIP 或 +IEX）、是否由 SIP 降级到 IEX，以及请求结束时间是否因实时数据权限被截断。 +网页在回测高级详情中显示这些来源信息，使频率正确但 tape 不同的运行也不会 +被误认为完全等价。 + +## 代码契约 + +`dashboard/backend/infrastructure/market_data/frequency.py` 提供： + +- `normalize_bar_timeframe()`：将 `1min`、`5Min`、`hourly` 等别名归一化为 + `1m`、`5m`、`60m`。 +- `TradingFrequency`：校验源周期、决策周期、决策频率、执行周期和估值周期。 +- `TradingFrequency.minute_source_hourly_decisions()`：返回目标 5m 源数据、 + 1h 决策契约。 + +`MarketProfile` 保留旧的 `timeframe` 字段作为决策 bar 周期，并新增： + +```text +source_timeframe +decision_frequency +execution_timeframe +valuation_frequency +``` + +当前 Alpaca profile 已记录目标契约；iFinD 和 vn.py profile 继续使用现有 +`60m` 数据能力。 + +## Alpaca 数据层 + +`AlpacaDataLoader` 保持默认 `60m`，保证旧调用方兼容；分钟数据调用方式： + +```python +loader = AlpacaDataLoader( + api_key="...", + secret_key="...", + source_timeframe="5m", +) +bars = loader.fetch_bars(["AAPL"], "2026-01-01", "2026-01-03") +``` + +也可以通过 provider factory 显式选择： + +```python +provider = create_market_data_provider( + "alpaca", + source_timeframe="5m", +) +``` + +支持的 Alpaca 源周期为 `1m`、`5m` 和 `60m`。每次 fetch 的 +`last_fetch["source_timeframe"]` 会记录实际请求周期。默认不传 +`source_timeframe` 时仍请求 profile 的决策周期，防止旧小时回测行为改变。