diff --git a/docs/research/crypto_combo_accounting_correction_20261003.zh-CN.md b/docs/research/crypto_combo_accounting_correction_20261003.zh-CN.md new file mode 100644 index 0000000..f09c775 --- /dev/null +++ b/docs/research/crypto_combo_accounting_correction_20261003.zh-CN.md @@ -0,0 +1,13 @@ +# Crypto combo 历史收益口径修正（2026-10-03） + +`crypto_combo_backtest_20260628.json` 是旧版资金流口径的只读历史结果。它把每日 DCA 入金混入账户权益百分比变化，因此其中的年化收益、Sharpe、回撤和总收益均不代表投资绩效；本说明不覆盖或重写该 artifact。 + +当前 Crypto combo runner 以现金流调整后的单位净值收益计算 TWR、CAGR、Sharpe 和回撤，并另外报告 XIRR、期初/期末权益、累计注资、累计提现及净利润。该研究代理明确假设每笔 DCA 在当日收盘时到达，并按同一收盘价买入；窗口化指标使用窗口前一日的实际权益作为期初资本。现金流计时必须在同一评估序列中保持一致，TWR 依现金流时点切分并链乘。 + +XIRR 将同一自然日的现金流先合并并移除净额为零的日期。若净现金流存在多次正负号变化，XIRR 标为歧义并留空，不从数值搜索找到的单个候选根推断唯一解；该搜索不保证找到全部根。这样可避免非传统现金流因网格漏检而输出看似唯一的回报率。 + +combo 中的山寨币仍是由 ETH 收益缩放并加噪声生成的 synthetic proxy。它不是实际策略重放，也不提供真实成交、结算或成本证据。当前 proxy 未建模费用与滑点，结果显式标记 `cost_status=not_modelled`；为保持既有消费者数值兼容而输出的零成本数值不代表已测量成本。旧 artifact 没有在本次修正中重跑，也不得用于晋级或证明策略效果。 + +另有独立 `backtest.combo_replay` 研究入口，直接调用真实 `crypto_equity_combo.build_target_weights`，消费调用方提供并带 as-of/available-at 的 indicators、universe、benchmark 快照及 OHLC；它不生成 ETH 收益代理。信号只在 t 日收盘形成，t+1 自然日开盘按 `_rebalance_holdings` 调仓，随后以 t+1 收盘价估值；策略组合权益来自信号日收盘时已知的现金和持仓。显式 `fee_bps` / `slippage_bps` 是合成执行假设，按成交名义金额从现金扣除，并非测得的实际成本；本入口不支持外部现金流。当前工程案例仅用 synthetic 输入验证时序与账本，不构成策略表现、晋级或 live 证据。 + +现金流调整与绩效报告口径参照 [GIPS Standards Handbook for Firms](https://www.gipsstandards.org/standards/gips-standards-for-firms/gips-standards-handbook-for-firms/)；此处仅声明本模拟器的现金流时间约定，不代表 GIPS 合规认证。 diff --git a/scripts/research_crypto_combo_backtest.py b/scripts/research_crypto_combo_backtest.py index 8304848..ebf3949 100644 --- a/scripts/research_crypto_combo_backtest.py +++ b/scripts/research_crypto_combo_backtest.py @@ -233,6 +233,9 @@ def run_backtest( "orchestrator": { "equity": pd.Series(dtype=float), "metrics": payload["metrics"], + "accounting": payload["accounting"], + "cost_status": payload["cost_status"], + "simulation_model": payload["simulation_model"], "profile": payload["profile"], "source": payload["source"], } @@ -293,9 +296,24 @@ def run_backtest( metrics_c = _compute_metrics(equity_c, "Dynamic Combo") return { - "Pure BTC DCA": {"equity": equity_a, "metrics": metrics_a}, - "Static Combo": {"equity": equity_b, "metrics": metrics_b}, - "Dynamic Combo": {"equity": equity_c, "metrics": metrics_c}, + "Pure BTC DCA": { + "equity": equity_a, + "metrics": metrics_a, + "simulation_model": "historical_btc_dca_no_cost_model", + "cost_status": "not_modelled", + }, + "Static Combo": { + "equity": equity_b, + "metrics": metrics_b, + "simulation_model": "synthetic_alt_proxy_not_strategy_replay", + "cost_status": "not_modelled", + }, + "Dynamic Combo": { + "equity": equity_c, + "metrics": metrics_c, + "simulation_model": "synthetic_alt_proxy_not_strategy_replay", + "cost_status": "not_modelled", + }, } @@ -310,7 +328,10 @@ def _simulate_btc_dca( price = btc_close.reindex(idx) btc_units = pd.Series(DCA_AMOUNT_USD / price, index=idx) cum_units = btc_units.cumsum() - return cum_units * price + equity = cum_units * price + equity.attrs["external_flows"] = pd.Series(DCA_AMOUNT_USD, index=idx) + equity.attrs["flow_timing"] = "end_of_day" + return equity def _simulate_combo( @@ -385,15 +406,26 @@ def _simulate_combo( ) portfolio_values.append(btc_value + alt_value + cash_held) - return pd.Series(portfolio_values, index=idx) + equity = pd.Series(portfolio_values, index=idx) + equity.attrs["external_flows"] = pd.Series(DCA_AMOUNT_USD, index=idx) + equity.attrs["flow_timing"] = "end_of_day" + equity.attrs["simulation_model"] = "synthetic_alt_proxy_not_strategy_replay" + return equity def _compute_metrics( equity: pd.Series, label: str, ) -> dict[str, Any]: - """Compute per-period metrics for a single equity curve.""" + """Compute cashflow-adjusted per-period metrics for a single equity curve.""" + from crypto_strategies.backtest.live_pool_simulator import ( + _cashflow_accounting_metrics, + _cashflow_adjusted_returns, + _performance_metrics, + ) + periods_metrics: dict[str, Any] = {} + all_flows = equity.attrs.get("external_flows", pd.Series(0.0, index=equity.index)) for period_name, (start_str, end_str) in PERIODS.items(): start_ts = pd.Timestamp(start_str) end_ts = pd.Timestamp(end_str) @@ -404,21 +436,47 @@ def _compute_metrics( "max_drawdown": 0.0, "sharpe": 0.0, "total_return": 0.0, + "cumulative_contributions": 0.0, + "cumulative_withdrawals": 0.0, + "net_profit": 0.0, + "xirr": None, + "xirr_status": "unavailable_empty_period", + "cost_status": "not_modelled", } continue - start_val = sub.iloc[0] - end_val = sub.iloc[-1] - total_ret = end_val / start_val - 1.0 if start_val > 0 else 0.0 - ann_ret = _safe_annual_return(sub) - mdd = _safe_max_drawdown(sub) - sharpe = _safe_sharpe(sub) + flows = all_flows.reindex(sub.index).fillna(0.0).astype(float) + prior = equity.loc[equity.index < sub.index[0]] + initial_equity = float(prior.iloc[-1]) if not prior.empty else 0.0 + returns = _cashflow_adjusted_returns( + sub, + flows, + initial_equity=initial_equity, + flow_timing="end", + ) + perf = _performance_metrics(returns) + accounting = _cashflow_accounting_metrics( + sub, + flows, + initial_equity=initial_equity, + initial_equity_date=prior.index[-1] if not prior.empty else None, + ) periods_metrics[period_name] = { - "annual_return": round(float(ann_ret), 4), - "max_drawdown": round(float(mdd), 4), - "sharpe": round(float(sharpe), 4), - "total_return": round(float(total_ret), 4), + "annual_return": round(float(perf["CAGR"]), 4), + "max_drawdown": round(float(perf["Max Drawdown"]), 4), + "sharpe": round(float(perf["Sharpe"]), 4), + "total_return": round(float(perf["total_return"]), 4), + "twr_total_return": round(float((1.0 + returns).prod() - 1.0), 4), + "xirr": accounting["xirr"], + "xirr_status": accounting["xirr_status"], + "initial_equity": round(float(accounting["initial_equity"]), 4), + "ending_equity": round(float(accounting["ending_equity"]), 4), + "cumulative_contributions": round(float(accounting["cumulative_contributions"]), 4), + "cumulative_withdrawals": round(float(accounting["cumulative_withdrawals"]), 4), + "net_profit": round(float(accounting["net_profit"]), 4), + "flow_timing": accounting["flow_timing"], + "cost_status": "not_modelled", } return periods_metrics @@ -532,6 +590,9 @@ def main() -> None: "profile": payload["profile"], "metrics": payload["metrics"], "source": payload["source"], + "accounting": payload["accounting"], + "cost_status": payload["cost_status"], + "simulation_model": payload["simulation_model"], "orchestrator": True, }, indent=2, @@ -543,7 +604,11 @@ def main() -> None: # Strip equity curves for JSON output (too large) json_results: dict[str, Any] = {} for strat_name, data in results.items(): - json_results[strat_name] = {"metrics": data["metrics"]} + json_results[strat_name] = { + "metrics": data["metrics"], + "simulation_model": data["simulation_model"], + "cost_status": data["cost_status"], + } json.dump(json_results, sys.stdout, indent=2) print() else: diff --git a/src/crypto_strategies/backtest/combo_replay.py b/src/crypto_strategies/backtest/combo_replay.py new file mode 100644 index 0000000..28938bd --- /dev/null +++ b/src/crypto_strategies/backtest/combo_replay.py @@ -0,0 +1,491 @@ +"""Point-in-time replay for the Crypto Equity Combo strategy. + +This research-only runner consumes caller-supplied daily snapshots. It uses a +signal day's close to build targets, trades at the next calendar day's open, +and marks the resulting holdings at that day's close. Synthetic runs are +engineering checks, not promotion evidence. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +import math +from typing import Any + +import numpy as np +import pandas as pd + +from crypto_strategies.backtest.live_pool_simulator import _rebalance_holdings +from crypto_strategies.strategies import crypto_equity_combo +from crypto_strategies.strategies.crypto_trend_rotation import REQUIRED_FEATURE_COLUMNS + + +@dataclass(frozen=True) +class ComboSignalSnapshot: + """Explicit-date PIT inputs required to evaluate one close-time signal.""" + + signal_date: Any + indicators_as_of: Any + indicators_available_at: Any + indicators: Mapping[str, Mapping[str, Any]] + universe_as_of: Any + universe_available_at: Any + universe_source_version: str + universe: Sequence[str] + benchmark_as_of: Any + benchmark_available_at: Any + benchmark: Mapping[str, Any] + + +@dataclass(frozen=True) +class ComboReplayResult: + daily: pd.DataFrame + trades: pd.DataFrame + final_state: dict[str, Any] + simulation_model: str = "actual_strategy_replay" + evidence_kind: str = "synthetic" + research_only: bool = True + promotion_eligible: bool = False + live_ready: bool = False + size_zero_required: bool = True + no_order: bool = True + cost_status: str = "explicit_synthetic_assumption" + + +def _daily_date(value: Any, name: str) -> pd.Timestamp: + try: + timestamp = pd.Timestamp(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"{name} must be a valid calendar date") from exc + if pd.isna(timestamp) or timestamp.tz is not None or timestamp != timestamp.normalize(): + raise ValueError(f"{name} must be a timezone-naive midnight date") + return timestamp.normalize() + + +def _normalise_ohlc(ohlc: pd.DataFrame) -> pd.DataFrame: + if not isinstance(ohlc, pd.DataFrame) or not isinstance(ohlc.index, pd.MultiIndex): + raise ValueError("ohlc must be a DataFrame indexed by (date, symbol)") + if ohlc.index.nlevels != 2 or set(ohlc.index.names) != {"date", "symbol"}: + raise ValueError("ohlc index levels must be named date and symbol") + if not {"open", "close"}.issubset(ohlc.columns): + raise ValueError("ohlc must include open and close columns") + frame = ohlc.reset_index().copy() + frame["date"] = [_daily_date(value, "ohlc date") for value in frame["date"]] + frame["symbol"] = frame["symbol"].astype(str).str.strip().str.upper() + if (frame["symbol"] == "").any(): + raise ValueError("ohlc symbols must be non-empty") + if frame.duplicated(["date", "symbol"]).any(): + raise ValueError("duplicate date/symbol OHLC rows are not allowed") + return frame.set_index(["date", "symbol"]).sort_index() + + +def _normalise_indicator_map( + indicators: Mapping[str, Mapping[str, Any]], + universe: tuple[str, ...], + indicators_as_of: pd.Timestamp, + close_prices: pd.DataFrame, +) -> dict[str, dict[str, Any]]: + if not isinstance(indicators, Mapping): + raise ValueError("indicators snapshot must be a mapping") + normalised: dict[str, dict[str, Any]] = {} + for raw_symbol, raw_payload in indicators.items(): + symbol = str(raw_symbol).strip().upper() + if not symbol or symbol in normalised or not isinstance(raw_payload, Mapping): + raise ValueError("indicator symbols must be unique with mapping payloads") + normalised[symbol] = dict(raw_payload) + + required_fields = REQUIRED_FEATURE_COLUMNS - {"symbol"} + for symbol in ("BTCUSDT", *universe): + payload = normalised.get(symbol) + if payload is None: + message = "BTCUSDT indicators are required" if symbol == "BTCUSDT" else f"missing indicators for {symbol}" + raise ValueError(message) + missing = required_fields - set(payload) + if missing: + raise ValueError(f"missing required indicators for {symbol}: {', '.join(sorted(missing))}") + for field in required_fields: + value = payload[field] + if isinstance(value, (bool, np.bool_)) or not isinstance(value, (int, float, np.number)) or not math.isfinite(float(value)): + raise ValueError(f"indicator {symbol}.{field} must be finite") + if payload["close"] <= 0.0: + raise ValueError(f"indicator {symbol}.close must be positive") + if symbol == "BTCUSDT" and not isinstance(payload.get("regime_on"), bool): + raise ValueError("BTCUSDT indicators require an explicit boolean regime_on") + close = _required_price(close_prices, symbol, "close", indicators_as_of) + if not math.isclose(float(payload["close"]), close, rel_tol=1e-10, abs_tol=1e-10): + raise ValueError(f"{symbol} indicator close must match OHLC close on indicators_as_of") + return normalised + + +def _required_price( + prices: pd.DataFrame, + symbol: str, + column: str, + date: pd.Timestamp, +) -> float: + try: + value = float(prices.loc[(date, symbol), column]) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError(f"missing required {column} for {symbol} on {date.date()}") from exc + if not math.isfinite(value) or value <= 0.0: + raise ValueError(f"required {column} for {symbol} must be finite and positive") + return value + + +def _validate_snapshot(snapshot: ComboSignalSnapshot) -> tuple[pd.Timestamp, tuple[str, ...]]: + if not isinstance(snapshot, ComboSignalSnapshot): + raise ValueError("snapshots must contain ComboSignalSnapshot values") + signal_date = _daily_date(snapshot.signal_date, "signal_date") + for label, as_of_value, available_value in ( + ("indicators", snapshot.indicators_as_of, snapshot.indicators_available_at), + ("universe", snapshot.universe_as_of, snapshot.universe_available_at), + ("benchmark", snapshot.benchmark_as_of, snapshot.benchmark_available_at), + ): + as_of = _daily_date(as_of_value, f"{label}_as_of") + available_at = _daily_date(available_value, f"{label}_available_at") + if available_at < as_of: + raise ValueError(f"{label} snapshot availability precedes its as_of date") + if as_of > signal_date or available_at > signal_date: + raise ValueError(f"{label} snapshot available after signal date") + universe = tuple(str(symbol).strip().upper() for symbol in snapshot.universe) + if not universe or any(not symbol for symbol in universe) or len(set(universe)) != len(universe): + raise ValueError("universe must contain unique non-empty symbols") + if "BTCUSDT" in universe: + raise ValueError("universe must contain trend candidates only, not BTCUSDT") + if not str(snapshot.universe_source_version).strip(): + raise ValueError("universe_source_version is required") + if not isinstance(snapshot.benchmark, Mapping) or not isinstance(snapshot.benchmark.get("regime_on"), bool): + raise ValueError("benchmark snapshot requires an explicit boolean regime_on") + return signal_date, universe + + +def _validate_state( + initial_state: Mapping[str, Any], + *, + first_signal_date: pd.Timestamp, + first_universe: tuple[str, ...], +) -> dict[str, Any]: + if not isinstance(initial_state, Mapping) or not initial_state: + raise ValueError("initial_state must be a non-empty rotation state") + state = dict(initial_state) + if not state.get("rotation_pool_symbols"): + raise ValueError("initial_state requires non-empty rotation_pool_symbols") + if not str(state.get("trend_pool_version", "")).strip(): + raise ValueError("initial_state requires trend_pool_version") + state_as_of = _daily_date(state.get("trend_pool_as_of_date"), "initial trend_pool_as_of_date") + if state_as_of >= first_signal_date: + raise ValueError("initial rotation state must predate the first signal date") + cached_pool = tuple(str(symbol).strip().upper() for symbol in state["rotation_pool_symbols"]) + if not cached_pool or len(set(cached_pool)) != len(cached_pool): + raise ValueError("initial rotation pool must contain unique symbols") + if not set(cached_pool).issubset(first_universe): + raise ValueError("initial rotation pool must be within the first PIT universe") + return state + + +def run_crypto_equity_combo_replay( + snapshots: Sequence[ComboSignalSnapshot], + ohlc: pd.DataFrame, + *, + initial_cash: float, + fee_bps: float, + slippage_bps: float, + initial_state: Mapping[str, Any], + strategy_kwargs: Mapping[str, Any] | None = None, +) -> ComboReplayResult: + """Replay actual combo targets from a close signal at the following open. + + External flows are unsupported and therefore zero throughout the replay. + Fee and slippage rates are explicit assumptions, not measured execution + costs. The returned result is always synthetic research-only evidence. + """ + if len(snapshots) == 0: + raise ValueError("at least one signal snapshot is required") + if isinstance(initial_cash, (bool, np.bool_)) or not isinstance(initial_cash, (int, float, np.number)): + raise ValueError("initial_cash must be finite and positive") + cash = float(initial_cash) + if not math.isfinite(cash) or cash <= 0.0: + raise ValueError("initial_cash must be finite and positive") + if ( + isinstance(fee_bps, (bool, np.bool_)) + or isinstance(slippage_bps, (bool, np.bool_)) + or not isinstance(fee_bps, (int, float, np.number)) + or not isinstance(slippage_bps, (int, float, np.number)) + or not math.isfinite(float(fee_bps)) + or not math.isfinite(float(slippage_bps)) + or float(fee_bps) < 0.0 + or float(slippage_bps) < 0.0 + ): + raise ValueError("fee_bps and slippage_bps must be finite and non-negative") + fee_rate = float(fee_bps) / 10_000.0 + slippage_rate = float(slippage_bps) / 10_000.0 + cost_rate = fee_rate + slippage_rate + if cost_rate >= 1.0: + raise ValueError("combined transaction cost rate must be less than 1.0") + + allowed_strategy_kwargs = { + "btc_weight", "trend_weight", "dynamic_mode", "dynamic_regime_mode", + "dynamic_regime_off_cut", "rotation_top_n", "weight_mode", + "allow_rotation_refresh", "circuit_breaker_enabled", "btc_drawdown_threshold", + "vol_scaling_enabled", "target_vol", "max_leverage", "smart_multiplier_enabled", + } + forbidden_strategy_kwargs = { + "prices", "indicators_map", "universe_snapshot", "benchmark_snapshot", + "portfolio", "state", "as_of", "translator", "derived_indicators", + "zscore_exit_context", + } + strategy_options = dict(strategy_kwargs or {}) + if forbidden_strategy_kwargs & strategy_options.keys(): + raise ValueError("strategy_kwargs cannot override point-in-time replay inputs") + unknown_strategy_options = strategy_options.keys() - allowed_strategy_kwargs + if unknown_strategy_options: + raise ValueError(f"unsupported strategy configuration: {', '.join(sorted(unknown_strategy_options))}") + for key, value in strategy_options.items(): + values = value if isinstance(value, (tuple, list)) else (value,) + if any( + isinstance(item, Mapping) + or not isinstance(item, (str, bool, int, float, np.number)) + or ( + isinstance(item, (int, float, np.number)) + and not isinstance(item, (bool, np.bool_)) + and not math.isfinite(float(item)) + ) + for item in values + ): + raise ValueError(f"strategy configuration {key} must use finite scalar values") + + raw_signal_dates = [ + _daily_date(snapshot.signal_date, "signal_date") + for snapshot in snapshots + if isinstance(snapshot, ComboSignalSnapshot) + ] + if len(raw_signal_dates) != len(snapshots): + raise ValueError("snapshots must contain ComboSignalSnapshot values") + if len(set(raw_signal_dates)) != len(raw_signal_dates): + raise ValueError("signal dates must be unique") + prepared = [_validate_snapshot(snapshot) for snapshot in snapshots] + signal_dates = [date for date, _ in prepared] + if signal_dates != sorted(signal_dates): + raise ValueError("signal snapshots must be chronological") + if len(signal_dates) > 1 and not pd.DatetimeIndex(signal_dates).equals( + pd.date_range(signal_dates[0], periods=len(signal_dates), freq="D") + ): + raise ValueError("signal snapshots must be consecutive calendar days") + + prices = _normalise_ohlc(ohlc) + available_dates = set(prices.index.get_level_values("date")) + symbols = tuple(sorted(set(prices.index.get_level_values("symbol")))) + if "BTCUSDT" not in symbols: + raise ValueError("OHLC must include BTCUSDT") + if any(date not in available_dates or date + pd.Timedelta(days=1) not in available_dates for date in signal_dates): + raise ValueError("OHLC must include each signal date and its next natural calendar day") + + holdings = pd.Series(0.0, index=symbols, dtype=float) + state = _validate_state( + initial_state, + first_signal_date=signal_dates[0], + first_universe=prepared[0][1], + ) + daily_records: list[dict[str, Any]] = [] + trade_records: list[dict[str, Any]] = [] + + for snapshot, (signal_date, universe) in zip(snapshots, prepared, strict=True): + effective_date = signal_date + pd.Timedelta(days=1) + close_prices = prices.loc[signal_date] + if not isinstance(close_prices, pd.DataFrame): + raise ValueError("OHLC signal-date rows must contain symbol prices") + required_signal_symbols = {"BTCUSDT", *universe} | set(holdings[holdings > 0.0].index) + signal_close = { + symbol: _required_price(prices, symbol, "close", signal_date) + for symbol in required_signal_symbols + } + indicators = _normalise_indicator_map( + snapshot.indicators, + universe, + _daily_date(snapshot.indicators_as_of, "indicators_as_of"), + prices, + ) + held_values = { + symbol: float(holdings.loc[symbol]) * signal_close[symbol] + for symbol in holdings[holdings > 0.0].index + } + signal_equity = cash + sum(held_values.values()) + if not math.isfinite(signal_equity) or signal_equity <= 0.0: + raise ValueError("signal-time portfolio equity must remain finite and positive") + btc_value = held_values.get("BTCUSDT", 0.0) + portfolio = { + "total_equity": signal_equity, + "buying_power": cash, + "cash_balance": cash, + "positions": [ + {"symbol": symbol, "market_value": market_value} + for symbol, market_value in held_values.items() + ], + "metadata": {"dca_value": btc_value}, + } + + state["trend_pool_version"] = str(snapshot.universe_source_version).strip() + state["trend_pool_as_of_date"] = _daily_date( + snapshot.universe_as_of, "universe_as_of" + ).strftime("%Y-%m-%d") + targets, metadata = crypto_equity_combo.build_target_weights( + prices=signal_close, + indicators_map=indicators, + universe_snapshot=universe, + benchmark_snapshot=dict(snapshot.benchmark), + portfolio=portfolio, + state=state, + as_of=signal_date.strftime("%Y-%m-%d"), + **strategy_options, + ) + if not isinstance(targets, Mapping): + raise ValueError("strategy target weights must be a mapping") + if not isinstance(metadata, Mapping) or metadata.get("error"): + raise ValueError("strategy metadata error prevents successful replay") + try: + strategy_equity = float(metadata.get("total_equity")) + except (TypeError, ValueError) as exc: + raise ValueError("strategy metadata must include signal-time total_equity") from exc + if not math.isfinite(strategy_equity) or not math.isclose( + strategy_equity, signal_equity, rel_tol=1e-10, abs_tol=1e-8 + ): + raise ValueError("strategy metadata total_equity must equal signal-time ledger equity") + trend_metadata = metadata.get("trend_leg") + if not isinstance(trend_metadata, Mapping) or trend_metadata.get("error"): + raise ValueError("trend strategy metadata error prevents successful replay") + + target_values: dict[str, float] = {} + allowed_symbols = {"BTCUSDT", *universe} + for raw_symbol, raw_weight in targets.items(): + symbol = str(raw_symbol).strip().upper() + if not symbol or symbol not in allowed_symbols or symbol not in holdings.index: + raise ValueError(f"target weights reference unknown symbol: {raw_symbol}") + if isinstance(raw_weight, (bool, np.bool_)) or not isinstance(raw_weight, (int, float, np.number)): + raise ValueError("target weights must be finite non-negative numbers") + weight = float(raw_weight) + if not math.isfinite(weight) or weight < 0.0: + raise ValueError("target weights must be finite non-negative numbers") + target_values[symbol] = weight + if sum(target_values.values()) > 1.0 + 1e-10: + raise ValueError("target weights gross exposure must not exceed 1.0") + + btc_metadata = metadata.get("btc_leg") + dca_metadata = btc_metadata.get("dca_metadata") if isinstance(btc_metadata, Mapping) else None + required_dca_metadata = {"regime", "actionable", "planned_investment_usd"} + if not isinstance(dca_metadata, Mapping) or not required_dca_metadata.issubset(dca_metadata): + raise ValueError("BTC DCA metadata missing; fallback signals cannot be replayed successfully") + if not isinstance(dca_metadata["actionable"], bool): + raise ValueError("BTC DCA actionable metadata must be boolean") + try: + planned_investment = float(dca_metadata["planned_investment_usd"]) + except (TypeError, ValueError) as exc: + raise ValueError("BTC DCA planned investment metadata must be finite") from exc + if not math.isfinite(planned_investment) or planned_investment < 0.0: + raise ValueError("BTC DCA planned investment metadata must be finite and non-negative") + + target_series = pd.Series(0.0, index=symbols, dtype=float) + for symbol, weight in target_values.items(): + target_series.loc[symbol] = weight + + execution_open = pd.Series(np.nan, index=symbols, dtype=float) + for symbol in set(holdings[holdings > 0.0].index) | set(target_values): + execution_open.loc[symbol] = _required_price(prices, symbol, "open", effective_date) + pretrade_open_equity = cash + float( + (holdings * execution_open.fillna(0.0)).sum() + ) + previous_holdings = holdings.copy() + holdings, cash, transaction_cost, sale_notional, purchase_notional = _rebalance_holdings( + holdings, + cash, + execution_open, + target_series, + cost_rate=cost_rate, + ) + fee = transaction_cost * (fee_rate / cost_rate) if cost_rate else 0.0 + slippage = transaction_cost * (slippage_rate / cost_rate) if cost_rate else 0.0 + execution_equity_after_cost = cash + float( + (holdings * execution_open.fillna(0.0)).sum() + ) + if not math.isclose( + execution_equity_after_cost + transaction_cost, + pretrade_open_equity, + rel_tol=1e-9, + abs_tol=1e-8, + ): + raise ValueError("execution cash, holdings, and costs do not reconcile") + if cash < 0.0: + raise ValueError("execution would overdraw cash") + + current_trades: list[dict[str, Any]] = [] + for symbol in symbols: + quantity_delta = float(holdings.loc[symbol] - previous_holdings.loc[symbol]) + if abs(quantity_delta) <= 1e-12: + continue + open_price = float(execution_open.loc[symbol]) + notional = abs(quantity_delta) * open_price + trade_fee = notional * fee_rate + trade_slippage = notional * slippage_rate + trade = { + "signal_date": signal_date, + "effective_date": effective_date, + "symbol": symbol, + "side": "buy" if quantity_delta > 0.0 else "sell", + "quantity": abs(quantity_delta), + "reference_open": open_price, + "notional": notional, + "target_weight": target_values.get(symbol, 0.0), + "fee": trade_fee, + "slippage": trade_slippage, + "cost": trade_fee + trade_slippage, + } + current_trades.append(trade) + trade_records.append(trade) + if not math.isclose( + sum(row["cost"] for row in current_trades), + transaction_cost, + rel_tol=1e-8, + abs_tol=1e-8, + ): + raise ValueError("trade cost records do not reconcile to the cash ledger") + + execution_close = pd.Series(np.nan, index=symbols, dtype=float) + held_symbols = tuple(holdings[holdings > 0.0].index) + for symbol in held_symbols: + execution_close.loc[symbol] = _required_price(prices, symbol, "close", effective_date) + ending_equity = cash + float((holdings * execution_close.fillna(0.0)).sum()) + if not math.isfinite(ending_equity) or ending_equity <= 0.0: + raise ValueError("ending portfolio equity must remain finite and positive") + + daily_records.append( + { + "signal_date": signal_date, + "effective_date": effective_date, + "target_weights": dict(target_values), + "actual_shares": { + symbol: float(holdings.loc[symbol]) for symbol in held_symbols + }, + "cash": cash, + "fee": fee, + "slippage": slippage, + "transaction_cost": transaction_cost, + "external_flow": 0.0, + "signal_total_equity": signal_equity, + "pretrade_open_equity": pretrade_open_equity, + "execution_equity_after_cost": execution_equity_after_cost, + "execution_prices": { + symbol: float(execution_open.loc[symbol]) for symbol in held_symbols + }, + "ending_equity": ending_equity, + "cost_status": "explicit_synthetic_assumption", + "simulation_model": "actual_strategy_replay", + "evidence_kind": "synthetic", + } + ) + cash = float(cash) + + return ComboReplayResult( + daily=pd.DataFrame(daily_records), + trades=pd.DataFrame(trade_records), + final_state=state, + ) diff --git a/src/crypto_strategies/backtest/combo_simulator.py b/src/crypto_strategies/backtest/combo_simulator.py index 8fece3c..ea68cc1 100644 --- a/src/crypto_strategies/backtest/combo_simulator.py +++ b/src/crypto_strategies/backtest/combo_simulator.py @@ -2,13 +2,19 @@ from __future__ import annotations +import math from dataclasses import dataclass from typing import Any, Literal import numpy as np import pandas as pd -from crypto_strategies.backtest.live_pool_simulator import LivePoolBacktestResult, _performance_metrics +from crypto_strategies.backtest.live_pool_simulator import ( + LivePoolBacktestResult, + _cashflow_accounting_metrics, + _cashflow_adjusted_returns, + _performance_metrics, +) from crypto_strategies.strategies.crypto_equity_combo import ( DEFAULT_BTC_WEIGHT, DEFAULT_TREND_WEIGHT, @@ -65,11 +71,11 @@ def _compute_sma(series: pd.Series, window: int) -> pd.Series: return series.rolling(window=window, min_periods=window).mean() -def _combo_daily_returns( +def _combo_simulation( close: pd.DataFrame, *, combo_config: CryptoComboBacktestConfig, -) -> pd.Series: +) -> tuple[pd.Series, pd.Series, pd.Series]: btc_col = BTC_SYMBOL if BTC_SYMBOL in close.columns else close.columns[0] eth_col = ETH_SYMBOL if ETH_SYMBOL in close.columns else close.columns[min(1, len(close.columns) - 1)] @@ -77,7 +83,8 @@ def _combo_daily_returns( eth_close = close[eth_col].dropna() idx = btc_close.index.intersection(eth_close.index).sort_values() if len(idx) < combo_config.min_history_days: - return pd.Series(dtype=float) + empty = pd.Series(dtype=float) + return empty, empty, empty eth_returns = eth_close.pct_change().dropna() alt_returns = _simulate_alt_returns(eth_returns.reindex(idx).fillna(0.0)) @@ -103,6 +110,7 @@ def _combo_daily_returns( ) portfolio_values: list[float] = [] + external_flows: list[float] = [] alt_positions: dict[str, float] = {} btc_units = 0.0 cash_held = 0.0 @@ -119,6 +127,9 @@ def _combo_daily_returns( btc_alloc = combo_config.dca_amount_usd * combo_config.btc_weight trend_alloc = combo_config.dca_amount_usd * trend_weight btc_units += (btc_alloc + extra_btc_alloc) / btc_p + cash_held += combo_config.dca_amount_usd * ( + 1.0 - combo_config.btc_weight - combo_config.trend_weight + ) alt_prices_today: dict[str, float] = {} alt_candidates: list[str] = [] @@ -144,9 +155,20 @@ def _combo_daily_returns( for alt in ALTS ) portfolio_values.append(btc_value + alt_value + cash_held) + external_flows.append(combo_config.dca_amount_usd) equity = pd.Series(portfolio_values, index=idx) - return equity.pct_change().fillna(0.0) + flows = pd.Series(external_flows, index=idx, dtype=float) + returns = _cashflow_adjusted_returns(equity, flows, flow_timing="end") + return returns, equity, flows + + +def _combo_daily_returns( + close: pd.DataFrame, + *, + combo_config: CryptoComboBacktestConfig, +) -> pd.Series: + return _combo_simulation(close, combo_config=combo_config)[0] def run_combo_backtest( @@ -156,14 +178,46 @@ def run_combo_backtest( universe_symbols: Any = None, ) -> LivePoolBacktestResult: combo = combo_config or CryptoComboBacktestConfig() + if not math.isfinite(float(combo.dca_amount_usd)) or combo.dca_amount_usd <= 0.0: + raise ValueError("dca_amount_usd must be finite and positive") + weights = (float(combo.btc_weight), float(combo.trend_weight)) + if any(not math.isfinite(weight) or weight < 0.0 for weight in weights): + raise ValueError("btc_weight and trend_weight must be finite and non-negative") + if sum(weights) > 1.0: + raise ValueError("btc_weight and trend_weight must sum to at most 1.0") + if ( + not math.isfinite(float(combo.dynamic_trend_cut)) + or not 0.0 <= float(combo.dynamic_trend_cut) <= 1.0 + ): + raise ValueError("dynamic_trend_cut must be finite and between 0.0 and 1.0") symbols = tuple(universe_symbols or (BTC_SYMBOL, ETH_SYMBOL)) close = build_close_matrix(market_history, symbols=symbols) if len(close) < int(combo.min_history_days): raise ValueError( f"market_history requires at least {int(combo.min_history_days)} overlapping trading days" ) - returns = _combo_daily_returns(close, combo_config=combo) - return LivePoolBacktestResult(metrics=_performance_metrics(returns), returns=returns) + returns, equity, flows = _combo_simulation(close, combo_config=combo) + accounting = _cashflow_accounting_metrics( + equity, + flows, + initial_equity=0.0, + ) + metrics = _performance_metrics(returns) + accounting["TWR_total_return"] = float((1.0 + returns).prod() - 1.0) + accounting["TWR_CAGR"] = float(metrics["CAGR"]) + accounting["TWR_Sharpe"] = float(metrics["Sharpe"]) + metrics.update({key: value for key, value in accounting.items() if isinstance(value, float)}) + if accounting["xirr"] is not None: + metrics["XIRR"] = float(accounting["xirr"]) + return LivePoolBacktestResult( + metrics=metrics, + returns=returns, + cost_status="not_modelled", + accounting_status=str(accounting["status"]), + accounting=accounting, + equity=equity, + external_flows=flows, + ) __all__ = [ diff --git a/src/crypto_strategies/backtest/live_pool_simulator.py b/src/crypto_strategies/backtest/live_pool_simulator.py index 2a57555..438d8fa 100644 --- a/src/crypto_strategies/backtest/live_pool_simulator.py +++ b/src/crypto_strategies/backtest/live_pool_simulator.py @@ -17,6 +17,197 @@ class LivePoolBacktestResult: trade_log: pd.DataFrame = field(default_factory=lambda: pd.DataFrame( columns=["signal_date", "effective_date", "turnover", "fee", "slippage", "cost"] )) + cost_status: str = "not_modelled" + accounting_status: str = "unavailable" + accounting: dict[str, Any] = field(default_factory=dict) + equity: pd.Series = field(default_factory=lambda: pd.Series(dtype=float)) + external_flows: pd.Series = field(default_factory=lambda: pd.Series(dtype=float)) + + +def _cashflow_adjusted_returns( + equity: pd.Series, + external_flows: pd.Series, + *, + initial_equity: float = 0.0, + flow_timing: str = "end", +) -> pd.Series: + """Return daily time-weighted returns with explicitly timed external flows.""" + if not math.isfinite(float(initial_equity)) or initial_equity < 0.0: + raise ValueError("initial_equity must be finite and non-negative") + if flow_timing not in {"start", "end"}: + raise ValueError("flow_timing must be 'start' or 'end'") + if not equity.index.equals(external_flows.index): + raise ValueError("equity and external_flows must have identical indexes") + values = equity.astype(float) + flows = external_flows.astype(float) + if not np.isfinite(values.to_numpy()).all() or not np.isfinite(flows.to_numpy()).all(): + raise ValueError("equity and external_flows must be finite") + returns: list[float] = [] + prior_equity = float(initial_equity) + for value, flow in zip(values, flows, strict=True): + value = float(value) + flow = float(flow) + if prior_equity == 0.0: + if flow_timing != "end" or flow <= 0.0 or value <= 0.0: + raise ValueError("first period requires positive initial cash or initial_equity") + period_return = 0.0 + elif flow_timing == "end": + investment_value = value - flow + if prior_equity <= 0.0 or investment_value <= 0.0 or value <= 0.0: + raise ValueError("wealth must remain positive after external cash flows") + period_return = investment_value / prior_equity - 1.0 + else: + invested_equity = prior_equity + flow + if invested_equity <= 0.0 or value <= 0.0: + raise ValueError("wealth must remain positive after external cash flows") + period_return = value / invested_equity - 1.0 + if not math.isfinite(period_return) or period_return <= -1.0: + raise ValueError("net return must be finite and greater than -1.0") + if abs(period_return) < 1e-12: + period_return = 0.0 + if value <= 0.0: + raise ValueError("wealth must remain positive after external cash flows") + returns.append(period_return) + prior_equity = value + return pd.Series(returns, index=equity.index, dtype=float) + + +def _cashflow_accounting_metrics( + equity: pd.Series, + external_flows: pd.Series, + *, + initial_equity: float = 0.0, + initial_equity_date: Any | None = None, +) -> dict[str, Any]: + """Summarize portfolio cash flows; flows are signed and dated at period end.""" + if not equity.index.equals(external_flows.index): + raise ValueError("equity and external_flows must have identical indexes") + values = equity.astype(float) + flows = external_flows.astype(float) + if not np.isfinite(values.to_numpy()).all() or not np.isfinite(flows.to_numpy()).all(): + raise ValueError("equity and external_flows must be finite") + if not math.isfinite(float(initial_equity)) or initial_equity < 0.0: + raise ValueError("initial_equity must be finite and non-negative") + if values.empty: + return { + "status": "unavailable", + "xirr_status": "unavailable", + "xirr": None, + "initial_equity": float(initial_equity), + "ending_equity": float(initial_equity), + "cumulative_contributions": 0.0, + "cumulative_withdrawals": 0.0, + "net_contributions": 0.0, + "net_profit": 0.0, + "flow_timing": "end_of_day", + "initial_equity_date": None, + } + if (values <= 0.0).any(): + raise ValueError("wealth must remain positive") + dates = pd.DatetimeIndex(pd.to_datetime(values.index)).normalize() + investor_flows = -flows.to_numpy(dtype=float) + xirr_status: str + if initial_equity > 0.0: + if initial_equity_date is None: + xirr, xirr_status = None, "unavailable_missing_initial_equity_date" + else: + initial_date = pd.Timestamp(initial_equity_date).normalize() + if initial_date >= dates[0]: + raise ValueError("initial_equity_date must precede the first cash-flow date") + investor_flows = np.insert(investor_flows, 0, -float(initial_equity)) + dates = dates.insert(0, initial_date) + xirr = None + xirr_status = "unavailable" + else: + xirr = None + xirr_status = "unavailable" + if initial_equity <= 0.0 or initial_equity_date is not None: + investor_flows[-1] += float(values.iloc[-1]) + xirr, xirr_status = _solve_xirr(dates, investor_flows) + contributions = float(flows.clip(lower=0.0).sum()) + withdrawals = float(-flows.clip(upper=0.0).sum()) + net_contributions = float(initial_equity) + contributions - withdrawals + return { + "status": "computed", + "xirr_status": xirr_status, + "xirr": xirr, + "initial_equity": float(initial_equity), + "ending_equity": float(values.iloc[-1]), + "cumulative_contributions": contributions, + "cumulative_withdrawals": withdrawals, + "net_contributions": net_contributions, + "net_profit": float(values.iloc[-1]) - net_contributions, + "flow_timing": "end_of_day", + "cashflow_start_date": pd.Timestamp(values.index[0]).date().isoformat(), + "cashflow_end_date": pd.Timestamp(values.index[-1]).date().isoformat(), + "initial_equity_date": ( + pd.Timestamp(initial_equity_date).date().isoformat() + if initial_equity > 0.0 and initial_equity_date is not None + else None + ), + } + + +def _solve_xirr(dates: pd.DatetimeIndex, cashflows: np.ndarray) -> tuple[float | None, str]: + dates = pd.DatetimeIndex(pd.to_datetime(dates)).normalize() + cashflows = np.asarray(cashflows, dtype=float) + if len(dates) != len(cashflows) or not np.isfinite(cashflows).all(): + return None, "unavailable_invalid_cashflows" + if len(dates) and dates.nunique() == 1: + return None, "unavailable_no_elapsed_time" + grouped = pd.Series(cashflows, index=dates).groupby(level=0, sort=True).sum() + grouped = grouped.loc[grouped != 0.0] + dates = pd.DatetimeIndex(grouped.index) + cashflows = grouped.to_numpy(dtype=float) + if not len(dates): + return None, "unavailable_insufficient_signs" + day_offsets = (dates - dates[0]).days.to_numpy(dtype=float) / 365.25 + if np.ptp(day_offsets) == 0.0: + return None, "unavailable_no_elapsed_time" + if not (cashflows < 0).any() or not (cashflows > 0).any(): + return None, "unavailable_insufficient_signs" + signs = np.sign(cashflows) + sign_changes = int(np.count_nonzero(signs[1:] != signs[:-1])) + + def npv(rate: float) -> float: + try: + with np.errstate(over="raise", invalid="raise", divide="raise"): + return float(np.sum(cashflows / np.power(1.0 + rate, day_offsets))) + except (FloatingPointError, ZeroDivisionError): + return math.copysign(math.inf, float(cashflows[0])) + + grid = np.unique(np.concatenate((np.linspace(-0.999999, 1.0, 1000), np.geomspace(2.0, 1001.0, 800) - 1.0, [0.0]))) + roots: list[float] = [] + previous_rate = float(grid[0]) + previous_value = npv(previous_rate) + for rate_value in grid[1:]: + rate = float(rate_value) + value = npv(rate) + tolerance = 1e-12 * max(1.0, float(np.abs(cashflows).sum())) + if abs(value) <= tolerance: + roots.append(rate) + elif math.isfinite(value) and math.isfinite(previous_value) and value * previous_value < 0.0: + low, high = previous_rate, rate + low_value = previous_value + for _ in range(100): + mid = (low + high) / 2.0 + mid_value = npv(mid) + if abs(mid_value) <= 1e-10 or high - low <= 1e-12: + low = high = mid + break + if low_value * mid_value <= 0.0: + high = mid + else: + low, low_value = mid, mid_value + roots.append((low + high) / 2.0) + previous_rate, previous_value = rate, value + if len(roots) > 1: + return None, "ambiguous_multiple_roots" + if sign_changes > 1: + return None, "ambiguous_nonconventional_cashflows" + if not roots: + return None, "unavailable_no_root" + return min(roots, key=abs), "computed" def _performance_metrics( diff --git a/src/crypto_strategies/backtest/orchestrator_research.py b/src/crypto_strategies/backtest/orchestrator_research.py index 984eda9..48d26cf 100644 --- a/src/crypto_strategies/backtest/orchestrator_research.py +++ b/src/crypto_strategies/backtest/orchestrator_research.py @@ -91,6 +91,9 @@ def run_combo_profile_backtest( "start_date": result.start_date.isoformat() if result.start_date else None, "end_date": result.end_date.isoformat() if result.end_date else None, "metrics": _result_to_metrics(result), + "accounting": runner.last_accounting_metrics, + "cost_status": runner.cost_status, + "simulation_model": "synthetic_alt_proxy_not_strategy_replay", "source": "CryptoEquityComboBacktestRunner", "run_id": getattr(result, "run_id", None), } diff --git a/src/crypto_strategies/backtest/orchestrator_runner.py b/src/crypto_strategies/backtest/orchestrator_runner.py index 4db08c3..fa8a2ac 100644 --- a/src/crypto_strategies/backtest/orchestrator_runner.py +++ b/src/crypto_strategies/backtest/orchestrator_runner.py @@ -10,7 +10,12 @@ import pandas as pd from crypto_strategies.backtest.combo_simulator import ComboMode, CryptoComboBacktestConfig, run_combo_backtest -from crypto_strategies.backtest.live_pool_simulator import _performance_metrics, run_live_pool_rotation_backtest +from crypto_strategies.backtest.live_pool_simulator import ( + _cashflow_accounting_metrics, + _cashflow_adjusted_returns, + _performance_metrics, + run_live_pool_rotation_backtest, +) from crypto_strategies.strategies.crypto_equity_combo import PROFILE_NAME as CRYPTO_EQUITY_COMBO_PROFILE try: @@ -122,6 +127,7 @@ def _metrics_to_result( start_date: date | None, end_date: date | None, run_duration_seconds: float, + cost_model: str = "", ) -> Any: if BacktestResult is None: raise ImportError("quant_platform_kit is required to build BacktestResult") @@ -146,6 +152,7 @@ def _metrics_to_result( source_script="crypto_strategies.backtest.orchestrator_runner", computed_at=datetime.now(timezone.utc).isoformat(), run_duration_seconds=run_duration_seconds, + cost_model=cost_model, periods_per_year=CRYPTO_PERIODS_PER_YEAR, calendar_id=CRYPTO_CALENDAR_ID, ) @@ -234,6 +241,8 @@ def __init__( self._synthetic_days = int(synthetic_days) self._last_daily_returns = pd.Series(dtype=float) self._run_return_history: list[pd.Series] = [] + self._last_accounting_metrics: dict[str, Any] = {} + self._cost_status = "not_modelled" @property def last_daily_returns(self) -> pd.Series: @@ -243,6 +252,14 @@ def last_daily_returns(self) -> pd.Series: def run_return_history(self) -> tuple[pd.Series, ...]: return tuple(item.copy() for item in self._run_return_history) + @property + def last_accounting_metrics(self) -> dict[str, Any]: + return dict(self._last_accounting_metrics) + + @property + def cost_status(self) -> str: + return self._cost_status + def run( self, strategy_profile: str, @@ -283,10 +300,38 @@ def run( min_history_days=min_history_days, ), ) - self._last_daily_returns = _slice_daily_returns( - result.returns, - start_date=start_date, - end_date=end_date, + flow_dates = pd.to_datetime(result.external_flows.index, utc=False).tz_localize(None).normalize() + mask = pd.Series(True, index=result.external_flows.index) + if start_date is not None: + mask &= flow_dates >= pd.Timestamp(start_date) + if end_date is not None: + mask &= flow_dates <= pd.Timestamp(end_date) + window_flows = result.external_flows.loc[mask] + window_equity = result.equity.loc[mask] + if not window_flows.empty: + first_window_date = window_flows.index[0] + prior_equity = result.equity.loc[result.equity.index < first_window_date] + initial_equity = float(prior_equity.iloc[-1]) if not prior_equity.empty else 0.0 + initial_equity_date = prior_equity.index[-1] if not prior_equity.empty else None + self._last_daily_returns = _cashflow_adjusted_returns( + window_equity, + window_flows, + initial_equity=initial_equity, + flow_timing="end", + ) + self._last_accounting_metrics = _cashflow_accounting_metrics( + window_equity, + window_flows, + initial_equity=initial_equity, + initial_equity_date=initial_equity_date, + ) + else: + self._last_daily_returns = pd.Series(dtype=float) + self._last_accounting_metrics = _cashflow_accounting_metrics( + pd.Series(dtype=float), pd.Series(dtype=float) + ) + self._last_accounting_metrics["TWR_total_return"] = float( + (1.0 + self._last_daily_returns).prod() - 1.0 ) self._run_return_history.append(self._last_daily_returns.copy()) elapsed = (datetime.now(timezone.utc) - started).total_seconds() @@ -300,6 +345,7 @@ def run( start_date=start_date or (eval_frame["date"].min().date() if not eval_frame.empty else None), end_date=end_date or (eval_frame["date"].max().date() if not eval_frame.empty else None), run_duration_seconds=elapsed, + cost_model="not_modelled_synthetic_proxy", ) diff --git a/src/crypto_strategies/strategies/crypto_trend_rotation.py b/src/crypto_strategies/strategies/crypto_trend_rotation.py index 78996fb..d788f01 100644 --- a/src/crypto_strategies/strategies/crypto_trend_rotation.py +++ b/src/crypto_strategies/strategies/crypto_trend_rotation.py @@ -17,6 +17,7 @@ from __future__ import annotations +import math from typing import Any import pandas as pd @@ -152,39 +153,46 @@ def _apply_volatility_scaling( ) -> dict[str, float]: """Scale position weights inversely by volatility. - When vol_scaling_enabled, each weight is scaled so the - portfolio-level volatility stays near target_vol. - max_leverage caps the total exposure (1.0 = 100%). + When enabled, weighted single-asset volatility scales exposure toward + target_vol. This proxy does not estimate covariance portfolio risk. + max_leverage caps final gross exposure; scaling never increases weights. """ - if not coerce_bool(vol_scaling_enabled, default=True): - return weights if not weights: return weights - total_weight = sum(weights.values()) + if not all(math.isfinite(float(weight)) and float(weight) >= 0.0 for weight in weights.values()): + raise ValueError("weights must be finite and non-negative") + total_weight = sum(float(weight) for weight in weights.values()) if total_weight <= 0: return weights - # Estimate portfolio vol as weighted average of individual vols + target_vol = float(target_vol) + max_lev = float(max_leverage) + if not math.isfinite(target_vol) or target_vol <= 0.0: + raise ValueError("target_vol must be finite and positive") + if not math.isfinite(max_lev) or max_lev < 0.0: + raise ValueError("max_leverage must be finite and non-negative") + max_scale = min(1.0, max_lev / total_weight) + if not coerce_bool(vol_scaling_enabled, default=True): + return {symbol: weight * max_scale for symbol, weight in weights.items()} + + # Estimate risk from weighted single-asset vol; this is not covariance risk. weighted_vol = 0.0 vol_sum = 0.0 for symbol, weight in weights.items(): indicators = indicators_map.get(symbol, {}) vol20 = coerce_float(indicators.get("vol20"), default=float("nan")) - if not pd.isna(vol20) and vol20 > 0: + if math.isfinite(float(vol20)) and vol20 > 0: weighted_vol += weight * vol20 vol_sum += weight - if vol_sum <= 0 or weighted_vol <= 0: - return weights - - avg_vol = weighted_vol / vol_sum - target_vol = float(target_vol) - max_lev = float(max_leverage) + # Missing vol blocks exposure; an unknown risk estimate must not bypass either cap. + if vol_sum < total_weight: + return {symbol: 0.0 for symbol in weights} - # Scale: if current vol > target, reduce; if < target, allow up to max_leverage - scale = min(max_lev, target_vol / avg_vol) if avg_vol > 0 else max_lev - scale = max(0.5, min(1.0, scale)) # clamp to [0.5, 1.0] to avoid extreme moves + # Weighted single-asset vol is an exposure proxy, not covariance portfolio vol. + portfolio_vol_proxy = weighted_vol + scale = min(max_scale, target_vol / portfolio_vol_proxy) if portfolio_vol_proxy > 0 else max_scale return {symbol: weight * scale for symbol, weight in weights.items()} @@ -266,6 +274,8 @@ def build_target_weights( circuit_breaker_enabled = coerce_bool(config.get("circuit_breaker_enabled"), default=True) btc_drawdown_threshold = float(config.get("btc_drawdown_threshold", 0.30)) vol_scaling_enabled = coerce_bool(config.get("vol_scaling_enabled"), default=True) + target_vol = float(config.get("target_vol", 0.40)) + max_leverage = float(config.get("max_leverage", 1.0)) # Extract BTC benchmark from indicators_map btc_snapshot = _extract_btc_snapshot(indicators_map) @@ -307,6 +317,20 @@ def build_target_weights( weights_map, indicators_map, vol_scaling_enabled=vol_scaling_enabled, + target_vol=target_vol, + max_leverage=max_leverage, + ) + volatility_status = ( + "disabled" + if not vol_scaling_enabled + else "blocked_missing_volatility" + if any( + not math.isfinite(coerce_float(indicators_map.get(symbol, {}).get("vol20"), default=float("nan"))) + or coerce_float(indicators_map.get(symbol, {}).get("vol20"), default=float("nan")) <= 0.0 + for symbol, weight in weights_map.items() + if weight > 0.0 + ) + else "proxy_scaled" ) return { @@ -314,6 +338,7 @@ def build_target_weights( "weight": scaled_weights.get(sym, selected_candidates[sym]["weight"]), "relative_score": selected_candidates[sym]["relative_score"], "abs_momentum": selected_candidates[sym]["abs_momentum"], + "volatility_scaling_status": volatility_status, } for sym in selected_candidates } @@ -356,6 +381,8 @@ def compute_signals( "circuit_breaker_enabled": kwargs.get("circuit_breaker_enabled", True), "btc_drawdown_threshold": kwargs.get("btc_drawdown_threshold", 0.30), "vol_scaling_enabled": kwargs.get("vol_scaling_enabled", True), + "target_vol": kwargs.get("target_vol", 0.40), + "max_leverage": kwargs.get("max_leverage", 1.0), } frame = _to_indicator_frame(feature_snapshot) @@ -412,6 +439,7 @@ def compute_signals( "weight": float(payload["weight"]), "relative_score": float(payload["relative_score"]), "abs_momentum": float(payload["abs_momentum"]), + "volatility_scaling_status": payload["volatility_scaling_status"], } for sym, payload in selected.items() }, diff --git a/tests/test_backtest_correctness.py b/tests/test_backtest_correctness.py index a5d748f..27a302b 100644 --- a/tests/test_backtest_correctness.py +++ b/tests/test_backtest_correctness.py @@ -7,6 +7,7 @@ from pathlib import Path import pandas as pd +import numpy as np import pytest from crypto_strategies.backtest.live_pool_simulator import run_live_pool_rotation_backtest @@ -237,6 +238,125 @@ def test_cost_that_wipes_out_portfolio_fails_closed(final_open: float) -> None: run_live_pool_rotation_backtest(panel, top_n=1, fee_bps=20_000) +def test_cashflow_adjusted_returns_exclude_end_of_day_contributions_and_withdrawals() -> None: + from crypto_strategies.backtest.live_pool_simulator import _cashflow_adjusted_returns + + dates = pd.date_range("2024-01-01", periods=3, freq="D") + equity = pd.Series([100.0, 190.0, 180.0], index=dates) + flows = pd.Series([100.0, 100.0, -10.0], index=dates) + returns = _cashflow_adjusted_returns(equity, flows, flow_timing="end") + assert returns.tolist() == pytest.approx([0.0, -0.1, 0.0]) + + +def test_cashflow_adjusted_returns_support_start_of_period_flows() -> None: + from crypto_strategies.backtest.live_pool_simulator import _cashflow_adjusted_returns + + dates = pd.date_range("2024-01-01", periods=2, freq="D") + equity = pd.Series([110.0, 210.0], index=dates) + flows = pd.Series([0.0, 100.0], index=dates) + returns = _cashflow_adjusted_returns( + equity, flows, initial_equity=100.0, flow_timing="start" + ) + assert returns.tolist() == pytest.approx([0.1, 0.0]) + + +@pytest.mark.parametrize( + ("equity", "flows"), + [([100.0, float("nan")], [100.0, 0.0]), ([100.0, float("inf")], [100.0, 0.0]), ([100.0, -1.0], [100.0, 0.0])], +) +def test_cashflow_adjusted_returns_reject_nonfinite_or_nonpositive_wealth(equity, flows) -> None: + from crypto_strategies.backtest.live_pool_simulator import _cashflow_adjusted_returns + + dates = pd.date_range("2024-01-01", periods=2, freq="D") + with pytest.raises(ValueError): + _cashflow_adjusted_returns( + pd.Series(equity, index=dates), pd.Series(flows, index=dates), flow_timing="end" + ) + + +def test_flat_contributions_have_zero_xirr_and_profit() -> None: + from crypto_strategies.backtest.live_pool_simulator import _cashflow_accounting_metrics + + dates = pd.date_range("2024-01-01", periods=365, freq="D") + flows = pd.Series(100.0, index=dates) + equity = flows.cumsum() + result = _cashflow_accounting_metrics(equity, flows) + assert result["xirr"] == 0.0 + assert result["net_profit"] == 0.0 + assert result["cumulative_contributions"] == 36_500.0 + assert result["cumulative_withdrawals"] == 0.0 + assert result["cashflow_start_date"] == dates[0].date().isoformat() + assert result["cashflow_end_date"] == dates[-1].date().isoformat() + + +def test_cashflow_summary_counts_withdrawals_and_reports_net_profit() -> None: + from crypto_strategies.backtest.live_pool_simulator import _cashflow_accounting_metrics + + dates = pd.date_range("2024-01-01", periods=3, freq="D") + flows = pd.Series([100.0, 100.0, -10.0], index=dates) + equity = pd.Series([100.0, 190.0, 180.0], index=dates) + result = _cashflow_accounting_metrics(equity, flows) + assert result["cumulative_contributions"] == 200.0 + assert result["cumulative_withdrawals"] == 10.0 + assert result["net_contributions"] == 190.0 + assert result["net_profit"] == -10.0 + assert result["xirr_status"] == "computed" + + +def test_xirr_uses_actual_initial_capital_date_and_flags_multiple_roots() -> None: + from crypto_strategies.backtest.live_pool_simulator import ( + _cashflow_accounting_metrics, + _solve_xirr, + ) + + cashflow_day = pd.Timestamp("2024-01-10") + equity = pd.Series([110.0], index=pd.DatetimeIndex([cashflow_day])) + flows = pd.Series([0.0], index=equity.index) + result = _cashflow_accounting_metrics( + equity, + flows, + initial_equity=100.0, + initial_equity_date=pd.Timestamp("2024-01-01"), + ) + assert result["initial_equity_date"] == "2024-01-01" + assert result["xirr"] == pytest.approx(1.1 ** (365.25 / 9) - 1.0) + + dates = pd.DatetimeIndex( + pd.Timestamp("2020-01-01") + + pd.to_timedelta([0.0, 365.25, 730.5], unit="D") + ) + xirr, status = _solve_xirr(dates, np.array([-100.0, 230.0, -132.0])) + assert xirr is None + assert status == "ambiguous_multiple_roots" + + xirr, status = _solve_xirr( + pd.DatetimeIndex([cashflow_day, cashflow_day]), np.array([-100.0, 110.0]) + ) + assert xirr is None + assert status == "unavailable_no_elapsed_time" + + +def test_xirr_conservatively_rejects_nonconventional_cashflows() -> None: + from crypto_strategies.backtest.live_pool_simulator import _solve_xirr + + dates = pd.date_range("2020-01-01", periods=4, freq="365D") + xirr, status = _solve_xirr( + dates, + np.array([-100.0, 370.05, -451.13, 181.5825]), + ) + assert xirr is None + assert status in {"ambiguous_nonconventional_cashflows", "ambiguous_multiple_roots"} + + +def test_xirr_merges_same_day_cashflows_before_classifying_sign_changes() -> None: + from crypto_strategies.backtest.live_pool_simulator import _solve_xirr + + dates = pd.DatetimeIndex(["2020-01-01", "2020-01-01", "2021-01-01", "2022-12-31"]) + xirr, status = _solve_xirr(dates, np.array([-100.0, 50.0, -50.0, 110.0])) + assert xirr is not None + assert status == "computed" + + def test_synthetic_panel_digest_is_stable_across_hash_seeds() -> None: code = ( "import hashlib, sys; " diff --git a/tests/test_combo_orchestrator.py b/tests/test_combo_orchestrator.py index 3f111ba..9d2ac24 100644 --- a/tests/test_combo_orchestrator.py +++ b/tests/test_combo_orchestrator.py @@ -1,6 +1,11 @@ from __future__ import annotations +import io +import json +import sys import unittest +from contextlib import redirect_stderr, redirect_stdout +from unittest.mock import patch import pandas as pd @@ -36,6 +41,57 @@ def test_run_combo_backtest_dynamic_mode(self) -> None: ) self.assertGreater(result.metrics["Trading Days"], 0) + def test_flat_prices_with_daily_contributions_have_zero_return(self) -> None: + dates = pd.date_range("2024-01-01", periods=365, freq="D") + history = pd.DataFrame( + [{"date": day, "symbol": symbol, "close": 100.0} + for day in dates for symbol in ("BTCUSDT", "ETHUSDT")] + ) + zero_alt_returns = pd.DataFrame(0.0, index=dates, columns=("ETH", "SOL", "AVAX", "MATIC", "DOT")) + with patch("crypto_strategies.backtest.combo_simulator._simulate_alt_returns", return_value=zero_alt_returns): + result = run_combo_backtest( + history, + combo_config=CryptoComboBacktestConfig( + combo_mode="static", min_history_days=260, dca_amount_usd=100.0 + ), + ) + self.assertAlmostEqual(result.metrics["ending_equity"], 36_500.0) + self.assertAlmostEqual(result.metrics["net_profit"], 0.0) + self.assertEqual(result.metrics["CAGR"], 0.0) + self.assertEqual(result.metrics["Sharpe"], 0.0) + self.assertEqual(result.metrics["TWR_total_return"], 0.0) + self.assertEqual(result.metrics["XIRR"], 0.0) + self.assertEqual(result.cost_status, "not_modelled") + self.assertEqual(result.accounting_status, "computed") + + def test_unused_combo_budget_stays_in_cash_and_overallocation_is_rejected(self) -> None: + dates = pd.date_range("2024-01-01", periods=365, freq="D") + history = pd.DataFrame( + [{"date": day, "symbol": symbol, "close": 100.0} + for day in dates for symbol in ("BTCUSDT", "ETHUSDT")] + ) + zero_alt_returns = pd.DataFrame(0.0, index=dates, columns=("ETH", "SOL", "AVAX", "MATIC", "DOT")) + with patch("crypto_strategies.backtest.combo_simulator._simulate_alt_returns", return_value=zero_alt_returns): + result = run_combo_backtest( + history, + combo_config=CryptoComboBacktestConfig( + btc_weight=0.3, trend_weight=0.5, combo_mode="static", min_history_days=260 + ), + ) + self.assertAlmostEqual(result.metrics["ending_equity"], 36_500.0) + with self.assertRaisesRegex(ValueError, "sum to at most 1.0"): + run_combo_backtest( + history, + combo_config=CryptoComboBacktestConfig( + btc_weight=0.6, trend_weight=0.5, min_history_days=260 + ), + ) + with self.assertRaisesRegex(ValueError, "dca_amount_usd must be finite and positive"): + run_combo_backtest( + history, + combo_config=CryptoComboBacktestConfig(dca_amount_usd=float("nan"), min_history_days=260), + ) + class ComboOrchestratorResearchTests(unittest.TestCase): def test_run_combo_profile_backtest_with_fixture_history(self) -> None: @@ -47,6 +103,115 @@ def test_run_combo_profile_backtest_with_fixture_history(self) -> None: self.assertEqual(payload["profile"], CRYPTO_EQUITY_COMBO_PROFILE) self.assertEqual(payload["source"], "CryptoEquityComboBacktestRunner") self.assertGreater(payload["metrics"]["days"], 0) + self.assertEqual(payload["cost_status"], "not_modelled") + self.assertEqual(payload["simulation_model"], "synthetic_alt_proxy_not_strategy_replay") + self.assertEqual(payload["accounting"]["flow_timing"], "end_of_day") + self.assertIn("cumulative_contributions", payload["accounting"]) + self.assertIn("xirr_status", payload["accounting"]) + + def test_direct_and_orchestrator_combo_returns_share_twr_accounting(self) -> None: + from crypto_strategies.backtest.orchestrator_runner import CryptoEquityComboBacktestRunner + + history = _fixture_history() + config = CryptoComboBacktestConfig(min_history_days=260, combo_mode="static") + direct = run_combo_backtest(history, combo_config=config) + runner = CryptoEquityComboBacktestRunner(market_history=history) + runner.run( + CRYPTO_EQUITY_COMBO_PROFILE, + {"min_history_days": 260, "combo_mode": "static"}, + ) + pd.testing.assert_series_equal(runner.last_daily_returns, direct.returns) + self.assertAlmostEqual( + runner.last_accounting_metrics["TWR_total_return"], + direct.metrics["TWR_total_return"], + ) + + def test_legacy_metrics_exclude_flat_asset_contributions(self) -> None: + from scripts.research_crypto_combo_backtest import _compute_metrics + + dates = pd.date_range("2021-01-01", "2026-06-28", freq="D") + flows = pd.Series(100.0, index=dates) + equity = flows.cumsum() + equity.attrs["external_flows"] = flows + result = _compute_metrics(equity, "flat proxy") + full = result["Full Period"] + self.assertEqual(full["total_return"], 0.0) + self.assertEqual(full["twr_total_return"], 0.0) + self.assertEqual(full["annual_return"], 0.0) + self.assertEqual(full["net_profit"], 0.0) + self.assertEqual(full["xirr"], 0.0) + self.assertEqual(full["cost_status"], "not_modelled") + + def test_legacy_backtest_labels_proxy_and_unmodelled_costs(self) -> None: + from scripts.research_crypto_combo_backtest import run_backtest + + dates = pd.date_range("2021-01-01", periods=600, freq="D") + prices = pd.DataFrame( + {"btc_close": 100.0, "eth_close": 50.0}, + index=dates, + ) + with patch("scripts.research_crypto_combo_backtest._simulate_alt_returns") as simulate: + simulate.return_value = pd.DataFrame( + 0.0, + index=dates[1:], + columns=("ETH", "SOL", "AVAX", "MATIC", "DOT"), + ) + result = run_backtest(prices, orchestrator=False) + self.assertEqual(result["Static Combo"]["simulation_model"], "synthetic_alt_proxy_not_strategy_replay") + self.assertEqual(result["Dynamic Combo"]["simulation_model"], "synthetic_alt_proxy_not_strategy_replay") + self.assertTrue(all(item["cost_status"] == "not_modelled" for item in result.values())) + + def test_main_json_preserves_orchestrator_accounting_and_legacy_status(self) -> None: + from scripts import research_crypto_combo_backtest as research + + prices = pd.DataFrame({"btc_close": [1.0], "eth_close": [1.0]}, index=pd.date_range("2024-01-01", periods=1)) + orchestrator_payload = { + "profile": CRYPTO_EQUITY_COMBO_PROFILE, + "metrics": {"total_return": 0.1}, + "accounting": {"xirr": 0.1, "net_profit": 10.0}, + "cost_status": "not_modelled", + "simulation_model": "synthetic_alt_proxy_not_strategy_replay", + "source": "CryptoEquityComboBacktestRunner", + } + stdout = io.StringIO() + stderr = io.StringIO() + with ( + patch.object(sys, "argv", ["research_crypto_combo_backtest.py"]), + patch.object(research, "load_crypto_data", return_value=prices), + patch.object(research, "run_backtest", return_value={"orchestrator": orchestrator_payload}), + redirect_stdout(stdout), + redirect_stderr(stderr), + ): + research.main() + emitted = json.loads(stdout.getvalue()) + self.assertEqual(emitted["accounting"], orchestrator_payload["accounting"]) + self.assertEqual(emitted["cost_status"], "not_modelled") + self.assertEqual(emitted["simulation_model"], "synthetic_alt_proxy_not_strategy_replay") + + legacy_payload = { + name: { + "metrics": {}, + "simulation_model": model, + "cost_status": "not_modelled", + } + for name, model in ( + ("Pure BTC DCA", "historical_btc_dca_no_cost_model"), + ("Static Combo", "synthetic_alt_proxy_not_strategy_replay"), + ("Dynamic Combo", "synthetic_alt_proxy_not_strategy_replay"), + ) + } + stdout = io.StringIO() + with ( + patch.object(sys, "argv", ["research_crypto_combo_backtest.py", "--legacy", "--json-output"]), + patch.object(research, "load_crypto_data", return_value=prices), + patch.object(research, "run_backtest", return_value=legacy_payload), + redirect_stdout(stdout), + redirect_stderr(io.StringIO()), + ): + research.main() + emitted = json.loads(stdout.getvalue()) + self.assertEqual(emitted["Static Combo"]["simulation_model"], "synthetic_alt_proxy_not_strategy_replay") + self.assertEqual(emitted["Static Combo"]["cost_status"], "not_modelled") if __name__ == "__main__": diff --git a/tests/test_crypto_combo_replay.py b/tests/test_crypto_combo_replay.py new file mode 100644 index 0000000..0eb9359 --- /dev/null +++ b/tests/test_crypto_combo_replay.py @@ -0,0 +1,323 @@ +from __future__ import annotations + +from dataclasses import replace + +import pandas as pd +import pytest + +from crypto_strategies.backtest.combo_replay import ( + ComboSignalSnapshot, + run_crypto_equity_combo_replay, +) + + +SIGNAL_DATES = pd.date_range("2024-01-01", periods=2, freq="D") +ALL_DATES = pd.date_range("2024-01-01", periods=3, freq="D") +SYMBOLS = ("BTCUSDT", "ETHUSDT", "SOLUSDT") + + +def _indicators(day: pd.Timestamp) -> dict[str, dict[str, float | bool]]: + offsets = (day - SIGNAL_DATES[0]).days + btc_close = (100.0, 120.0)[offsets] + eth_close = (100.0, 270.0)[offsets] + sol_close = (50.0, 58.0)[offsets] + return { + "BTCUSDT": { + "close": btc_close, + "sma20": 90.0, + "sma60": 85.0, + "sma200": 80.0, + "roc20": 0.05, + "roc60": 0.10, + "roc120": 0.20, + "vol20": 0.20, + "avg_quote_vol_30": 1_000_000.0, + "avg_quote_vol_90": 900_000.0, + "avg_quote_vol_180": 800_000.0, + "trend_persist_90": 0.8, + "age_days": 2000.0, + "regime_on": True, + }, + "ETHUSDT": { + "close": eth_close, + "sma20": 90.0, + "sma60": 85.0, + "sma200": 80.0, + "roc20": 0.60, + "roc60": 0.80, + "roc120": 1.00, + "vol20": 0.25, + "avg_quote_vol_30": 2_000_000.0, + "avg_quote_vol_90": 1_800_000.0, + "avg_quote_vol_180": 1_600_000.0, + "trend_persist_90": 0.9, + "age_days": 1800.0, + }, + "SOLUSDT": { + "close": sol_close, + "sma20": 45.0, + "sma60": 40.0, + "sma200": 35.0, + "roc20": 0.30, + "roc60": 0.40, + "roc120": 0.50, + "vol20": 0.30, + "avg_quote_vol_30": 1_500_000.0, + "avg_quote_vol_90": 1_300_000.0, + "avg_quote_vol_180": 1_100_000.0, + "trend_persist_90": 0.7, + "age_days": 1500.0, + }, + } + + +def _snapshots() -> list[ComboSignalSnapshot]: + return [ + ComboSignalSnapshot( + signal_date=day, + indicators_as_of=day, + indicators_available_at=day, + indicators=_indicators(day), + universe_as_of=day, + universe_available_at=day, + universe_source_version="synthetic-universe-v1", + universe=("ETHUSDT", "SOLUSDT"), + benchmark_as_of=day, + benchmark_available_at=day, + benchmark={"regime_on": True, "ma200": 80.0, "ma200_slope": 0.01}, + ) + for day in SIGNAL_DATES + ] + + +def _ohlc() -> pd.DataFrame: + opens = { + "BTCUSDT": [100.0, 110.0, 115.0], + "ETHUSDT": [100.0, 250.0, 300.0], + "SOLUSDT": [50.0, 55.0, 60.0], + } + closes = { + "BTCUSDT": [100.0, 120.0, 118.0], + "ETHUSDT": [100.0, 270.0, 310.0], + "SOLUSDT": [50.0, 58.0, 62.0], + } + rows = [ + (day, symbol, opens[symbol][index], closes[symbol][index]) + for index, day in enumerate(ALL_DATES) + for symbol in SYMBOLS + ] + return pd.DataFrame( + [(open_price, close_price) for _, _, open_price, close_price in rows], + index=pd.MultiIndex.from_tuples( + [(day, symbol) for day, symbol, _, _ in rows], names=("date", "symbol") + ), + columns=("open", "close"), + ) + + +def _run(snapshots=None, ohlc=None, **kwargs): + call_kwargs = { + "initial_cash": 1_000.0, + "fee_bps": 10.0, + "slippage_bps": 20.0, + "initial_state": { + "rotation_pool_symbols": ["ETHUSDT", "SOLUSDT"], + "trend_pool_version": "synthetic-universe-v1", + "trend_pool_as_of_date": "2023-12-31", + }, + "strategy_kwargs": { + "btc_weight": 0.2, + "trend_weight": 0.6, + "dynamic_mode": False, + "rotation_top_n": 1, + "weight_mode": "equal", + "vol_scaling_enabled": False, + "circuit_breaker_enabled": False, + "smart_multiplier_enabled": False, + }, + } + call_kwargs.update(kwargs) + return run_crypto_equity_combo_replay( + _snapshots() if snapshots is None else snapshots, + _ohlc() if ohlc is None else ohlc, + **call_kwargs, + ) + + +def test_replay_calls_real_combo_strategy_and_trades_next_open_without_lookahead() -> None: + from crypto_strategies.strategies import crypto_equity_combo + + captured: list[tuple[dict, dict]] = [] + original = crypto_equity_combo.build_target_weights + + def capture(*args, **kwargs): + targets, metadata = original(*args, **kwargs) + captured.append((kwargs["prices"], kwargs["portfolio"])) + return targets, metadata + + crypto_equity_combo.build_target_weights = capture + try: + result = _run() + finally: + crypto_equity_combo.build_target_weights = original + + first = result.daily.iloc[0] + assert result.simulation_model == "actual_strategy_replay" + assert result.evidence_kind == "synthetic" + assert result.research_only is True + assert first["signal_date"] == SIGNAL_DATES[0] + assert first["effective_date"] == SIGNAL_DATES[1] + assert first["target_weights"]["ETHUSDT"] == pytest.approx(0.6) + assert first["target_weights"]["BTCUSDT"] > 0.0 + assert sum(first["target_weights"].values()) < 1.0 + assert first["actual_shares"]["BTCUSDT"] > 0.0 + assert first["actual_shares"]["ETHUSDT"] > 0.0 + assert captured[0][0]["ETHUSDT"] == 100.0 + assert captured[0][1]["total_equity"] == pytest.approx(1_000.0) + assert first["actual_shares"]["ETHUSDT"] < 1_000.0 / 250.0 + assert first["cash"] > 0.0 + assert len(result.daily) == 2 + assert result.final_state["rotation_pool_source_as_of_date"] == "2024-01-02" + assert result.daily.iloc[1]["signal_total_equity"] == pytest.approx( + result.daily.iloc[0]["ending_equity"] + ) + + +def test_replay_costs_reconcile_to_cash_and_purchase_never_overdraws() -> None: + result = _run() + + for _, row in result.daily.iterrows(): + execution_equity_after_cost = row["cash"] + sum( + shares * row["execution_prices"][symbol] + for symbol, shares in row["actual_shares"].items() + ) + assert row["transaction_cost"] == pytest.approx(row["fee"] + row["slippage"]) + assert execution_equity_after_cost + row["transaction_cost"] == pytest.approx( + row["pretrade_open_equity"] + ) + assert row["cash"] >= 0.0 + assert row["external_flow"] == 0.0 + + assert result.cost_status == "explicit_synthetic_assumption" + + +def test_future_feature_availability_is_rejected() -> None: + snapshots = _snapshots() + snapshots[0] = replace( + snapshots[0], indicators_available_at=SIGNAL_DATES[0] + pd.Timedelta(days=1) + ) + with pytest.raises(ValueError, match="available after signal date"): + _run(snapshots=snapshots) + + snapshots[0] = replace( + snapshots[0], indicators_as_of=SIGNAL_DATES[0], + indicators_available_at=SIGNAL_DATES[0] - pd.Timedelta(days=1), + ) + with pytest.raises(ValueError, match="availability precedes its as_of"): + _run(snapshots=snapshots) + + snapshots = _snapshots() + mismatched = dict(snapshots[0].indicators) + mismatched["ETHUSDT"] = {**mismatched["ETHUSDT"], "close": 101.0} + snapshots[0] = replace(snapshots[0], indicators=mismatched) + with pytest.raises(ValueError, match="indicator close must match OHLC close"): + _run(snapshots=snapshots) + + +def test_duplicate_ohlc_symbol_and_missing_held_quote_are_rejected() -> None: + duplicate = pd.concat([_ohlc(), _ohlc().iloc[[0]]]) + with pytest.raises(ValueError, match="duplicate date/symbol"): + _run(ohlc=duplicate) + + missing_close = _ohlc() + missing_close.loc[(SIGNAL_DATES[1], "ETHUSDT"), "close"] = float("nan") + with pytest.raises(ValueError, match="required close"): + _run(ohlc=missing_close) + + +def test_missing_btc_signal_metadata_fallback_is_not_successful_replay() -> None: + from crypto_strategies.strategies import crypto_btc_dca + + original = crypto_btc_dca.build_rebalance_plan + + def fail_plan(*args, **kwargs): + raise ValueError("synthetic BTC plan failure") + + crypto_btc_dca.build_rebalance_plan = fail_plan + try: + with pytest.raises(ValueError, match="BTC DCA metadata"): + _run() + finally: + crypto_btc_dca.build_rebalance_plan = original + + +def test_missing_btc_indicators_and_invalid_targets_are_rejected() -> None: + snapshots = _snapshots() + snapshots[0] = replace( + snapshots[0], indicators={key: value for key, value in snapshots[0].indicators.items() if key != "BTCUSDT"} + ) + with pytest.raises(ValueError, match="BTCUSDT indicators"): + _run(snapshots=snapshots) + + from crypto_strategies.strategies import crypto_equity_combo + + original = crypto_equity_combo.build_target_weights + crypto_equity_combo.build_target_weights = lambda **kwargs: ( + {"ETHUSDT": float("nan")}, + { + "total_equity": kwargs["portfolio"]["total_equity"], + "btc_leg": { + "dca_metadata": { + "regime": "ordinary_dca", + "actionable": True, + "planned_investment_usd": 0.0, + } + }, + "trend_leg": {}, + }, + ) + try: + with pytest.raises(ValueError, match="target weights"): + _run() + crypto_equity_combo.build_target_weights = lambda **kwargs: ( + {"ETHUSDT": 1.2}, + { + "total_equity": kwargs["portfolio"]["total_equity"], + "btc_leg": { + "dca_metadata": { + "regime": "ordinary_dca", + "actionable": True, + "planned_investment_usd": 0.0, + } + }, + "trend_leg": {}, + }, + ) + with pytest.raises(ValueError, match="gross exposure must not exceed"): + _run() + finally: + crypto_equity_combo.build_target_weights = original + + +def test_future_initial_rotation_state_and_unchecked_strategy_data_are_rejected() -> None: + future_state = { + "rotation_pool_symbols": ["ETHUSDT"], + "trend_pool_version": "synthetic-universe-v1", + "trend_pool_as_of_date": "2024-01-02", + } + with pytest.raises(ValueError, match="must predate the first signal date"): + _run(initial_state=future_state) + + with pytest.raises(ValueError, match="cannot override point-in-time replay inputs"): + _run(strategy_kwargs={"derived_indicators": {"BTCUSDT": {"roc20": 0.9}}}) + + +def test_duplicate_signal_dates_and_missing_target_open_are_rejected() -> None: + snapshots = _snapshots() + snapshots[1] = replace(snapshots[1], signal_date=snapshots[0].signal_date) + with pytest.raises(ValueError, match="signal dates must be unique"): + _run(snapshots=snapshots) + + missing_open = _ohlc().drop(index=(SIGNAL_DATES[1], "ETHUSDT")) + with pytest.raises(ValueError, match="missing required open"): + _run(ohlc=missing_open) diff --git a/tests/test_crypto_trend_rotation.py b/tests/test_crypto_trend_rotation.py index 406fb15..626c154 100644 --- a/tests/test_crypto_trend_rotation.py +++ b/tests/test_crypto_trend_rotation.py @@ -98,6 +98,73 @@ def test_compute_signals_with_valid_data(self) -> None: self.assertIn("managed_symbols", metadata) self.assertIn("selected_candidates", metadata) + def test_compute_signals_honors_low_volatility_target_and_leverage_cap(self) -> None: + feature_snapshot = pd.DataFrame([ + { + "symbol": symbol, + "close": close, + "sma20": close * 0.95, + "sma60": close * 0.90, + "sma200": close * 0.80, + "roc20": 0.20, + "roc60": 0.35, + "roc120": 0.60, + "vol20": 2.0, + "avg_quote_vol_30": 60_000_000.0, + "avg_quote_vol_90": 50_000_000.0, + "avg_quote_vol_180": 45_000_000.0, + "trend_persist_90": 0.80, + "age_days": 500, + } + for symbol, close in (("SOLUSDT", 180.0), ("ETHUSDT", 3000.0)) + ]) + weights, _, _, _, metadata = compute_signals( + feature_snapshot, + [], + target_vol=0.40, + max_leverage=0.20, + ) + self.assertIsNotNone(weights) + self.assertAlmostEqual(sum(weights.values()), 0.20) + self.assertTrue(all( + payload["volatility_scaling_status"] == "proxy_scaled" + for payload in metadata["selected_candidates"].values() + )) + self.assertTrue(all( + payload["volatility_scaling_status"] == "proxy_scaled" + for payload in metadata["selected_candidates"].values() + )) + + def test_scaling_caps_gross_exposure_for_large_weights_and_unknown_volatility(self) -> None: + from crypto_strategies.strategies.crypto_trend_rotation import _apply_volatility_scaling + + for weights in ({"A": 1.0}, {"A": 1.2, "B": 0.8}): + scaled = _apply_volatility_scaling( + weights, + {symbol: {"vol20": 2.0} for symbol in weights}, + target_vol=0.40, + max_leverage=0.20, + ) + self.assertAlmostEqual(sum(scaled.values()), 0.20) + + capped_without_vol_scaling = _apply_volatility_scaling( + {"A": 0.8, "B": 0.7}, + {"A": {"vol20": 0.2}, "B": {"vol20": 0.3}}, + vol_scaling_enabled=False, + max_leverage=0.20, + ) + self.assertAlmostEqual(sum(capped_without_vol_scaling.values()), 0.20) + + blocked_unknown = _apply_volatility_scaling( + {"A": 1.0}, {"A": {"vol20": float("nan")}}, max_leverage=0.20 + ) + self.assertEqual(blocked_unknown, {"A": 0.0}) + + zero_cap = _apply_volatility_scaling( + {"A": 1.0}, {"A": {"vol20": 0.2}}, max_leverage=0.0 + ) + self.assertEqual(zero_cap, {"A": 0.0}) + def test_compute_signals_empty_snapshot(self) -> None: """An empty feature snapshot should return None weights.""" feature_snapshot = pd.DataFrame() diff --git a/tests/test_orchestrator_runner.py b/tests/test_orchestrator_runner.py index 9c12cc4..99a0a61 100644 --- a/tests/test_orchestrator_runner.py +++ b/tests/test_orchestrator_runner.py @@ -89,6 +89,9 @@ def test_run_returns_backtest_result(self) -> None: self.assertLessEqual(runner.last_daily_returns.index.max().date(), date(2024, 6, 1)) self.assertEqual(result.observation_count, len(runner.last_daily_returns)) self.assertEqual(len(runner.run_return_history), 1) + self.assertEqual(result.cost_model, "not_modelled_synthetic_proxy") + self.assertEqual(runner.cost_status, "not_modelled") + self.assertIn("net_profit", runner.last_accounting_metrics) def test_invalid_combo_mode_raises(self) -> None: runner = CryptoEquityComboBacktestRunner(synthetic_days=1600)