Spaces:
Running
Running
File size: 3,768 Bytes
74d4035 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | #!/usr/bin/env python3
"""Schedule loop_backtest.py to start once after N hours (macOS launchd)."""
from __future__ import annotations
import argparse
import datetime as dt
import os
import plistlib
import subprocess
import sys
from pathlib import Path
def _pick_python(repo_root: Path) -> str:
venv_python = repo_root / ".venv" / "bin" / "python"
if venv_python.exists():
return str(venv_python)
return sys.executable
def _write_plist(plist_path: Path, label: str, run_at: dt.datetime, repo_root: Path, reset: bool) -> dict:
python_bin = _pick_python(repo_root)
script_path = repo_root / "research" / "loop_backtest.py"
args = [python_bin, str(script_path)]
if reset:
args.append("--reset")
payload = {
"Label": label,
"ProgramArguments": args,
"WorkingDirectory": str(repo_root),
"StartCalendarInterval": {
"Year": run_at.year,
"Month": run_at.month,
"Day": run_at.day,
"Hour": run_at.hour,
"Minute": run_at.minute,
},
"RunAtLoad": False,
"StandardOutPath": "/tmp/papertrade_loop_backtest_delayed.log",
"StandardErrorPath": "/tmp/papertrade_loop_backtest_delayed.err",
}
plist_path.parent.mkdir(parents=True, exist_ok=True)
with plist_path.open("wb") as f:
plistlib.dump(payload, f)
return {
"python": python_bin,
"script": str(script_path),
"stdout": payload["StandardOutPath"],
"stderr": payload["StandardErrorPath"],
}
def _launch_load(plist_path: Path) -> tuple[bool, str]:
uid = os.getuid()
target = f"gui/{uid}"
subprocess.run(
["launchctl", "bootout", target, str(plist_path)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
bootstrap = subprocess.run(
["launchctl", "bootstrap", target, str(plist_path)],
capture_output=True,
text=True,
check=False,
)
if bootstrap.returncode == 0:
return True, "bootstrap"
legacy = subprocess.run(
["launchctl", "load", "-w", str(plist_path)],
capture_output=True,
text=True,
check=False,
)
if legacy.returncode == 0:
return True, "load"
err = (bootstrap.stderr or "").strip() or (legacy.stderr or "").strip() or "unknown launchctl error"
return False, err
def main() -> int:
parser = argparse.ArgumentParser(description="Schedule loop_backtest.py after a delay")
parser.add_argument("--hours", type=float, default=5.0, help="Delay before running loop_backtest (default: 5)")
parser.add_argument("--reset", action="store_true", help="Pass --reset to loop_backtest.py")
parser.add_argument("--label", default="com.papertrade.loopbacktest.once", help="launchd label")
args = parser.parse_args()
repo_root = Path(__file__).resolve().parents[1]
run_at = dt.datetime.now() + dt.timedelta(hours=args.hours)
plist_path = Path.home() / "Library" / "LaunchAgents" / f"{args.label}.plist"
details = _write_plist(plist_path, args.label, run_at, repo_root, args.reset)
ok, method_or_err = _launch_load(plist_path)
if not ok:
print(f"FAILED to load launch agent: {method_or_err}")
return 1
print("Scheduled delayed loop backtest")
print(f"label={args.label}")
print(f"plist={plist_path}")
print(f"run_at_local={run_at.strftime('%Y-%m-%d %H:%M')}")
print(f"python={details['python']}")
print(f"script={details['script']}")
print(f"stdout={details['stdout']}")
print(f"stderr={details['stderr']}")
print(f"load_method={method_or_err}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|