From ac1fa54bb629429b15502c951bb59daefbcdefad Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 24 Sep 2026 09:28:16 +0800 Subject: [PATCH] feat(research): persist trial outcomes and daily ledger Co-Authored-By: Codex --- .../strategy_lifecycle/contracts.py | 479 ++++++++++++++ .../strategy_lifecycle/performance_store.py | 424 ++++++++++++ tests/test_lifecycle_performance_store.py | 603 ++++++++++++++++++ 3 files changed, 1506 insertions(+) diff --git a/src/quant_platform_kit/strategy_lifecycle/contracts.py b/src/quant_platform_kit/strategy_lifecycle/contracts.py index 014c43d8..9c58c6f8 100644 --- a/src/quant_platform_kit/strategy_lifecycle/contracts.py +++ b/src/quant_platform_kit/strategy_lifecycle/contracts.py @@ -3,6 +3,8 @@ from __future__ import annotations import enum +import json +import math from dataclasses import dataclass, field from datetime import date from typing import Any, Mapping @@ -772,6 +774,483 @@ def to_dict(self) -> dict[str, object]: } +# ── Research trial ledger ─────────────────────────────────────────── +# Independent of BacktestResult / promotion. Synthetic storage is not a grant. + + +_IDENTITY_PLACEHOLDERS = frozenset({"unknown", "default", "none", "null", "na", "n/a"}) +_RESEARCH_NON_SUCCESS = frozenset({"failed", "rejected", "aborted"}) + + +class ResearchTrialStatus(str, enum.Enum): + """Lifecycle of one research attempt. Terminal states do not promote.""" + + STARTED = "started" + SUCCEEDED = "succeeded" + FAILED = "failed" + REJECTED = "rejected" + ABORTED = "aborted" + + +def _reject_identity_placeholder(value: str) -> None: + if value.casefold() in _IDENTITY_PLACEHOLDERS: + raise ValueError("identity_placeholder") + + +def _label(value: object) -> str: + if not isinstance(value, str) or not value or value != value.strip(): + raise ValueError("identity") + if len(value) > 200 or any(ord(char) < 32 or ord(char) == 127 for char in value): + raise ValueError("identity") + _reject_identity_placeholder(value) + return value + + +def _raw_identity(value: object) -> str: + if not isinstance(value, str) or not value or any(char.isspace() for char in value): + raise ValueError("identity") + if len(value) > 500 or any(ord(char) < 32 or ord(char) == 127 for char in value): + raise ValueError("identity") + _reject_identity_placeholder(value) + return value + + +def _run_identity(value: object) -> str: + """Backtest run ids may contain internal spaces; they are not reason text.""" + + if not isinstance(value, str) or not value or value != value.strip(): + raise ValueError("identity") + if len(value) > 500 or any(ord(char) < 32 or ord(char) == 127 for char in value): + raise ValueError("identity") + if any(char.isspace() and char != " " for char in value): + raise ValueError("identity") + _reject_identity_placeholder(value) + return value + + +def _require_date(value: object) -> date: + if type(value) is not date: + raise ValueError("window") + return value + + +def _finite_number(value: object) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError("invalid_number") + number = float(value) + if not math.isfinite(number): + raise ValueError("invalid_number") + return number + + +def _reason_code(value: object, *, allow_empty: bool) -> str: + if not isinstance(value, str): + raise ValueError("reason_code") + if value == "" and allow_empty: + return "" + alphabet = "abcdefghijklmnopqrstuvwxyz0123456789_" + if ( + not value + or len(value) > 64 + or value[0] not in "abcdefghijklmnopqrstuvwxyz" + or any(char not in alphabet for char in value) + ): + raise ValueError("reason_code") + return value + + +def _cost_inputs(value: object, *, allow_empty: bool) -> dict[str, float]: + if isinstance(value, (str, bytes)) or not isinstance(value, Mapping): + raise ValueError("cost_inputs") + parsed: dict[str, float] = {} + for key, item in value.items(): + if not isinstance(key, str) or not key or any(char.isspace() for char in key): + raise ValueError("cost_inputs") + number = _finite_number(item) + if number < 0: + raise ValueError("cost_inputs") + parsed[key] = number + if not parsed and not allow_empty: + raise ValueError("cost_inputs") + return parsed + + +def _symbol(value: object) -> str: + if not isinstance(value, str) or not value or len(value) > 32: + raise ValueError("position_mark") + if any(not (char.isalnum() or char in "._-") for char in value): + raise ValueError("position_mark") + return value + + +@dataclass(frozen=True) +class ResearchPositionMark: + """One position's quantity and marked value. Zero quantity has zero value.""" + + symbol: str + quantity: float + valuation: float + + def __post_init__(self) -> None: + symbol = _symbol(self.symbol) + quantity = _finite_number(self.quantity) + valuation = _finite_number(self.valuation) + if (quantity == 0.0) != (valuation == 0.0): + raise ValueError("position_mark") + object.__setattr__(self, "symbol", symbol) + object.__setattr__(self, "quantity", quantity) + object.__setattr__(self, "valuation", valuation) + + def to_dict(self) -> dict[str, object]: + return { + "symbol": self.symbol, + "quantity": self.quantity, + "valuation": self.valuation, + } + + +def _position_marks(value: object) -> tuple[ResearchPositionMark, ...]: + if isinstance(value, (str, bytes)) or not isinstance(value, (tuple, list)): + raise ValueError("position_mark") + marks: list[ResearchPositionMark] = [] + seen: set[str] = set() + for item in value: + if not isinstance(item, ResearchPositionMark): + raise ValueError("position_mark") + if item.symbol in seen: + raise ValueError("position_mark") + seen.add(item.symbol) + marks.append(item) + return tuple(marks) + + +@dataclass(frozen=True) +class ResearchLedgerDay: + """End-of-session cash, marks, flows, fees, NAV, and that session's return.""" + + session_date: date + cash: float + positions: tuple[ResearchPositionMark, ...] + trade_net_cashflow: float + fees: float + nav: float + daily_return: float + + def __post_init__(self) -> None: + session_date = _require_date(self.session_date) + cash = _finite_number(self.cash) + positions = _position_marks(self.positions) + trade_net_cashflow = _finite_number(self.trade_net_cashflow) + fees = _finite_number(self.fees) + if fees < 0: + raise ValueError("ledger_fee") + nav = _finite_number(self.nav) + if nav <= 0: + raise ValueError("ledger_nav") + daily_return = _finite_number(self.daily_return) + expected_nav = cash + sum(mark.valuation for mark in positions) + if not math.isclose(nav, expected_nav, rel_tol=0.0, abs_tol=1e-9): + raise ValueError("ledger_nav") + object.__setattr__(self, "session_date", session_date) + object.__setattr__(self, "cash", cash) + object.__setattr__(self, "positions", positions) + object.__setattr__(self, "trade_net_cashflow", trade_net_cashflow) + object.__setattr__(self, "fees", fees) + object.__setattr__(self, "nav", nav) + object.__setattr__(self, "daily_return", daily_return) + + def to_dict(self) -> dict[str, object]: + return { + "session_date": self.session_date.isoformat(), + "cash": self.cash, + "positions": [mark.to_dict() for mark in self.positions], + "trade_net_cashflow": self.trade_net_cashflow, + "fees": self.fees, + "nav": self.nav, + "daily_return": self.daily_return, + } + + +@dataclass(frozen=True) +class ResearchDailyLedger: + """Complete daily book for one research trial. + + ``initial_session_date`` is the first marked session. Its cash, positions, + and NAV are not a return observation. ``days`` are strictly later sessions + and each carries that session's return. ``window_start`` is the initial + session; ``observation_count`` counts only those later return days. + """ + + trial_id: str + domain: str + strategy_profile: str + run_id: str + param_version: int + input_id: str + calendar_id: str + periods_per_year: float + cost_source: str + cost_inputs: Mapping[str, float] + initial_session_date: date + initial_nav: float + initial_cash: float + initial_positions: tuple[ResearchPositionMark, ...] + days: tuple[ResearchLedgerDay, ...] + synthetic: bool + + def __post_init__(self) -> None: + trial_id = _raw_identity(self.trial_id) + domain = _label(self.domain) + strategy_profile = _label(self.strategy_profile) + run_id = _run_identity(self.run_id) + if type(self.param_version) is not int or self.param_version <= 0: + raise ValueError("param_version") + input_id = _raw_identity(self.input_id) + calendar_id = _label(self.calendar_id) + number = _finite_number(self.periods_per_year) + try: + periods = validate_periods_per_year(number) + except ValueError as exc: + raise ValueError("periods_per_year") from exc + cost_source = _raw_identity(self.cost_source) + cost_inputs = _cost_inputs(self.cost_inputs, allow_empty=False) + initial_session_date = _require_date(self.initial_session_date) + initial_nav = _finite_number(self.initial_nav) + initial_cash = _finite_number(self.initial_cash) + initial_positions = _position_marks(self.initial_positions) + if initial_nav <= 0: + raise ValueError("ledger_nav") + expected_initial = initial_cash + sum(mark.valuation for mark in initial_positions) + if not math.isclose(initial_nav, expected_initial, rel_tol=0.0, abs_tol=1e-9): + raise ValueError("ledger_nav") + if isinstance(self.days, (str, bytes)) or not isinstance(self.days, (tuple, list)) or not self.days: + raise ValueError("ledger_dates") + days = tuple(self.days) + previous_date = initial_session_date + for day in days: + if not isinstance(day, ResearchLedgerDay): + raise ValueError("ledger_dates") + if day.session_date <= previous_date: + raise ValueError("ledger_dates") + previous_date = day.session_date + if type(self.synthetic) is not bool: + raise ValueError("synthetic") + previous_cash = initial_cash + previous_nav = initial_nav + for day in days: + expected_cash = previous_cash + day.trade_net_cashflow - day.fees + if not math.isclose(day.cash, expected_cash, rel_tol=0.0, abs_tol=1e-9): + raise ValueError("ledger_cash") + expected_return = day.nav / previous_nav - 1.0 + if day.daily_return != expected_return: + raise ValueError("ledger_return") + previous_cash = day.cash + previous_nav = day.nav + object.__setattr__(self, "trial_id", trial_id) + object.__setattr__(self, "domain", domain) + object.__setattr__(self, "strategy_profile", strategy_profile) + object.__setattr__(self, "run_id", run_id) + object.__setattr__(self, "input_id", input_id) + object.__setattr__(self, "calendar_id", calendar_id) + object.__setattr__(self, "periods_per_year", periods) + object.__setattr__(self, "cost_source", cost_source) + object.__setattr__(self, "cost_inputs", cost_inputs) + object.__setattr__(self, "initial_session_date", initial_session_date) + object.__setattr__(self, "initial_nav", initial_nav) + object.__setattr__(self, "initial_cash", initial_cash) + object.__setattr__(self, "initial_positions", initial_positions) + object.__setattr__(self, "days", days) + + @property + def window_start(self) -> date: + return self.initial_session_date + + @property + def window_end(self) -> date: + return self.days[-1].session_date + + @property + def observation_count(self) -> int: + return len(self.days) + + @property + def total_return(self) -> float: + return self.days[-1].nav / self.initial_nav - 1.0 + + @property + def total_fees(self) -> float: + return float(sum(day.fees for day in self.days)) + + def to_dict(self) -> dict[str, object]: + return { + "trial_id": self.trial_id, + "domain": self.domain, + "strategy_profile": self.strategy_profile, + "run_id": self.run_id, + "param_version": self.param_version, + "input_id": self.input_id, + "calendar_id": self.calendar_id, + "periods_per_year": self.periods_per_year, + "cost_source": self.cost_source, + "cost_inputs": dict(self.cost_inputs), + "initial_session_date": self.initial_session_date.isoformat(), + "initial_nav": self.initial_nav, + "initial_cash": self.initial_cash, + "initial_positions": [mark.to_dict() for mark in self.initial_positions], + "days": [day.to_dict() for day in self.days], + "synthetic": self.synthetic, + } + + +def _actual_params(value: object, *, required: bool) -> dict[str, Any] | None: + if value is None: + if required: + raise ValueError("actual_config_unknown") + return None + if type(value) is not dict: + raise ValueError("actual_params") + try: + parsed = json.loads(json.dumps(value, sort_keys=True, allow_nan=False)) + except (TypeError, ValueError) as exc: + raise ValueError("actual_params") from exc + if type(parsed) is not dict or any(type(key) is not str or not key for key in parsed): + raise ValueError("actual_params") + return parsed + + +def _optional_identity(value: object, *, required: bool) -> str | None: + if value is None: + if required: + raise ValueError("identity") + return None + return _raw_identity(value) + + +@dataclass(frozen=True) +class ResearchTrialRecord: + """One research attempt. Unknown actual params stay null, never a filler. + + Failed, rejected, and aborted trials store no run or return metric. They + may keep known cost inputs. Succeeded points at one stored result and ledger. + """ + + trial_id: str + domain: str + strategy_profile: str + status: ResearchTrialStatus + candidate_config_id: str + actual_params: Mapping[str, Any] | None + param_set_id: str | None + source_revision: str | None + input_id: str + window_start: date + window_end: date + calendar_id: str + periods_per_year: float + cost_source: str | None + cost_inputs: Mapping[str, float] + reason_code: str + synthetic: bool + run_id: str | None + param_version: int | None + + def __post_init__(self) -> None: + status = self.status + if isinstance(status, str) and not isinstance(status, ResearchTrialStatus): + try: + status = ResearchTrialStatus(status) + except ValueError as exc: + raise ValueError("status") from exc + if not isinstance(status, ResearchTrialStatus): + raise ValueError("status") + succeeded = status is ResearchTrialStatus.SUCCEEDED + trial_id = _raw_identity(self.trial_id) + domain = _label(self.domain) + strategy_profile = _label(self.strategy_profile) + candidate_config_id = _raw_identity(self.candidate_config_id) + actual_params = _actual_params(self.actual_params, required=succeeded) + param_set_id = _optional_identity(self.param_set_id, required=succeeded) + source_revision = _optional_identity(self.source_revision, required=succeeded) + input_id = _raw_identity(self.input_id) + window_start = _require_date(self.window_start) + window_end = _require_date(self.window_end) + if window_end < window_start: + raise ValueError("window") + calendar_id = _label(self.calendar_id) + try: + periods = validate_periods_per_year(_finite_number(self.periods_per_year)) + except ValueError as exc: + raise ValueError("periods_per_year") from exc + cost_source = _optional_identity(self.cost_source, required=succeeded) + if type(self.synthetic) is not bool: + raise ValueError("synthetic") + cost_inputs = _cost_inputs(self.cost_inputs, allow_empty=not succeeded) + if succeeded: + if not isinstance(self.run_id, str): + raise ValueError("research_trial_result_link") + run_id: str | None = _run_identity(self.run_id) + if type(self.param_version) is not int or self.param_version <= 0: + raise ValueError("param_version") + param_version: int | None = self.param_version + reason_code = _reason_code(self.reason_code, allow_empty=True) + if reason_code != "": + raise ValueError("reason_code") + else: + if self.run_id is not None or self.param_version is not None: + raise ValueError("research_trial_result_link") + run_id = None + param_version = None + if status is ResearchTrialStatus.STARTED: + reason_code = _reason_code(self.reason_code, allow_empty=True) + if reason_code != "": + raise ValueError("reason_code") + elif status.value not in _RESEARCH_NON_SUCCESS: + raise ValueError("status") + else: + reason_code = _reason_code(self.reason_code, allow_empty=False) + object.__setattr__(self, "status", status) + object.__setattr__(self, "trial_id", trial_id) + object.__setattr__(self, "domain", domain) + object.__setattr__(self, "strategy_profile", strategy_profile) + object.__setattr__(self, "candidate_config_id", candidate_config_id) + object.__setattr__(self, "actual_params", actual_params) + object.__setattr__(self, "param_set_id", param_set_id) + object.__setattr__(self, "source_revision", source_revision) + object.__setattr__(self, "input_id", input_id) + object.__setattr__(self, "window_start", window_start) + object.__setattr__(self, "window_end", window_end) + object.__setattr__(self, "calendar_id", calendar_id) + object.__setattr__(self, "periods_per_year", periods) + object.__setattr__(self, "cost_source", cost_source) + object.__setattr__(self, "cost_inputs", cost_inputs) + object.__setattr__(self, "reason_code", reason_code) + object.__setattr__(self, "run_id", run_id) + object.__setattr__(self, "param_version", param_version) + + def to_dict(self) -> dict[str, object]: + return { + "trial_id": self.trial_id, + "domain": self.domain, + "strategy_profile": self.strategy_profile, + "status": self.status.value, + "candidate_config_id": self.candidate_config_id, + "actual_params": None if self.actual_params is None else dict(self.actual_params), + "param_set_id": self.param_set_id, + "source_revision": self.source_revision, + "input_id": self.input_id, + "window_start": self.window_start.isoformat(), + "window_end": self.window_end.isoformat(), + "calendar_id": self.calendar_id, + "periods_per_year": self.periods_per_year, + "cost_source": self.cost_source, + "cost_inputs": dict(self.cost_inputs), + "reason_code": self.reason_code, + "synthetic": self.synthetic, + "run_id": self.run_id, + "param_version": self.param_version, + } + + # ── Safe Update ───────────────────────────────────────────────────── diff --git a/src/quant_platform_kit/strategy_lifecycle/performance_store.py b/src/quant_platform_kit/strategy_lifecycle/performance_store.py index 5f80893d..526c7c53 100644 --- a/src/quant_platform_kit/strategy_lifecycle/performance_store.py +++ b/src/quant_platform_kit/strategy_lifecycle/performance_store.py @@ -6,10 +6,16 @@ gs://{bucket}/daily/{domain}/{strategy}/{date}.json gs://{bucket}/backtest/{domain}/{strategy}/backtest_v{n}_{stamp}.json gs://{bucket}/backtest/{domain}/{strategy}/runs/{run_digest}/backtest_v{n}.json + gs://{bucket}/research_trial/{identity_digest}/started.json + gs://{bucket}/research_trial/{identity_digest}/terminal.json + gs://{bucket}/research_trial/{identity_digest}/ledger.json gs://{bucket}/drift/{domain}/{strategy}/drift_{date}.json gs://{bucket}/optimization/{domain}/{strategy}/proposal_v{n}_{stamp}.json gs://{bucket}/dashboard/aggregated_health.json gs://{bucket}/audit/updates/{strategy}/{entry_id}.json + +Research objects use one digest of the original domain, profile, and trial id. +When cloud_bucket is set, that bucket is the only research authority. """ from __future__ import annotations @@ -17,6 +23,7 @@ import hashlib import json import math +import os import tempfile from collections.abc import Mapping, Sequence from dataclasses import dataclass @@ -30,6 +37,11 @@ BacktestValidationIdentity, DriftResult, OptimizationProposal, + ResearchDailyLedger, + ResearchLedgerDay, + ResearchPositionMark, + ResearchTrialRecord, + ResearchTrialStatus, StrategyHealthScore, StrategyPerformanceSnapshot, UpdateLogEntry, @@ -365,6 +377,216 @@ def load_latest_backtest(self, domain: str, strategy_profile: str) -> BacktestRe selected.sort(key=lambda item: item[0]) return selected[-1][1] + # ── research trials ────────────────────────────────────────── + # One backend: cloud when cloud_bucket is set, otherwise local exclusive + # create. A cloud or local readback is not proof of the other backend. + + def _research_key(self, domain: str, strategy_profile: str, trial_id: str, name: str) -> str: + material = f"{domain}\0{strategy_profile}\0{trial_id}".encode() + digest = hashlib.sha256(material).hexdigest() + return f"research_trial/{digest}/{name}.json" + + def save_research_ledger(self, ledger: ResearchDailyLedger) -> None: + """Create the one ledger for this trial. An existing identical object is kept.""" + + if type(ledger) is not ResearchDailyLedger: + raise ValueError("research_ledger_malformed") + self._create_same( + self._research_key(ledger.domain, ledger.strategy_profile, ledger.trial_id, "ledger"), + _research_payload(ledger), + ) + + def load_research_ledger( + self, + domain: str, + strategy_profile: str, + trial_id: str, + run_id: str, + param_version: int, + ) -> ResearchDailyLedger | None: + if type(param_version) is not int or param_version <= 0: + return None + ledger = _research_ledger_from_dict( + self._read_research_json(self._research_key(domain, strategy_profile, trial_id, "ledger")) + ) + if ( + ledger is None + or ledger.domain != domain + or ledger.strategy_profile != strategy_profile + or ledger.trial_id != trial_id + or ledger.run_id != run_id + or ledger.param_version != param_version + ): + return None + return ledger + + def save_research_trial(self, trial: ResearchTrialRecord) -> None: + """Create started or one terminal. Succeeded is stored only after result and ledger.""" + + if type(trial) is not ResearchTrialRecord: + raise ValueError("research_trial_malformed") + name = "started" if trial.status is ResearchTrialStatus.STARTED else "terminal" + other = "terminal" if name == "started" else "started" + self._require_research_pair(trial, other) + if trial.status is ResearchTrialStatus.SUCCEEDED: + problem = self._research_success_problem(trial) + if problem is not None: + raise ValueError(problem) + self._create_same( + self._research_key(trial.domain, trial.strategy_profile, trial.trial_id, name), + _research_payload(trial), + ) + + def load_research_trial( + self, + domain: str, + strategy_profile: str, + trial_id: str, + *, + run_id: str | None = None, + param_version: int | None = None, + ) -> ResearchTrialRecord | None: + started_text = self._research_text(self._research_key(domain, strategy_profile, trial_id, "started")) + terminal_text = self._research_text(self._research_key(domain, strategy_profile, trial_id, "terminal")) + started = _research_trial_from_dict(_research_object(started_text)) if started_text is not None else None + terminal = _research_trial_from_dict(_research_object(terminal_text)) if terminal_text is not None else None + if terminal_text is not None: + if ( + terminal is None + or terminal.domain != domain + or terminal.strategy_profile != strategy_profile + or terminal.trial_id != trial_id + or ( + started_text is not None + and (started is None or not _research_trial_continues(started, terminal)) + ) + ): + return None + record = terminal + else: + record = started + if record is None or record.domain != domain or record.strategy_profile != strategy_profile or record.trial_id != trial_id: + return None + if record.status is ResearchTrialStatus.SUCCEEDED and self._research_success_problem(record) is not None: + return None + if run_id is not None and record.run_id != run_id: + return None + if param_version is not None and (type(param_version) is not int or record.param_version != param_version): + return None + return record + + def _research_success_problem(self, trial: ResearchTrialRecord) -> str | None: + if not isinstance(trial.run_id, str) or type(trial.param_version) is not int: + return "research_trial_result_missing" + result = self._read_saved_backtest(trial.domain, trial.strategy_profile, trial.run_id, trial.param_version) + if result is None: + return "research_trial_result_missing" + ledger = self.load_research_ledger( + trial.domain, trial.strategy_profile, trial.trial_id, trial.run_id, trial.param_version + ) + if ledger is None: + if self._research_text(self._research_key(trial.domain, trial.strategy_profile, trial.trial_id, "ledger")) is None: + return "research_trial_ledger_missing" + return "research_trial_result_mismatch" + if not _research_result_matches(result, trial, ledger): + return "research_trial_result_mismatch" + return None + + def _require_research_pair(self, trial: ResearchTrialRecord, other_name: str) -> None: + text = self._research_text(self._research_key(trial.domain, trial.strategy_profile, trial.trial_id, other_name)) + if text is None: + return + other = _research_trial_from_dict(_research_object(text)) + if other is None: + raise ValueError("research_trial_malformed") + started, terminal = (trial, other) if other_name == "terminal" else (other, trial) + if not _research_trial_continues(started, terminal): + raise ValueError("research_trial_conflict") + + def _read_saved_backtest(self, domain: str, strategy_profile: str, run_id: str, param_version: int) -> BacktestResult | None: + key = self._backtest_key( + BacktestResult( + strategy_profile=strategy_profile, + domain=domain, + param_set_id="stored", + params={}, + run_id=run_id, + param_version=param_version, + ) + ) + data = self._read_research_json(key) + if data is None: + return None + return _backtest_from_dict(data) + + def _read_research_json(self, key: str) -> dict[str, Any] | None: + return _research_object(self._research_text(key)) + + def _research_text(self, key: str) -> str | None: + if self.cloud_bucket: + store = self._object_store() + uri = self._cloud_uri(key) + try: + present = bool(store.exists(uri)) + except Exception as exc: + raise ValueError("research_store_unavailable") from exc + if not present: + return None + try: + return str(store.read_text(uri)) + except Exception as exc: + raise ValueError("research_store_unavailable") from exc + path = self._local_path(key) + if not path.exists(): + return None + try: + return path.read_text(encoding="utf-8") + except OSError as exc: + raise ValueError("research_store_unavailable") from exc + + def _create_same(self, key: str, payload: Mapping[str, Any]) -> None: + text = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + if self.cloud_bucket: + self._cloud_create_same(self._cloud_uri(key), text) + return + path = self._local_path(key) + path.parent.mkdir(parents=True, exist_ok=True) + try: + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + except FileExistsError: + self._require_same_local(path, text) + return + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(text) + handle.flush() + os.fsync(handle.fileno()) + + def _cloud_create_same(self, uri: str, text: str) -> None: + store = self._object_store() + try: + present = bool(store.exists(uri)) + except Exception as exc: + raise ValueError("research_store_unavailable") from exc + if not present: + try: + created = bool(store.create_text(uri, text, content_type="application/json")) + except Exception as exc: + raise ValueError("research_store_unavailable") from exc + if created: + return + try: + current = str(store.read_text(uri)) + except Exception as exc: + raise ValueError("research_store_unavailable") from exc + _research_same_text(current, text, ledger=uri.endswith("/ledger.json")) + + def _require_same_local(self, path: Path, text: str) -> None: + try: + current = path.read_text(encoding="utf-8") + except OSError as exc: + raise ValueError("research_store_unavailable") from exc + _research_same_text(current, text, ledger=path.name == "ledger.json") + # ── optimization ───────────────────────────────────────────── def _proposal_key(self, proposal: OptimizationProposal) -> str: @@ -854,3 +1076,205 @@ def _audit_from_dict(data: Mapping[str, Any]) -> UpdateLogEntry | None: ) except Exception: return None + + + +def _research_payload(record: ResearchTrialRecord | ResearchDailyLedger) -> dict[str, Any]: + return {"schema_version": SCHEMA_VERSION, **record.to_dict()} + + +def _research_object(text: str | None) -> dict[str, Any] | None: + if not text: + return None + try: + data = json.loads(text) + except Exception: + return None + return data if isinstance(data, dict) else None + + +def _research_same_text(current: str, expected: str, *, ledger: bool) -> None: + if current == expected: + return + code = "research_ledger_malformed" if ledger else "research_trial_malformed" + parsed = ( + _research_ledger_from_dict(_research_object(current)) + if ledger + else _research_trial_from_dict(_research_object(current)) + ) + if parsed is None: + raise ValueError(code) + raise ValueError("research_trial_conflict") + + +def _research_date(value: object) -> date: + if not isinstance(value, str) or not value: + raise ValueError("window") + parsed = date.fromisoformat(value) + if parsed.isoformat() != value: + raise ValueError("window") + return parsed + + +_POSITION_FIELDS = frozenset({"symbol", "quantity", "valuation"}) +_DAY_FIELDS = frozenset({"session_date", "cash", "positions", "trade_net_cashflow", "fees", "nav", "daily_return"}) +_LEDGER_FIELDS = frozenset({ + "schema_version", "trial_id", "domain", "strategy_profile", "run_id", "param_version", + "input_id", "calendar_id", "periods_per_year", "cost_source", "cost_inputs", + "initial_session_date", "initial_nav", "initial_cash", "initial_positions", "days", "synthetic", +}) +_TRIAL_FIELDS = frozenset({ + "schema_version", "trial_id", "domain", "strategy_profile", "status", "candidate_config_id", + "actual_params", "param_set_id", "source_revision", "input_id", "window_start", "window_end", + "calendar_id", "periods_per_year", "cost_source", "cost_inputs", "reason_code", "synthetic", + "run_id", "param_version", +}) + + +def _research_positions(value: object) -> tuple[ResearchPositionMark, ...]: + if not isinstance(value, list): + raise ValueError("position_mark") + marks: list[ResearchPositionMark] = [] + for item in value: + if not isinstance(item, dict) or set(item) != _POSITION_FIELDS: + raise ValueError("position_mark") + marks.append(ResearchPositionMark(symbol=item["symbol"], quantity=item["quantity"], valuation=item["valuation"])) + return tuple(marks) + + +def _research_days(value: object) -> tuple[ResearchLedgerDay, ...]: + if not isinstance(value, list) or not value: + raise ValueError("ledger_dates") + days: list[ResearchLedgerDay] = [] + for item in value: + if not isinstance(item, dict) or set(item) != _DAY_FIELDS: + raise ValueError("ledger_dates") + days.append(ResearchLedgerDay( + session_date=_research_date(item["session_date"]), + cash=item["cash"], + positions=_research_positions(item["positions"]), + trade_net_cashflow=item["trade_net_cashflow"], + fees=item["fees"], + nav=item["nav"], + daily_return=item["daily_return"], + )) + return tuple(days) + + +def _research_ledger_from_dict(data: Mapping[str, Any] | None) -> ResearchDailyLedger | None: + if not isinstance(data, dict) or set(data) != _LEDGER_FIELDS or data.get("schema_version") != SCHEMA_VERSION: + return None + try: + return ResearchDailyLedger( + trial_id=data["trial_id"], + domain=data["domain"], + strategy_profile=data["strategy_profile"], + run_id=data["run_id"], + param_version=data["param_version"], + input_id=data["input_id"], + calendar_id=data["calendar_id"], + periods_per_year=data["periods_per_year"], + cost_source=data["cost_source"], + cost_inputs=data["cost_inputs"], + initial_session_date=_research_date(data["initial_session_date"]), + initial_nav=data["initial_nav"], + initial_cash=data["initial_cash"], + initial_positions=_research_positions(data["initial_positions"]), + days=_research_days(data["days"]), + synthetic=data["synthetic"], + ) + except Exception: + return None + + +def _research_trial_from_dict(data: Mapping[str, Any] | None) -> ResearchTrialRecord | None: + if not isinstance(data, dict) or set(data) != _TRIAL_FIELDS or data.get("schema_version") != SCHEMA_VERSION: + return None + try: + return ResearchTrialRecord( + trial_id=data["trial_id"], + domain=data["domain"], + strategy_profile=data["strategy_profile"], + status=data["status"], + candidate_config_id=data["candidate_config_id"], + actual_params=data["actual_params"], + param_set_id=data["param_set_id"], + source_revision=data["source_revision"], + input_id=data["input_id"], + window_start=_research_date(data["window_start"]), + window_end=_research_date(data["window_end"]), + calendar_id=data["calendar_id"], + periods_per_year=data["periods_per_year"], + cost_source=data["cost_source"], + cost_inputs=data["cost_inputs"], + reason_code=data["reason_code"], + synthetic=data["synthetic"], + run_id=data["run_id"], + param_version=data["param_version"], + ) + except Exception: + return None + + +def _research_trial_continues(started: ResearchTrialRecord, terminal: ResearchTrialRecord) -> bool: + if started.status is not ResearchTrialStatus.STARTED or terminal.status is ResearchTrialStatus.STARTED: + return False + if ( + started.trial_id != terminal.trial_id + or started.domain != terminal.domain + or started.strategy_profile != terminal.strategy_profile + or started.candidate_config_id != terminal.candidate_config_id + or started.input_id != terminal.input_id + or started.window_start != terminal.window_start + or started.window_end != terminal.window_end + or started.calendar_id != terminal.calendar_id + or started.periods_per_year != terminal.periods_per_year + or started.synthetic is not terminal.synthetic + ): + return False + if started.actual_params is not None and started.actual_params != terminal.actual_params: + return False + if started.param_set_id is not None and started.param_set_id != terminal.param_set_id: + return False + if started.source_revision is not None and started.source_revision != terminal.source_revision: + return False + if started.cost_source is not None and started.cost_source != terminal.cost_source: + return False + if started.cost_inputs and dict(started.cost_inputs) != dict(terminal.cost_inputs): + return False + return True + + +def _research_result_matches(result: BacktestResult, trial: ResearchTrialRecord, ledger: ResearchDailyLedger) -> bool: + if trial.actual_params is None or not trial.param_set_id or not trial.source_revision: + return False + return ( + dict(result.params) == dict(trial.actual_params) + and result.param_set_id == trial.param_set_id + and result.source_revision == trial.source_revision + and result.run_id == trial.run_id + and result.param_version == trial.param_version + and result.domain == trial.domain + and result.strategy_profile == trial.strategy_profile + and result.start_date == trial.window_start + and result.end_date == trial.window_end + and result.calendar_id == trial.calendar_id + and result.periods_per_year == trial.periods_per_year + and result.cost_model == trial.cost_source + and dict(result.cost_inputs) == dict(trial.cost_inputs) + and result.observation_count == ledger.observation_count + and result.total_return == ledger.total_return + and ledger.trial_id == trial.trial_id + and ledger.run_id == trial.run_id + and ledger.param_version == trial.param_version + and ledger.domain == trial.domain + and ledger.strategy_profile == trial.strategy_profile + and ledger.input_id == trial.input_id + and ledger.calendar_id == trial.calendar_id + and ledger.periods_per_year == trial.periods_per_year + and ledger.cost_source == trial.cost_source + and dict(ledger.cost_inputs) == dict(trial.cost_inputs) + and ledger.window_start == trial.window_start + and ledger.window_end == trial.window_end + and ledger.synthetic is trial.synthetic + ) diff --git a/tests/test_lifecycle_performance_store.py b/tests/test_lifecycle_performance_store.py index 9094c171..3405b24f 100644 --- a/tests/test_lifecycle_performance_store.py +++ b/tests/test_lifecycle_performance_store.py @@ -2,6 +2,7 @@ from __future__ import annotations +import dataclasses import hashlib import json import tempfile @@ -13,6 +14,11 @@ from quant_platform_kit.strategy_lifecycle.contracts import ( BacktestResult, BacktestValidationIdentity, + ResearchDailyLedger, + ResearchLedgerDay, + ResearchPositionMark, + ResearchTrialRecord, + ResearchTrialStatus, ) from quant_platform_kit.strategy_lifecycle.performance_store import ( DEFAULT_LOCAL_ROOT, @@ -586,5 +592,602 @@ def test_malformed_cost_inputs_or_identity_fail_closed(self) -> None: self.assertIsNone(loaded, run_id) + +_PARAMS = {"lookback": 20} +_COSTS = {"commission_bps": 1.0, "slippage_bps": 0.5} +_INITIAL = date(2024, 1, 1) +_START = date(2024, 1, 2) +_END = date(2024, 1, 3) + + +def _identity_digest(domain: str, profile: str, trial_id: str) -> str: + return hashlib.sha256(f"{domain}\0{profile}\0{trial_id}".encode()).hexdigest() + + +def _research_file(root: Path, domain: str, profile: str, trial_id: str, name: str) -> Path: + return root / "research_trial" / _identity_digest(domain, profile, trial_id) / f"{name}.json" + + +def _mark_day(session: date, cash: float, quantity: float, valuation: float, flow: float, fees: float, nav: float, previous: float) -> ResearchLedgerDay: + positions = () if quantity == 0 else (ResearchPositionMark("SOXL", quantity, valuation),) + return ResearchLedgerDay(session, cash, positions, flow, fees, nav, nav / previous - 1.0) + + +def _ledger(trial_id: str = "trial-a", run_id: str = "run-a", version: int = 1, domain: str = "us_equity", profile: str = "global_etf_rotation") -> ResearchDailyLedger: + return ResearchDailyLedger( + trial_id=trial_id, + domain=domain, + strategy_profile=profile, + run_id=run_id, + param_version=version, + input_id="input-a", + calendar_id="XNYS", + periods_per_year=252.0, + cost_source="synthetic_cost_v1", + cost_inputs=dict(_COSTS), + initial_session_date=_INITIAL, + initial_nav=100.0, + initial_cash=100.0, + initial_positions=(), + days=( + _mark_day(_START, 59, 2, 40, -40, 1, 99, 100), + _mark_day(_END, 59, 2, 50, 0, 0, 109, 99), + ), + synthetic=True, + ) + + +def _trial(status: ResearchTrialStatus, **overrides: object) -> ResearchTrialRecord: + succeeded = status is ResearchTrialStatus.SUCCEEDED + payload: dict[str, object] = { + "trial_id": "trial-a", + "domain": "us_equity", + "strategy_profile": "global_etf_rotation", + "status": status, + "candidate_config_id": "candidate-a", + "actual_params": dict(_PARAMS) if succeeded else None, + "param_set_id": "set-a" if succeeded else None, + "source_revision": "rev-a" if succeeded else None, + "input_id": "input-a", + "window_start": _INITIAL, + "window_end": _END, + "calendar_id": "XNYS", + "periods_per_year": 252.0, + "cost_source": "synthetic_cost_v1", + "cost_inputs": dict(_COSTS), + "reason_code": "" if succeeded or status is ResearchTrialStatus.STARTED else "config_unparsed", + "synthetic": True, + "run_id": "run-a" if succeeded else None, + "param_version": 1 if succeeded else None, + } + payload.update(overrides) + return ResearchTrialRecord(**payload) + + +def _result(ledger: ResearchDailyLedger, **overrides: object) -> BacktestResult: + payload: dict[str, object] = { + "strategy_profile": ledger.strategy_profile, + "domain": ledger.domain, + "param_set_id": "set-a", + "params": dict(_PARAMS), + "param_version": ledger.param_version, + "run_id": ledger.run_id, + "start_date": ledger.window_start, + "end_date": ledger.window_end, + "observation_count": ledger.observation_count, + "total_return": ledger.total_return, + "calendar_id": ledger.calendar_id, + "periods_per_year": ledger.periods_per_year, + "cost_model": ledger.cost_source, + "cost_inputs": dict(ledger.cost_inputs), + "source_revision": "rev-a", + "sharpe_ratio": 1.0, + "computed_at": "2026-01-01T00:00:00Z", + } + payload.update(overrides) + return BacktestResult(**payload) + + +class _ResearchCloud: + """In-memory object store. A successful round trip here is not GCS persistence.""" + + def __init__(self) -> None: + self.objects: dict[str, str] = {} + self.fail_exists = False + self.fail_read = False + self.fail_create = False + self.create_calls = 0 + + def exists(self, uri: str) -> bool: + if self.fail_exists: + raise OSError("exists down") + return uri in self.objects + + def read_text(self, uri: str) -> str: + if self.fail_read: + raise OSError("read down") + return self.objects[uri] + + def read_bytes(self, uri: str) -> bytes: + return self.read_text(uri).encode() + + def write_bytes(self, uri: str, data: bytes, content_type: str = "application/octet-stream") -> str: + self.objects[uri] = data.decode() + return uri + + def create_text(self, uri: str, data: str, content_type: str = "text/plain") -> bool: + self.create_calls += 1 + if self.fail_create: + raise OSError("create down") + if uri in self.objects: + return False + self.objects[uri] = data + return True + + def list(self, prefix: str) -> list[str]: + return [uri for uri in self.objects if uri.startswith(prefix)] + + +class ResearchTrialLedgerStoreTest(unittest.TestCase): + def test_rejected_before_parse_keeps_null_actual_params_and_cost_inputs(self) -> None: + for field in dataclasses.fields(ResearchTrialRecord): + self.assertIs(field.default, dataclasses.MISSING) + self.assertIs(field.default_factory, dataclasses.MISSING) + with self.assertRaises(ValueError) as unknown: + _trial(ResearchTrialStatus.REJECTED, actual_params="unknown") + with self.assertRaises(ValueError) as blank: + _trial(ResearchTrialStatus.REJECTED, actual_params="") + self.assertEqual(str(unknown.exception), "actual_params") + self.assertEqual(str(blank.exception), "actual_params") + trial = _trial(ResearchTrialStatus.REJECTED, actual_params=None, cost_inputs=dict(_COSTS)) + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + store = PerformanceStore(local_root=root) + store.save_research_trial(trial) + loaded = store.load_research_trial("us_equity", "global_etf_rotation", "trial-a") + path = _research_file(root, "us_equity", "global_etf_rotation", "trial-a", "terminal") + payload = json.loads(path.read_text()) + self.assertFalse(any(item.parts[len(root.parts)] == "backtest" for item in root.rglob("*") if item != root)) + self.assertEqual(loaded, trial) + self.assertIsNone(loaded.actual_params) + self.assertIsNone(loaded.run_id) + self.assertIsNone(loaded.param_version) + self.assertEqual(dict(loaded.cost_inputs), _COSTS) + self.assertIsNone(payload["actual_params"]) + self.assertEqual(path.parent.name, _identity_digest("us_equity", "global_etf_rotation", "trial-a")) + self.assertEqual(path.name, "terminal.json") + for banned in ("sharpe_ratio", "total_return", "promotion_eligible"): + self.assertNotIn(banned, payload) + + def test_orphan_ledger_still_allows_failed_and_aborted_records(self) -> None: + started = _trial(ResearchTrialStatus.STARTED, actual_params=None, cost_inputs={}, reason_code="") + changed = _trial(ResearchTrialStatus.STARTED, actual_params=None, cost_inputs={}, reason_code="", input_id="input-b") + failed = _trial(ResearchTrialStatus.FAILED, reason_code="result_rejected", cost_inputs=dict(_COSTS), actual_params=None) + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + store = PerformanceStore(local_root=root) + store.save_research_trial(started) + store.save_research_trial(started) + started_path = _research_file(root, "us_equity", "global_etf_rotation", "trial-a", "started") + started_bytes = started_path.read_text() + with self.assertRaises(ValueError) as changed_error: + store.save_research_trial(changed) + self.assertEqual(str(changed_error.exception), "research_trial_conflict") + self.assertEqual(started_path.read_text(), started_bytes) + store.save_research_ledger(_ledger()) + store.save_research_trial(failed) + loaded = store.load_research_trial("us_equity", "global_etf_rotation", "trial-a") + ledger = store.load_research_ledger("us_equity", "global_etf_rotation", "trial-a", "run-a", 1) + terminal = _research_file(root, "us_equity", "global_etf_rotation", "trial-a", "terminal") + planted = json.loads(terminal.read_text()) + planted["sharpe_ratio"] = 1.2 + terminal.write_text(json.dumps(planted)) + planted_text = terminal.read_text() + self.assertIsNone(store.load_research_trial("us_equity", "global_etf_rotation", "trial-a")) + with self.assertRaises(ValueError) as malformed: + store.save_research_trial(failed) + self.assertEqual(str(malformed.exception), "research_trial_malformed") + self.assertEqual(terminal.read_text(), planted_text) + self.assertIn("sharpe_ratio", terminal.read_text()) + self.assertEqual(len(list(root.rglob("started.json"))), 1) + self.assertEqual(loaded, failed) + self.assertIsNone(loaded.run_id) + self.assertEqual(dict(loaded.cost_inputs), _COSTS) + self.assertEqual(ledger.run_id, "run-a") + with self.assertRaises(ValueError) as linked: + _trial(ResearchTrialStatus.FAILED, run_id="run-a", reason_code="result_rejected") + self.assertEqual(str(linked.exception), "research_trial_result_link") + aborted = _trial(ResearchTrialStatus.ABORTED, trial_id="trial-b", reason_code="operator_abort", cost_inputs=dict(_COSTS)) + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + store = PerformanceStore(local_root=root) + store.save_research_ledger(_ledger(trial_id="trial-b")) + store.save_research_trial(aborted) + loaded_aborted = store.load_research_trial("us_equity", "global_etf_rotation", "trial-b") + self.assertEqual(loaded_aborted, aborted) + self.assertIsNone(loaded_aborted.run_id) + self.assertEqual(dict(loaded_aborted.cost_inputs), _COSTS) + + def test_same_params_keep_distinct_trials(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + store = PerformanceStore(local_root=root) + saved = [] + for trial_id, run_id in (("trial-a", "run-a"), ("trial-b", "run-b")): + ledger = _ledger(trial_id=trial_id, run_id=run_id) + trial = _trial(ResearchTrialStatus.SUCCEEDED, trial_id=trial_id, run_id=run_id) + store.save_backtest_result(_result(ledger)) + store.save_research_ledger(ledger) + store.save_research_trial(trial) + saved.append(trial) + self.assertEqual( + store.load_research_trial("us_equity", "global_etf_rotation", trial_id, run_id=run_id, param_version=1), + trial, + ) + self.assertIsNone(store.load_research_trial("us_equity", "global_etf_rotation", "trial-a", run_id="run-b")) + self.assertIsNone(store.load_research_trial("us_equity", "global_etf_rotation", "trial-a", param_version=2)) + names = {path.relative_to(root).parts[0] for path in root.rglob("*.json")} + self.assertEqual(names, {"research_trial", "backtest"}) + self.assertEqual(saved[0].actual_params, saved[1].actual_params) + self.assertNotEqual(saved[0].trial_id, saved[1].trial_id) + + def test_nul_in_domain_or_profile_is_rejected(self) -> None: + colliding = (("a\0b", "c"), ("a", "b\0c")) + for domain, profile in colliding: + with self.subTest(domain=domain, profile=profile): + with self.assertRaises(ValueError) as caught: + _trial(ResearchTrialStatus.REJECTED, domain=domain, strategy_profile=profile) + self.assertEqual(str(caught.exception), "identity") + + def test_original_identity_does_not_collapse_cleaned_keys(self) -> None: + spaced = _trial(ResearchTrialStatus.REJECTED, domain="us equity", strategy_profile="global etf", actual_params=None) + hyphen = _trial(ResearchTrialStatus.REJECTED, domain="us-equity", strategy_profile="global-etf", actual_params=None) + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + store = PerformanceStore(local_root=root) + store.save_research_trial(spaced) + store.save_research_trial(hyphen) + self.assertEqual(store.load_research_trial("us equity", "global etf", "trial-a"), spaced) + self.assertEqual(store.load_research_trial("us-equity", "global-etf", "trial-a"), hyphen) + self.assertIsNone(store.load_research_trial("us equity", "global-etf", "trial-a")) + self.assertIsNone(store.load_research_trial("us-equity", "global etf", "trial-a")) + path = _research_file(root, "us equity", "global etf", "trial-a", "terminal") + planted = json.loads(path.read_text()) + planted["trial_id"] = "trial-other" + path.write_text(json.dumps(planted)) + self.assertIsNone(store.load_research_trial("us equity", "global etf", "trial-a")) + with self.assertRaises(ValueError) as conflict: + store.save_research_trial(spaced) + self.assertEqual(str(conflict.exception), "research_trial_conflict") + self.assertEqual(json.loads(path.read_text())["trial_id"], "trial-other") + + def test_illegal_ledger_is_rejected_and_planted_cash_is_not_rewritten(self) -> None: + with self.assertRaises(ValueError) as bad_bool: + _mark_day(_START, 59, True, 40, -40, 1, 99, 100) + with self.assertRaises(ValueError) as bad_nan: + _mark_day(_START, float("nan"), 2, 40, -40, 1, 99, 100) + with self.assertRaises(ValueError) as bad_fee: + _mark_day(_START, 59, 2, 40, -40, -1, 99, 100) + self.assertEqual(str(bad_bool.exception), "invalid_number") + self.assertEqual(str(bad_nan.exception), "invalid_number") + self.assertEqual(str(bad_fee.exception), "ledger_fee") + day = _mark_day(_START, 59, 2, 40, -40, 1, 99, 100) + earlier = _mark_day(date(2024, 1, 1), 100, 0, 0, 0, 0, 100, 100) + with self.assertRaises(ValueError) as bad_dates: + ResearchDailyLedger( + trial_id="trial-a", domain="us_equity", strategy_profile="global_etf_rotation", + run_id="run-a", param_version=1, input_id="input-a", calendar_id="XNYS", + periods_per_year=252.0, cost_source="synthetic_cost_v1", cost_inputs=dict(_COSTS), + initial_session_date=_INITIAL, initial_nav=100.0, initial_cash=100.0, + initial_positions=(), days=(day, earlier), synthetic=True, + ) + self.assertEqual(str(bad_dates.exception), "ledger_dates") + with self.assertRaises(ValueError) as same_session: + ResearchDailyLedger( + trial_id="trial-a", domain="us_equity", strategy_profile="global_etf_rotation", + run_id="run-a", param_version=1, input_id="input-a", calendar_id="XNYS", + periods_per_year=252.0, cost_source="synthetic_cost_v1", cost_inputs=dict(_COSTS), + initial_session_date=_START, initial_nav=100.0, initial_cash=100.0, + initial_positions=(), days=(day,), synthetic=True, + ) + self.assertEqual(str(same_session.exception), "ledger_dates") + short_cash = _mark_day(_START, 50, 2, 40, -40, 1, 90, 100) + with self.assertRaises(ValueError) as bad_cash: + ResearchDailyLedger( + trial_id="trial-a", domain="us_equity", strategy_profile="global_etf_rotation", + run_id="run-a", param_version=1, input_id="input-a", calendar_id="XNYS", + periods_per_year=252.0, cost_source="synthetic_cost_v1", cost_inputs=dict(_COSTS), + initial_session_date=_INITIAL, initial_nav=100.0, initial_cash=100.0, + initial_positions=(), days=(short_cash,), synthetic=True, + ) + self.assertEqual(str(bad_cash.exception), "ledger_cash") + broken = _mark_day(_START, 59, 2, 40, -40, 1, 99, 100) + object.__setattr__(broken, "daily_return", 0.0) + with self.assertRaises(ValueError) as bad_return: + ResearchDailyLedger( + trial_id="trial-a", domain="us_equity", strategy_profile="global_etf_rotation", + run_id="run-a", param_version=1, input_id="input-a", calendar_id="XNYS", + periods_per_year=252.0, cost_source="synthetic_cost_v1", cost_inputs=dict(_COSTS), + initial_session_date=_INITIAL, initial_nav=100.0, initial_cash=100.0, + initial_positions=(), days=(broken,), synthetic=True, + ) + self.assertEqual(str(bad_return.exception), "ledger_return") + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + store = PerformanceStore(local_root=root) + store.save_research_ledger(_ledger()) + path = _research_file(root, "us_equity", "global_etf_rotation", "trial-a", "ledger") + planted = json.loads(path.read_text()) + planted["days"][0]["cash"] = 1 + path.write_text(json.dumps(planted)) + planted_text = path.read_text() + self.assertIsNone(store.load_research_ledger("us_equity", "global_etf_rotation", "trial-a", "run-a", 1)) + with self.assertRaises(ValueError) as malformed: + store.save_research_ledger(_ledger()) + self.assertEqual(str(malformed.exception), "research_ledger_malformed") + self.assertEqual(path.read_text(), planted_text) + + def test_terminal_resave_is_identical_only(self) -> None: + trial = _trial(ResearchTrialStatus.REJECTED, actual_params=None, cost_inputs=dict(_COSTS)) + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + store = PerformanceStore(local_root=root) + store.save_research_trial(trial) + store.save_research_trial(trial) + path = _research_file(root, "us_equity", "global_etf_rotation", "trial-a", "terminal") + original = path.read_text() + other = _trial(ResearchTrialStatus.REJECTED, actual_params=None, cost_inputs=dict(_COSTS), reason_code="operator_abort") + with self.assertRaises(ValueError) as conflict: + store.save_research_trial(other) + self.assertEqual(str(conflict.exception), "research_trial_conflict") + self.assertEqual(path.read_text(), original) + + def test_success_is_written_last_and_binds_result_fields(self) -> None: + ledger = _ledger() + result = _result(ledger) + started = _trial(ResearchTrialStatus.STARTED, actual_params=dict(_PARAMS), param_set_id="set-a", source_revision="rev-a", cost_inputs={}) + success = _trial(ResearchTrialStatus.SUCCEEDED) + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + store = PerformanceStore(local_root=root) + store.save_research_trial(started) + with self.assertRaises(ValueError) as missing_result: + store.save_research_trial(success) + self.assertEqual(str(missing_result.exception), "research_trial_result_missing") + self.assertFalse(_research_file(root, "us_equity", "global_etf_rotation", "trial-a", "terminal").exists()) + store.save_backtest_result(result) + with self.assertRaises(ValueError) as missing_ledger: + store.save_research_trial(success) + self.assertEqual(str(missing_ledger.exception), "research_trial_ledger_missing") + self.assertFalse(_research_file(root, "us_equity", "global_etf_rotation", "trial-a", "ledger").exists()) + self.assertFalse(_research_file(root, "us_equity", "global_etf_rotation", "trial-a", "terminal").exists()) + store.save_research_ledger(ledger) + store.save_research_trial(success) + store.save_research_trial(success) + loaded = store.load_research_trial("us_equity", "global_etf_rotation", "trial-a", run_id="run-a", param_version=1) + ledger_path = _research_file(root, "us_equity", "global_etf_rotation", "trial-a", "ledger") + ledger_bytes = ledger_path.read_text() + with self.assertRaises(ValueError) as second_ledger: + store.save_research_ledger(_ledger(run_id="run-b")) + self.assertEqual(str(second_ledger.exception), "research_trial_conflict") + self.assertEqual(ledger_path.read_text(), ledger_bytes) + self.assertTrue(_run_file(root, "run-a").exists()) + self.assertEqual(loaded, success) + self.assertEqual(dict(loaded.actual_params), dict(result.params)) + self.assertEqual(loaded.param_set_id, result.param_set_id) + self.assertEqual(loaded.source_revision, result.source_revision) + self.assertEqual(dict(loaded.cost_inputs), dict(result.cost_inputs)) + self.assertEqual(loaded.calendar_id, result.calendar_id) + self.assertEqual(loaded.run_id, result.run_id) + self.assertEqual((ledger.initial_nav, ledger.days[0].nav, ledger.days[1].nav), (100.0, 99.0, 109.0)) + self.assertEqual(ledger.initial_session_date, _INITIAL) + self.assertEqual(ledger.window_start, _INITIAL) + self.assertEqual(ledger.days[0].session_date, _START) + self.assertEqual(ledger.observation_count, 2) + self.assertEqual(result.start_date, _INITIAL) + self.assertEqual(result.end_date, _END) + self.assertEqual(result.observation_count, 2) + self.assertEqual(loaded.window_start, result.start_date) + self.assertEqual(ledger.observation_count, result.observation_count) + self.assertEqual(ledger.total_return, result.total_return) + + def test_param_or_source_mismatch_does_not_create_terminal(self) -> None: + ledger = _ledger() + cases = { + "params": {"params": {"lookback": 21}}, + "source_revision": {"source_revision": "rev-other"}, + "param_set_id": {"param_set_id": "set-other"}, + "calendar": {"calendar_id": "XNAS"}, + "cost_inputs": {"cost_inputs": {"commission_bps": 9.0}}, + "cost_model": {"cost_model": "other_cost"}, + "observation_count": {"observation_count": 9}, + "total_return": {"total_return": 0.5}, + "window": {"start_date": _START}, + } + for name, overrides in cases.items(): + with self.subTest(name=name): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + store = PerformanceStore(local_root=root) + store.save_backtest_result(_result(ledger, **overrides)) + store.save_research_ledger(ledger) + with self.assertRaises(ValueError) as mismatch: + store.save_research_trial(_trial(ResearchTrialStatus.SUCCEEDED)) + self.assertEqual(str(mismatch.exception), "research_trial_result_mismatch") + self.assertFalse(_research_file(root, "us_equity", "global_etf_rotation", "trial-a", "terminal").exists()) + + def test_terminal_keeps_known_started_identity(self) -> None: + started = _trial( + ResearchTrialStatus.STARTED, + actual_params={"lookback": 20}, + param_set_id=None, + source_revision=None, + cost_source="synthetic_cost_v1", + cost_inputs=dict(_COSTS), + ) + replacements = { + "candidate": {"candidate_config_id": "candidate-b"}, + "input": {"input_id": "input-b"}, + "window": {"window_end": date(2024, 1, 4)}, + "calendar": {"calendar_id": "XNAS"}, + "periods": {"periods_per_year": 365.25}, + "synthetic": {"synthetic": False}, + "actual_params": {"actual_params": {"lookback": 21}}, + "cost_inputs": {"cost_inputs": {"commission_bps": 9.0}}, + "cost_source": {"cost_source": "other_cost"}, + } + filled = _trial( + ResearchTrialStatus.FAILED, + reason_code="result_rejected", + actual_params={"lookback": 20}, + param_set_id="set-a", + source_revision="rev-a", + cost_inputs=dict(_COSTS), + ) + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + store = PerformanceStore(local_root=root) + store.save_research_trial(started) + started_path = _research_file(root, "us_equity", "global_etf_rotation", "trial-a", "started") + started_bytes = started_path.read_text() + for name, overrides in replacements.items(): + with self.subTest(name=name): + payload = { + "reason_code": "result_rejected", + "actual_params": {"lookback": 20}, + "cost_inputs": dict(_COSTS), + } + payload.update(overrides) + with self.assertRaises(ValueError) as conflict: + store.save_research_trial(_trial(ResearchTrialStatus.FAILED, **payload)) + self.assertEqual(str(conflict.exception), "research_trial_conflict") + self.assertFalse(_research_file(root, "us_equity", "global_etf_rotation", "trial-a", "terminal").exists()) + store.save_research_trial(filled) + store.save_research_trial(filled) + store.save_research_trial(started) + self.assertEqual(store.load_research_trial("us_equity", "global_etf_rotation", "trial-a"), filled) + self.assertEqual(started_path.read_text(), started_bytes) + with self.assertRaises(ValueError) as rewritten: + store.save_research_trial(_trial( + ResearchTrialStatus.STARTED, + candidate_config_id="candidate-b", + actual_params={"lookback": 20}, + cost_inputs=dict(_COSTS), + )) + self.assertEqual(str(rewritten.exception), "research_trial_conflict") + self.assertEqual(started_path.read_text(), started_bytes) + self.assertEqual(store.load_research_trial("us_equity", "global_etf_rotation", "trial-a"), filled) + planted = json.loads(_research_file(root, "us_equity", "global_etf_rotation", "trial-a", "terminal").read_text()) + planted["candidate_config_id"] = "candidate-b" + terminal_path = _research_file(root, "us_equity", "global_etf_rotation", "trial-a", "terminal") + terminal_path.write_text(json.dumps(planted)) + self.assertIsNone(store.load_research_trial("us_equity", "global_etf_rotation", "trial-a")) + self.assertEqual(json.loads(terminal_path.read_text())["candidate_config_id"], "candidate-b") + open_started = _trial( + ResearchTrialStatus.STARTED, + trial_id="trial-b", + actual_params=None, + param_set_id=None, + source_revision=None, + cost_source=None, + cost_inputs={}, + ) + completed = _trial( + ResearchTrialStatus.FAILED, + trial_id="trial-b", + reason_code="result_rejected", + actual_params={"lookback": 20}, + param_set_id="set-a", + source_revision="rev-a", + cost_source="synthetic_cost_v1", + cost_inputs=dict(_COSTS), + ) + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + store = PerformanceStore(local_root=root) + store.save_research_trial(completed) + with self.assertRaises(ValueError) as late_started: + store.save_research_trial(_trial( + ResearchTrialStatus.STARTED, + trial_id="trial-b", + candidate_config_id="candidate-b", + cost_inputs=dict(_COSTS), + )) + self.assertEqual(str(late_started.exception), "research_trial_conflict") + self.assertFalse(_research_file(root, "us_equity", "global_etf_rotation", "trial-b", "started").exists()) + self.assertEqual(store.load_research_trial("us_equity", "global_etf_rotation", "trial-b"), completed) + store.save_research_trial(open_started) + store.save_research_trial(open_started) + self.assertEqual(store.load_research_trial("us_equity", "global_etf_rotation", "trial-b"), completed) + + def test_cloud_faults_fail_closed_without_local_fallback(self) -> None: + trial = _trial(ResearchTrialStatus.REJECTED, actual_params=None, cost_inputs=dict(_COSTS)) + digest = _identity_digest("us_equity", "global_etf_rotation", "trial-a") + uri = f"gs://lifecycle-bucket/production/research_trial/{digest}/terminal.json" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + local_path = _research_file(root, "us_equity", "global_etf_rotation", "trial-a", "terminal") + local_path.parent.mkdir(parents=True) + local_path.write_text('{"trial_id":"local-only"}') + cloud = _ResearchCloud() + store = PerformanceStore(cloud_bucket="lifecycle-bucket", cloud_prefix="production", local_root=root) + with patch("quant_platform_kit.strategy_lifecycle.performance_store.get_object_store", return_value=cloud): + cloud.fail_exists = True + with self.assertRaises(ValueError) as exists_error: + store.save_research_trial(trial) + self.assertEqual(str(exists_error.exception), "research_store_unavailable") + self.assertEqual(cloud.create_calls, 0) + with self.assertRaises(ValueError) as load_error: + store.load_research_trial("us_equity", "global_etf_rotation", "trial-a") + self.assertEqual(str(load_error.exception), "research_store_unavailable") + cloud.fail_exists = False + cloud.objects[uri] = "{}" + cloud.fail_read = True + with self.assertRaises(ValueError) as read_error: + store.save_research_trial(trial) + self.assertEqual(str(read_error.exception), "research_store_unavailable") + self.assertEqual(cloud.create_calls, 0) + self.assertEqual(cloud.objects[uri], "{}") + cloud.fail_read = False + cloud.objects.clear() + self.assertIsNone(store.load_research_trial("us_equity", "global_etf_rotation", "trial-a")) + cloud.fail_create = True + with self.assertRaises(ValueError) as create_error: + store.save_research_trial(trial) + self.assertEqual(str(create_error.exception), "research_store_unavailable") + self.assertEqual(cloud.objects, {}) + cloud.fail_create = False + store.save_research_trial(trial) + self.assertEqual(local_path.read_text(), '{"trial_id":"local-only"}') + local_path.write_text('{"reason_code":"local-copy"}') + loaded = store.load_research_trial("us_equity", "global_etf_rotation", "trial-a") + self.assertEqual(json.loads(cloud.objects[uri])["reason_code"], "config_unparsed") + self.assertEqual(loaded, trial) + self.assertEqual(local_path.read_text(), '{"reason_code":"local-copy"}') + + def test_cloud_success_uses_only_the_bucket(self) -> None: + ledger = _ledger() + trial = _trial(ResearchTrialStatus.SUCCEEDED) + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + cloud = _ResearchCloud() + store = PerformanceStore(cloud_bucket="lifecycle-bucket", cloud_prefix="production", local_root=root) + with patch("quant_platform_kit.strategy_lifecycle.performance_store.get_object_store", return_value=cloud): + store.save_backtest_result(_result(ledger)) + store.save_research_ledger(ledger) + store.save_research_trial(trial) + loaded = store.load_research_trial("us_equity", "global_etf_rotation", "trial-a", run_id="run-a", param_version=1) + loaded_ledger = store.load_research_ledger("us_equity", "global_etf_rotation", "trial-a", "run-a", 1) + uris = list(cloud.objects) + research = [uri for uri in uris if "/research_trial/" in uri] + self.assertEqual(loaded, trial) + self.assertEqual(loaded_ledger.total_fees, 1.0) + self.assertTrue(research) + self.assertTrue(any("/backtest/" in uri for uri in uris)) + self.assertTrue(all(uri.startswith("gs://lifecycle-bucket/production/") for uri in uris)) + self.assertTrue(all("/research_trial/" in uri or "/backtest/" in uri for uri in uris)) + self.assertFalse(any("research_trial" in path.parts for path in root.rglob("*"))) + + if __name__ == "__main__": unittest.main()