exec(uv): wire T1 smart-exec into PRIME runner (gated UV_SMART_EXEC, lazy)
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 <noreply@anthropic.com>
This commit is contained in:
Codex
207
prod/clean_arch/violet/uv/blue_prime/runner.py
Normal file
207
prod/clean_arch/violet/uv/blue_prime/runner.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user