from __future__ import annotations import argparse, json, os, time, uuid from collections import Counter, deque from datetime import datetime, timezone from decimal import Decimal from pathlib import Path from typing import Any from src.execution.cost_model import CostModel, TradeCostInput from src.execution.order_intent import OrderIntent from src.execution.paper_executor import PaperExecutor from src.execution.paper_lifecycle import PaperCandle, simulate_paper_lifecycle from src.hyperliquid.market_data import HyperliquidMarketData from src.market.context import CoinMarketContext from src.market.confluence import score_coin_confluence from src.research.tradingview_context import classify_tradingview_impulse, context_to_journal_dict, load_tradingview_latest from src.risk.data_quality_gate import DataQualityGate from src.risk.pretrade_risk_gate import PretradeRiskGate, RiskContext from src.strategies.v76_hl_confirmed_squeeze_hybrid import MarketContext, build_v76_order_intent, strict_anti_chase_blockers from src.strategies.v76_fee_aware_anti_chase import AntiChaseContext, anti_chase_blockers, build_fee_aware_anti_chase_intent from src.strategies.swing_trend_retest import SwingRetestContext, build_swing_trend_retest_intent ROOT = Path("runtime/experiments") BASE_ID = "candidate_v76_hl_confirmed_squeeze_hybrid" STRICT_ID = "candidate_v76_strict_live_candidate" RESEARCH_ID = "candidate_v76_research_probe" ANTI_CHASE_ID = "candidate_v76_fee_aware_anti_chase" SWING_ID = "candidate_swing_trend_retest_research" def _json_default(value: Any) -> Any: if isinstance(value, Decimal): return str(value) if isinstance(value, datetime): return value.isoformat() return str(value) def _append_jsonl(path: Path, row: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("a", encoding="utf-8") as handle: handle.write(json.dumps(row, default=_json_default, sort_keys=True) + "\n") def _runtime_dir(strategy_id: str) -> Path: return ROOT / strategy_id def _pid_path(strategy_id: str) -> Path: return _runtime_dir(strategy_id) / "bot.pid" def variant_params(strategy_id: str) -> dict[str, Any]: if strategy_id == STRICT_ID: return { "expected_move_mult": Decimal("3"), "min_breadth": 3, "research": False, "max_spread": Decimal("0.12"), "wld_risk_multiplier": Decimal("0.25"), "strict_anti_chase": True, "confluence_gate": True, "min_confluence_score": Decimal("60"), # Fresh lifecycle evidence showed the edge is either immediate or not # present: TP/trailing winners mostly close inside one tick, while # 17+ tick time-exits were the dominant PnL drag. Keep this paper-only # until a new lifecycle scorecard proves improvement. "max_hold_ticks": 8, "dead_fish_ticks": 4, "dead_fish_min_mfe_pct": Decimal("0.18"), "wld_max_hold_ticks": 4, "wld_dead_fish_ticks": 2, "wld_dead_fish_min_mfe_pct": Decimal("0.25"), } if strategy_id == RESEARCH_ID: return {"expected_move_mult": Decimal("2"), "min_breadth": 2, "research": True, "max_spread": Decimal("0.20"), "wld_risk_multiplier": Decimal("0.50")} if strategy_id == ANTI_CHASE_ID: return { "expected_move_mult": Decimal("4"), "min_breadth": 4, "research": True, "max_spread": Decimal("0.08"), "wld_risk_multiplier": Decimal("0"), "anti_chase": True, "confluence_gate": True, "min_confluence_score": Decimal("75"), } if strategy_id == SWING_ID: return { "expected_move_mult": Decimal("5"), "min_breadth": 3, "research": True, "max_spread": Decimal("0.08"), "wld_risk_multiplier": Decimal("0"), "swing_retest": True, "max_hold_ticks": 96, "dead_fish_ticks": 16, "dead_fish_min_mfe_pct": Decimal("0.30"), "confluence_gate": True, "min_confluence_score": Decimal("72"), } return {"expected_move_mult": Decimal("3"), "min_breadth": 3, "research": False, "max_spread": Decimal("0.15"), "wld_risk_multiplier": Decimal("0.25")} def _trend_state(mid: Decimal, prev: Decimal | None) -> str: if prev is None: return "unknown" return "up" if mid > prev else ("down" if mid < prev else "flat") def _market_trend_state(mids: dict[str, Decimal], prev_mids: dict[str, Any]) -> dict[str, Any]: comparable = 0 up = 0 down = 0 for coin, mid in mids.items(): raw = prev_mids.get(coin) if isinstance(prev_mids, dict) else None if raw is None: continue prev = _to_decimal(raw) if prev <= 0: continue comparable += 1 if mid > prev: up += 1 elif mid < prev: down += 1 if comparable < 3: direction = "unknown" breadth = Decimal("0") else: breadth = (Decimal(up) / Decimal(comparable) * Decimal("100")).quantize(Decimal("0.01")) if breadth >= Decimal("60"): direction = "up" elif breadth <= Decimal("40"): direction = "down" else: direction = "mixed" return {"direction": direction, "positive_pct": breadth, "comparable": comparable, "up": up, "down": down} def _context(coin: str, mid: Decimal, prev: Decimal | None, dq_allowed: bool, params: dict[str, Any], top_share: Decimal) -> MarketContext: trend_up = prev is not None and mid > prev previous = prev or mid move_pct = (mid / previous - Decimal("1")) * Decimal("100") if previous else Decimal("0") recent_high = previous * Decimal("1.006") if params.get("strict_anti_chase") else mid * (Decimal("0.999") if trend_up else Decimal("1.001")) pullback = ((recent_high - mid) / recent_high * Decimal("100")) if recent_high else Decimal("0") sma_fast = mid * (Decimal("0.9990") if params.get("strict_anti_chase") else Decimal("1.0005") if trend_up else Decimal("0.9995")) sma_slow = mid * (Decimal("0.9975") if params.get("strict_anti_chase") else Decimal("1")) baseline = mid * Decimal("1.002") if not trend_up else mid * Decimal("0.95") vwap_proxy = (previous + mid + sma_fast + sma_slow) / Decimal("4") vwap_distance_pct = max(Decimal("0"), (mid - vwap_proxy) / vwap_proxy * Decimal("100")) if vwap_proxy > 0 else Decimal("0") upper_wick_pct = max(Decimal("0"), (recent_high - max(mid, previous)) / mid * Decimal("100")) if mid > 0 else Decimal("0") candle_body_pct = abs(mid - previous) / previous * Decimal("100") if previous > 0 else Decimal("0") leakage_blocked = coin == "WLD" and top_share > Decimal("60") and not params["research"] risk_mult = params["wld_risk_multiplier"] if coin == "WLD" and top_share > Decimal("50") else Decimal("1") return MarketContext( coin=coin, symbol=f"{coin}/USDC:USDC", current_price=mid, recent_high=recent_high, sma_fast=sma_fast, sma_slow=sma_slow, volume_24h=Decimal("50000000"), breadth_positive_candidates=int(params["min_breadth"]), relative_strength_rank=1, baseline_price=baseline, wallet_equity_usdc=Decimal("75"), coin_leakage_blocked=leakage_blocked, risk_multiplier=risk_mult, data_quality_allowed=dq_allowed, strict_anti_chase=bool(params.get("strict_anti_chase")), retest_low=mid * Decimal("0.994"), atr_pct=Decimal("1.20"), move_15m_pct=move_pct, move_1h_pct=move_pct * Decimal("1.5"), pullback_from_high_pct=pullback, spread_pct=Decimal("0.03"), expected_slippage_pct=Decimal("0.02"), funding_rate_hourly_pct=Decimal("0"), vwap_distance_pct=vwap_distance_pct, upper_wick_pct=upper_wick_pct, candle_body_pct=candle_body_pct, open_interest_change_pct=Decimal("0"), premium_pct=Decimal("0"), ) def _anti_chase_context(coin: str, mid: Decimal, prev: Decimal | None, dq_allowed: bool, params: dict[str, Any], spread_pct: Decimal, *, market_trend: dict[str, Any] | None = None, recent_stop_cooldown: bool = False, tradingview_context: Any | None = None) -> AntiChaseContext: previous = prev or mid move_pct = (mid / previous - Decimal("1")) * Decimal("100") if previous else Decimal("0") recent_high = max(mid, previous * Decimal("1.01")) pullback = ((recent_high - mid) / recent_high * Decimal("100")) if recent_high else Decimal("0") return AntiChaseContext( coin=coin, symbol=f"{coin}/USDC:USDC", current_price=mid, recent_high=recent_high, retest_low=mid * Decimal("0.992"), sma_fast=mid * Decimal("0.996"), sma_slow=mid * Decimal("0.988"), atr_pct=Decimal("1.20"), move_15m_pct=move_pct, move_1h_pct=move_pct * Decimal("1.5"), pullback_from_high_pct=pullback, breadth_positive_candidates=int(params["min_breadth"]), relative_strength_rank=2, volume_24h=Decimal("50000000"), spread_pct=spread_pct, expected_slippage_pct=Decimal("0.02"), funding_rate_hourly_pct=Decimal("0"), wallet_equity_usdc=Decimal("75"), market_trend_direction=str((market_trend or {}).get("direction") or "unknown"), market_breadth_positive_pct=_to_decimal((market_trend or {}).get("positive_pct"), "0"), recent_stop_cooldown=recent_stop_cooldown, tradingview_context=tradingview_context, data_quality_allowed=dq_allowed, ) def _swing_context(coin: str, mid: Decimal, prev: Decimal | None, dq_allowed: bool, params: dict[str, Any], spread_pct: Decimal) -> SwingRetestContext: previous = prev or mid trend_up = prev is not None and mid >= previous swing_high = max(mid * Decimal("1.018"), previous * Decimal("1.025")) pullback = ((swing_high - mid) / swing_high * Decimal("100")) if swing_high > 0 else Decimal("0") sma_slow = mid * (Decimal("0.970") if trend_up else Decimal("1.005")) sma_fast = mid * (Decimal("0.990") if trend_up else Decimal("0.998")) retest_low = min(previous, sma_fast) * Decimal("0.995") trend_strength = ((sma_fast / sma_slow) - Decimal("1")) * Decimal("100") if sma_slow > 0 else Decimal("0") reclaim = ((mid / retest_low) - Decimal("1")) * Decimal("100") if retest_low > 0 else Decimal("0") return SwingRetestContext( coin=coin, symbol=f"{coin}/USDC:USDC", current_price=mid, sma_fast=sma_fast, sma_slow=sma_slow, swing_high=swing_high, retest_low=retest_low, atr_pct=Decimal("1.30"), trend_strength_pct=trend_strength, pullback_from_high_pct=pullback, reclaim_pct=reclaim, volume_24h=Decimal("50000000"), spread_pct=spread_pct, expected_slippage_pct=Decimal("0.02"), funding_rate_hourly_pct=Decimal("0"), market_breadth=int(params["min_breadth"]), relative_strength_rank=2, wallet_equity_usdc=Decimal("75"), data_quality_allowed=dq_allowed, ) def _to_decimal(value: Any, default: str = "0") -> Decimal: try: return Decimal(str(value)) except Exception: return Decimal(default) def _load_latest_confluence(runtime_root: Path = Path("runtime")) -> dict[str, Any]: path = runtime_root / "reports" / "market_confluence_latest.json" if not path.exists(): return {} try: loaded = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError: return {} return loaded if isinstance(loaded, dict) else {} def _confluence_gate(*, coin: str, dq_ctx: CoinMarketContext, params: dict[str, Any], latest: dict[str, Any]) -> dict[str, Any]: """Paper-only confluence gate for v76 strict entries. Uses the persisted Market Confluence Light snapshot when available. If no snapshot is present, falls back to a local read-only score from the current tick context. Never grants live permissions. """ min_score = _to_decimal(params.get("min_confluence_score"), "60") raw_conf_obj = latest.get("confluence") raw_conf: dict[str, Any] = raw_conf_obj if isinstance(raw_conf_obj, dict) else {} raw_scores_obj = raw_conf.get("scores") raw_scores: dict[str, Any] = raw_scores_obj if isinstance(raw_scores_obj, dict) else {} row = raw_scores.get(coin.upper()) if isinstance(raw_scores, dict) else None if isinstance(row, dict): recommendation = str(row.get("recommendation") or "block_new_entry") score = _to_decimal(row.get("final_score")) blockers = [str(item) for item in (row.get("blockers") or [])] components = row.get("components") if isinstance(row.get("components"), dict) else {} source = "market_confluence_latest" else: local = score_coin_confluence(dq_ctx, min_final_score=min_score) recommendation = local.recommendation score = local.final_score blockers = list(local.blockers) components = local.components source = "local_tick_context" depth_penalty_pct = max( _to_decimal(components.get("buy_impact_1k_pct")), _to_decimal(components.get("sell_impact_1k_pct")), ) if isinstance(components, dict) else Decimal("0") allowed = recommendation == "paper_candidate" and score >= min_score and not blockers reasons = [] if allowed else ["confluence_block_new_entry", *[f"confluence_{reason}" for reason in blockers]] return { "source": source, "score": str(score), "recommendation": recommendation, "allowed": allowed, "blockers": blockers, "components": components, "depth_penalty_pct": str(depth_penalty_pct), "reasons": reasons, "live_order_allowed": False, "mainnet_signed_action": False, } def _intent_from_position(coin: str, pos: dict[str, Any], strategy_id: str) -> OrderIntent | None: entry = _to_decimal(pos.get("entry")) size = _to_decimal(pos.get("size")) stop_loss = _to_decimal(pos.get("stop_loss")) take_profit = _to_decimal(pos.get("take_profit")) if pos.get("take_profit") is not None else None if entry <= 0 or size <= 0 or stop_loss <= 0: return None return OrderIntent( strategy_id=strategy_id, symbol=f"{coin}/USDC:USDC", coin=str(coin).upper(), side="buy", reduce_only=False, order_type="market", tif="Ioc", size=size, price=None, trigger_price=None, stop_loss=stop_loss, take_profit=take_profit, client_order_id=str(pos.get("client_order_id") or "paper-lifecycle-close"), reason=str(pos.get("entry_reason") or "paper_lifecycle"), risk_usd=_to_decimal(pos.get("risk_usd")), estimated_notional_usd=entry * size, ) def _close_lifecycle_positions(*, state: dict[str, Any], mids: dict[str, Decimal], runtime_dir: Path, strategy_id: str, research: bool) -> tuple[int, set[str]]: open_positions = state.setdefault("open_positions", {}) if not isinstance(open_positions, dict) or not open_positions: state["open_positions"] = {} if not isinstance(open_positions, dict) else open_positions return 0, set() closed = 0 closed_coins: set[str] = set() prev_mids = state.get("prev_mids", {}) if isinstance(state.get("prev_mids"), dict) else {} for coin, pos in list(open_positions.items()): coin = str(coin).upper() mid = mids.get(coin) if mid is None or not isinstance(pos, dict): continue intent = _intent_from_position(coin, pos, strategy_id) if intent is None: continue previous = _to_decimal(prev_mids.get(coin), str(pos.get("entry") or mid)) entry_price = _to_decimal(pos.get("entry")) tick_high = max(previous, mid) tick_low = min(previous, mid) high_watermark = max(_to_decimal(pos.get("high_watermark"), str(entry_price)), tick_high) low_watermark = min(_to_decimal(pos.get("low_watermark"), str(entry_price)), tick_low) pos["high_watermark"] = str(high_watermark) pos["low_watermark"] = str(low_watermark) ticks_held = int(pos.get("ticks_held") or 0) + 1 pos["ticks_held"] = ticks_held max_hold_ticks = int(pos.get("max_hold_ticks") or 24) dead_fish_ticks = int(pos.get("dead_fish_ticks") or 0) dead_fish_min_mfe_pct = _to_decimal(pos.get("dead_fish_min_mfe_pct"), "0") candle = PaperCandle(open=previous, high=tick_high, low=tick_low, close=mid) result = simulate_paper_lifecycle( intent, entry_price=entry_price, candles=[candle], fee_rate=Decimal("0.00045"), slippage_pct=_to_decimal(pos.get("slippage_pct"), "0.02"), funding_rate_hourly_pct=_to_decimal(pos.get("funding_rate_hourly_pct"), "0"), trailing_stop_pct=_to_decimal(pos.get("trailing_stop_pct")) if pos.get("trailing_stop_pct") is not None else None, max_fill_notional_usd=_to_decimal(pos.get("max_fill_notional_usd"), str(entry_price * intent.size)), entry_liquidity=str(pos.get("entry_liquidity") or "taker"), exit_liquidity=str(pos.get("exit_liquidity") or "taker"), depth_slippage_pct=_to_decimal(pos.get("slippage_pct"), "0.02"), stop_gap_slippage_pct=_to_decimal(pos.get("stop_gap_slippage_pct"), str(_to_decimal(pos.get("slippage_pct"), "0.02"))), ) pos["mfe_pct"] = str(max(_to_decimal(pos.get("mfe_pct")), result.mfe_pct)) pos["mae_pct"] = str(min(_to_decimal(pos.get("mae_pct")), result.mae_pct)) # Keep lifecycle positions open unless a real exit trigger fired, max hold # is reached, or a paper-only dead-fish rule says the setup failed to move. if result.exit_reason == "time_exit" and ticks_held < max_hold_ticks: if not (dead_fish_ticks and ticks_held >= dead_fish_ticks and _to_decimal(pos.get("mfe_pct"), "0") < dead_fish_min_mfe_pct): continue row = result.to_journal_row() if result.exit_reason == "time_exit": row["candles_held"] = ticks_held row["mfe_pct"] = pos["mfe_pct"] row["mae_pct"] = pos["mae_pct"] if dead_fish_ticks and ticks_held < max_hold_ticks: row["exit_reason"] = "dead_fish_time_exit" row["dead_fish_ticks"] = dead_fish_ticks row["dead_fish_min_mfe_pct"] = str(dead_fish_min_mfe_pct) row.update({"timestamp": datetime.now(timezone.utc).isoformat(), "event": "exit", "research": research}) _append_jsonl(runtime_dir / "trade_journal.jsonl", row) del open_positions[coin] closed += 1 closed_coins.add(coin) return closed, closed_coins def _open_lifecycle_position(*, state: dict[str, Any], coin: str, mid: Decimal, intent: OrderIntent, fill: Any, strategy_id: str, research: bool, expected_move_pct: Decimal, expected_vs_cost: Decimal, runtime_dir: Path, strict_anti_chase: bool = False, depth_slippage_pct: Decimal = Decimal("0.02"), max_hold_ticks: int = 24, dead_fish_ticks: int = 0, dead_fish_min_mfe_pct: Decimal = Decimal("0")) -> None: entry_fill_price = _to_decimal(getattr(fill, "fill_price", mid), str(mid)) state.setdefault("open_positions", {})[coin] = { "entry": str(entry_fill_price), "mark_at_entry": str(mid), "size": str(intent.size), "stop_loss": str(intent.stop_loss), "take_profit": str((mid * (Decimal("1") + expected_move_pct / Decimal("100"))).quantize(Decimal("0.00000001"))), "trailing_stop_pct": "1.20", "opened_at": datetime.now(timezone.utc).isoformat(), "strategy_id": strategy_id, "entry_reason": intent.reason, "client_order_id": intent.client_order_id, "risk_usd": str(intent.risk_usd), "entry_fee_usd": str(fill.entry_fee_usd), "spread_cost_usd": str(fill.spread_cost_usd), "slippage_cost_usd": str(fill.slippage_cost_usd), "slippage_pct": str(depth_slippage_pct), "entry_liquidity": "taker", "exit_liquidity": "taker", "max_fill_notional_usd": str(intent.estimated_notional_usd if intent.estimated_notional_usd > 0 else mid * intent.size), "funding_rate_hourly_pct": "0", "mfe_pct": "0", "mae_pct": "0", "high_watermark": str(mid), "low_watermark": str(mid), "ticks_held": 0, "max_hold_ticks": max_hold_ticks, "dead_fish_ticks": dead_fish_ticks, "dead_fish_min_mfe_pct": str(dead_fish_min_mfe_pct), } _append_jsonl(runtime_dir / "trade_journal.jsonl", { "timestamp": datetime.now(timezone.utc).isoformat(), "event": "entry", "strategy_id": strategy_id, "coin": coin, "paper_trading": True, "research": research, "entry_price": str(entry_fill_price), "mark_at_entry": str(mid), "size": str(intent.size), "stop_loss": str(intent.stop_loss), "take_profit": state["open_positions"][coin]["take_profit"], "entry_fee_usd": str(fill.entry_fee_usd), "spread_cost_usd": str(fill.spread_cost_usd), "slippage_cost_usd": str(fill.slippage_cost_usd), "mainnet_signed_action": False, "live_order_allowed": False, "extra": {"risk_usd": str(intent.risk_usd), "expected_move_vs_cost": str(expected_vs_cost), "lifecycle": "entry_fill_stop_tp_trailing", "strict_anti_chase": strict_anti_chase, "max_hold_ticks": max_hold_ticks, "dead_fish_ticks": dead_fish_ticks, "dead_fish_min_mfe_pct": str(dead_fish_min_mfe_pct)}, }) def _hold_policy_for_coin(coin: str, params: dict[str, Any]) -> dict[str, Any]: coin = coin.upper() max_hold_ticks = int(params.get("max_hold_ticks") or 24) dead_fish_ticks = int(params.get("dead_fish_ticks") or 0) dead_fish_min_mfe_pct = _to_decimal(params.get("dead_fish_min_mfe_pct"), "0") if coin == "WLD": max_hold_ticks = int(params.get("wld_max_hold_ticks") or max_hold_ticks) dead_fish_ticks = int(params.get("wld_dead_fish_ticks") or dead_fish_ticks) dead_fish_min_mfe_pct = _to_decimal(params.get("wld_dead_fish_min_mfe_pct"), str(dead_fish_min_mfe_pct)) return {"max_hold_ticks": max_hold_ticks, "dead_fish_ticks": dead_fish_ticks, "dead_fish_min_mfe_pct": dead_fish_min_mfe_pct} def _close_anti_chase_positions(*, state: dict[str, Any], mids: dict[str, Decimal], runtime_dir: Path, strategy_id: str, research: bool) -> tuple[int, set[str]]: open_positions = state.setdefault("open_positions", {}) if not isinstance(open_positions, dict) or not open_positions: state["open_positions"] = {} if not isinstance(open_positions, dict) else open_positions return 0, set() closed = 0 closed_coins: set[str] = set() for coin, pos in list(open_positions.items()): mid = mids.get(str(coin).upper()) if mid is None or not isinstance(pos, dict): continue entry = _to_decimal(pos.get("entry")) size = _to_decimal(pos.get("size")) stop_loss = _to_decimal(pos.get("stop_loss")) if entry <= 0 or size <= 0: continue exit_reason = None if stop_loss > 0 and mid <= stop_loss: exit_reason = "stop_loss" elif mid >= entry * Decimal("1.012"): exit_reason = "profit_target" if exit_reason is None: continue if exit_reason == "stop_loss": cooldowns = state.setdefault("cooldowns", {}) if isinstance(cooldowns, dict): cooldowns[str(coin).upper()] = 6 entry_fee = _to_decimal(pos.get("entry_fee_usd")) spread_cost = _to_decimal(pos.get("spread_cost_usd")) slippage_cost = _to_decimal(pos.get("slippage_cost_usd")) exit_fee = (mid * size * Decimal("0.0004")).quantize(Decimal("0.00000001")) gross = (mid - entry) * size net = gross - entry_fee - exit_fee - spread_cost - slippage_cost _append_jsonl(runtime_dir / "trade_journal.jsonl", { "timestamp": datetime.now(timezone.utc).isoformat(), "event": "exit", "strategy_id": strategy_id, "coin": str(coin).upper(), "paper_trading": True, "research": research, "entry_price": str(entry), "exit_price": str(mid), "size": str(size), "exit_reason": exit_reason, "realized_pnl_usd": str(net), "net_pnl_usd": str(net), "gross_pnl_usd": str(gross), "entry_fee_usd": str(entry_fee), "exit_fee_usd": str(exit_fee), "spread_cost_usd": str(spread_cost), "slippage_cost_usd": str(slippage_cost), "mainnet_signed_action": False, "live_order_allowed": False, "extra": { "tradingview_context": pos.get("tradingview_context") if isinstance(pos.get("tradingview_context"), dict) else None, "tradingview_impulse": pos.get("tradingview_impulse") if isinstance(pos.get("tradingview_impulse"), dict) else None, }, }) del open_positions[coin] closed += 1 closed_coins.add(str(coin).upper()) return closed, closed_coins def scan_once(strategy_id: str, coins: list[str], *, env: str = "mainnet") -> dict[str, Any]: if os.getenv("CTB_LIVE_TRADING_ALLOWED", "").lower() == "true": raise PermissionError("v76 paper runtime refuses CTB_LIVE_TRADING_ALLOWED=true") params = variant_params(strategy_id) confluence_latest = _load_latest_confluence() if params.get("confluence_gate") else {} tradingview_contexts = load_tradingview_latest() runtime_dir = _runtime_dir(strategy_id) runtime_dir.mkdir(parents=True, exist_ok=True) state_path = runtime_dir / "state.json" state = json.loads(state_path.read_text(encoding="utf-8")) if state_path.exists() else {"prev_mids": {}, "recent_trade_coins": []} md = HyperliquidMarketData(env=env) try: mids = {k: Decimal(str(v)) for k, v in md.get_all_mids().items() if k in coins} except Exception as exc: event = { "timestamp": datetime.now(timezone.utc).isoformat(), "strategy_id": strategy_id, "event": "market_data_unavailable", "env": env, "error_type": type(exc).__name__, "message": str(exc)[:300], "mainnet_signed_action": False, "paper_trading": True, "action": "skip_iteration_no_new_entries", } _append_jsonl(runtime_dir / "runtime_health.jsonl", event) for coin in coins: _append_jsonl(runtime_dir / "signal_journal.jsonl", {"timestamp": event["timestamp"], "coin": coin, "strategy_id": strategy_id, "setup_type": "blocked", "signal_score": "0", "would_enter": False, "block_reason": ["api_degraded"], "spread_pct": None, "depth_ok": False, "expected_roundtrip_cost_pct": None, "expected_move_pct": None, "expected_move_vs_cost": None, "funding": None, "data_quality_allowed": False, "breadth_state": "unknown_api_degraded", "trend_state": "unknown_api_degraded", "leakage_state": {}, "risk_gate_result": {"allowed": False, "reasons": ["api_degraded"]}, "final_decision": "blocked:api_degraded"}) return {"status": "degraded", "strategy_id": strategy_id, "runtime_dir": str(runtime_dir), "signals_seen": len(coins), "trades": 0, "blocked": {"api_degraded": len(coins)}, "mainnet_signed_action": False, "paper_trading": True, "error_type": type(exc).__name__} dq_gate = DataQualityGate(max_spread_pct=params["max_spread"]) cost_model = CostModel() executor = PaperExecutor(cost_model=cost_model) closed_positions, closed_this_tick = (0, set()) if params.get("anti_chase"): closed_positions, closed_this_tick = _close_anti_chase_positions(state=state, mids=mids, runtime_dir=runtime_dir, strategy_id=strategy_id, research=bool(params["research"])) else: closed_positions, closed_this_tick = _close_lifecycle_positions(state=state, mids=mids, runtime_dir=runtime_dir, strategy_id=strategy_id, research=bool(params["research"])) prev_mids_state = state.get("prev_mids", {}) if isinstance(state.get("prev_mids"), dict) else {} market_trend = _market_trend_state(mids, prev_mids_state) cooldowns = state.setdefault("cooldowns", {}) if not isinstance(cooldowns, dict): cooldowns = {} state["cooldowns"] = cooldowns recent = deque(state.get("recent_trade_coins", []), maxlen=30) top_coin, top_count = Counter(recent).most_common(1)[0] if recent else (None, 0) top_share = Decimal(str(round(top_count / len(recent) * 100, 4))) if recent else Decimal("0") trades = 0 blocked: Counter[str] = Counter() signals = 0 for coin in coins: mid = mids.get(coin) if mid is None: continue prev_raw = state.get("prev_mids", {}).get(coin) prev = Decimal(str(prev_raw)) if prev_raw is not None else None spread_pct = Decimal("0.03") dq_ctx = CoinMarketContext( coin=coin, rsi=Decimal("55"), sma_fast=mid, sma_slow=mid, atr=mid * Decimal("0.01"), funding=Decimal("0"), volume=Decimal("50000000"), mid=mid, spread_pct=spread_pct, timestamp=datetime.now(timezone.utc), reliability_score=Decimal("0.99"), stale_data=False, l2_available=True, open_interest=Decimal("1"), premium=Decimal("0"), oracle_px=mid, mark_px=mid, day_ntl_vlm=Decimal("50000000"), bid_depth_notional_5=Decimal("50000"), ask_depth_notional_5=Decimal("50000"), buy_impact_1k_pct=Decimal("0.02"), sell_impact_1k_pct=Decimal("0.02"), buy_impact_5k_pct=Decimal("0.03"), sell_impact_5k_pct=Decimal("0.03"), ) dq = dq_gate.evaluate(dq_ctx) tv_ctx = tradingview_contexts.get(coin.upper()) confluence_gate = _confluence_gate(coin=coin, dq_ctx=dq_ctx, params=params, latest=confluence_latest) if params.get("confluence_gate") else {"allowed": True, "reasons": [], "source": "disabled", "score": None, "recommendation": "not_evaluated", "blockers": [], "live_order_allowed": False, "mainnet_signed_action": False} tv_impulse = classify_tradingview_impulse(tv_ctx, side="long") if params.get("anti_chase"): cooldown_remaining = int(cooldowns.get(coin, 0) or 0) ctx = _anti_chase_context( coin, mid, prev, dq.allowed, params, spread_pct, market_trend=market_trend, recent_stop_cooldown=cooldown_remaining > 0, tradingview_context=tv_ctx, ) pre_intent_blockers = anti_chase_blockers(ctx) intent = None if coin in state.get("open_positions", {}) or pre_intent_blockers else build_fee_aware_anti_chase_intent(ctx, client_order_id="paper-" + uuid.uuid4().hex) elif params.get("swing_retest"): ctx = _swing_context(coin, mid, prev, dq.allowed, params, spread_pct) intent = None if coin in state.get("open_positions", {}) else build_swing_trend_retest_intent(ctx, client_order_id="paper-" + uuid.uuid4().hex) pre_intent_blockers = [] else: ctx = _context(coin, mid, prev, dq.allowed, params, top_share) pre_intent_blockers = strict_anti_chase_blockers(ctx) if params.get("strict_anti_chase") else [] intent = build_v76_order_intent(ctx, client_order_id="paper-" + uuid.uuid4().hex) setup_type = "blocked" if intent and intent.reason == "confirmed_squeeze_breakout": setup_type = "squeeze_breakout" elif intent and intent.reason == "survival_rebound_sma_reclaim": setup_type = "survival_reclaim" elif intent and intent.reason == "fee_aware_breakout_retest": setup_type = "fee_aware_anti_chase_retest" elif intent and intent.reason == "swing_trend_retest_reclaim": setup_type = "swing_trend_retest_reclaim" expected_move_pct = Decimal("3.20") if setup_type == "swing_trend_retest_reclaim" else Decimal("1.44") if setup_type == "fee_aware_anti_chase_retest" else Decimal("1.25") if setup_type == "squeeze_breakout" else Decimal("0.75") if setup_type == "fee_aware_anti_chase_retest" and tv_impulse.get("category") == "buy_opportunity": # Paper-only reaction boost: a strong MCP buy impulse can make an # otherwise valid retest setup worth measuring, but it cannot bypass # anti-chase, confluence, cost, data-quality, risk, or live gates. expected_move_pct = max(expected_move_pct, Decimal("1.80")) expected_hold_hours = Decimal("24") if setup_type == "swing_trend_retest_reclaim" else Decimal("4") if setup_type == "fee_aware_anti_chase_retest" else Decimal("2") cost = cost_model.estimate(TradeCostInput(notional_usd=Decimal("15"), expected_move_pct=expected_move_pct, half_spread_pct=spread_pct / 2, depth_penalty_pct=_to_decimal(confluence_gate.get("depth_penalty_pct")), hold_hours=expected_hold_hours)) expected_vs_cost = expected_move_pct / cost.roundtrip_cost_pct if cost.roundtrip_cost_pct else Decimal("99") reasons: list[str] = list(pre_intent_blockers) risk_gate_result: dict[str, Any] = {"allowed": False, "reasons": ["no_intent"]} if not dq.allowed: reasons.append("data_quality") if not confluence_gate.get("allowed", True): reasons.extend(str(reason) for reason in confluence_gate.get("reasons", [])) if expected_vs_cost < params["expected_move_mult"]: reasons.append("cost") if getattr(ctx, "coin_leakage_blocked", False): reasons.append("leakage") if params.get("anti_chase") and coin in state.get("open_positions", {}): reasons.append("already_open") if not params.get("anti_chase") and coin in state.get("open_positions", {}): reasons.append("already_open") if params.get("anti_chase") and coin in closed_this_tick: reasons.append("closed_this_tick") if not params.get("anti_chase") and coin in closed_this_tick: reasons.append("closed_this_tick") if intent: risk = PretradeRiskGate(Decimal("10"), 1, Decimal("15"), Decimal("0.50")).evaluate(intent, RiskContext(0, False, False)) risk_gate_result = {"allowed": risk.allowed, "reasons": list(risk.reasons)} if not risk.allowed: reasons.append("risk") if not intent and not reasons: reasons.append("no_setup_or_no_reclaim") reasons = list(dict.fromkeys(reasons)) would_enter = bool(intent and not reasons) final_decision = "blocked:" + "+".join(reasons) if would_enter and intent is not None: depth_penalty_pct = _to_decimal(confluence_gate.get("depth_penalty_pct")) fill = executor.execute(intent, mark_price=mid, half_spread_pct=spread_pct / 2, expected_move_pct=expected_move_pct, depth_penalty_pct=depth_penalty_pct, hold_hours=Decimal("2")) if params.get("anti_chase"): state.setdefault("open_positions", {})[coin] = {"entry": str(mid), "size": str(intent.size), "stop_loss": str(intent.stop_loss), "opened_at": datetime.now(timezone.utc).isoformat(), "strategy_id": strategy_id, "entry_fee_usd": str(fill.entry_fee_usd), "spread_cost_usd": str(fill.spread_cost_usd), "slippage_cost_usd": str(fill.slippage_cost_usd), "tradingview_context": context_to_journal_dict(tv_ctx), "tradingview_impulse": tv_impulse} _append_jsonl(runtime_dir / "trade_journal.jsonl", {"timestamp": datetime.now(timezone.utc).isoformat(), "event": "entry", "strategy_id": strategy_id, "coin": coin, "paper_trading": True, "research": params["research"], "entry_price": str(mid), "size": str(intent.size), "stop_loss": str(intent.stop_loss), "entry_fee_usd": str(fill.entry_fee_usd), "spread_cost_usd": str(fill.spread_cost_usd), "slippage_cost_usd": str(fill.slippage_cost_usd), "mainnet_signed_action": False, "live_order_allowed": False, "extra": {"risk_usd": str(intent.risk_usd), "expected_move_vs_cost": str(expected_vs_cost), "tradingview_context": context_to_journal_dict(tv_ctx), "tradingview_impulse": tv_impulse}}) final_decision = "paper_opened" else: _open_lifecycle_position( state=state, coin=coin, mid=mid, intent=intent, fill=fill, strategy_id=strategy_id, research=bool(params["research"]), expected_move_pct=expected_move_pct, expected_vs_cost=expected_vs_cost, runtime_dir=runtime_dir, strict_anti_chase=bool(params.get("strict_anti_chase")), depth_slippage_pct=depth_penalty_pct if depth_penalty_pct > 0 else Decimal("0.02"), **_hold_policy_for_coin(coin, params), ) final_decision = "paper_opened_lifecycle" recent.append(coin) trades += 1 else: blocked.update(reasons) _append_jsonl(runtime_dir / "signal_journal.jsonl", {"timestamp": datetime.now(timezone.utc).isoformat(), "coin": coin, "strategy_id": strategy_id, "setup_type": setup_type, "signal_score": str(expected_vs_cost), "would_enter": would_enter, "block_reason": reasons, "spread_pct": str(spread_pct), "depth_ok": True, "expected_roundtrip_cost_pct": str(cost.roundtrip_cost_pct), "expected_move_pct": str(expected_move_pct), "expected_move_vs_cost": str(expected_vs_cost), "funding": "0", "data_quality_allowed": dq.allowed, "confluence_gate": confluence_gate, "tradingview_context": context_to_journal_dict(tv_ctx), "tradingview_impulse": tv_impulse, "breadth_state": f"min={params['min_breadth']}", "trend_state": _trend_state(mid, prev), "market_trend": market_trend, "leakage_state": {"top_coin": top_coin, "top_coin_share": str(top_share), "wld_throttled": coin == "WLD" and top_share > Decimal("50")}, "risk_gate_result": risk_gate_result, "final_decision": final_decision}) state.setdefault("prev_mids", {})[coin] = str(mid) signals += 1 state["recent_trade_coins"] = list(recent) for coin, remaining in list(cooldowns.items()): try: next_remaining = int(remaining) - 1 except Exception: next_remaining = 0 if next_remaining > 0: cooldowns[coin] = next_remaining else: cooldowns.pop(coin, None) state_path.write_text(json.dumps(state, indent=2, sort_keys=True), encoding="utf-8") return {"status": "ok", "strategy_id": strategy_id, "runtime_dir": str(runtime_dir), "signals_seen": signals, "trades": trades, "closed_positions": closed_positions, "blocked": dict(blocked), "mainnet_signed_action": False, "paper_trading": True} def supervisor_status(strategy_ids: list[str]) -> dict[str, Any]: out: dict[str, Any] = {} for strategy_id in strategy_ids: runtime_dir = _runtime_dir(strategy_id) pid = None pid_file = _pid_path(strategy_id) if pid_file.exists() and pid_file.read_text(encoding="utf-8").strip().isdigit(): pid = int(pid_file.read_text(encoding="utf-8")) running = bool(pid and Path(f"/proc/{pid}").exists()) env_ok = False cmd_ok = False paper_only = False if running: try: env = Path(f"/proc/{pid}/environ").read_bytes().decode("utf-8", "ignore") cmd = Path(f"/proc/{pid}/cmdline").read_bytes().decode("utf-8", "ignore") env_ok = f"CTB_STRATEGY_ID={strategy_id}" in env and "CTB_PAPER_TRADING=true" in env and "CTB_DRY_RUN=false" in env cmd_ok = "src.tools.v76_paper_runtime" in cmd and f"--strategy-id\x00{strategy_id}" in cmd paper_only = env_ok and "CTB_LIVE_TRADING_ALLOWED=true" not in env and "CTB_LIVE_ORDER_ALLOWED=true" not in env and "HL_MAINNET_SIGNED_ACTION=true" not in env except Exception: env_ok = False cmd_ok = False paper_only = False now = datetime.now(timezone.utc) journals = {} for name in ("trade_journal.jsonl", "signal_journal.jsonl", "runtime_health.jsonl"): path = runtime_dir / name if path.exists(): stat = path.stat() age_seconds = max(0, int((now - datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc)).total_seconds())) journals[name] = {"exists": True, "size_bytes": stat.st_size, "age_seconds": age_seconds, "fresh": age_seconds <= 900} else: journals[name] = {"exists": False, "size_bytes": 0, "age_seconds": None, "fresh": False} out[strategy_id] = { "pid": pid, "running": running, "env_ok": env_ok, "cmd_ok": cmd_ok, "paper_only": paper_only, "runtime_dir": str(runtime_dir), "trade_journal": str(runtime_dir / "trade_journal.jsonl"), "signal_journal": str(runtime_dir / "signal_journal.jsonl"), "journals": journals, } return out def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="v76 PaperRuntime: Mainnet market data read-only, PaperExecutor only, no signed actions.") parser.add_argument("--strategy-id", default=BASE_ID) parser.add_argument("--coins", default="BTC,ETH,SOL,LINK,WLD,SUI,ENA,BCH") parser.add_argument("--iterations", type=int, default=1) parser.add_argument("--interval-seconds", type=int, default=60) parser.add_argument("--status", action="store_true") parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) if args.status: payload = {"status": "ok", "supervisor": supervisor_status([BASE_ID, STRICT_ID, RESEARCH_ID, ANTI_CHASE_ID, SWING_ID])} else: os.environ.setdefault("CTB_PAPER_TRADING", "true") os.environ.setdefault("CTB_DRY_RUN", "false") os.environ["CTB_STRATEGY_ID"] = args.strategy_id _runtime_dir(args.strategy_id).mkdir(parents=True, exist_ok=True) _pid_path(args.strategy_id).write_text(str(os.getpid()), encoding="utf-8") coins = [c.strip().upper() for c in args.coins.split(",") if c.strip()] results = [] for idx in range(args.iterations): results.append(scan_once(args.strategy_id, coins)) if idx < args.iterations - 1: time.sleep(args.interval_seconds) payload = {"status": "ok", "results": results, "supervisor": supervisor_status([args.strategy_id])} print(json.dumps(payload, indent=2, default=_json_default, sort_keys=True) if args.json else payload) return 0 if __name__ == "__main__": raise SystemExit(main())