from __future__ import annotations import argparse import fcntl import json import os import time import uuid from datetime import datetime, timezone from decimal import Decimal from pathlib import Path from typing import Any from src.hyperliquid.market_data import HyperliquidMarketData from src.research.external_market_context import load_external_context from src.strategies.confirmed_range_reversion_v78_2 import ( STRATEGY_ID, STRATEGY_VERSION, build_maker_intent, classify_true_range, confirmed_range_reversion, ) from src.tools.v77_trend_retest_runtime import _asset_contexts, _cross_section, _fetch_feature ROOT = Path("runtime/experiments") EXTERNAL_CONTEXT = Path("runtime/research/external_market_context_latest.json") MAKER_FEE_RATE = Decimal("0.00015") TAKER_FEE_RATE = Decimal("0.00045") MIN_EXIT_SLIPPAGE_PCT = Decimal("0.02") MAX_WAIT_BARS = 2 MAX_HOLD_BARS = 8 QUEUE_CROSS_PCT = Decimal("0.005") def _d(value: Any, default: str = "0") -> Decimal: try: return Decimal(str(value)) except Exception: return Decimal(default) def _now() -> str: return datetime.now(timezone.utc).isoformat() def _append(path: Path, row: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) row.setdefault("timestamp", _now()) row.update({"paper_trading": True, "research_only": True, "live_order_allowed": False, "mainnet_signed_action": False}) with path.open("a", encoding="utf-8") as handle: handle.write(json.dumps(row, sort_keys=True, default=str) + "\n") def _save(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(json.dumps(payload, indent=2, sort_keys=True, default=str), encoding="utf-8") tmp.replace(path) def _assert_paper_only() -> None: if os.getenv("CTB_STRATEGY_ID") != STRATEGY_ID: raise PermissionError(f"v78.2 requires CTB_STRATEGY_ID={STRATEGY_ID}") if os.getenv("CTB_PAPER_TRADING", "").lower() != "true": raise PermissionError("v78.2 requires CTB_PAPER_TRADING=true") for key in ("CTB_LIVE_TRADING_ALLOWED", "CTB_LIVE_ORDER_ALLOWED", "HL_MAINNET_SIGNED_ACTION"): if os.getenv(key, "").lower() != "false": raise PermissionError(f"v78.2 requires {key}=false") def _exit_fill(raw_exit: Decimal, feature: Any, side: str) -> Decimal: impact = feature.sell_impact_1k_pct if side == "long" else feature.buy_impact_1k_pct slip_pct = max(MIN_EXIT_SLIPPAGE_PCT, feature.spread_pct / Decimal("2"), impact) return raw_exit * (Decimal("1") - slip_pct / Decimal("100")) if side == "long" else raw_exit * (Decimal("1") + slip_pct / Decimal("100")) def _manage_open(state: dict[str, Any], features: dict[str, Any], runtime: Path) -> int: positions = state.setdefault("open_positions", {}) closed = 0 for coin, pos in list(positions.items()): feature = features.get(coin) if feature is None or int(feature.candle_ts) <= int(pos["last_candle_ts"]): continue previous_candle_ts = int(pos["last_candle_ts"]) current_candle_ts = int(feature.candle_ts) elapsed_bars = max(1, (current_candle_ts - previous_candle_ts) // 900_000) pos["last_candle_ts"] = current_candle_ts pos["bars_held"] = int(pos.get("bars_held") or 0) + elapsed_bars side, entry, stop, target, size = str(pos["side"]), _d(pos["entry_price"]), _d(pos["stop_loss"]), _d(pos["take_profit"]), _d(pos["size"]) funding_delta = entry * size * feature.funding_rate_hourly_pct / Decimal("100") * Decimal(elapsed_bars) * Decimal("0.25") if side == "short": funding_delta = -funding_delta pos["accrued_funding_usd"] = str(_d(pos.get("accrued_funding_usd")) + funding_delta) favorable = (feature.high_15m - entry) / entry * Decimal("100") if side == "long" else (entry - feature.low_15m) / entry * Decimal("100") adverse = (entry - feature.low_15m) / entry * Decimal("100") if side == "long" else (feature.high_15m - entry) / entry * Decimal("100") pos["mfe_pct"] = str(max(_d(pos.get("mfe_pct")), favorable)) pos["mae_pct"] = str(max(_d(pos.get("mae_pct")), adverse)) stop_hit = feature.low_15m <= stop if side == "long" else feature.high_15m >= stop target_hit = feature.high_15m >= target if side == "long" else feature.low_15m <= target timed_out = int(pos["bars_held"]) >= MAX_HOLD_BARS if not stop_hit and not target_hit and not timed_out: continue reason = "stop_loss" if stop_hit else "vwap_take_profit" if target_hit else "no_reversion_time_exit" if stop_hit: raw_exit = min(stop, feature.open_15m) if side == "long" else max(stop, feature.open_15m) else: raw_exit = target if target_hit else feature.close_15m exit_price = _exit_fill(raw_exit, feature, side) gross = (exit_price - entry) * size if side == "long" else (entry - exit_price) * size entry_fee = _d(pos["entry_fee_usd"]) exit_fee = exit_price * size * TAKER_FEE_RATE funding = _d(pos.get("accrued_funding_usd")) net = gross - entry_fee - exit_fee - funding _append(runtime / "trade_journal.jsonl", {**pos, "event": "exit", "exit_price": str(exit_price), "exit_reason": reason, "gross_pnl_usd": str(gross), "fees_usd": str(entry_fee + exit_fee), "funding_cost_usd": str(funding), "net_pnl_usd": str(net), "proxy_inputs_used": False}) del positions[coin] closed += 1 return closed def _manage_pending(state: dict[str, Any], features: dict[str, Any], runtime: Path, *, range_allowed: bool = True) -> int: pending = state.setdefault("pending_orders", {}) positions = state.setdefault("open_positions", {}) filled = 0 for coin, order in list(pending.items()): feature = features.get(coin) if feature is None or int(feature.candle_ts) <= int(order["last_candle_ts"]): continue previous_candle_ts = int(order["last_candle_ts"]) current_candle_ts = int(feature.candle_ts) elapsed_bars = max(1, (current_candle_ts - previous_candle_ts) // 900_000) order["last_candle_ts"] = current_candle_ts order["wait_bars"] = int(order.get("wait_bars") or 0) + elapsed_bars if elapsed_bars > 1: del pending[coin] _append(runtime / "order_journal.jsonl", {**order, "event": "maker_cancel", "reason": "data_gap_invalidates_fill_path"}) continue if not range_allowed or feature.adx_15m > Decimal("25") or not feature.data_quality_allowed: del pending[coin] _append(runtime / "order_journal.jsonl", {**order, "event": "maker_cancel", "reason": "range_or_data_quality_invalidated"}) continue limit = _d(order["entry_price"]) queue_cross = limit * QUEUE_CROSS_PCT / Decimal("100") touched = feature.low_15m <= limit - queue_cross if order["side"] == "long" else feature.high_15m >= limit + queue_cross if touched: entry_fee = limit * _d(order["size"]) * MAKER_FEE_RATE position = {**order, "entry_fee_usd": str(entry_fee), "accrued_funding_usd": "0", "bars_held": 0, "mfe_pct": "0", "mae_pct": "0", "last_candle_ts": int(feature.candle_ts)} positions[coin] = position del pending[coin] _append(runtime / "trade_journal.jsonl", {**position, "event": "entry", "entry_order_type": "limit_Alo", "proxy_inputs_used": False}) side = str(position["side"]) stop = _d(position["stop_loss"]) same_bar_stop = feature.low_15m <= stop if side == "long" else feature.high_15m >= stop if same_bar_stop: raw_exit = min(stop, feature.open_15m) if side == "long" else max(stop, feature.open_15m) exit_price = _exit_fill(raw_exit, feature, side) size = _d(position["size"]) gross = (exit_price - limit) * size if side == "long" else (limit - exit_price) * size entry_fee = _d(position["entry_fee_usd"]) exit_fee = exit_price * size * TAKER_FEE_RATE _append(runtime / "trade_journal.jsonl", {**position, "event": "exit", "exit_price": str(exit_price), "exit_reason": "same_bar_stop_loss", "gross_pnl_usd": str(gross), "fees_usd": str(entry_fee + exit_fee), "funding_cost_usd": "0", "net_pnl_usd": str(gross - entry_fee - exit_fee), "same_bar_ambiguity_policy": "stop_enforced_target_deferred", "proxy_inputs_used": False}) del positions[coin] filled += 1 elif int(order["wait_bars"]) >= MAX_WAIT_BARS: del pending[coin] _append(runtime / "order_journal.jsonl", {**order, "event": "maker_cancel", "reason": "not_filled_before_expiry"}) return filled def _edge_multiple(intent: Any, feature: Any) -> Decimal: entry, target = _d(intent.price), _d(intent.take_profit) expected = abs(target - entry) / entry * Decimal("100") if entry > 0 else Decimal("0") exit_friction = max(MIN_EXIT_SLIPPAGE_PCT, feature.spread_pct / Decimal("2"), feature.sell_impact_1k_pct if intent.side == "buy" else feature.buy_impact_1k_pct) roundtrip = (MAKER_FEE_RATE + TAKER_FEE_RATE) * Decimal("100") + exit_friction return expected / roundtrip if roundtrip > 0 else Decimal("0") def _edge_cost_allowed(intent: Any, feature: Any) -> bool: return _edge_multiple(intent, feature) >= Decimal("3") def scan_once(coins: list[str], *, md: Any | None = None) -> dict[str, Any]: _assert_paper_only() runtime = ROOT / STRATEGY_ID state_path = runtime / "state.json" state = json.loads(state_path.read_text(encoding="utf-8")) if state_path.exists() else {} run_id = str(os.getenv("CTB_RUN_ID") or state.get("run_id") or uuid.uuid4().hex) state.update({"run_id": run_id, "strategy_version": STRATEGY_VERSION}) source = md or HyperliquidMarketData("mainnet") now_ms = int(time.time() * 1000) contexts, errors, raw = {}, {}, {} try: contexts = _asset_contexts(source.get_meta_and_asset_ctxs()) except Exception as exc: errors["asset_contexts"] = type(exc).__name__ for coin in coins: try: raw[coin] = _fetch_feature(source, coin, now_ms=now_ms, asset_ctx=contexts.get(coin, {})) except Exception as exc: errors[coin] = f"{type(exc).__name__}:{str(exc)[:100]}" features = _cross_section(raw) lifecycle_coins = set(state.get("open_positions") or {}) | set(state.get("pending_orders") or {}) for coin in lifecycle_coins - set(features): errors[coin] = "lifecycle_feature_missing" external = load_external_context(EXTERNAL_CONTEXT) range_decision = classify_true_range(features) closed = _manage_open(state, features, runtime) filled = _manage_pending(state, features, runtime, range_allowed=range_decision.allowed) opened = 0 blocked: dict[str, int] = {} last_windows = state.setdefault("last_signal_window_by_coin", {}) portfolio_busy = bool(state.get("open_positions") or state.get("pending_orders")) evaluations: list[dict[str, Any]] = [] for coin, feature in features.items(): decision = confirmed_range_reversion(feature, range_allowed=range_decision.allowed, external=external) reasons = list(decision.blockers) if last_windows.get(coin) == feature.data_window_id: reasons.append("duplicate_data_window") if portfolio_busy: reasons.append("single_position_or_pending_limit") intent = build_maker_intent(feature, decision) if not reasons else None edge_multiple = _edge_multiple(intent, feature) if intent is not None else Decimal("0") if intent is not None and edge_multiple < Decimal("3"): reasons.append("expected_vwap_edge_below_3x_cost") evaluations.append({"coin": coin, "feature": feature, "decision": decision, "intent": intent, "edge_multiple": edge_multiple, "reasons": reasons}) eligible = [row for row in evaluations if row["intent"] is not None and not row["reasons"]] selected = max(eligible, key=lambda row: (row["edge_multiple"], row["coin"])) if eligible else None for row in evaluations: coin, feature, decision, intent = row["coin"], row["feature"], row["decision"], row["intent"] reasons = row["reasons"] if selected is not None and row is not selected and intent is not None and not reasons: reasons.append("lower_ranked_simultaneous_candidate") if selected is row and intent is not None: order = {"strategy_id": STRATEGY_ID, "strategy_version": STRATEGY_VERSION, "run_id": run_id, "data_window_id": feature.data_window_id, "coin": coin, "side": decision.side, "entry_price": str(intent.price), "size": str(intent.size), "estimated_notional_usd": str(intent.estimated_notional_usd), "risk_usd": str(intent.risk_usd), "stop_loss": str(intent.stop_loss), "take_profit": str(intent.take_profit), "funding_rate_hourly_pct": str(feature.funding_rate_hourly_pct), "source_attribution": list(decision.source_attribution), "external_gate_profile": "coingecko_plus_hyperliquid_native" if coin == "HYPE" else "coingecko_plus_binance_crowding", "edge_to_cost_multiple": str(row["edge_multiple"]), "submitted_candle_ts": int(feature.candle_ts), "last_candle_ts": int(feature.candle_ts), "wait_bars": 0} state.setdefault("pending_orders", {})[coin] = order last_windows[coin] = feature.data_window_id opened += 1 _append(runtime / "order_journal.jsonl", {**order, "event": "maker_submitted", "order_type": "limit_Alo"}) for reason in reasons: blocked[reason] = blocked.get(reason, 0) + 1 _append(runtime / "signal_journal.jsonl", {"strategy_id": STRATEGY_ID, "strategy_version": STRATEGY_VERSION, "run_id": run_id, "data_window_id": feature.data_window_id, "coin": coin, "range_regime": range_decision.to_dict(), "decision": decision.to_dict(), "edge_to_cost_multiple": str(row["edge_multiple"]), "runtime_blockers": list(dict.fromkeys(reasons)), "features": {"adx_15m": str(feature.adx_15m), "bb_width_pct": str(feature.bb_width_pct), "bb_upper_15m": str(feature.bb_upper_15m), "bb_lower_15m": str(feature.bb_lower_15m), "vwap_15m": str(feature.vwap_15m), "best_bid": str(feature.best_bid), "best_ask": str(feature.best_ask), "rsi_15m": str(feature.rsi_15m), "volume_ratio": str(feature.volume_ratio), "sma_1h_spread_pct": str(feature.sma_1h_spread_pct), "sma_4h_spread_pct": str(feature.sma_4h_spread_pct)}, "proxy_inputs_used": False}) _save(state_path, state) result = {"status": "degraded" if errors else "ok", "strategy_id": STRATEGY_ID, "strategy_version": STRATEGY_VERSION, "run_id": run_id, "features_loaded": len(features), "range_regime": range_decision.to_dict(), "submitted": opened, "filled": filled, "closed": closed, "blocked": blocked, "data_errors": errors, "paper_trading": True, "research_only": True, "live_order_allowed": False, "mainnet_signed_action": False} _append(runtime / "runtime_health.jsonl", {"event": "scan_complete", **result}) return result def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="v78.2 confirmed range reversion; paper-only maker entries.") parser.add_argument("--strategy-id", default=STRATEGY_ID, choices=[STRATEGY_ID]) parser.add_argument("--coins", default="BTC,ETH,SOL,LINK,HYPE") parser.add_argument("--iterations", type=int, default=1) parser.add_argument("--interval-seconds", type=int, default=60) parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) os.environ.setdefault("CTB_STRATEGY_ID", STRATEGY_ID) os.environ.setdefault("CTB_RUN_ID", uuid.uuid4().hex) runtime = ROOT / STRATEGY_ID runtime.mkdir(parents=True, exist_ok=True) lock_handle = (runtime / "runtime.lock").open("w") try: fcntl.flock(lock_handle, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: print(json.dumps({"status": "blocked", "reason": "runtime_lock_held", "strategy_id": STRATEGY_ID})) return 2 (runtime / "bot.pid").write_text(str(os.getpid()), encoding="utf-8") results = [] for idx in range(args.iterations): results.append(scan_once([coin.strip().upper() for coin in args.coins.split(",") if coin.strip()])) if idx < args.iterations - 1: time.sleep(args.interval_seconds) payload = {"status": "ok" if all(row["status"] == "ok" for row in results) else "degraded", "results": results, "paper_trading": True, "research_only": True, "live_order_allowed": False, "mainnet_signed_action": False} print(json.dumps(payload, indent=2, sort_keys=True, default=str) if args.json else payload) return 0 if __name__ == "__main__": raise SystemExit(main())