from __future__ import annotations import argparse 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.execution.cost_model import CostModel from src.execution.maker_taker_shadow import advance_shadow, submit_shadow from src.execution.paper_executor import PaperExecutor from src.hyperliquid.market_data import HyperliquidMarketData from src.research.external_market_context import load_external_context from src.strategies.v78_research_archetypes import ( LIQUIDATION_REVERSAL_ID, REGIME_ROUTER_ID, RELATIVE_VALUE_ID, STRATEGY_VERSION, ResearchDecision, build_paper_intent, classify_regime, confirmed_breakout_retest, funding_carry_advice, liquidation_sweep_reversal, portfolio_allocation_advice, range_mean_reversion, relative_value_pair, ) 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") SUPPORTED = {REGIME_ROUTER_ID, RELATIVE_VALUE_ID, LIQUIDATION_REVERSAL_ID} FEE_RATE = Decimal("0.00045") MAX_HOLD_BARS = 16 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.setdefault("paper_trading", True) row.setdefault("research_only", 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_research_only() -> None: if os.getenv("CTB_PAPER_TRADING", "").lower() != "true": raise PermissionError("v78 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("v78 research runtime requires explicit false live/signed flags: " + ",".join(unsafe)) def _manage_positions(state: dict[str, Any], features: dict[str, Any], runtime_dir: Path, run_id: str) -> int: closed = 0 positions = state.setdefault("open_positions", {}) for coin, pos in list(positions.items()): feature = features.get(coin) if feature is None or int(pos.get("last_candle_ts") or 0) >= feature.candle_ts: continue pos["last_candle_ts"] = feature.candle_ts pos["bars_held"] = int(pos.get("bars_held") or 0) + 1 side = str(pos.get("side")) entry = _d(pos.get("entry_price")) stop = _d(pos.get("stop_loss")) target = _d(pos.get("take_profit")) size = _d(pos.get("size")) 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 reason = "stop_loss" if stop_hit else "take_profit" if target_hit else "time_exit" if int(pos["bars_held"]) >= MAX_HOLD_BARS else None if not reason: continue raw_exit = stop if stop_hit else target if target_hit else feature.close_15m slip = max(Decimal("0.02"), feature.spread_pct / Decimal("2"), feature.sell_impact_1k_pct if side == "long" else feature.buy_impact_1k_pct) exit_price = raw_exit * (Decimal("1") - slip / Decimal("100")) if side == "long" else raw_exit * (Decimal("1") + slip / Decimal("100")) gross = (exit_price - entry) * size if side == "long" else (entry - exit_price) * size fees = entry * size * FEE_RATE + exit_price * size * FEE_RATE funding = entry * size * _d(pos.get("funding_rate_hourly_pct")) / Decimal("100") * Decimal(int(pos["bars_held"])) * Decimal("0.25") if side == "short": funding = -funding net = gross - fees - funding _append(runtime_dir / "trade_journal.jsonl", {"event": "exit", "strategy_id": pos.get("strategy_id"), "strategy_version": STRATEGY_VERSION, "run_id": pos.get("run_id") or run_id, "data_window_id": pos.get("data_window_id"), "strategy_family": pos.get("strategy_family"), "source_attribution": pos.get("source_attribution"), "coin": coin, "side": side, "entry_price": str(entry), "exit_price": str(exit_price), "size": str(size), "exit_reason": reason, "bars_held": pos["bars_held"], "gross_pnl_usd": str(gross), "fees_usd": str(fees), "funding_cost_usd": str(funding), "net_pnl_usd": str(net), "proxy_inputs_used": False}) del positions[coin] closed += 1 return closed def _open(state: dict[str, Any], runtime_dir: Path, run_id: str, feature: Any, decision: ResearchDecision, strategy_id: str, executor: PaperExecutor) -> bool: if feature.coin in state.setdefault("open_positions", {}): return False intent = build_paper_intent(feature, decision, strategy_id=strategy_id) if intent is None: return False impact = feature.buy_impact_1k_pct if decision.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, depth_penalty_pct=impact, hold_hours=Decimal("4")) if fill.blocked_by_cost: return False entry = _d(fill.fill_price) state["open_positions"][feature.coin] = {"strategy_id": strategy_id, "strategy_version": STRATEGY_VERSION, "strategy_family": decision.strategy_family, "source_attribution": list(decision.source_attribution), "run_id": run_id, "data_window_id": feature.data_window_id, "side": decision.side, "entry_price": str(entry), "size": str(intent.size), "stop_loss": str(intent.stop_loss), "take_profit": str(intent.take_profit), "funding_rate_hourly_pct": str(feature.funding_rate_hourly_pct), "bars_held": 0, "last_candle_ts": feature.candle_ts} _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, "strategy_family": decision.strategy_family, "source_attribution": list(decision.source_attribution), "coin": feature.coin, "side": decision.side, "entry_price": str(entry), "size": str(intent.size), "stop_loss": str(intent.stop_loss), "take_profit": str(intent.take_profit), "entry_fee_usd": str(fill.fee_usd), "proxy_inputs_used": False}) return True def scan_once(strategy_id: str, coins: list[str], *, md: Any | None = None) -> dict[str, Any]: _assert_research_only() if strategy_id not in SUPPORTED: raise ValueError(f"unsupported strategy_id: {strategy_id}") runtime_dir = ROOT / strategy_id 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.update({"run_id": run_id, "strategy_version": STRATEGY_VERSION}) resolved_md = md or HyperliquidMarketData("mainnet") 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 = {} for coin in coins: try: raw[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)[:100]}" features = _cross_section(raw) external = load_external_context(EXTERNAL_CONTEXT) fred = (((external.get("sources") or {}).get("fred_macro") or {}).get("data") or {}) vix = _d((fred.get("vix_close") or {}).get("value")) event_red = vix >= Decimal("35") regime = classify_regime(features, event_risk_red=event_red) closed = _manage_positions(state, features, runtime_dir, run_id) maker_state = state.setdefault("maker_taker_shadow", {}) maker_events = advance_shadow(maker_state, features) for event in maker_events: _append(runtime_dir / "maker_taker_shadow_journal.jsonl", event) executor = PaperExecutor(cost_model=CostModel()) opened = 0 blocked: dict[str, int] = {} last_windows = state.setdefault("last_entry_window_by_coin", {}) decisions: dict[str, ResearchDecision] = {} if strategy_id == REGIME_ROUTER_ID: for coin, feature in features.items(): if regime.regime in {"trend_up", "trend_down"}: decision = confirmed_breakout_retest(feature, regime=regime.regime, external=external) elif regime.regime == "range": decision = range_mean_reversion(feature, regime=regime.regime, external=external) elif regime.regime == "high_volatility": decision = liquidation_sweep_reversal(feature, external=external) else: decision = ResearchDecision("regime_router", "no_trade", None, Decimal("0"), ("regime_no_trade",), "regime_block") decisions[coin] = decision elif strategy_id == LIQUIDATION_REVERSAL_ID: decisions = {coin: liquidation_sweep_reversal(feature, external=external) for coin, feature in features.items()} else: pair = relative_value_pair(features, regime=regime.regime, external=external) _append(runtime_dir / "pair_signal_journal.jsonl", {"strategy_id": strategy_id, "strategy_version": STRATEGY_VERSION, "run_id": run_id, "regime": regime.to_dict(), "pair_decision": pair, "external_context_status": external.get("status"), "external_context_fresh": external.get("fresh")}) if pair.get("action") == "paper_pair_candidate": legs = list(pair.get("legs") or []) pair_runtime_blockers: list[str] = [] for leg in legs: leg_coin = str(leg.get("coin")) leg_feature = features.get(leg_coin) if leg_coin in state.get("open_positions", {}): pair_runtime_blockers.append(f"pair_leg_already_open:{leg_coin}") if leg_feature is None: pair_runtime_blockers.append(f"pair_leg_feature_missing:{leg_coin}") elif last_windows.get(leg_coin) == leg_feature.data_window_id: pair_runtime_blockers.append(f"pair_leg_duplicate_window:{leg_coin}") if pair_runtime_blockers: _append(runtime_dir / "pair_signal_journal.jsonl", {"strategy_id": strategy_id, "strategy_version": STRATEGY_VERSION, "run_id": run_id, "event": "pair_runtime_blocked", "blockers": pair_runtime_blockers, "research_only": True}) else: for leg in legs: coin = str(leg.get("coin")) feature = features.get(coin) side = str(leg.get("side")) if feature: decisions[coin] = ResearchDecision("market_neutral_relative_value", "paper_candidate", side, Decimal(str(leg.get("score") or 0)), (), "cross_sectional_pair", ("hyperliquid_live_readonly_ohlcv_l2", "coingecko_markets", "binance_futures_crowding")) for coin, decision in decisions.items(): feature = features[coin] reasons = list(decision.blockers) if last_windows.get(coin) == feature.data_window_id: reasons.append("duplicate_data_window") if coin in state.get("open_positions", {}): reasons.append("already_open") effective = ResearchDecision(decision.strategy_family, "no_trade" if reasons else decision.action, decision.side, decision.score, tuple(dict.fromkeys(reasons)), decision.reason, decision.source_attribution) if not reasons and _open(state, runtime_dir, run_id, feature, effective, strategy_id, executor): opened += 1 last_windows[coin] = feature.data_window_id if effective.strategy_family == "range_mean_reversion": position = state["open_positions"][coin] shadow_event = submit_shadow( maker_state, coin=coin, side=str(effective.side), feature=feature, size=_d(position["size"]), stop_loss=_d(position["stop_loss"]), take_profit=_d(position["take_profit"]), source_window_id=feature.data_window_id, strategy_id=strategy_id, ) _append(runtime_dir / "maker_taker_shadow_journal.jsonl", shadow_event) else: for reason in effective.blockers: 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, "data_window_id": feature.data_window_id, "coin": coin, "regime": regime.to_dict(), "decision": effective.to_dict(), "external_context_status": external.get("status"), "external_context_fresh": external.get("fresh"), "external_blockers": external.get("blockers"), "proxy_inputs_used": False}) advisory = {"funding_carry": funding_carry_advice(features), "portfolio_allocation": portfolio_allocation_advice(features, external)} _save(runtime_dir / "advisory_latest.json", advisory) _save(state_path, state) result = {"status": "degraded" if errors and not features else "ok", "strategy_id": strategy_id, "strategy_version": STRATEGY_VERSION, "run_id": run_id, "regime": regime.to_dict(), "features_loaded": len(features), "opened": opened, "closed": closed, "blocked": blocked, "external_context_status": external.get("status"), "external_context_fresh": external.get("fresh"), "data_errors": errors, "paper_trading": True, "research_only": 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="v78 multi-family research runtime; paper-only and no signed actions.") parser.add_argument("--strategy-id", choices=sorted(SUPPORTED), default=REGIME_ROUTER_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") coins = [coin.strip().upper() for coin in args.coins.split(",") if coin.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": "degraded" if any(row["status"] == "degraded" for row in results) else "ok", "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())