diff --git a/README.md b/README.md index 02db6b3..49282c0 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,7 @@ uv run --no-sync python scripts/check_qpk_pin_consistency.py ## Useful docs +- [`docs/historical_price_basis.md`](docs/historical_price_basis.md): explicit `PRICE_BASIS` / `price_basis`, coherent OHLC, and unavailable-basis failures - [`docs/hk_equity_runtime.md`](docs/hk_equity_runtime.md) - [`docs/ibkr_paper_top50_switch.md`](docs/ibkr_paper_top50_switch.md) - [`docs/ibkr_runtime_rollout.md`](docs/ibkr_runtime_rollout.md) diff --git a/application/price_basis.py b/application/price_basis.py new file mode 100644 index 0000000..157eb9e --- /dev/null +++ b/application/price_basis.py @@ -0,0 +1,120 @@ +"""Explicit historical OHLC contract; no corporate-action basis substitution.""" + +from __future__ import annotations + +import json +import logging +from enum import StrEnum +from math import isclose, isfinite + + +class PriceBasis(StrEnum): + RAW_UNADJUSTED = "raw_unadjusted" + SPLIT_ADJUSTED = "split_adjusted" + SPLIT_DIVIDEND_ADJUSTED = "split_dividend_adjusted" + TOTAL_RETURN_INDEX = "total_return_index" + + +DEFAULT_PRICE_BASIS = PriceBasis.SPLIT_ADJUSTED.value +VOLUME_BASIS = "provider_native_unscaled" +OHLC_FIELDS = ("open", "high", "low", "close") + + +def validate_price_basis(value) -> str: + if not isinstance(value, str) or value not in tuple(PriceBasis): + # Do not echo an arbitrary configuration value into diagnostics. + raise ValueError("price_basis must be one of: " + ", ".join(PriceBasis)) + return str(value) + + +def basis_diagnostic(requested, *, provider, source_columns=(), resolved=None, reason=None): + return { + "requested_basis": requested, + "resolved_basis": resolved, + "provider": provider, + "source_columns": list(source_columns), + "volume_basis": VOLUME_BASIS, + "reason": reason, + } + + +class PriceBasisError(ValueError): + """A requested history basis could not be proven or produced. + + Only bounded contract metadata is logged; never provider exception text, + prices, account identifiers, credentials or request arguments. + """ + + def __init__(self, requested, *, provider, reason, source_columns=()): + self.diagnostic = basis_diagnostic( + requested, provider=provider, source_columns=source_columns, reason=reason, + ) + message = "price_basis_unavailable " + json.dumps(self.diagnostic, sort_keys=True) + logging.getLogger(__name__).warning(message) + super().__init__(message) + + +def require_supported_ohlc_basis(requested, *, provider): + requested = validate_price_basis(requested) + if requested == PriceBasis.TOTAL_RETURN_INDEX: + raise PriceBasisError(requested, provider=provider, reason="total_return_index_is_not_ohlc") + if requested == PriceBasis.RAW_UNADJUSTED: + raise PriceBasisError(requested, provider=provider, reason="raw_prices_not_provided") + return requested + + +class HistoricalCandles(list): + """Numeric candles stay unchanged; one batch carries non-sensitive metadata. + + A custom history loader must declare every OHLC field's basis. Unknown or + conflicting metadata is rejected at the adapter boundary. + """ + + def __init__(self, rows, *, price_basis, provider, source_columns, field_basis=None): + super().__init__(rows) + self.diagnostic = basis_diagnostic( + validate_price_basis(price_basis), provider=provider, + source_columns=source_columns, resolved=price_basis, + ) + self.field_basis = dict(field_basis) if field_basis is not None else { + field: price_basis for field in OHLC_FIELDS + } + + +def validate_candles(candles, requested): + if not isinstance(candles, HistoricalCandles): + raise PriceBasisError(requested, provider="history_loader", reason="missing_basis_metadata") + diagnostic = candles.diagnostic + provider = diagnostic["provider"] + columns = diagnostic["source_columns"] + + def fail(reason): + raise PriceBasisError(requested, provider=provider, source_columns=columns, reason=reason) + + if (diagnostic["resolved_basis"] != requested + or diagnostic["requested_basis"] != requested + or any(candles.field_basis.get(field) != requested for field in OHLC_FIELDS)): + fail("mixed_or_mismatched_basis") + if not candles: + fail("empty_history") + for row in candles: + try: + prices = [float(row[field]) for field in OHLC_FIELDS] + volume = float(row["volume"]) + if not all(isfinite(value) and value > 0 for value in prices): + fail("invalid_price") + if not isfinite(volume) or volume < 0: + fail("invalid_volume") + opening, high, low, close = prices + # A scaled O/H/L boundary can differ from Adj Close by one rounding + # unit. Tolerate floating-point noise without changing any value. + lower, upper = min(opening, close), max(opening, close) + if ((low > lower and not isclose(low, lower, rel_tol=1e-12)) + or (high < upper and not isclose(high, upper, rel_tol=1e-12)) + or (low > high and not isclose(low, high, rel_tol=1e-12))): + fail("inconsistent_ohlc") + except (KeyError, TypeError, ValueError, OverflowError) as exc: + if isinstance(exc, PriceBasisError): + raise + fail("missing_or_invalid_ohlcv") + return candles diff --git a/application/runtime_strategy_adapters.py b/application/runtime_strategy_adapters.py index f8dcb71..ffad486 100644 --- a/application/runtime_strategy_adapters.py +++ b/application/runtime_strategy_adapters.py @@ -3,13 +3,19 @@ from __future__ import annotations from dataclasses import dataclass -from math import ceil +from math import ceil, isfinite +import json +import logging import re from typing import Any import pandas as pd from application import signal_provenance +from application.price_basis import ( + DEFAULT_PRICE_BASIS, HistoricalCandles, PriceBasis, PriceBasisError, + basis_diagnostic, require_supported_ohlc_basis, validate_candles, validate_price_basis, +) from quant_platform_kit.common.strategy_plugins import ( build_strategy_plugin_alert_messages, build_strategy_plugin_error_notification_lines, @@ -75,26 +81,62 @@ def _frame_column(frame: pd.DataFrame, name: str) -> pd.Series | None: _YFINANCE_PRICE_BASIS: dict[str, str] = {} -_IBKR_HISTORY_NOTE = "IBKR TRADES bars (split-adjusted); ADJUSTED_LAST suppressed for signals" -class SplitOnlyHistoricalIB: - """Proxy around an ``ib`` client for signal-history requests only (RS-08 fix 3). +class BasisHistoricalIB: + """Constrain the pinned QPK loader to exactly one IBKR price basis. - QPK requests ``ADJUSTED_LAST`` (dividend-adjusted) first and falls back to - ``TRADES`` when that returns nothing. Answering ``ADJUSTED_LAST`` with no bars - makes QPK use split-adjusted ``TRADES`` bars, matching the yfinance ``Close`` - basis. Every other attribute is delegated unchanged; only historical-data - calls made through this proxy are affected. + For split-only requests, skip QPK's first ADJUSTED_LAST probe. For adjusted + requests, forbid its TRADES retry even if QPK catches the initial error. + This is endpoint selection, never a price-basis fallback. """ - def __init__(self, ib: Any) -> None: + def __init__(self, ib: Any, price_basis=DEFAULT_PRICE_BASIS, *, ohlc=False) -> None: self._ib = ib + self.price_basis = require_supported_ohlc_basis(price_basis, provider="ibkr") + self.endpoint = "TRADES" if self.price_basis == PriceBasis.SPLIT_ADJUSTED else "ADJUSTED_LAST" + self.failure = None + self.resolved = False + self.ohlc = ohlc def reqHistoricalData(self, *args: Any, **kwargs: Any) -> Any: # noqa: N802 - ib_insync API name - if str(kwargs.get("whatToShow") or "").upper() == "ADJUSTED_LAST": - return [] - return self._ib.reqHistoricalData(*args, **kwargs) + endpoint = str(kwargs.get("whatToShow") or "").upper() + if self.failure is not None: + raise self.failure + if endpoint != self.endpoint: + if endpoint == "ADJUSTED_LAST" and self.endpoint == "TRADES": + return [] + raise PriceBasisError(self.price_basis, provider="ibkr", reason="basis_fallback_forbidden", + source_columns=(self.endpoint,)) + try: + bars = self._ib.reqHistoricalData(*args, **kwargs) + except Exception: + self.failure = PriceBasisError(self.price_basis, provider="ibkr", reason="provider_request_failed", + source_columns=(self.endpoint,)) + raise self.failure from None + if not bars: + self.failure = PriceBasisError(self.price_basis, provider="ibkr", reason=f"empty_{self.endpoint}", + source_columns=(self.endpoint,)) + raise self.failure + if self.ohlc: + # Check native fields BEFORE QPK's legacy candle conversion, which + # otherwise synthesizes missing O/H/L from C and missing volume as 0. + try: + rows = [{field: getattr(bar, field) for field in ("open", "high", "low", "close", "volume")} + for bar in bars] + validate_candles(HistoricalCandles( + rows, price_basis=self.price_basis, provider="ibkr", + source_columns=(self.endpoint, "open", "high", "low", "close", "volume"), + ), self.price_basis) + except AttributeError: + self.failure = PriceBasisError(self.price_basis, provider="ibkr", reason="missing_native_ohlcv", + source_columns=(self.endpoint,)) + raise self.failure from None + except PriceBasisError as exc: + self.failure = exc + raise + self.resolved = True + return bars def __getattr__(self, name: str) -> Any: return getattr(self._ib, name) @@ -105,46 +147,74 @@ def yfinance_price_basis(symbol: str) -> str: return _YFINANCE_PRICE_BASIS.get(str(symbol or "").strip().upper(), signal_provenance.UNKNOWN) -def _coerce_yfinance_candles(frame: pd.DataFrame, *, symbol: str | None = None) -> list[dict[str, Any]]: +def _coerce_yfinance_candles(frame: pd.DataFrame, *, symbol: str | None = None, + price_basis: str = DEFAULT_PRICE_BASIS) -> list[dict[str, Any]]: + basis = require_supported_ohlc_basis(price_basis, provider="yfinance") + if symbol: + _YFINANCE_PRICE_BASIS.pop(str(symbol).strip().upper(), None) if frame is None or frame.empty: - return [] - # RS-08 fix 3: signals use split-only prices (backtest standard ยง4.4, same - # basis as Schwab). With auto_adjust=False yfinance "Close" is split-adjusted - # only; "Adj Close" (dividend-adjusted) is used only if "Close" is absent. - close_series = _frame_column(frame, "Close") - basis = signal_provenance.SPLIT_ONLY - if close_series is None: - close_series = _frame_column(frame, "Adj Close") - basis = signal_provenance.DIVIDEND_ADJUSTED - if close_series is not None and symbol: - _YFINANCE_PRICE_BASIS[str(symbol).strip().upper()] = basis - if close_series is None: - return [] - open_series = _frame_column(frame, "Open") - high_series = _frame_column(frame, "High") - low_series = _frame_column(frame, "Low") - volume_series = _frame_column(frame, "Volume") + raise PriceBasisError(basis, provider="yfinance", reason="empty_history") + columns = ["Close", "Open", "High", "Low", "Volume"] + if basis == PriceBasis.SPLIT_DIVIDEND_ADJUSTED: + columns.append("Adj Close") + selected = {} + for column in columns: + series = _frame_column(frame, column) + if series is None: + raise PriceBasisError(basis, provider="yfinance", reason=f"missing_column:{column}", + source_columns=tuple(selected)) + # Multiple instruments/duplicate fields cannot be silently collapsed. + matches = [key for key in frame.columns if key == column or ( + isinstance(key, tuple) and column in key + )] + if len(matches) != 1: + raise PriceBasisError(basis, provider="yfinance", reason="ambiguous_columns", source_columns=columns) + selected[column] = series + if not frame.index.is_unique: + raise PriceBasisError(basis, provider="yfinance", reason="duplicate_rows", source_columns=columns) + # Native yfinance auto_adjust=False OHLC is split adjusted. An explicit + # per-column metadata declaration must agree with that source contract. + declared = frame.attrs.get("field_price_basis", {}) + expected = {name: PriceBasis.SPLIT_ADJUSTED.value for name in ("Open", "High", "Low", "Close")} + expected["Adj Close"] = PriceBasis.SPLIT_DIVIDEND_ADJUSTED.value + if any(name in declared and declared[name] != expected[name] for name in expected): + raise PriceBasisError(basis, provider="yfinance", reason="mixed_or_mismatched_basis", source_columns=columns) + close_series = selected["Close"] candles: list[dict[str, Any]] = [] - for as_of, close in close_series.dropna().items(): - close_value = float(close) - candles.append( - { + try: + for as_of, close in close_series.items(): + # Split+dividend OHLC: apply Adj Close / Close on EACH row to O/H/L; + # use Adj Close itself for C. Volume is NEVER scaled by this factor. + close_value = float(close) + factor = float(selected["Adj Close"].loc[as_of]) / close_value if "Adj Close" in selected else 1.0 + if not isfinite(close_value) or close_value <= 0 or not isfinite(factor) or factor <= 0: + raise ValueError("invalid factor") + row = { "as_of": pd.Timestamp(as_of).to_pydatetime(), - "open": float(open_series.get(as_of, close_value)) if open_series is not None else close_value, - "high": float(high_series.get(as_of, close_value)) if high_series is not None else close_value, - "low": float(low_series.get(as_of, close_value)) if low_series is not None else close_value, - "close": close_value, - "volume": float(volume_series.get(as_of, 0.0) or 0.0) if volume_series is not None else 0.0, + "open": float(selected["Open"].loc[as_of]) * factor, + "high": float(selected["High"].loc[as_of]) * factor, + "low": float(selected["Low"].loc[as_of]) * factor, + "close": float(selected["Adj Close"].loc[as_of]) if "Adj Close" in selected else close_value, + "volume": float(selected["Volume"].loc[as_of]), } - ) - return candles + candles.append(row) + except (TypeError, ValueError, ZeroDivisionError, OverflowError): + raise PriceBasisError(basis, provider="yfinance", reason="invalid_ohlcv_or_factor", source_columns=columns) from None + result = validate_candles(HistoricalCandles(candles, price_basis=basis, provider="yfinance", source_columns=columns), basis) + if symbol: + _YFINANCE_PRICE_BASIS[str(symbol).strip().upper()] = basis + logging.getLogger(__name__).info("price_basis_resolved %s", json.dumps(result.diagnostic, sort_keys=True)) + return result -def fetch_yfinance_historical_candles(symbol: str, *, duration: str = "2 Y", bar_size: str = "1 day") -> list[dict[str, Any]]: +def fetch_yfinance_historical_candles(symbol: str, *, duration: str = "2 Y", bar_size: str = "1 day", + price_basis: str = DEFAULT_PRICE_BASIS) -> list[dict[str, Any]]: + basis = require_supported_ohlc_basis(price_basis, provider="yfinance") + _YFINANCE_PRICE_BASIS.pop(str(symbol).strip().upper(), None) try: import yfinance as yf except Exception: - return [] + raise PriceBasisError(basis, provider="yfinance", reason="provider_unavailable") from None try: frame = yf.download( @@ -156,8 +226,8 @@ def fetch_yfinance_historical_candles(symbol: str, *, duration: str = "2 Y", bar threads=False, ) except Exception: - return [] - return _coerce_yfinance_candles(frame, symbol=str(symbol).strip().upper()) + raise PriceBasisError(basis, provider="yfinance", reason="provider_request_failed") from None + return _coerce_yfinance_candles(frame, symbol=str(symbol).strip().upper(), price_basis=basis) @dataclass(frozen=True) @@ -174,6 +244,10 @@ class IBKRRuntimeStrategyAdapters: build_strategy_plugin_report_payload_fn: Any = None load_configured_strategy_plugin_signals_fn: Any = None parse_strategy_plugin_mounts_fn: Any = None + price_basis: str = DEFAULT_PRICE_BASIS + + def __post_init__(self): + validate_price_basis(self.price_basis) def load_strategy_plugin_signals(self, raw_mounts): if not raw_mounts or self.parse_strategy_plugin_mounts_fn is None or self.load_configured_strategy_plugin_signals_fn is None: @@ -217,86 +291,89 @@ def build_strategy_plugin_alert_messages(self, signals): strategy_label=self.strategy_profile, ) - def _get_fallback_historical_candles(self, symbol, *, duration: str, bar_size: str) -> tuple[dict[str, Any], ...]: + def _get_fallback_historical_candles(self, symbol, *, duration: str, bar_size: str): + require_supported_ohlc_basis(self.price_basis, provider="history_loader") if self.fallback_historical_candles_fn is None: - return () + return None try: - return tuple(self.fallback_historical_candles_fn(symbol, duration=duration, bar_size=bar_size) or ()) + candles = self.fallback_historical_candles_fn( + symbol, duration=duration, bar_size=bar_size, price_basis=self.price_basis, + ) + except PriceBasisError: + raise except Exception: - return () + raise PriceBasisError(self.price_basis, provider="history_loader", reason="provider_request_failed") from None + return validate_candles(candles, self.price_basis) - def _fallback_source_and_basis(self, symbol) -> tuple[str, str]: - if self.fallback_historical_candles_fn is fetch_yfinance_historical_candles: - return "yfinance", yfinance_price_basis(symbol) - return "fallback_provider", signal_provenance.UNKNOWN + def _record_history(self, symbol, dates, diagnostic): + signal_provenance.record( + symbol, source=diagnostic["provider"], price_basis=self.price_basis, + dates=dates, basis_details=diagnostic, + ) + logging.getLogger(__name__).info("price_basis_resolved %s", json.dumps(diagnostic, sort_keys=True)) - def _record_fallback(self, symbol, dates) -> None: - source, basis = self._fallback_source_and_basis(symbol) - signal_provenance.record(symbol, source=source, price_basis=basis, dates=dates) + def _fetch_native(self, fn, ib, symbol, *, duration, bar_size, ohlc=False): + proxy = BasisHistoricalIB(ib, self.price_basis, ohlc=ohlc) + try: + result = fn(proxy, symbol, duration=duration, bar_size=bar_size) + except PriceBasisError: + raise + except Exception: + raise PriceBasisError(self.price_basis, provider="ibkr", reason="provider_request_failed", + source_columns=(proxy.endpoint,)) from None + if proxy.failure is not None: + raise proxy.failure + if not proxy.resolved: + raise PriceBasisError(self.price_basis, provider="ibkr", reason="unverified_provider_response", + source_columns=(proxy.endpoint,)) + return result, basis_diagnostic(self.price_basis, provider="ibkr", resolved=self.price_basis, + source_columns=(proxy.endpoint, "Open", "High", "Low", "Close", "Volume")) def get_historical_close(self, ib, symbol, duration="2 Y", bar_size="1 day"): fallback_candles = self._get_fallback_historical_candles(symbol, duration=duration, bar_size=bar_size) fallback_points = tuple( (candle["as_of"], candle["close"]) - for candle in fallback_candles - if "close" in candle + for candle in (fallback_candles or ()) ) if fallback_points: - self._record_fallback(symbol, [as_of for as_of, _ in fallback_points]) + self._record_history(symbol, [as_of for as_of, _ in fallback_points], fallback_candles.diagnostic) return pd.Series( data=[close for _, close in fallback_points], index=pd.to_datetime([as_of for as_of, _ in fallback_points]), ) - series = self.fetch_historical_price_series_fn( - SplitOnlyHistoricalIB(ib), - symbol, + series, diagnostic = self._fetch_native( + self.fetch_historical_price_series_fn, ib, symbol, duration=duration, bar_size=bar_size, ) points = tuple(series.points or ()) - signal_provenance.record( - symbol, - source="ibkr_historical", - price_basis=signal_provenance.SPLIT_ONLY, - dates=[point.as_of for point in points], - note=_IBKR_HISTORY_NOTE, + if not points or any(not isfinite(float(point.close)) or float(point.close) <= 0 for point in points): + raise PriceBasisError(self.price_basis, provider="ibkr", reason="empty_or_invalid_close", + source_columns=diagnostic["source_columns"]) + diagnostic["source_columns"] = [diagnostic["source_columns"][0], "Close"] + self._record_history(symbol, [point.as_of for point in points], diagnostic) + return pd.Series( + data=[point.close for point in points], + index=pd.to_datetime([point.as_of for point in points]), ) - if points: - return pd.Series( - data=[point.close for point in points], - index=pd.to_datetime([point.as_of for point in points]), - ) - return pd.Series(dtype=float) def get_historical_candles(self, ib, symbol, duration="2 Y", bar_size="1 day"): fallback_candles = self._get_fallback_historical_candles(symbol, duration=duration, bar_size=bar_size) if fallback_candles: - self._record_fallback(symbol, [candle.get("as_of") for candle in fallback_candles]) - return list(fallback_candles) + self._record_history(symbol, [candle["as_of"] for candle in fallback_candles], fallback_candles.diagnostic) + return fallback_candles - candles = self.fetch_historical_price_candles_fn( - SplitOnlyHistoricalIB(ib), - symbol, + candles, diagnostic = self._fetch_native( + self.fetch_historical_price_candles_fn, ib, symbol, duration=duration, bar_size=bar_size, + ohlc=True, ) - candle_dates: list[Any] = [] - if isinstance(candles, (list, tuple)): # never consume a one-shot iterator - try: - candle_dates = [ - candle.get("as_of") if isinstance(candle, dict) else getattr(candle, "as_of", None) - for candle in candles - ] - except Exception: - candle_dates = [] - signal_provenance.record( - symbol, - source="ibkr_historical", - price_basis=signal_provenance.SPLIT_ONLY, - dates=candle_dates, - note=_IBKR_HISTORY_NOTE, - ) + candles = validate_candles(HistoricalCandles( + candles, price_basis=self.price_basis, provider="ibkr", source_columns=diagnostic["source_columns"], + ), self.price_basis) + self._record_history(symbol, [candle["as_of"] for candle in candles], diagnostic) return candles def compute_signals(self, ib, current_holdings, *, strategy_plugin_signals=()): @@ -337,7 +414,10 @@ def build_runtime_strategy_adapters( build_strategy_plugin_report_payload_fn=None, load_configured_strategy_plugin_signals_fn=None, parse_strategy_plugin_mounts_fn=None, + price_basis=None, ) -> IBKRRuntimeStrategyAdapters: + config = getattr(strategy_runtime, "merged_runtime_config", {}) or {} + basis = validate_price_basis(config.get("price_basis", DEFAULT_PRICE_BASIS) if price_basis is None else price_basis) return IBKRRuntimeStrategyAdapters( strategy_runtime=strategy_runtime, strategy_profile=str(strategy_profile), @@ -351,4 +431,5 @@ def build_runtime_strategy_adapters( build_strategy_plugin_report_payload_fn=build_strategy_plugin_report_payload_fn, load_configured_strategy_plugin_signals_fn=load_configured_strategy_plugin_signals_fn, parse_strategy_plugin_mounts_fn=parse_strategy_plugin_mounts_fn, + price_basis=basis, ) diff --git a/application/signal_provenance.py b/application/signal_provenance.py index d46f326..602d492 100644 --- a/application/signal_provenance.py +++ b/application/signal_provenance.py @@ -13,11 +13,11 @@ from datetime import date, datetime, timezone from typing import Any -SCHEMA_VERSION = "qsl.signal_data_provenance.v1" -SPLIT_ONLY = "split_only" -DIVIDEND_ADJUSTED = "dividend_adjusted" -UNKNOWN = "unknown" -PRICE_BASIS_LABELS = frozenset({SPLIT_ONLY, DIVIDEND_ADJUSTED, UNKNOWN}) +from application.price_basis import PriceBasis + +SCHEMA_VERSION = "qsl.signal_data_provenance.v2" +UNKNOWN = "unknown_legacy" # Inventory only, never accepted by history loaders. +PRICE_BASIS_LABELS = frozenset({*PriceBasis, UNKNOWN}) _LOCK = threading.Lock() _RECORDS: dict[str, dict[str, Any]] = {} @@ -47,7 +47,8 @@ def _iso_date(value: Any) -> str | None: return None -def record(symbol: Any, *, source: str, price_basis: str, dates: Iterable[Any], note: str | None = None) -> None: +def record(symbol: Any, *, source: str, price_basis: str, dates: Iterable[Any], note: str | None = None, + basis_details: dict | None = None) -> None: """Record one history fetch. Best-effort: swallows every error.""" try: normalized = str(symbol or "").strip().upper() @@ -63,6 +64,8 @@ def record(symbol: Any, *, source: str, price_basis: str, dates: Iterable[Any], } if note: entry["note"] = str(note) + if basis_details is not None: + entry["basis_details"] = dict(basis_details) with _LOCK: previous = _RECORDS.get(normalized) entry["fetch_count"] = int(previous.get("fetch_count", 0)) + 1 if previous else 1 diff --git a/docs/historical_price_basis.md b/docs/historical_price_basis.md new file mode 100644 index 0000000..b946470 --- /dev/null +++ b/docs/historical_price_basis.md @@ -0,0 +1,70 @@ +# Historical signal price basis + +`PRICE_BASIS` is an optional, non-secret environment configuration key. The +equivalent strategy runtime JSON key is `price_basis`. Precedence is explicit +`PRICE_BASIS`, runtime JSON, strategy manifest default, then `split_adjusted`. +Environment values are validated before configuration clients are resolved; +the merged runtime value is validated before strategy evaluation. Only the four +exact, case-sensitive values below are accepted (no aliases or empty values). + +| Requested `price_basis` | yfinance (`auto_adjust=False`) | IBKR historical bars | +| --- | --- | --- | +| `split_adjusted` (default) | Native Open/High/Low/Close | TRADES only | +| `split_dividend_adjusted` | Adj Close, with O/H/L scaled per row | ADJUSTED_LAST only | +| `raw_unadjusted` | Unsupported: Close already adjusts for splits | Unsupported: TRADES already adjusts for splits | +| `total_return_index` | Rejected for these OHLC/history consumers | Rejected for these OHLC/history consumers | + +The default preserves #602's successful Close/OHLC/Volume output. Synthetic +golden fixtures freeze IEEE-754 output from main commit +`8e173b2bd1cb903f00ffbfe77e7f5f709aa00b7a`, including both MultiIndex orientations. +The intentional behavior change is **fail closed**: unavailable or unprovable +requested data now raises `PriceBasisError` rather than switching basis, returning +empty history, dropping invalid rows or inventing missing OHLC/volume. A missing +Close never selects Adj Close. Provider errors/empty responses also stop the +request, rather than implicitly switching providers. No fallback is retained as +a safety requirement. + +For dividend adjustment, each row uses `factor = Adj Close / Close`. Multiply +that row's Open, High and Low by the factor; use Adj Close directly for Close. +All four prices then share `split_dividend_adjusted`. This is the vendor's +dividend-aware research series, not an audited reinvested wealth index or an +executable price. `return_basis` and dividend cash accounting remain separate. +Do not book dividends twice when using this series for return research. + +Volume is declared as `volume_basis=provider_native_unscaled`: retain the vendor +value and units exactly, with no adapter split/dividend rescaling. Native zero +volume remains zero; missing, nonfinite or negative volume is unavailable and +fails. This label does not certify physical corporate-action normalization or +cross-vendor volume equivalence. + +The IBKR proxy selects exactly one endpoint while consuming the existing pinned +QPK loader. For split-only, it skips QPK's initial ADJUSTED_LAST probe and permits +TRADES. For split+dividend, ADJUSTED_LAST is passed to IBKR; an empty/error response +is remembered and its typed failure is re-raised if QPK attempts TRADES. Native +OHLC fields are checked before QPK's legacy conversion can synthesize missing +values. This policy belongs to this platform's signal adapter; shared QPK callers +and dependency pins are unchanged. + +`HistoricalCandles` carries batch metadata without changing candle dictionaries +or numeric values. Custom injected loaders must accept `price_basis`, return +this batch with each OHLC field's declared basis and preserve volume. Plain, +unlabelled or mixed-basis results fail. yfinance's known native column contract +is checked against optional `frame.attrs['field_price_basis']` declarations; +invalid OHLC ranges and ambiguous multi-symbol columns also fail. Range checks +alone cannot infer an undocumented vendor adjustment method. + +Success provenance (`qsl.signal_data_provenance.v2`) and logs include requested +basis, resolved basis, provider, source columns and volume basis. Typed failures +include the same fields, a null resolved basis and a bounded reason code. They +never log provider exception text, prices, account data or credentials. Legacy +recorder input without recognized metadata is labelled `unknown_legacy` for +inventory only; it is never an accepted request value or a loader success. + +This config controls historical loaders. External indicators and feature +snapshots retain their separate contracts; setting it does not rebuild or +certify their upstream inputs. It does not approve any deploy or replace a +cloud-only, frozen paired strategy comparison. + +Related proposals: [QPK #678](https://github.com/QuantStrategyLab/QuantPlatformKit/pull/678) +and [UESP #542](https://github.com/QuantStrategyLab/UsEquitySnapshotPipelines/pull/542). +Needs owner review before IBKR deploy. diff --git a/runtime_config_support.py b/runtime_config_support.py index 24b1955..53ad188 100644 --- a/runtime_config_support.py +++ b/runtime_config_support.py @@ -7,6 +7,8 @@ from pathlib import Path from typing import Any, Mapping, Callable +from application.price_basis import validate_price_basis + from quant_platform_kit.cloud import get_secret_store from quant_platform_kit.common.runtime_config import ( first_non_empty, @@ -172,6 +174,7 @@ class PlatformRuntimeSettings: strategy_config_source: str | None reconciliation_output_path: str | None dry_run_only: bool + price_basis: str | None = None ib_gateway_project_id: str | None = None feature_snapshot_fallback_mode: str | None = None feature_snapshot_fallback_cache_dir: str | None = None @@ -302,6 +305,10 @@ def load_platform_runtime_settings( logger: Callable[[str], None] = print, secret_client_factory: Callable[[], Any] | None = None, ) -> PlatformRuntimeSettings: + # Validate before any project/config clients (including secret stores). + configured_price_basis = os.getenv("PRICE_BASIS") + if configured_price_basis is not None: + configured_price_basis = validate_price_basis(configured_price_basis) project_id = project_id_resolver() account_group = resolve_account_group(os.getenv("ACCOUNT_GROUP")) group_config = load_account_group_config( @@ -413,6 +420,7 @@ def load_platform_runtime_settings( strategy_config_source=runtime_paths.strategy_config_source, reconciliation_output_path=runtime_paths.reconciliation_output_path, dry_run_only=resolve_dry_run_env(os.environ, "IBKR_DRY_RUN_ONLY"), + price_basis=configured_price_basis, feature_snapshot_fallback_mode=first_non_empty( os.getenv("IBKR_FEATURE_SNAPSHOT_FALLBACK_MODE"), os.getenv("FEATURE_SNAPSHOT_FALLBACK_MODE"), diff --git a/strategy_runtime.py b/strategy_runtime.py index da702cf..1cbefb7 100644 --- a/strategy_runtime.py +++ b/strategy_runtime.py @@ -47,6 +47,7 @@ ) from quant_platform_kit.risk.contracts import RuntimeRiskLimits, SmallAccountRiskHoldPolicy from runtime_config_support import PlatformRuntimeSettings +from application.price_basis import DEFAULT_PRICE_BASIS, PriceBasisError, validate_price_basis from us_equity_strategies.signals import resolve_external_market_signal_inputs from strategy_loader import ( load_strategy_definition, @@ -848,6 +849,8 @@ def _build_historical_close_map( "historical_price_fallback_failed | " f"profile={self.profile} symbol={symbol} error_type={type(exc).__name__} error={exc}" ) + if isinstance(exc, PriceBasisError): + raise continue latest_close = self._extract_latest_positive_close(price_history) if latest_close is not None: @@ -1768,6 +1771,9 @@ def load_strategy_runtime( merged_runtime_config = dict(entrypoint.manifest.default_config) merged_runtime_config.update(runtime_config) + merged_runtime_config["price_basis"] = validate_price_basis( + merged_runtime_config.get("price_basis", DEFAULT_PRICE_BASIS) + ) strategy_cash_reserve_ratio = float( merged_runtime_config.get( "execution_cash_reserve_ratio", @@ -1805,6 +1811,8 @@ def load_strategy_runtime( def _build_runtime_overrides(runtime_settings: PlatformRuntimeSettings) -> dict[str, Any]: overrides: dict[str, Any] = {} + if getattr(runtime_settings, "price_basis", None) is not None: + overrides["price_basis"] = validate_price_basis(runtime_settings.price_basis) reserved_cash_floor_usd = getattr(runtime_settings, "reserved_cash_floor_usd", 0.0) reserved_cash_ratio = getattr(runtime_settings, "reserved_cash_ratio", None) if float(reserved_cash_floor_usd or 0.0) > 0.0: diff --git a/tests/fixtures/price_basis_main_golden.json b/tests/fixtures/price_basis_main_golden.json new file mode 100644 index 0000000..4fb60e1 --- /dev/null +++ b/tests/fixtures/price_basis_main_golden.json @@ -0,0 +1,67 @@ +{ + "source_sha": "8e173b2bd1cb903f00ffbfe77e7f5f709aa00b7a", + "source_path": "application/runtime_strategy_adapters.py", + "synthetic_input": { + "Open": [ + 101.125, + 50.05, + 0.3 + ], + "High": [ + 103.75, + 52.0, + 0.4 + ], + "Low": [ + 99.125, + 49.99, + 0.2 + ], + "Close": [ + 102.3, + 50.125, + 0.35 + ], + "Adj Close": [ + 98.1, + 49.0, + 0.35 + ], + "Volume": [ + 10000.0, + 0.0, + 99.0 + ] + }, + "dates": [ + "2024-06-03", + "2024-06-04", + "2024-06-05" + ], + "main_output": [ + { + "as_of": "2024-06-03T00:00:00", + "open": "4059480000000000", + "high": "4059f00000000000", + "low": "4058c80000000000", + "close": "4059933333333333", + "volume": "40c3880000000000" + }, + { + "as_of": "2024-06-04T00:00:00", + "open": "4049066666666666", + "high": "404a000000000000", + "low": "4048feb851eb851f", + "close": "4049100000000000", + "volume": "0000000000000000" + }, + { + "as_of": "2024-06-05T00:00:00", + "open": "3fd3333333333333", + "high": "3fd999999999999a", + "low": "3fc999999999999a", + "close": "3fd6666666666666", + "volume": "4058c00000000000" + } + ] +} diff --git a/tests/test_price_basis_contract.py b/tests/test_price_basis_contract.py new file mode 100644 index 0000000..39eedf4 --- /dev/null +++ b/tests/test_price_basis_contract.py @@ -0,0 +1,282 @@ +import json +import struct +import sys +from pathlib import Path +from types import SimpleNamespace + +import pandas as pd +import pytest + +from application import signal_provenance +from application.price_basis import ( + HistoricalCandles, PriceBasis, PriceBasisError, validate_candles, validate_price_basis, +) +from application.runtime_strategy_adapters import ( + BasisHistoricalIB, _coerce_yfinance_candles, build_runtime_strategy_adapters, + fetch_yfinance_historical_candles, +) + + +def frame(): + # Synthetic dividend: split-only drops from 100 to 99, while adjusted + # history is continuous at 99. No real market data is used. + return pd.DataFrame({ + "Open": [100.0, 99.0], "High": [102.0, 101.0], "Low": [98.0, 97.0], + "Close": [100.0, 99.0], "Adj Close": [99.0, 99.0], "Volume": [1000.0, 1500.0], + }, index=pd.to_datetime(["2024-06-03", "2024-06-04"])) + + +def adapters(*, basis="split_adjusted", fallback=None, **kwargs): + from quant_platform_kit.ibkr.market_data import fetch_historical_price_candles, fetch_historical_price_series + def contract(*_a, **_k): + return SimpleNamespace() + return build_runtime_strategy_adapters( + strategy_runtime=SimpleNamespace(merged_runtime_config={"price_basis": basis}), + strategy_profile="synthetic", translator=lambda key: key, pacing_sec=0, + resolve_run_as_of_date_fn=lambda: None, + fetch_historical_price_series_fn=lambda ib, symbol, **kw: fetch_historical_price_series( + ib, symbol, stock_factory=contract, **kw), + fetch_historical_price_candles_fn=lambda ib, symbol, **kw: fetch_historical_price_candles( + ib, symbol, stock_factory=contract, **kw), + map_strategy_decision_fn=lambda *_a, **_kw: None, + fallback_historical_candles_fn=fallback, **kwargs, + ) + + +class FakeIB: + def __init__(self, *, empty=False, error=False): + self.calls = [] + self.empty = empty + self.error = error + + def qualifyContracts(self, contract): # noqa: N802 + return [contract] + + def reqHistoricalData(self, _contract, **kwargs): # noqa: N802 + self.calls.append(kwargs["whatToShow"]) + if self.error: + raise RuntimeError("SENSITIVE_PROVIDER_EXCEPTION_MUST_NOT_APPEAR") + if self.empty: + return [] + return [SimpleNamespace(date=pd.Timestamp("2024-06-03"), open=100.0, high=102.0, + low=98.0, close=100.0, volume=1000.0)] + + +def test_missing_close_never_uses_adjusted_close(): + frame = pd.DataFrame({"Open": [100.0], "Adj Close": [90.0]}, + index=pd.to_datetime(["2024-06-03"])) + with pytest.raises(PriceBasisError, match="missing_column:Close"): + _coerce_yfinance_candles(frame) + + +@pytest.mark.parametrize("basis", list(PriceBasis)) +def test_each_enum_is_valid_configuration(basis): + assert validate_price_basis(basis.value) == basis.value + assert adapters(basis=basis.value).price_basis == basis.value + + +@pytest.mark.parametrize("basis", ["", "adjusted", "split_only", "unknown_legacy", "SPLIT_ADJUSTED", None, 1, True]) +def test_invalid_configuration(basis): + with pytest.raises(ValueError, match="price_basis must be one of"): + validate_price_basis(basis) + + +@pytest.mark.parametrize("basis,reason", [("raw_unadjusted", "raw_prices_not_provided"), + ("total_return_index", "total_return_index_is_not_ohlc")]) +def test_unsupported_ohlc_fails_before_provider_access(basis, reason, monkeypatch): + monkeypatch.setitem(sys.modules, "yfinance", SimpleNamespace(download=lambda *_a, **_k: pytest.fail("provider called"))) + with pytest.raises(PriceBasisError, match=reason): + fetch_yfinance_historical_candles("SYN", price_basis=basis) + with pytest.raises(PriceBasisError, match=reason): + BasisHistoricalIB(None, basis) + with pytest.raises(PriceBasisError, match=reason): + adapters(basis=basis).get_historical_candles(None, "SYN") + + +def test_dividend_basis_requires_adj_close(): + with pytest.raises(PriceBasisError, match="missing_column:Adj Close"): + _coerce_yfinance_candles(frame().drop(columns="Adj Close"), price_basis="split_dividend_adjusted") + + +@pytest.mark.parametrize("column", ["Open", "High", "Low", "Volume"]) +def test_ohlcv_columns_must_not_be_fabricated(column): + with pytest.raises(PriceBasisError, match="missing_column:" + column): + _coerce_yfinance_candles(frame().drop(columns=column)) + + +@pytest.mark.parametrize("column,value", [("Close", 0), ("Close", float("nan")), ("Close", float("inf")), + ("Adj Close", float("nan")), ("Volume", float("nan")), ("Volume", -1)]) +def test_invalid_row_never_switches_basis_or_disappears(column, value): + data = frame() + data.loc[data.index[0], column] = value + with pytest.raises(PriceBasisError): + _coerce_yfinance_candles(data, price_basis="split_dividend_adjusted") + + +def test_dividend_factor_applies_to_all_prices_not_volume(): + data = frame() + split = _coerce_yfinance_candles(data) + adjusted = _coerce_yfinance_candles(data, price_basis="split_dividend_adjusted") + assert [row["close"] for row in split] == [100.0, 99.0] + assert [row["close"] for row in adjusted] == [99.0, 99.0] + for key in ("open", "high", "low"): + assert adjusted[0][key] == split[0][key] * 0.99 + assert adjusted[1][key] == split[1][key] + assert [row["volume"] for row in adjusted] == [1000.0, 1500.0] + assert adjusted.diagnostic["volume_basis"] == "provider_native_unscaled" + + +def test_dividend_ohlc_boundaries_allow_only_float_roundoff(): + data = frame() + for key in ("Open", "High", "Low", "Close"): + data[key] = [0.14, 0.14] + data["Adj Close"] = [0.11, 0.11] + result = _coerce_yfinance_candles(data, price_basis="split_dividend_adjusted") + assert result[0]["close"] == 0.11 + assert result[0]["low"] == 0.11000000000000001 + data["Low"] = [0.14001, 0.14] + with pytest.raises(PriceBasisError, match="inconsistent_ohlc"): + _coerce_yfinance_candles(data, price_basis="split_dividend_adjusted") + + +def test_four_for_one_split_is_continuous_and_is_not_raw(): + raw = [400.0, 100.0] # Contemporaneous units on either side of a 4:1 split. + data = frame() + for key in ("Open", "High", "Low", "Close", "Adj Close"): + data[key] = [raw[0] / 4, raw[1]] + split = _coerce_yfinance_candles(data) + assert [row["close"] for row in split] == [100.0, 100.0] + assert raw[0] / raw[1] == 4 + with pytest.raises(PriceBasisError, match="raw_prices_not_provided"): + _coerce_yfinance_candles(data, price_basis="raw_unadjusted") + + +def test_mixed_source_field_metadata_is_rejected(): + data = frame() + data.attrs["field_price_basis"] = {"Close": "split_dividend_adjusted", "Open": "split_adjusted"} + with pytest.raises(PriceBasisError, match="mixed_or_mismatched_basis"): + _coerce_yfinance_candles(data) + + +def test_mixed_custom_provider_ohlc_metadata_is_rejected(): + candles = _coerce_yfinance_candles(frame()) + candles.field_basis["close"] = "split_dividend_adjusted" + with pytest.raises(PriceBasisError, match="mixed_or_mismatched_basis"): + validate_candles(candles, "split_adjusted") + + +def test_inconsistent_ohlc_geometry_is_rejected(): + data = frame() + data["Close"] = [200.0, 99.0] + with pytest.raises(PriceBasisError, match="inconsistent_ohlc"): + _coerce_yfinance_candles(data) + + +@pytest.mark.parametrize("multiindex", [None, "field_first", "ticker_first"]) +def test_default_matches_frozen_current_main_float_bytes(multiindex): + fixture = json.loads((Path(__file__).parent / "fixtures/price_basis_main_golden.json").read_text()) + data = pd.DataFrame(fixture["synthetic_input"], index=pd.to_datetime(fixture["dates"])) + if multiindex: + data.columns = pd.MultiIndex.from_tuples([ + (key, "SYN") if multiindex == "field_first" else ("SYN", key) for key in data.columns]) + result = _coerce_yfinance_candles(data) + actual = [{"as_of": row["as_of"].isoformat(), **{ + key: struct.pack("!d", row[key]).hex() for key in ("open", "high", "low", "close", "volume")}} + for row in result] + assert fixture["source_sha"] == "8e173b2bd1cb903f00ffbfe77e7f5f709aa00b7a" + assert actual == fixture["main_output"] + + +def test_multiple_symbols_never_selects_first_silently(): + data = pd.concat({"SYN": frame(), "OTHER": frame()}, axis=1) + with pytest.raises(PriceBasisError, match="ambiguous_columns"): + _coerce_yfinance_candles(data) + + +@pytest.mark.parametrize("basis,endpoint", [("split_adjusted", "TRADES"), + ("split_dividend_adjusted", "ADJUSTED_LAST")]) +@pytest.mark.parametrize("method", ["get_historical_close", "get_historical_candles"]) +def test_native_qpk_integration_uses_exact_endpoint(basis, endpoint, method): + ib = FakeIB() + result = getattr(adapters(basis=basis), method)(ib, "SYN") + assert len(result) == 1 + assert ib.calls == [endpoint] + details = signal_provenance.snapshot()["symbols"]["SYN"]["basis_details"] + assert details["requested_basis"] == details["resolved_basis"] == basis + assert details["provider"] == "ibkr" + assert endpoint in details["source_columns"] + + +@pytest.mark.parametrize("method", ["get_historical_close", "get_historical_candles"]) +def test_adjusted_last_empty_never_retries_trades(method): + ib = FakeIB(empty=True) + with pytest.raises(PriceBasisError, match="empty_ADJUSTED_LAST") as error: + getattr(adapters(basis="split_dividend_adjusted"), method)(ib, "SYN") + assert ib.calls == ["ADJUSTED_LAST"] + assert error.value.diagnostic["resolved_basis"] is None + + +def test_native_empty_trades_is_diagnosable(): + with pytest.raises(PriceBasisError, match="empty_TRADES"): + adapters().get_historical_close(FakeIB(empty=True), "SYN") + + +def test_provider_errors_do_not_disclose_exception_text(caplog): + with pytest.raises(PriceBasisError, match="provider_request_failed") as error: + adapters(basis="split_dividend_adjusted").get_historical_close(FakeIB(error=True), "SYN") + assert "SENSITIVE_PROVIDER" not in str(error.value) + caplog.text + assert set(error.value.diagnostic) == { + "requested_basis", "resolved_basis", "provider", "source_columns", "volume_basis", "reason"} + + +@pytest.mark.parametrize("result", [[], [{"close": 1.0}]]) +def test_unlabelled_fallback_is_not_relabelled(result): + with pytest.raises(PriceBasisError, match="missing_basis_metadata"): + adapters(fallback=lambda *_a, **_kw: result).get_historical_close(None, "SYN") + + +def test_explicit_fallback_mismatch_is_rejected(): + candles = _coerce_yfinance_candles(frame(), price_basis="split_dividend_adjusted") + with pytest.raises(PriceBasisError, match="mixed_or_mismatched_basis"): + adapters(fallback=lambda *_a, **_kw: candles).get_historical_close(None, "SYN") + + +def test_requested_basis_reaches_wrapped_provider(monkeypatch): + seen = {} + def download(*_a, **kw): + seen.update(kw) + return frame() + monkeypatch.setitem(sys.modules, "yfinance", SimpleNamespace(download=download)) + loaded = adapters(basis="split_dividend_adjusted", fallback=fetch_yfinance_historical_candles) + assert list(loaded.get_historical_close(None, "SYN")) == [99.0, 99.0] + assert seen["auto_adjust"] is False + assert signal_provenance.snapshot()["symbols"]["SYN"]["basis_details"]["provider"] == "yfinance" + + +def test_fail_closed_is_not_swallowed_by_runtime_price_map(): + from strategy_runtime import LoadedStrategyRuntime + runtime = LoadedStrategyRuntime(entrypoint=SimpleNamespace(manifest=SimpleNamespace(profile="synthetic")), + runtime_adapter=SimpleNamespace(), runtime_settings=SimpleNamespace(), + logger=lambda _message: None) + def fail(*_a, **_kw): + raise PriceBasisError("split_adjusted", provider="yfinance", reason="missing_column:Close") + with pytest.raises(PriceBasisError, match="missing_column:Close"): + runtime._build_historical_close_map(None, fail, ("SYN",)) + + +def test_native_ohlc_is_checked_before_qpk_can_fabricate_fields(): + class CloseOnlyIB(FakeIB): + def reqHistoricalData(self, _contract, **kwargs): # noqa: N802 + self.calls.append(kwargs["whatToShow"]) + return [SimpleNamespace(date=pd.Timestamp("2024-06-03"), close=100.0)] + with pytest.raises(PriceBasisError, match="missing_native_ohlcv"): + adapters().get_historical_candles(CloseOnlyIB(), "SYN") + + +@pytest.mark.parametrize("basis", ["split_adjusted", "split_dividend_adjusted"]) +def test_yfinance_empty_does_not_switch_provider(basis, monkeypatch): + monkeypatch.setitem(sys.modules, "yfinance", SimpleNamespace(download=lambda *_a, **_kw: pd.DataFrame())) + ib = FakeIB() + with pytest.raises(PriceBasisError, match="empty_history"): + adapters(basis=basis, fallback=fetch_yfinance_historical_candles).get_historical_close(ib, "SYN") + assert ib.calls == [] diff --git a/tests/test_runtime_config_support.py b/tests/test_runtime_config_support.py index 8c23865..a36cd67 100644 --- a/tests/test_runtime_config_support.py +++ b/tests/test_runtime_config_support.py @@ -168,6 +168,7 @@ def test_load_platform_runtime_settings_uses_minimal_group_config(monkeypatch): settings = load_platform_runtime_settings(project_id_resolver=lambda: "project-1") assert settings.project_id == "project-1" + assert settings.price_basis is None # Effective runtime default is split_adjusted. assert settings.execution_backend == EXECUTION_BACKEND_GATEWAY assert settings.ib_gateway_instance_name == "ib-gateway" assert settings.ib_gateway_project_id is None @@ -2162,3 +2163,22 @@ def test_load_trusted_runtime_risk_policy_from_runtime_target_json(monkeypatch): def test_load_trusted_runtime_risk_policy_absent_when_missing(monkeypatch): monkeypatch.setenv("RUNTIME_TARGET_JSON", json.dumps({"platform_id": "ibkr"})) assert runtime_config_support_module._load_trusted_runtime_risk_policy() is None + + +@pytest.mark.parametrize("basis", ["raw_unadjusted", "split_adjusted", "split_dividend_adjusted", "total_return_index"]) +def test_price_basis_environment_accepts_only_the_shared_enum(monkeypatch, basis): + monkeypatch.setenv("PRICE_BASIS", basis) + monkeypatch.setenv("RUNTIME_TARGET_JSON", runtime_target_json(SAMPLE_STRATEGY_PROFILE)) + monkeypatch.setenv("ACCOUNT_GROUP", "paper") + monkeypatch.setenv("IB_ACCOUNT_GROUP_CONFIG_JSON", MINIMAL_GROUP_JSON) + settings = load_platform_runtime_settings(project_id_resolver=lambda: None) + assert settings.price_basis == basis + + +@pytest.mark.parametrize("basis", ["", "adjusted", "unknown_legacy", "split_only"]) +def test_price_basis_invalid_environment_fails_before_any_clients(monkeypatch, basis): + monkeypatch.setenv("PRICE_BASIS", basis) + def forbidden(): + pytest.fail("client resolution must not occur before price_basis validation") + with pytest.raises(ValueError, match="price_basis must be one of"): + load_platform_runtime_settings(project_id_resolver=forbidden, secret_client_factory=forbidden) diff --git a/tests/test_runtime_strategy_adapters.py b/tests/test_runtime_strategy_adapters.py index cc68f4a..320daba 100644 --- a/tests/test_runtime_strategy_adapters.py +++ b/tests/test_runtime_strategy_adapters.py @@ -2,9 +2,26 @@ import pandas as pd +from application.price_basis import HistoricalCandles + from application.runtime_strategy_adapters import _coerce_yfinance_candles, build_runtime_strategy_adapters +def _candle(day, close): + return {"as_of": pd.Timestamp(day), "open": close, "high": close, "low": close, + "close": close, "volume": 0.0} + + +class _FakeIB: + def reqHistoricalData(self, *_args, **_kwargs): # noqa: N802 + return [object()] + + +def _native_series(ib, points): + ib.reqHistoricalData("synthetic", whatToShow="TRADES") + return SimpleNamespace(points=points) + + def test_strategy_plugin_signals_are_loaded_reported_and_rendered(): observed = {} @@ -71,10 +88,10 @@ def fail_broker_history(*_args, **_kwargs): resolve_run_as_of_date_fn=lambda: None, fetch_historical_price_series_fn=fail_broker_history, fetch_historical_price_candles_fn=lambda *_args, **_kwargs: (), - fallback_historical_candles_fn=lambda symbol, **_kwargs: [ - {"as_of": pd.Timestamp("2026-05-21"), "close": 100.0}, - {"as_of": pd.Timestamp("2026-05-22"), "close": 101.0}, - ], + fallback_historical_candles_fn=lambda symbol, **_kwargs: HistoricalCandles([ + _candle("2026-05-21", 100.0), + _candle("2026-05-22", 101.0), + ], price_basis="split_adjusted", provider="synthetic", source_columns=("Close",)), map_strategy_decision_fn=lambda *_args, **_kwargs: (), ) @@ -88,7 +105,8 @@ def test_historical_candles_use_fallback_before_ibkr_history(): def fail_broker_history(*_args, **_kwargs): raise AssertionError("broker historical candles should not be called when fallback has data") - fallback = [{"as_of": pd.Timestamp("2026-05-22"), "close": 101.0}] + fallback = HistoricalCandles([_candle("2026-05-22", 101.0)], price_basis="split_adjusted", + provider="synthetic", source_columns=("OHLC",)) adapters = build_runtime_strategy_adapters( strategy_runtime=SimpleNamespace(), strategy_profile="tqqq_growth_income", @@ -104,25 +122,23 @@ def fail_broker_history(*_args, **_kwargs): assert adapters.get_historical_candles("fake-ib", "QQQ") == fallback -def test_historical_close_uses_ibkr_history_when_fallback_is_empty(): +def test_historical_close_uses_ibkr_history_when_fallback_is_disabled(): adapters = build_runtime_strategy_adapters( strategy_runtime=SimpleNamespace(), strategy_profile="tqqq_growth_income", translator=lambda key, **_kwargs: key, pacing_sec=0.0, resolve_run_as_of_date_fn=lambda: None, - fetch_historical_price_series_fn=lambda *_args, **_kwargs: SimpleNamespace( - points=( + fetch_historical_price_series_fn=lambda ib, *_args, **_kwargs: _native_series(ib, ( SimpleNamespace(as_of=pd.Timestamp("2026-05-21"), close=99.0), SimpleNamespace(as_of=pd.Timestamp("2026-05-22"), close=100.0), - ) - ), + )), fetch_historical_price_candles_fn=lambda *_args, **_kwargs: (), - fallback_historical_candles_fn=lambda symbol, **_kwargs: (), + fallback_historical_candles_fn=None, map_strategy_decision_fn=lambda *_args, **_kwargs: (), ) - history = adapters.get_historical_close("fake-ib", "QQQ") + history = adapters.get_historical_close(_FakeIB(), "QQQ") assert list(history) == [99.0, 100.0] diff --git a/tests/test_signal_provenance.py b/tests/test_signal_provenance.py index 1523b65..ce434ce 100644 --- a/tests/test_signal_provenance.py +++ b/tests/test_signal_provenance.py @@ -8,6 +8,21 @@ from application import runtime_strategy_adapters as rsa +def _candle(): + return {"as_of": datetime(2026, 10, 9), "open": 3.0, "high": 3.0, "low": 3.0, + "close": 3.0, "volume": 0.0} + + +class _FakeIB: + def reqHistoricalData(self, *_args, **_kwargs): # noqa: N802 + return [SimpleNamespace(**_candle())] + + +def _native(ib, result): + ib.reqHistoricalData("synthetic", whatToShow="TRADES") + return result + + @pytest.fixture(autouse=True) def _reset_provenance(): signal_provenance.reset() @@ -22,8 +37,8 @@ def _adapters(*, fallback, series_points=(), candles=()): translator=lambda key, **_kw: key, pacing_sec=0.0, resolve_run_as_of_date_fn=lambda: None, - fetch_historical_price_series_fn=lambda *_a, **_k: SimpleNamespace(points=tuple(series_points)), - fetch_historical_price_candles_fn=lambda *_a, **_k: list(candles), + fetch_historical_price_series_fn=lambda ib, *_a, **_k: _native(ib, SimpleNamespace(points=tuple(series_points))), + fetch_historical_price_candles_fn=lambda ib, *_a, **_k: _native(ib, list(candles)), map_strategy_decision_fn=lambda *_a, **_k: (), fallback_historical_candles_fn=fallback, ) @@ -36,23 +51,23 @@ def _frame(columns): def test_recorder_keeps_only_non_sensitive_fields(): - signal_provenance.record("soxx", source="yfinance", price_basis="dividend_adjusted", + signal_provenance.record("soxx", source="yfinance", price_basis="split_dividend_adjusted", dates=[datetime(2024, 10, 10), datetime(2026, 10, 9)]) - signal_provenance.record("SOXX", source="yfinance", price_basis="dividend_adjusted", + signal_provenance.record("SOXX", source="yfinance", price_basis="split_dividend_adjusted", dates=[datetime(2024, 10, 10), datetime(2026, 10, 9)]) snap = signal_provenance.snapshot() - assert snap["schema_version"] == "qsl.signal_data_provenance.v1" + assert snap["schema_version"] == "qsl.signal_data_provenance.v2" entry = snap["symbols"]["SOXX"] - assert entry == {"source": "yfinance", "price_basis": "dividend_adjusted", "rows": 2, + assert entry == {"source": "yfinance", "price_basis": "split_dividend_adjusted", "rows": 2, "first_date": "2024-10-10", "last_date": "2026-10-09", "fetch_count": 2} def test_recorder_normalizes_unknown_basis_and_never_raises(): signal_provenance.record("QQQ", source="x", price_basis="weird", dates=[1_728_518_400_000]) - signal_provenance.record(None, source="x", price_basis="split_only", dates=[]) - signal_provenance.record("BAD", source="x", price_basis="split_only", dates=object()) # not iterable + signal_provenance.record(None, source="x", price_basis="split_adjusted", dates=[]) + signal_provenance.record("BAD", source="x", price_basis="split_adjusted", dates=object()) # not iterable snap = signal_provenance.snapshot() - assert snap["symbols"]["QQQ"]["price_basis"] == "unknown" + assert snap["symbols"]["QQQ"]["price_basis"] == "unknown_legacy" assert snap["symbols"]["QQQ"]["first_date"] == "2024-10-10" assert "BAD" not in snap["symbols"] signal_provenance.reset() @@ -64,36 +79,37 @@ def test_yfinance_signal_uses_split_only_close_not_adj_close(): frame["Adj Close"] = [9.0, 10.0, 11.0] candles = rsa._coerce_yfinance_candles(frame, symbol="SOXL") assert [c["close"] for c in candles] == [10.0, 11.0, 12.0] # RS-08 fix 3: Close, not Adj Close - assert rsa.yfinance_price_basis("soxl") == "split_only" + assert rsa.yfinance_price_basis("soxl") == "split_adjusted" -def test_yfinance_adj_close_only_used_when_close_missing(): +def test_yfinance_adj_close_never_replaces_missing_close(): frame = _frame(["Open", "Adj Close"]) - candles = rsa._coerce_yfinance_candles(frame, symbol="QQQ") - assert len(candles) == 3 - assert rsa.yfinance_price_basis("QQQ") == "dividend_adjusted" + with pytest.raises(rsa.PriceBasisError, match="missing_column:Close"): + rsa._coerce_yfinance_candles(frame, symbol="QQQ") def test_yfinance_close_only_is_labelled_split_only(): - rsa._coerce_yfinance_candles(_frame(["Open", "Close"]), symbol="SOXX") - assert rsa.yfinance_price_basis("SOXX") == "split_only" + rsa._coerce_yfinance_candles(_frame(["Open", "High", "Low", "Close", "Volume"]), symbol="SOXX") + assert rsa.yfinance_price_basis("SOXX") == "split_adjusted" def test_adapter_records_yfinance_fallback_history(monkeypatch): - frame = _frame(["Close", "Adj Close"]) + frame = _frame(["Open", "High", "Low", "Close", "Adj Close", "Volume"]) - def fake_fetch(symbol, *, duration, bar_size): - return rsa._coerce_yfinance_candles(frame, symbol=symbol) + def fake_fetch(symbol, *, duration, bar_size, price_basis): + return rsa._coerce_yfinance_candles(frame, symbol=symbol, price_basis=price_basis) monkeypatch.setattr(rsa, "fetch_yfinance_historical_candles", fake_fetch) adapters = _adapters(fallback=fake_fetch) # identity check uses the module attribute, so the fake counts as the yfinance provider closes = adapters.get_historical_close(None, "SOXL", duration="420 D") - candles = adapters.get_historical_candles(None, "QQQ") + candles = adapters.get_historical_candles(_FakeIB(), "QQQ") assert list(closes.values) == [10.0, 11.0, 12.0] assert len(candles) == 3 symbols = signal_provenance.snapshot()["symbols"] - assert symbols["SOXL"] == {"source": "yfinance", "price_basis": "split_only", "rows": 3, + details = symbols["SOXL"].pop("basis_details") + assert details["requested_basis"] == details["resolved_basis"] == "split_adjusted" + assert symbols["SOXL"] == {"source": "yfinance", "price_basis": "split_adjusted", "rows": 3, "first_date": "2026-10-07", "last_date": "2026-10-09", "fetch_count": 1} assert symbols["QQQ"]["source"] == "yfinance" @@ -101,38 +117,38 @@ def fake_fetch(symbol, *, duration, bar_size): def test_adapter_records_ibkr_history_when_fallback_empty(): points = [SimpleNamespace(as_of=datetime(2026, 10, 8), close=1.0), SimpleNamespace(as_of=datetime(2026, 10, 9), close=2.0)] - candles = [{"as_of": datetime(2026, 10, 9), "close": 3.0}] - adapters = _adapters(fallback=lambda *_a, **_k: [], series_points=points, candles=candles) - closes = adapters.get_historical_close(None, "SOXX") - got = adapters.get_historical_candles(None, "QQQ") + candles = [_candle()] + adapters = _adapters(fallback=None, series_points=points, candles=candles) + closes = adapters.get_historical_close(_FakeIB(), "SOXX") + got = adapters.get_historical_candles(_FakeIB(), "QQQ") assert list(closes.values) == [1.0, 2.0] assert got == candles symbols = signal_provenance.snapshot()["symbols"] - assert symbols["SOXX"]["source"] == "ibkr_historical" - assert symbols["SOXX"]["price_basis"] == "split_only" + assert symbols["SOXX"]["source"] == "ibkr" + assert symbols["SOXX"]["price_basis"] == "split_adjusted" assert symbols["SOXX"]["rows"] == 2 - assert "TRADES" in symbols["SOXX"]["note"] + assert "TRADES" in symbols["SOXX"]["basis_details"]["source_columns"] assert symbols["QQQ"]["rows"] == 1 -def test_generator_candles_are_not_consumed(): +def test_generator_candles_are_validated_before_use(): def gen(): - yield {"as_of": datetime(2026, 10, 9), "close": 3.0} + yield _candle() adapters = rsa.build_runtime_strategy_adapters( strategy_runtime=SimpleNamespace(), strategy_profile="tqqq_growth_income", translator=lambda key, **_kw: key, pacing_sec=0.0, resolve_run_as_of_date_fn=lambda: None, fetch_historical_price_series_fn=lambda *_a, **_k: SimpleNamespace(points=()), - fetch_historical_price_candles_fn=lambda *_a, **_k: gen(), + fetch_historical_price_candles_fn=lambda ib, *_a, **_k: _native(ib, gen()), map_strategy_decision_fn=lambda *_a, **_k: (), - fallback_historical_candles_fn=lambda *_a, **_k: [], + fallback_historical_candles_fn=None, ) - result = adapters.get_historical_candles(None, "QQQ") + result = adapters.get_historical_candles(_FakeIB(), "QQQ") assert len(list(result)) == 1 def test_main_attaches_provenance_to_report(strategy_module): - signal_provenance.record("SOXL", source="yfinance", price_basis="dividend_adjusted", + signal_provenance.record("SOXL", source="yfinance", price_basis="split_dividend_adjusted", dates=[datetime(2026, 10, 9)]) report = {"diagnostics": {"result": "ok"}} strategy_module.attach_signal_data_provenance(report) @@ -149,19 +165,19 @@ def test_ibkr_history_requests_go_through_split_only_proxy(): def fetch_series(ib, symbol, **_kw): seen.append(ib) - return SimpleNamespace(points=()) + return _native(ib, SimpleNamespace(points=(SimpleNamespace(as_of=datetime(2026, 10, 9), close=3.0),))) adapters = rsa.build_runtime_strategy_adapters( strategy_runtime=SimpleNamespace(), strategy_profile="soxl_soxx_trend_income", translator=lambda key, **_kw: key, pacing_sec=0.0, resolve_run_as_of_date_fn=lambda: None, fetch_historical_price_series_fn=fetch_series, - fetch_historical_price_candles_fn=lambda ib, *_a, **_k: seen.append(ib) or [], + fetch_historical_price_candles_fn=lambda ib, *_a, **_k: seen.append(ib) or _native(ib, [_candle()]), map_strategy_decision_fn=lambda *_a, **_k: (), - fallback_historical_candles_fn=lambda *_a, **_k: [], + fallback_historical_candles_fn=None, ) - adapters.get_historical_close("raw-ib", "SOXX") - adapters.get_historical_candles("raw-ib", "QQQ") - assert all(isinstance(ib, rsa.SplitOnlyHistoricalIB) for ib in seen) and len(seen) == 2 + adapters.get_historical_close(_FakeIB(), "SOXX") + adapters.get_historical_candles(_FakeIB(), "QQQ") + assert all(isinstance(ib, rsa.BasisHistoricalIB) for ib in seen) and len(seen) == 2 def test_split_only_proxy_suppresses_adjusted_last_and_delegates_rest(): @@ -177,7 +193,7 @@ def reqHistoricalData(self, contract, **kwargs): # noqa: N802 def qualifyContracts(self, contract): # noqa: N802 return [contract] - proxy = rsa.SplitOnlyHistoricalIB(FakeIB()) + proxy = rsa.BasisHistoricalIB(FakeIB()) assert proxy.reqHistoricalData("c", whatToShow="ADJUSTED_LAST") == [] assert proxy.reqHistoricalData("c", whatToShow="TRADES") == ["bar"] assert proxy.qualifyContracts("c") == ["c"] @@ -199,6 +215,6 @@ def reqHistoricalData(self, contract, **kwargs): # noqa: N802 return [SimpleNamespace(date=datetime(2026, 10, 9), close=5.0)] series = market_data.fetch_historical_price_series( - rsa.SplitOnlyHistoricalIB(FakeIB()), "SOXX", stock_factory=lambda *a, **k: SimpleNamespace()) + rsa.BasisHistoricalIB(FakeIB()), "SOXX", stock_factory=lambda *a, **k: SimpleNamespace()) assert asked == ["TRADES"] assert [p.close for p in series.points] == [5.0] diff --git a/tests/test_strategy_runtime.py b/tests/test_strategy_runtime.py index e23ae1e..e30984d 100644 --- a/tests/test_strategy_runtime.py +++ b/tests/test_strategy_runtime.py @@ -1974,3 +1974,38 @@ def test_cash_only_runtime_overrides_force_option_overlays_off(): assert overrides["option_growth_overlay_enabled"] is False assert overrides["option_income_overlay_enabled"] is False assert overrides["cash_reserve_ratio"] == 0.03 + + +@pytest.mark.parametrize("file_value,env_value,expected", [ + ({}, None, "split_adjusted"), + ({"price_basis": "split_dividend_adjusted"}, None, "split_dividend_adjusted"), + ({"price_basis": "split_dividend_adjusted"}, "split_adjusted", "split_adjusted"), +]) +def test_price_basis_config_merge_reaches_historical_adapters(monkeypatch, file_value, env_value, expected): + from application.runtime_strategy_adapters import build_runtime_strategy_adapters + entrypoint = SimpleNamespace(manifest=SimpleNamespace( + profile="synthetic", required_inputs=frozenset({"feature_snapshot"}), default_config={})) + monkeypatch.setattr(strategy_runtime_module, "load_strategy_definition", lambda _raw: SimpleNamespace(profile="synthetic")) + monkeypatch.setattr(strategy_runtime_module, "load_strategy_entrypoint_for_profile", lambda _raw: entrypoint) + monkeypatch.setattr(strategy_runtime_module, "load_strategy_runtime_adapter_for_profile", lambda _raw: + StrategyRuntimeAdapter(runtime_parameter_loader=lambda **_kw: file_value)) + runtime = strategy_runtime_module.load_strategy_runtime( + "synthetic", runtime_settings=replace(_build_runtime_settings(), price_basis=env_value), + logger=lambda _message: None) + adapter = build_runtime_strategy_adapters( + strategy_runtime=runtime, strategy_profile="synthetic", translator=None, pacing_sec=0, + resolve_run_as_of_date_fn=None, fetch_historical_price_series_fn=None, + fetch_historical_price_candles_fn=None, map_strategy_decision_fn=None) + assert runtime.merged_runtime_config["price_basis"] == adapter.price_basis == expected + + +def test_price_basis_invalid_file_config_fails_at_runtime_validation(monkeypatch): + entrypoint = SimpleNamespace(manifest=SimpleNamespace( + profile="synthetic", required_inputs=frozenset({"feature_snapshot"}), default_config={})) + monkeypatch.setattr(strategy_runtime_module, "load_strategy_definition", lambda _raw: SimpleNamespace(profile="synthetic")) + monkeypatch.setattr(strategy_runtime_module, "load_strategy_entrypoint_for_profile", lambda _raw: entrypoint) + monkeypatch.setattr(strategy_runtime_module, "load_strategy_runtime_adapter_for_profile", lambda _raw: + StrategyRuntimeAdapter(runtime_parameter_loader=lambda **_kw: {"price_basis": "adjusted"})) + with pytest.raises(ValueError, match="price_basis must be one of"): + strategy_runtime_module.load_strategy_runtime( + "synthetic", runtime_settings=_build_runtime_settings(), logger=lambda _message: None)