diff --git a/src/quant_platform_kit/strategy_lifecycle/contracts.py b/src/quant_platform_kit/strategy_lifecycle/contracts.py index 9c58c6f8..5a023451 100644 --- a/src/quant_platform_kit/strategy_lifecycle/contracts.py +++ b/src/quant_platform_kit/strategy_lifecycle/contracts.py @@ -885,28 +885,70 @@ def _symbol(value: object) -> str: @dataclass(frozen=True) class ResearchPositionMark: - """One position's quantity and marked value. Zero quantity has zero value.""" + """One equity or option contract position and its signed marked value.""" symbol: str quantity: float valuation: float + option_underlying: str | None = None + option_right: str | None = None + option_strike: float | None = None + option_expiration: date | None = None + option_multiplier: float | None = None + option_premium_cashflow: float | None = None 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") + option_values = ( + self.option_underlying, self.option_right, self.option_strike, + self.option_expiration, self.option_multiplier, self.option_premium_cashflow, + ) + if all(value is None for value in option_values): + if (quantity == 0.0) != (valuation == 0.0): + raise ValueError("position_mark") + elif any(value is None for value in option_values): + raise ValueError("option_position") + else: + underlying = _symbol(self.option_underlying) + right = self.option_right + if right not in {"call", "put"} or quantity == 0.0 or ( + valuation != 0.0 and quantity * valuation < 0 + ): + raise ValueError("option_position") + strike = _finite_number(self.option_strike) + multiplier = _finite_number(self.option_multiplier) + premium_cashflow = _finite_number(self.option_premium_cashflow) + expiration = _require_date(self.option_expiration) + if strike <= 0 or multiplier <= 0 or quantity * premium_cashflow >= 0: + raise ValueError("option_position") + object.__setattr__(self, "option_underlying", underlying) + object.__setattr__(self, "option_right", right) + object.__setattr__(self, "option_strike", strike) + object.__setattr__(self, "option_expiration", expiration) + object.__setattr__(self, "option_multiplier", multiplier) + object.__setattr__(self, "option_premium_cashflow", premium_cashflow) object.__setattr__(self, "symbol", symbol) object.__setattr__(self, "quantity", quantity) object.__setattr__(self, "valuation", valuation) def to_dict(self) -> dict[str, object]: - return { + payload: dict[str, object] = { "symbol": self.symbol, "quantity": self.quantity, "valuation": self.valuation, } + if self.option_underlying is not None: + payload.update({ + "option_underlying": self.option_underlying, + "option_right": self.option_right, + "option_strike": self.option_strike, + "option_expiration": self.option_expiration.isoformat(), + "option_multiplier": self.option_multiplier, + "option_premium_cashflow": self.option_premium_cashflow, + }) + return payload def _position_marks(value: object) -> tuple[ResearchPositionMark, ...]: @@ -924,9 +966,341 @@ def _position_marks(value: object) -> tuple[ResearchPositionMark, ...]: return tuple(marks) +def _option_collateral_requirement(positions: tuple[ResearchPositionMark, ...]) -> float: + groups: dict[tuple[str, str, date, float], list[ResearchPositionMark]] = {} + for mark in positions: + if mark.option_underlying is not None: + key = (mark.option_underlying, mark.option_right, mark.option_expiration, mark.option_multiplier) + groups.setdefault(key, []).append(mark) + required = 0.0 + for (_, right, _, multiplier), marks in groups.items(): + shorts = sorted( + (mark for mark in marks if mark.quantity < 0), + key=lambda mark: mark.option_strike, + reverse=right == "put", + ) + longs = {mark.symbol: mark.quantity for mark in marks if mark.quantity > 0} + gross_risk = 0.0 + for short in shorts: + remaining = abs(short.quantity) + if right == "put": + protective = sorted( + (mark for mark in marks if mark.quantity > 0 and mark.option_strike < short.option_strike), + key=lambda mark: mark.option_strike, + reverse=True, + ) + else: + protective = sorted( + (mark for mark in marks if mark.quantity > 0 and mark.option_strike > short.option_strike), + key=lambda mark: mark.option_strike, + ) + for long in protective: + matched = min(remaining, longs[long.symbol]) + if matched <= 0: + continue + gross_risk += abs(short.option_strike - long.option_strike) * matched * multiplier + remaining -= matched + longs[long.symbol] -= matched + if remaining <= 1e-9: + break + if remaining > 1e-9: + raise ValueError("option_naked_short") + if shorts: + net_premium_credit = sum(mark.option_premium_cashflow for mark in marks) + required += max(0.0, gross_risk - net_premium_credit) + return required + + +def _equity_trade_quantities(value: object) -> dict[str, float] | None: + if value is None: + return None + if isinstance(value, (str, bytes)) or not isinstance(value, Mapping): + raise ValueError("ledger_equity_trade_quantity") + quantities: dict[str, float] = {} + for raw_symbol, raw_quantity in value.items(): + symbol = _symbol(raw_symbol) + quantity = _finite_number(raw_quantity) + if quantity == 0.0 or symbol in quantities: + raise ValueError("ledger_equity_trade_quantity") + quantities[symbol] = quantity + return dict(sorted(quantities.items())) + + +def _validate_option_ledger( + initial_positions: tuple[ResearchPositionMark, ...], + days: tuple[ResearchLedgerDay, ...], + initial_cash: float, + initial_restricted_cash: float, +) -> None: + previous = {mark.symbol: mark for mark in initial_positions} + previous_cash = initial_cash + previous_restricted = initial_restricted_cash + if initial_restricted_cash + 1e-9 < _option_collateral_requirement(initial_positions): + raise ValueError("ledger_option_collateral") + for day in days: + current = {mark.symbol: mark for mark in day.positions} + previous_options = {key: mark for key, mark in previous.items() if mark.option_underlying is not None} + current_options = {key: mark for key, mark in current.items() if mark.option_underlying is not None} + option_trade_events = [event for event in day.events if event.event_type == "option_trade"] + if len({event.symbol for event in option_trade_events}) != len(option_trade_events): + raise ValueError("ledger_option_trade") + option_changes = set(previous_options) ^ set(current_options) + expiring_symbols = { + symbol for symbol, mark in previous_options.items() if mark.option_expiration == day.session_date + } + trade_changes = option_changes - expiring_symbols + if {event.symbol for event in option_trade_events} != trade_changes: + raise ValueError("ledger_option_trade") + if trade_changes: + option_trade_cashflow = sum(event.amount for event in option_trade_events) + equity_quantity_changed = False + for symbol in set(previous) | set(current): + before, after = previous.get(symbol), current.get(symbol) + before_quantity = 0.0 if before is None or before.option_underlying is not None else before.quantity + after_quantity = 0.0 if after is None or after.option_underlying is not None else after.quantity + if before_quantity != after_quantity: + equity_quantity_changed = True + if equity_quantity_changed and day.equity_trade_cashflow is None: + raise ValueError("ledger_equity_trade_cashflow") + equity_trade_cashflow = day.equity_trade_cashflow or 0.0 + if not math.isclose( + day.trade_net_cashflow, option_trade_cashflow + equity_trade_cashflow, + rel_tol=0.0, abs_tol=1e-9, + ): + raise ValueError("ledger_option_trade") + for symbol in trade_changes: + event = next(event for event in option_trade_events if event.symbol == symbol) + before, after = previous_options.get(symbol), current_options.get(symbol) + if after is not None: + if after.option_expiration <= day.session_date or not math.isclose( + event.amount, after.option_premium_cashflow, rel_tol=0.0, abs_tol=1e-9 + ): + raise ValueError("ledger_option_trade") + elif before is not None and before.option_expiration <= day.session_date: + raise ValueError("ledger_option_settlement") + elif before is not None and event.amount * before.quantity <= 0: + raise ValueError("ledger_option_trade") + for symbol in set(previous_options) & set(current_options): + before, after = previous_options[symbol], current_options[symbol] + if ( + before.quantity != after.quantity + or before.option_underlying != after.option_underlying + or before.option_right != after.option_right + or before.option_strike != after.option_strike + or before.option_expiration != after.option_expiration + or before.option_multiplier != after.option_multiplier + or before.option_premium_cashflow != after.option_premium_cashflow + ): + raise ValueError("ledger_option_trade") + + settlements = [event for event in day.events if event.event_type == "option_settlement"] + settlement_by_underlying = {event.symbol: event for event in settlements} + if len(settlement_by_underlying) != len(settlements): + raise ValueError("ledger_option_settlement") + if not settlements and (day.equity_trade_quantities is not None or day.equity_trade_phase is not None): + raise ValueError("ledger_equity_trade_phase") + expiring: dict[str, list[ResearchPositionMark]] = {} + for mark in previous_options.values(): + if mark.option_expiration < day.session_date: + raise ValueError("ledger_option_expiration") + if mark.option_expiration == day.session_date: + if mark.symbol in current_options: + raise ValueError("ledger_option_settlement") + expiring.setdefault(mark.option_underlying, []).append(mark) + if set(settlement_by_underlying) != set(expiring): + raise ValueError("ledger_option_settlement") + settlement_cashflow = 0.0 + deliveries: dict[str, float] = {} + for underlying, legs in expiring.items(): + spot = settlement_by_underlying[underlying].settlement_price + for leg in legs: + intrinsic = ( + max(spot - leg.option_strike, 0.0) + if leg.option_right == "call" + else max(leg.option_strike - spot, 0.0) + ) + if intrinsic == 0.0: + continue + signed_contracts = leg.quantity * leg.option_multiplier + if leg.option_right == "call": + settlement_cashflow -= signed_contracts * leg.option_strike + deliveries[underlying] = deliveries.get(underlying, 0.0) + signed_contracts + else: + settlement_cashflow += signed_contracts * leg.option_strike + deliveries[underlying] = deliveries.get(underlying, 0.0) - signed_contracts + current_equity = next(( + mark for mark in current.values() + if mark.symbol == underlying and mark.option_underlying is None + ), None) + current_shares = 0.0 if current_equity is None else current_equity.quantity + if current_equity is not None and not math.isclose( + current_equity.valuation, current_shares * spot, rel_tol=0.0, abs_tol=1e-9 + ): + raise ValueError("ledger_option_delivery") + if settlements: + if day.equity_trade_quantities is not None and day.equity_trade_cashflow is None: + raise ValueError("ledger_equity_trade_quantity") + if day.equity_trade_cashflow is not None and day.equity_trade_quantities is None: + raise ValueError("ledger_equity_trade_quantity") + option_trade_cashflow = sum(event.amount for event in option_trade_events) + equity_trade_cashflow = day.equity_trade_cashflow or 0.0 + if not math.isclose( + day.trade_net_cashflow, option_trade_cashflow + equity_trade_cashflow, + rel_tol=0.0, abs_tol=1e-9, + ): + raise ValueError("ledger_option_settlement") + declared_quantities = day.equity_trade_quantities or {} + previous_equity = { + symbol: mark.quantity for symbol, mark in previous.items() if mark.option_underlying is None + } + current_equity = { + symbol: mark.quantity for symbol, mark in current.items() if mark.option_underlying is None + } + for symbol in set(previous_equity) | set(current_equity) | set(deliveries) | set(declared_quantities): + actual = current_equity.get(symbol, 0.0) - previous_equity.get(symbol, 0.0) + trade_quantity = declared_quantities.get(symbol, 0.0) + if ( + day.equity_trade_phase == "before_settlement" + and previous_equity.get(symbol, 0.0) + trade_quantity < -1e-9 + ): + raise ValueError("ledger_equity_trade_quantity") + expected = deliveries.get(symbol, 0.0) + trade_quantity + if not math.isclose(actual, expected, rel_tol=0.0, abs_tol=1e-9): + raise ValueError("ledger_equity_trade_quantity") + if not math.isclose(day.option_settlement_cashflow, settlement_cashflow, rel_tol=0.0, abs_tol=1e-9): + raise ValueError("ledger_option_settlement") + collateral_change = sum( + event.amount for event in day.events if event.event_type == "collateral_change" + ) + if not math.isclose( + day.restricted_cash, previous_restricted + collateral_change, rel_tol=0.0, abs_tol=1e-9 + ): + raise ValueError("ledger_collateral") + if day.restricted_cash + 1e-9 < _option_collateral_requirement(day.positions): + raise ValueError("ledger_option_collateral") + if settlements and day.restricted_cash > _option_collateral_requirement(day.positions) + 1e-9: + raise ValueError("ledger_collateral_release") + if settlement_cashflow < 0: + option_trade_cashflow = sum(event.amount for event in option_trade_events) + equity_cash_before_settlement = ( + (day.equity_trade_cashflow or 0.0) + if day.equity_trade_phase == "before_settlement" else 0.0 + ) + fees_before_settlement = day.fees if day.equity_trade_phase == "before_settlement" else 0.0 + cash_before_settlement = ( + previous_cash + + option_trade_cashflow + equity_cash_before_settlement - fees_before_settlement + + day.income_cashflow + day.external_cashflow + ) + collateral_release = max(0.0, previous_restricted - day.restricted_cash) + available = cash_before_settlement - previous_restricted + collateral_release + if -settlement_cashflow > available + 1e-9: + raise ValueError("ledger_option_funding") + previous = current + previous_cash = day.cash + previous_restricted = day.restricted_cash + + +@dataclass(frozen=True) +class ResearchLedgerEvent: + """One explicitly identified corporate-action, option, or collateral event.""" + + event_id: str + event_type: str + symbol: str + amount: float | None = None + per_share: float | None = None + ratio: float | None = None + reference_event_id: str | None = None + settlement_price: float | None = None + + def __post_init__(self) -> None: + event_id = _raw_identity(self.event_id) + event_type = _label(self.event_type) + symbol = _label(self.symbol) + if event_type == "dividend_accrual": + if ( + self.amount is not None or self.ratio is not None or self.reference_event_id is not None + or self.settlement_price is not None + ): + raise ValueError("ledger_event_fields") + per_share = _finite_number(self.per_share) + if per_share <= 0: + raise ValueError("ledger_event_fields") + elif event_type == "dividend_payment": + if ( + self.per_share is not None or self.ratio is not None or self.settlement_price is not None + or not self.reference_event_id + ): + raise ValueError("ledger_event_fields") + amount = _finite_number(self.amount) + if amount <= 0: + raise ValueError("ledger_event_fields") + object.__setattr__(self, "amount", amount) + object.__setattr__(self, "reference_event_id", _raw_identity(self.reference_event_id)) + elif event_type == "split": + if ( + self.amount is not None or self.per_share is not None or self.reference_event_id is not None + or self.settlement_price is not None + ): + raise ValueError("ledger_event_fields") + ratio = _finite_number(self.ratio) + if ratio <= 0: + raise ValueError("ledger_event_fields") + object.__setattr__(self, "ratio", ratio) + elif event_type == "option_trade": + if any(value is not None for value in (self.per_share, self.ratio, self.reference_event_id, self.settlement_price)): + raise ValueError("ledger_event_fields") + amount = _finite_number(self.amount) + if amount == 0.0: + raise ValueError("ledger_event_fields") + symbol = _symbol(symbol) + object.__setattr__(self, "amount", amount) + elif event_type == "option_settlement": + if any(value is not None for value in (self.amount, self.per_share, self.ratio, self.reference_event_id)): + raise ValueError("ledger_event_fields") + settlement_price = _finite_number(self.settlement_price) + if settlement_price <= 0: + raise ValueError("ledger_event_fields") + symbol = _symbol(symbol) + object.__setattr__(self, "settlement_price", settlement_price) + elif event_type == "collateral_change": + if any(value is not None for value in (self.per_share, self.ratio, self.reference_event_id, self.settlement_price)): + raise ValueError("ledger_event_fields") + amount = _finite_number(self.amount) + if amount == 0.0 or symbol != "CASH": + raise ValueError("ledger_event_fields") + object.__setattr__(self, "amount", amount) + else: + raise ValueError("ledger_event_type") + object.__setattr__(self, "event_id", event_id) + object.__setattr__(self, "event_type", event_type) + object.__setattr__(self, "symbol", symbol) + if event_type == "dividend_accrual": + object.__setattr__(self, "per_share", per_share) + + def to_dict(self) -> dict[str, object]: + payload: dict[str, object] = { + "event_id": self.event_id, + "event_type": self.event_type, + "symbol": self.symbol, + } + if self.amount is not None: + payload["amount"] = self.amount + if self.per_share is not None: + payload["per_share"] = self.per_share + if self.ratio is not None: + payload["ratio"] = self.ratio + if self.reference_event_id is not None: + payload["reference_event_id"] = self.reference_event_id + if self.settlement_price is not None: + payload["settlement_price"] = self.settlement_price + return payload + + @dataclass(frozen=True) class ResearchLedgerDay: - """End-of-session cash, marks, flows, fees, NAV, and that session's return.""" + """End-of-session book with events ordered split, accrual, payment, trade, valuation.""" session_date: date cash: float @@ -935,32 +1309,84 @@ class ResearchLedgerDay: fees: float nav: float daily_return: float + income_cashflow: float = 0.0 + external_cashflow: float = 0.0 + dividend_receivable: float = 0.0 + declared_event_ids: tuple[str, ...] | None = None + events: tuple[ResearchLedgerEvent, ...] = () + restricted_cash: float = 0.0 + option_settlement_cashflow: float = 0.0 + equity_trade_cashflow: float | None = None + equity_trade_quantities: Mapping[str, float] | None = None + equity_trade_phase: str | None = None 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) + income_cashflow = _finite_number(self.income_cashflow) + external_cashflow = _finite_number(self.external_cashflow) + dividend_receivable = _finite_number(self.dividend_receivable) + restricted_cash = _finite_number(self.restricted_cash) + option_settlement_cashflow = _finite_number(self.option_settlement_cashflow) + equity_trade_cashflow = ( + None if self.equity_trade_cashflow is None else _finite_number(self.equity_trade_cashflow) + ) + equity_trade_quantities = _equity_trade_quantities(self.equity_trade_quantities) + equity_trade_phase = self.equity_trade_phase + if equity_trade_phase is not None and ( + not isinstance(equity_trade_phase, str) + or equity_trade_phase not in {"before_settlement", "after_settlement"} + ): + raise ValueError("ledger_equity_trade_phase") + if (equity_trade_quantities is None) != (equity_trade_phase is None): + raise ValueError("ledger_equity_trade_phase") fees = _finite_number(self.fees) if fees < 0: raise ValueError("ledger_fee") nav = _finite_number(self.nav) - if nav <= 0: + if nav <= 0 or dividend_receivable < 0 or restricted_cash < 0 or restricted_cash > max(cash, 0.0) + 1e-9: raise ValueError("ledger_nav") daily_return = _finite_number(self.daily_return) - expected_nav = cash + sum(mark.valuation for mark in positions) + expected_nav = cash + sum(mark.valuation for mark in positions) + dividend_receivable if not math.isclose(nav, expected_nav, rel_tol=0.0, abs_tol=1e-9): raise ValueError("ledger_nav") + if self.declared_event_ids is None: + declared_event_ids = None + else: + if isinstance(self.declared_event_ids, (str, bytes)) or not isinstance(self.declared_event_ids, (tuple, list)): + raise ValueError("ledger_event_set") + declared_event_ids = tuple(_raw_identity(item) for item in self.declared_event_ids) + if len(set(declared_event_ids)) != len(declared_event_ids): + raise ValueError("ledger_event_set") + if isinstance(self.events, (str, bytes)) or not isinstance(self.events, (tuple, list)): + raise ValueError("ledger_event") + events = tuple(self.events) + if any(not isinstance(event, ResearchLedgerEvent) for event in events): + raise ValueError("ledger_event") + if len({event.event_id for event in events}) != len(events): + raise ValueError("ledger_event_duplicate") 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, "income_cashflow", income_cashflow) + object.__setattr__(self, "external_cashflow", external_cashflow) + object.__setattr__(self, "dividend_receivable", dividend_receivable) + object.__setattr__(self, "declared_event_ids", declared_event_ids) + object.__setattr__(self, "events", events) + object.__setattr__(self, "restricted_cash", restricted_cash) + object.__setattr__(self, "option_settlement_cashflow", option_settlement_cashflow) + object.__setattr__(self, "equity_trade_cashflow", equity_trade_cashflow) + object.__setattr__(self, "equity_trade_quantities", equity_trade_quantities) + object.__setattr__(self, "equity_trade_phase", equity_trade_phase) 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 { + payload: dict[str, object] = { "session_date": self.session_date.isoformat(), "cash": self.cash, "positions": [mark.to_dict() for mark in self.positions], @@ -969,6 +1395,26 @@ def to_dict(self) -> dict[str, object]: "nav": self.nav, "daily_return": self.daily_return, } + if self.income_cashflow != 0.0: + payload["income_cashflow"] = self.income_cashflow + if self.external_cashflow != 0.0: + payload["external_cashflow"] = self.external_cashflow + if self.dividend_receivable != 0.0: + payload["dividend_receivable"] = self.dividend_receivable + if self.declared_event_ids is not None: + payload["declared_event_ids"] = list(self.declared_event_ids) + if self.events: + payload["events"] = [event.to_dict() for event in self.events] + if self.restricted_cash != 0.0: + payload["restricted_cash"] = self.restricted_cash + if self.option_settlement_cashflow != 0.0: + payload["option_settlement_cashflow"] = self.option_settlement_cashflow + if self.equity_trade_cashflow is not None: + payload["equity_trade_cashflow"] = self.equity_trade_cashflow + if self.equity_trade_quantities is not None: + payload["equity_trade_quantities"] = dict(self.equity_trade_quantities) + payload["equity_trade_phase"] = self.equity_trade_phase + return payload @dataclass(frozen=True) @@ -997,6 +1443,7 @@ class ResearchDailyLedger: initial_positions: tuple[ResearchPositionMark, ...] days: tuple[ResearchLedgerDay, ...] synthetic: bool + initial_restricted_cash: float = 0.0 def __post_init__(self) -> None: trial_id = _raw_identity(self.trial_id) @@ -1017,12 +1464,15 @@ def __post_init__(self) -> None: initial_session_date = _require_date(self.initial_session_date) initial_nav = _finite_number(self.initial_nav) initial_cash = _finite_number(self.initial_cash) + initial_restricted_cash = _finite_number(self.initial_restricted_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 initial_restricted_cash < 0 or initial_restricted_cash > max(initial_cash, 0.0) + 1e-9: + raise ValueError("ledger_collateral") 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) @@ -1037,15 +1487,93 @@ def __post_init__(self) -> None: raise ValueError("synthetic") previous_cash = initial_cash previous_nav = initial_nav + previous_receivable = 0.0 + previous_quantities = {mark.symbol: mark.quantity for mark in initial_positions} + seen_event_ids: set[str] = set() + open_dividends: dict[str, tuple[str, float]] = {} + complete_event_mode = any( + day.declared_event_ids is not None or day.events or day.dividend_receivable != 0.0 for day in days + ) + if complete_event_mode and any(day.declared_event_ids is None for day in days): + raise ValueError("ledger_event_set") for day in days: - expected_cash = previous_cash + day.trade_net_cashflow - day.fees + expected_cash = ( + previous_cash + day.trade_net_cashflow - day.fees + day.income_cashflow + day.external_cashflow + + day.option_settlement_cashflow + ) 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 + return_basis = previous_nav + day.external_cashflow + if return_basis <= 0: + raise ValueError("ledger_return") + expected_return = day.nav / return_basis - 1.0 if day.daily_return != expected_return: raise ValueError("ledger_return") + event_ids = {event.event_id for event in day.events} + if seen_event_ids.intersection(event_ids): + raise ValueError("ledger_event_duplicate") + seen_event_ids.update(event_ids) + if day.declared_event_ids is not None and set(day.declared_event_ids) != event_ids: + raise ValueError("ledger_event_set") + phases = { + "split": 0, "dividend_accrual": 1, "dividend_payment": 2, + "option_trade": 3, "option_settlement": 4, "collateral_change": 5, + } + event_phases = [phases[event.event_type] for event in day.events] + if event_phases != sorted(event_phases): + raise ValueError("ledger_event_order") + if any(event.event_type == "split" for event in day.events) and ( + day.trade_net_cashflow != 0.0 or day.fees != 0.0 + ): + raise ValueError("ledger_split_trade") + quantities = dict(previous_quantities) + split_symbols: set[str] = set() + for event in day.events: + if event.event_type == "split": + old_quantity = quantities.get(event.symbol, 0.0) + if old_quantity <= 0: + raise ValueError("ledger_split_quantity") + if event.symbol in split_symbols: + raise ValueError("ledger_split_quantity") + split_symbols.add(event.symbol) + quantities[event.symbol] = old_quantity * event.ratio + if split_symbols: + current_quantities = {mark.symbol: mark.quantity for mark in day.positions} + for symbol in set(previous_quantities) | set(current_quantities): + expected_quantity = quantities.get(symbol, 0.0) + if not math.isclose( + current_quantities.get(symbol, 0.0), expected_quantity, rel_tol=0.0, abs_tol=1e-9 + ): + raise ValueError("ledger_split_quantity") + expected_receivable = previous_receivable + payment_total = 0.0 + for event in day.events: + if event.event_type == "dividend_accrual": + quantity = quantities.get(event.symbol, 0.0) + if quantity <= 0: + raise ValueError("ledger_dividend_quantity") + amount = quantity * event.per_share + expected_receivable += amount + open_dividends[event.event_id] = (event.symbol, amount) + elif event.event_type == "dividend_payment": + accrued = open_dividends.get(event.reference_event_id) + if accrued is None or accrued[0] != event.symbol or not math.isclose( + accrued[1], event.amount, rel_tol=0.0, abs_tol=1e-9 + ): + raise ValueError("ledger_dividend_pair") + del open_dividends[event.reference_event_id] + expected_receivable -= event.amount + payment_total += event.amount + if complete_event_mode and not math.isclose( + payment_total, day.income_cashflow, rel_tol=0.0, abs_tol=1e-9 + ): + raise ValueError("ledger_dividend_cashflow") + if not math.isclose(day.dividend_receivable, expected_receivable, rel_tol=0.0, abs_tol=1e-9): + raise ValueError("ledger_dividend_receivable") previous_cash = day.cash previous_nav = day.nav + previous_receivable = day.dividend_receivable + previous_quantities = {mark.symbol: mark.quantity for mark in day.positions} object.__setattr__(self, "trial_id", trial_id) object.__setattr__(self, "domain", domain) object.__setattr__(self, "strategy_profile", strategy_profile) @@ -1060,6 +1588,8 @@ def __post_init__(self) -> None: object.__setattr__(self, "initial_cash", initial_cash) object.__setattr__(self, "initial_positions", initial_positions) object.__setattr__(self, "days", days) + object.__setattr__(self, "initial_restricted_cash", initial_restricted_cash) + _validate_option_ledger(initial_positions, days, initial_cash, initial_restricted_cash) @property def window_start(self) -> date: @@ -1073,16 +1603,23 @@ def window_end(self) -> date: def observation_count(self) -> int: return len(self.days) + @property + def events_complete(self) -> bool: + """Whether every return day declares its complete event-ID set.""" + return all(day.declared_event_ids is not None for day in self.days) + @property def total_return(self) -> float: - return self.days[-1].nav / self.initial_nav - 1.0 + if all(day.external_cashflow == 0.0 for day in self.days): + return self.days[-1].nav / self.initial_nav - 1.0 + return math.prod(1.0 + day.daily_return for day in self.days) - 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 { + payload: dict[str, object] = { "trial_id": self.trial_id, "domain": self.domain, "strategy_profile": self.strategy_profile, @@ -1100,6 +1637,9 @@ def to_dict(self) -> dict[str, object]: "days": [day.to_dict() for day in self.days], "synthetic": self.synthetic, } + if self.initial_restricted_cash != 0.0: + payload["initial_restricted_cash"] = self.initial_restricted_cash + return payload def _actual_params(value: object, *, required: bool) -> dict[str, Any] | None: @@ -1118,6 +1658,20 @@ def _actual_params(value: object, *, required: bool) -> dict[str, Any] | None: return parsed +def _research_identity(value: object) -> dict[str, Any] | None: + if value is None: + return None + if not isinstance(value, Mapping) or not value or any(type(key) is not str or not key for key in value): + raise ValueError("research_identity") + try: + parsed = json.loads(json.dumps(dict(value), sort_keys=True, allow_nan=False)) + except (TypeError, ValueError) as exc: + raise ValueError("research_identity") from exc + if type(parsed) is not dict or any(type(key) is not str or not key for key in parsed): + raise ValueError("research_identity") + return parsed + + def _optional_identity(value: object, *, required: bool) -> str | None: if value is None: if required: @@ -1153,6 +1707,7 @@ class ResearchTrialRecord: synthetic: bool run_id: str | None param_version: int | None + research_identity: Mapping[str, Any] | None = None def __post_init__(self) -> None: status = self.status @@ -1169,6 +1724,7 @@ def __post_init__(self) -> None: strategy_profile = _label(self.strategy_profile) candidate_config_id = _raw_identity(self.candidate_config_id) actual_params = _actual_params(self.actual_params, required=succeeded) + research_identity = _research_identity(self.research_identity) 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) @@ -1214,6 +1770,7 @@ def __post_init__(self) -> None: 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, "research_identity", research_identity) object.__setattr__(self, "param_set_id", param_set_id) object.__setattr__(self, "source_revision", source_revision) object.__setattr__(self, "input_id", input_id) @@ -1228,7 +1785,7 @@ def __post_init__(self) -> None: object.__setattr__(self, "param_version", param_version) def to_dict(self) -> dict[str, object]: - return { + payload: dict[str, object] = { "trial_id": self.trial_id, "domain": self.domain, "strategy_profile": self.strategy_profile, @@ -1249,6 +1806,9 @@ def to_dict(self) -> dict[str, object]: "run_id": self.run_id, "param_version": self.param_version, } + if self.research_identity is not None: + payload["research_identity"] = dict(self.research_identity) + return payload # ── 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 526c7c53..d86d411b 100644 --- a/src/quant_platform_kit/strategy_lifecycle/performance_store.py +++ b/src/quant_platform_kit/strategy_lifecycle/performance_store.py @@ -38,6 +38,7 @@ DriftResult, OptimizationProposal, ResearchDailyLedger, + ResearchLedgerEvent, ResearchLedgerDay, ResearchPositionMark, ResearchTrialRecord, @@ -1117,18 +1118,29 @@ def _research_date(value: object) -> date: _POSITION_FIELDS = frozenset({"symbol", "quantity", "valuation"}) +_OPTION_POSITION_FIELDS = _POSITION_FIELDS | { + "option_underlying", "option_right", "option_strike", "option_expiration", "option_multiplier", + "option_premium_cashflow", +} _DAY_FIELDS = frozenset({"session_date", "cash", "positions", "trade_net_cashflow", "fees", "nav", "daily_return"}) +_DAY_OPTIONAL_FIELDS = _DAY_FIELDS | { + "income_cashflow", "external_cashflow", "dividend_receivable", "declared_event_ids", "events", + "restricted_cash", "option_settlement_cashflow", "equity_trade_cashflow", + "equity_trade_quantities", "equity_trade_phase", +} _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", }) +_LEDGER_COLLATERAL_FIELDS = _LEDGER_FIELDS | {"initial_restricted_cash"} _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", }) +_TRIAL_IDENTITY_FIELDS = _TRIAL_FIELDS | {"research_identity"} def _research_positions(value: object) -> tuple[ResearchPositionMark, ...]: @@ -1136,9 +1148,12 @@ def _research_positions(value: object) -> tuple[ResearchPositionMark, ...]: raise ValueError("position_mark") marks: list[ResearchPositionMark] = [] for item in value: - if not isinstance(item, dict) or set(item) != _POSITION_FIELDS: + if not isinstance(item, dict) or frozenset(item) not in {_POSITION_FIELDS, _OPTION_POSITION_FIELDS}: raise ValueError("position_mark") - marks.append(ResearchPositionMark(symbol=item["symbol"], quantity=item["quantity"], valuation=item["valuation"])) + parsed = dict(item) + if "option_expiration" in parsed: + parsed["option_expiration"] = _research_date(parsed["option_expiration"]) + marks.append(ResearchPositionMark(**parsed)) return tuple(marks) @@ -1147,8 +1162,28 @@ def _research_days(value: object) -> tuple[ResearchLedgerDay, ...]: raise ValueError("ledger_dates") days: list[ResearchLedgerDay] = [] for item in value: - if not isinstance(item, dict) or set(item) != _DAY_FIELDS: + if ( + not isinstance(item, dict) + or not _DAY_FIELDS.issubset(item) + or not frozenset(item).issubset(_DAY_OPTIONAL_FIELDS) + ): raise ValueError("ledger_dates") + parsed_events = [] + for event in item.get("events", []): + if not isinstance(event, dict): + raise ValueError("ledger_event") + event_type = event.get("event_type") + event_fields = { + "split": {"event_id", "event_type", "symbol", "ratio"}, + "dividend_accrual": {"event_id", "event_type", "symbol", "per_share"}, + "dividend_payment": {"event_id", "event_type", "symbol", "amount", "reference_event_id"}, + "option_trade": {"event_id", "event_type", "symbol", "amount"}, + "option_settlement": {"event_id", "event_type", "symbol", "settlement_price"}, + "collateral_change": {"event_id", "event_type", "symbol", "amount"}, + }.get(event_type) + if event_fields is None or frozenset(event) != event_fields: + raise ValueError("ledger_event_fields") + parsed_events.append(ResearchLedgerEvent(**event)) days.append(ResearchLedgerDay( session_date=_research_date(item["session_date"]), cash=item["cash"], @@ -1157,12 +1192,26 @@ def _research_days(value: object) -> tuple[ResearchLedgerDay, ...]: fees=item["fees"], nav=item["nav"], daily_return=item["daily_return"], + income_cashflow=item.get("income_cashflow", 0.0), + external_cashflow=item.get("external_cashflow", 0.0), + dividend_receivable=item.get("dividend_receivable", 0.0), + declared_event_ids=item.get("declared_event_ids"), + events=tuple(parsed_events), + restricted_cash=item.get("restricted_cash", 0.0), + option_settlement_cashflow=item.get("option_settlement_cashflow", 0.0), + equity_trade_cashflow=item.get("equity_trade_cashflow"), + equity_trade_quantities=item.get("equity_trade_quantities"), + equity_trade_phase=item.get("equity_trade_phase"), )) 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: + if ( + not isinstance(data, dict) + or frozenset(data) not in {_LEDGER_FIELDS, _LEDGER_COLLATERAL_FIELDS} + or data.get("schema_version") != SCHEMA_VERSION + ): return None try: return ResearchDailyLedger( @@ -1182,13 +1231,18 @@ def _research_ledger_from_dict(data: Mapping[str, Any] | None) -> ResearchDailyL initial_positions=_research_positions(data["initial_positions"]), days=_research_days(data["days"]), synthetic=data["synthetic"], + initial_restricted_cash=data.get("initial_restricted_cash", 0.0), ) 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: + if ( + not isinstance(data, dict) + or frozenset(data) not in {_TRIAL_FIELDS, _TRIAL_IDENTITY_FIELDS} + or data.get("schema_version") != SCHEMA_VERSION + ): return None try: return ResearchTrialRecord( @@ -1211,6 +1265,7 @@ def _research_trial_from_dict(data: Mapping[str, Any] | None) -> ResearchTrialRe synthetic=data["synthetic"], run_id=data["run_id"], param_version=data["param_version"], + research_identity=data.get("research_identity"), ) except Exception: return None @@ -1230,9 +1285,12 @@ def _research_trial_continues(started: ResearchTrialRecord, terminal: ResearchTr or started.calendar_id != terminal.calendar_id or started.periods_per_year != terminal.periods_per_year or started.synthetic is not terminal.synthetic + or not _research_json_mapping_equal(started.research_identity, terminal.research_identity) ): return False - if started.actual_params is not None and started.actual_params != terminal.actual_params: + if started.actual_params is not None and not _research_json_mapping_equal( + 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 @@ -1245,11 +1303,30 @@ def _research_trial_continues(started: ResearchTrialRecord, terminal: ResearchTr return True +def _research_json_mapping_equal(left: Mapping[str, Any] | None, right: Mapping[str, Any] | None) -> bool: + try: + return json.dumps(left, sort_keys=True, separators=(",", ":"), allow_nan=False) == json.dumps( + right, sort_keys=True, separators=(",", ":"), allow_nan=False + ) + except (TypeError, ValueError): + return False + + 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 + result_params = dict(result.params) + if trial.research_identity is None: + if "research_identity" in result_params: + return False + else: + result_identity = result_params.pop("research_identity", None) + if not isinstance(result_identity, Mapping) or not _research_json_mapping_equal( + result_identity, trial.research_identity + ): + return False return ( - dict(result.params) == dict(trial.actual_params) + _research_json_mapping_equal(result_params, 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 diff --git a/tests/test_lifecycle_performance_store.py b/tests/test_lifecycle_performance_store.py index 3405b24f..bb8d5537 100644 --- a/tests/test_lifecycle_performance_store.py +++ b/tests/test_lifecycle_performance_store.py @@ -15,6 +15,7 @@ BacktestResult, BacktestValidationIdentity, ResearchDailyLedger, + ResearchLedgerEvent, ResearchLedgerDay, ResearchPositionMark, ResearchTrialRecord, @@ -23,6 +24,7 @@ from quant_platform_kit.strategy_lifecycle.performance_store import ( DEFAULT_LOCAL_ROOT, PerformanceStore, + SCHEMA_VERSION, ) @@ -608,9 +610,74 @@ def _research_file(root: Path, domain: str, profile: str, trial_id: str, name: s 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 _mark_day( + session: date, + cash: float, + quantity: float, + valuation: float, + flow: float, + fees: float, + nav: float, + previous: float, + income_cashflow: float = 0.0, + external_cashflow: float = 0.0, + dividend_receivable: float = 0.0, + declared_event_ids: tuple[str, ...] | None = None, + events: tuple[ResearchLedgerEvent, ...] = (), + restricted_cash: float = 0.0, + option_settlement_cashflow: float = 0.0, + position_marks: tuple[ResearchPositionMark, ...] | None = None, + equity_trade_cashflow: float | None = None, + equity_trade_quantities: dict[str, float] | None = None, + equity_trade_phase: str | None = None, +) -> ResearchLedgerDay: + positions = position_marks if position_marks is not None else ( + () if quantity == 0 else (ResearchPositionMark("SOXL", quantity, valuation),) + ) + return ResearchLedgerDay( + session, + cash, + positions, + flow, + fees, + nav, + nav / (previous + external_cashflow) - 1.0, + income_cashflow, + external_cashflow, + dividend_receivable, + declared_event_ids, + events, + restricted_cash, + option_settlement_cashflow, + equity_trade_cashflow, + equity_trade_quantities, + equity_trade_phase, + ) + + +def _option_mark( + symbol: str, + quantity: float, + valuation: float, + *, + right: str, + strike: float, + expiration: date, + premium_cashflow: float, + underlying: str = "SOXL", + multiplier: float = 100.0, +) -> ResearchPositionMark: + return ResearchPositionMark( + symbol=symbol, + quantity=quantity, + valuation=valuation, + option_underlying=underlying, + option_right=right, + option_strike=strike, + option_expiration=expiration, + option_multiplier=multiplier, + option_premium_cashflow=premium_cashflow, + ) 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: @@ -729,8 +796,802 @@ def list(self, prefix: str) -> list[str]: class ResearchTrialLedgerStoreTest(unittest.TestCase): + def test_external_cashflow_is_excluded_from_daily_and_total_return_and_matches_result(self) -> None: + day = _mark_day( + _END, + cash=200.0, + quantity=0.0, + valuation=0.0, + flow=0.0, + fees=0.0, + nav=200.0, + previous=100.0, + external_cashflow=100.0, + ) + ledger = 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,), synthetic=True, + ) + self.assertEqual(ledger.days[0].daily_return, 0.0) + self.assertEqual(ledger.total_return, 0.0) + self.assertEqual(ledger.to_dict()["days"][0]["external_cashflow"], 100.0) + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + store = PerformanceStore(local_root=root) + result = _result(ledger) + store.save_backtest_result(result) + store.save_research_ledger(ledger) + store.save_research_trial(_trial(ResearchTrialStatus.SUCCEEDED)) + loaded = store.load_research_trial("us_equity", "global_etf_rotation", "trial-a") + self.assertEqual(result.total_return, 0.0) + self.assertEqual(loaded.status, ResearchTrialStatus.SUCCEEDED) + + def test_legacy_non_integer_nav_total_return_uses_original_formula(self) -> None: + first_return = 1.2 / 1.1 - 1.0 + second_return = 1.3 / 1.2 - 1.0 + self.assertNotEqual((1.0 + first_return) * (1.0 + second_return) - 1.0, 1.3 / 1.1 - 1.0) + ledger = 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=1.1, initial_cash=1.1, initial_positions=(), + days=( + _mark_day(_START, 1.2, 0.0, 0.0, 0.1, 0.0, 1.2, 1.1), + _mark_day(_END, 1.3, 0.0, 0.0, 0.1, 0.0, 1.3, 1.2), + ), synthetic=True, + ) + self.assertEqual(ledger.total_return, ledger.days[-1].nav / ledger.initial_nav - 1.0) + result = _result(ledger) + self.assertEqual(result.total_return, ledger.total_return) + + def test_explicit_empty_event_declaration_is_complete_and_round_trips(self) -> None: + ledger = 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=(_mark_day( + _START, 100.0, 0.0, 0.0, 0.0, 0.0, 100.0, 100.0, declared_event_ids=(), + ),), synthetic=True, + ) + self.assertTrue(ledger.events_complete) + self.assertEqual(ledger.days[0].to_dict()["declared_event_ids"], []) + with tempfile.TemporaryDirectory() as tmp: + store = PerformanceStore(local_root=Path(tmp)) + store.save_research_ledger(ledger) + loaded = store.load_research_ledger("us_equity", "global_etf_rotation", "trial-a", "run-a", 1) + self.assertEqual(loaded, ledger) + + def test_long_leaps_purchase_keeps_nav_flat_and_round_trips(self) -> None: + leaps = _option_mark( + "SOXL-2026-01-16-90C", 1.0, 9000.0, right="call", strike=90.0, + expiration=date(2026, 1, 16), premium_cashflow=-9000.0, + ) + day = ResearchLedgerDay( + _START, 1000.0, (leaps,), -9000.0, 0.0, 10000.0, 0.0, + declared_event_ids=("leaps-buy",), + events=(ResearchLedgerEvent("leaps-buy", "option_trade", leaps.symbol, amount=-9000.0),), + ) + ledger = 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=10000.0, initial_cash=10000.0, + initial_positions=(), days=(day,), synthetic=True, + ) + self.assertEqual(ledger.days[0].nav, ledger.initial_nav) + with tempfile.TemporaryDirectory() as tmp: + store = PerformanceStore(local_root=Path(tmp)) + store.save_research_ledger(ledger) + loaded = store.load_research_ledger("us_equity", "global_etf_rotation", "trial-a", "run-a", 1) + self.assertEqual(loaded, ledger) + self.assertEqual(loaded.days[0].positions[0].symbol, "SOXL-2026-01-16-90C") + + def test_zero_mark_option_can_expire_worthless_without_share_delivery(self) -> None: + call = _option_mark( + "SOXL-2024-01-03-90C", 1.0, 0.0, right="call", strike=90.0, + expiration=_END, premium_cashflow=-100.0, + ) + expired = _mark_day( + _END, 1000.0, 0.0, 0.0, 0.0, 0.0, 1000.0, 1000.0, + declared_event_ids=("call-expiry",), + events=(ResearchLedgerEvent("call-expiry", "option_settlement", "SOXL", settlement_price=75.0),), + ) + ledger = 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=1000.0, initial_cash=1000.0, + initial_positions=(call,), days=(expired,), synthetic=True, + ) + self.assertEqual(ledger.days[0].nav, ledger.initial_nav) + self.assertEqual(ledger.days[0].option_settlement_cashflow, 0.0) + + def test_option_close_cashflow_sign_matches_open_position_side(self) -> None: + short = _option_mark( + "SOXL-2024-01-17-90P", -1.0, -900.0, right="put", strike=90.0, + expiration=date(2024, 1, 17), premium_cashflow=900.0, + ) + long = _option_mark( + "SOXL-2024-01-17-80P", 1.0, 700.0, right="put", strike=80.0, + expiration=date(2024, 1, 17), premium_cashflow=-700.0, + ) + for short_close, long_close in ((100.0, 100.0), (-100.0, -100.0)): + with self.subTest(short_close=short_close, long_close=long_close): + flow = short_close + long_close + day = _mark_day( + _START, 1000.0 + flow, 0.0, 0.0, flow, 0.0, 1000.0 + flow, 800.0, + declared_event_ids=("short-close", "long-close", "collateral-release"), + events=( + ResearchLedgerEvent("short-close", "option_trade", short.symbol, amount=short_close), + ResearchLedgerEvent("long-close", "option_trade", long.symbol, amount=long_close), + ResearchLedgerEvent("collateral-release", "collateral_change", "CASH", amount=-800.0), + ), + restricted_cash=0.0, + ) + with self.assertRaisesRegex(ValueError, "ledger_option_trade"): + 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=800.0, initial_cash=1000.0, + initial_positions=(short, long), days=(day,), synthetic=True, + initial_restricted_cash=800.0, + ) + + def test_put_credit_spread_settles_physical_legs_and_releases_collateral(self) -> None: + expiration = _END + short = _option_mark( + "SOXL-2024-01-03-90P", -1.0, -900.0, right="put", strike=90.0, + expiration=expiration, premium_cashflow=900.0, + ) + long = _option_mark( + "SOXL-2024-01-03-80P", 1.0, 700.0, right="put", strike=80.0, + expiration=expiration, premium_cashflow=-700.0, + ) + opened = _mark_day( + _START, 1200.0, 0.0, 0.0, 200.0, 0.0, 1000.0, 1000.0, + declared_event_ids=("short-open", "long-open", "collateral-lock"), + events=( + ResearchLedgerEvent("short-open", "option_trade", short.symbol, amount=900.0), + ResearchLedgerEvent("long-open", "option_trade", long.symbol, amount=-700.0), + ResearchLedgerEvent("collateral-lock", "collateral_change", "CASH", amount=800.0), + ), + restricted_cash=800.0, + position_marks=(short, long), + ) + settled = _mark_day( + _END, 200.0, 0.0, 0.0, 0.0, 0.0, 200.0, 1000.0, + declared_event_ids=("expiry", "collateral-release"), + events=( + ResearchLedgerEvent("expiry", "option_settlement", "SOXL", settlement_price=75.0), + ResearchLedgerEvent("collateral-release", "collateral_change", "CASH", amount=-800.0), + ), + restricted_cash=0.0, + option_settlement_cashflow=-1000.0, + ) + ledger = 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=1000.0, initial_cash=1000.0, + initial_positions=(), days=(opened, settled), synthetic=True, + ) + self.assertEqual(opened.nav, 1000.0) + self.assertEqual(opened.restricted_cash, 800.0) + self.assertEqual(settled.cash, 200.0) + self.assertEqual(settled.nav, 200.0) + self.assertEqual(settled.daily_return, -0.8) + self.assertEqual(ledger.total_return, -0.8) + with tempfile.TemporaryDirectory() as tmp: + store = PerformanceStore(local_root=Path(tmp)) + store.save_research_ledger(ledger) + loaded = store.load_research_ledger("us_equity", "global_etf_rotation", "trial-a", "run-a", 1) + self.assertEqual(loaded, ledger) + + repeated_settlement = _mark_day( + date(2024, 1, 4), 200.0, 0.0, 0.0, 0.0, 0.0, 200.0, 200.0, + declared_event_ids=("expiry-repeat",), + events=(ResearchLedgerEvent("expiry-repeat", "option_settlement", "SOXL", settlement_price=75.0),), + ) + with self.assertRaisesRegex(ValueError, "ledger_option_settlement"): + 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=1000.0, initial_cash=1000.0, + initial_positions=(), days=(opened, settled, repeated_settlement), synthetic=True, + ) + + def test_equity_rebalance_and_option_trade_cashflows_are_explicitly_decomposed(self) -> None: + short = _option_mark( + "SOXL-2024-01-17-90P", -1.0, -900.0, right="put", strike=90.0, + expiration=_END, premium_cashflow=900.0, + ) + long = _option_mark( + "SOXL-2024-01-17-80P", 1.0, 700.0, right="put", strike=80.0, + expiration=_END, premium_cashflow=-700.0, + ) + equity = ResearchPositionMark("SOXL", 2.0, 200.0) + opened = _mark_day( + _START, 1100.0, 2.0, 200.0, 100.0, 0.0, 1100.0, 1100.0, + declared_event_ids=("short-open", "long-open", "collateral-lock"), + events=( + ResearchLedgerEvent("short-open", "option_trade", short.symbol, amount=900.0), + ResearchLedgerEvent("long-open", "option_trade", long.symbol, amount=-700.0), + ResearchLedgerEvent("collateral-lock", "collateral_change", "CASH", amount=800.0), + ), + restricted_cash=800.0, + position_marks=(equity, short, long), + equity_trade_cashflow=-100.0, + ) + ledger = 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=1100.0, initial_cash=1000.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), + days=(opened,), synthetic=True, + ) + with tempfile.TemporaryDirectory() as tmp: + store = PerformanceStore(local_root=Path(tmp)) + store.save_research_ledger(ledger) + loaded = store.load_research_ledger("us_equity", "global_etf_rotation", "trial-a", "run-a", 1) + self.assertEqual(loaded, ledger) + self.assertEqual(loaded.days[0].trade_net_cashflow, 100.0) + self.assertEqual(loaded.days[0].equity_trade_cashflow, -100.0) + + def test_option_trade_day_rejects_missing_or_unbalanced_equity_cashflow(self) -> None: + short = _option_mark( + "SOXL-2024-01-17-90P", -1.0, -900.0, right="put", strike=90.0, + expiration=_END, premium_cashflow=900.0, + ) + long = _option_mark( + "SOXL-2024-01-17-80P", 1.0, 700.0, right="put", strike=80.0, + expiration=_END, premium_cashflow=-700.0, + ) + common = dict( + session_date=_START, cash=1100.0, positions=(ResearchPositionMark("SOXL", 2.0, 200.0), short, long), + trade_net_cashflow=100.0, fees=0.0, nav=1100.0, daily_return=0.0, + declared_event_ids=("short-open", "long-open"), + events=( + ResearchLedgerEvent("short-open", "option_trade", short.symbol, amount=900.0), + ResearchLedgerEvent("long-open", "option_trade", long.symbol, amount=-700.0), + ), + ) + for equity_cashflow in (None, -90.0): + with self.subTest(equity_cashflow=equity_cashflow), self.assertRaises(ValueError): + day = ResearchLedgerDay(**common, equity_trade_cashflow=equity_cashflow) + 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=1100.0, initial_cash=1000.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), + days=(day,), synthetic=True, + ) + def test_single_itm_put_assignment_requires_full_cash_and_stock_delivery(self) -> None: + expiration = _END + short = _option_mark( + "SOXL-2024-01-03-90P", -1.0, -900.0, right="put", strike=90.0, + expiration=expiration, premium_cashflow=900.0, + ) + long = _option_mark( + "SOXL-2024-01-03-80P", 1.0, 700.0, right="put", strike=80.0, + expiration=expiration, premium_cashflow=-700.0, + ) + opened = _mark_day( + _START, 10200.0, 0.0, 0.0, 200.0, 0.0, 10000.0, 10000.0, + declared_event_ids=("short-open", "long-open", "collateral-lock"), + events=( + ResearchLedgerEvent("short-open", "option_trade", short.symbol, amount=900.0), + ResearchLedgerEvent("long-open", "option_trade", long.symbol, amount=-700.0), + ResearchLedgerEvent("collateral-lock", "collateral_change", "CASH", amount=800.0), + ), + restricted_cash=800.0, + position_marks=(short, long), + ) + stock = ResearchPositionMark("SOXL", 100.0, 8500.0) + expiration_day = _mark_day( + _END, 1200.0, 100.0, 8500.0, 0.0, 0.0, 9700.0, 10000.0, + declared_event_ids=("expiry", "collateral-release"), + events=( + ResearchLedgerEvent("expiry", "option_settlement", "SOXL", settlement_price=85.0), + ResearchLedgerEvent("collateral-release", "collateral_change", "CASH", amount=-800.0), + ), + restricted_cash=0.0, + option_settlement_cashflow=-9000.0, + ) + ledger = 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=10000.0, initial_cash=10000.0, + initial_positions=(), days=(opened, dataclasses.replace(expiration_day, positions=(stock,))), + synthetic=True, + ) + self.assertEqual(ledger.days[-1].cash, 1200.0) + self.assertEqual(ledger.days[-1].positions[0].quantity, 100.0) + self.assertEqual(ledger.days[-1].nav, 9700.0) + + with self.assertRaisesRegex(ValueError, "ledger_event_type"): + ResearchLedgerEvent("early-assignment", "option_assignment", "SOXL", amount=-9000.0) + + funded_open = _mark_day( + _START, 1200.0, 0.0, 0.0, 200.0, 0.0, 1000.0, 1000.0, + declared_event_ids=("short-open", "long-open", "collateral-lock"), + events=( + ResearchLedgerEvent("short-open", "option_trade", short.symbol, amount=900.0), + ResearchLedgerEvent("long-open", "option_trade", long.symbol, amount=-700.0), + ResearchLedgerEvent("collateral-lock", "collateral_change", "CASH", amount=800.0), + ), restricted_cash=800.0, position_marks=(short, long), + ) + explicitly_funded = _mark_day( + _END, 200.0, 100.0, 8500.0, 0.0, 0.0, 8700.0, 1000.0, + external_cashflow=8000.0, + declared_event_ids=("expiry", "collateral-release"), + events=( + ResearchLedgerEvent("expiry", "option_settlement", "SOXL", settlement_price=85.0), + ResearchLedgerEvent("collateral-release", "collateral_change", "CASH", amount=-800.0), + ), restricted_cash=0.0, option_settlement_cashflow=-9000.0, + ) + funded_ledger = 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=1000.0, initial_cash=1000.0, + initial_positions=(), days=(funded_open, explicitly_funded), synthetic=True, + ) + self.assertEqual(funded_ledger.days[-1].external_cashflow, 8000.0) + self.assertEqual(funded_ledger.days[-1].cash, 200.0) + + def test_option_settlement_allows_declared_same_day_equity_trades_and_fees(self) -> None: + call = _option_mark( + "TQQQ-2024-01-03-90C", 1.0, 100.0, right="call", strike=90.0, + expiration=_END, premium_cashflow=-100.0, underlying="TQQQ", + ) + + def make_ledger( + initial_cash: float, + *, + initial_equity: bool = False, + include_quantities: bool = True, + declared_equity_cashflow: float | None = 1800.0, + trade_phase: str | None = "after_settlement", + ) -> ResearchDailyLedger: + starting_stock = ResearchPositionMark("TQQQ", 20.0, 2000.0) if initial_equity else None + initial_positions = (call,) if starting_stock is None else (call, starting_stock) + settlement_cashflow = -9000.0 + net_equity_flow = 1800.0 + fees = 5.0 + ending_cash = initial_cash + net_equity_flow - fees + settlement_cashflow + ending_shares = 100.0 if initial_equity else 80.0 + ending_nav = ending_cash + ending_shares * 100.0 + 2 * 100.0 + day = _mark_day( + _END, ending_cash, ending_shares, ending_shares * 100.0, + net_equity_flow, fees, ending_nav, + initial_cash + call.valuation + (2000.0 if initial_equity else 0.0), + declared_event_ids=("call-expiry",), + events=(ResearchLedgerEvent( + "call-expiry", "option_settlement", "TQQQ", settlement_price=100.0, + ),), + option_settlement_cashflow=settlement_cashflow, + equity_trade_cashflow=declared_equity_cashflow, + equity_trade_quantities={"TQQQ": -20.0, "SPY": 2.0} if include_quantities else None, + equity_trade_phase=trade_phase if include_quantities else None, + position_marks=( + ResearchPositionMark("TQQQ", ending_shares, ending_shares * 100.0), + ResearchPositionMark("SPY", 2.0, 200.0), + ), + ) + 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=initial_cash + call.valuation + (2000.0 if initial_equity else 0.0), + initial_cash=initial_cash, initial_positions=initial_positions, days=(day,), synthetic=True, + ) + + ledger = make_ledger(9900.0) + self.assertEqual(ledger.days[0].cash, 2695.0) + self.assertEqual(ledger.days[0].nav, 10895.0) + with tempfile.TemporaryDirectory() as tmp: + store = PerformanceStore(local_root=Path(tmp)) + store.save_research_ledger(ledger) + loaded = store.load_research_ledger("us_equity", "global_etf_rotation", "trial-a", "run-a", 1) + self.assertEqual(loaded, ledger) + self.assertEqual(loaded.days[0].equity_trade_quantities, {"SPY": 2.0, "TQQQ": -20.0}) + + with self.assertRaisesRegex(ValueError, "ledger_equity_trade_quantity"): + make_ledger(9900.0, trade_phase="before_settlement") + + before_settlement = make_ledger( + 8500.0, initial_equity=True, trade_phase="before_settlement", + ) + self.assertEqual(before_settlement.days[0].cash, 1295.0) + self.assertEqual(before_settlement.days[0].nav, 11495.0) + + with self.assertRaisesRegex(ValueError, "ledger_equity_trade_quantity"): + make_ledger(9900.0, include_quantities=False) + with self.assertRaisesRegex(ValueError, "ledger_option_settlement"): + make_ledger(9900.0, declared_equity_cashflow=1700.0) + with self.assertRaisesRegex(ValueError, "ledger_option_funding"): + make_ledger(8000.0) + + def test_insufficient_collateral_funding_missing_settlement_and_repeat_fail_closed(self) -> None: + expiration = _END + short = _option_mark( + "SOXL-2024-01-03-90P", -1.0, -900.0, right="put", strike=90.0, + expiration=expiration, premium_cashflow=900.0, + ) + long = _option_mark( + "SOXL-2024-01-03-80P", 1.0, 700.0, right="put", strike=80.0, + expiration=expiration, premium_cashflow=-700.0, + ) + events = ( + ResearchLedgerEvent("short-open", "option_trade", short.symbol, amount=900.0), + ResearchLedgerEvent("long-open", "option_trade", long.symbol, amount=-700.0), + ResearchLedgerEvent("collateral-lock", "collateral_change", "CASH", amount=700.0), + ) + under_collateralized = _mark_day( + _START, 1200.0, 0.0, 0.0, 200.0, 0.0, 1000.0, 1000.0, + declared_event_ids=tuple(event.event_id for event in events), events=events, restricted_cash=700.0, + position_marks=(short, long), + ) + base = dict( + 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=1000.0, initial_cash=1000.0, + initial_positions=(), synthetic=True, + ) + with self.assertRaisesRegex(ValueError, "ledger_option_collateral"): + ResearchDailyLedger(**base, days=(under_collateralized,)) + + funded_events = ( + events[0], events[1], + ResearchLedgerEvent("collateral-lock", "collateral_change", "CASH", amount=800.0), + ) + funded_open = dataclasses.replace( + under_collateralized, + declared_event_ids=tuple(event.event_id for event in funded_events), + events=funded_events, + restricted_cash=800.0, + ) + + insufficient_settlement = _mark_day( + _END, -7800.0, 100.0, 8500.0, 0.0, 0.0, 700.0, 1000.0, + declared_event_ids=("expiry", "collateral-release"), + events=( + ResearchLedgerEvent("expiry", "option_settlement", "SOXL", settlement_price=85.0), + ResearchLedgerEvent("collateral-release", "collateral_change", "CASH", amount=-800.0), + ), restricted_cash=0.0, option_settlement_cashflow=-9000.0, + ) + with self.assertRaisesRegex(ValueError, "ledger_option_funding"): + ResearchDailyLedger(**base, days=(funded_open, insufficient_settlement)) + + missing_settlement = _mark_day( + _END, 1200.0, 0.0, 0.0, 0.0, 0.0, 1000.0, 1000.0, + declared_event_ids=(), restricted_cash=800.0, position_marks=(short, long), + ) + with self.assertRaisesRegex(ValueError, "ledger_option_settlement"): + ResearchDailyLedger(**base, days=(funded_open, missing_settlement)) + + def test_dividend_accrual_and_payment_preserve_nav_and_read_back(self) -> None: + accrual = ResearchLedgerEvent("div-ex-1", "dividend_accrual", "SOXL", per_share=5.0) + payment = ResearchLedgerEvent( + "div-pay-1", "dividend_payment", "SOXL", amount=5.0, reference_event_id="div-ex-1" + ) + ex_day = _mark_day( + _START, 0.0, 1.0, 95.0, 0.0, 0.0, 100.0, 100.0, + dividend_receivable=5.0, declared_event_ids=("div-ex-1",), events=(accrual,), + ) + pay_day = _mark_day( + _END, 5.0, 1.0, 95.0, 0.0, 0.0, 100.0, 100.0, income_cashflow=5.0, + declared_event_ids=("div-pay-1",), events=(payment,), + ) + ledger = 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=0.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), + days=(ex_day, pay_day), synthetic=True, + ) + self.assertEqual((ex_day.nav, ex_day.dividend_receivable, ex_day.daily_return), (100.0, 5.0, 0.0)) + self.assertEqual((pay_day.nav, pay_day.dividend_receivable, pay_day.daily_return), (100.0, 0.0, 0.0)) + self.assertEqual(ledger.total_return, 0.0) + self.assertTrue(ledger.events_complete) + with tempfile.TemporaryDirectory() as tmp: + store = PerformanceStore(local_root=Path(tmp)) + store.save_research_ledger(ledger) + loaded = store.load_research_ledger("us_equity", "global_etf_rotation", "trial-a", "run-a", 1) + self.assertEqual(loaded, ledger) + self.assertTrue(all(day.declared_event_ids is not None for day in loaded.days)) + + def test_split_preserves_wealth_and_validates_quantity(self) -> None: + day = _mark_day( + _START, 0.0, 2.0, 100.0, 0.0, 0.0, 100.0, 100.0, + declared_event_ids=("split-1",), + events=(ResearchLedgerEvent("split-1", "split", "SOXL", ratio=2.0),), + ) + ledger = 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=0.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), days=(day,), synthetic=True, + ) + self.assertEqual(ledger.days[0].positions[0].quantity * 50.0, 100.0) + with self.assertRaisesRegex(ValueError, "ledger_split_quantity"): + 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=0.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), + days=(_mark_day( + _START, 0.0, 3.0, 150.0, 0.0, 0.0, 150.0, 100.0, + declared_event_ids=("split-1",), + events=(ResearchLedgerEvent("split-1", "split", "SOXL", ratio=2.0),), + ),), synthetic=True, + ) + + def test_event_declarations_pairing_duplicates_and_split_trade_fail_closed(self) -> None: + accrual = ResearchLedgerEvent("ex-1", "dividend_accrual", "SOXL", per_share=1.0) + day = _mark_day( + _START, 0.0, 1.0, 99.0, 0.0, 0.0, 100.0, 100.0, + dividend_receivable=1.0, declared_event_ids=("wrong-id",), events=(accrual,), + ) + with self.assertRaisesRegex(ValueError, "ledger_event_set"): + 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=0.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), days=(day,), synthetic=True, + ) + payment = ResearchLedgerEvent( + "pay-1", "dividend_payment", "SOXL", amount=1.0, reference_event_id="missing-ex" + ) + with self.assertRaisesRegex(ValueError, "ledger_dividend_pair"): + 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=0.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), + days=(_mark_day( + _START, 1.0, 1.0, 99.0, 0.0, 0.0, 100.0, 100.0, income_cashflow=1.0, + declared_event_ids=("pay-1",), events=(payment,), + ),), synthetic=True, + ) + split = ResearchLedgerEvent("split-2", "split", "SOXL", ratio=2.0) + with self.assertRaisesRegex(ValueError, "ledger_split_trade"): + 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=0.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), + days=(_mark_day( + _START, -10.0, 2.0, 110.0, -10.0, 0.0, 100.0, 100.0, + declared_event_ids=("split-2",), events=(split,), + ),), synthetic=True, + ) + valid_day = _mark_day( + _START, 0.0, 1.0, 99.0, 0.0, 0.0, 100.0, 100.0, + dividend_receivable=1.0, declared_event_ids=("ex-1",), events=(accrual,), + ) + repeated = ResearchLedgerEvent("ex-1", "dividend_accrual", "SOXL", per_share=1.0) + with self.assertRaisesRegex(ValueError, "ledger_event_duplicate"): + 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=0.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), + days=(valid_day, _mark_day( + _END, 0.0, 1.0, 98.0, 0.0, 0.0, 100.0, 100.0, + dividend_receivable=2.0, declared_event_ids=("ex-1",), events=(repeated,), + )), synthetic=True, + ) + with self.assertRaisesRegex(ValueError, "ledger_dividend_pair"): + 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=0.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), + days=( + valid_day, + _mark_day( + _END, 1.0, 1.0, 99.0, 0.0, 0.0, 100.0, 100.0, income_cashflow=1.0, + declared_event_ids=("pay-1",), events=(ResearchLedgerEvent( + "pay-1", "dividend_payment", "SOXL", amount=1.0, reference_event_id="ex-1" + ),), + ), + _mark_day( + date(2024, 1, 4), 2.0, 1.0, 99.0, 0.0, 0.0, 101.0, 100.0, income_cashflow=1.0, + declared_event_ids=("pay-retry",), events=(ResearchLedgerEvent( + "pay-retry", "dividend_payment", "SOXL", amount=1.0, + reference_event_id="ex-1", + ),), + ), + ), synthetic=True, + ) + with self.assertRaisesRegex(ValueError, "ledger_event_set"): + 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=0.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), + days=(day, _mark_day(_END, 0.0, 1.0, 100.0, 0.0, 0.0, 100.0, 100.0)), synthetic=True, + ) + with self.assertRaisesRegex(ValueError, "ledger_event_set"): + 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=0.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), + days=(_mark_day( + _START, 0.0, 1.0, 99.0, 0.0, 0.0, 100.0, 100.0, + dividend_receivable=1.0, events=(accrual,), + ),), synthetic=True, + ) + with self.assertRaisesRegex(ValueError, "ledger_dividend_cashflow"): + 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=0.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), + days=( + _mark_day( + _START, 0.0, 1.0, 99.0, 0.0, 0.0, 100.0, 100.0, + dividend_receivable=1.0, declared_event_ids=("ex-1",), events=(accrual,), + ), + _mark_day( + _END, 2.0, 1.0, 99.0, 0.0, 0.0, 101.0, 100.0, income_cashflow=2.0, + declared_event_ids=("pay-extra",), events=(ResearchLedgerEvent( + "pay-extra", "dividend_payment", "SOXL", amount=1.0, + reference_event_id="ex-1", + ),), + ), + ), synthetic=True, + ) + + def test_income_cashflow_conserves_cash_and_preserves_legacy_zero_payload(self) -> None: + income_day = _mark_day( + _START, + cash=105.0, + quantity=1.0, + valuation=110.0, + flow=0.0, + fees=0.0, + nav=215.0, + previous=200.0, + income_cashflow=5.0, + ) + ledger = 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=200.0, + initial_cash=100.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), + days=(income_day,), + synthetic=True, + ) + self.assertEqual(ledger.days[0].cash, 105.0) + self.assertEqual(ledger.days[0].nav, 215.0) + self.assertAlmostEqual(ledger.days[0].daily_return, 0.075) + self.assertEqual(ledger.days[0].to_dict()["income_cashflow"], 5.0) + with tempfile.TemporaryDirectory() as tmp: + store = PerformanceStore(local_root=Path(tmp)) + store.save_research_ledger(ledger) + loaded_income = store.load_research_ledger( + "us_equity", "global_etf_rotation", "trial-a", "run-a", 1 + ) + self.assertEqual(loaded_income, ledger) + self.assertEqual(loaded_income.days[0].income_cashflow, 5.0) + + old_ledger = _ledger() + old_payload = {"schema_version": SCHEMA_VERSION, **old_ledger.to_dict()} + old_day_payload = old_payload["days"][0] + self.assertNotIn("income_cashflow", old_day_payload) + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + path = root / "research_trial" / _identity_digest("us_equity", "global_etf_rotation", "trial-a") / "ledger.json" + path.parent.mkdir(parents=True) + path.write_text(json.dumps(old_payload), encoding="utf-8") + loaded_old = PerformanceStore(local_root=root).load_research_ledger( + "us_equity", "global_etf_rotation", "trial-a", "run-a", 1 + ) + self.assertEqual(loaded_old, old_ledger) + self.assertEqual(loaded_old.days[0].income_cashflow, 0.0) + self.assertEqual(loaded_old.days[0].dividend_receivable, 0.0) + self.assertIsNone(loaded_old.days[0].declared_event_ids) + self.assertFalse(loaded_old.events_complete) + + def test_tampered_cash_does_not_match_income_cashflow(self) -> None: + income_day = _mark_day( + _START, + cash=105.0, + quantity=1.0, + valuation=110.0, + flow=0.0, + fees=0.0, + nav=215.0, + previous=200.0, + income_cashflow=5.0, + ) + ledger = 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=200.0, initial_cash=100.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), + days=(income_day,), synthetic=True, + ) + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + store = PerformanceStore(local_root=root) + store.save_research_ledger(ledger) + path = root / "research_trial" / _identity_digest("us_equity", "global_etf_rotation", "trial-a") / "ledger.json" + planted = json.loads(path.read_text()) + planted["days"][0]["cash"] = 104.0 + path.write_text(json.dumps(planted), encoding="utf-8") + self.assertIsNone(store.load_research_ledger("us_equity", "global_etf_rotation", "trial-a", "run-a", 1)) + + def test_event_ledger_readback_rejects_tampered_cash_and_nav(self) -> None: + event = ResearchLedgerEvent("ex-1", "dividend_accrual", "SOXL", per_share=5.0) + day = _mark_day( + _START, 0.0, 1.0, 95.0, 0.0, 0.0, 100.0, 100.0, + dividend_receivable=5.0, declared_event_ids=("ex-1",), events=(event,), + ) + ledger = 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=0.0, + initial_positions=(ResearchPositionMark("SOXL", 1.0, 100.0),), days=(day,), synthetic=True, + ) + for key, value in (("cash", 1.0), ("nav", 101.0)): + with self.subTest(key=key), tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + store = PerformanceStore(local_root=root) + store.save_research_ledger(ledger) + path = root / "research_trial" / _identity_digest( + "us_equity", "global_etf_rotation", "trial-a" + ) / "ledger.json" + payload = json.loads(path.read_text()) + payload["days"][0][key] = value + path.write_text(json.dumps(payload), encoding="utf-8") + self.assertIsNone(store.load_research_ledger( + "us_equity", "global_etf_rotation", "trial-a", "run-a", 1 + )) + def test_rejected_before_parse_keeps_null_actual_params_and_cost_inputs(self) -> None: for field in dataclasses.fields(ResearchTrialRecord): + if field.name == "research_identity": + self.assertIsNone(field.default) + continue self.assertIs(field.default, dataclasses.MISSING) self.assertIs(field.default_factory, dataclasses.MISSING) with self.assertRaises(ValueError) as unknown: @@ -759,6 +1620,30 @@ def test_rejected_before_parse_keeps_null_actual_params_and_cost_inputs(self) -> for banned in ("sharpe_ratio", "total_return", "promotion_eligible"): self.assertNotIn(banned, payload) + def test_legacy_trial_payload_without_research_identity_round_trips(self) -> None: + legacy = { + "schema_version": SCHEMA_VERSION, + **_trial(ResearchTrialStatus.REJECTED, actual_params=None).to_dict(), + } + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + path = _research_file(root, "us_equity", "global_etf_rotation", "trial-a", "terminal") + path.parent.mkdir(parents=True) + path.write_text(json.dumps(legacy), encoding="utf-8") + loaded = PerformanceStore(local_root=root).load_research_trial( + "us_equity", "global_etf_rotation", "trial-a" + ) + self.assertEqual(loaded, _trial(ResearchTrialStatus.REJECTED, actual_params=None)) + self.assertIsNone(loaded.research_identity) + self.assertNotIn("research_identity", legacy) + + def test_research_identity_requires_a_nonempty_json_mapping(self) -> None: + for identity in ([], {}, {"source": object()}, {"source": float("nan")}): + with self.subTest(identity=identity): + with self.assertRaises(ValueError) as malformed: + _trial(ResearchTrialStatus.REJECTED, actual_params=None, research_identity=identity) + self.assertEqual(str(malformed.exception), "research_identity") + 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") @@ -866,10 +1751,13 @@ def test_illegal_ledger_is_rejected_and_planted_cash_is_not_rewritten(self) -> N _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_income: + _mark_day(_START, 100, 0, 0, 0, 0, 100, 100, income_cashflow=float("inf")) 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_income.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) @@ -991,6 +1879,116 @@ def test_success_is_written_last_and_binds_result_fields(self) -> None: self.assertEqual(ledger.observation_count, result.observation_count) self.assertEqual(ledger.total_return, result.total_return) + def test_research_identity_round_trips_and_must_match_result_params(self) -> None: + identity = {"producer": "research-runner", "source": {"revision": "rev-a", "input": "input-a"}} + ledger = _ledger() + started = _trial( + ResearchTrialStatus.STARTED, + actual_params=dict(_PARAMS), + param_set_id="set-a", + source_revision="rev-a", + cost_inputs={}, + research_identity=identity, + ) + success = _trial(ResearchTrialStatus.SUCCEEDED, research_identity=identity) + result = _result(ledger, params={**_PARAMS, "research_identity": identity}) + with tempfile.TemporaryDirectory() as tmp: + store = PerformanceStore(local_root=Path(tmp)) + store.save_backtest_result(result) + store.save_research_ledger(ledger) + store.save_research_trial(started) + store.save_research_trial(success) + loaded = store.load_research_trial("us_equity", "global_etf_rotation", "trial-a") + self.assertEqual(loaded, success) + self.assertEqual(dict(loaded.research_identity), identity) + + for params in (_PARAMS, {**_PARAMS, "research_identity": {"producer": "other"}}): + with self.subTest(params=params): + with tempfile.TemporaryDirectory() as tmp: + store = PerformanceStore(local_root=Path(tmp)) + store.save_backtest_result(_result(ledger, params=params)) + store.save_research_ledger(ledger) + with self.assertRaises(ValueError) as mismatch: + store.save_research_trial(success) + self.assertEqual(str(mismatch.exception), "research_trial_result_mismatch") + self.assertFalse( + _research_file(Path(tmp), "us_equity", "global_etf_rotation", "trial-a", "terminal").exists() + ) + + def test_started_trial_rejects_changed_research_identity(self) -> None: + started = _trial(ResearchTrialStatus.STARTED, research_identity={"producer": "runner-a"}) + terminal = _trial( + ResearchTrialStatus.FAILED, + reason_code="result_rejected", + research_identity={"producer": "runner-b"}, + ) + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + store = PerformanceStore(local_root=root) + store.save_research_trial(started) + with self.assertRaises(ValueError) as conflict: + store.save_research_trial(terminal) + self.assertEqual(str(conflict.exception), "research_trial_conflict") + self.assertFalse( + _research_file(root, "us_equity", "global_etf_rotation", "trial-a", "terminal").exists() + ) + + def test_research_identity_comparison_preserves_json_value_types(self) -> None: + started = _trial(ResearchTrialStatus.STARTED, research_identity={"schema": {"version": True}}) + terminal = _trial( + ResearchTrialStatus.FAILED, + reason_code="result_rejected", + research_identity={"schema": {"version": 1}}, + ) + with tempfile.TemporaryDirectory() as tmp: + store = PerformanceStore(local_root=Path(tmp)) + store.save_research_trial(started) + with self.assertRaises(ValueError) as conflict: + store.save_research_trial(terminal) + self.assertEqual(str(conflict.exception), "research_trial_conflict") + + ledger = _ledger() + success = _trial( + ResearchTrialStatus.SUCCEEDED, + research_identity={"schema": {"version": 1}}, + ) + result = _result( + ledger, + params={**_PARAMS, "research_identity": {"schema": {"version": True}}}, + ) + with tempfile.TemporaryDirectory() as tmp: + store = PerformanceStore(local_root=Path(tmp)) + store.save_backtest_result(result) + store.save_research_ledger(ledger) + with self.assertRaises(ValueError) as mismatch: + store.save_research_trial(success) + self.assertEqual(str(mismatch.exception), "research_trial_result_mismatch") + + def test_actual_params_comparison_preserves_json_value_types(self) -> None: + started = _trial(ResearchTrialStatus.STARTED, actual_params={"flag": True}) + terminal = _trial( + ResearchTrialStatus.FAILED, + reason_code="result_rejected", + actual_params={"flag": 1}, + ) + with tempfile.TemporaryDirectory() as tmp: + store = PerformanceStore(local_root=Path(tmp)) + store.save_research_trial(started) + with self.assertRaises(ValueError) as conflict: + store.save_research_trial(terminal) + self.assertEqual(str(conflict.exception), "research_trial_conflict") + + ledger = _ledger() + success = _trial(ResearchTrialStatus.SUCCEEDED, actual_params={"flag": 1}) + result = _result(ledger, params={"flag": True}) + with tempfile.TemporaryDirectory() as tmp: + store = PerformanceStore(local_root=Path(tmp)) + store.save_backtest_result(result) + store.save_research_ledger(ledger) + with self.assertRaises(ValueError) as mismatch: + store.save_research_trial(success) + self.assertEqual(str(mismatch.exception), "research_trial_result_mismatch") + def test_param_or_source_mismatch_does_not_create_terminal(self) -> None: ledger = _ledger() cases = { @@ -1033,6 +2031,7 @@ def test_terminal_keeps_known_started_identity(self) -> None: "periods": {"periods_per_year": 365.25}, "synthetic": {"synthetic": False}, "actual_params": {"actual_params": {"lookback": 21}}, + "research_identity": {"research_identity": {"producer": "other"}}, "cost_inputs": {"cost_inputs": {"commission_bps": 9.0}}, "cost_source": {"cost_source": "other_cost"}, }