From 2e9f99f9f1814a1adbba716632f0626115286a3e Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 29 Jun 2026 05:58:19 +0800 Subject: [PATCH] feat: enhance crypto DCA strategies with smart sizing and exit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - crypto_btc_dca: add AHR999 cycle multiplier, drawdown-based sizing, Z-score exit (逃顶), execution window, and bilingual i18n - crypto_trend_rotation: fix empty btc_snapshot bug, add circuit breaker and volatility-based position scaling - crypto_equity_combo: simplify to pure combinator, delegate BTC leg to enhanced smart DCA - Extract shared utilities to _utils.py - Add backtest script comparing 5 DCA strategies (2021-2026) - Update catalog configs and entrypoints Co-Authored-By: Claude --- scripts/research_crypto_dca_backtest.py | 632 ++++++++++++ src/crypto_strategies/_utils.py | 93 ++ src/crypto_strategies/catalog.py | 55 +- src/crypto_strategies/entrypoints/__init__.py | 32 + .../strategies/crypto_btc_dca.py | 930 +++++++++++++++++- .../strategies/crypto_equity_combo.py | 340 ++++--- .../strategies/crypto_trend_rotation.py | 251 ++++- tests/test_crypto_btc_dca.py | 389 +++++++- tests/test_crypto_trend_rotation.py | 14 +- 9 files changed, 2516 insertions(+), 220 deletions(-) create mode 100644 scripts/research_crypto_dca_backtest.py create mode 100644 src/crypto_strategies/_utils.py diff --git a/scripts/research_crypto_dca_backtest.py b/scripts/research_crypto_dca_backtest.py new file mode 100644 index 0000000..ec3242a --- /dev/null +++ b/scripts/research_crypto_dca_backtest.py @@ -0,0 +1,632 @@ +"""BTC DCA strategy backtest: compare ordinary vs smart DCA with escape. + +Simulates five BTC DCA strategies (2021-2026): + + A. Ordinary DCA — fixed monthly buy on day 25, no smart sizing + B. Smart (AHR999) — AHR999 cycle multiplier only + C. Smart (Drawdown) — drawdown-based multiplier only + D. Smart (Full) — AHR999 + drawdown, pick max multiplier + E. Smart + Escape — Full smart + Z-score exit to USDT + +Metrics: CAGR / Max Drawdown / Sharpe / Calmar / total return +Periods: Full / Bear 2022 / Bull 2023-2024 / Recent 2025-2026 + +Usage +----- + cd CryptoStrategies + PYTHONPATH=src:scripts python scripts/research_crypto_dca_backtest.py +""" + +from __future__ import annotations + +import argparse +import math +import sys +from typing import Any + +import numpy as np +import pandas as pd + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +START_DATE = "2021-01-01" +END_DATE = "2026-06-28" + +DCA_AMOUNT_USD = 100.0 # monthly base investment +MONTHLY_DAY = 25 +MONTHLY_WINDOW = 5 + +# AHR999 thresholds (matching the strategy config) +AHR999_BOTTOM = 0.45 +AHR999_ACCUMULATION = 0.80 +AHR999_DCA = 1.20 + +AHR999_BOTTOM_MULT = 3.0 +AHR999_ACCUMULATION_MULT = 2.25 +AHR999_DCA_MULT = 1.50 +AHR999_EXPENSIVE_MULT = 0.0 # skip buy + +# Drawdown thresholds +MILD_DD = 0.12 +DEEP_DD = 0.25 +SEVERE_DD = 0.40 +MILD_DISCOUNT_GAP = 0.08 +DEEP_DISCOUNT_GAP = 0.18 +EXPENSIVE_GAP = 0.30 +VERY_EXPENSIVE_GAP = 0.60 + +MILD_MULT = 1.50 +DEEP_MULT = 2.25 +SEVERE_MULT = 3.0 + +# Z-score exit thresholds (simulated — real implementation uses MVRV Z-Score) +ZSCORE_EXIT_SOFT = 7.0 # risk_reduced → 50% BTC +ZSCORE_EXIT_HARD = 9.0 # risk_off → 25% BTC + +BITCOIN_GENESIS = pd.Timestamp("2009-01-03") + +PERIODS: dict[str, tuple[str, str]] = { + "Full Period": (START_DATE, END_DATE), + "Bear (2022)": ("2022-01-01", "2022-12-31"), + "Bull (2023-2024)": ("2023-01-01", "2024-12-31"), + "Recent (2025-2026)": ("2025-01-01", END_DATE), +} + + +# --------------------------------------------------------------------------- +# Data loading +# --------------------------------------------------------------------------- + + +def load_btc_data() -> pd.DataFrame: + """Download BTC-USD daily data from yfinance.""" + try: + import yfinance as yf + raw = yf.download( + "BTC-USD", + start=START_DATE, + end=END_DATE, + auto_adjust=True, + progress=False, + threads=False, + ) + except ImportError: + from quant_strategy_plugins.yfinance_prices import download_price_history + raw = download_price_history(["BTC-USD"], start=START_DATE, end=END_DATE) + raw = raw.pivot_table(index="as_of", columns="symbol", values="close", aggfunc="last") + + closes = raw["Close"].copy() if "Close" in raw.columns else raw["close"].copy() + if isinstance(closes, pd.DataFrame): + closes = closes.iloc[:, 0] + closes = closes.astype(float).dropna() + closes.index = pd.to_datetime(closes.index).normalize() + return closes.sort_index() + + +# --------------------------------------------------------------------------- +# Indicators +# --------------------------------------------------------------------------- + + +def _sma(series: pd.Series, window: int) -> pd.Series: + return series.rolling(window=window, min_periods=window).mean() + + +def _gma(series: pd.Series, window: int) -> pd.Series: + """Geometric moving average.""" + log_values = np.log(series.clip(lower=1e-10)) + return np.exp(log_values.rolling(window=window, min_periods=window).mean()) + + +def _estimate_price(as_of: pd.Timestamp) -> float: + """Bitcoin age-based fair price estimate (power-law model).""" + age_days = max(1, (as_of.normalize() - BITCOIN_GENESIS).days) + return float(10 ** (5.84 * math.log10(age_days) - 17.01)) + + +def _ahr999_from_series(price, sma200, gma200, estimate) -> float: + """Compute AHR999 index: (price/gma200) * (price/estimate_price).""" + if sma200 <= 0 or gma200 <= 0 or estimate <= 0: + return float("nan") + return float((price / gma200) * (price / estimate)) + + +def _mayer_multiple(price, sma200) -> float: + """Mayer Multiple: price / SMA200.""" + return float(price / sma200) if sma200 > 0 else float("nan") + + +def _simulate_zscore(price, sma200, gma200) -> float: + """Approximate MVRV Z-Score using market-value to realized-value ratio. + + This is a coarse proxy — real MVRV Z-Score requires on-chain data. + Uses (Mayer Multiple - 1) / rolling_std as a rough stand-in. + """ + if sma200 <= 0 or gma200 <= 0: + return float("nan") + # Simplified: use Mayer Multiple deviation from 1.0 as proxy + mm = price / sma200 + return (mm - 1.0) / 0.5 # rough normalization + + +def compute_indicators(btc_close: pd.Series) -> pd.DataFrame: + """Compute all needed indicators from price series.""" + df = pd.DataFrame({"close": btc_close}, index=btc_close.index) + df["sma200"] = _sma(df["close"], 200) + df["gma200"] = _gma(df["close"], 200) + df["sma50"] = _sma(df["close"], 50) + df["high252"] = df["close"].rolling(252, min_periods=1).max() + df["drawdown_252d"] = 1.0 - df["close"] / df["high252"] + df["sma200_gap"] = df["close"] / df["sma200"] - 1.0 + df["rsi14"] = _compute_rsi(df["close"], 14) + df["vol20"] = df["close"].pct_change().rolling(20).std() + + df["ahr999"] = df.apply( + lambda r: _ahr999_from_series(r["close"], r["sma200"], r["gma200"], + _estimate_price(r.name)), + axis=1, + ) + df["mayer_multiple"] = df.apply(lambda r: _mayer_multiple(r["close"], r["sma200"]), axis=1) + df["zscore_proxy"] = df.apply( + lambda r: _simulate_zscore(r["close"], r["sma200"], r["gma200"]), axis=1, + ) + + return df + + +def _compute_rsi(series: pd.Series, period: int = 14) -> pd.Series: + delta = series.diff() + gain = delta.clip(lower=0) + loss = (-delta).clip(lower=0) + avg_gain = gain.rolling(window=period, min_periods=period).mean() + avg_loss = loss.rolling(window=period, min_periods=period).mean() + rs = avg_gain / avg_loss.replace(0, np.nan) + return 100.0 - (100.0 / (1.0 + rs)) + + +# --------------------------------------------------------------------------- +# Multiplier logic (matching the strategy implementation) +# --------------------------------------------------------------------------- + + +def get_ahr999_multiplier(ahr999: float) -> tuple[float, str]: + if pd.isna(ahr999): + return 1.0, "normal" + if ahr999 <= AHR999_BOTTOM: + return AHR999_BOTTOM_MULT, "ahr999_bottom" + if ahr999 <= AHR999_ACCUMULATION: + return AHR999_ACCUMULATION_MULT, "ahr999_accumulation" + if ahr999 <= AHR999_DCA: + return AHR999_DCA_MULT, "ahr999_dca" + return AHR999_EXPENSIVE_MULT, "ahr999_expensive" + + +def get_drawdown_multiplier( + drawdown: float, sma_gap: float, rsi: float, +) -> tuple[float, str]: + if pd.isna(drawdown) or pd.isna(sma_gap): + return 1.0, "normal" + if drawdown >= SEVERE_DD: + return SEVERE_MULT, "severe_pullback" + if drawdown >= DEEP_DD or sma_gap <= -abs(DEEP_DISCOUNT_GAP): + return DEEP_MULT, "deep_pullback" + if drawdown >= MILD_DD or sma_gap <= -abs(MILD_DISCOUNT_GAP): + return MILD_MULT, "mild_pullback" + if sma_gap >= VERY_EXPENSIVE_GAP and drawdown <= 0.05 and not pd.isna(rsi) and rsi >= 75: + return 1.0, "very_expensive_overbought" + if sma_gap >= EXPENSIVE_GAP and drawdown <= 0.05: + return 1.0, "expensive" + return 1.0, "normal" + + +def get_smart_multiplier(row: pd.Series) -> tuple[float, str, str]: + """Get combined multiplier: max of AHR999 and drawdown multipliers. + + Returns (multiplier, regime, source). + """ + ahr999 = row.get("ahr999", float("nan")) + dd = row.get("drawdown_252d", 0.0) + gap = row.get("sma200_gap", 0.0) + rsi = row.get("rsi14", float("nan")) + + a_mult, a_regime = get_ahr999_multiplier(ahr999) + d_mult, d_regime = get_drawdown_multiplier(dd, gap, rsi) + + if a_mult >= d_mult: + return a_mult, a_regime, "ahr999" + return d_mult, d_regime, "drawdown" + + +def should_zscore_exit(row: pd.Series) -> tuple[bool, str, float]: + """Check Z-score exit signal. + + Returns (should_exit, route, target_btc_exposure). + """ + zscore = row.get("zscore_proxy", float("nan")) + if pd.isna(zscore): + return False, "normal", 1.0 + if zscore >= ZSCORE_EXIT_HARD: + return True, "risk_off", 0.25 + if zscore >= ZSCORE_EXIT_SOFT: + return True, "risk_reduced", 0.50 + return False, "normal", 1.0 + + +# --------------------------------------------------------------------------- +# Simulation +# --------------------------------------------------------------------------- + + +def run_strategies( + btc_close: pd.Series, + indicators: pd.DataFrame, + warm_up: int = 252, +) -> dict[str, pd.Series]: + """Simulate all five strategies and return equity curves.""" + idx = indicators.index[warm_up:] + equity_curves: dict[str, pd.Series] = {} + + # A. Ordinary DCA + equity_curves["A. Ordinary DCA"] = _simulate_ordinary_dca(indicators, idx) + + # B. Smart (AHR999 only) + equity_curves["B. Smart (AHR999)"] = _simulate_smart_dca( + indicators, idx, use_ahr999=True, use_drawdown=False, use_escape=False, + ) + + # C. Smart (Drawdown only) + equity_curves["C. Smart (Drawdown)"] = _simulate_smart_dca( + indicators, idx, use_ahr999=False, use_drawdown=True, use_escape=False, + ) + + # D. Smart (Full) + equity_curves["D. Smart (Full)"] = _simulate_smart_dca( + indicators, idx, use_ahr999=True, use_drawdown=True, use_escape=False, + ) + + # E. Smart + Escape + equity_curves["E. Smart + Escape"] = _simulate_smart_dca( + indicators, idx, use_ahr999=True, use_drawdown=True, use_escape=True, + ) + + return equity_curves + + +def _is_execution_day(day: int, month: int, year: int) -> bool: + """Check if this day falls within the execution window (day 25 ± 5).""" + return abs(day - MONTHLY_DAY) < MONTHLY_WINDOW + + +def _simulate_ordinary_dca( + indicators: pd.DataFrame, + idx: pd.DatetimeIndex, +) -> pd.Series: + """Simulate ordinary monthly DCA.""" + btc_units = 0.0 + cash_held = 0.0 + equity = [] + + for date in idx: + price = float(indicators.loc[date, "close"]) + if pd.isna(price) or price <= 0: + if equity: + equity.append(equity[-1]) + else: + equity.append(0.0) + continue + + if _is_execution_day(date.day, date.month, date.year): + btc_units += DCA_AMOUNT_USD / price + + btc_value = btc_units * price + equity.append(btc_value + cash_held) + + return pd.Series(equity, index=idx) + + +def _simulate_smart_dca( + indicators: pd.DataFrame, + idx: pd.DatetimeIndex, + *, + use_ahr999: bool = True, + use_drawdown: bool = True, + use_escape: bool = False, +) -> pd.Series: + """Simulate smart DCA with configurable features.""" + btc_units = 0.0 + usdt_units = 0.0 # cash parked in USDT + equity = [] + escape_active = False + escape_target_btc = 1.0 # proportion of portfolio in BTC + + for date in idx: + row = indicators.loc[date] + price = float(row["close"]) + if pd.isna(price) or price <= 0: + if equity: + equity.append(equity[-1]) + else: + equity.append(0.0) + continue + + # Check Z-score exit + if use_escape: + should_exit, escape_route, target_btc = should_zscore_exit(row) + if should_exit: + escape_active = True + escape_target_btc = target_btc + else: + escape_active = False + escape_target_btc = 1.0 + + # On execution day, buy BTC with smart sizing + if _is_execution_day(date.day, date.month, date.year): + multiplier = 1.0 + regime = "ordinary_dca" + + if use_ahr999 or use_drawdown: + ahr999 = row.get("ahr999", float("nan")) + dd = row.get("drawdown_252d", 0.0) + gap = row.get("sma200_gap", 0.0) + rsi = row.get("rsi14", float("nan")) + + if use_ahr999 and use_drawdown: + multiplier, regime, _ = get_smart_multiplier(row) + elif use_ahr999: + multiplier, regime = get_ahr999_multiplier(ahr999) + elif use_drawdown: + multiplier, regime = get_drawdown_multiplier(dd, gap, rsi) + + invest_amount = max(0.0, DCA_AMOUNT_USD * max(0.0, multiplier)) + if multiplier > 0.0 and invest_amount > 0: + btc_units += min(invest_amount, DCA_AMOUNT_USD * 3.0) / price + + # Z-score exit: rebalance between BTC and USDT + if escape_active: + total_value = btc_units * price + usdt_units + target_btc_value = total_value * escape_target_btc + if total_value > 0: + # Sell BTC to USDT if over target + current_btc_value = btc_units * price + if current_btc_value > target_btc_value: + excess = current_btc_value - target_btc_value + btc_units -= excess / price + usdt_units += excess + elif usdt_units > 0: + # Buy back BTC from USDT if under target + deficit = target_btc_value - current_btc_value + buy_back = min(deficit, usdt_units) + btc_units += buy_back / price + usdt_units -= buy_back + + total_value = btc_units * price + usdt_units + equity.append(total_value) + + return pd.Series(equity, index=idx) + + +# --------------------------------------------------------------------------- +# Metrics +# --------------------------------------------------------------------------- + + +def _cagr(equity: pd.Series) -> float: + """CAGR from first positive value to last value.""" + nonzero = equity[equity > 0] + if len(nonzero) < 2: + return 0.0 + start_val = nonzero.iloc[0] + end_val = nonzero.iloc[-1] + if start_val <= 0: + return 0.0 + total_return = end_val / start_val - 1.0 + n_days = max(len(nonzero), 1) + years = n_days / 365.25 + if years <= 0: + return 0.0 + return (1.0 + total_return) ** (1.0 / years) - 1.0 + + +def _max_drawdown(equity: pd.Series) -> float: + nonzero = equity[equity > 0] + if len(nonzero) < 2: + return 0.0 + rolling_max = nonzero.expanding().max() + dd = (nonzero - rolling_max) / rolling_max + return float(dd.min()) + + +def _sharpe(equity: pd.Series) -> float: + nonzero = equity[equity > 0] + if len(nonzero) < 2: + return 0.0 + daily_returns = nonzero.pct_change().dropna() + if daily_returns.empty or daily_returns.std() == 0: + return 0.0 + return float(np.sqrt(365.25) * daily_returns.mean() / daily_returns.std()) + + +def _calmar(equity: pd.Series) -> float: + nonzero = equity[equity > 0] + series = nonzero if len(nonzero) > 1 else equity + c = _cagr(series) + mdd = abs(_max_drawdown(series)) + return c / mdd if mdd > 0.001 else 0.0 + + +def compute_all_metrics( + equity_curves: dict[str, pd.Series], +) -> dict[str, dict[str, dict[str, float]]]: + """Compute per-period metrics for each strategy.""" + results: dict[str, dict[str, dict[str, float]]] = {} + + for name, equity in equity_curves.items(): + results[name] = {} + for period_name, (start, end) in PERIODS.items(): + sub = equity.loc[start:end] + if sub.empty or len(sub) < 2: + results[name][period_name] = { + "cagr": 0.0, "max_dd": 0.0, "sharpe": 0.0, + "calmar": 0.0, "total_return": 0.0, "final_value": 0.0, + } + continue + + nonzero_sub = sub[sub > 0] + if len(nonzero_sub) >= 2: + total_ret = nonzero_sub.iloc[-1] / nonzero_sub.iloc[0] - 1.0 + elif len(sub) >= 2 and sub.iloc[0] > 0: + total_ret = sub.iloc[-1] / sub.iloc[0] - 1.0 + else: + total_ret = 0.0 + # Use nonzero series for CAGR calculation + nonzero = sub[sub > 0] + cagr_val = _cagr(nonzero) if len(nonzero) > 1 else _cagr(sub) + mdd_val = _max_drawdown(nonzero) if len(nonzero) > 1 else _max_drawdown(sub) + shrp_val = _sharpe(nonzero) if len(nonzero) > 1 else _sharpe(sub) + cal_val = cagr_val / abs(mdd_val) if abs(mdd_val) > 0.001 else 0.0 + results[name][period_name] = { + "cagr": round(cagr_val, 4), + "max_dd": round(mdd_val, 4), + "sharpe": round(shrp_val, 4), + "calmar": round(cal_val, 4), + "total_return": round(float(total_ret), 4), + "final_value": round(float(sub.iloc[-1]), 2), + } + + return results + + +# --------------------------------------------------------------------------- +# Reporting +# --------------------------------------------------------------------------- + + +def print_report( + equity_curves: dict[str, pd.Series], + metrics: dict[str, dict[str, dict[str, float]]], +) -> None: + """Print formatted report.""" + strategy_names = list(equity_curves.keys()) + period_names = list(PERIODS.keys()) + + print() + print("=" * 90) + print(" BTC DCA STRATEGY BACKTEST (2021-2026)") + print("=" * 90) + print(f" Monthly investment: ${DCA_AMOUNT_USD:.0f} on day {MONTHLY_DAY} ± {MONTHLY_WINDOW}d") + print(f" AHR999 thresholds: bottom≤{AHR999_BOTTOM} accum≤{AHR999_ACCUMULATION} dca≤{AHR999_DCA}") + print(f" Smart multipliers: bottom={AHR999_BOTTOM_MULT}x accum={AHR999_ACCUMULATION_MULT}x " + f"dca={AHR999_DCA_MULT}x severe={SEVERE_MULT}x") + print() + print(f" Note: CAGR/TotalRet computed from first non-zero equity (first DCA buy).") + print() + + # Full period summary + print(" Full Period Summary:") + print(f" {'Strategy':<28s} {'CAGR':>8s} {'MDD':>8s} {'Sharpe':>8s} {'Calmar':>8s} {'TotRet':>9s} {'Final':>10s}") + print(" " + "-" * 79) + for name in strategy_names: + m = metrics[name]["Full Period"] + print( + f" {name:<28s} {m['cagr']:>7.1%} {m['max_dd']:>7.1%} " + f"{m['sharpe']:>7.2f} {m['calmar']:>7.2f} " + f"{m['total_return']:>8.1%} ${m['final_value']:>9,.0f}" + ) + + # Period breakdowns + for period in period_names[1:]: + print(f"\n {period}:") + print(f" {'Strategy':<28s} {'CAGR':>8s} {'MDD':>8s} {'Sharpe':>8s} {'Calmar':>8s} {'TotRet':>9s}") + print(" " + "-" * 69) + for name in strategy_names: + m = metrics[name][period] + print( + f" {name:<28s} {m['cagr']:>7.1%} {m['max_dd']:>7.1%} " + f"{m['sharpe']:>7.2f} {m['calmar']:>7.2f} " + f"{m['total_return']:>8.1%}" + ) + + # Winner analysis + print() + print(" Best Strategy Per Period:") + for period in period_names: + best = max(strategy_names, key=lambda n: metrics[n][period]["calmar"]) + print(f" {period:<20s} → {best} (Calmar: {metrics[best][period]['calmar']:.2f})") + + print() + + +def print_regime_distribution(indicators: pd.DataFrame) -> None: + """Print AHR999 regime distribution over the backtest period.""" + warm_up = 252 + sub = indicators.iloc[warm_up:] + if sub.empty: + return + + regimes: dict[str, int] = {"bottom (≤0.45)": 0, "accumulation (0.45-0.80)": 0, + "dca (0.80-1.20)": 0, "expensive (>1.20)": 0, "no_data": 0} + for _, row in sub.iterrows(): + ahr999 = row["ahr999"] + if pd.isna(ahr999): + regimes["no_data"] += 1 + elif ahr999 <= 0.45: + regimes["bottom (≤0.45)"] += 1 + elif ahr999 <= 0.80: + regimes["accumulation (0.45-0.80)"] += 1 + elif ahr999 <= 1.20: + regimes["dca (0.80-1.20)"] += 1 + else: + regimes["expensive (>1.20)"] += 1 + + total = sum(regimes.values()) or 1 + print(" AHR999 Regime Distribution:") + for regime, count in regimes.items(): + pct = count / total * 100 + bar = "█" * int(pct / 2) + print(f" {regime:<25s}: {count:>5d} ({pct:5.1f}%) {bar}") + print() + + +# --------------------------------------------------------------------------- +# Entrypoint +# --------------------------------------------------------------------------- + + +def main() -> None: + parser = argparse.ArgumentParser(description="BTC DCA strategy backtest") + parser.add_argument("--json", action="store_true", help="Output JSON") + args = parser.parse_args() + + print("Loading BTC price data ...", file=sys.stderr) + btc_close = load_btc_data() + print(f" Loaded {len(btc_close)} days: {btc_close.index[0].date()} → {btc_close.index[-1].date()}", + file=sys.stderr) + + print("Computing indicators ...", file=sys.stderr) + indicators = compute_indicators(btc_close) + print(f" Indicator range: {indicators.index[0].date()} → {indicators.index[-1].date()}", + file=sys.stderr) + + print("Running strategies ...", file=sys.stderr) + equity_curves = run_strategies(btc_close, indicators) + metrics = compute_all_metrics(equity_curves) + + if args.json: + import json + output = { + name: { + period: {k: float(v) for k, v in pd.items()} + for period, pd in periods.items() + } + for name, periods in metrics.items() + } + json.dump(output, sys.stdout, indent=2) + else: + print_report(equity_curves, metrics) + print_regime_distribution(indicators) + + +if __name__ == "__main__": + main() diff --git a/src/crypto_strategies/_utils.py b/src/crypto_strategies/_utils.py new file mode 100644 index 0000000..fa2d84f --- /dev/null +++ b/src/crypto_strategies/_utils.py @@ -0,0 +1,93 @@ +"""Shared utilities for crypto strategies (no cross-package dependency).""" + +from __future__ import annotations + +import logging +from typing import Any + +import pandas as pd + +logger = logging.getLogger(__name__) + + +def coerce_float(value: Any, default: float = 0.0) -> float: + try: + numeric = float(value) + except (TypeError, ValueError): + return default + if pd.isna(numeric): + return default + return numeric + + +def coerce_bool(value: Any, default: bool = False) -> bool: + if value is None: + return default + if isinstance(value, bool): + return value + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in {"1", "true", "yes", "y", "on"}: + return True + if normalized in {"0", "false", "no", "n", "off"}: + return False + return bool(value) + + +def normalize_symbol(symbol: object) -> str: + return str(symbol or "").strip().upper() + + +def as_clamped_ratio(value: Any, *, default: float) -> float: + numeric = coerce_float(value, default=float("nan")) + if pd.isna(numeric): + return float(default) + return max(0.0, min(1.0, float(numeric))) + + +def payload_numeric(payload: dict[str, Any], *keys: str) -> float: + lowered = {str(key).strip().lower(): value for key, value in payload.items()} + for key in keys: + value = lowered.get(key.lower()) + numeric = coerce_float(value, default=float("nan")) + if not pd.isna(numeric): + return numeric + return float("nan") + + +# --------------------------------------------------------------------------- +# i18n +# --------------------------------------------------------------------------- + + +def translate_with_fallback( + translator, + key: str, + *, + fallback_en: str, + fallback_zh: str, + **kwargs: object, +) -> str: + if translator is None: + template = fallback_zh + else: + try: + translated = translator(key, **kwargs) + except Exception: + translated = key + if translated == key: + template = fallback_zh if translator_uses_zh(translator) else fallback_en + else: + return str(translated) + try: + return template.format(**{str(k): v for k, v in kwargs.items()}) + except (KeyError, ValueError): + return template + + +def translator_uses_zh(translator) -> bool: + try: + sample = str(translator("no_trades")) + except Exception: + return False + return any("一" <= char <= "鿿" for char in sample) diff --git a/src/crypto_strategies/catalog.py b/src/crypto_strategies/catalog.py index 5086126..8bcf004 100644 --- a/src/crypto_strategies/catalog.py +++ b/src/crypto_strategies/catalog.py @@ -71,6 +71,51 @@ } CRYPTO_BTC_DCA_DEFAULT_CONFIG = { + "base_investment_usd": 100.0, + "max_investment_usd": None, + "cash_reserve_usd": 0.0, + "min_investment_usd": 5.0, + "smart_multiplier_enabled": True, + "cycle_indicator_enabled": True, + "cadence": "monthly", + "monthly_day": 25, + "monthly_window_calendar_days": 5, + "weekly_day": 4, + "weekly_window_calendar_days": 4, + "quarterly_months": (1, 4, 7, 10), + "quarterly_day": 25, + "quarterly_window_calendar_days": 5, + # Drawdown thresholds + "mild_drawdown_threshold": 0.12, + "deep_drawdown_threshold": 0.25, + "severe_drawdown_threshold": 0.40, + "mild_discount_gap": 0.08, + "deep_discount_gap": 0.18, + "expensive_gap": 0.30, + "very_expensive_gap": 0.60, + "shallow_drawdown_threshold": 0.05, + "overbought_rsi": 75.0, + "base_multiplier": 1.0, + "mild_pullback_multiplier": 1.50, + "deep_pullback_multiplier": 2.25, + "severe_pullback_multiplier": 3.0, + "expensive_multiplier": 1.0, + "very_expensive_multiplier": 1.0, + # AHR999 thresholds + "ahr999_bottom_threshold": 0.45, + "ahr999_accumulation_threshold": 0.80, + "ahr999_dca_threshold": 1.20, + "ahr999_bottom_multiplier": 3.0, + "ahr999_accumulation_multiplier": 2.25, + "ahr999_dca_multiplier": 1.50, + "ahr999_expensive_multiplier": 0.0, + # Z-score exit + "zscore_exit_enabled": True, + "zscore_exit_parking_symbol": "USDT", + "zscore_exit_risk_reduced_exposure": 0.50, + "zscore_exit_risk_off_exposure": 0.25, + "zscore_exit_allow_outside_execution_window": True, + # Legacy params (kept for backward compat) "target_ratio_min": 0.0, "target_ratio_max": 0.65, "ratio_base": 0.14, @@ -87,12 +132,20 @@ "weight_mode": "inverse_vol", "allow_rotation_refresh": True, "atr_multiplier": 2.5, + "circuit_breaker_enabled": True, + "btc_drawdown_threshold": 0.30, + "vol_scaling_enabled": True, + "target_vol": 0.40, + "max_leverage": 1.0, } CRYPTO_EQUITY_COMBO_DEFAULT_CONFIG = { "btc_weight": 0.30, "trend_weight": 0.70, "dynamic_mode": True, + "circuit_breaker_enabled": True, + "btc_drawdown_threshold": 0.30, + "vol_scaling_enabled": True, } STRATEGY_DEFINITIONS: dict[str, StrategyDefinition] = { @@ -191,7 +244,7 @@ display_name="Crypto BTC DCA", description="Dynamic BTC DCA strategy that targets a single BTCUSDT position with equity-scaled allocation.", aliases=(), - cadence="daily", + cadence="daily_check_monthly_execution", asset_scope="btc_only", benchmark="BTC", role="crypto_core_accumulation", diff --git a/src/crypto_strategies/entrypoints/__init__.py b/src/crypto_strategies/entrypoints/__init__.py index b85778a..7189c08 100644 --- a/src/crypto_strategies/entrypoints/__init__.py +++ b/src/crypto_strategies/entrypoints/__init__.py @@ -274,19 +274,27 @@ def evaluate_crypto_live_pool_rotation(ctx: StrategyContext) -> StrategyDecision def evaluate_crypto_btc_dca(ctx: StrategyContext) -> StrategyDecision: from crypto_strategies.strategies.crypto_btc_dca import compute_signals + config = _merge_runtime_config(ctx, crypto_btc_dca_manifest.default_config) prices = _require_market_data(ctx, "market_prices") portfolio = _resolve_portfolio_snapshot(ctx) account_metrics = _resolve_account_metrics(ctx) total_equity = account_metrics["total_equity"] + derived_indicators = ctx.market_data.get("derived_indicators") + translator = _resolve_translator(config) result = compute_signals( prices=prices, portfolio=portfolio, total_equity=total_equity, state=dict(ctx.state), + derived_indicators=derived_indicators, + translator=translator, + **config, ) btc_target_ratio = float(result.get("btc_target_ratio", 0.0)) + metadata = result.get("metadata", {}) if isinstance(result.get("metadata"), dict) else {} + positions = [ PositionTarget( symbol="BTCUSDT", @@ -307,11 +315,23 @@ def evaluate_crypto_btc_dca(ctx: StrategyContext) -> StrategyDecision: risk_flags: tuple[str, ...] = () if btc_target_ratio <= 0.0: risk_flags += ("no_btc_allocation",) + if not metadata.get("actionable", True): + risk_flags += ("no_execute",) diagnostics = { "btc_target_ratio": btc_target_ratio, "total_equity": total_equity, "profile": result.get("profile"), + "signal_description": metadata.get("signal_description", ""), + "status_description": metadata.get("status_description", ""), + "regime": metadata.get("regime", "ordinary_dca"), + "multiplier": metadata.get("multiplier", 1.0), + "smart_multiplier_enabled": metadata.get("smart_multiplier_enabled", True), + "planned_investment_usd": metadata.get("planned_investment_usd", 0.0), + "in_execution_window": metadata.get("in_execution_window", True), + "zscore_exit": metadata.get("zscore_exit", {}), + "ahr999": metadata.get("ahr999", float("nan")), + "mayer_multiple": metadata.get("mayer_multiple", float("nan")), } return StrategyDecision( @@ -338,6 +358,7 @@ def evaluate_crypto_trend_rotation(ctx: StrategyContext) -> StrategyDecision: config = _merge_runtime_config(ctx, crypto_trend_rotation_manifest.default_config) feature_snapshot = _require_market_data(ctx, "derived_indicators") prices = _require_market_data(ctx, "market_prices") + translator = _resolve_translator(config) # Build a feature_snapshot from indicators_map import pandas as pd @@ -351,11 +372,15 @@ def evaluate_crypto_trend_rotation(ctx: StrategyContext) -> StrategyDecision: weights, signal_desc, is_emergency, debug_str, metadata = compute_signals( feature_snapshot=feature_frame, current_holdings=list(prices.keys()), + translator=translator, trend_pool_size=int(config.get("trend_pool_size", 5)), rotation_top_n=int(config.get("rotation_top_n", 2)), weight_mode=str(config.get("weight_mode", "inverse_vol")), allow_rotation_refresh=bool(config.get("allow_rotation_refresh", True)), atr_multiplier=float(config.get("atr_multiplier", 2.5)), + circuit_breaker_enabled=bool(config.get("circuit_breaker_enabled", True)), + btc_drawdown_threshold=float(config.get("btc_drawdown_threshold", 0.30)), + vol_scaling_enabled=bool(config.get("vol_scaling_enabled", True)), ) positions: list[PositionTarget] = [] @@ -418,6 +443,7 @@ def evaluate_crypto_equity_combo(ctx: StrategyContext) -> StrategyDecision: benchmark_snapshot = _require_market_data(ctx, "benchmark_snapshot") portfolio = _resolve_portfolio_snapshot(ctx) universe_snapshot = list(_require_market_data(ctx, "universe_snapshot")) + translator = _resolve_translator(config) weights, signal_desc, has_cash_residual, status_desc, metadata = compute_signals( prices=prices, @@ -426,9 +452,15 @@ def evaluate_crypto_equity_combo(ctx: StrategyContext) -> StrategyDecision: benchmark_snapshot=benchmark_snapshot, portfolio=portfolio, state=dict(ctx.state), + translator=translator, btc_weight=float(config.get("btc_weight", 0.30)), trend_weight=float(config.get("trend_weight", 0.70)), dynamic_mode=bool(config.get("dynamic_mode", True)), + smart_multiplier_enabled=bool(config.get("smart_multiplier_enabled", True)), + cycle_indicator_enabled=bool(config.get("cycle_indicator_enabled", True)), + zscore_exit_enabled=bool(config.get("zscore_exit_enabled", True)), + circuit_breaker_enabled=bool(config.get("circuit_breaker_enabled", True)), + vol_scaling_enabled=bool(config.get("vol_scaling_enabled", True)), ) positions: list[PositionTarget] = [] diff --git a/src/crypto_strategies/strategies/crypto_btc_dca.py b/src/crypto_strategies/strategies/crypto_btc_dca.py index 7a3dd3b..07db034 100644 --- a/src/crypto_strategies/strategies/crypto_btc_dca.py +++ b/src/crypto_strategies/strategies/crypto_btc_dca.py @@ -1,24 +1,198 @@ -"""BTC DCA standalone strategy. +"""BTC DCA standalone strategy with smart sizing and cycle-aware exit. Profile: crypto_btc_dca Domain: crypto -Source: market_prices +Source: derived_indicators + portfolio_snapshot -This is a pure DCA strategy targeting only BTCUSDT with a dynamic target -ratio based on total portfolio equity. It produces a single PositionTarget -and BudgetIntent per evaluation. +This is a smart DCA strategy targeting BTCUSDT. It supports: + +- **Smart multiplier**: AHR999 cycle indicator (priority) or drawdown-based sizing. + When enabled, the base DCA amount is scaled by a regime-dependent multiplier + (0.0x → skip, 1.0x → normal, up to 3.0x → aggressive). +- **Z-score exit (逃顶)**: MVRV Z-Score driven position control that reduces BTC + exposure to 50 % (risk_reduced) or 25 % (risk_off), parking freed capital in + USDT. +- **Execution window**: Monthly (day 25 ± 5 days), weekly (Thursday ± 4 days), + or quarterly with configurable windows. +- **i18n**: Chinese / English bilingual diagnostics. + +Strategy Contract +----------------- +``compute_signals`` returns a dict compatible with the combo strategy: + + { + "signals": {"BTCUSDT": {"target_weight": 1.0, "btc_target_ratio": ...}}, + "total_equity": ..., + "btc_target_ratio": ..., + "profile": "crypto_btc_dca", + "metadata": { ... full diagnostics ... }, + } + +``build_target_weights`` returns {"BTCUSDT": 1.0} — the ratio is applied +externally by the entrypoint / combo. """ from __future__ import annotations +import logging import math +from collections.abc import Mapping, Sequence from typing import Any -CN_EQUITY_DOMAIN: str = "crypto" -SIGNAL_SOURCE: str = "market_prices" -STATUS_ICON: str = "\U0001f351" # BTC emoji (peach) +import pandas as pd + +from crypto_strategies._utils import ( + as_clamped_ratio, + coerce_bool, + coerce_float, + normalize_symbol, + payload_numeric, + translate_with_fallback, +) + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +CRYPTO_DOMAIN: str = "crypto" +SIGNAL_SOURCE: str = "derived_indicators+portfolio_snapshot" +STATUS_ICON: str = "₿" PROFILE_NAME: str = "crypto_btc_dca" +DEFAULT_SIGNAL_SYMBOL = "BTCUSDT" +DEFAULT_PARKING_SYMBOL = "USDT" +BITCOIN_GENESIS_DATE = pd.Timestamp("2009-01-03") + +ZSCORE_EXIT_PROFILE = "btc_zscore_exit" +ZSCORE_EXIT_POSITION_ROUTES = frozenset({"normal", "risk_on", "risk_reduced", "risk_off"}) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _as_timestamp(value: object) -> pd.Timestamp: + if value is None: + return pd.Timestamp.now(tz="UTC").tz_localize(None).normalize() + timestamp = pd.Timestamp(value) + if timestamp.tzinfo is not None: + timestamp = timestamp.tz_convert("UTC").tz_localize(None) + return timestamp.normalize() + + +def _localized_regime(regime: str, translator) -> str: + labels: dict[str, tuple[str, str]] = { + "ordinary_dca": ("ordinary DCA", "普通定投"), + "normal": ("normal", "正常"), + "expensive": ("expensive", "偏贵"), + "very_expensive_overbought": ("very expensive and overbought", "极贵且超买"), + "mild_pullback": ("mild pullback", "温和回撤"), + "deep_pullback": ("deep pullback", "深度回撤"), + "severe_pullback": ("severe pullback", "严重回撤"), + "ahr999_bottom": ("AHR999 bottom zone", "AHR999 底部区"), + "ahr999_accumulation": ("AHR999 accumulation zone", "AHR999 囤币区"), + "ahr999_dca": ("AHR999 DCA zone", "AHR999 定投区"), + "ahr999_expensive": ("AHR999 expensive zone", "AHR999 偏贵区"), + } + fallback_en, fallback_zh = labels.get(regime, (regime, regime)) + return translate_with_fallback( + translator, + f"btc_dca_regime_{regime}", + fallback_en=fallback_en, + fallback_zh=fallback_zh, + ) + + +def _localized_skip_reason(skip_reason: str, translator) -> str: + labels: dict[str, tuple[str, str]] = { + "outside_execution_window": ("outside execution window", "不在执行窗口"), + "valuation_too_expensive": ("valuation too expensive", "估值过贵"), + "insufficient_cash": ("insufficient cash", "可投资现金不足"), + } + fallback_en, fallback_zh = labels.get(skip_reason, (skip_reason, skip_reason)) + return translate_with_fallback( + translator, + f"btc_dca_skip_{skip_reason}", + fallback_en=fallback_en, + fallback_zh=fallback_zh, + ) + + +# --------------------------------------------------------------------------- +# Execution window +# --------------------------------------------------------------------------- + + +def _is_in_execution_window( + as_of: object, + *, + cadence: str, + monthly_day: int, + monthly_window_calendar_days: int, + weekly_day: int, + weekly_window_calendar_days: int, + quarterly_months: object, + quarterly_day: int, + quarterly_window_calendar_days: int, +) -> tuple[bool, str]: + timestamp = _as_timestamp(as_of) + cadence_key = str(cadence or "monthly").strip().lower() + + if cadence_key == "weekly": + day = int(max(0, min(6, weekly_day))) + window = int(max(1, min(7, weekly_window_calendar_days))) + days_since_start = (int(timestamp.weekday()) - day) % 7 + return days_since_start < window, f"weekly_day={day} window={window}d" + + if cadence_key == "quarterly": + months = _normalize_quarterly_months(quarterly_months) + start_day = int(max(1, min(31, quarterly_day))) + window = int(max(1, quarterly_window_calendar_days)) + months_text = ",".join(str(m) for m in months) + return ( + timestamp.month in months and start_day <= int(timestamp.day) < start_day + window, + f"quarterly_months={months_text} day={start_day} window={window}d", + ) + + if cadence_key != "monthly": + raise ValueError("cadence must be 'monthly', 'weekly', or 'quarterly'") + + start_day = int(max(1, min(31, monthly_day))) + window = int(max(1, monthly_window_calendar_days)) + return ( + start_day <= int(timestamp.day) < start_day + window, + f"monthly_day={start_day} window={window}d", + ) + + +def _normalize_quarterly_months(raw_months: object) -> tuple[int, ...]: + if raw_months is None: + candidates: object = (1, 4, 7, 10) + elif isinstance(raw_months, str): + candidates = raw_months.replace(";", ",").split(",") + else: + candidates = raw_months + months: list[int] = [] + try: + iterator = iter(candidates) # type: ignore[arg-type] + except TypeError: + iterator = iter((candidates,)) + for item in iterator: + try: + month = int(item) + except (TypeError, ValueError): + continue + if 1 <= month <= 12 and month not in months: + months.append(month) + return tuple(months) or (1, 4, 7, 10) + + +# --------------------------------------------------------------------------- +# Dynamic BTC target ratio (from crypto_live_pool_rotation.core) +# --------------------------------------------------------------------------- + def get_dynamic_btc_target_ratio(total_equity: float) -> float: """Compute the target BTC allocation ratio based on total equity. @@ -31,6 +205,688 @@ def get_dynamic_btc_target_ratio(total_equity: float) -> float: return min(0.65, max(0.0, ratio)) +# --------------------------------------------------------------------------- +# Indicator extraction +# --------------------------------------------------------------------------- + + +def _resolve_indicator_payload( + indicator_snapshot: Mapping[str, object] | None, + symbol: str, +) -> Mapping[str, object] | None: + if not isinstance(indicator_snapshot, Mapping): + return None + # Direct match: snapshot itself contains indicator keys + if any( + str(key).lower() in {"ahr999", "ahr_999", "ahr999_gma", "close", "price"} + for key in indicator_snapshot + ): + return indicator_snapshot + + candidates = { + symbol, symbol.upper(), + symbol.replace("-", ""), symbol.replace("-", "").upper(), + "BTC", "BTC-USD", "BTCUSDT", + } + for key in candidates: + value = indicator_snapshot.get(key) + if isinstance(value, Mapping): + return value + normalized_snapshot = { + str(key).strip().upper().replace("-", ""): value + for key, value in indicator_snapshot.items() + } + for key in candidates: + value = normalized_snapshot.get(key.upper().replace("-", "")) + if isinstance(value, Mapping): + return value + return None + + +def _indicator_from_payload(symbol: str, payload: Mapping[str, object]) -> dict[str, float | None]: + price = payload_numeric(payload, "close", "price", "last", "last_price") + sma200 = payload_numeric(payload, "sma200", "ma200", "sma_200") + high252 = payload_numeric(payload, "high252", "high_252", "high252d", "high_252d") + if pd.isna(price) or pd.isna(sma200): + return {} + if pd.isna(high252): + high252 = max(price, sma200) + drawdown = payload_numeric(payload, "drawdown_252d", "drawdown252", "drawdown") + if pd.isna(drawdown): + drawdown = 0.0 if high252 <= 0.0 else max(0.0, 1.0 - price / high252) + sma_gap = payload_numeric(payload, "sma200_gap", "gap_vs_sma200", "price_vs_sma200") + if pd.isna(sma_gap) and sma200 > 0.0: + sma_gap = price / sma200 - 1.0 + rsi14 = payload_numeric(payload, "rsi14", "rsi_14", "rsi") + return { + "price": float(price), + "sma200": float(sma200), + "high252": float(high252), + "drawdown_252d": float(drawdown if not pd.isna(drawdown) else 0.0), + "sma200_gap": float(sma_gap if not pd.isna(sma_gap) else 0.0), + "rsi14": None if pd.isna(rsi14) else float(rsi14), + } + + +# --------------------------------------------------------------------------- +# Cycle metrics (AHR999, Mayer Multiple) +# --------------------------------------------------------------------------- + + +def _bitcoin_age_estimate_price(as_of: object) -> float: + timestamp = pd.Timestamp(as_of) if as_of is not None else pd.Timestamp.now(tz="UTC") + if timestamp.tzinfo is not None: + timestamp = timestamp.tz_convert("UTC").tz_localize(None) + age_days = max(1, int((timestamp.normalize() - BITCOIN_GENESIS_DATE).days)) + return float(10 ** (5.84 * math.log10(age_days) - 17.01)) + + +def _cycle_metrics_from_payload( + payload: Mapping[str, object] | None, + *, + as_of: object, +) -> dict[str, float | str]: + if not isinstance(payload, Mapping): + return {} + ahr999_gma = payload_numeric(payload, "ahr999_gma", "ahr999", "ahr_999", "ahr999_index") + ahr999_sma = payload_numeric(payload, "ahr999_sma", "ahr999_sma200") + mayer = payload_numeric(payload, "mayer_multiple", "mayer", "price_sma200_ratio") + price = payload_numeric(payload, "close", "price", "last", "last_price") + sma200 = payload_numeric(payload, "sma200", "ma200", "sma_200") + estimate_price = payload_numeric(payload, "ahr999_estimate_price", "estimate_price") + if pd.isna(estimate_price): + estimate_price = _bitcoin_age_estimate_price(as_of) + if pd.isna(mayer) and not pd.isna(price) and not pd.isna(sma200) and sma200 > 0.0: + mayer = price / sma200 + if pd.isna(ahr999_sma) and not pd.isna(price) and not pd.isna(sma200) and sma200 > 0.0 and estimate_price > 0.0: + ahr999_sma = (price / sma200) * (price / estimate_price) + if pd.isna(ahr999_gma): + ahr999_gma = ahr999_sma + if pd.isna(ahr999_gma): + return {} + metrics: dict[str, float | str] = {"ahr999": float(ahr999_gma)} + if not pd.isna(ahr999_sma): + metrics["ahr999_sma"] = float(ahr999_sma) + if not pd.isna(mayer): + metrics["mayer_multiple"] = float(mayer) + if not pd.isna(estimate_price): + metrics["ahr999_estimate_price"] = float(estimate_price) + if metrics: + metrics["cycle_indicator_source"] = "derived_indicators" + return metrics + + +# --------------------------------------------------------------------------- +# Multiplier logic +# --------------------------------------------------------------------------- + + +def _determine_multiplier( + indicator: dict[str, float | None], + *, + mild_drawdown_threshold: float, + deep_drawdown_threshold: float, + severe_drawdown_threshold: float, + mild_discount_gap: float, + deep_discount_gap: float, + expensive_gap: float, + very_expensive_gap: float, + shallow_drawdown_threshold: float, + overbought_rsi: float, + mild_pullback_multiplier: float, + deep_pullback_multiplier: float, + severe_pullback_multiplier: float, + expensive_multiplier: float, + very_expensive_multiplier: float, + base_multiplier: float, +) -> tuple[float, str, dict[str, float]]: + drawdown = float(indicator.get("drawdown_252d", 0.0) or 0.0) + sma_gap = float(indicator.get("sma200_gap", 0.0) or 0.0) + rsi14 = indicator.get("rsi14") + rsi_value = float(rsi14) if rsi14 is not None else float("nan") + + metrics = { + "drawdown_252d": drawdown, + "sma200_gap": sma_gap, + "rsi14": rsi_value, + } + + if drawdown >= severe_drawdown_threshold: + return float(severe_pullback_multiplier), "severe_pullback", metrics + if drawdown >= deep_drawdown_threshold or sma_gap <= -abs(float(deep_discount_gap)): + return float(deep_pullback_multiplier), "deep_pullback", metrics + if drawdown >= mild_drawdown_threshold or sma_gap <= -abs(float(mild_discount_gap)): + return float(mild_pullback_multiplier), "mild_pullback", metrics + if sma_gap >= very_expensive_gap and drawdown <= shallow_drawdown_threshold and not pd.isna(rsi_value) and rsi_value >= overbought_rsi: + return float(very_expensive_multiplier), "very_expensive_overbought", metrics + if sma_gap >= expensive_gap and drawdown <= shallow_drawdown_threshold: + return float(expensive_multiplier), "expensive", metrics + return float(base_multiplier), "normal", metrics + + +def _determine_cycle_multiplier( + cycle_metrics: Mapping[str, object], + *, + ahr999_bottom_threshold: float, + ahr999_accumulation_threshold: float, + ahr999_dca_threshold: float, + ahr999_bottom_multiplier: float, + ahr999_accumulation_multiplier: float, + ahr999_dca_multiplier: float, + ahr999_expensive_multiplier: float, + base_multiplier: float, +) -> tuple[float, str]: + ahr999 = coerce_float(cycle_metrics.get("ahr999"), default=float("nan")) + if pd.isna(ahr999): + return float(base_multiplier), "normal" + if ahr999 <= float(ahr999_bottom_threshold): + return float(ahr999_bottom_multiplier), "ahr999_bottom" + if ahr999 <= float(ahr999_accumulation_threshold): + return float(ahr999_accumulation_multiplier), "ahr999_accumulation" + if ahr999 <= float(ahr999_dca_threshold): + return float(ahr999_dca_multiplier), "ahr999_dca" + return float(ahr999_expensive_multiplier), "ahr999_expensive" + + +# --------------------------------------------------------------------------- +# Z-score exit (逃顶) +# --------------------------------------------------------------------------- + + +def _find_zscore_exit_payload( + metadata: Mapping[str, object] | None, + explicit_context: Mapping[str, object] | None, +) -> tuple[Mapping[str, object] | None, str]: + sources: list[tuple[str, object, bool]] = [] + if isinstance(explicit_context, Mapping): + sources.append(("explicit", explicit_context, True)) + if isinstance(metadata, Mapping): + sources.append(("metadata.btc_zscore_exit", metadata.get(ZSCORE_EXIT_PROFILE), True)) + sources.append(("metadata.strategy_plugins", metadata.get("strategy_plugins"), False)) + + for source_name, source, _allow_pluginless in sources: + if not isinstance(source, (Mapping, Sequence)) or isinstance(source, (str, bytes, bytearray)): + continue + for payload in _iter_mapping_payloads(source): + plugin = str(payload.get("plugin") or payload.get("profile") or "").strip().lower() + if plugin == ZSCORE_EXIT_PROFILE: + return payload, source_name + if not _allow_pluginless and not plugin: + continue + if bool({"position_control", "target_allocations", "route", "canonical_route"} & set(payload)): + return payload, source_name + return None, "" + + +def _iter_mapping_payloads(value, *, _depth: int = 0): + if _depth > 4: + return + if isinstance(value, Mapping): + yield value + for item in value.values(): + yield from _iter_mapping_payloads(item, _depth=_depth + 1) + elif isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)): + for item in value: + yield from _iter_mapping_payloads(item, _depth=_depth + 1) + + +def _resolve_zscore_exit_context( + portfolio_metadata: Mapping[str, object] | None, + *, + enabled: object, + explicit_context: Mapping[str, object] | None, + parking_symbol: str, + risk_reduced_exposure: float, + risk_off_exposure: float, +) -> dict[str, object]: + config_enabled = coerce_bool(enabled, default=True) + + payload, source = _find_zscore_exit_payload(portfolio_metadata, explicit_context) + if not isinstance(payload, Mapping): + return { + "enabled": bool(config_enabled), + "found": False, "source": "", "active": False, "applied": False, + "route": "", "parking_symbol": parking_symbol, + "target_btc_exposure": 1.0, "target_parking_exposure": 0.0, + } + + position_control = payload.get("position_control") + if not isinstance(position_control, Mapping): + position_control = {} + route = str( + position_control.get("final_route") + or position_control.get("route") + or payload.get("canonical_route") + or payload.get("route") + or "" + ).strip().lower() + + target_allocations = ( + position_control.get("target_allocations") + or position_control.get("target_exposure") + or payload.get("target_allocations") + ) + if isinstance(target_allocations, Mapping): + target_allocations = dict(target_allocations) + else: + target_allocations = {} + + target_btc_exposure = as_clamped_ratio( + coerce_float(target_allocations.get("BTCUSDT"), default=float("nan")), + default=1.0, + ) + if pd.isna(coerce_float(target_allocations.get("BTCUSDT"), default=float("nan"))): + for raw_value in ( + position_control.get("target_btc_exposure"), + payload.get("target_btc_exposure"), + ): + numeric = coerce_float(raw_value, default=float("nan")) + if not pd.isna(numeric): + target_btc_exposure = as_clamped_ratio(numeric, default=1.0) + break + if route == "risk_reduced": + target_btc_exposure = as_clamped_ratio(risk_reduced_exposure, default=0.50) + elif route == "risk_off": + target_btc_exposure = as_clamped_ratio(risk_off_exposure, default=0.25) + + active = route in ZSCORE_EXIT_POSITION_ROUTES + + return { + "enabled": bool(config_enabled), + "found": True, "source": source, "active": bool(active), + "applied": bool(config_enabled and active), + "route": route, + "parking_symbol": parking_symbol, + "target_btc_exposure": float(target_btc_exposure), + "target_parking_exposure": float(1.0 - target_btc_exposure), + } + + +# --------------------------------------------------------------------------- +# Portfolio helpers +# --------------------------------------------------------------------------- + + +def _portfolio_metrics(snapshot) -> dict[str, float]: + """Extract key portfolio metrics from a snapshot object or dict.""" + if isinstance(snapshot, Mapping): + total_equity = float(snapshot.get("total_equity", 0.0) or 0.0) + buying_power = float(snapshot.get("buying_power", snapshot.get("cash_balance", 0.0)) or 0.0) + positions = snapshot.get("positions", ()) + else: + total_equity = float(getattr(snapshot, "total_equity", 0.0) or 0.0) + buying_power = float( + getattr(snapshot, "buying_power", getattr(snapshot, "cash_balance", 0.0)) or 0.0 + ) + positions = getattr(snapshot, "positions", ()) + + btc_value = 0.0 + for pos in positions or (): + if isinstance(pos, Mapping): + symbol = str(pos.get("symbol", "")).strip().upper() + market_value = float(pos.get("market_value", 0.0) or 0.0) + else: + symbol = str(getattr(pos, "symbol", "")).strip().upper() + market_value = float(getattr(pos, "market_value", 0.0) or 0.0) + if symbol == "BTCUSDT": + btc_value += market_value + + if isinstance(snapshot, Mapping): + metadata = snapshot.get("metadata", {}) or {} + else: + metadata = getattr(snapshot, "metadata", {}) or {} + if isinstance(metadata, Mapping): + btc_value = float(metadata.get("dca_value", btc_value)) + + return { + "total_equity": total_equity, + "buying_power": max(0.0, buying_power), + "btc_value": max(0.0, btc_value), + } + + +# --------------------------------------------------------------------------- +# Core: build_rebalance_plan +# --------------------------------------------------------------------------- + + +def build_rebalance_plan( + portfolio, + *, + as_of=None, + signal_symbol: str = DEFAULT_SIGNAL_SYMBOL, + base_investment_usd: float = 100.0, + max_investment_usd: float | None = None, + cash_reserve_usd: float = 0.0, + min_investment_usd: float = 5.0, + smart_multiplier_enabled: bool = True, + cycle_indicator_enabled: bool = True, + cadence: str = "monthly", + monthly_day: int = 25, + monthly_window_calendar_days: int = 5, + weekly_day: int = 4, + weekly_window_calendar_days: int = 4, + quarterly_months: object = (1, 4, 7, 10), + quarterly_day: int = 25, + quarterly_window_calendar_days: int = 5, + # Drawdown thresholds + mild_drawdown_threshold: float = 0.12, + deep_drawdown_threshold: float = 0.25, + severe_drawdown_threshold: float = 0.40, + mild_discount_gap: float = 0.08, + deep_discount_gap: float = 0.18, + expensive_gap: float = 0.30, + very_expensive_gap: float = 0.60, + shallow_drawdown_threshold: float = 0.05, + overbought_rsi: float = 75.0, + base_multiplier: float = 1.0, + mild_pullback_multiplier: float = 1.50, + deep_pullback_multiplier: float = 2.25, + severe_pullback_multiplier: float = 3.0, + expensive_multiplier: float = 1.0, + very_expensive_multiplier: float = 1.0, + # AHR999 thresholds + ahr999_bottom_threshold: float = 0.45, + ahr999_accumulation_threshold: float = 0.80, + ahr999_dca_threshold: float = 1.20, + ahr999_bottom_multiplier: float = 3.0, + ahr999_accumulation_multiplier: float = 2.25, + ahr999_dca_multiplier: float = 1.50, + ahr999_expensive_multiplier: float = 0.0, + # Z-score exit + zscore_exit_enabled: bool = True, + zscore_exit_parking_symbol: str = DEFAULT_PARKING_SYMBOL, + zscore_exit_risk_reduced_exposure: float = 0.50, + zscore_exit_risk_off_exposure: float = 0.25, + zscore_exit_allow_outside_execution_window: bool = True, + # External data + derived_indicators: Mapping[str, object] | None = None, + zscore_exit_context: Mapping[str, object] | None = None, + # Extension + translator=None, +) -> dict[str, object]: + """Produce a full rebalance plan for the BTC DCA strategy.""" + symbol = normalize_symbol(signal_symbol) + if not symbol: + symbol = DEFAULT_SIGNAL_SYMBOL + + # --- Execution window --- + is_window, window_text = _is_in_execution_window( + as_of, + cadence=cadence, + monthly_day=monthly_day, + monthly_window_calendar_days=monthly_window_calendar_days, + weekly_day=weekly_day, + weekly_window_calendar_days=weekly_window_calendar_days, + quarterly_months=quarterly_months, + quarterly_day=quarterly_day, + quarterly_window_calendar_days=quarterly_window_calendar_days, + ) + + # --- Portfolio state --- + pm = _portfolio_metrics(portfolio) + total_equity = pm["total_equity"] + buying_power = pm["buying_power"] + btc_value = pm["btc_value"] + reserved_cash = max(0.0, coerce_float(cash_reserve_usd)) + investable_cash = max(0.0, buying_power - reserved_cash) + + # --- Portfolio metadata (for z-score exit lookup) --- + if isinstance(portfolio, Mapping): + portfolio_metadata = portfolio.get("metadata", {}) or {} + else: + portfolio_metadata = getattr(portfolio, "metadata", {}) or {} + + # --- Z-score exit --- + zscore_exit = _resolve_zscore_exit_context( + portfolio_metadata, + enabled=zscore_exit_enabled, + explicit_context=zscore_exit_context, + parking_symbol=normalize_symbol(zscore_exit_parking_symbol) or DEFAULT_PARKING_SYMBOL, + risk_reduced_exposure=float(zscore_exit_risk_reduced_exposure), + risk_off_exposure=float(zscore_exit_risk_off_exposure), + ) + + # --- Smart multiplier --- + smart_enabled = coerce_bool(smart_multiplier_enabled, default=True) + cycle_metrics: dict[str, float | str] = {} + indicator: dict[str, float | None] = {} + regime = "ordinary_dca" + multiplier = 1.0 + aggregate_metrics: dict[str, float] = { + "drawdown_252d": float("nan"), + "sma200_gap": float("nan"), + "rsi14": float("nan"), + } + + if smart_enabled: + payload = _resolve_indicator_payload(derived_indicators, symbol) + indicator = _indicator_from_payload(symbol, payload) if payload else {} + + if coerce_bool(cycle_indicator_enabled, default=True): + cycle_metrics = _cycle_metrics_from_payload(payload, as_of=as_of) + + if cycle_metrics and not pd.isna(coerce_float(cycle_metrics.get("ahr999"), default=float("nan"))): + multiplier, regime = _determine_cycle_multiplier( + cycle_metrics, + ahr999_bottom_threshold=float(ahr999_bottom_threshold), + ahr999_accumulation_threshold=float(ahr999_accumulation_threshold), + ahr999_dca_threshold=float(ahr999_dca_threshold), + ahr999_bottom_multiplier=float(ahr999_bottom_multiplier), + ahr999_accumulation_multiplier=float(ahr999_accumulation_multiplier), + ahr999_dca_multiplier=float(ahr999_dca_multiplier), + ahr999_expensive_multiplier=float(ahr999_expensive_multiplier), + base_multiplier=float(base_multiplier), + ) + if indicator: + aggregate_metrics.update({ + "drawdown_252d": float(indicator.get("drawdown_252d", 0.0) or 0.0), + "sma200_gap": float(indicator.get("sma200_gap", 0.0) or 0.0), + "rsi14": float(indicator.get("rsi14") or float("nan")), + }) + elif indicator: + multiplier, regime, aggregate_metrics = _determine_multiplier( + indicator, + mild_drawdown_threshold=float(mild_drawdown_threshold), + deep_drawdown_threshold=float(deep_drawdown_threshold), + severe_drawdown_threshold=float(severe_drawdown_threshold), + mild_discount_gap=float(mild_discount_gap), + deep_discount_gap=float(deep_discount_gap), + expensive_gap=float(expensive_gap), + very_expensive_gap=float(very_expensive_gap), + shallow_drawdown_threshold=float(shallow_drawdown_threshold), + overbought_rsi=float(overbought_rsi), + mild_pullback_multiplier=float(mild_pullback_multiplier), + deep_pullback_multiplier=float(deep_pullback_multiplier), + severe_pullback_multiplier=float(severe_pullback_multiplier), + expensive_multiplier=float(expensive_multiplier), + very_expensive_multiplier=float(very_expensive_multiplier), + base_multiplier=float(base_multiplier), + ) + # else: no indicator data available, use base_multiplier + + # --- Update aggregate metrics with cycle data --- + if cycle_metrics: + aggregate_metrics.update({ + "ahr999": float(coerce_float(cycle_metrics.get("ahr999"), default=float("nan"))), + "ahr999_sma": float(coerce_float(cycle_metrics.get("ahr999_sma"), default=float("nan"))), + "mayer_multiple": float(coerce_float(cycle_metrics.get("mayer_multiple"), default=float("nan"))), + "cycle_indicator_source": str(cycle_metrics.get("cycle_indicator_source", "none")), + }) + + # --- Calculate investment amount --- + regime_multiplier = float(multiplier if smart_enabled else 1.0) + requested_investment = max(0.0, float(base_investment_usd) * max(0.0, regime_multiplier)) + if max_investment_usd is not None: + requested_investment = min(requested_investment, max(0.0, float(max_investment_usd))) + planned_investment = min(requested_investment, investable_cash) + cash_capped = investable_cash < requested_investment + + # --- Actionability --- + skip_reason = None + actionable = True + if not is_window: + skip_reason = "outside_execution_window" + actionable = False + elif regime_multiplier <= 0.0 or requested_investment <= 0.0: + skip_reason = "valuation_too_expensive" + actionable = False + elif planned_investment < float(min_investment_usd): + skip_reason = "insufficient_cash" + actionable = False + + # --- Z-score exit overlay --- + zscore_overlay_applied = bool(zscore_exit["applied"]) + if zscore_overlay_applied and not is_window and not coerce_bool( + zscore_exit_allow_outside_execution_window, default=True, + ): + zscore_overlay_applied = False + zscore_exit["applied"] = False + + planned_investment = float(planned_investment if actionable or zscore_overlay_applied else 0.0) + + # --- Target values --- + target_btc_value = btc_value + if zscore_overlay_applied: + controlled_value = btc_value + planned_investment + target_btc_value = controlled_value * float(zscore_exit["target_btc_exposure"]) + actionable = True + skip_reason = None + elif actionable: + target_btc_value = btc_value + planned_investment + + target_values = {symbol: target_btc_value} + + # --- Signal description --- + localized_regime = _localized_regime(regime, translator) + displayed_planned = float(planned_investment if actionable else 0.0) + cash_for_display = float(investable_cash) + + if smart_enabled: + signal_desc = translate_with_fallback( + translator, + "btc_dca_smart_signal", + fallback_en=( + "BTC Smart DCA {regime}: multiplier {multiplier}, " + "planned buy ${planned_investment} from cash ${available_cash}" + ), + fallback_zh=( + "BTC 智能定投 {regime}: 倍数 {multiplier}，计划买入 ${planned_investment}，" + "现金 ${available_cash}" + ), + regime=localized_regime, + multiplier=f"{regime_multiplier:.2f}x", + planned_investment=f"{displayed_planned:,.2f}", + available_cash=f"{cash_for_display:,.2f}", + ) + else: + signal_desc = translate_with_fallback( + translator, + "btc_dca_ordinary_signal", + fallback_en="BTC ordinary DCA: planned buy ${planned_investment} from cash ${available_cash}", + fallback_zh="BTC 普通定投：计划买入 ${planned_investment}，现金 ${available_cash}", + planned_investment=f"{displayed_planned:,.2f}", + available_cash=f"{cash_for_display:,.2f}", + ) + + if cash_capped and planned_investment > 0.0: + signal_desc = translate_with_fallback( + translator, + "btc_dca_cash_capped", + fallback_en="{signal} | cash capped from ${requested_investment}", + fallback_zh="{signal} | 因现金限制，低于请求金额 ${requested_investment}", + signal=signal_desc, + requested_investment=f"{requested_investment:,.2f}", + ) + + if skip_reason: + signal_desc = translate_with_fallback( + translator, + "btc_dca_skip", + fallback_en="{signal} | skip: {skip_reason}", + fallback_zh="{signal} | 跳过：{skip_reason}", + signal=signal_desc, + skip_reason=_localized_skip_reason(skip_reason, translator), + ) + + if zscore_exit["found"]: + signal_desc = translate_with_fallback( + translator, + "btc_dca_zscore_overlay", + fallback_en=( + "{signal} | Z-Score exit: {route}, target {btc_pct} BTC / " + "{parking_pct} {parking_symbol}" + ), + fallback_zh=( + "{signal} | Z-Score 逃顶: {route}，目标 {btc_pct} BTC / " + "{parking_pct} {parking_symbol}" + ), + signal=signal_desc, + route=str(zscore_exit["route"] or "inactive"), + btc_pct=f"{float(zscore_exit['target_btc_exposure']):.0%}", + parking_pct=f"{float(zscore_exit['target_parking_exposure']):.0%}", + parking_symbol=str(zscore_exit["parking_symbol"]), + ) + + # --- Status description --- + if smart_enabled: + cycle_text = "" + ahr999_val = aggregate_metrics.get("ahr999") + if ahr999_val is not None and not pd.isna(ahr999_val): + cycle_text = f", AHR999 {float(ahr999_val):.2f}" + status_desc = translate_with_fallback( + translator, + "btc_dca_status", + fallback_en="{window} | drawdown {drawdown}, gap vs SMA200 {sma_gap}{cycle_text}", + fallback_zh="{window} | 回撤 {drawdown}，SMA200偏离 {sma_gap}{cycle_text}", + window=window_text, + drawdown=f"{float(aggregate_metrics.get('drawdown_252d', float('nan'))):.1%}", + sma_gap=f"{float(aggregate_metrics.get('sma200_gap', float('nan'))):.1%}", + cycle_text=cycle_text, + ) + else: + status_desc = translate_with_fallback( + translator, + "btc_dca_status_ordinary", + fallback_en="{window} | ordinary DCA", + fallback_zh="{window} | 普通定投", + window=window_text, + ) + + return { + "actionable": actionable, + "skip_reason": skip_reason, + "target_values": target_values if actionable else {}, + "managed_symbols": (symbol,), + "signal_symbol": symbol, + "signal_description": signal_desc, + "status_description": status_desc, + "regime": regime, + "multiplier": float(regime_multiplier), + "regime_multiplier": float(regime_multiplier), + "smart_multiplier_enabled": bool(smart_enabled), + "base_investment_usd": float(base_investment_usd), + "requested_investment_usd": float(requested_investment), + "planned_investment_usd": float(displayed_planned), + "available_cash": float(buying_power), + "reserved_cash": float(reserved_cash), + "investable_cash": float(investable_cash), + "cash_capped": bool(cash_capped), + "min_investment_usd": float(min_investment_usd), + "execution_window": window_text, + "in_execution_window": bool(is_window), + "zscore_exit": zscore_exit, + "total_equity": float(total_equity), + "btc_value": float(btc_value), + **aggregate_metrics, + } + + +# --------------------------------------------------------------------------- +# Public API (contract-compatible with existing callers) +# --------------------------------------------------------------------------- + + def build_target_weights( prices: dict[str, float | None], portfolio: Any, @@ -41,8 +897,7 @@ def build_target_weights( Always returns {BTCUSDT: 1.0} since this is a single-asset DCA strategy. The BTC target ratio is computed via get_dynamic_btc_target_ratio but the weight returned here always reflects a 100% allocation of the available - DCA budget to BTCUSDT. The ratio is applied externally (e.g. in budget - computation). + DCA budget to BTCUSDT. The ratio is applied externally. """ return {"BTCUSDT": 1.0} @@ -52,15 +907,48 @@ def compute_signals( portfolio: Any, total_equity: float, state: dict[str, Any] | None = None, + *, + derived_indicators: Mapping[str, object] | None = None, + translator=None, + **kwargs: Any, ) -> dict[str, Any]: """Compute DCA signals for BTC. - Returns a dict with BTC target weight, the dynamic ratio, and the equity. + Returns a dict compatible with both the standalone entrypoint and the + combo strategy. """ if state is None: state = {} + + # Build full rebalance plan if we have portfolio + config + plan: dict[str, object] = {} + portfolio_for_plan = portfolio + + # Extract runtime config from kwargs (combo-style) or use defaults + smart_enabled = coerce_bool( + kwargs.get("smart_multiplier_enabled", True), default=True, + ) + base_amount = coerce_float( + kwargs.get("base_investment_usd", 100.0), default=100.0, + ) + + try: + plan = build_rebalance_plan( + portfolio_for_plan, + as_of=kwargs.get("as_of"), + smart_multiplier_enabled=smart_enabled, + base_investment_usd=base_amount, + derived_indicators=derived_indicators, + translator=translator, + **{k: v for k, v in kwargs.items() if k not in ("as_of",)}, + ) + except (ValueError, TypeError) as exc: + logger.warning("btc_dca build_rebalance_plan failed, falling back: %s", exc) + plan = {} + ratio = get_dynamic_btc_target_ratio(total_equity) - return { + + result: dict[str, Any] = { "signals": { "BTCUSDT": { "target_weight": 1.0, @@ -72,6 +960,24 @@ def compute_signals( "profile": PROFILE_NAME, } + if plan: + result["metadata"] = { + "actionable": plan.get("actionable", True), + "regime": plan.get("regime", "ordinary_dca"), + "multiplier": plan.get("multiplier", 1.0), + "smart_multiplier_enabled": plan.get("smart_multiplier_enabled", smart_enabled), + "planned_investment_usd": plan.get("planned_investment_usd", 0.0), + "signal_description": plan.get("signal_description", ""), + "status_description": plan.get("status_description", ""), + "zscore_exit": plan.get("zscore_exit", {}), + "in_execution_window": plan.get("in_execution_window", True), + "execution_window": plan.get("execution_window", ""), + "ahr999": plan.get("ahr999", float("nan")), + "mayer_multiple": plan.get("mayer_multiple", float("nan")), + } + + return result + def extract_managed_symbols( state: dict[str, Any] | None = None, diff --git a/src/crypto_strategies/strategies/crypto_equity_combo.py b/src/crypto_strategies/strategies/crypto_equity_combo.py index 0de6e2c..b07ecbd 100644 --- a/src/crypto_strategies/strategies/crypto_equity_combo.py +++ b/src/crypto_strategies/strategies/crypto_equity_combo.py @@ -1,16 +1,13 @@ """Crypto equity combo strategy — BTC DCA + Trend Rotation. -Combines two crypto sub-strategies (BTC DCA and Trend Rotation) into a single -weight-allocated portfolio. +Combines two enhanced independent strategies into a single weight-allocated +portfolio. BTC leg delegates to ``crypto_btc_dca.compute_signals`` so that +smart sizing (AHR999, drawdown multipliers, Z-score exit) is active when +configured. -Static mode ------------ -Fixed weights per leg (default: 30/70 BTC/trend). - -Dynamic mode ------------- -Regime-based adjustment: when the benchmark snapshot signals regime_off, -reduce the trend leg weight by 50 % and re-allocate the freed budget to BTC. +Static mode: fixed weights per leg (default: 30/70 BTC/trend). +Dynamic mode: regime-based adjustment — when BTC is below SMA200, reduce +trend leg by 50 % and re-allocate to BTC. Usage ----- @@ -19,44 +16,126 @@ from __future__ import annotations +import logging from typing import Any +from crypto_strategies._utils import coerce_float, translate_with_fallback +from crypto_strategies.strategies.crypto_trend_rotation import ( + _apply_volatility_scaling, + _check_circuit_breaker, + _extract_btc_snapshot, +) -from crypto_strategies.strategies import crypto_btc_dca -from crypto_strategies.strategies import crypto_trend_rotation +logger = logging.getLogger(__name__) CRYPTO_EQUITY_DOMAIN = "crypto_equity" SIGNAL_SOURCE = "combo" STATUS_ICON = "\U0001f500" PROFILE_NAME = "crypto_equity_combo" -# Default static weights DEFAULT_BTC_WEIGHT = 0.30 DEFAULT_TREND_WEIGHT = 0.70 +DYNAMIC_REGIME_OFF_CUT = 0.50 + + +def _compute_btc_leg( + total_equity: float, + btc_weight: float, + *, + prices: dict[str, float] | None = None, + portfolio: Any = None, + derived_indicators: dict[str, Any] | None = None, + translator=None, + **kwargs: Any, +) -> dict[str, float]: + """Compute BTC leg target using the enhanced smart DCA strategy. + + Delegates to ``crypto_btc_dca.compute_signals`` to incorporate + AHR999 cycle multiplier, drawdown sizing, and Z-score exit signals. + Falls back to equity-scaled ratio on error. + """ + from crypto_strategies.strategies.crypto_btc_dca import ( + compute_signals, + get_dynamic_btc_target_ratio, + ) -# Dynamic mode thresholds -DYNAMIC_REGIME_OFF_CUT = 0.50 # reduce trend by 50 % when regime_off + smart_ratio: float | None = None + try: + result = compute_signals( + prices=prices or {}, + portfolio=portfolio, + total_equity=total_equity, + derived_indicators=derived_indicators, + translator=translator, + **kwargs, + ) + metadata = result.get("metadata", {}) if isinstance(result, dict) else {} + regime = str(metadata.get("regime", "")) + # Use smart multiplier only when the strategy is in an accumulation regime + if regime and regime not in ("ordinary_dca", "ahr999_expensive"): + multiplier = float(metadata.get("multiplier", 1.0)) + base_ratio = get_dynamic_btc_target_ratio(total_equity) + # Scale ratio by multiplier, clamped to [base_ratio, base_ratio * 3] + smart_ratio = min(base_ratio * max(1.0, multiplier), base_ratio * 3.0) + smart_ratio = min(0.65, max(0.0, smart_ratio)) + except (ValueError, TypeError) as exc: + logger.debug("btc_dca smart signals unavailable (non-critical): %s", exc) + + if smart_ratio is None: + from crypto_strategies.strategies.crypto_btc_dca import ( + get_dynamic_btc_target_ratio, + ) + smart_ratio = get_dynamic_btc_target_ratio(total_equity) + + return {"BTCUSDT": smart_ratio * float(btc_weight)} + + +def _compute_trend_leg( + indicators_map: dict[str, dict[str, Any]], + prices: dict[str, float], + universe_snapshot: list[str], + state: dict[str, Any], + trend_weight: float, + trend_pool_size: int = 5, + rotation_top_n: int = 2, + weight_mode: str = "inverse_vol", + vol_scaling_enabled: bool = True, +) -> dict[str, float]: + """Compute trend leg targets using rotation logic.""" + from crypto_strategies.strategies.crypto_live_pool_rotation.core import ( + select_rotation_weights, + ) + from crypto_strategies.strategies.crypto_live_pool_rotation.rotation import ( + resolve_authoritative_rotation_pool, + ) -# BTC leg defaults -BTC_DEFAULT_CONFIG: dict[str, Any] = {} + btc_snapshot = _extract_btc_snapshot(indicators_map) + blocked, _ = _check_circuit_breaker(btc_snapshot) + if blocked: + return {} -# Trend rotation leg defaults -TREND_DEFAULT_CONFIG: dict[str, Any] = {} + trend_pool = resolve_authoritative_rotation_pool( + state, + trend_universe_symbols=list(universe_snapshot), + trend_pool_size=trend_pool_size, + allow_refresh=True, + ) + candidates = select_rotation_weights( + indicators_map, prices, btc_snapshot, trend_pool, + rotation_top_n, weight_mode=weight_mode, + ) + if not candidates: + return {} -def _clean_kwargs(kwargs: dict[str, Any]) -> dict[str, Any]: - """Remove combo-level keys before passing to sub-strategies.""" - ignored = { - "btc_weight", - "trend_weight", - "dynamic_mode", - "translator", - "signal_text_fn", - "execution_cash_reserve_ratio", - "rebalance_frequency", - "run_as_of", + raw_weights = { + sym: float(payload["weight"]) * float(trend_weight) + for sym, payload in candidates.items() } - return {k: v for k, v in kwargs.items() if k not in ignored} + return _apply_volatility_scaling( + raw_weights, indicators_map, + vol_scaling_enabled=vol_scaling_enabled, + ) def build_target_weights( @@ -70,75 +149,40 @@ def build_target_weights( btc_weight: float = DEFAULT_BTC_WEIGHT, trend_weight: float = DEFAULT_TREND_WEIGHT, dynamic_mode: bool = True, - btc_config: dict[str, Any] | None = None, - trend_config: dict[str, Any] | None = None, + translator=None, **kwargs: Any, ) -> tuple[dict[str, float], dict[str, object]]: - """Compute combined target weights from both sub-strategies. - - Parameters - ---------- - prices : dict or None - Current price map (symbol -> float). - indicators_map : dict or None - Derived indicator map (symbol -> dict). - universe_snapshot : dict or None - Trend universe snapshot. - benchmark_snapshot : dict or None - Benchmark snapshot used for regime detection. - portfolio : dict or None - Current portfolio snapshot. - state : dict or None - Strategy state (may be mutated by sub-strategies). - btc_weight, trend_weight : float - Allocation weights for each leg. Should sum to 1.0. - dynamic_mode : bool - If True, reduce trend allocation when the benchmark snapshot - signals ``regime_on == False``. - btc_config, trend_config : dict or None - Overrides passed to each sub-strategy's ``build_target_weights``. - **kwargs : Any - Ignored (compatibility with runtime entrypoint). - """ - resolved_btc = dict(BTC_DEFAULT_CONFIG) - resolved_btc.update(btc_config or {}) - - resolved_trend = dict(TREND_DEFAULT_CONFIG) - resolved_trend.update(trend_config or {}) - - # Compute each leg - btc_weights: dict[str, float] = {} - trend_raw_weights: dict[str, float] = {} - trend_metadata: dict[str, object] = {} - - if prices is not None and indicators_map is not None and benchmark_snapshot is not None: - try: - btc_weights, _ = crypto_btc_dca.build_target_weights( - prices=prices, - indicators_map=indicators_map, - benchmark_snapshot=benchmark_snapshot, - portfolio=portfolio, - state=state, - **resolved_btc, - ) - except Exception: - btc_weights = {} - - try: - trend_raw_weights, _, trend_metadata = crypto_trend_rotation.build_target_weights( - prices=prices, - indicators_map=indicators_map, - universe_snapshot=universe_snapshot, - benchmark_snapshot=benchmark_snapshot, - portfolio=portfolio, - state=state, - **resolved_trend, - ) - except Exception: - trend_raw_weights = {} - - # Determine effective weights (dynamic adjustment) - regime_off = bool(benchmark_snapshot.get("regime_on", True)) is False if benchmark_snapshot else False + """Compute combined target weights from both sub-strategies.""" + if prices is None or indicators_map is None: + return {}, {"error": "missing required inputs"} + + prices = {str(k).strip().upper(): float(v or 0) for k, v in prices.items() if v is not None} + + if isinstance(universe_snapshot, dict): + universe_symbols = list(universe_snapshot.keys()) + elif isinstance(universe_snapshot, (list, tuple)): + universe_symbols = list(universe_snapshot) + else: + universe_symbols = list(prices) + + total_equity = 100000.0 + if isinstance(portfolio, dict): + total_equity = coerce_float(portfolio.get("total_equity"), default=100000.0) + elif portfolio is not None: + total_equity = coerce_float(getattr(portfolio, "total_equity", None), default=100000.0) + + state = state or {} + + # Dynamic regime adjustment + regime_off = False + if benchmark_snapshot: + regime_on = benchmark_snapshot.get("regime_on") + if regime_on is not None: + regime_off = not bool(regime_on) + else: + btc_snapshot = _extract_btc_snapshot(indicators_map or {}) + regime_off = not btc_snapshot.get("regime_on", True) + if dynamic_mode and regime_off: effective_btc = btc_weight + trend_weight * DYNAMIC_REGIME_OFF_CUT effective_trend = trend_weight * (1.0 - DYNAMIC_REGIME_OFF_CUT) @@ -146,39 +190,51 @@ def build_target_weights( effective_btc = btc_weight effective_trend = trend_weight - # Combine weights - all_symbols = set(btc_weights) | set(trend_raw_weights) + # Compute legs + btc_weights = _compute_btc_leg( + total_equity, effective_btc, + prices=prices, portfolio=portfolio, + derived_indicators=indicators_map, + translator=translator, + **kwargs, + ) + + trend_weights: dict[str, float] = {} + try: + trend_weights = _compute_trend_leg( + indicators_map, prices, universe_symbols, state, effective_trend, + trend_pool_size=int(kwargs.get("trend_pool_size", 5)), + rotation_top_n=int(kwargs.get("rotation_top_n", 2)), + vol_scaling_enabled=bool(kwargs.get("vol_scaling_enabled", True)), + ) + except (ValueError, TypeError, KeyError) as exc: + logger.warning("trend_leg failed, using empty weights: %s", exc) + + # Combine + all_symbols = set(btc_weights) | set(trend_weights) combined: dict[str, float] = {} for symbol in all_symbols: - bw = btc_weights.get(symbol, 0.0) - tw = trend_raw_weights.get(symbol, 0.0) - combined[symbol] = bw * effective_btc + tw * effective_trend + combined[symbol] = btc_weights.get(symbol, 0.0) + trend_weights.get(symbol, 0.0) - # Normalize to ensure sum <= 1.0 total = sum(combined.values()) - if total > 0.0: - scale = min(1.0, 1.0 / total) if total > 1.0 else 1.0 - if scale < 1.0: - combined = {s: w * scale for s, w in combined.items()} + if total > 1.0: + combined = {s: w / total for s, w in combined.items()} metadata: dict[str, object] = { "combo": { "btc_weight": effective_btc, "trend_weight": effective_trend, + "base_btc_weight": btc_weight, + "base_trend_weight": trend_weight, }, - "legs": { - "btc": {"weights": btc_weights, "configured_weight": btc_weight}, - "trend": { - "weights": trend_raw_weights, - "configured_weight": trend_weight, - }, - }, + "btc_leg": {"weights": btc_weights}, + "trend_leg": {"weights": trend_weights}, "regime_off": regime_off, "dynamic_mode": dynamic_mode, "gross_exposure": sum(combined.values()), "selected_count": len(combined), + "total_equity": total_equity, } - return combined, metadata @@ -193,12 +249,13 @@ def compute_signals( benchmark_snapshot: dict[str, Any] | None = None, portfolio: dict[str, Any] | None = None, state: dict[str, Any] | None = None, + translator=None, **kwargs: Any, ): - kwargs.pop("translator", None) - kwargs.pop("signal_text_fn", None) - kwargs.pop("execution_cash_reserve_ratio", None) + """Compute combo signals with dynamic regime-based allocation. + Returns (weights, signal_desc, has_cash_residual, status_desc, metadata). + """ weights, metadata = build_target_weights( prices=prices, indicators_map=indicators_map, @@ -206,25 +263,48 @@ def compute_signals( benchmark_snapshot=benchmark_snapshot, portfolio=portfolio, state=state, + translator=translator, **kwargs, ) combo_meta = metadata.get("combo", {}) + if isinstance(combo_meta, dict): + btw = combo_meta.get("btc_weight", 0) + trw = combo_meta.get("trend_weight", 0) + else: + btw, trw = 0, 0 + regime_off = metadata.get("regime_off", False) regime_label = "regime_off" if regime_off else "regime_on" - selected = ",".join(weights.keys()) if weights else "cash" - signal_desc = ( - f"combo {regime_label} selected={selected} " - f"gross={metadata['gross_exposure']:.0%} " - f"btc={combo_meta.get('btc_weight', 0):.0%} " - f"trend={combo_meta.get('trend_weight', 0):.0%}" + gross = float(metadata.get("gross_exposure", 0.0)) + + if weights: + selected = ",".join( + f"{s}({w:.1%})" for s, w in sorted(weights.items(), key=lambda x: -x[1])[:6] + ) + else: + selected = "cash" + + signal_desc = translate_with_fallback( + translator, + "combo_signal", + fallback_en=( + f"combo {regime_label} selected={selected} " + f"gross={gross:.0%} btc={btw:.0%} trend={trw:.0%}" + ), + fallback_zh=( + f"组合策略 {regime_label} 选中={selected} " + f"总仓位={gross:.0%} BTC={btw:.0%} 趋势={trw:.0%}" + ), ) - status_desc = ( - f"{regime_label} | " - f"btc={combo_meta.get('btc_weight', 0):.0%} " - f"trend={combo_meta.get('trend_weight', 0):.0%}" + status_desc = translate_with_fallback( + translator, + "combo_status", + fallback_en=f"{regime_label} | btc={btw:.0%} trend={trw:.0%} | {len(weights)} positions", + fallback_zh=f"{regime_label} | BTC={btw:.0%} 趋势={trw:.0%} | {len(weights)} 个仓位", ) - has_cash_residual = metadata["gross_exposure"] < 0.999 + + has_cash_residual = gross < 0.999 return ( weights, diff --git a/src/crypto_strategies/strategies/crypto_trend_rotation.py b/src/crypto_strategies/strategies/crypto_trend_rotation.py index 44ec053..78996fb 100644 --- a/src/crypto_strategies/strategies/crypto_trend_rotation.py +++ b/src/crypto_strategies/strategies/crypto_trend_rotation.py @@ -1,12 +1,18 @@ -"""Trend rotation standalone strategy (stripped of BTC DCA). +"""Trend rotation standalone strategy (enhanced). Profile: crypto_trend_rotation Domain: crypto -Source: feature_snapshot +Source: feature_snapshot + derived_indicators (BTC benchmark) -This strategy reuses the core rank/weight/budget/sell logic from +This strategy reuses the core rank/weight/sell logic from crypto_live_pool_rotation but does NOT allocate any budget to BTC. It is a pure altcoin trend-rotation signal. + +Enhancements over the original stripped version: +- Proper BTC benchmark snapshot integration (fixes the empty {} bug) +- Volatility-based position sizing +- Market drawdown circuit breaker +- i18n signal descriptions """ from __future__ import annotations @@ -15,6 +21,11 @@ import pandas as pd +from crypto_strategies._utils import ( + coerce_bool, + coerce_float, + translate_with_fallback, +) from crypto_strategies.strategies.crypto_live_pool_rotation.core import ( select_rotation_weights, ) @@ -44,6 +55,142 @@ } ) +# --- BTC benchmark extraction --- + + +def _extract_btc_snapshot(indicators_map: dict[str, dict[str, Any]]) -> dict[str, Any]: + """Extract BTC benchmark data from the indicators map. + + Looks for BTCUSDT, BTC-USD, or BTC key in the map and extracts + the necessary roc and regime fields. + """ + btc_keys = ("BTCUSDT", "BTC-USD", "BTC", "BTCUSDT.P") + btc_data = None + for key in btc_keys: + candidate = indicators_map.get(key) + if isinstance(candidate, dict) and candidate: + btc_data = candidate + break + + if btc_data is None: + # Search case-insensitively + for k, v in indicators_map.items(): + if isinstance(v, dict) and str(k).upper().replace("-", "").replace(".", "") == "BTCUSDT": + btc_data = v + break + + if btc_data is None: + # Return a default "regime on" snapshot so rotation can proceed + return { + "regime_on": True, + "btc_roc20": 0.05, + "btc_roc60": 0.10, + "btc_roc120": 0.20, + } + + roc20 = coerce_float(btc_data.get("roc20"), default=0.05) + roc60 = coerce_float(btc_data.get("roc60"), default=0.05) + roc120 = coerce_float(btc_data.get("roc120"), default=0.05) + close = coerce_float(btc_data.get("close"), default=0.0) + sma200 = coerce_float(btc_data.get("sma200"), default=0.0) + + # Regime on = BTC price above SMA200 (long-term uptrend) + regime_on = bool(close > sma200 if sma200 > 0 else True) + # Also check explicit regime field if present + explicit_regime = btc_data.get("regime_on") + if explicit_regime is not None: + regime_on = coerce_bool(explicit_regime, default=True) + + return { + "regime_on": regime_on, + "btc_roc20": float(roc20), + "btc_roc60": float(roc60), + "btc_roc120": float(roc120), + } + + +# --- Market circuit breaker --- + + +def _check_circuit_breaker( + btc_snapshot: dict[str, Any], + *, + circuit_breaker_enabled: bool = True, + btc_drawdown_threshold: float = 0.30, +) -> tuple[bool, str]: + """Check if trend rotation should be suspended due to market stress. + + Returns (blocked, reason). + """ + if not coerce_bool(circuit_breaker_enabled, default=True): + return False, "" + + if not btc_snapshot.get("regime_on", True): + return True, "btc_below_sma200" + + # Check for extreme BTC drawdown + btc_close = coerce_float(btc_snapshot.get("close"), default=float("nan")) + btc_sma200 = coerce_float(btc_snapshot.get("sma200"), default=float("nan")) + if not pd.isna(btc_close) and not pd.isna(btc_sma200) and btc_sma200 > 0: + drawdown = 1.0 - btc_close / btc_sma200 + if drawdown > float(btc_drawdown_threshold): + return True, f"btc_drawdown_{drawdown:.0%}_exceeds_{btc_drawdown_threshold:.0%}" + + return False, "" + + +# --- Volatility-based position sizing --- + + +def _apply_volatility_scaling( + weights: dict[str, float], + indicators_map: dict[str, dict[str, Any]], + *, + vol_scaling_enabled: bool = True, + target_vol: float = 0.40, + max_leverage: float = 1.0, +) -> 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%). + """ + if not coerce_bool(vol_scaling_enabled, default=True): + return weights + if not weights: + return weights + + total_weight = sum(weights.values()) + if total_weight <= 0: + return weights + + # Estimate portfolio vol as weighted average of individual vols + 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: + 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) + + # 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 + + return {symbol: weight * scale for symbol, weight in weights.items()} + + +# --- Core --- + def _to_indicator_frame(feature_snapshot) -> pd.DataFrame: """Normalise a feature snapshot into a DataFrame indexed by symbol.""" @@ -77,14 +224,18 @@ def build_target_weights( Steps ----- - 1. Resolve the authoritative rotation pool from universe snapshot. - 2. Select rotation weights (top N by relative strength) from that pool. - 3. Return the altcoin weights dict -- no BTC allocation. + 1. Extract BTC benchmark from indicators_map (fixes the empty {} bug). + 2. Check market circuit breaker. + 3. Resolve the authoritative rotation pool from universe snapshot. + 4. Select rotation weights (top N by relative strength) from that pool. + 5. Apply volatility-based position scaling. + 6. Return the altcoin weights dict — no BTC allocation. Parameters ---------- indicators_map : dict[str, dict] - Symbol -> indicator-dict (close, sma*, roc*, vol*, etc.). + Symbol → indicator-dict (close, sma*, roc*, vol*, etc.). + Must contain BTCUSDT (or equivalent) for benchmark extraction. universe_snapshot : list[str] Full candidate universe provided by the upstream platform. prices : dict[str, float] @@ -95,20 +246,38 @@ def build_target_weights( Runtime configuration keys: - trend_pool_size (int, default 5) - rotation_top_n (int, default 2) - - weight_mode (str, default ``"inverse_vol"``) + - weight_mode (str, default "inverse_vol") - allow_rotation_refresh (bool, default True) + - circuit_breaker_enabled (bool, default True) + - btc_drawdown_threshold (float, default 0.30) + - vol_scaling_enabled (bool, default True) Returns ------- dict[str, dict] Selected candidates keyed by symbol, each holding ``weight``, ``relative_score``, and ``abs_momentum``. Empty dict when no - candidates pass filters. + candidates pass filters or circuit breaker is active. """ trend_pool_size = int(config.get("trend_pool_size", 5)) rotation_top_n = int(config.get("rotation_top_n", 2)) weight_mode = str(config.get("weight_mode", "inverse_vol")) allow_refresh = bool(config.get("allow_rotation_refresh", True)) + 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) + + # Extract BTC benchmark from indicators_map + btc_snapshot = _extract_btc_snapshot(indicators_map) + + # Circuit breaker check + blocked, block_reason = _check_circuit_breaker( + btc_snapshot, + circuit_breaker_enabled=circuit_breaker_enabled, + btc_drawdown_threshold=btc_drawdown_threshold, + ) + if blocked: + return {} trend_pool = resolve_authoritative_rotation_pool( state, @@ -120,12 +289,34 @@ def build_target_weights( selected_candidates = select_rotation_weights( indicators_map, prices, - {}, + btc_snapshot, # FIXED: was {} — now passes real BTC benchmark data trend_pool, rotation_top_n, weight_mode=weight_mode, ) - return selected_candidates + + if not selected_candidates: + return {} + + # Apply volatility scaling + weights_map = { + sym: float(payload["weight"]) + for sym, payload in selected_candidates.items() + } + scaled_weights = _apply_volatility_scaling( + weights_map, + indicators_map, + vol_scaling_enabled=vol_scaling_enabled, + ) + + return { + sym: { + "weight": scaled_weights.get(sym, selected_candidates[sym]["weight"]), + "relative_score": selected_candidates[sym]["relative_score"], + "abs_momentum": selected_candidates[sym]["abs_momentum"], + } + for sym in selected_candidates + } def compute_signals( @@ -144,32 +335,27 @@ def compute_signals( ---------- feature_snapshot : pd.DataFrame or list[dict] Rows containing the required feature columns. + Must include BTCUSDT row for benchmark extraction. current_holdings : list-like Symbols currently held (used for sell-reason checks). translator : callable or None - Optional translation helper for sell-reason text. + Optional translation helper. **kwargs - Forwarded to the internal rotation helpers. Supported keys: - - - trend_pool_size (int) - - rotation_top_n (int) - - weight_mode (str) - - allow_rotation_refresh (bool) - - atr_multiplier (float) + Forwarded to the internal rotation helpers. Returns ------- tuple ``(weights, signal_desc, is_emergency, debug_str, metadata)`` """ - _ = translator - del translator - config = { "trend_pool_size": kwargs.get("trend_pool_size", 5), "rotation_top_n": kwargs.get("rotation_top_n", 2), "weight_mode": kwargs.get("weight_mode", "inverse_vol"), "allow_rotation_refresh": kwargs.get("allow_rotation_refresh", True), + "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), } frame = _to_indicator_frame(feature_snapshot) @@ -191,9 +377,15 @@ def compute_signals( ) if not selected: + signal_desc = translate_with_fallback( + translator, + "trend_rotation_no_candidates", + fallback_en="crypto_trend_rotation: no candidates passed filters", + fallback_zh="加密货币趋势轮动：无候选币种通过筛选", + ) return ( None, - "crypto_trend_rotation: no candidates passed filters", + signal_desc, False, "no_candidates", {"managed_symbols": tuple(universe_symbols), "profile": PROFILE_NAME}, @@ -202,13 +394,19 @@ def compute_signals( weights = {sym: float(payload["weight"]) for sym, payload in selected.items()} selected_symbols = ", ".join( f"{sym}({payload['relative_score']:.3f})" - for sym, payload in selected.items() + for sym, payload in sorted(selected.items(), key=lambda x: -x[1]["relative_score"]) + ) + signal_desc = translate_with_fallback( + translator, + "trend_rotation_selected", + fallback_en=f"crypto_trend_rotation selected: {selected_symbols}", + fallback_zh=f"加密货币趋势轮动选中：{selected_symbols}", ) - signal_desc = f"crypto_trend_rotation selected: {selected_symbols}" metadata: dict[str, Any] = { "managed_symbols": tuple(universe_symbols), "profile": PROFILE_NAME, + "selected_count": len(selected), "selected_candidates": { sym: { "weight": float(payload["weight"]), @@ -235,8 +433,7 @@ def extract_managed_symbols( feature_snapshot : pd.DataFrame or list[dict] Feature rows with at least a ``symbol`` column. benchmark_symbol : str or None - Ignored in the altcoin-only context (kept for interface - compatibility). + Ignored in the altcoin-only context. Returns ------- diff --git a/tests/test_crypto_btc_dca.py b/tests/test_crypto_btc_dca.py index 1dc7d72..f8946a5 100644 --- a/tests/test_crypto_btc_dca.py +++ b/tests/test_crypto_btc_dca.py @@ -1,102 +1,401 @@ -"""Tests for the crypto_btc_dca standalone strategy module.""" +"""Tests for the enhanced crypto_btc_dca strategy module.""" from __future__ import annotations import math import unittest +from unittest.mock import MagicMock + +import pandas as pd from crypto_strategies.strategies.crypto_btc_dca import ( PROFILE_NAME, + CRYPTO_DOMAIN, + SIGNAL_SOURCE, + STATUS_ICON, + DEFAULT_SIGNAL_SYMBOL, + DEFAULT_PARKING_SYMBOL, + build_rebalance_plan, + build_target_weights, compute_signals, extract_managed_symbols, get_dynamic_btc_target_ratio, + _determine_cycle_multiplier, + _determine_multiplier, + _is_in_execution_window, ) -class CryptoBtcDcaModuleTest(unittest.TestCase): - """Verify the module-level constants and standalone helpers.""" +# --------------------------------------------------------------------------- +# Portfolio stub +# --------------------------------------------------------------------------- + + +class FakePortfolio: + def __init__( + self, + total_equity: float = 50000.0, + buying_power: float = 5000.0, + btc_units: float = 0.0, + btc_price: float = 60000.0, + metadata: dict | None = None, + ): + self.total_equity = total_equity + self.buying_power = buying_power + self.cash_balance = buying_power + self.positions = ( + [type("Pos", (), {"symbol": "BTCUSDT", "quantity": btc_units, + "market_value": btc_units * btc_price})()] + if btc_units > 0 + else () + ) + self.metadata = metadata or {} + +def _zh_translator(key: str, **kwargs) -> str: + return key + + +# --------------------------------------------------------------------------- +# Core module tests +# --------------------------------------------------------------------------- + + +class CryptoBtcDcaModuleTest(unittest.TestCase): def test_profile_name(self) -> None: self.assertEqual(PROFILE_NAME, "crypto_btc_dca") - def test_get_dynamic_btc_target_ratio_at_minimum_equity(self) -> None: - """At very low equity the ratio should be at its base value ~0.14.""" + def test_domain(self) -> None: + self.assertEqual(CRYPTO_DOMAIN, "crypto") + + def test_signal_source(self) -> None: + self.assertEqual(SIGNAL_SOURCE, "derived_indicators+portfolio_snapshot") + + def test_status_icon(self) -> None: + self.assertEqual(STATUS_ICON, "₿") + + def test_default_signal_symbol(self) -> None: + self.assertEqual(DEFAULT_SIGNAL_SYMBOL, "BTCUSDT") + + def test_default_parking_symbol(self) -> None: + self.assertEqual(DEFAULT_PARKING_SYMBOL, "USDT") + + +# --------------------------------------------------------------------------- +# Dynamic BTC target ratio +# --------------------------------------------------------------------------- + + +class DynamicBtcTargetRatioTest(unittest.TestCase): + def test_at_minimum_equity(self) -> None: ratio = get_dynamic_btc_target_ratio(0.0) - # safe_equity is clamped to 1.0, so ratio = 0.14 + 0.16 * log1p(1/10000) expected = 0.14 + 0.16 * math.log1p(1.0 / 10000.0) self.assertAlmostEqual(ratio, expected) - def test_get_dynamic_btc_target_ratio_at_maximum_equity(self) -> None: - """At very high equity the ratio should be clamped to 0.65.""" + def test_at_maximum_equity(self) -> None: ratio = get_dynamic_btc_target_ratio(1_000_000_000.0) self.assertAlmostEqual(ratio, 0.65) - def test_get_dynamic_btc_target_ratio_mid_equity(self) -> None: - """At 10000 equity: 0.14 + 0.16 * log1p(1) = 0.14 + 0.16*ln(2) ~ 0.2509.""" + def test_mid_equity(self) -> None: ratio = get_dynamic_btc_target_ratio(10_000.0) expected = 0.14 + 0.16 * math.log1p(1.0) self.assertAlmostEqual(ratio, expected, places=6) - def test_get_dynamic_btc_target_ratio_negative_equity(self) -> None: - """Negative equity is internally clamped to 1.0, giving the base ratio ~0.14.""" + def test_negative_equity(self) -> None: ratio = get_dynamic_btc_target_ratio(-1.0) expected = 0.14 + 0.16 * math.log1p(1.0 / 10000.0) self.assertAlmostEqual(ratio, expected) - def test_get_dynamic_btc_target_ratio_monotonic(self) -> None: - """Ratio should be non-decreasing with equity.""" + def test_monotonic(self) -> None: ratios = [get_dynamic_btc_target_ratio(e) for e in [0, 1_000, 10_000, 100_000, 1_000_000]] for prev, curr in zip(ratios, ratios[1:]): self.assertGreaterEqual(curr, prev) - def test_compute_signals_returns_btcusdt(self) -> None: - """compute_signals should produce a signal for BTCUSDT.""" - prices = {"BTCUSDT": 60000.0, "ETHUSDT": 3000.0} - portfolio = None - total_equity = 50_000.0 - result = compute_signals( - prices=prices, - portfolio=portfolio, - total_equity=total_equity, +# --------------------------------------------------------------------------- +# Execution window +# --------------------------------------------------------------------------- + + +class ExecutionWindowTest(unittest.TestCase): + def test_monthly_day25_in_window(self) -> None: + in_window, text = _is_in_execution_window( + "2026-05-26", cadence="monthly", monthly_day=25, + monthly_window_calendar_days=5, weekly_day=4, + weekly_window_calendar_days=4, quarterly_months=(1, 4, 7, 10), + quarterly_day=25, quarterly_window_calendar_days=5, + ) + self.assertTrue(in_window) + self.assertIn("monthly_day=25", text) + + def test_monthly_day30_outside_window(self) -> None: + in_window, _ = _is_in_execution_window( + "2026-05-30", cadence="monthly", monthly_day=25, + monthly_window_calendar_days=5, weekly_day=4, + weekly_window_calendar_days=4, quarterly_months=(1, 4, 7, 10), + quarterly_day=25, quarterly_window_calendar_days=5, + ) + self.assertFalse(in_window) + + def test_monthly_day25_start_of_window(self) -> None: + in_window, _ = _is_in_execution_window( + "2026-05-25", cadence="monthly", monthly_day=25, + monthly_window_calendar_days=5, weekly_day=4, + weekly_window_calendar_days=4, quarterly_months=(1, 4, 7, 10), + quarterly_day=25, quarterly_window_calendar_days=5, + ) + self.assertTrue(in_window) + + def test_weekly_thursday_in_window(self) -> None: + # 2026-05-28 is a Thursday (weekday=3) + in_window, text = _is_in_execution_window( + "2026-05-28", cadence="weekly", weekly_day=3, + weekly_window_calendar_days=4, monthly_day=25, + monthly_window_calendar_days=5, quarterly_months=(1, 4, 7, 10), + quarterly_day=25, quarterly_window_calendar_days=5, + ) + self.assertTrue(in_window) + self.assertIn("weekly_day=3", text) + + def test_quarterly_in_window(self) -> None: + in_window, text = _is_in_execution_window( + "2026-04-26", cadence="quarterly", quarterly_months=(1, 4, 7, 10), + quarterly_day=25, quarterly_window_calendar_days=5, + monthly_day=25, monthly_window_calendar_days=5, + weekly_day=4, weekly_window_calendar_days=4, ) + self.assertTrue(in_window) + self.assertIn("quarterly_months=1,4,7,10", text) - signals = result.get("signals", {}) - self.assertIn("BTCUSDT", signals) - self.assertAlmostEqual(signals["BTCUSDT"]["target_weight"], 1.0) + +# --------------------------------------------------------------------------- +# Multiplier logic +# --------------------------------------------------------------------------- + + +class MultiplierTest(unittest.TestCase): + def test_ahr999_bottom(self) -> None: + mult, regime = _determine_cycle_multiplier( + {"ahr999": 0.30}, + ahr999_bottom_threshold=0.45, ahr999_accumulation_threshold=0.80, + ahr999_dca_threshold=1.20, ahr999_bottom_multiplier=3.0, + ahr999_accumulation_multiplier=2.25, ahr999_dca_multiplier=1.50, + ahr999_expensive_multiplier=0.0, base_multiplier=1.0, + ) + self.assertEqual(mult, 3.0) + self.assertEqual(regime, "ahr999_bottom") + + def test_ahr999_accumulation(self) -> None: + mult, regime = _determine_cycle_multiplier( + {"ahr999": 0.70}, + ahr999_bottom_threshold=0.45, ahr999_accumulation_threshold=0.80, + ahr999_dca_threshold=1.20, ahr999_bottom_multiplier=3.0, + ahr999_accumulation_multiplier=2.25, ahr999_dca_multiplier=1.50, + ahr999_expensive_multiplier=0.0, base_multiplier=1.0, + ) + self.assertEqual(mult, 2.25) + self.assertEqual(regime, "ahr999_accumulation") + + def test_ahr999_expensive_skips(self) -> None: + mult, regime = _determine_cycle_multiplier( + {"ahr999": 1.50}, + ahr999_bottom_threshold=0.45, ahr999_accumulation_threshold=0.80, + ahr999_dca_threshold=1.20, ahr999_bottom_multiplier=3.0, + ahr999_accumulation_multiplier=2.25, ahr999_dca_multiplier=1.50, + ahr999_expensive_multiplier=0.0, base_multiplier=1.0, + ) + self.assertEqual(mult, 0.0) + self.assertEqual(regime, "ahr999_expensive") + + def test_ahr999_missing_returns_base(self) -> None: + mult, regime = _determine_cycle_multiplier( + {}, + ahr999_bottom_threshold=0.45, ahr999_accumulation_threshold=0.80, + ahr999_dca_threshold=1.20, ahr999_bottom_multiplier=3.0, + ahr999_accumulation_multiplier=2.25, ahr999_dca_multiplier=1.50, + ahr999_expensive_multiplier=0.0, base_multiplier=1.0, + ) + self.assertEqual(mult, 1.0) + self.assertEqual(regime, "normal") + + def test_drawdown_severe(self) -> None: + mult, regime, _ = _determine_multiplier( + {"drawdown_252d": 0.45, "sma200_gap": -0.30, "rsi14": None}, + mild_drawdown_threshold=0.12, deep_drawdown_threshold=0.25, + severe_drawdown_threshold=0.40, mild_discount_gap=0.08, + deep_discount_gap=0.18, expensive_gap=0.30, very_expensive_gap=0.60, + shallow_drawdown_threshold=0.05, overbought_rsi=75.0, + mild_pullback_multiplier=1.50, deep_pullback_multiplier=2.25, + severe_pullback_multiplier=3.0, expensive_multiplier=1.0, + very_expensive_multiplier=1.0, base_multiplier=1.0, + ) + self.assertEqual(mult, 3.0) + self.assertEqual(regime, "severe_pullback") + + +# --------------------------------------------------------------------------- +# build_rebalance_plan tests +# --------------------------------------------------------------------------- + + +class BuildRebalancePlanTest(unittest.TestCase): + def test_ordinary_dca_defaults(self) -> None: + plan = build_rebalance_plan( + FakePortfolio(), + as_of="2026-05-26", + smart_multiplier_enabled=False, + ) + self.assertTrue(plan["actionable"]) + self.assertEqual(plan["regime"], "ordinary_dca") + self.assertEqual(plan["multiplier"], 1.0) + self.assertAlmostEqual(plan["planned_investment_usd"], 100.0) + self.assertIn("BTCUSDT", plan["target_values"]) + + def test_smart_multiplier_enabled_by_default(self) -> None: + plan = build_rebalance_plan( + FakePortfolio(), + as_of="2026-05-26", + ) + self.assertTrue(plan["smart_multiplier_enabled"]) + + def test_ahr999_bottom_with_indicator_data(self) -> None: + plan = build_rebalance_plan( + FakePortfolio(buying_power=5000.0), + as_of="2026-05-26", + smart_multiplier_enabled=True, + derived_indicators={ + "BTCUSDT": { + "ahr999": 0.29, + "mayer_multiple": 0.85, + } + }, + ) + self.assertTrue(plan["actionable"]) + self.assertEqual(plan["regime"], "ahr999_bottom") + self.assertEqual(plan["multiplier"], 3.0) + self.assertAlmostEqual(plan["requested_investment_usd"], 300.0) + + def test_ahr999_expensive_skips(self) -> None: + plan = build_rebalance_plan( + FakePortfolio(buying_power=5000.0), + as_of="2026-05-26", + smart_multiplier_enabled=True, + derived_indicators={ + "BTCUSDT": { + "ahr999": 1.50, + } + }, + ) + self.assertFalse(plan["actionable"]) + self.assertEqual(plan["skip_reason"], "valuation_too_expensive") + self.assertEqual(plan["regime"], "ahr999_expensive") + self.assertEqual(plan["multiplier"], 0.0) + + def test_skips_outside_execution_window(self) -> None: + plan = build_rebalance_plan( + FakePortfolio(), + as_of="2026-05-30", + ) + self.assertFalse(plan["actionable"]) + self.assertEqual(plan["skip_reason"], "outside_execution_window") + + def test_skips_insufficient_cash(self) -> None: + plan = build_rebalance_plan( + FakePortfolio(buying_power=2.0), + as_of="2026-05-26", + ) + self.assertFalse(plan["actionable"]) + self.assertEqual(plan["skip_reason"], "insufficient_cash") + + def test_zscore_exit_reduces_btc_exposure(self) -> None: + plan = build_rebalance_plan( + FakePortfolio(buying_power=500.0, btc_price=60000.0, btc_units=1.0), + as_of="2026-05-26", + zscore_exit_enabled=True, + zscore_exit_context={ + "plugin": "btc_zscore_exit", + "canonical_route": "risk_reduced", + "position_control": { + "final_route": "risk_reduced", + "target_allocations": {"BTCUSDT": 0.50, "USDT": 0.50}, + }, + }, + ) + self.assertTrue(plan["zscore_exit"]["applied"]) + self.assertEqual(plan["zscore_exit"]["route"], "risk_reduced") + self.assertAlmostEqual(plan["zscore_exit"]["target_btc_exposure"], 0.50) + + def test_i18n_chinese_signal(self) -> None: + plan = build_rebalance_plan( + FakePortfolio(buying_power=2.0), + as_of="2026-05-26", + translator=_zh_translator, + ) + self.assertIn("BTC", plan["signal_description"]) + self.assertIn("skip", plan["signal_description"]) + + +# --------------------------------------------------------------------------- +# compute_signals tests +# --------------------------------------------------------------------------- + + +class ComputeSignalsTest(unittest.TestCase): + def test_returns_btcusdt(self) -> None: + prices = {"BTCUSDT": 60000.0, "ETHUSDT": 3000.0} + result = compute_signals(prices=prices, portfolio=None, total_equity=50000.0) + self.assertIn("BTCUSDT", result["signals"]) + self.assertAlmostEqual(result["signals"]["BTCUSDT"]["target_weight"], 1.0) self.assertIn("btc_target_ratio", result) - self.assertIn("profile", result) - self.assertIn("total_equity", result) self.assertEqual(result["profile"], PROFILE_NAME) - self.assertEqual(result["total_equity"], total_equity) - def test_compute_signals_with_state(self) -> None: - """compute_signals should accept an optional state dict.""" + def test_with_state(self) -> None: prices = {"BTCUSDT": 60000.0} - state: dict = {"some_key": "value"} result = compute_signals( - prices=prices, - portfolio=None, - total_equity=10_000.0, - state=state, + prices=prices, portfolio=None, total_equity=10000.0, state={"k": "v"}, ) self.assertIn("BTCUSDT", result["signals"]) - def test_extract_managed_symbols(self) -> None: - """Managed symbols should only include BTCUSDT.""" - symbols = extract_managed_symbols() - self.assertEqual(symbols, ("BTCUSDT",)) + def test_with_derived_indicators_enables_smart_mode(self) -> None: + prices = {"BTCUSDT": 60000.0} + portfolio = FakePortfolio(buying_power=5000.0) + result = compute_signals( + prices=prices, + portfolio=portfolio, + total_equity=50000.0, + derived_indicators={"BTCUSDT": {"ahr999": 0.30}}, + as_of="2026-05-26", + ) + metadata = result.get("metadata", {}) + self.assertEqual(metadata.get("regime"), "ahr999_bottom") + self.assertEqual(metadata.get("multiplier"), 3.0) + self.assertTrue(metadata.get("smart_multiplier_enabled")) - def test_compute_signals_btc_target_ratio_matches_function(self) -> None: - """The btc_target_ratio in compute_signals output should match get_dynamic_btc_target_ratio.""" + def test_btc_target_ratio_matches_function(self) -> None: equity = 75_000.0 result = compute_signals( + prices={"BTCUSDT": 60000.0}, portfolio=None, total_equity=equity, + ) + expected = get_dynamic_btc_target_ratio(equity) + self.assertAlmostEqual(result["btc_target_ratio"], expected) + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +class PublicApiTest(unittest.TestCase): + def test_build_target_weights_returns_btc(self) -> None: + weights = build_target_weights( prices={"BTCUSDT": 60000.0}, portfolio=None, - total_equity=equity, + total_equity=50000.0, ) - expected_ratio = get_dynamic_btc_target_ratio(equity) - self.assertAlmostEqual(result["btc_target_ratio"], expected_ratio) + self.assertEqual(weights, {"BTCUSDT": 1.0}) + + def test_extract_managed_symbols(self) -> None: + self.assertEqual(extract_managed_symbols(), ("BTCUSDT",)) if __name__ == "__main__": diff --git a/tests/test_crypto_trend_rotation.py b/tests/test_crypto_trend_rotation.py index cf47b00..406fb15 100644 --- a/tests/test_crypto_trend_rotation.py +++ b/tests/test_crypto_trend_rotation.py @@ -85,14 +85,18 @@ def test_compute_signals_with_valid_data(self) -> None: current_holdings=[], ) - # The rotation pool requires btc_snapshot data via select_rotation_weights, - # which is called with {} in the standalone trend module. As a result, - # candidates are None (no regime_on signal). Verify the fallback metadata. - self.assertIsNone(weights) + # After enhancement: _extract_btc_snapshot provides default BTC benchmark + # data (regime_on=True) when no BTCUSDT row is present, so rotation now + # finds valid candidates from the test data. Both altcoins have: + # price > sma20, price > sma60, price > sma200, + # positive rel_20/60/120, positive abs_momentum => valid candidates. + self.assertIsNotNone(weights) + self.assertTrue(len(weights) > 0) self.assertFalse(is_emergency) - self.assertEqual(debug_str, "no_candidates") + self.assertEqual(debug_str, "ok") self.assertEqual(metadata["profile"], PROFILE_NAME) self.assertIn("managed_symbols", metadata) + self.assertIn("selected_candidates", metadata) def test_compute_signals_empty_snapshot(self) -> None: """An empty feature snapshot should return None weights."""