from __future__ import annotations import argparse import os import subprocess import sys from dataclasses import dataclass from pathlib import Path from typing import Sequence from config import BotConfig, RuntimePaths from strategy_registry import StrategyPreset, load_strategy_presets PARAM_ENV = { "max_total_trades": "CTB_MAX_TOTAL_TRADES", "default_leverage": "CTB_DEFAULT_LEVERAGE", "crash_roe_trigger_pct": "CTB_CRASH_ROE_TRIGGER_PCT", "trade_size_usd": "CTB_TRADE_SIZE_USD", "wallet_equity_usdc": "CTB_WALLET_EQUITY_USDC", "risk_per_trade_pct": "CTB_RISK_PER_TRADE_PCT", "position_sizing_mode": "CTB_POSITION_SIZING_MODE", "max_position_margin_pct": "CTB_MAX_POSITION_MARGIN_PCT", "max_position_notional_pct": "CTB_MAX_POSITION_NOTIONAL_PCT", "min_order_notional_usd": "CTB_MIN_ORDER_NOTIONAL_USD", "crash_window_minutes": "CTB_CRASH_WINDOW_MINUTES", "flash_crash_trigger_pct": "CTB_FLASH_CRASH_TRIGGER_PCT", "atr_sl_multiplier": "CTB_ATR_SL_MULTIPLIER", "break_even_activation_pct": "CTB_BREAK_EVEN_ACTIVATION_PCT", "v_shape_activation_pct": "CTB_V_SHAPE_ACTIVATION_PCT", "v_shape_trail_dist_pct": "CTB_V_SHAPE_TRAIL_DIST_PCT", "dead_fish_time_limit_mins": "CTB_DEAD_FISH_TIME_LIMIT_MINS", "max_hard_stop_pct": "CTB_MAX_HARD_STOP_PCT", "min_volume_24h": "CTB_MIN_VOLUME_24H", "cooldown_minutes": "CTB_COOLDOWN_MINUTES", "max_daily_trades": "CTB_MAX_DAILY_TRADES", "max_daily_loss_usd": "CTB_MAX_DAILY_LOSS_USD", "strategy_family": "CTB_STRATEGY_FAMILY", "squeeze_lookback_ticks": "CTB_SQUEEZE_LOOKBACK_TICKS", "squeeze_max_band_width_pct": "CTB_SQUEEZE_MAX_BAND_WIDTH_PCT", "squeeze_breakout_pct": "CTB_SQUEEZE_BREAKOUT_PCT", "trend_pullback_min_rsi": "CTB_TREND_PULLBACK_MIN_RSI", "trend_pullback_max_rsi": "CTB_TREND_PULLBACK_MAX_RSI", "trend_pullback_max_distance_pct": "CTB_TREND_PULLBACK_MAX_DISTANCE_PCT", "mean_reversion_max_rsi": "CTB_MEAN_REVERSION_MAX_RSI", "mean_reversion_band_tolerance_pct": "CTB_MEAN_REVERSION_BAND_TOLERANCE_PCT", "multi_day_min_trend_pct": "CTB_MULTI_DAY_MIN_TREND_PCT", "multi_day_max_pullback_distance_pct": "CTB_MULTI_DAY_MAX_PULLBACK_DISTANCE_PCT", "relative_strength_lookback_ticks": "CTB_RELATIVE_STRENGTH_LOOKBACK_TICKS", "relative_strength_top_n": "CTB_RELATIVE_STRENGTH_TOP_N", "relative_strength_min_momentum_pct": "CTB_RELATIVE_STRENGTH_MIN_MOMENTUM_PCT", "relative_strength_min_positive_candidates": "CTB_RELATIVE_STRENGTH_MIN_POSITIVE_CANDIDATES", "allowed_coins": "CTB_ALLOWED_COINS", } @dataclass(frozen=True) class StrategyCommandPlan: strategy_id: str command: list[str] env: dict[str, str] log_path: Path pid_path: Path def build_strategy_env(preset: StrategyPreset, runtime_dir: str | Path) -> dict[str, str]: experiment_root = Path(runtime_dir).expanduser() / "experiments" / preset.strategy_id env = { "CTB_STRATEGY_ID": preset.strategy_id, "CTB_RUNTIME_DIR": str(experiment_root), "CTB_PAPER_TRADING": "true", "CTB_DRY_RUN": "false", "CTB_TLS_VERIFY": "true", "PYTHONUNBUFFERED": "1", } for key, value in preset.parameters.items(): env_name = PARAM_ENV.get(key) if env_name: if isinstance(value, (list, tuple)): env[env_name] = ",".join(str(item) for item in value) else: env[env_name] = str(value) return env def build_strategy_command_plan(presets: Sequence[StrategyPreset], runtime_dir: str | Path, *, max_parallel: int = 5) -> list[StrategyCommandPlan]: selected = list(presets)[:max_parallel] plans: list[StrategyCommandPlan] = [] for preset in selected: experiment_root = Path(runtime_dir).expanduser() / "experiments" / preset.strategy_id plans.append(StrategyCommandPlan( strategy_id=preset.strategy_id, command=[sys.executable, "AutoTrader.py"], env=build_strategy_env(preset, runtime_dir), log_path=experiment_root / "bot.log", pid_path=experiment_root / "bot.pid", )) return plans def render_command_plan(plans: Sequence[StrategyCommandPlan]) -> str: lines = ["Multi-Strategy Paper Command Plan"] for plan in plans: lines.append(f"- {plan.strategy_id}: {' '.join(plan.command)} | log={plan.log_path}") return "\n".join(lines) def launch_plans(plans: Sequence[StrategyCommandPlan]) -> list[subprocess.Popen]: processes = [] for plan in plans: plan.log_path.parent.mkdir(parents=True, exist_ok=True) log_file = plan.log_path.open("a", encoding="utf-8") env = {**os.environ, **plan.env} process = subprocess.Popen(plan.command, env=env, stdout=log_file, stderr=subprocess.STDOUT) plan.pid_path.write_text(str(process.pid), encoding="utf-8") processes.append(process) return processes def main(argv: list[str] | None = None) -> int: cfg = BotConfig.from_file() runtime_paths = RuntimePaths.from_config(cfg) parser = argparse.ArgumentParser(description="Plan or launch isolated paper AutoTrader processes for selected strategies.") parser.add_argument("--runtime-dir", default=str(runtime_paths.runtime_dir)) parser.add_argument("--max-parallel", type=int, default=5, help="Safety cap; do not launch every preset at once against live APIs.") parser.add_argument("--strategies", nargs="*", default=None) parser.add_argument("--launch", action="store_true", help="Actually launch paper processes. Default only prints the plan.") args = parser.parse_args(argv) all_presets = load_strategy_presets() strategy_ids = args.strategies or list(all_presets.keys()) presets = [all_presets[sid] for sid in strategy_ids if sid in all_presets] plans = build_strategy_command_plan(presets, args.runtime_dir, max_parallel=args.max_parallel) print(render_command_plan(plans)) if args.launch: processes = launch_plans(plans) print("Launched PIDs: " + ", ".join(str(p.pid) for p in processes)) else: print("Dry plan only. Re-run with --launch to start paper processes.") return 0 if __name__ == "__main__": raise SystemExit(main())