#!/usr/bin/env python3 from __future__ import annotations import json import os import subprocess import sys from decimal import Decimal from pathlib import Path from datetime import datetime, timezone REPO = Path('/home/agent/projects/CryptoTradingBot/Crypto_Agent') STRATEGY = 'candidate_v76_fee_aware_anti_chase' RUNTIME = REPO / 'runtime/experiments' / STRATEGY STATE_DIR = Path('/home/agent/.local/state/CryptoTradingBot') MARKER = STATE_DIR / 'research_live_decision_watchdog_markers.json' LOG = STATE_DIR / 'anti_chase_paper_watchdog.log' def load_markers() -> dict: try: return json.loads(MARKER.read_text()) if MARKER.exists() else {} except Exception: return {} def save_markers(markers: dict) -> None: MARKER.parent.mkdir(parents=True, exist_ok=True) MARKER.write_text(json.dumps(markers, indent=2, sort_keys=True), encoding='utf-8') def run(cmd: list[str], timeout: int = 60) -> subprocess.CompletedProcess: return subprocess.run(cmd, cwd=REPO, text=True, capture_output=True, timeout=timeout) def supervisor() -> dict: cp = run([sys.executable, '-m', 'src.tools.v76_paper_runtime', '--status', '--json']) if cp.returncode != 0: return {'error': cp.stderr[-500:] or cp.stdout[-500:]} try: return json.loads(cp.stdout) except Exception as exc: return {'error': f'bad supervisor json: {exc}'} def trade_stats() -> dict: p = RUNTIME / 'trade_journal.jsonl' rows = [] if p.exists(): for line in p.read_text(encoding='utf-8', errors='replace').splitlines(): if line.strip(): try: rows.append(json.loads(line)) except json.JSONDecodeError: pass entries = [r for r in rows if r.get('event') == 'entry'] exits = [r for r in rows if r.get('event') == 'exit'] pnls = [Decimal(str(r.get('net_pnl_usd') or r.get('realized_pnl_usd') or '0')) for r in exits] wins = [p for p in pnls if p > 0] losses = [p for p in pnls if p < 0] total = sum(pnls, Decimal('0')) last20 = sum(pnls[-20:], Decimal('0')) if pnls else Decimal('0') pf = Decimal('999') if wins and not losses else (sum(wins, Decimal('0')) / abs(sum(losses, Decimal('0'))) if losses else Decimal('0')) state = {} sp = RUNTIME / 'state.json' if sp.exists(): try: state = json.loads(sp.read_text(encoding='utf-8')) except Exception: state = {} return { 'entries': len(entries), 'closed': len(exits), 'net': total, 'last20': last20, 'wins': len(wins), 'losses': len(losses), 'winrate': (Decimal(len(wins)) / Decimal(len(exits)) * Decimal('100')) if exits else Decimal('0'), 'pf': pf, 'open': len(state.get('open_positions', {}) or {}), 'open_coins': sorted((state.get('open_positions', {}) or {}).keys()), } def copy_status_text() -> str: cp = subprocess.run([sys.executable, '/home/agent/.hermes/scripts/ctb_copy_observation_status.py'], cwd=REPO, text=True, capture_output=True, timeout=60) return cp.stdout.strip() if cp.returncode == 0 else '' def start_paper_run() -> None: env = os.environ.copy() env.update({'CTB_STRATEGY_ID': STRATEGY, 'CTB_PAPER_TRADING': 'true', 'CTB_DRY_RUN': 'false'}) LOG.parent.mkdir(parents=True, exist_ok=True) log = LOG.open('ab') subprocess.Popen( [sys.executable, '-m', 'src.tools.v76_paper_runtime', '--strategy-id', STRATEGY, '--iterations', '360', '--interval-seconds', '60', '--json'], cwd=REPO, env=env, stdout=log, stderr=subprocess.STDOUT, start_new_session=True, ) def live_gate(stats: dict) -> tuple[bool, list[str]]: reasons = [] if stats['closed'] < 30: reasons.append('sample_too_small') if stats['net'] <= 0: reasons.append('net_pnl_not_positive') if stats['last20'] <= 0: reasons.append('last20_not_positive') if not (stats['winrate'] >= Decimal('45') or stats['pf'] >= Decimal('1.25')): reasons.append('winrate_pf_below_gate') return not reasons, reasons def main() -> int: markers = load_markers() messages: list[str] = [] sup = supervisor() strategy_sup = ((sup.get('supervisor') or {}).get(STRATEGY) or {}) if isinstance(sup, dict) else {} running = bool(strategy_sup.get('running') and strategy_sup.get('env_ok')) if not running: # Continue only paper/shadow research; no signing/orders. start_paper_run() messages.append('⚙️ Anti-Chase Paper-Run war gestoppt und wurde read-only/paper neu gestartet. Keine Orders, kein Signing.') stats = trade_stats() ready, reasons = live_gate(stats) # Refresh reports opportunistically; no alert on failure. run([sys.executable, '-m', 'src.tools.analyze_trade_timing', '--strategy-id', STRATEGY, '--output', 'runtime/reports/trade_timing_analysis_latest.md'], timeout=180) run([sys.executable, 'paper_scorecard_report.py', '--output', 'runtime/reports/paper_scorecard_latest.md', '--max-bots', '10'], timeout=180) milestone = (stats['closed'] // 10) * 10 if ready and not markers.get('live_ready_alerted'): messages.append( '🟢 Entscheidung nötig: Anti-Chase Paper erfüllt die quantitativen Live-Gates. ' f"closed={stats['closed']}, net={stats['net']:.4f}, last20={stats['last20']:.4f}, " f"PF={stats['pf']:.2f}, winrate={stats['winrate']:.1f}%. Nächster Schritt: Live-Preflight/Reconcile/Stops/Alerts und explizites Sir-GO." ) markers['live_ready_alerted'] = datetime.now(timezone.utc).isoformat() elif stats['closed'] >= 30 and not ready: key = f"negative_milestone_{milestone}" if milestone >= 30 and not markers.get(key): messages.append( '🔴 Genug Paper-Info für eine Research-Entscheidung, aber NICHT live-ready: ' f"closed={stats['closed']}, net={stats['net']:.4f}, last20={stats['last20']:.4f}, " f"PF={stats['pf']:.2f}, winrate={stats['winrate']:.1f}%, Blocker={', '.join(reasons)}. " 'Empfehlung: Strategie/Entry/Exit-Regeln überarbeiten statt live gehen.' ) markers[key] = datetime.now(timezone.utc).isoformat() # Copy final/near-final one-shot alert if it reaches full 84 and allowed remains 0. copy_text = copy_status_text() if 'Fortschritt: 84/84' in copy_text and 'allowed 0' in copy_text and not markers.get('copy_84_zero_allowed'): messages.append('📌 Copy Observation 7d ist vollständig und bleibt bei allowed=0. Entscheidung: Copy nicht live; Watchlist/Copy-Shadow-v2 statt Executor.') markers['copy_84_zero_allowed'] = datetime.now(timezone.utc).isoformat() save_markers(markers) if messages: print('\n'.join(messages)) return 0 if __name__ == '__main__': raise SystemExit(main())