from __future__ import annotations from dataclasses import dataclass, field from decimal import Decimal from typing import Any, Literal from src.execution.order_intent import OrderIntent StrategyFamily = Literal[ "v76_anti_chase", "market_regime", "copy", "copy_confirmation", "research_sampler", ] @dataclass(frozen=True) class MarketRegimeContext: btc_trend: str eth_trend: str choppy: bool = False @property def long_allowed(self) -> bool: return not self.choppy and self.btc_trend == "bullish" and self.eth_trend == "bullish" @property def short_allowed(self) -> bool: return not self.choppy and self.btc_trend == "bearish" and self.eth_trend == "bearish" @dataclass(frozen=True) class CopyObservationEvidence: cycles_completed: int expected_cycles: int allowed_signals: int shadow_decisions: int data_quality_ok: bool @property def complete(self) -> bool: return self.expected_cycles > 0 and self.cycles_completed >= self.expected_cycles @dataclass(frozen=True) class PromotionEvidence: closed_trades: int total_net_pnl: Decimal last_20_net_pnl: Decimal win_rate_pct: Decimal profit_factor: Decimal top_coin_share_pct: Decimal short_scalp_share_pct: Decimal max_drawdown_usd: Decimal replay_only: bool = False simulator_sanity_ok: bool = True @dataclass(frozen=True) class StrategyCandidate: strategy_id: str family: StrategyFamily intent: OrderIntent | None promotion: PromotionEvidence expected_move_vs_cost: Decimal = Decimal("0") anti_chase_ok: bool = False retest_confirmed: bool = False copy_confirmation_ok: bool = False max_effective_leverage: Decimal = Decimal("1") max_open_positions: int = 1 @dataclass(frozen=True) class GlobalStrategyContext: market_regime: MarketRegimeContext copy_observation: CopyObservationEvidence reconcile_clean: bool stops_confirmed: bool alerts_confirmed: bool kill_switch_active: bool live_entries_blocked: bool = True @dataclass(frozen=True) class GlobalStrategyDecision: allowed: bool status: str recommended_mode: str reasons: tuple[str, ...] = () risk_limits: dict[str, Any] = field(default_factory=dict) def _promotion_blockers(evidence: PromotionEvidence) -> list[str]: reasons: list[str] = [] if evidence.closed_trades < 30: reasons.append("sample_too_small") if evidence.total_net_pnl <= 0: reasons.append("paper_total_pnl_not_positive") if evidence.last_20_net_pnl <= 0: reasons.append("last_20_not_positive") if evidence.win_rate_pct < Decimal("45") and evidence.profit_factor < Decimal("1.25"): reasons.append("winrate_and_profit_factor_below_gate") if evidence.top_coin_share_pct > Decimal("35"): reasons.append("coin_concentration") if evidence.short_scalp_share_pct > Decimal("25"): reasons.append("short_scalp_dominance") if evidence.replay_only: reasons.append("replay_only_not_promotable") if not evidence.simulator_sanity_ok: reasons.append("simulator_sanity_failed") return reasons def evaluate_global_strategy(candidate: StrategyCandidate, context: GlobalStrategyContext) -> GlobalStrategyDecision: """Evaluate whether a strategy may progress from research/paper to tiny live preview. This is intentionally conservative. It combines the hard lessons from live fills, paper artefact reviews, copy observation, and v76 paper work into one no-trade-first gate. It never executes orders; it only returns a decision and reasons. """ reasons: list[str] = [] intent = candidate.intent if intent is None: reasons.append("no_order_intent") else: if context.market_regime.choppy: reasons.append("market_regime_choppy_no_trade") elif intent.side == "buy" and not context.market_regime.long_allowed: reasons.append("market_regime_blocks_long") elif intent.side == "sell" and not context.market_regime.short_allowed: reasons.append("market_regime_blocks_short") if candidate.family == "copy": # Direct copy execution remains blocked until observation produces enough # actual allowed, PnL-evaluable shadow entries. Current no-allowed runs # are useful research, not live permission. if context.copy_observation.allowed_signals <= 0: reasons.append("copy_observation_has_no_allowed_signals") if not context.copy_observation.complete: reasons.append("copy_observation_incomplete") if not context.copy_observation.data_quality_ok: reasons.append("copy_data_quality_not_ok") elif candidate.family == "copy_confirmation": if not candidate.copy_confirmation_ok: reasons.append("copy_confirmation_missing") else: if candidate.expected_move_vs_cost < Decimal("4"): reasons.append("expected_move_below_cost_hurdle") if not candidate.anti_chase_ok: reasons.append("anti_chase_not_confirmed") if not candidate.retest_confirmed: reasons.append("retest_not_confirmed") reasons.extend(_promotion_blockers(candidate.promotion)) if not context.reconcile_clean: reasons.append("reconcile_not_clean") if not context.stops_confirmed: reasons.append("stops_not_confirmed") if not context.alerts_confirmed: reasons.append("alerts_not_confirmed") if context.kill_switch_active: reasons.append("kill_switch_active") if candidate.max_effective_leverage > Decimal("1"): reasons.append("leverage_above_tiny_live_cap") if candidate.max_open_positions > 1: reasons.append("too_many_initial_open_positions") allowed = not reasons if allowed: return GlobalStrategyDecision( allowed=True, status="tiny_live_preview_ready", recommended_mode="tiny_live_preview", risk_limits={ "max_effective_leverage": str(candidate.max_effective_leverage), "max_open_positions": candidate.max_open_positions, "daily_loss_cap_required": True, "reduce_only_stop_required_after_fill": True, "no_averaging_down": True, }, ) if candidate.family == "copy": mode = "copy_research_only" status = "research" elif any(r in reasons for r in ("reconcile_not_clean", "stops_not_confirmed", "alerts_not_confirmed", "kill_switch_active")): mode = "paper_shadow_only" status = "blocked" else: mode = "paper_shadow_only" status = "candidate" if not _promotion_blockers(candidate.promotion) else "research" return GlobalStrategyDecision( allowed=False, status=status, recommended_mode=mode, reasons=tuple(dict.fromkeys(reasons)), risk_limits={ "max_effective_leverage": "1", "max_open_positions": 1, "live_entries_allowed": False, }, )