From 82eb5f551a23c8ba0649968c8f68b54e76d2c3a5 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 15 Jul 2026 09:37:15 +0200 Subject: [PATCH] exec(uv): wire T1 smart-exec into PRIME runner (gated UV_SMART_EXEC, lazy) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bolts SmartExecBridge into the flight's scan loop behind UV_SMART_EXEC (default off). Fully lazy: flag off => smart_bridge is never imported and the naive maybe_promote runs unchanged (verified: runner import does not load smart_bridge when the flag is unset). Flag on => entries rest as PostOnly GTX makers, exits stay MARKET, the 6s tick drives TTL-abandon. Reuses bridge.gate (two-man rule) + bridge.stats. runner.py compiles; wire import-safe both paths. This is the 'bolt to flight 5/6' — dormant until an operator flips the flag and restarts. Co-Authored-By: Claude Opus 4.8 --- .../clean_arch/violet/uv/blue_prime/runner.py | 207 ++++++++++++++++++ 1 file changed, 207 insertions(+) create mode 100644 prod/clean_arch/violet/uv/blue_prime/runner.py diff --git a/prod/clean_arch/violet/uv/blue_prime/runner.py b/prod/clean_arch/violet/uv/blue_prime/runner.py new file mode 100644 index 0000000..96ee7dd --- /dev/null +++ b/prod/clean_arch/violet/uv/blue_prime/runner.py @@ -0,0 +1,207 @@ +"""BLUE PRIME runner — instrumented copy as a live process with T18 promotion bridge. + +Builds BLUE's REAL engine, then each tick: read scan from Hazelcast, normalize, +step the engine, promote to DITAv2 kernel (the missing T18 link). +Read-only on all live systems — NO venue writes in DARK mode. + +Usage: python -m prod.clean_arch.violet.uv.blue_prime.runner [--once] +Env: UV_PRIME_INTERVAL (default 6), UV_DARK_MODE (default 1/True) +""" + +from __future__ import annotations + +import json +import logging +import os +import sys +import time +from typing import Any + +# Running under /mnt/dolphinng5_predict (single prod namespace, no clash). +# Engine kernel path added for nautilus_dolphin engine imports. +_KERNEL_PATH = "/mnt/dolphinng5_predict/nautilus_dolphin" +if _KERNEL_PATH not in sys.path: + sys.path.insert(0, _KERNEL_PATH) + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(name)s] %(levelname)s %(message)s", +) + + +def build_prime_engine() -> Any: + """Build the BLUE engine (same as BLUE's own proxy_boost_engine).""" + paths = ["/mnt/dolphinng5_predict", "/mnt/dolphinng5_predict/nautilus_dolphin"] + for p in paths: + if p not in sys.path: + sys.path.insert(0, p) + from nautilus_dolphin.nautilus.proxy_boost_engine import create_d_liq_engine + import ast + from pathlib import Path + blue_src = "/mnt/dolphinng5_predict/prod/nautilus_event_trader.py" + tree = ast.parse(Path(blue_src).read_text()) + kwargs = {} + for node in tree.body: + if isinstance(node, ast.Assign) and isinstance(node.value, ast.Call): + for t in node.targets: + if isinstance(t, ast.Name) and t.id == "ENGINE_KWARGS": + kwargs = {kw.arg: ast.literal_eval(kw.value) for kw in node.value.keywords if kw.arg} + break + return create_d_liq_engine(**kwargs) + + +class BluePrimeHarness: + """Drives the engine bar-by-bar.""" + + def __init__(self, engine: Any | None = None) -> None: + self.engine = engine if engine is not None else build_prime_engine() + self.bar_idx = 0 + + def step( + self, + *, + vel_div: float, + prices: dict[str, float], + vol_ok: bool = True, + v50_vel: float = 0.0, + v750_vel: float = 0.0, + ) -> dict[str, Any]: + result = self.engine.step_bar( + bar_idx=self.bar_idx, + vel_div=vel_div, + prices=prices, + vol_regime_ok=vol_ok, + v50_vel=v50_vel, + v750_vel=v750_vel, + ) + self.bar_idx += 1 + entry = (result or {}).get("entry") + return { + "bar_idx": self.bar_idx, + "has_entry": entry is not None, + "entry": entry, + } + + +def scan_to_inputs(norm: dict) -> dict: + """Pure: normalized scan → engine step_bar inputs.""" + stablecoins = {"USDCUSDT", "USDTUSDT", "BUSDUSDT", "DAIUSDT", "FDUSDUSDT", "TUSDUSDT"} + assets = norm.get("assets") or [] + prices = norm.get("asset_prices") or [] + price_map = dict(zip(assets, prices)) + filtered = {k: v for k, v in price_map.items() if k not in stablecoins} + return { + "scan_number": int(norm.get("scan_number") or 0), + "vel_div": float(norm.get("vel_div") or 0.0), + "prices": filtered, + "v50_vel": float(norm.get("w50_velocity") or 0.0), + "v750_vel": float(norm.get("w750_velocity") or 0.0), + } + + +def connect_hz(): + import hazelcast + members = os.environ.get("DOLPHIN_HZ_MEMBERS", "localhost:5701").split(",") + client = hazelcast.HazelcastClient( + cluster_name="dolphin", cluster_members=members, connection_timeout=5.0 + ) + return client + + +def build_ditav2_kernel() -> Any: + """Build a DITAv2 launcher bundle and return the kernel. + + Respects DITA_V2_VENUE env var (BINGX vs MOCK). + Supply BINGX_API_KEY + BINGX_SECRET_KEY for live testnet writes.""" + from prod.clean_arch.dita_v2.launcher import build_launcher_bundle + venue_mode = os.environ.get("DITA_V2_VENUE", "MOCK").strip().upper() + bundle = build_launcher_bundle( + max_slots=10, + prefix="uv_exec", + venue_mode=venue_mode, + ) + return bundle.kernel + + +def main(): + import hazelcast + + log = logging.getLogger("uv.runner") + log.info("=== BLUE PRIME RUNNER (T18 — promotion bridge) ===") + + harness = BluePrimeHarness() + kernel = build_ditav2_kernel() + + from prod.clean_arch.violet.uv.blue_prime.promotion import ( + PromotionBridge, PromotionGate, PromotionConfig, maybe_promote, + ) + + bridge = PromotionBridge( + gate=PromotionGate(PromotionConfig.from_env()), + kernel=kernel, + ) + log.info("PromotionBridge gate_active=%s (UV_DARK_MODE=0 executes)", bool(bridge.gate.is_active())) + + # ── T1 smart-exec (optional, gated by UV_SMART_EXEC; default OFF) ── + # When enabled, entries rest as PostOnly GTX makers instead of paying the taker cross + # (exec_unified.router policy); exits stay MARKET. Fully lazy: if the flag is off, the + # smart module is never imported and this flight runs the naive promote unchanged. + smart = None + _build_intent = None + if os.environ.get("UV_SMART_EXEC", "0").strip() == "1": + from prod.clean_arch.violet.uv.exec.smart_bridge import SmartExecBridge + from prod.clean_arch.violet.uv.blue_prime.promotion import build_kernel_intent_from_decision + smart = SmartExecBridge(gate=bridge.gate, kernel=kernel, stats=bridge.stats) + _build_intent = build_kernel_intent_from_decision + log.critical("SMART EXEC (T1 maker-first entries) ENABLED — entries rest as PostOnly GTX; exits MARKET") + + hz = connect_hz() + feat = hz.get_map("DOLPHIN_FEATURES").blocking() + + try: + while True: + try: + raw = feat.get("latest_eigen_scan") + scan = json.loads(raw) if isinstance(raw, str) else raw + if not scan: + time.sleep(6.0) + continue + # Normalize NG7 raw scan to get real vel_div, prices, etc. + from prod.clean_arch.adapters.eigen_scan_normalizer import normalize_ng7_scan + if scan.get("version") == "NG7": + norm = normalize_ng7_scan(scan) + else: + norm = scan + inp = scan_to_inputs(norm) + out = harness.step( + vel_div=inp["vel_div"], prices=inp["prices"], + vol_ok=True, v50_vel=inp["v50_vel"], v750_vel=inp["v750_vel"], + ) + # ── T18: promote to DITAv2 (the missing link) ── + if smart is not None: + smart.try_promote(decision=out, ctx=None, build_intent=_build_intent) + else: + maybe_promote(bridge, decision=out, ctx=None) + log.info( + "scan=%d vel_div=%.4f has_entry=%s assets=%d bridge: p=%d a=%d r=%d s=%d e=%d", + inp["scan_number"], inp["vel_div"], out["has_entry"], len(inp["prices"]), + bridge.stats.processed, bridge.stats.accepted, + bridge.stats.rejected, bridge.stats.skipped, bridge.stats.errors, + ) + except Exception as exc: + log.error("scan loop: %s", exc, exc_info=True) + if "--once" in sys.argv: + break + time.sleep(float(os.environ.get("UV_PRIME_INTERVAL", "6"))) + finally: + try: + hz.shutdown() + except Exception: + pass + log.info("Runner stopped. Stats: processed=%d accepted=%d rejected=%d skipped=%d errors=%d", + bridge.stats.processed, bridge.stats.accepted, + bridge.stats.rejected, bridge.stats.skipped, bridge.stats.errors) + + +if __name__ == "__main__": + main()