from pathlib import Path import json import os import subprocess import sys import time from sqlalchemy import text ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) REQUIRED_PIPELINE_TABLES = ( "venues", "instruments", "candles", "market_sessions", "swings", "dealing_ranges", "liquidity_pools", "sweep_events", "displacement_events", "structure_events", "fvg_zones", "bias_snapshots", "trade_signals", "trade_executions", "backtest_runs", "backtest_results", ) REQUIRED_PIPELINE_SCHEMA = { "venues": ("id", "code"), "instruments": ("id", "symbol", "venue_id"), "candles": ("id", "instrument_id", "timeframe", "ts_open", "ts_close", "open", "high", "low", "close"), "market_sessions": ("id", "instrument_id", "session_code", "session_date", "start_ts", "end_ts", "high", "low", "midpoint", "status", "meta"), "swings": ("id", "instrument_id", "timeframe", "kind", "strength", "price", "ts", "left_bars", "right_bars", "confirmed", "meta"), "dealing_ranges": ("id", "instrument_id", "timeframe", "high", "low", "equilibrium", "state", "ts_start", "meta"), "liquidity_pools": ("id", "instrument_id", "timeframe", "pool_type", "side", "price_low", "price_high", "reference_ids", "status", "created_ts", "updated_ts", "meta"), "sweep_events": ("id", "instrument_id", "timeframe", "side_swept", "pool_id", "sweep_high", "sweep_low", "sweep_ts", "close_back_inside", "validation_level", "status", "meta"), "displacement_events": ("id", "instrument_id", "timeframe", "direction", "start_ts", "end_ts", "strength_score", "body_ratio", "has_fvg", "is_valid", "meta"), "structure_events": ("id", "instrument_id", "timeframe", "event_type", "direction", "broken_level", "reference_swing_id", "displacement_event_id", "accepted", "signal_relevant", "ts", "meta"), "fvg_zones": ("id", "instrument_id", "timeframe", "direction", "lower", "upper", "start_ts", "end_ts", "status", "fill_pct", "related_displacement_id", "meta"), "bias_snapshots": ("id", "instrument_id", "bias_tf", "setup_tf", "trigger_tf", "snapshot_ts", "daily_bias", "execution_intensity", "draw_side", "reason_codes", "meta"), "trade_signals": ("id", "instrument_id", "model_code", "side", "status", "bias_snapshot_id", "entry_type", "entry_low", "entry_high", "stop_loss", "tp1", "tp2", "rr_tp1", "rr_tp2", "created_ts", "expires_ts", "invalidations", "evidence", "meta"), "trade_executions": ("id", "signal_id", "execution_mode", "side", "entry_price", "stop_loss", "take_profit_1", "take_profit_2", "size", "risk_pct", "status", "opened_ts", "meta"), "backtest_runs": ("id", "model_code", "instrument_id", "timeframe_set", "started_ts", "config", "summary"), "backtest_results": ("id", "backtest_run_id", "signal_id", "outcome", "mfe", "mae", "tp1_hit", "tp2_hit", "invalidated_before_entry", "error_tags", "meta"), } ACTION_CATALOG = { "fix_offline_checks_before_database_work": { "command": [sys.executable, "scripts/run_offline_checks.py"], "requires_database": False, "description": "Fix failing database-free checks before any database-dependent work.", }, "wait_for_database_recovery": { "command": [sys.executable, "scripts/run_autonomous_cycle.py"], "requires_database": False, "description": "Retry the autonomous cycle after PostgreSQL recovery or on the next heartbeat.", }, "run_migrations_before_pipeline": { "command": [sys.executable, "scripts/run_migrations.py"], "requires_database": True, "description": "Apply pending migrations before data validation or pipeline execution.", }, "inspect_migration_output_and_schema": { "command": [sys.executable, "scripts/run_migrations.py"], "requires_database": True, "description": "Inspect failed migration output and schema readiness before retrying.", }, "verify_seed_migrations_or_seed_data": { "command": [sys.executable, "scripts/run_database_readiness.py"], "requires_database": True, "description": "Verify required venues and instruments before the guarded pipeline.", }, "run_okx_backfill_or_load_replay_sample": { "command": [sys.executable, "scripts/run_okx_backfill.py"], "requires_database": True, "description": "Load enough candles for data readiness before the guarded pipeline.", }, "inspect_backfill_output_and_data_readiness": { "command": [sys.executable, "scripts/run_database_readiness.py"], "requires_database": True, "description": "Inspect failed candle backfill output and rerun database readiness.", }, "run_guarded_pipeline_when_ready": { "command": [sys.executable, "scripts/run_pipeline.py"], "requires_database": True, "description": "Run the guarded closed loop after readiness passes.", }, "run_guarded_pipeline": { "command": [sys.executable, "scripts/run_pipeline.py"], "requires_database": True, "description": "Run the guarded closed loop after readiness passes.", }, "inspect_pipeline_output_and_quality_gates": { "command": [sys.executable, "scripts/run_pipeline.py"], "requires_database": True, "description": "Inspect pipeline output and backtest quality gates before advancing.", }, "review_pipeline_summary_and_reports": { "command": [sys.executable, "scripts/run_reports.py"], "requires_database": True, "description": "Review reports after a completed guarded pipeline.", }, "continue_offline_automation": { "command": [sys.executable, "scripts/run_offline_checks.py"], "requires_database": False, "description": "Continue database-free automation and verification.", }, } def run_autonomous_cycle(env=None, runner=subprocess.run, db_probe=None, schema_probe=None, cwd=ROOT, sleeper=time.sleep) -> int: env = dict(os.environ if env is None else env) db_probe = db_probe or probe_database schema_probe = schema_probe or probe_database_schema output_json = _get_bool(env, "AUTONOMOUS_OUTPUT_JSON", False) require_db = _get_bool(env, "AUTONOMOUS_REQUIRE_DB", False) require_schema = _get_bool(env, "AUTONOMOUS_REQUIRE_SCHEMA", True) apply_migrations = _get_bool(env, "AUTONOMOUS_APPLY_MIGRATIONS", True) require_data_readiness = _get_bool(env, "AUTONOMOUS_REQUIRE_DATA_READINESS", True) backfill_candles = _get_bool(env, "AUTONOMOUS_BACKFILL_CANDLES", True) run_pipeline = _get_bool(env, "AUTONOMOUS_RUN_PIPELINE", True) max_db_probes = max(_get_int(env, "AUTONOMOUS_MAX_DB_PROBES", 1), 1) probe_interval_seconds = max(_get_float(env, "AUTONOMOUS_PROBE_INTERVAL_SECONDS", 0.0), 0.0) summary = { "mode": "starting", "passed": False, "steps": [], "database": None, "database_attempts": [], "schema": None, "next_actions": [], "action_plan": [], } offline_env = dict(env) offline_env["OFFLINE_CHECKS_OUTPUT_JSON"] = "1" offline = _run_step( name="offline_checks", command=[sys.executable, "scripts/run_offline_checks.py"], runner=runner, cwd=cwd, env=offline_env, ) summary["steps"].append(offline) if offline["exit_code"] != 0: summary["mode"] = "offline_checks_failed" return _finish(summary=summary, exit_code=offline["exit_code"], output_json=output_json) database = _probe_database_with_retries( db_probe=db_probe, max_db_probes=max_db_probes, probe_interval_seconds=probe_interval_seconds, sleeper=sleeper, ) summary["database"] = database summary["database_attempts"] = database.get("attempts", []) if not database["available"]: summary["mode"] = "waiting_for_database" summary["passed"] = not require_db return _finish(summary=summary, exit_code=2 if require_db else 0, output_json=output_json) schema = schema_probe() if require_schema else {"ready": True, "skipped": True, "missing_tables": []} summary["schema"] = schema if not schema["ready"]: if apply_migrations: migration_env = dict(env) migration_env["MIGRATIONS_OUTPUT_JSON"] = "1" migrations = _run_step( name="migrations", command=[sys.executable, "scripts/run_migrations.py"], runner=runner, cwd=cwd, env=migration_env, ) summary["steps"].append(migrations) if migrations["exit_code"] != 0: summary["mode"] = "migration_failed" summary["passed"] = False return _finish(summary=summary, exit_code=migrations["exit_code"], output_json=output_json) schema = schema_probe() summary["schema"] = schema if schema["ready"]: summary["migrations_applied"] = True else: summary["migrations_applied"] = False summary["mode"] = "database_schema_not_ready" summary["passed"] = False if not schema["ready"]: return _finish(summary=summary, exit_code=2, output_json=output_json) if require_data_readiness: readiness_env = dict(env) readiness_env["READINESS_OUTPUT_JSON"] = "1" readiness = _run_database_readiness_step(runner=runner, cwd=cwd, env=readiness_env) summary["steps"].append(readiness) if readiness["exit_code"] != 0 and backfill_candles and readiness_needs_candle_backfill(readiness.get("summary") or {}): backfill_env = dict(env) backfill_env["OKX_BACKFILL_OUTPUT_JSON"] = "1" backfill_env.setdefault("OKX_BACKFILL_TIMEFRAMES", "1m") backfill_env.setdefault("OKX_TIMEOUT", "10") backfill = _run_step( name="okx_backfill", command=[sys.executable, "scripts/run_okx_backfill.py"], runner=runner, cwd=cwd, env=backfill_env, ) summary["steps"].append(backfill) if backfill["exit_code"] != 0: summary["mode"] = "backfill_failed" summary["passed"] = False return _finish(summary=summary, exit_code=backfill["exit_code"], output_json=output_json) readiness = _run_database_readiness_step(runner=runner, cwd=cwd, env=readiness_env) summary["steps"].append(readiness) if readiness["exit_code"] != 0: summary["mode"] = "database_readiness_failed" summary["passed"] = False return _finish(summary=summary, exit_code=readiness["exit_code"], output_json=output_json) if not run_pipeline: summary["mode"] = "database_ready_pipeline_skipped" summary["passed"] = True return _finish(summary=summary, exit_code=0, output_json=output_json) pipeline_env = dict(env) pipeline_env.setdefault("BACKTEST_REQUIRE_QUALITY_GATE", "1") pipeline_env.setdefault("PIPELINE_OUTPUT_JSON", "1") pipeline = _run_step( name="pipeline", command=[sys.executable, "scripts/run_pipeline.py"], runner=runner, cwd=cwd, env=pipeline_env, ) summary["steps"].append(pipeline) summary["mode"] = "pipeline_completed" if pipeline["exit_code"] == 0 else "pipeline_failed" summary["passed"] = pipeline["exit_code"] == 0 return _finish(summary=summary, exit_code=pipeline["exit_code"], output_json=output_json) def _run_database_readiness_step(runner, cwd: Path, env: dict) -> dict: return _run_step( name="database_readiness", command=[sys.executable, "scripts/run_database_readiness.py"], runner=runner, cwd=cwd, env=env, ) def readiness_needs_candle_backfill(readiness_summary: dict) -> bool: reasons = set(readiness_summary.get("reasons") or []) blocking_reasons = {"database_unavailable", "schema_not_ready", "missing_required_venues", "missing_required_instruments"} if reasons & blocking_reasons: return False next_actions = set(readiness_summary.get("next_actions") or []) candle_reasons = {"not_enough_candles", "not_enough_candle_span", "candles_too_stale"} return bool(reasons & candle_reasons) or "run_okx_backfill_or_load_replay_sample" in next_actions def probe_database() -> dict: try: from src.db.engine import engine with engine.connect() as connection: value = connection.execute(text("select 1")).scalar_one() return {"available": True, "value": value} except Exception as exc: return {"available": False, "error_type": exc.__class__.__name__, "error": str(exc)} def probe_database_schema(required_tables: tuple[str, ...] = REQUIRED_PIPELINE_TABLES) -> dict: try: from src.db.engine import engine required_schema = build_required_schema(required_tables=required_tables) existing_tables = set() existing_columns: dict[str, set[str]] = {} missing_tables = [] with engine.connect() as connection: for table in required_tables: exists = connection.execute(text("select to_regclass(:table_name)"), {"table_name": f"public.{table}"}).scalar() if exists is None: missing_tables.append(table) continue existing_tables.add(table) columns = connection.execute( text("select column_name from information_schema.columns where table_schema = 'public' and table_name = :table_name"), {"table_name": table}, ).scalars().all() existing_columns[table] = set(columns) readiness = evaluate_schema_readiness( required_schema=required_schema, existing_tables=existing_tables, existing_columns=existing_columns, ) readiness["missing_tables"] = missing_tables or readiness["missing_tables"] readiness["ready"] = len(readiness["missing_tables"]) == 0 and len(readiness["missing_columns"]) == 0 return readiness except Exception as exc: return { "ready": False, "required_tables": list(required_tables), "missing_tables": list(required_tables), "missing_columns": {}, "error_type": exc.__class__.__name__, "error": str(exc), } def build_required_schema(required_tables: tuple[str, ...] = REQUIRED_PIPELINE_TABLES) -> dict[str, tuple[str, ...]]: return {table: REQUIRED_PIPELINE_SCHEMA.get(table, ()) for table in required_tables} def evaluate_schema_readiness(required_schema: dict[str, tuple[str, ...]], existing_tables: set[str], existing_columns: dict[str, set[str]]) -> dict: missing_tables = [table for table in required_schema if table not in existing_tables] missing_columns = {} for table, required_columns in required_schema.items(): if table in missing_tables: continue missing = [column for column in required_columns if column not in existing_columns.get(table, set())] if missing: missing_columns[table] = missing return { "ready": len(missing_tables) == 0 and len(missing_columns) == 0, "required_tables": list(required_schema), "missing_tables": missing_tables, "missing_columns": missing_columns, } def _probe_database_with_retries(db_probe, max_db_probes: int, probe_interval_seconds: float, sleeper) -> dict: attempts = [] last_result = None for attempt in range(1, max_db_probes + 1): last_result = dict(db_probe()) last_result["attempt"] = attempt attempts.append(last_result) if last_result["available"]: break if attempt < max_db_probes and probe_interval_seconds > 0: sleeper(probe_interval_seconds) result = dict(last_result or {"available": False, "error_type": "NoProbeResult", "error": "no database probe executed"}) result["attempts"] = attempts return result def _run_step(name: str, command: list[str], runner, cwd: Path, env: dict) -> dict: result = runner(command, cwd=cwd, capture_output=True, text=True, check=False, env=env) parsed_stdout = _parse_json(result.stdout) return { "name": name, "command": command, "status": "passed" if result.returncode == 0 else "failed", "exit_code": result.returncode, "stdout": result.stdout, "stderr": result.stderr, "summary": parsed_stdout, } def _parse_json(value: str): if not value: return None try: return json.loads(value) except json.JSONDecodeError: return None def _finish(summary: dict, exit_code: int, output_json: bool) -> int: summary["next_actions"] = build_cycle_next_actions(summary) summary["action_plan"] = build_cycle_action_plan(summary["next_actions"]) if output_json: print(json.dumps(summary, ensure_ascii=False, sort_keys=True)) return exit_code print(f"Autonomous mode: {summary['mode']}") print(f"Next actions: {', '.join(summary['next_actions'])}") for action in summary["action_plan"]: command = " ".join(action["command"]) db_hint = "db" if action["requires_database"] else "no-db" print(f"Action plan [{db_hint}]: {action['action']} -> {command}") for step in summary["steps"]: print(f"Step {step['name']}: {step['status']} exit_code={step['exit_code']}") if step["exit_code"] != 0: if step["stdout"]: print(step["stdout"].rstrip()) if step["stderr"]: print(step["stderr"].rstrip()) database = summary.get("database") if database: print(f"Database available: {database['available']}") if not database["available"]: print(f"Database probe error: {database.get('error_type')}") schema = summary.get("schema") if schema: print(f"Database schema ready: {schema['ready']}") if schema.get("missing_tables"): print(f"Missing tables: {', '.join(schema['missing_tables'])}") if schema.get("missing_columns"): missing_columns = [ f"{table}.{column}" for table, columns in schema["missing_columns"].items() for column in columns ] print(f"Missing columns: {', '.join(missing_columns)}") return exit_code def build_cycle_next_actions(summary: dict) -> list[str]: mode = summary.get("mode") if mode == "offline_checks_failed": return ["fix_offline_checks_before_database_work"] if mode == "waiting_for_database": return ["wait_for_database_recovery"] if mode == "database_schema_not_ready": return ["run_migrations_before_pipeline"] if mode == "migration_failed": return ["inspect_migration_output_and_schema"] if mode == "backfill_failed": return ["inspect_backfill_output_and_data_readiness"] if mode == "database_readiness_failed": for step in summary.get("steps", []): if step.get("name") == "database_readiness": step_summary = step.get("summary") or {} next_actions = step_summary.get("next_actions") if next_actions: return next_actions return ["inspect_database_readiness_reasons"] if mode == "database_ready_pipeline_skipped": return ["run_guarded_pipeline_when_ready"] if mode == "pipeline_failed": return ["inspect_pipeline_output_and_quality_gates"] if mode == "pipeline_completed": return ["review_pipeline_summary_and_reports"] return ["continue_offline_automation"] def build_cycle_action_plan(next_actions: list[str]) -> list[dict]: plan = [] for action in next_actions: catalog_item = ACTION_CATALOG.get(action, ACTION_CATALOG["continue_offline_automation"]) plan.append( { "action": action, "command": list(catalog_item["command"]), "requires_database": catalog_item["requires_database"], "description": catalog_item["description"], } ) return plan def _get_bool(env, key: str, default: bool) -> bool: value = env.get(key) if value is None or value == "": return default return str(value).strip().lower() in {"1", "true", "yes", "on"} def _get_int(env, key: str, default: int) -> int: value = env.get(key) if value is None or value == "": return default return int(value) def _get_float(env, key: str, default: float) -> float: value = env.get(key) if value is None or value == "": return default return float(value) def main() -> int: return run_autonomous_cycle() if __name__ == "__main__": raise SystemExit(main())