from __future__ import annotations from collections import Counter, defaultdict from dataclasses import dataclass from decimal import Decimal, ROUND_DOWN from typing import Any from src.execution.cost_model import CostModel, TradeCostInput from src.execution.order_intent import OrderIntent from src.research.tradingview_context import TradingViewContext, tradingview_blockers STRATEGY_ID = "candidate_v76_fee_aware_anti_chase" ALLOWED_COINS = {"BTC", "ETH", "SOL", "LINK"} @dataclass(frozen=True) class AntiChaseContext: coin: str symbol: str current_price: Decimal recent_high: Decimal retest_low: Decimal sma_fast: Decimal sma_slow: Decimal atr_pct: Decimal move_15m_pct: Decimal move_1h_pct: Decimal pullback_from_high_pct: Decimal breadth_positive_candidates: int relative_strength_rank: int volume_24h: Decimal spread_pct: Decimal expected_slippage_pct: Decimal funding_rate_hourly_pct: Decimal wallet_equity_usdc: Decimal market_trend_direction: str = "unknown" market_breadth_positive_pct: Decimal = Decimal("0") recent_stop_cooldown: bool = False tradingview_context: TradingViewContext | None = None data_quality_allowed: bool = True data_quality_reasons: tuple[str, ...] = () @dataclass(frozen=True) class FillHistoryRow: coin: str side: str px: Decimal sz: Decimal closed_pnl: Decimal fee: Decimal minutes_held: Decimal = Decimal("0") def _quantize_size(notional: Decimal, price: Decimal) -> Decimal: if price <= 0: return Decimal("0") return (notional / price).quantize(Decimal("0.00000001"), rounding=ROUND_DOWN) def _expected_edge_pct(ctx: AntiChaseContext) -> Decimal: # Conservative edge proxy: a retest setup may reasonably target roughly 1.2 ATR, # capped so tiny ATR regimes do not invent edge from noise. return max(Decimal("0"), ctx.atr_pct * Decimal("1.20")) def _stop_distance_pct(ctx: AntiChaseContext) -> Decimal: return max(Decimal("0.80"), ctx.atr_pct * Decimal("0.85")) def anti_chase_blockers(ctx: AntiChaseContext) -> list[str]: coin = ctx.coin.upper() blockers: list[str] = [] if coin not in ALLOWED_COINS: blockers.append("coin_not_live_primary") if not ctx.data_quality_allowed: blockers.append("data_quality") if ctx.recent_stop_cooldown: blockers.append("recent_stop_cooldown") if ctx.market_trend_direction == "down": blockers.append("market_trend_down") if ctx.market_trend_direction != "unknown" and ctx.market_breadth_positive_pct < Decimal("50"): blockers.append("market_breadth_not_supportive") blockers.extend(tradingview_blockers(ctx.tradingview_context, side="long", max_age_seconds=3600)) if ctx.volume_24h < Decimal("25000000"): blockers.append("weak_liquidity") if ctx.breadth_positive_candidates < 4: blockers.append("weak_market_breadth") if ctx.relative_strength_rank > 3: blockers.append("weak_relative_strength") overextended = ctx.move_15m_pct > Decimal("1.20") or ctx.move_1h_pct > Decimal("3.00") no_meaningful_retest = ctx.pullback_from_high_pct < Decimal("0.45") if overextended and no_meaningful_retest: blockers.append("anti_chase_overextended_no_retest") reclaim_from_retest_pct = ((ctx.current_price / ctx.retest_low) - Decimal("1")) * Decimal("100") if ctx.retest_low > 0 else Decimal("0") retest_held = ctx.current_price >= ctx.sma_fast >= ctx.sma_slow and reclaim_from_retest_pct >= Decimal("0.30") below_recent_high = ctx.current_price <= ctx.recent_high if not (retest_held and below_recent_high): blockers.append("no_confirmed_retest_hold") expected_edge = _expected_edge_pct(ctx) stop_distance = _stop_distance_pct(ctx) if expected_edge < stop_distance * Decimal("1.40"): blockers.append("risk_reward_below_1_4x") cost = CostModel(min_slippage_pct=ctx.expected_slippage_pct).estimate( TradeCostInput( notional_usd=Decimal("15"), expected_move_pct=expected_edge, half_spread_pct=ctx.spread_pct / Decimal("2"), depth_penalty_pct=ctx.expected_slippage_pct, hold_hours=Decimal("4"), funding_rate_hourly_pct=ctx.funding_rate_hourly_pct, ) ) if expected_edge < cost.roundtrip_cost_pct * Decimal("4"): blockers.append("edge_below_4x_realistic_cost") return blockers def build_fee_aware_anti_chase_intent(ctx: AntiChaseContext, *, client_order_id: str) -> OrderIntent | None: if anti_chase_blockers(ctx): return None risk_pct = Decimal("0.15") stop_distance_pct = _stop_distance_pct(ctx) risk_usd = ctx.wallet_equity_usdc * risk_pct / Decimal("100") notional = min(ctx.wallet_equity_usdc * Decimal("0.15"), risk_usd / (stop_distance_pct / Decimal("100"))) if notional < Decimal("10"): return None stop_loss = min(ctx.sma_slow, ctx.retest_low * Decimal("0.9975"), ctx.current_price * (Decimal("1") - stop_distance_pct / Decimal("100"))) size = _quantize_size(notional, ctx.current_price) if size <= 0: return None return OrderIntent( strategy_id=STRATEGY_ID, symbol=ctx.symbol, coin=ctx.coin.upper(), side="buy", reduce_only=False, order_type="market", tif="Ioc", size=size, price=None, trigger_price=None, stop_loss=stop_loss.quantize(Decimal("0.00000001"), rounding=ROUND_DOWN), take_profit=None, client_order_id=client_order_id, reason="fee_aware_breakout_retest", risk_usd=risk_usd, estimated_notional_usd=notional, ) def evaluate_recent_fill_history(rows: list[FillHistoryRow]) -> dict[str, Any]: closed = [row for row in rows if row.closed_pnl != 0] closed_count = len(closed) wins = sum(1 for row in closed if row.closed_pnl > 0) losses = sum(1 for row in closed if row.closed_pnl < 0) total_closed_pnl = sum((row.closed_pnl for row in closed), Decimal("0")) total_fees = sum((row.fee for row in rows), Decimal("0")) short_hold_net = sum((row.closed_pnl - row.fee for row in closed if row.minutes_held < Decimal("60")), Decimal("0")) long_hold_net = sum((row.closed_pnl - row.fee for row in closed if row.minutes_held >= Decimal("240")), Decimal("0")) by_coin: defaultdict[str, Decimal] = defaultdict(lambda: Decimal("0")) counts = Counter(row.coin.upper() for row in closed) for row in closed: by_coin[row.coin.upper()] += row.closed_pnl - row.fee win_rate = (Decimal(wins) / Decimal(closed_count) * Decimal("100")).quantize(Decimal("0.01")) if closed_count else Decimal("0.00") blockers: list[str] = [] if closed_count < 30: blockers.append("sample_too_small") if total_closed_pnl - total_fees <= 0: blockers.append("net_pnl_negative_after_fees") if win_rate < Decimal("45"): blockers.append("win_rate_below_gate") if short_hold_net < 0: blockers.append("short_holds_negative") top_coin, top_count = counts.most_common(1)[0] if counts else (None, 0) top_share = (Decimal(top_count) / Decimal(closed_count) * Decimal("100")).quantize(Decimal("0.01")) if closed_count else Decimal("0.00") if top_share > Decimal("35"): blockers.append("coin_concentration") status = "candidate" if not blockers else "edge_failure" return { "status": status, "closed_count": closed_count, "wins": wins, "losses": losses, "win_rate_pct": win_rate, "closed_pnl": total_closed_pnl, "fees": total_fees, "net_after_fees": total_closed_pnl - total_fees, "short_hold_net_pnl": short_hold_net, "long_hold_net_pnl": long_hold_net, "top_coin": top_coin, "top_coin_share_pct": top_share, "per_coin_net_pnl": dict(sorted(by_coin.items())), "blockers": blockers, "recommendation": "paper_candidate_only" if status == "candidate" else "keep_live_entries_blocked", }