from __future__ import annotations import argparse import json import os import time import uuid from dataclasses import replace from datetime import datetime, timezone from decimal import Decimal from pathlib import Path from typing import Any from src.execution.cost_model import CostModel from src.execution.paper_executor import PaperExecutor from src.hyperliquid.market_data import HyperliquidMarketData, summarize_l2_book from src.market.trend_retest_features import TrendRetestFeatures, build_trend_retest_features from src.research.external_market_context import load_external_context from src.research.tradingview_context import classify_tradingview_impulse, context_to_journal_dict, load_tradingview_latest, tradingview_blockers from src.strategies.relative_strength_momentum_v77_2 import MOMENTUM_RESEARCH_ID, MOMENTUM_VERSION, build_momentum_intent, momentum_blockers from src.strategies.v78_research_archetypes import external_entry_blockers from src.strategies.trend_retest_anti_chase_v2 import ( LONG_STRATEGY_ID, SHORT_RESEARCH_ID, STRATEGY_VERSION, build_intent, entry_blockers, feature_journal, with_cross_section, ) ROOT = Path("runtime/experiments") SUPPORTED = {LONG_STRATEGY_ID: "long", SHORT_RESEARCH_ID: "short", MOMENTUM_RESEARCH_ID: "long"} FEE_RATE = Decimal("0.00045") TP1_FRACTION = Decimal("0.50") DEAD_FISH_BARS = 4 DEAD_FISH_MIN_MFE_PCT = Decimal("0.20") MAX_HOLD_BARS = 24 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 _strategy_version(strategy_id: str) -> str: return MOMENTUM_VERSION if strategy_id == MOMENTUM_RESEARCH_ID else STRATEGY_VERSION def _strategy_entry_blockers(strategy_id: str, feature: TrendRetestFeatures, *, tv: Any | None) -> list[str]: if strategy_id == MOMENTUM_RESEARCH_ID: return list(dict.fromkeys([*momentum_blockers(feature), *tradingview_blockers(tv, side="long", max_age_seconds=3600)])) return entry_blockers(feature, side=SUPPORTED[strategy_id], tv=tv) def _strategy_intent(strategy_id: str, feature: TrendRetestFeatures, *, side: str, client_order_id: str): if strategy_id == MOMENTUM_RESEARCH_ID: return build_momentum_intent(feature, client_order_id=client_order_id) return build_intent(feature, side=side, client_order_id=client_order_id) def _append(path: Path, row: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) row.setdefault("timestamp", _now()) row.setdefault("paper_trading", True) row.setdefault("live_order_allowed", False) row.setdefault("mainnet_signed_action", False) path.open("a", encoding="utf-8").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_PAPER_TRADING", "").lower() != "true": raise PermissionError("v77 research runtime requires CTB_PAPER_TRADING=true") unsafe = [key for key in ("CTB_LIVE_TRADING_ALLOWED", "CTB_LIVE_ORDER_ALLOWED", "HL_MAINNET_SIGNED_ACTION") if os.getenv(key, "").lower() != "false"] if unsafe: raise PermissionError("v77 research runtime requires explicit false live/signed flags: " + ",".join(unsafe)) def _asset_contexts(payload: Any) -> dict[str, dict[str, Any]]: if not isinstance(payload, list) or len(payload) < 2 or not isinstance(payload[0], dict) or not isinstance(payload[1], list): return {} universe = payload[0].get("universe") or [] out: dict[str, dict[str, Any]] = {} for idx, meta in enumerate(universe): if idx < len(payload[1]) and isinstance(meta, dict) and isinstance(payload[1][idx], dict): out[str(meta.get("name") or "").upper()] = payload[1][idx] return out def _fetch_feature(md: Any, coin: str, *, now_ms: int, asset_ctx: dict[str, Any]) -> TrendRetestFeatures: intervals = {"15m": 130 * 15 * 60_000, "1h": 80 * 60 * 60_000, "4h": 60 * 4 * 60 * 60_000} candles = {name: md.get_candles(coin, name, now_ms - span, now_ms) for name, span in intervals.items()} l2 = summarize_l2_book(md.get_l2_book(coin)) if l2.spread_pct is None: raise ValueError("spread_unavailable") return build_trend_retest_features( coin=coin, candles_15m=candles["15m"], candles_1h=candles["1h"], candles_4h=candles["4h"], now_ms=now_ms, spread_pct=l2.spread_pct, bid_depth_notional_5=l2.bid_depth_notional_5, ask_depth_notional_5=l2.ask_depth_notional_5, buy_impact_1k_pct=l2.buy_impact_1k_pct if l2.buy_impact_1k_pct is not None else Decimal("-1"), sell_impact_1k_pct=l2.sell_impact_1k_pct if l2.sell_impact_1k_pct is not None else Decimal("-1"), best_bid=l2.best_bid or Decimal("0"), best_ask=l2.best_ask or Decimal("0"), funding_rate_hourly_pct=_d(asset_ctx.get("funding")) * Decimal("100"), open_interest=_d(asset_ctx.get("openInterest")), premium_pct=_d(asset_ctx.get("premium")) * Decimal("100"), ) def _cross_section(features: dict[str, TrendRetestFeatures]) -> dict[str, TrendRetestFeatures]: ordered_strength = sorted(features, key=lambda coin: features[coin].move_1h_pct, reverse=True) ordered_weakness = list(reversed(ordered_strength)) strength = {coin: idx + 1 for idx, coin in enumerate(ordered_strength)} weakness = {coin: idx + 1 for idx, coin in enumerate(ordered_weakness)} positive = sum(1 for feature in features.values() if feature.move_1h_pct > 0) breadth = Decimal(positive) / Decimal(len(features)) * Decimal("100") if features else Decimal("0") return { coin: with_cross_section(feature, strength_rank=strength[coin], weakness_rank=weakness[coin], breadth_positive_pct=breadth) for coin, feature in features.items() } def _confluence_blockers(coin: str, side: str, path: Path = Path("runtime/reports/market_confluence_latest.json"), *, max_age_seconds: int = 5400, strategy_id: str | None = None) -> list[str]: if not path.exists(): return ["confluence_report_missing"] if time.time() - path.stat().st_mtime > max_age_seconds: return ["confluence_report_stale"] try: payload = json.loads(path.read_text(encoding="utf-8")) except Exception: return ["confluence_report_invalid"] row = (((payload.get("confluence") or {}).get("scores") or {}).get(coin)) if isinstance(payload, dict) else None if not isinstance(row, dict): return ["confluence_coin_missing"] blockers = [str(value) for value in (row.get("blockers") or [])] reliability = _d(((row.get("components") or {}).get("reliability"))) universal = [value for value in blockers if value in {"weak_liquidity", "event_risk_red", "data_unreliable"}] if reliability < Decimal("0.80"): universal.append("low_reliability") if side == "short" or strategy_id == MOMENTUM_RESEARCH_ID: return [f"confluence_{value}" for value in dict.fromkeys(universal)] recommendation = str(row.get("recommendation") or "block_new_entry") if recommendation != "paper_candidate": return ["confluence_block_new_entry", *[f"confluence_{value}" for value in blockers]] return [] def _segment_pnl(*, side: str, entry: Decimal, exit_price: Decimal, size: Decimal) -> Decimal: return (exit_price - entry) * size if side == "long" else (entry - exit_price) * size def _exit_price(*, side: str, raw: Decimal, slip_pct: Decimal) -> Decimal: return raw * (Decimal("1") - slip_pct / Decimal("100")) if side == "long" else raw * (Decimal("1") + slip_pct / Decimal("100")) def _journal_exit( *, runtime_dir: Path, strategy_id: str, run_id: str, coin: str, pos: dict[str, Any], exit_price: Decimal, exit_size: Decimal, reason: str, event: str, final: bool, ) -> Decimal: entry = _d(pos.get("entry")) original_size = _d(pos.get("original_size"), str(pos.get("size") or 0)) entry_fee_total = _d(pos.get("entry_fee_usd")) allocation = exit_size / original_size if original_size > 0 else Decimal("0") # Entry/exit slippage and spread are already embedded in the adverse fill # prices. Subtracting them again would double-count execution costs. entry_cost_share = entry_fee_total * allocation exit_fee = exit_price * exit_size * FEE_RATE exit_slippage_embedded = exit_price * exit_size * _d(pos.get("exit_slippage_pct"), "0.02") / Decimal("100") funding = entry * exit_size * _d(pos.get("funding_rate_hourly_pct")) / Decimal("100") * Decimal(int(pos.get("bars_held") or 0)) * Decimal("0.25") if str(pos.get("side")) == "short": funding = -funding gross = _segment_pnl(side=str(pos.get("side")), entry=entry, exit_price=exit_price, size=exit_size) segment_net = gross - entry_cost_share - exit_fee - funding accumulated = _d(pos.get("realized_net_pnl_accum")) lifecycle_net = accumulated + segment_net _append(runtime_dir / "trade_journal.jsonl", { "event": event, "strategy_id": strategy_id, "strategy_version": str(pos.get("strategy_version") or _strategy_version(strategy_id)), "run_id": run_id, "data_window_id": pos.get("data_window_id"), "setup": pos.get("setup"), "coin": coin, "side": pos.get("side"), "entry_price": str(entry), "exit_price": str(exit_price), "size": str(exit_size), "remaining_size": "0" if final else str(_d(pos.get("size")) - exit_size), "exit_reason": reason, "bars_held": int(pos.get("bars_held") or 0), "gross_pnl_usd": str(gross), "segment_net_pnl_usd": str(segment_net), "net_pnl_usd": str(lifecycle_net if final else segment_net), "lifecycle_net_pnl_usd": str(lifecycle_net), "entry_cost_share_usd": str(entry_cost_share), "exit_fee_usd": str(exit_fee), "exit_slippage_embedded_usd": str(exit_slippage_embedded), "funding_cost_usd": str(funding), "mfe_pct": str(pos.get("mfe_pct") or "0"), "mae_pct": str(pos.get("mae_pct") or "0"), "tp1_hit": bool(pos.get("tp1_hit")), "proxy_inputs_used": False, "research": True, }) return segment_net def _manage_positions( *, state: dict[str, Any], features: dict[str, TrendRetestFeatures], runtime_dir: Path, strategy_id: str, run_id: str, ) -> tuple[int, set[str]]: positions = state.setdefault("open_positions", {}) closed = 0 closed_coins: set[str] = set() for coin, pos in list(positions.items()): feature = features.get(coin) if feature is None or not isinstance(pos, dict) or int(pos.get("last_managed_candle_ts") or 0) >= feature.candle_ts: continue side = str(pos.get("side")) entry = _d(pos.get("entry")) size = _d(pos.get("size")) stop = _d(pos.get("stop_loss")) tp1 = _d(pos.get("tp1")) slip = max(_d(pos.get("exit_slippage_pct"), "0.02"), feature.spread_pct / Decimal("2"), feature.buy_impact_1k_pct if side == "long" else feature.sell_impact_1k_pct) pos["last_managed_candle_ts"] = feature.candle_ts pos["bars_held"] = int(pos.get("bars_held") or 0) + 1 if side == "long": favorable = (feature.high_15m - entry) / entry * Decimal("100") adverse = (feature.low_15m - entry) / entry * Decimal("100") pos["high_watermark"] = str(max(_d(pos.get("high_watermark"), str(entry)), feature.high_15m)) stop_touched = feature.low_15m <= stop tp_touched = feature.high_15m >= tp1 else: favorable = (entry - feature.low_15m) / entry * Decimal("100") adverse = (entry - feature.high_15m) / entry * Decimal("100") pos["low_watermark"] = str(min(_d(pos.get("low_watermark"), str(entry)), feature.low_15m)) stop_touched = feature.high_15m >= stop tp_touched = feature.low_15m <= tp1 pos["mfe_pct"] = str(max(_d(pos.get("mfe_pct")), favorable)) pos["mae_pct"] = str(min(_d(pos.get("mae_pct")), adverse)) reason: str | None = None raw_exit: Decimal | None = None tp1_hit_this_bar = False if stop_touched: gap = (side == "long" and feature.open_15m < stop) or (side == "short" and feature.open_15m > stop) raw_exit = feature.open_15m if gap else stop reason = "stop_gap_slippage" if gap else ("trailing_or_break_even_stop" if pos.get("tp1_hit") else "stop_loss") elif not pos.get("tp1_hit") and tp_touched: partial_size = size * TP1_FRACTION partial_price = _exit_price(side=side, raw=tp1, slip_pct=slip) net = _journal_exit(runtime_dir=runtime_dir, strategy_id=strategy_id, run_id=str(pos.get("run_id") or run_id), coin=coin, pos=pos, exit_price=partial_price, exit_size=partial_size, reason="tp1_partial_50pct", event="partial_exit", final=False) pos["realized_net_pnl_accum"] = str(_d(pos.get("realized_net_pnl_accum")) + net) pos["size"] = str(size - partial_size) pos["tp1_hit"] = True tp1_hit_this_bar = True pos["stop_loss"] = str(entry) size = size - partial_size elif int(pos.get("bars_held") or 0) >= DEAD_FISH_BARS and _d(pos.get("mfe_pct")) < DEAD_FISH_MIN_MFE_PCT: raw_exit = feature.close_15m reason = "dead_fish_time_exit" elif int(pos.get("bars_held") or 0) >= MAX_HOLD_BARS: raw_exit = feature.close_15m reason = "max_hold_time_exit" if reason is None and pos.get("tp1_hit") and not tp1_hit_this_bar: trail_pct = max(Decimal("0.35"), feature.atr_pct * Decimal("0.80")) if side == "long": trail = _d(pos.get("high_watermark")) * (Decimal("1") - trail_pct / Decimal("100")) pos["stop_loss"] = str(max(entry, _d(pos.get("stop_loss")), trail)) else: trail = _d(pos.get("low_watermark")) * (Decimal("1") + trail_pct / Decimal("100")) pos["stop_loss"] = str(min(entry, _d(pos.get("stop_loss")), trail)) if reason is not None and raw_exit is not None: final_price = _exit_price(side=side, raw=raw_exit, slip_pct=slip) _journal_exit(runtime_dir=runtime_dir, strategy_id=strategy_id, run_id=str(pos.get("run_id") or run_id), coin=coin, pos=pos, exit_price=final_price, exit_size=_d(pos.get("size")), reason=reason, event="exit", final=True) del positions[coin] closed += 1 closed_coins.add(coin) return closed, closed_coins def scan_once(strategy_id: str, coins: list[str], *, env: str = "mainnet", md: Any | None = None) -> dict[str, Any]: _assert_paper_only() if strategy_id not in SUPPORTED: raise ValueError(f"unsupported strategy_id: {strategy_id}") side = SUPPORTED[strategy_id] strategy_version = _strategy_version(strategy_id) runtime_dir = ROOT / 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 {} run_id = str(os.getenv("CTB_RUN_ID") or state.get("run_id") or uuid.uuid4().hex) state["run_id"] = run_id state["strategy_version"] = strategy_version resolved_md: Any = md if md is not None else HyperliquidMarketData(env=env) now_ms = int(time.time() * 1000) errors: dict[str, str] = {} try: contexts = _asset_contexts(resolved_md.get_meta_and_asset_ctxs()) except Exception as exc: contexts = {} errors["asset_contexts"] = type(exc).__name__ raw_features: dict[str, TrendRetestFeatures] = {} for coin in coins: try: raw_features[coin] = _fetch_feature(resolved_md, coin, now_ms=now_ms, asset_ctx=contexts.get(coin, {})) except Exception as exc: errors[coin] = f"{type(exc).__name__}:{str(exc)[:120]}" features = _cross_section(raw_features) closed, closed_this_tick = _manage_positions(state=state, features=features, runtime_dir=runtime_dir, strategy_id=strategy_id, run_id=run_id) tv_contexts = load_tradingview_latest() external_context = load_external_context(Path("runtime/research/external_market_context_latest.json")) executor = PaperExecutor(cost_model=CostModel()) opened = 0 blocked: dict[str, int] = {} last_windows = state.setdefault("last_entry_window_by_coin", {}) for coin in coins: feature = features.get(coin) if feature is None: reasons = ["live_feature_data_unavailable"] for reason in reasons: blocked[reason] = blocked.get(reason, 0) + 1 _append(runtime_dir / "signal_journal.jsonl", {"strategy_id": strategy_id, "strategy_version": strategy_version, "run_id": run_id, "coin": coin, "side": side, "would_enter": False, "block_reason": reasons, "data_error": errors.get(coin), "proxy_inputs_used": False, "final_decision": "blocked:" + "+".join(reasons), "research": True}) continue tv = tv_contexts.get(coin) reasons = _strategy_entry_blockers(strategy_id, feature, tv=tv) reasons.extend(_confluence_blockers(coin, side, strategy_id=strategy_id)) external_reasons, external_sources = external_entry_blockers(coin, side, external_context) reasons.extend(external_reasons) if coin in state.get("open_positions", {}): reasons.append("already_open") if coin in closed_this_tick: reasons.append("closed_this_tick") if last_windows.get(coin) == feature.data_window_id: reasons.append("duplicate_data_window") reasons = list(dict.fromkeys(reasons)) intent = None if reasons else _strategy_intent(strategy_id, feature, side=side, client_order_id="paper-v77-" + uuid.uuid4().hex) if intent is None and not reasons: reasons.append("risk_sizing_below_minimum") would_enter = intent is not None and not reasons tv_impulse = classify_tradingview_impulse(tv, side=side) decision = "blocked:" + "+".join(reasons) if would_enter and intent is not None: impact = feature.buy_impact_1k_pct if side == "long" else feature.sell_impact_1k_pct fill = executor.execute(intent, mark_price=feature.current_price, half_spread_pct=feature.spread_pct / Decimal("2"), expected_move_pct=feature.atr_pct * Decimal("1.20"), depth_penalty_pct=impact, hold_hours=Decimal("4")) if fill.blocked_by_cost: reasons.append(f"cost_model_{fill.blocked_reason}") would_enter = False for reason in reasons: blocked[reason] = blocked.get(reason, 0) + 1 decision = "blocked:" + "+".join(reasons) _append(runtime_dir / "signal_journal.jsonl", {"strategy_id": strategy_id, "strategy_version": strategy_version, "run_id": run_id, "coin": coin, "side": side, "would_enter": False, "block_reason": reasons, "features": feature_journal(feature), "tradingview_context": context_to_journal_dict(tv), "tradingview_impulse": tv_impulse, "external_context_status": external_context.get("status"), "external_context_fresh": external_context.get("fresh"), "external_source_attribution": list(external_sources), "final_decision": decision, "proxy_inputs_used": False, "research": True}) continue entry = _d(fill.fill_price) state.setdefault("open_positions", {})[coin] = { "strategy_id": strategy_id, "strategy_version": strategy_version, "run_id": run_id, "data_window_id": feature.data_window_id, "setup": intent.reason, "side": side, "entry": str(entry), "size": str(intent.size), "original_size": str(intent.size), "stop_loss": str(intent.stop_loss), "tp1": str(intent.take_profit), "tp1_hit": False, "entry_fee_usd": str(fill.fee_usd), "entry_execution_slippage_embedded_usd": str(abs(entry - feature.current_price) * intent.size), "exit_slippage_pct": str(max(Decimal("0.02"), impact, feature.spread_pct / Decimal("2"))), "funding_rate_hourly_pct": str(feature.funding_rate_hourly_pct), "entry_candle_ts": feature.candle_ts, "last_managed_candle_ts": feature.candle_ts, "bars_held": 0, "mfe_pct": "0", "mae_pct": "0", "high_watermark": str(entry), "low_watermark": str(entry), "realized_net_pnl_accum": "0", "tradingview_impulse": tv_impulse, "external_source_attribution": list(external_sources), } last_windows[coin] = feature.data_window_id _append(runtime_dir / "trade_journal.jsonl", {"event": "entry", "strategy_id": strategy_id, "strategy_version": strategy_version, "run_id": run_id, "data_window_id": feature.data_window_id, "setup": intent.reason, "coin": coin, "side": side, "entry_price": str(entry), "size": str(intent.size), "stop_loss": str(intent.stop_loss), "tp1": str(intent.take_profit), "entry_fee_usd": str(fill.fee_usd), "entry_execution_slippage_embedded_usd": str(abs(entry - feature.current_price) * intent.size), "proxy_inputs_used": False, "research": True, "tradingview_impulse": tv_impulse, "external_source_attribution": list(external_sources)}) opened += 1 decision = "paper_opened_v77" else: for reason in reasons: blocked[reason] = blocked.get(reason, 0) + 1 _append(runtime_dir / "signal_journal.jsonl", {"strategy_id": strategy_id, "strategy_version": strategy_version, "run_id": run_id, "coin": coin, "side": side, "would_enter": would_enter, "block_reason": reasons, "features": feature_journal(feature), "tradingview_context": context_to_journal_dict(tv), "tradingview_impulse": tv_impulse, "external_context_status": external_context.get("status"), "external_context_fresh": external_context.get("fresh"), "external_source_attribution": list(external_sources), "final_decision": decision, "proxy_inputs_used": False, "research": True}) _save(state_path, state) status = "degraded" if errors and not features else "ok" result = {"status": status, "strategy_id": strategy_id, "strategy_version": strategy_version, "run_id": run_id, "side": side, "signals_seen": len(coins), "features_loaded": len(features), "opened": opened, "closed": closed, "blocked": blocked, "data_errors": errors, "runtime_dir": str(runtime_dir), "paper_trading": True, "live_order_allowed": False, "mainnet_signed_action": False} _append(runtime_dir / "runtime_health.jsonl", {"event": "scan_complete", **result}) return result def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="v77 live-data trend-retest paper runtime; refuses all live/signed flags.") parser.add_argument("--strategy-id", choices=sorted(SUPPORTED), default=LONG_STRATEGY_ID) parser.add_argument("--coins", default="BTC,ETH,SOL,LINK") 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_PAPER_TRADING", "true") os.environ.setdefault("CTB_LIVE_TRADING_ALLOWED", "false") os.environ.setdefault("CTB_LIVE_ORDER_ALLOWED", "false") os.environ.setdefault("HL_MAINNET_SIGNED_ACTION", "false") os.environ["CTB_STRATEGY_ID"] = args.strategy_id os.environ.setdefault("CTB_RUN_ID", uuid.uuid4().hex) runtime_dir = ROOT / args.strategy_id runtime_dir.mkdir(parents=True, exist_ok=True) (runtime_dir / "bot.pid").write_text(str(os.getpid()), encoding="utf-8") results = [] coins = [coin.strip().upper() for coin in args.coins.split(",") if coin.strip()] 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": "degraded" if any(row["status"] == "degraded" for row in results) else "ok", "results": results, "paper_trading": 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())