795 lines
33 KiB
Python
795 lines
33 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from urllib.error import HTTPError
|
|
from unittest.mock import patch
|
|
|
|
import pandas as pd
|
|
|
|
from strategy29.data import binance_history
|
|
from strategy29.common.models import MarketDataBundle, Regime
|
|
from strategy32.backtest.simulator import Strategy32Backtester, Strategy32MomentumCarryBacktester
|
|
from strategy32.config import PROFILE_V5_BASELINE, PROFILE_V7_DEFAULT, Strategy32Budgets, Strategy32Config, build_strategy32_config
|
|
from strategy32.live.executor import LiveExecutionConfig, LiveFuturesExecutor
|
|
from strategy32.live.runtime import (
|
|
BEST_CASH_OVERLAY,
|
|
LiveMonitorConfig,
|
|
_capital_summary,
|
|
_apply_weekly_macro_filter,
|
|
_combine_targets,
|
|
_completed_bar_time,
|
|
_execution_refinement_states,
|
|
_refine_execution_targets,
|
|
_expand_core_targets,
|
|
_heartbeat_slot,
|
|
_overlay_signal_strengths,
|
|
_select_live_hard_filter_symbols,
|
|
_weekly_macro_filter_state,
|
|
)
|
|
from strategy32.research.soft_router import (
|
|
CashOverlayCandidate,
|
|
MacroScaleSpec,
|
|
SoftRouterCandidate,
|
|
compose_cash_overlay_curve,
|
|
compose_soft_router_curve,
|
|
)
|
|
from strategy32.routing.router import Strategy32Router
|
|
from strategy32.universe import filter_momentum_frame, limit_correlated_symbols, rank_momentum_universe, score_momentum_universe, select_dynamic_universe, select_strategic_universe
|
|
|
|
|
|
def make_bundle(bars: int = 260) -> MarketDataBundle:
|
|
timestamps = pd.date_range("2025-01-01", periods=bars, freq="4h", tz="UTC")
|
|
prices = {
|
|
"BTC": pd.DataFrame(
|
|
{
|
|
"timestamp": timestamps,
|
|
"open": [100.0 + i * 0.20 for i in range(bars)],
|
|
"high": [100.3 + i * 0.20 for i in range(bars)],
|
|
"low": [99.8 + i * 0.20 for i in range(bars)],
|
|
"close": [100.0 + i * 0.20 for i in range(bars)],
|
|
"volume": [2_000_000.0] * bars,
|
|
}
|
|
),
|
|
"ETH": pd.DataFrame(
|
|
{
|
|
"timestamp": timestamps,
|
|
"open": [50.0 + i * 0.25 for i in range(bars)],
|
|
"high": [50.3 + i * 0.25 for i in range(bars)],
|
|
"low": [49.8 + i * 0.25 for i in range(bars)],
|
|
"close": [50.0 + i * 0.25 for i in range(bars)],
|
|
"volume": [2_500_000.0] * bars,
|
|
}
|
|
),
|
|
"SOL": pd.DataFrame(
|
|
{
|
|
"timestamp": timestamps,
|
|
"open": [20.0 + i * 0.18 for i in range(bars)],
|
|
"high": [20.2 + i * 0.18 for i in range(bars)],
|
|
"low": [19.9 + i * 0.18 for i in range(bars)],
|
|
"close": [20.0 + i * 0.18 for i in range(bars)],
|
|
"volume": [1_800_000.0] * bars,
|
|
}
|
|
),
|
|
}
|
|
funding = {
|
|
"ETH": pd.DataFrame(
|
|
{
|
|
"timestamp": timestamps,
|
|
"funding_rate": [0.00015] * bars,
|
|
"basis": [0.0100 - i * 0.00001 for i in range(bars)],
|
|
}
|
|
),
|
|
"SOL": pd.DataFrame(
|
|
{
|
|
"timestamp": timestamps,
|
|
"funding_rate": [0.00012] * bars,
|
|
"basis": [0.0090 - i * 0.00001 for i in range(bars)],
|
|
}
|
|
),
|
|
}
|
|
return MarketDataBundle(prices=prices, funding=funding)
|
|
|
|
|
|
def make_execution_prices(bundle: MarketDataBundle, *, blocked_symbols: set[str] | None = None) -> dict[str, pd.DataFrame]:
|
|
blocked_symbols = blocked_symbols or set()
|
|
start = pd.Timestamp(bundle.prices["BTC"]["timestamp"].iloc[0])
|
|
end = pd.Timestamp(bundle.prices["BTC"]["timestamp"].iloc[-1])
|
|
timestamps = pd.date_range(start, end, freq="1h", tz="UTC")
|
|
prices: dict[str, pd.DataFrame] = {}
|
|
for symbol in bundle.prices:
|
|
base = 100.0 if symbol == "BTC" else 50.0
|
|
if symbol in blocked_symbols:
|
|
closes = [base - i * 0.15 for i in range(len(timestamps))]
|
|
else:
|
|
closes = [base + i * 0.08 for i in range(len(timestamps))]
|
|
prices[symbol] = pd.DataFrame(
|
|
{
|
|
"timestamp": timestamps,
|
|
"open": closes,
|
|
"high": [value + 0.2 for value in closes],
|
|
"low": [value - 0.2 for value in closes],
|
|
"close": closes,
|
|
"volume": [1_000_000.0] * len(timestamps),
|
|
}
|
|
)
|
|
return prices
|
|
|
|
|
|
class Strategy32Tests(unittest.TestCase):
|
|
def test_router_disables_spread(self) -> None:
|
|
decision = Strategy32Router(Strategy32Config().budgets).decide(Regime.STRONG_UP)
|
|
self.assertEqual(decision.spread_budget_pct, 0.0)
|
|
self.assertGreater(decision.momentum_budget_pct, 0.0)
|
|
|
|
def test_dynamic_universe_picks_highest_volume_symbols(self) -> None:
|
|
bundle = make_bundle()
|
|
bundle.prices["SOL"]["volume"] = 100_000.0
|
|
selected = select_dynamic_universe(
|
|
bundle.prices,
|
|
timestamp=bundle.prices["BTC"]["timestamp"].iloc[-1],
|
|
min_history_bars=120,
|
|
lookback_bars=30,
|
|
max_symbols=1,
|
|
min_avg_dollar_volume=1_000_000.0,
|
|
)
|
|
self.assertEqual(selected, ["ETH"])
|
|
|
|
def test_dynamic_universe_supports_unlimited_selection(self) -> None:
|
|
bundle = make_bundle()
|
|
selected = select_dynamic_universe(
|
|
bundle.prices,
|
|
timestamp=bundle.prices["BTC"]["timestamp"].iloc[-1],
|
|
min_history_bars=120,
|
|
lookback_bars=30,
|
|
max_symbols=0,
|
|
min_avg_dollar_volume=1_000_000.0,
|
|
)
|
|
self.assertEqual(selected, ["ETH", "SOL"])
|
|
|
|
def test_live_hard_filter_uses_daily_cut_before_ranking(self) -> None:
|
|
bundle = make_bundle()
|
|
bundle.prices["SOL"]["volume"] = 10_000.0
|
|
selected = _select_live_hard_filter_symbols(
|
|
bundle.prices,
|
|
timestamp=bundle.prices["BTC"]["timestamp"].iloc[-1],
|
|
config=LiveMonitorConfig(
|
|
hard_filter_min_history_bars=120,
|
|
hard_filter_lookback_bars=30,
|
|
hard_filter_min_avg_dollar_volume=1_000_000.0,
|
|
),
|
|
)
|
|
self.assertEqual(selected, ["BTC", "ETH"])
|
|
|
|
def test_backtester_daily_hard_filter_cache_sticks_within_day(self) -> None:
|
|
bundle = make_bundle(bars=12)
|
|
bundle.prices["ETH"]["volume"] = 60_000.0
|
|
bundle.prices["SOL"]["volume"] = 1.0
|
|
bundle.prices["SOL"].loc[2, "volume"] = 120_000.0
|
|
bundle.prices["SOL"].loc[5, "volume"] = 1.0
|
|
config = Strategy32Config(
|
|
hard_filter_refresh_cadence="1d",
|
|
hard_filter_min_history_bars=1,
|
|
hard_filter_lookback_bars=1,
|
|
hard_filter_min_avg_dollar_volume=1_000_000.0,
|
|
)
|
|
backtester = Strategy32MomentumCarryBacktester(config, bundle)
|
|
first_ts = bundle.prices["BTC"]["timestamp"].iloc[2]
|
|
later_same_day_ts = bundle.prices["BTC"]["timestamp"].iloc[5]
|
|
initial = backtester._hard_filter_symbols(first_ts, min_history_bars=1)
|
|
same_day = backtester._hard_filter_symbols(later_same_day_ts, min_history_bars=1)
|
|
self.assertIn("SOL", initial)
|
|
self.assertIn("SOL", same_day)
|
|
|
|
def test_backtester_intraday_hard_filter_changes_without_daily_cache(self) -> None:
|
|
bundle = make_bundle(bars=12)
|
|
bundle.prices["ETH"]["volume"] = 60_000.0
|
|
bundle.prices["SOL"]["volume"] = 1.0
|
|
bundle.prices["SOL"].loc[2, "volume"] = 120_000.0
|
|
bundle.prices["SOL"].loc[5, "volume"] = 1.0
|
|
config = Strategy32Config(
|
|
hard_filter_refresh_cadence="4h",
|
|
hard_filter_min_history_bars=1,
|
|
hard_filter_lookback_bars=1,
|
|
hard_filter_min_avg_dollar_volume=1_000_000.0,
|
|
)
|
|
backtester = Strategy32MomentumCarryBacktester(config, bundle)
|
|
first_ts = bundle.prices["BTC"]["timestamp"].iloc[2]
|
|
later_same_day_ts = bundle.prices["BTC"]["timestamp"].iloc[5]
|
|
initial = backtester._hard_filter_symbols(first_ts, min_history_bars=1)
|
|
same_day = backtester._hard_filter_symbols(later_same_day_ts, min_history_bars=1)
|
|
self.assertIn("SOL", initial)
|
|
self.assertNotIn("SOL", same_day)
|
|
|
|
def test_weekly_macro_filter_flags_downtrend_as_risk_off(self) -> None:
|
|
timestamps = pd.date_range("2024-01-01", periods=400, freq="1D", tz="UTC")
|
|
prices = {
|
|
"BTC": pd.DataFrame(
|
|
{
|
|
"timestamp": timestamps,
|
|
"open": [300.0 - i * 0.4 for i in range(len(timestamps))],
|
|
"high": [301.0 - i * 0.4 for i in range(len(timestamps))],
|
|
"low": [299.0 - i * 0.4 for i in range(len(timestamps))],
|
|
"close": [300.0 - i * 0.4 for i in range(len(timestamps))],
|
|
"volume": [1_000_000.0] * len(timestamps),
|
|
}
|
|
)
|
|
}
|
|
macro = _weekly_macro_filter_state(
|
|
prices,
|
|
timestamp=timestamps[-1],
|
|
config=LiveMonitorConfig(macro_filter_fast_weeks=10, macro_filter_slow_weeks=30),
|
|
)
|
|
self.assertFalse(macro["risk_on"])
|
|
|
|
def test_weekly_macro_filter_removes_tradeable_core_targets_when_risk_off(self) -> None:
|
|
filtered = _apply_weekly_macro_filter(
|
|
[
|
|
{"instrument": "perp:ETH", "tradeable": True, "source": "core", "weight": 0.3},
|
|
{"instrument": "carry:SOL", "tradeable": False, "source": "core", "weight": 0.1},
|
|
],
|
|
macro_state={"risk_on": False},
|
|
)
|
|
self.assertEqual(filtered, [{"instrument": "carry:SOL", "tradeable": False, "source": "core", "weight": 0.1}])
|
|
|
|
def test_execution_refinement_blocks_extended_entry(self) -> None:
|
|
bars = 64
|
|
timestamps = pd.date_range("2025-01-01", periods=bars, freq="1h", tz="UTC")
|
|
prices = {
|
|
"ETH": pd.DataFrame(
|
|
{
|
|
"timestamp": timestamps,
|
|
"open": [100.0 + i * 0.4 for i in range(bars)],
|
|
"high": [100.3 + i * 0.4 for i in range(bars)],
|
|
"low": [99.7 + i * 0.4 for i in range(bars)],
|
|
"close": [100.0 + i * 0.4 for i in range(bars - 1)] + [140.0],
|
|
"volume": [1_000_000.0] * bars,
|
|
}
|
|
)
|
|
}
|
|
states = _execution_refinement_states(
|
|
prices,
|
|
timestamp=timestamps[-1],
|
|
config=LiveMonitorConfig(
|
|
execution_refinement_lookback_bars=48,
|
|
execution_refinement_fast_ema=8,
|
|
execution_refinement_slow_ema=21,
|
|
execution_refinement_scale_down_gap=0.008,
|
|
execution_refinement_max_chase_gap=0.018,
|
|
execution_refinement_max_recent_return=0.03,
|
|
execution_refinement_scale_down_factor=0.5,
|
|
),
|
|
)
|
|
self.assertEqual(states["ETH"]["action"], "block")
|
|
|
|
def test_refined_execution_targets_scale_positive_perp_only(self) -> None:
|
|
refined = _refine_execution_targets(
|
|
[
|
|
{"instrument": "perp:ETH", "tradeable": True, "weight": 0.4},
|
|
{"instrument": "perp:BTC", "tradeable": True, "weight": -0.2},
|
|
],
|
|
refinement_states={"ETH": {"action": "scale_down", "scale": 0.5, "reason": "slightly_extended"}},
|
|
)
|
|
self.assertEqual(refined[0]["weight"], 0.2)
|
|
self.assertEqual(refined[0]["desired_weight"], 0.4)
|
|
self.assertEqual(refined[1]["weight"], -0.2)
|
|
|
|
def test_executor_entry_only_refinement_does_not_force_close_existing_position(self) -> None:
|
|
class FakeClient:
|
|
def __init__(self) -> None:
|
|
self.orders: list[dict[str, object]] = []
|
|
|
|
def get_balance(self):
|
|
return [{"asset": "USDT", "balance": "1000"}]
|
|
|
|
def get_position_risk(self):
|
|
return [{"symbol": "ETHUSDT", "positionAmt": "1", "markPrice": "100", "entryPrice": "100", "notional": "100", "unRealizedProfit": "0"}]
|
|
|
|
def get_ticker_price(self, symbol):
|
|
return {"symbol": symbol, "price": "100"}
|
|
|
|
def get_exchange_info(self):
|
|
return {
|
|
"symbols": [
|
|
{
|
|
"symbol": "ETHUSDT",
|
|
"baseAsset": "ETH",
|
|
"quoteAsset": "USDT",
|
|
"filters": [
|
|
{"filterType": "LOT_SIZE", "stepSize": "0.001", "minQty": "0.001"},
|
|
{"filterType": "MIN_NOTIONAL", "notional": "5"},
|
|
],
|
|
}
|
|
]
|
|
}
|
|
|
|
def set_leverage(self, symbol, leverage):
|
|
return {"symbol": symbol, "leverage": leverage}
|
|
|
|
def place_market_order(self, **kwargs):
|
|
self.orders.append(kwargs)
|
|
return {"status": "FILLED", **kwargs}
|
|
|
|
executor = LiveFuturesExecutor(
|
|
FakeClient(),
|
|
LiveExecutionConfig(
|
|
enabled=True,
|
|
leverage=2,
|
|
min_target_notional_usd=25.0,
|
|
min_rebalance_notional_usd=10.0,
|
|
close_orphan_positions=True,
|
|
entry_only_refinement=True,
|
|
),
|
|
)
|
|
result = executor.reconcile(
|
|
{
|
|
"generated_at": "2026-03-16T00:00:00Z",
|
|
"universe": {"quote_by_symbol": {"ETH": "USDT"}},
|
|
"execution_targets": [
|
|
{
|
|
"instrument": "perp:ETH",
|
|
"tradeable": True,
|
|
"weight": 0.0,
|
|
"desired_weight": 0.4,
|
|
"refinement_action": "block",
|
|
"refinement_reason": "too_extended",
|
|
}
|
|
],
|
|
}
|
|
)
|
|
self.assertEqual(result.orders, [])
|
|
|
|
def test_capital_summary_extracts_usdt_and_usdc(self) -> None:
|
|
summary = _capital_summary(
|
|
{
|
|
"balances": [
|
|
{"asset": "USDT", "balance": "1200.5"},
|
|
{"asset": "USDC", "balance": "300.25"},
|
|
{"asset": "BTC", "balance": "0.1"},
|
|
]
|
|
}
|
|
)
|
|
self.assertEqual(summary, {"usdt": 1200.5, "usdc": 300.25, "total_quote": 1500.75})
|
|
|
|
def test_momentum_universe_prefers_stronger_symbol(self) -> None:
|
|
bundle = make_bundle()
|
|
ranked = rank_momentum_universe(
|
|
bundle.prices,
|
|
bundle.funding,
|
|
btc_symbol="BTC",
|
|
timestamp=bundle.prices["BTC"]["timestamp"].iloc[-1],
|
|
candidate_symbols=["ETH", "SOL"],
|
|
min_history_bars=120,
|
|
liquidity_lookback_bars=30,
|
|
short_lookback_bars=18,
|
|
long_lookback_bars=72,
|
|
overheat_funding_rate=0.00025,
|
|
max_symbols=2,
|
|
)
|
|
self.assertEqual(ranked[0], "SOL")
|
|
|
|
def test_momentum_quality_filter_drops_overheated_symbol(self) -> None:
|
|
bundle = make_bundle()
|
|
bundle.prices["PEPE"] = bundle.prices["ETH"].copy()
|
|
bundle.prices["PEPE"]["close"] = [10.0 + i * 0.80 for i in range(len(bundle.prices["PEPE"]))]
|
|
bundle.prices["PEPE"]["volume"] = [3_000_000.0] * len(bundle.prices["PEPE"])
|
|
bundle.funding["PEPE"] = pd.DataFrame(
|
|
{
|
|
"timestamp": bundle.prices["PEPE"]["timestamp"],
|
|
"funding_rate": [0.0008] * len(bundle.prices["PEPE"]),
|
|
"basis": [0.012] * len(bundle.prices["PEPE"]),
|
|
}
|
|
)
|
|
frame = score_momentum_universe(
|
|
bundle.prices,
|
|
bundle.funding,
|
|
btc_symbol="BTC",
|
|
timestamp=bundle.prices["BTC"]["timestamp"].iloc[-1],
|
|
candidate_symbols=["ETH", "PEPE"],
|
|
min_history_bars=120,
|
|
liquidity_lookback_bars=30,
|
|
short_lookback_bars=18,
|
|
long_lookback_bars=72,
|
|
overheat_funding_rate=0.00025,
|
|
)
|
|
filtered = filter_momentum_frame(
|
|
frame,
|
|
min_score=0.0,
|
|
min_relative_strength=-1.0,
|
|
min_7d_return=-1.0,
|
|
max_7d_return=0.35,
|
|
min_positive_bar_ratio=0.0,
|
|
max_short_volatility=1.0,
|
|
max_latest_funding_rate=0.00045,
|
|
max_beta=10.0,
|
|
)
|
|
self.assertIn("ETH", filtered["symbol"].tolist())
|
|
self.assertNotIn("PEPE", filtered["symbol"].tolist())
|
|
|
|
def test_strategic_universe_keeps_symbols_with_positive_edge(self) -> None:
|
|
bundle = make_bundle()
|
|
selected = select_strategic_universe(
|
|
bundle.prices,
|
|
bundle.funding,
|
|
btc_symbol="BTC",
|
|
timestamp=bundle.prices["BTC"]["timestamp"].iloc[-1],
|
|
min_history_bars=120,
|
|
lookback_bars=30,
|
|
min_avg_dollar_volume=1_000_000.0,
|
|
short_lookback_bars=18,
|
|
long_lookback_bars=72,
|
|
overheat_funding_rate=0.00025,
|
|
carry_lookback_bars=21,
|
|
carry_expected_horizon_bars=18,
|
|
carry_roundtrip_cost_pct=0.0020,
|
|
carry_basis_risk_multiplier=1.0,
|
|
momentum_min_score=0.0,
|
|
momentum_min_relative_strength=-1.0,
|
|
momentum_min_7d_return=-1.0,
|
|
momentum_max_7d_return=1.0,
|
|
momentum_min_positive_bar_ratio=0.0,
|
|
momentum_max_short_volatility=1.0,
|
|
momentum_max_latest_funding_rate=1.0,
|
|
momentum_max_beta=10.0,
|
|
carry_min_expected_edge=-1.0,
|
|
max_symbols=0,
|
|
)
|
|
self.assertIn("ETH", selected)
|
|
self.assertIn("SOL", selected)
|
|
|
|
def test_correlation_limit_drops_duplicate_path(self) -> None:
|
|
bundle = make_bundle()
|
|
bundle.prices["LINK"] = bundle.prices["ETH"].copy()
|
|
bundle.prices["LINK"]["close"] = [30.0 + i * 0.10 for i in range(len(bundle.prices["LINK"]))]
|
|
bundle.prices["SOL"]["close"] = [20.0 + i * 0.18 + (0.9 if i % 2 == 0 else -0.7) for i in range(len(bundle.prices["SOL"]))]
|
|
limited = limit_correlated_symbols(
|
|
bundle.prices,
|
|
timestamp=bundle.prices["BTC"]["timestamp"].iloc[-1],
|
|
candidate_symbols=["ETH", "LINK", "SOL"],
|
|
lookback_bars=36,
|
|
max_pairwise_correlation=0.80,
|
|
max_symbols=2,
|
|
)
|
|
self.assertEqual(limited, ["ETH", "SOL"])
|
|
|
|
def test_backtester_runs(self) -> None:
|
|
result = Strategy32Backtester(
|
|
Strategy32Config(
|
|
symbols=["BTC", "ETH", "SOL"],
|
|
momentum_min_history_bars=120,
|
|
momentum_max_7d_return=1.0,
|
|
momentum_min_positive_bar_ratio=0.0,
|
|
momentum_max_short_volatility=1.0,
|
|
momentum_max_beta=10.0,
|
|
momentum_max_latest_funding_rate=1.0,
|
|
),
|
|
make_bundle(),
|
|
).run()
|
|
self.assertGreater(result.total_trades, 0)
|
|
self.assertIn("momentum", result.engine_pnl)
|
|
|
|
def test_backtester_execution_refinement_blocks_entry_and_logs_rejection(self) -> None:
|
|
bundle = make_bundle()
|
|
bundle.prices = {"BTC": bundle.prices["BTC"], "ETH": bundle.prices["ETH"]}
|
|
bundle.funding = {"ETH": bundle.funding["ETH"]}
|
|
config = Strategy32Config(
|
|
symbols=["BTC", "ETH"],
|
|
budgets=Strategy32Budgets(
|
|
strong_up_carry=0.0,
|
|
up_carry=0.0,
|
|
sideways_carry=0.0,
|
|
down_carry=0.0,
|
|
strong_up_sideways=0.0,
|
|
up_sideways=0.0,
|
|
sideways_sideways=0.0,
|
|
down_sideways=0.0,
|
|
),
|
|
)
|
|
result = Strategy32Backtester(
|
|
config,
|
|
bundle,
|
|
execution_prices=make_execution_prices(bundle, blocked_symbols={"ETH"}),
|
|
).run(close_final_positions=False)
|
|
summary = result.metadata.get("rejection_summary", {})
|
|
self.assertGreater(summary.get("execution_refinement_blocked", 0), 0)
|
|
final_positions = result.metadata.get("final_positions", [])
|
|
self.assertTrue(all(position["engine"] != "momentum" for position in final_positions))
|
|
|
|
def test_backtester_rejection_logging_records_empty_universe(self) -> None:
|
|
bundle = make_bundle()
|
|
config = Strategy32Config(
|
|
symbols=["BTC", "ETH", "SOL"],
|
|
universe_min_avg_dollar_volume=1_000_000_000_000.0,
|
|
budgets=Strategy32Budgets(
|
|
strong_up_carry=0.0,
|
|
up_carry=0.0,
|
|
sideways_carry=0.0,
|
|
down_carry=0.0,
|
|
strong_up_sideways=0.0,
|
|
up_sideways=0.0,
|
|
sideways_sideways=0.0,
|
|
down_sideways=0.0,
|
|
),
|
|
)
|
|
result = Strategy32Backtester(config, bundle).run()
|
|
summary = result.metadata.get("rejection_summary", {})
|
|
self.assertGreater(summary.get("tradeable_universe_empty", 0), 0)
|
|
|
|
def test_backtester_liquidity_and_momentum_fallback_can_restore_candidates(self) -> None:
|
|
bundle = make_bundle()
|
|
config = Strategy32Config(
|
|
symbols=["BTC", "ETH", "SOL"],
|
|
momentum_min_score=10.0,
|
|
carry_min_expected_edge=10.0,
|
|
universe_fallback_min_avg_dollar_volume=1_000_000.0,
|
|
universe_fallback_top_n=2,
|
|
momentum_fallback_min_score=0.0,
|
|
momentum_fallback_min_relative_strength=-1.0,
|
|
momentum_fallback_min_7d_return=-1.0,
|
|
budgets=Strategy32Budgets(
|
|
strong_up_carry=0.0,
|
|
up_carry=0.0,
|
|
sideways_carry=0.0,
|
|
down_carry=0.0,
|
|
strong_up_sideways=0.0,
|
|
up_sideways=0.0,
|
|
sideways_sideways=0.0,
|
|
down_sideways=0.0,
|
|
),
|
|
)
|
|
result = Strategy32Backtester(
|
|
config,
|
|
bundle,
|
|
execution_prices=make_execution_prices(bundle),
|
|
).run(close_final_positions=False)
|
|
summary = result.metadata.get("rejection_summary", {})
|
|
self.assertGreater(summary.get("dynamic_universe_fallback_used", 0), 0)
|
|
self.assertGreater(summary.get("momentum_filter_fallback_used", 0), 0)
|
|
final_positions = result.metadata.get("final_positions", [])
|
|
self.assertTrue(any(position["engine"] == "momentum" for position in final_positions))
|
|
|
|
def test_trade_start_blocks_warmup_trades(self) -> None:
|
|
bundle = make_bundle()
|
|
trade_start = bundle.prices["BTC"]["timestamp"].iloc[-40]
|
|
result = Strategy32Backtester(
|
|
Strategy32Config(symbols=["BTC", "ETH", "SOL"]),
|
|
bundle,
|
|
trade_start=trade_start,
|
|
).run()
|
|
self.assertTrue(all(trade.entry_time >= trade_start for trade in result.trades))
|
|
|
|
def test_profile_helpers_select_expected_feature_flags(self) -> None:
|
|
default_cfg = build_strategy32_config(PROFILE_V7_DEFAULT)
|
|
self.assertFalse(default_cfg.enable_sideways_engine)
|
|
self.assertTrue(default_cfg.enable_strong_kill_switch)
|
|
self.assertTrue(default_cfg.enable_daily_trend_filter)
|
|
self.assertFalse(default_cfg.enable_expanded_hedge)
|
|
self.assertFalse(default_cfg.enable_max_holding_exit)
|
|
|
|
baseline_cfg = build_strategy32_config(PROFILE_V5_BASELINE)
|
|
self.assertTrue(baseline_cfg.enable_sideways_engine)
|
|
self.assertFalse(baseline_cfg.enable_strong_kill_switch)
|
|
self.assertFalse(baseline_cfg.enable_daily_trend_filter)
|
|
self.assertFalse(baseline_cfg.enable_expanded_hedge)
|
|
self.assertFalse(baseline_cfg.enable_max_holding_exit)
|
|
|
|
self.assertFalse(Strategy32Config().enable_sideways_engine)
|
|
self.assertTrue(Strategy32Config().enable_strong_kill_switch)
|
|
self.assertTrue(Strategy32Config().enable_daily_trend_filter)
|
|
|
|
def test_binance_history_fetch_uses_stale_cache_on_http_error(self) -> None:
|
|
url = "https://example.com/test.json"
|
|
|
|
class FakeResponse:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def read(self) -> bytes:
|
|
return json.dumps({"ok": True}).encode("utf-8")
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with (
|
|
patch.object(binance_history, "DEFAULT_CACHE_DIR", Path(tmpdir)),
|
|
patch.object(binance_history, "DEFAULT_HTTP_RETRIES", 1),
|
|
patch.object(binance_history, "urlopen", side_effect=[FakeResponse(), HTTPError(url, 418, "blocked", hdrs=None, fp=None)]),
|
|
):
|
|
first = binance_history._fetch_json(url, ttl_seconds=0)
|
|
second = binance_history._fetch_json(url, ttl_seconds=0)
|
|
self.assertEqual(first, {"ok": True})
|
|
self.assertEqual(second, {"ok": True})
|
|
|
|
def test_soft_router_weights_remain_bounded(self) -> None:
|
|
timestamps = list(pd.date_range("2025-01-01", periods=4, freq="4h", tz="UTC"))
|
|
score_frame = pd.DataFrame(
|
|
{
|
|
"timestamp": timestamps[:-1],
|
|
"core_score": [0.8, 0.2, 0.1],
|
|
"panic_score": [0.0, 0.6, 0.1],
|
|
"choppy_score": [0.1, 0.7, 0.8],
|
|
"distribution_score": [0.1, 0.2, 0.9],
|
|
}
|
|
)
|
|
returns = pd.Series([0.01, -0.02, 0.03], index=pd.DatetimeIndex(timestamps[1:], name="timestamp"))
|
|
curve, weights = compose_soft_router_curve(
|
|
timestamps=timestamps,
|
|
score_frame=score_frame,
|
|
core_returns=returns,
|
|
cap_returns=returns * 0.5,
|
|
chop_returns=returns * 0.25,
|
|
dist_returns=returns * -0.10,
|
|
candidate=SoftRouterCandidate(
|
|
regime_profile="base",
|
|
core_filter="overheat_tolerant",
|
|
cap_engine="cap_btc_rebound",
|
|
chop_engine="chop_inverse_carry",
|
|
dist_engine="dist_inverse_carry_strict",
|
|
core_floor=0.1,
|
|
cap_max_weight=0.4,
|
|
chop_max_weight=0.3,
|
|
dist_max_weight=0.2,
|
|
chop_blend_floor=0.15,
|
|
),
|
|
)
|
|
self.assertEqual(len(curve), 4)
|
|
total_weights = weights[["core_weight", "cap_weight", "chop_weight", "dist_weight", "cash_weight"]].sum(axis=1)
|
|
self.assertTrue(((total_weights - 1.0).abs() < 1e-9).all())
|
|
|
|
def test_cash_overlay_respects_core_cash_budget(self) -> None:
|
|
timestamps = list(pd.date_range("2025-01-01", periods=4, freq="4h", tz="UTC"))
|
|
score_frame = pd.DataFrame(
|
|
{
|
|
"timestamp": timestamps[:-1],
|
|
"core_score": [0.2, 0.8, 0.1],
|
|
"panic_score": [0.8, 0.1, 0.0],
|
|
"choppy_score": [0.6, 0.7, 0.2],
|
|
"distribution_score": [0.2, 0.9, 0.8],
|
|
}
|
|
)
|
|
core_returns = pd.Series([0.01, 0.00, 0.02], index=pd.DatetimeIndex(timestamps[1:], name="timestamp"))
|
|
overlay_returns = pd.Series([0.02, 0.01, -0.01], index=pd.DatetimeIndex(timestamps[1:], name="timestamp"))
|
|
core_exposure_frame = pd.DataFrame(
|
|
{
|
|
"timestamp": timestamps[:-1],
|
|
"cash_pct": [0.50, 0.30, 0.80],
|
|
}
|
|
)
|
|
curve, weights = compose_cash_overlay_curve(
|
|
timestamps=timestamps,
|
|
score_frame=score_frame,
|
|
core_returns=core_returns,
|
|
core_exposure_frame=core_exposure_frame,
|
|
cap_returns=overlay_returns,
|
|
chop_returns=overlay_returns,
|
|
dist_returns=overlay_returns,
|
|
candidate=CashOverlayCandidate(
|
|
regime_profile="loose_positive",
|
|
core_filter="overheat_tolerant",
|
|
cap_engine="cap_btc_rebound",
|
|
chop_engine="chop_inverse_carry_strict",
|
|
dist_engine="dist_inverse_carry_strict",
|
|
cap_cash_weight=0.80,
|
|
chop_cash_weight=0.80,
|
|
dist_cash_weight=0.80,
|
|
cap_threshold=0.20,
|
|
chop_threshold=0.20,
|
|
dist_threshold=0.20,
|
|
core_block_threshold=0.50,
|
|
),
|
|
)
|
|
self.assertEqual(len(curve), 4)
|
|
self.assertTrue((weights["overlay_total"] <= weights["core_cash_pct"] + 1e-9).all())
|
|
|
|
def test_cash_overlay_macro_scale_reduces_core_return_and_frees_cash(self) -> None:
|
|
timestamps = list(pd.date_range("2025-01-01", periods=4, freq="4h", tz="UTC"))
|
|
score_frame = pd.DataFrame(
|
|
{
|
|
"timestamp": timestamps[:-1],
|
|
"core_score": [0.1, 0.1, 0.1],
|
|
"panic_score": [0.0, 0.0, 0.0],
|
|
"choppy_score": [0.0, 0.0, 0.0],
|
|
"distribution_score": [0.0, 0.0, 0.0],
|
|
}
|
|
)
|
|
core_returns = pd.Series([0.10, 0.10, 0.10], index=pd.DatetimeIndex(timestamps[1:], name="timestamp"))
|
|
core_exposure_frame = pd.DataFrame(
|
|
{
|
|
"timestamp": timestamps[:-1],
|
|
"cash_pct": [0.20, 0.20, 0.20],
|
|
}
|
|
)
|
|
curve, weights = compose_cash_overlay_curve(
|
|
timestamps=timestamps,
|
|
score_frame=score_frame,
|
|
core_returns=core_returns,
|
|
core_exposure_frame=core_exposure_frame,
|
|
cap_returns=core_returns * 0.0,
|
|
chop_returns=core_returns * 0.0,
|
|
dist_returns=core_returns * 0.0,
|
|
candidate=CashOverlayCandidate(
|
|
regime_profile="loose_positive",
|
|
core_filter="overheat_tolerant",
|
|
cap_engine="cap_btc_rebound",
|
|
chop_engine="chop_inverse_carry_strict",
|
|
dist_engine="dist_inverse_carry_strict",
|
|
cap_cash_weight=0.0,
|
|
chop_cash_weight=0.0,
|
|
dist_cash_weight=0.0,
|
|
cap_threshold=0.20,
|
|
chop_threshold=0.20,
|
|
dist_threshold=0.20,
|
|
core_block_threshold=0.50,
|
|
),
|
|
macro_scale_map=pd.Series(
|
|
[0.50, 0.50, 0.50],
|
|
index=pd.DatetimeIndex(timestamps[:-1], name="timestamp"),
|
|
dtype=float,
|
|
),
|
|
)
|
|
self.assertAlmostEqual(float(weights["macro_scale"].iloc[0]), 0.50)
|
|
self.assertAlmostEqual(float(weights["core_cash_pct"].iloc[0]), 0.60)
|
|
self.assertAlmostEqual(float(curve.iloc[-1]), 1000.0 * (1.05 ** 3), places=6)
|
|
|
|
def test_expand_core_targets_adds_btc_hedge(self) -> None:
|
|
targets = _expand_core_targets(
|
|
[
|
|
{
|
|
"engine": "momentum",
|
|
"symbol": "ETH",
|
|
"value": 250.0,
|
|
"meta": {"hedge_ratio": 0.4},
|
|
},
|
|
{
|
|
"engine": "carry",
|
|
"symbol": "SOL",
|
|
"value": 100.0,
|
|
"meta": {},
|
|
},
|
|
],
|
|
final_equity=1000.0,
|
|
)
|
|
by_instrument = {row["instrument"]: row for row in targets}
|
|
self.assertAlmostEqual(float(by_instrument["perp:ETH"]["weight"]), 0.25)
|
|
self.assertAlmostEqual(float(by_instrument["perp:BTC"]["weight"]), -0.10)
|
|
self.assertFalse(bool(by_instrument["carry:SOL"]["tradeable"]))
|
|
|
|
def test_overlay_signal_strengths_block_core_scores(self) -> None:
|
|
signals = _overlay_signal_strengths(
|
|
BEST_CASH_OVERLAY,
|
|
{
|
|
"core_score": 0.90,
|
|
"panic_score": 0.50,
|
|
"choppy_score": 0.80,
|
|
"distribution_score": 0.90,
|
|
},
|
|
)
|
|
self.assertGreater(signals["cap_signal"], 0.0)
|
|
self.assertLess(signals["chop_signal"], 0.25)
|
|
self.assertLess(signals["dist_signal"], 0.35)
|
|
|
|
def test_combine_targets_aggregates_same_instrument(self) -> None:
|
|
combined = _combine_targets(
|
|
[{"instrument": "perp:BTC", "weight": -0.10, "tradeable": True, "source": "core", "note": "hedge"}],
|
|
[{"instrument": "perp:BTC", "weight": -0.05, "tradeable": True, "source": "overlay", "note": "cap"}],
|
|
equity=1000.0,
|
|
)
|
|
self.assertEqual(len(combined), 1)
|
|
self.assertAlmostEqual(float(combined[0]["weight"]), -0.15)
|
|
self.assertAlmostEqual(float(combined[0]["notional_usd"]), -150.0)
|
|
|
|
def test_completed_bar_time_aligns_to_4h(self) -> None:
|
|
ts = pd.Timestamp("2026-03-16 09:17:00+00:00")
|
|
self.assertEqual(_completed_bar_time(ts, "4h"), pd.Timestamp("2026-03-16 08:00:00+00:00"))
|
|
|
|
def test_heartbeat_slot_uses_half_hour_boundaries(self) -> None:
|
|
self.assertEqual(_heartbeat_slot(pd.Timestamp("2026-03-16 09:17:00+00:00")), (2026, 3, 16, 9, 0))
|
|
self.assertEqual(_heartbeat_slot(pd.Timestamp("2026-03-16 09:31:00+00:00")), (2026, 3, 16, 9, 30))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|