from __future__ import annotations import uuid from dataclasses import asdict, dataclass from decimal import Decimal from typing import Any from src.execution.order_intent import OrderIntent from src.market.trend_retest_features import TrendRetestFeatures from src.strategies.v78_research_archetypes import external_entry_blockers STRATEGY_ID = "research_v78_2_confirmed_range_reversion" STRATEGY_VERSION = "v78.2.0" RISK_BUDGET_USD = Decimal("0.075") MAX_NOTIONAL_USD = Decimal("10") MAKER_FEE_RATE = Decimal("0.00015") TAKER_FEE_RATE = Decimal("0.00045") MIN_EXIT_SLIPPAGE_PCT = Decimal("0.02") @dataclass(frozen=True) class TrueRangeDecision: allowed: bool regime: str confidence: Decimal blockers: tuple[str, ...] def to_dict(self) -> dict[str, Any]: row = asdict(self) row["confidence"] = str(self.confidence) row["blockers"] = list(self.blockers) return row @dataclass(frozen=True) class RangeEntryDecision: action: str side: str | None blockers: tuple[str, ...] reason: str = "confirmed_band_reentry_to_vwap" strategy_family: str = "confirmed_range_reversion_v2" research_only: bool = True live_order_allowed: bool = False mainnet_signed_action: bool = False source_attribution: tuple[str, ...] = () def to_dict(self) -> dict[str, Any]: row = asdict(self) row["blockers"] = list(self.blockers) row["source_attribution"] = list(self.source_attribution) return row def classify_true_range(features: dict[str, TrendRetestFeatures]) -> TrueRangeDecision: primary = [features.get("BTC"), features.get("ETH")] if any(feature is None for feature in primary): return TrueRangeDecision(False, "no_trade", Decimal("1"), ("btc_eth_range_inputs_missing",)) rows = [feature for feature in primary if feature is not None] blockers: list[str] = [] avg_adx = sum((row.adx_15m for row in rows), Decimal("0")) / Decimal(len(rows)) avg_1h_spread = sum((abs(row.sma_1h_spread_pct) for row in rows), Decimal("0")) / Decimal(len(rows)) avg_4h_spread = sum((abs(row.sma_4h_spread_pct) for row in rows), Decimal("0")) / Decimal(len(rows)) avg_bb_width = sum((row.bb_width_pct for row in rows), Decimal("0")) / Decimal(len(rows)) if avg_adx > Decimal("22") or any(row.adx_15m > Decimal("25") for row in rows): blockers.append("primary_adx_not_range") if avg_1h_spread > Decimal("0.8") or any(abs(row.sma_1h_spread_pct) > Decimal("1.2") for row in rows): blockers.append("primary_1h_trend_not_flat") if avg_4h_spread > Decimal("1.5") or any(abs(row.sma_4h_spread_pct) > Decimal("2") for row in rows): blockers.append("primary_4h_trend_not_flat") if avg_bb_width <= Decimal("0.20") or avg_bb_width > Decimal("4"): blockers.append("primary_bollinger_width_not_range") if any(not row.data_quality_allowed or row.vwap_15m <= 0 or row.bb_lower_15m <= 0 or row.bb_upper_15m <= 0 for row in rows): blockers.append("primary_range_features_unreliable") return TrueRangeDecision(not blockers, "confirmed_range" if not blockers else "no_trade", Decimal("0.9") if not blockers else Decimal("0.2"), tuple(blockers)) def _external_blockers(feature: TrendRetestFeatures, side: str, external: dict[str, Any] | None) -> tuple[list[str], tuple[str, ...]]: blockers, attribution = external_entry_blockers(feature.coin, side, external) if feature.coin == "HYPE": blockers = [value for value in blockers if value not in {"crowding_context_missing"}] return blockers, attribution def confirmed_range_reversion(feature: TrendRetestFeatures, *, range_allowed: bool, external: dict[str, Any] | None) -> RangeEntryDecision: blockers: list[str] = [] attribution: tuple[str, ...] = () if not range_allowed: blockers.append("true_range_regime_not_confirmed") long_shape = feature.low_15m <= feature.bb_lower_15m and feature.close_15m > feature.bb_lower_15m and feature.close_15m < feature.vwap_15m short_shape = feature.high_15m >= feature.bb_upper_15m and feature.close_15m < feature.bb_upper_15m and feature.close_15m > feature.vwap_15m long_setup = long_shape and feature.rsi_15m <= Decimal("40") short_setup = short_shape and feature.rsi_15m >= Decimal("60") side = "long" if long_setup else "short" if short_setup else None if side is None: if feature.rsi_15m <= Decimal("40"): blockers.append("lower_band_reentry_missing") elif feature.rsi_15m >= Decimal("60"): blockers.append("upper_band_reentry_missing") else: blockers.append("rsi_not_at_range_extreme") if feature.adx_15m > Decimal("25"): blockers.append("coin_adx_too_trending") if feature.bb_width_pct <= Decimal("0.20") or feature.bb_width_pct > Decimal("5"): blockers.append("coin_bollinger_width_not_range") if feature.volume_ratio > Decimal("1.8"): blockers.append("volume_breakout_do_not_fade") if feature.best_bid <= 0 or feature.best_ask <= feature.best_bid: blockers.append("live_l2_quotes_missing_or_crossed") else: live_mid = (feature.best_bid + feature.best_ask) / Decimal("2") dislocation = abs(live_mid - feature.close_15m) / feature.close_15m * Decimal("100") if feature.close_15m > 0 else Decimal("999") if dislocation > Decimal("0.25"): blockers.append("live_mid_dislocated_from_signal_close") if side == "long" and (feature.sma_4h_spread_pct < Decimal("-1.5") or feature.sma_1h_spread_pct < Decimal("-0.8")): blockers.append("higher_timeframe_bearish_blocks_long") if side == "short" and (feature.sma_4h_spread_pct > Decimal("1.5") or feature.sma_1h_spread_pct > Decimal("0.8")): blockers.append("higher_timeframe_bullish_blocks_short") if not feature.data_quality_allowed: blockers.extend(feature.data_quality_reasons) if side: external_blockers, attribution = _external_blockers(feature, side, external) blockers.extend(external_blockers) return RangeEntryDecision("paper_candidate" if side and not blockers else "no_trade", side, tuple(dict.fromkeys(blockers)), source_attribution=attribution) def build_maker_intent(feature: TrendRetestFeatures, decision: RangeEntryDecision) -> OrderIntent | None: if decision.action != "paper_candidate" or decision.side not in {"long", "short"}: return None entry = feature.best_bid if decision.side == "long" else feature.best_ask buffer_pct = max(feature.atr_pct * Decimal("0.25"), Decimal("0.15")) stop = feature.low_15m * (Decimal("1") - buffer_pct / Decimal("100")) if decision.side == "long" else feature.high_15m * (Decimal("1") + buffer_pct / Decimal("100")) if (decision.side == "long" and feature.vwap_15m <= entry) or (decision.side == "short" and feature.vwap_15m >= entry): return None stop_pct = abs(entry - stop) / entry * Decimal("100") if stop_pct <= 0: return None max_size_by_notional = MAX_NOTIONAL_USD / entry exit_impact = feature.sell_impact_1k_pct if decision.side == "long" else feature.buy_impact_1k_pct exit_slippage_pct = max(MIN_EXIT_SLIPPAGE_PCT, feature.spread_pct / Decimal("2"), exit_impact) all_in_risk_per_unit = abs(entry - stop) + entry * MAKER_FEE_RATE + stop * TAKER_FEE_RATE + stop * exit_slippage_pct / Decimal("100") max_size_by_risk = RISK_BUDGET_USD / all_in_risk_per_unit size = min(max_size_by_notional, max_size_by_risk) * Decimal("0.999999") notional = size * entry return OrderIntent( strategy_id=STRATEGY_ID, symbol=f"{feature.coin}/USDC:USDC", coin=feature.coin, side="buy" if decision.side == "long" else "sell", reduce_only=False, order_type="limit", tif="Alo", size=size, price=entry, trigger_price=None, stop_loss=stop, take_profit=feature.vwap_15m, client_order_id="paper-v782-" + uuid.uuid4().hex, reason=decision.reason, risk_usd=all_in_risk_per_unit * size, estimated_notional_usd=notional, )