from __future__ import annotations import argparse import json import os import subprocess import sys from collections import Counter from datetime import datetime, timezone from decimal import Decimal from pathlib import Path from typing import Any STALE_JOURNAL_SECONDS = 15 * 60 from src.tools.hl_reconcile_watchdog import assess as assess_reconcile from src.tools.hl_reconcile_watchdog import _load_json_from_reconcile from src.tools.v76_paper_runtime import ANTI_CHASE_ID, BASE_ID, RESEARCH_ID, STRICT_ID, SWING_ID, supervisor_status STRATEGY_ALIASES = { "base": BASE_ID, "strict": STRICT_ID, "research": RESEARCH_ID, "research_probe": RESEARCH_ID, "anti_chase": ANTI_CHASE_ID, "fee_aware_anti_chase": ANTI_CHASE_ID, "swing": SWING_ID, "swing_retest": SWING_ID, "swing_trend_retest": SWING_ID, BASE_ID: BASE_ID, STRICT_ID: STRICT_ID, RESEARCH_ID: RESEARCH_ID, ANTI_CHASE_ID: ANTI_CHASE_ID, SWING_ID: SWING_ID, } def _repo_root() -> Path: return Path(__file__).resolve().parents[2] def _parse_strategies(value: str) -> list[str]: out: list[str] = [] for part in value.split(","): key = part.strip() if not key: continue out.append(STRATEGY_ALIASES.get(key, key)) return out def _read_jsonl(path: Path) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] if not path.exists(): return rows for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): if not line.strip(): continue try: rows.append(json.loads(line)) except json.JSONDecodeError: continue return rows def _file_meta(path: Path, *, now: datetime | None = None) -> dict[str, Any]: if not path.exists(): return {"exists": False, "size_bytes": 0, "age_seconds": None, "fresh": False} now = now or datetime.now(timezone.utc) stat = path.stat() mtime = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc) age = max(0, int((now - mtime).total_seconds())) return { "exists": True, "size_bytes": stat.st_size, "age_seconds": age, "fresh": age <= STALE_JOURNAL_SECONDS, "mtime": mtime.isoformat(), } def _process_identity(pid: int | None, strategy_id: str) -> dict[str, Any]: if not pid: return {"running": False, "env_ok": False, "cmd_ok": False, "paper_only": False} proc = Path(f"/proc/{pid}") if not proc.exists(): return {"running": False, "env_ok": False, "cmd_ok": False, "paper_only": False} try: raw_env = (proc / "environ").read_bytes().decode("utf-8", "ignore") raw_cmd = (proc / "cmdline").read_bytes().decode("utf-8", "ignore") except Exception: return {"running": True, "env_ok": False, "cmd_ok": False, "paper_only": False} env_ok = f"CTB_STRATEGY_ID={strategy_id}" in raw_env and "CTB_PAPER_TRADING=true" in raw_env and "CTB_DRY_RUN=false" in raw_env paper_only = env_ok and "CTB_LIVE_TRADING_ALLOWED=true" not in raw_env and "CTB_LIVE_ORDER_ALLOWED=true" not in raw_env and "HL_MAINNET_SIGNED_ACTION=true" not in raw_env cmd_ok = "src.tools.v76_paper_runtime" in raw_cmd and f"--strategy-id\x00{strategy_id}" in raw_cmd return {"running": True, "env_ok": env_ok, "cmd_ok": cmd_ok, "paper_only": paper_only} def _decimal(value: Any) -> Decimal: try: return Decimal(str(value)) except Exception: return Decimal("0") def _lifecycle_summary(strategy_id: str) -> dict[str, Any]: runtime_dir = _repo_root() / "runtime" / "experiments" / strategy_id trade_journal = runtime_dir / "trade_journal.jsonl" signal_journal = runtime_dir / "signal_journal.jsonl" trades = _read_jsonl(trade_journal) signals = _read_jsonl(signal_journal) state_path = runtime_dir / "state.json" state: dict[str, Any] = {} if state_path.exists(): try: state = json.loads(state_path.read_text(encoding="utf-8")) except json.JSONDecodeError: state = {} raw_open_positions = state.get("open_positions") open_positions: dict[str, Any] = raw_open_positions if isinstance(raw_open_positions, dict) else {} entries = [row for row in trades if row.get("event") == "entry"] exits = [row for row in trades if row.get("event") == "exit" and row.get("exit_reason")] exit_reasons = Counter(str(row.get("exit_reason") or "unknown") for row in exits) open_by_coin = Counter(str(coin).upper() for coin in open_positions.keys()) live_flags = sum(1 for row in trades if row.get("live_order_allowed") is True or row.get("mainnet_signed_action") is True) pnls = [_decimal(row.get("net_pnl_usd") or row.get("realized_pnl_usd")) for row in exits] closed_pnl = sum(pnls, Decimal("0")) wins = [p for p in pnls if p > 0] losses = [p for p in pnls if p < 0] gross_win = sum(wins, Decimal("0")) gross_loss_abs = abs(sum(losses, Decimal("0"))) profit_factor = gross_win / gross_loss_abs if gross_loss_abs > 0 else (Decimal("999") if gross_win > 0 else Decimal("0")) winrate = Decimal(len(wins)) / Decimal(len(pnls)) * Decimal("100") if pnls else Decimal("0") open_notional = Decimal("0") for pos in open_positions.values(): if isinstance(pos, dict): open_notional += _decimal(pos.get("entry")) * _decimal(pos.get("size")) recent_signal_reasons: Counter[str] = Counter() for row in signals[-500:]: reasons = row.get("block_reason") if isinstance(reasons, list): recent_signal_reasons.update(str(reason) for reason in reasons[:5]) elif reasons: recent_signal_reasons.update([str(reasons)]) decision = row.get("final_decision") if decision: recent_signal_reasons.update([str(decision).split(":", 1)[0]]) now = datetime.now(timezone.utc) return { "strategy_id": strategy_id, "entries": len(entries), "exits": len(exits), "open_positions": len(open_positions), "open_by_coin": dict(open_by_coin.most_common()), "open_notional_usd": str(open_notional.quantize(Decimal("0.01"))), "exit_reasons": dict(exit_reasons.most_common()), "closed_net_pnl_usd": str(closed_pnl.quantize(Decimal("0.01"))), "winrate_pct": str(winrate.quantize(Decimal("0.01"))), "profit_factor": str(profit_factor.quantize(Decimal("0.01"))), "recent_signal_blockers": dict(recent_signal_reasons.most_common(12)), "trade_journal_meta": _file_meta(trade_journal, now=now), "signal_journal_meta": _file_meta(signal_journal, now=now), "live_flags": live_flags, "paper_only": live_flags == 0, } def _lifecycle_summaries(strategy_ids: list[str]) -> dict[str, Any]: return {strategy_id: _lifecycle_summary(strategy_id) for strategy_id in strategy_ids} def _start_strategy(strategy_id: str, *, iterations: int, interval_seconds: int, coins: str) -> dict[str, Any]: root = _repo_root() runtime_dir = root / "runtime" / "experiments" / strategy_id runtime_dir.mkdir(parents=True, exist_ok=True) log_path = runtime_dir / "supervisor.log" env = os.environ.copy() env.update({ "CTB_PAPER_TRADING": "true", "CTB_DRY_RUN": "false", "CTB_STRATEGY_ID": strategy_id, }) env.pop("CTB_LIVE_TRADING_ALLOWED", None) cmd = [ sys.executable, "-m", "src.tools.v76_paper_runtime", "--strategy-id", strategy_id, "--coins", coins, "--iterations", str(iterations), "--interval-seconds", str(interval_seconds), "--json", ] handle = log_path.open("ab") proc = subprocess.Popen(cmd, cwd=root, env=env, stdout=handle, stderr=subprocess.STDOUT, start_new_session=True) return {"strategy_id": strategy_id, "started": True, "pid": proc.pid, "log": str(log_path)} def ensure_running(strategy_ids: list[str], *, iterations: int, interval_seconds: int, coins: str) -> dict[str, Any]: reconcile_raw = _load_json_from_reconcile("mainnet") reconcile = assess_reconcile(reconcile_raw) before = supervisor_status(strategy_ids) started: list[dict[str, Any]] = [] blocked_reason = None if reconcile["block_new_entries"] or reconcile["critical"]: blocked_reason = "mainnet_reconcile_not_clean" else: for strategy_id, row in before.items(): if not row.get("running") or not row.get("env_ok") or not row.get("cmd_ok") or not row.get("paper_only"): started.append(_start_strategy(strategy_id, iterations=iterations, interval_seconds=interval_seconds, coins=coins)) after = supervisor_status(strategy_ids) payload = { "timestamp": datetime.now(timezone.utc).isoformat(), "status": "blocked" if blocked_reason else "ok", "mode": "paper_only", "mainnet_signed_action": False, "reconcile": reconcile, "blocked_reason": blocked_reason, "before": before, "started": started, "after": after, "lifecycle": _lifecycle_summaries(strategy_ids), } report_path = _repo_root() / "runtime" / "reports" / "v76_paper_supervisor_latest.json" report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text(json.dumps(payload, indent=2, sort_keys=True, default=str), encoding="utf-8") return payload def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Safe v76 PaperRuntime supervisor: read-only reconcile first, paper-only starts, no signed actions.") parser.add_argument("--strategies", default="strict,research_probe,swing") parser.add_argument("--coins", default="BTC,ETH,SOL,LINK,WLD,SUI,ENA,BCH") parser.add_argument("--iterations", type=int, default=1440) parser.add_argument("--interval-seconds", type=int, default=60) parser.add_argument("--ensure-running", action="store_true") parser.add_argument("--status", action="store_true") parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) strategy_ids = _parse_strategies(args.strategies) if args.ensure_running: payload = ensure_running(strategy_ids, iterations=args.iterations, interval_seconds=args.interval_seconds, coins=args.coins) else: payload = { "status": "ok", "mode": "paper_only", "mainnet_signed_action": False, "supervisor": supervisor_status(strategy_ids), "lifecycle": _lifecycle_summaries(strategy_ids), } print(json.dumps(payload, indent=2, sort_keys=True, default=str) if args.json else payload) return 0 if payload.get("status") == "ok" else 2 if __name__ == "__main__": raise SystemExit(main())