#!/usr/bin/env python3 """Resumable manual replay control worker launcher. This worker does not confirm trades by itself. It launches small Hermes one-shot runs that must inspect TradingView replay for a small batch and write non-final manual review decisions. The wrapper maintains a cursor and lock so work can continue until the queue is complete without overwriting existing artifacts. """ from __future__ import annotations import json import os import re import subprocess import time from collections import Counter from datetime import datetime, timezone from pathlib import Path ROOT = Path('/home/aryy/.hermes/profiles/finance/backtests/xauusd_2026') HYBRID = ROOT / 'hybrid' REVIEWS = HYBRID / 'reviews' MANIFEST = REVIEWS / 'manual_replay_readiness_manifest.json' STATE = REVIEWS / 'manual_replay_progress.json' LOCK = REVIEWS / 'manual_replay_worker.lock' LOG = HYBRID / 'reports' / 'manual_replay_worker.log' LOG.parent.mkdir(parents=True, exist_ok=True) BATCH_SIZE = 3 MAX_CHILD_RUNS = int(os.environ.get('MANUAL_REPLAY_MAX_CHILD_RUNS', '0') or 0) # 0 = until done PROMPT_TEMPLATE = """ You are Aryy Finance manual replay reviewer. Process the next chronological mini-batch of XAUUSD hybrid manual replay items. Project root: /home/aryy/.hermes/profiles/finance/backtests/xauusd_2026 Read these files: - hybrid/reviews/manual_replay_readiness_manifest.json - hybrid/reviews/manual_replay_progress.json - hybrid/data/ohlcv_W.json, D, H4, H1, M30, M15, M5 as needed Process exactly {batch_size} items starting at manifest.manual_replay_items[{start_index}] unless the queue ends. Rules: - Analysis-only. Do not trade, do not execute broker/order actions. - Do not overwrite existing journal/backtest data blindly. - Do not send raw OHLCV in final output; use Python/local scripts to compute summaries/evidence. - Use TradingView MCP Bar Replay when possible for chronological review. If TradingView cannot jump/render a historical date, use local OHLCV chronological evidence and mark decision as 'needs_more_context' rather than accepting. - Do NOT promote to journal unless a candidate is clearly valid with HTF narrative, M30 structure, M15/M5 execution, Fib 50/62 + PD array, and chronological entry/SL/TP evidence. When in doubt, keep as rejected/shadow/needs_more_context. - For this autonomous batch, prefer conservative labels; confirmed journal promotion should be rare and must include exact Entry/SL/TP evidence. If not fully certain, do not promote. - Append/write a mini-batch result file: hybrid/reviews/manual_replay_results/replay_result_{start_index:05d}_{end_index:05d}.json - Update hybrid/reviews/manual_replay_progress.json cursor atomically after writing results. - Validate all changed JSON with python3 -m json.tool. Result schema per item: {{ "candidate_id": "...", "timestamp": "...", "decision": "accepted|rejected|shadow|needs_more_context", "reason": "concise chronological reason", "htf_context": "summary", "m30_structure": "summary", "m15_m5_execution": "summary", "entry": null or number, "stop": null or number, "target": null or number, "r": null or number, "promoted_to_journal": false }} Final response: one compact line with processed count, decisions breakdown, result file, next_index. """.strip() def now_iso() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace('+00:00', 'Z') def log(msg: str) -> None: line = f'{now_iso()} {msg}' print(line, flush=True) with LOG.open('a', encoding='utf-8') as handle: handle.write(line + '\n') def load(path: Path) -> dict: return json.loads(path.read_text(encoding='utf-8')) def save(path: Path, obj: dict) -> None: path.write_text(json.dumps(obj, indent=2, ensure_ascii=False) + '\n', encoding='utf-8') def normalize_state(obj: dict) -> dict: """Accept legacy/simple child-written progress schemas and normalize them.""" if 'cursor' in obj and isinstance(obj.get('cursor'), dict): obj.setdefault('runs', []) obj.setdefault('metadata', {'purpose': 'manual_replay_progress_state', 'created_at': now_iso(), 'source': 'hybrid/reviews/manual_replay_readiness_manifest.json', 'isFinalBacktest': False}) return obj current_index = int(obj.get('current_index', obj.get('next_index', 0)) or 0) total_processed = int(obj.get('total_processed', obj.get('processed_count', current_index)) or current_index) return { 'metadata': {'purpose': 'manual_replay_progress_state', 'created_at': now_iso(), 'source': 'hybrid/reviews/manual_replay_readiness_manifest.json', 'isFinalBacktest': False, 'notes': 'Normalized from legacy/simple schema.'}, 'cursor': {'next_index': current_index, 'processed_count': total_processed, 'accepted_count': int(obj.get('accepted_count', 0) or 0), 'rejected_count': int(obj.get('rejected_count', 0) or 0), 'shadow_count': int(obj.get('shadow_count', 0) or 0), 'needs_more_context_count': int(obj.get('needs_more_context_count', 0) or 0), 'last_candidate_id': obj.get('last_candidate_id'), 'last_updated': obj.get('last_updated')}, 'runs': obj.get('runs', []), } def ensure_state() -> dict: if not STATE.exists(): obj = { 'metadata': {'purpose': 'manual_replay_progress_state', 'created_at': now_iso(), 'source': 'hybrid/reviews/manual_replay_readiness_manifest.json', 'isFinalBacktest': False}, 'cursor': {'next_index': 0, 'processed_count': 0, 'accepted_count': 0, 'rejected_count': 0, 'shadow_count': 0, 'needs_more_context_count': 0, 'last_candidate_id': None, 'last_updated': None}, 'runs': [], } save(STATE, obj) return obj raw = load(STATE) obj = normalize_state(raw) if 'cursor' not in raw: save(STATE, obj) return obj def result_items(payload: object) -> list[dict]: if isinstance(payload, list): return [x for x in payload if isinstance(x, dict)] if isinstance(payload, dict): for key in ('items', 'results', 'decisions'): val = payload.get(key) if isinstance(val, list): return [x for x in val if isinstance(x, dict)] return [] def derive_state_from_results() -> dict: cursor = 0 counts: Counter[str] = Counter() last = None files: list[str] = [] for path in sorted((REVIEWS / 'manual_replay_results').glob('replay_result_*.json')): match = re.match(r'replay_result_(\d+)_(\d+)\.json$', path.name) if not match: continue start, end = map(int, match.groups()) if start != cursor: break items = result_items(load(path)) cursor = end files.append(str(path.relative_to(ROOT))) for item in items: counts[item.get('decision', 'unknown')] += 1 last = item.get('candidate_id') or last return { 'metadata': {'purpose': 'manual_replay_progress_state', 'created_at': now_iso(), 'source': 'hybrid/reviews/manual_replay_readiness_manifest.json', 'isFinalBacktest': False, 'notes': 'Derived from contiguous replay_result files.'}, 'cursor': {'next_index': cursor, 'processed_count': cursor, 'accepted_count': counts.get('accepted', 0), 'rejected_count': counts.get('rejected', 0), 'shadow_count': counts.get('shadow', 0), 'needs_more_context_count': counts.get('needs_more_context', 0), 'last_candidate_id': last, 'last_updated': now_iso()}, 'runs': [{'derived_at': now_iso(), 'result_files': files, 'decision_counts': dict(counts)}], } def main() -> int: os.chdir(ROOT) if LOCK.exists(): log(f'LOCK_EXISTS {LOCK}; exiting') return 2 LOCK.write_text(json.dumps({'started_at': now_iso(), 'pid': os.getpid()}, indent=2) + '\n') try: manifest = load(MANIFEST) total = len(manifest.get('manual_replay_items', [])) runs = 0 while True: state = ensure_state() start = int(state.get('cursor', {}).get('next_index', 0)) if start >= total: log(f'COMPLETE next_index={start} total={total}') return 0 end = min(start + BATCH_SIZE, total) if MAX_CHILD_RUNS and runs >= MAX_CHILD_RUNS: log(f'STOP max_child_runs={MAX_CHILD_RUNS} next_index={start}') return 0 runs += 1 prompt = PROMPT_TEMPLATE.format(batch_size=BATCH_SIZE, start_index=start, end_index=end) env = os.environ.copy() env['HERMES_PROFILE'] = 'finance' cmd = ['hermes', '--provider', 'custom:9router', '--model', 'ag/gemini-pro-agent', '-z', prompt] log(f'RUN_CHILD run={runs} start={start} end={end}') proc = subprocess.run(cmd, cwd=str(ROOT), env=env, text=True, capture_output=True, timeout=1800) log(f'HERMES_EXIT code={proc.returncode}') if proc.stdout.strip(): log('STDOUT ' + proc.stdout.strip().replace('\n', ' | ')[-4000:]) if proc.stderr.strip(): log('STDERR ' + proc.stderr.strip().replace('\n', ' | ')[-4000:]) if proc.returncode != 0: log('STOP nonzero child exit') return proc.returncode # Verify cursor advanced to avoid infinite loops. If child writes a stale # or incompatible progress schema but the result file exists, derive the # canonical cursor from contiguous replay_result files and continue. new_state = ensure_state() new_index = int(new_state.get('cursor', {}).get('next_index', 0)) if new_index <= start: derived = derive_state_from_results() derived_index = int(derived.get('cursor', {}).get('next_index', 0)) if derived_index > start: save(STATE, derived) log(f'CURSOR_DERIVED start={start} old_new_index={new_index} derived_index={derived_index}') else: log(f'STOP cursor_not_advanced start={start} new_index={new_index} derived_index={derived_index}') return 3 time.sleep(1) finally: try: LOCK.unlink(missing_ok=True) except Exception as exc: log(f'LOCK_REMOVE_ERROR {exc}') if __name__ == '__main__': raise SystemExit(main())