|
|
"""ORB paper trading in-process service layer.
|
|
|
|
|
|
Provides:
|
|
|
- ORBAutoScheduler: asyncio-based intraday scheduler
|
|
|
- orb_auto_scheduler: module-level singleton
|
|
|
- State persistence to .orb_auto_state.json
|
|
|
|
|
|
Schedule is computed dynamically from strategy params (orb_minutes,
|
|
|
order_timeout_minutes, sim_bar_minutes) rather than being hardcoded.
|
|
|
|
|
|
Daily event sequence:
|
|
|
9:30 + orb_minutes → orb_detect (daily bars + 5-min ORB bars → candidates)
|
|
|
orb_end + 1 min → breakout (Oracle snapshot price check, every 1 min)
|
|
|
...repeat until order_timeout_minutes elapses
|
|
|
orb_end + N×sim_bar → stop_check (fetch 5-min bars, aggregate, check stops)
|
|
|
...repeat until 15:50
|
|
|
15:55 → eod_exit
|
|
|
16:00 → post_close
|
|
|
"""
|
|
|
from __future__ import annotations
|
|
|
|
|
|
import asyncio
|
|
|
import datetime as dt
|
|
|
import json
|
|
|
import logging
|
|
|
import os
|
|
|
import signal
|
|
|
import subprocess
|
|
|
import sys
|
|
|
import time
|
|
|
import traceback
|
|
|
import uuid
|
|
|
from datetime import datetime, timezone
|
|
|
from pathlib import Path
|
|
|
from typing import Any
|
|
|
from zoneinfo import ZoneInfo
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
# Mirror scheduler activity into journal/events.db so the Logs/Health UI can
|
|
|
# group ORB phases by job_run_id alongside PEAD events. We write directly to
|
|
|
# EventsStore rather than via structlog because the scheduler runs in the
|
|
|
# daemon process where source inference from a PrintLogger is unreliable.
|
|
|
try:
|
|
|
from libs.common.logging import bind_job_run_id as _bind_job_run_id
|
|
|
except Exception: # pragma: no cover
|
|
|
_bind_job_run_id = None # type: ignore[assignment]
|
|
|
|
|
|
|
|
|
def _scheduler_emit(event: str, level: str = "INFO", **fields: Any) -> None:
|
|
|
try:
|
|
|
from apps.web.services.events_store import EventsStore
|
|
|
payload: dict[str, Any] = {
|
|
|
"ts_utc": datetime.now(timezone.utc).isoformat(),
|
|
|
"source": "auto_scheduler",
|
|
|
"level": level,
|
|
|
"event": event,
|
|
|
"scheduler": "orb",
|
|
|
}
|
|
|
payload.update(fields)
|
|
|
EventsStore.get().write(payload)
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
|
|
|
# kind → public phase name (distinct from PEAD's run_open/run_close so health
|
|
|
# card rows do not collide).
|
|
|
_PHASE_NAME_MAP = {
|
|
|
"pre_screen": "orb_pre_screen",
|
|
|
"orb_detect": "orb_detect",
|
|
|
"breakout": "orb_breakout",
|
|
|
"stop_check": "orb_stop",
|
|
|
"eod_exit": "orb_eod",
|
|
|
"post_close": "orb_post_close",
|
|
|
}
|
|
|
|
|
|
_TZ_ET = ZoneInfo("America/New_York")
|
|
|
_TZ_PHOENIX = ZoneInfo("America/Phoenix") # UTC-7 always (no DST)
|
|
|
_DEFAULT_DB = "data/paper/orb.db"
|
|
|
|
|
|
_MARKET_OPEN = dt.time(9, 30) # ET
|
|
|
_EOD_EXIT = dt.time(15, 55) # ET
|
|
|
_POST_CLOSE = dt.time(16, 0) # ET
|
|
|
|
|
|
|
|
|
# ── Dynamic schedule builder ──────────────────────────────────────────────────
|
|
|
|
|
|
def build_schedule(
|
|
|
date: dt.date,
|
|
|
orb_minutes: int = 10,
|
|
|
order_timeout_minutes: int = 45,
|
|
|
sim_bar_minutes: int = 90,
|
|
|
) -> list[dict[str, Any]]:
|
|
|
"""Build today's event list from strategy parameters.
|
|
|
|
|
|
Args:
|
|
|
date: The trading date.
|
|
|
orb_minutes: Length of the ORB window (e.g. 10 → window is 9:30–9:40).
|
|
|
order_timeout_minutes: How many minutes after ORB end to keep looking for
|
|
|
breakouts (one check per minute).
|
|
|
sim_bar_minutes: Interval between stop-management checks after breakout
|
|
|
window closes (matches the sim_bar_minutes strategy param).
|
|
|
|
|
|
Returns:
|
|
|
List of event dicts sorted chronologically. Each dict has:
|
|
|
name, label, kind, et_dt (aware datetime in ET)
|
|
|
|
|
|
Note: breakout checks run every sim_bar_minutes (matching the backtest bar
|
|
|
aggregation interval), up to order_timeout_minutes after *market open*
|
|
|
(not after ORB end) — consistent with orb_simulator.py timeout semantics.
|
|
|
"""
|
|
|
mkt_open = dt.datetime(date.year, date.month, date.day, 9, 30, tzinfo=_TZ_ET)
|
|
|
orb_end = mkt_open + dt.timedelta(minutes=orb_minutes)
|
|
|
eod = dt.datetime(date.year, date.month, date.day, 15, 55, tzinfo=_TZ_ET)
|
|
|
post = dt.datetime(date.year, date.month, date.day, 16, 0, tzinfo=_TZ_ET)
|
|
|
|
|
|
events: list[dict[str, Any]] = []
|
|
|
|
|
|
# ── 사전 스크리닝: 9:20 ET (daily bars + enrichment + quality filter) ────────
|
|
|
pre_screen_time = mkt_open - dt.timedelta(minutes=10)
|
|
|
events.append({
|
|
|
"name": "pre_screen",
|
|
|
"label": f"사전 스크리닝 ({pre_screen_time.strftime('%H:%M')} ET)",
|
|
|
"kind": "pre_screen",
|
|
|
"et_dt": pre_screen_time,
|
|
|
})
|
|
|
|
|
|
# ── ORB window monitoring: 9:30 → orb_end-1, every minute (no-op) ─────────
|
|
|
for i in range(orb_minutes):
|
|
|
t = mkt_open + dt.timedelta(minutes=i)
|
|
|
events.append({
|
|
|
"name": f"orb_monitor_{i + 1}",
|
|
|
"label": f"ORB 윈도우 ({t.strftime('%H:%M')} ET, {i + 1}/{orb_minutes}분)",
|
|
|
"kind": "orb_monitor",
|
|
|
"et_dt": t,
|
|
|
})
|
|
|
|
|
|
# ── ORB detection: one bar after window close (bars are published ~5 min late)
|
|
|
detect_dt = orb_end + dt.timedelta(minutes=5)
|
|
|
events.append({
|
|
|
"name": "orb_detect",
|
|
|
"label": f"ORB 감지 ({detect_dt.strftime('%H:%M')} ET)",
|
|
|
"kind": "orb_detect",
|
|
|
"et_dt": detect_dt,
|
|
|
})
|
|
|
|
|
|
# ── Breakout checks: every sim_bar_minutes from first bar close to timeout ──
|
|
|
# Timeout is measured from market open (matches orb_simulator.py semantics).
|
|
|
timeout_dt = mkt_open + dt.timedelta(minutes=order_timeout_minutes)
|
|
|
i = 1
|
|
|
t = orb_end + dt.timedelta(minutes=sim_bar_minutes)
|
|
|
while t <= timeout_dt:
|
|
|
if t >= eod:
|
|
|
break
|
|
|
events.append({
|
|
|
"name": f"breakout_{i}",
|
|
|
"label": f"브레이크아웃 ({t.strftime('%H:%M')} ET, +{i * sim_bar_minutes}분)",
|
|
|
"kind": "breakout",
|
|
|
"et_dt": t,
|
|
|
})
|
|
|
t += dt.timedelta(minutes=sim_bar_minutes)
|
|
|
i += 1
|
|
|
|
|
|
# ── Stop checks: every sim_bar_minutes from orb_end ───────────────────────
|
|
|
t = orb_end + dt.timedelta(minutes=sim_bar_minutes)
|
|
|
idx = 1
|
|
|
while t < eod:
|
|
|
events.append({
|
|
|
"name": f"stop_{idx}",
|
|
|
"label": f"스톱 체크 ({t.strftime('%H:%M')} ET, +{sim_bar_minutes}분)",
|
|
|
"kind": "stop_check",
|
|
|
"et_dt": t,
|
|
|
})
|
|
|
t += dt.timedelta(minutes=sim_bar_minutes)
|
|
|
idx += 1
|
|
|
|
|
|
# ── EOD + post-close ──────────────────────────────────────────────────────
|
|
|
events.append({"name": "eod_exit", "label": "EOD 청산 (15:55 ET)",
|
|
|
"kind": "eod_exit", "et_dt": eod})
|
|
|
events.append({"name": "post_close", "label": "마감 후 스냅샷 (16:00 ET)",
|
|
|
"kind": "post_close", "et_dt": post})
|
|
|
|
|
|
return sorted(events, key=lambda e: e["et_dt"])
|
|
|
|
|
|
|
|
|
def _load_session_params(db_path: str, session_name: str) -> dict[str, Any]:
|
|
|
"""Load strategy params (orb_minutes, sim_bar_minutes, order_timeout_minutes)
|
|
|
from the session's config YAML. Returns defaults on any error.
|
|
|
"""
|
|
|
defaults = {"orb_minutes": 10, "sim_bar_minutes": 90, "order_timeout_minutes": 45}
|
|
|
try:
|
|
|
from apps.orb_trader.state import ORBStateManager
|
|
|
from apps.intraday_bt.run import _load_config_yaml
|
|
|
session = ORBStateManager(db_path).get_session(session_name)
|
|
|
if session is None:
|
|
|
return defaults
|
|
|
raw = _load_config_yaml(Path(session.config_path)) or {}
|
|
|
orb = raw.get("orb_strategy", {})
|
|
|
return {
|
|
|
"orb_minutes": orb.get("orb_minutes", defaults["orb_minutes"]),
|
|
|
"sim_bar_minutes": orb.get("sim_bar_minutes", defaults["sim_bar_minutes"]),
|
|
|
"order_timeout_minutes": orb.get("order_timeout_minutes", defaults["order_timeout_minutes"]),
|
|
|
}
|
|
|
except Exception:
|
|
|
return defaults
|
|
|
|
|
|
|
|
|
|
|
|
# ── State file helpers ────────────────────────────────────────────────────────
|
|
|
|
|
|
def _state_file_path(db_path: str) -> Path:
|
|
|
return Path(db_path).parent / ".orb_auto_state.json"
|
|
|
|
|
|
|
|
|
def load_orb_saved_state(db_path: str) -> dict[str, Any] | None:
|
|
|
sf = _state_file_path(db_path)
|
|
|
if sf.exists():
|
|
|
try:
|
|
|
return json.loads(sf.read_text())
|
|
|
except Exception:
|
|
|
pass
|
|
|
return None
|
|
|
|
|
|
|
|
|
# ── ORBAutoScheduler ──────────────────────────────────────────────────────────
|
|
|
|
|
|
class ORBAutoScheduler:
|
|
|
"""In-process ORB intraday auto-trader.
|
|
|
|
|
|
Schedule is built dynamically each trading day from the sessions' strategy
|
|
|
params (orb_minutes, order_timeout_minutes, sim_bar_minutes).
|
|
|
"""
|
|
|
|
|
|
def __init__(self) -> None:
|
|
|
self._task: asyncio.Task | None = None # type: ignore[type-arg]
|
|
|
self._sessions: list[str] = []
|
|
|
self._db_path: str = _DEFAULT_DB
|
|
|
self._dry_run: bool = False
|
|
|
self._log_lines: list[str] = []
|
|
|
self._completed: set[str] = set()
|
|
|
self._today_schedule: list[dict[str, Any]] = [] # built each trading day
|
|
|
self._engines: dict[str, Any] = {}
|
|
|
self._engine_date: str = ""
|
|
|
|
|
|
@property
|
|
|
def running(self) -> bool:
|
|
|
return self._task is not None and not self._task.done()
|
|
|
|
|
|
def _log_file_path(self) -> Path:
|
|
|
return Path(self._db_path).parent / "orb_scheduler.log"
|
|
|
|
|
|
def _pid_file_path(self) -> Path:
|
|
|
return Path(self._db_path).parent / ".orb_scheduler.pid"
|
|
|
|
|
|
def _schedule_file_path(self) -> Path:
|
|
|
return Path(self._db_path).parent / ".orb_schedule_state.json"
|
|
|
|
|
|
def _save_schedule_state(self) -> None:
|
|
|
"""Persist today's schedule + completed set so web API can read it."""
|
|
|
try:
|
|
|
schedule_data = []
|
|
|
for ev in self._today_schedule:
|
|
|
schedule_data.append({
|
|
|
"name": ev["name"],
|
|
|
"kind": ev["kind"],
|
|
|
"session": ev.get("session", ""),
|
|
|
"label": ev["label"],
|
|
|
"et_iso": ev["et_dt"].isoformat(),
|
|
|
})
|
|
|
self._schedule_file_path().write_text(json.dumps({
|
|
|
"schedule": schedule_data,
|
|
|
"completed": list(self._completed),
|
|
|
}))
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
def _load_persisted_log(self, max_lines: int = 3000) -> list[str]:
|
|
|
"""Load existing log lines from file (survives server restarts).
|
|
|
Trims to last max_lines if the file has grown too large.
|
|
|
"""
|
|
|
try:
|
|
|
lf = self._log_file_path()
|
|
|
if lf.exists():
|
|
|
lines = lf.read_text(encoding="utf-8").splitlines()
|
|
|
if len(lines) > max_lines:
|
|
|
# Keep last max_lines; rewrite trimmed file
|
|
|
lines = lines[-max_lines:]
|
|
|
try:
|
|
|
lf.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
except Exception:
|
|
|
pass
|
|
|
return lines
|
|
|
except Exception:
|
|
|
pass
|
|
|
return []
|
|
|
|
|
|
def start(self, sessions: list[str], db_path: str, dry_run: bool = False) -> None:
|
|
|
self._db_path = db_path # Set FIRST so _pid_file_path() uses the right dir
|
|
|
if self.running:
|
|
|
raise RuntimeError("ORBAutoScheduler already running")
|
|
|
self._sessions = sessions
|
|
|
self._dry_run = dry_run
|
|
|
self._log_lines = self._load_persisted_log() # restore previous logs
|
|
|
self._completed = set()
|
|
|
self._today_schedule = []
|
|
|
self._engines = {}
|
|
|
self._engine_date = ""
|
|
|
self._save_state()
|
|
|
# Write PID file so ORBDaemonController can detect us
|
|
|
try:
|
|
|
self._pid_file_path().write_text(str(os.getpid()))
|
|
|
except Exception:
|
|
|
pass
|
|
|
self._task = asyncio.create_task(self._run_loop())
|
|
|
|
|
|
def stop(self) -> None:
|
|
|
if self._task and not self._task.done():
|
|
|
self._task.cancel()
|
|
|
# Remove PID file and schedule state on intentional stop
|
|
|
try:
|
|
|
pf = self._pid_file_path()
|
|
|
if pf.exists():
|
|
|
pf.unlink()
|
|
|
except Exception:
|
|
|
pass
|
|
|
try:
|
|
|
sf = self._schedule_file_path()
|
|
|
if sf.exists():
|
|
|
sf.unlink()
|
|
|
except Exception:
|
|
|
pass
|
|
|
self._clear_state()
|
|
|
|
|
|
def shutdown(self) -> None:
|
|
|
"""Server shutdown — cancel task but keep state for auto-restart."""
|
|
|
if self._task and not self._task.done():
|
|
|
self._task.cancel()
|
|
|
|
|
|
def clear_log(self) -> None:
|
|
|
"""Wipe in-memory log and the persisted log file."""
|
|
|
self._log_lines = []
|
|
|
try:
|
|
|
lf = self._log_file_path()
|
|
|
if lf.exists():
|
|
|
lf.unlink()
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
def get_log(self, lines: int = 200) -> str:
|
|
|
return "\n".join(self._log_lines[-lines:])
|
|
|
|
|
|
def get_log_tail(self, lines: int = 80) -> list[str]:
|
|
|
return self._log_lines[-lines:]
|
|
|
|
|
|
@property
|
|
|
def log_line_count(self) -> int:
|
|
|
return len(self._log_lines)
|
|
|
|
|
|
def get_status(self) -> dict[str, Any]:
|
|
|
now_et = self._now_et()
|
|
|
today = now_et.date()
|
|
|
schedule_view = []
|
|
|
for ev in self._today_schedule:
|
|
|
ev_dt = ev["et_dt"]
|
|
|
past = ev_dt <= now_et
|
|
|
wait = (ev_dt - now_et).total_seconds()
|
|
|
schedule_view.append({
|
|
|
"name": ev["name"],
|
|
|
"kind": ev["kind"],
|
|
|
"session": ev.get("session", ""),
|
|
|
"label": ev["label"],
|
|
|
"et_time": ev_dt.strftime("%H:%M ET"),
|
|
|
"et_iso": ev_dt.isoformat(),
|
|
|
"past": past,
|
|
|
"done": ev["name"] in self._completed,
|
|
|
"wait_secs": max(0, wait),
|
|
|
})
|
|
|
return {
|
|
|
"running": self.running,
|
|
|
"sessions": self._sessions,
|
|
|
"dry_run": self._dry_run,
|
|
|
"schedule": schedule_view,
|
|
|
"log_tail": list(self._log_lines), # full log, not truncated
|
|
|
"log_line_count": self.log_line_count,
|
|
|
}
|
|
|
|
|
|
# ── State persistence ──────────────────────────────────────────────────────
|
|
|
|
|
|
def _save_state(self) -> None:
|
|
|
try:
|
|
|
_state_file_path(self._db_path).write_text(json.dumps({
|
|
|
"running": True,
|
|
|
"sessions": self._sessions,
|
|
|
"dry_run": self._dry_run,
|
|
|
"db_path": self._db_path,
|
|
|
}))
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
def _clear_state(self) -> None:
|
|
|
try:
|
|
|
sf = _state_file_path(self._db_path)
|
|
|
if sf.exists():
|
|
|
sf.unlink()
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
# ── Helpers ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _log(self, msg: str) -> None:
|
|
|
ts = datetime.now(tz=_TZ_PHOENIX).strftime("%H:%M MST")
|
|
|
line = f"{ts} {msg}"
|
|
|
self._log_lines.append(line)
|
|
|
log.info("[ORBScheduler] %s", msg)
|
|
|
# Persist to file so logs survive server restarts
|
|
|
try:
|
|
|
with self._log_file_path().open("a", encoding="utf-8") as f:
|
|
|
f.write(line + "\n")
|
|
|
except Exception:
|
|
|
pass
|
|
|
# Mirror to events.db with source=auto_scheduler so the Logs/Health UI
|
|
|
# can show scheduler activity inline with engine events.
|
|
|
_scheduler_emit("scheduler_log", message=msg)
|
|
|
|
|
|
def _now_et(self) -> dt.datetime:
|
|
|
return dt.datetime.now(tz=_TZ_ET)
|
|
|
|
|
|
def _is_trading_day(self, date: dt.date) -> bool:
|
|
|
try:
|
|
|
from libs.common.time_utils import is_trading_day
|
|
|
return is_trading_day(date)
|
|
|
except Exception:
|
|
|
return date.weekday() < 5
|
|
|
|
|
|
def _next_trading_day(self, from_date: dt.date) -> dt.date:
|
|
|
check = from_date + dt.timedelta(days=1)
|
|
|
for _ in range(14):
|
|
|
if self._is_trading_day(check):
|
|
|
return check
|
|
|
check += dt.timedelta(days=1)
|
|
|
raise RuntimeError("No trading day found in next 14 days")
|
|
|
|
|
|
@staticmethod
|
|
|
def _fmt_countdown(seconds: float) -> str:
|
|
|
if seconds <= 0:
|
|
|
return "now"
|
|
|
h = int(seconds // 3600)
|
|
|
m = int((seconds % 3600) // 60)
|
|
|
s = int(seconds % 60)
|
|
|
if h > 0:
|
|
|
return f"{h}h {m:02d}m"
|
|
|
if m > 0:
|
|
|
return f"{m}m {s:02d}s"
|
|
|
return f"{s}s"
|
|
|
|
|
|
def _get_active_sessions(self) -> list[str]:
|
|
|
try:
|
|
|
from apps.orb_trader.state import ORBStateManager
|
|
|
return [
|
|
|
s.session_name
|
|
|
for s in ORBStateManager(self._db_path).list_sessions()
|
|
|
if s.status == "active"
|
|
|
]
|
|
|
except Exception:
|
|
|
return []
|
|
|
|
|
|
def _mark_past_events_completed(self, now_et: dt.datetime) -> None:
|
|
|
"""Mark already-past events as completed after a schedule rebuild."""
|
|
|
for ev in self._today_schedule:
|
|
|
if ev["et_dt"] <= now_et and ev["name"] not in self._completed:
|
|
|
self._completed.add(ev["name"])
|
|
|
self._log(f"Past (skipped): {ev['label']}")
|
|
|
|
|
|
def _build_today_schedule(self, date: dt.date, sessions: list[str]) -> list[dict[str, Any]]:
|
|
|
"""Build per-session schedules and merge into one sorted timeline.
|
|
|
|
|
|
Each event is tagged with its session so the loop can dispatch
|
|
|
the right engine for the right session at the right time.
|
|
|
Sessions that no longer exist in the DB are skipped with a warning.
|
|
|
"""
|
|
|
from apps.orb_trader.state import ORBStateManager
|
|
|
state_mgr = ORBStateManager(self._db_path)
|
|
|
|
|
|
combined: list[dict[str, Any]] = []
|
|
|
valid_sessions: list[str] = []
|
|
|
for session_name in sessions:
|
|
|
# Verify session still exists in DB
|
|
|
session_obj = state_mgr.get_session(session_name)
|
|
|
if session_obj is None:
|
|
|
self._log(f" WARNING: session '{session_name}' not found in DB — skipping")
|
|
|
continue
|
|
|
valid_sessions.append(session_name)
|
|
|
params = _load_session_params(self._db_path, session_name)
|
|
|
events = build_schedule(date, **params)
|
|
|
for ev in events:
|
|
|
ev["session"] = session_name
|
|
|
ev["name"] = f"{session_name}:{ev['name']}"
|
|
|
combined.extend(events)
|
|
|
self._log(
|
|
|
f" {session_name}: orb={params['orb_minutes']}min, "
|
|
|
f"timeout={params['order_timeout_minutes']}min, "
|
|
|
f"bar={params['sim_bar_minutes']}min → {len(events)} events"
|
|
|
)
|
|
|
|
|
|
if not valid_sessions and sessions:
|
|
|
# All specified sessions are gone — fall back to all active sessions
|
|
|
active = self._get_active_sessions()
|
|
|
self._log(
|
|
|
f" All specified sessions missing; falling back to active sessions: "
|
|
|
f"{', '.join(active) or 'none'}"
|
|
|
)
|
|
|
return self._build_today_schedule(date, active)
|
|
|
|
|
|
combined.sort(key=lambda e: (e["et_dt"], e["session"]))
|
|
|
self._log(f"Total schedule: {len(combined)} events across {len(valid_sessions)} session(s)")
|
|
|
return combined
|
|
|
|
|
|
def _get_or_create_engine(self, session_name: str, date_str: str) -> Any:
|
|
|
if self._engine_date != date_str:
|
|
|
self._engines = {}
|
|
|
self._engine_date = date_str
|
|
|
|
|
|
if session_name not in self._engines:
|
|
|
try:
|
|
|
from apps.orb_trader.state import ORBStateManager
|
|
|
from apps.orb_trader.engine import make_orb_engine
|
|
|
state_mgr = ORBStateManager(self._db_path)
|
|
|
session = state_mgr.get_session(session_name)
|
|
|
if session is None:
|
|
|
return None
|
|
|
self._engines[session_name] = make_orb_engine(
|
|
|
session, self._db_path, log_callback=self._log
|
|
|
)
|
|
|
except Exception as exc:
|
|
|
self._log(f" ERROR creating engine for {session_name}: {exc}")
|
|
|
return None
|
|
|
|
|
|
return self._engines[session_name]
|
|
|
|
|
|
# ── Engine operation runners ───────────────────────────────────────────────
|
|
|
|
|
|
async def _run_trading(self, kind: str, sessions: list[str], date_str: str) -> None:
|
|
|
# ORB 윈도우 모니터링은 no-op (범위 형성 중, 장중 데이터는 orb_detect에서 일괄 fetch)
|
|
|
if kind == "orb_monitor":
|
|
|
self._log(f" ORB 윈도우 모니터링 중...")
|
|
|
return
|
|
|
|
|
|
from apps.orb_trader.state import ORBStateManager
|
|
|
state_mgr = ORBStateManager(self._db_path)
|
|
|
|
|
|
for session_name in sessions:
|
|
|
session = state_mgr.get_session(session_name)
|
|
|
if session is None or session.status != "active":
|
|
|
self._log(f" {session_name}: skipped (not active)")
|
|
|
continue
|
|
|
if self._dry_run:
|
|
|
self._log(f" [DRY] {kind} — {session_name}")
|
|
|
continue
|
|
|
|
|
|
engine = self._get_or_create_engine(session_name, date_str)
|
|
|
if engine is None:
|
|
|
continue
|
|
|
|
|
|
# Bind a fresh job_run_id per (session, kind) so every engine event
|
|
|
# in this run groups into a single timeline in the Logs/Health UI.
|
|
|
phase_name = _PHASE_NAME_MAP.get(kind, kind)
|
|
|
phase_run_id = str(uuid.uuid4())
|
|
|
if _bind_job_run_id is not None:
|
|
|
try:
|
|
|
_bind_job_run_id(phase_run_id)
|
|
|
except Exception:
|
|
|
pass
|
|
|
phase_started = time.monotonic()
|
|
|
_scheduler_emit(
|
|
|
"phase_started",
|
|
|
job_run_id=phase_run_id,
|
|
|
phase=phase_name, session_id=session.session_id,
|
|
|
session_name=session_name, category="lifecycle",
|
|
|
)
|
|
|
|
|
|
self._log(f" ▶ {kind} — {session_name}")
|
|
|
phase_status = "success"
|
|
|
try:
|
|
|
def _run_sync(e=engine, k=kind, d=date_str) -> dict[str, Any]:
|
|
|
import asyncio as _asyncio
|
|
|
loop = _asyncio.new_event_loop()
|
|
|
_asyncio.set_event_loop(loop)
|
|
|
try:
|
|
|
if k == "pre_screen":
|
|
|
return e.run_pre_screen(d)
|
|
|
elif k == "orb_detect":
|
|
|
return e.run_orb_detection(d)
|
|
|
elif k == "breakout":
|
|
|
return e.run_breakout_check(d)
|
|
|
elif k == "stop_check":
|
|
|
return e.run_stop_check(d)
|
|
|
elif k == "eod_exit":
|
|
|
return e.run_eod_exit(d)
|
|
|
elif k == "post_close":
|
|
|
return e.run_post_close(d)
|
|
|
return {}
|
|
|
finally:
|
|
|
loop.close()
|
|
|
|
|
|
summary = await asyncio.to_thread(_run_sync)
|
|
|
self._log(f" ✓ {session_name}: {summary}")
|
|
|
except Exception as exc:
|
|
|
tb = traceback.format_exc()
|
|
|
phase_status = "failed"
|
|
|
self._log(f" ERROR {session_name}: {exc}")
|
|
|
log.error("ORB engine error: %s\n%s", exc, tb)
|
|
|
_scheduler_emit(
|
|
|
"phase_completed",
|
|
|
level="ERROR",
|
|
|
job_run_id=phase_run_id,
|
|
|
phase=phase_name, status="failed",
|
|
|
session_id=session.session_id, session_name=session_name,
|
|
|
category="lifecycle",
|
|
|
duration_s=round(time.monotonic() - phase_started, 3),
|
|
|
error=str(exc),
|
|
|
)
|
|
|
continue
|
|
|
_scheduler_emit(
|
|
|
"phase_completed",
|
|
|
job_run_id=phase_run_id,
|
|
|
phase=phase_name, status=phase_status,
|
|
|
session_id=session.session_id, session_name=session_name,
|
|
|
category="lifecycle",
|
|
|
duration_s=round(time.monotonic() - phase_started, 3),
|
|
|
)
|
|
|
|
|
|
# ── Run-now: manually trigger detection for a late-added session ─────────
|
|
|
|
|
|
async def run_session_now(self, session_name: str) -> dict[str, Any]:
|
|
|
"""Run ORB detection immediately for a session that missed the morning window.
|
|
|
|
|
|
After detection, injects the session's remaining today-events into the
|
|
|
active schedule so breakout checks, stop checks, and EOD exit still fire.
|
|
|
|
|
|
All exceptions are caught and logged so the fire-and-forget task never
|
|
|
silently vanishes.
|
|
|
"""
|
|
|
# Log immediately so the user knows the task started — even before any
|
|
|
# DB / engine work that might fail.
|
|
|
self._log(f"🔄 지금 시작 요청: {session_name}")
|
|
|
|
|
|
try:
|
|
|
from apps.orb_trader.state import ORBStateManager
|
|
|
|
|
|
now_et = self._now_et()
|
|
|
today = now_et.date()
|
|
|
date_str = today.isoformat()
|
|
|
|
|
|
state_mgr = ORBStateManager(self._db_path)
|
|
|
session = state_mgr.get_session(session_name)
|
|
|
if session is None:
|
|
|
self._log(f" ❌ 세션 '{session_name}' 을 DB에서 찾을 수 없음")
|
|
|
return {"error": f"Session '{session_name}' not found"}
|
|
|
|
|
|
# Already ran if daily_state.phase is beyond pre_screen/idle
|
|
|
daily = state_mgr.get_daily_state(session.session_id, date_str)
|
|
|
if daily.phase not in ("idle", "", None, "pre_screen"):
|
|
|
self._log(f" ⚠ {session_name}: 오늘 이미 실행됨 (phase={daily.phase})")
|
|
|
return {"error": f"already ran today (phase={daily.phase})"}
|
|
|
|
|
|
# Phase 1: Pre-market screening (narrows universe for faster orb_detect)
|
|
|
self._log(f" 사전 스크리닝 실행 중...")
|
|
|
await self._run_trading("pre_screen", [session_name], date_str)
|
|
|
|
|
|
self._log(f" ORB 감지 실행 중 (현재 시세 기준)...")
|
|
|
|
|
|
# Phase 2: ORB detection — fetches intraday bars for pre-screened tickers
|
|
|
await self._run_trading("orb_detect", [session_name], date_str)
|
|
|
|
|
|
# Phase 2: Immediate breakout check using current snapshot prices.
|
|
|
# This is the core of "지금 시작" — the scheduled breakout windows have
|
|
|
# already passed, so we check once manually right now.
|
|
|
self._log(f" 현재 가격으로 브레이크아웃 즉시 체크...")
|
|
|
await self._run_trading("breakout", [session_name], date_str)
|
|
|
|
|
|
# Phase 3: Inject remaining future events (stop checks + EOD exit only).
|
|
|
# Mark all scheduled breakout events as done — we already ran one above.
|
|
|
params = _load_session_params(self._db_path, session_name)
|
|
|
all_events = build_schedule(today, **params)
|
|
|
for ev in all_events:
|
|
|
ev["session"] = session_name
|
|
|
ev["name"] = f"{session_name}:{ev['name']}"
|
|
|
|
|
|
existing_names = {e["name"] for e in self._today_schedule}
|
|
|
injected = 0
|
|
|
for ev in all_events:
|
|
|
if ev["kind"] == "breakout":
|
|
|
self._completed.add(ev["name"])
|
|
|
elif ev["et_dt"] <= now_et:
|
|
|
self._completed.add(ev["name"])
|
|
|
elif ev["name"] not in existing_names:
|
|
|
self._today_schedule.append(ev)
|
|
|
injected += 1
|
|
|
|
|
|
self._today_schedule.sort(key=lambda e: (e["et_dt"], e.get("session", "")))
|
|
|
self._save_schedule_state()
|
|
|
self._log(f" ✓ 완료 — 스톱/EOD 이벤트 {injected}개 추가됨")
|
|
|
return {"session": session_name, "injected": injected}
|
|
|
|
|
|
except Exception as exc:
|
|
|
tb = traceback.format_exc()
|
|
|
self._log(f" ❌ 지금 시작 오류: {exc}")
|
|
|
log.error("run_session_now error: %s\n%s", exc, tb)
|
|
|
return {"error": str(exc)}
|
|
|
|
|
|
# ── Main scheduler loop ────────────────────────────────────────────────────
|
|
|
|
|
|
async def _run_loop(self) -> None:
|
|
|
resolved = self._sessions or self._get_active_sessions()
|
|
|
if not resolved:
|
|
|
self._log("No active ORB sessions. Stopping.")
|
|
|
return
|
|
|
|
|
|
self._log(f"ORB auto-scheduler started — sessions: {', '.join(resolved)}")
|
|
|
if self._dry_run:
|
|
|
self._log("DRY RUN — orders will not be placed")
|
|
|
|
|
|
last_schedule_date: dt.date | None = None
|
|
|
|
|
|
try:
|
|
|
while True:
|
|
|
now_et = self._now_et()
|
|
|
today = now_et.date()
|
|
|
date_str = today.isoformat()
|
|
|
|
|
|
# ── New trading day: rebuild schedule ─────────────────────────
|
|
|
if last_schedule_date != today:
|
|
|
self._completed.clear()
|
|
|
last_schedule_date = today
|
|
|
|
|
|
if not self._sessions:
|
|
|
fresh = self._get_active_sessions()
|
|
|
if set(fresh) != set(resolved):
|
|
|
self._log(f"Sessions refreshed: {', '.join(fresh) or 'none'}")
|
|
|
resolved = fresh
|
|
|
|
|
|
self._log(
|
|
|
f"━━━ {today.strftime('%a %Y-%m-%d')} "
|
|
|
f"— sessions: {', '.join(resolved) or 'none'} ━━━"
|
|
|
)
|
|
|
|
|
|
if not self._is_trading_day(today):
|
|
|
self._today_schedule = []
|
|
|
next_td = self._next_trading_day(today)
|
|
|
self._log(f"Non-trading day. Next: {next_td}")
|
|
|
else:
|
|
|
self._today_schedule = self._build_today_schedule(today, resolved)
|
|
|
# Skip events already past
|
|
|
self._mark_past_events_completed(now_et)
|
|
|
self._save_schedule_state()
|
|
|
|
|
|
if not self._is_trading_day(today):
|
|
|
await asyncio.sleep(1800)
|
|
|
continue
|
|
|
|
|
|
if not resolved:
|
|
|
await asyncio.sleep(300)
|
|
|
continue
|
|
|
|
|
|
# Detect stale sessions in both explicit and auto-discovery modes.
|
|
|
# Covers the case where a session is deleted/replaced after the
|
|
|
# midnight auto-discovery refresh but before trading hours.
|
|
|
active_now = self._get_active_sessions()
|
|
|
active_resolved = [s for s in resolved if s in active_now]
|
|
|
if not active_resolved and active_now:
|
|
|
self._log(
|
|
|
"Active sessions changed; "
|
|
|
f"refreshing to: {', '.join(active_now)}"
|
|
|
)
|
|
|
resolved = active_now
|
|
|
self._today_schedule = self._build_today_schedule(today, resolved)
|
|
|
self._mark_past_events_completed(now_et)
|
|
|
self._save_schedule_state()
|
|
|
|
|
|
pending = [ev for ev in self._today_schedule if ev["name"] not in self._completed]
|
|
|
|
|
|
if not pending:
|
|
|
next_td = self._next_trading_day(today)
|
|
|
# Wake up just before the ORB detect event of next trading day
|
|
|
# (schedule isn't built yet, assume market open 9:30 ET)
|
|
|
wake_et = dt.datetime(
|
|
|
next_td.year, next_td.month, next_td.day, 9, 25, tzinfo=_TZ_ET
|
|
|
)
|
|
|
wait = (wake_et - now_et).total_seconds()
|
|
|
self._log(
|
|
|
f"All done today. Sleeping until "
|
|
|
f"{wake_et.strftime('%I:%M %p ET')} on {next_td} "
|
|
|
f"({self._fmt_countdown(wait)})"
|
|
|
)
|
|
|
await asyncio.sleep(min(wait, 3600))
|
|
|
continue
|
|
|
|
|
|
next_ev = pending[0]
|
|
|
next_et = next_ev["et_dt"]
|
|
|
wait = (next_et - now_et).total_seconds()
|
|
|
|
|
|
if wait > 90:
|
|
|
label = next_ev.get("label", next_ev["name"])
|
|
|
sess = next_ev.get("session", "")
|
|
|
self._log(f"Next: [{sess}] {label} — {self._fmt_countdown(wait)}")
|
|
|
await asyncio.sleep(min(wait - 60, 600))
|
|
|
continue
|
|
|
|
|
|
if wait > 0:
|
|
|
await asyncio.sleep(wait)
|
|
|
|
|
|
# ── Execute all events at this time slot (per session) ─────────
|
|
|
batch = [ev for ev in pending if ev["et_dt"] == next_et]
|
|
|
for ev in batch:
|
|
|
session = ev.get("session", "")
|
|
|
self._log(f"▶ [{session}] {ev['label']}")
|
|
|
try:
|
|
|
await self._run_trading(ev["kind"], [session] if session else resolved, date_str)
|
|
|
except Exception as exc:
|
|
|
self._log(f" ERROR [{session}]: {exc}")
|
|
|
self._completed.add(ev["name"])
|
|
|
self._save_schedule_state()
|
|
|
|
|
|
except asyncio.CancelledError:
|
|
|
self._log("ORB auto-scheduler stopped.")
|
|
|
raise
|
|
|
|
|
|
|
|
|
# ── ORBDaemonController ───────────────────────────────────────────────────────
|
|
|
|
|
|
_PROJECT_ROOT = Path(__file__).parent.parent.parent
|
|
|
|
|
|
|
|
|
class ORBDaemonController:
|
|
|
"""Proxy that manages the ORB scheduler as a separate subprocess.
|
|
|
|
|
|
The daemon (apps.orb_trader.daemon) runs independently of the web server,
|
|
|
so HMR / server restarts do not kill active trading sessions.
|
|
|
|
|
|
Communication:
|
|
|
.orb_scheduler.pid — daemon writes its PID on startup
|
|
|
.orb_auto_state.json — sessions / dry_run / db_path (written by ORBAutoScheduler)
|
|
|
.orb_schedule_state.json— today's schedule + completed set (written by daemon)
|
|
|
orb_scheduler.log — human-readable log (written by daemon)
|
|
|
.orb_trigger_*.json — trigger files (web writes, daemon polls every 3 s)
|
|
|
"""
|
|
|
|
|
|
def __init__(self) -> None:
|
|
|
self._db_path: str = _DEFAULT_DB
|
|
|
self._proc: subprocess.Popen | None = None # type: ignore[type-arg]
|
|
|
|
|
|
def _pid_file(self) -> Path:
|
|
|
return Path(self._db_path).parent / ".orb_scheduler.pid"
|
|
|
|
|
|
def _trigger_dir(self) -> Path:
|
|
|
return Path(self._db_path).parent
|
|
|
|
|
|
@property
|
|
|
def running(self) -> bool:
|
|
|
"""True if the daemon process is alive (PID file + kill-0 check, zombie-safe)."""
|
|
|
pid_file = self._pid_file()
|
|
|
if not pid_file.exists():
|
|
|
return False
|
|
|
try:
|
|
|
pid = int(pid_file.read_text().strip())
|
|
|
os.kill(pid, 0) # signal 0: probe without sending anything
|
|
|
# kill-0 succeeds for zombie processes too — check /proc or ps to exclude zombies
|
|
|
try:
|
|
|
import subprocess as _sp
|
|
|
result = _sp.run(
|
|
|
["ps", "-p", str(pid), "-o", "stat="],
|
|
|
capture_output=True, text=True, timeout=2,
|
|
|
)
|
|
|
stat = result.stdout.strip()
|
|
|
if stat.startswith("Z"):
|
|
|
# Zombie — treat as dead, clean up PID file
|
|
|
try:
|
|
|
pid_file.unlink()
|
|
|
except Exception:
|
|
|
pass
|
|
|
return False
|
|
|
except Exception:
|
|
|
pass # if ps fails, trust kill-0 result
|
|
|
return True
|
|
|
except (ValueError, ProcessLookupError, PermissionError, OSError):
|
|
|
# Process dead or PID stale — remove the file
|
|
|
try:
|
|
|
pid_file.unlink()
|
|
|
except Exception:
|
|
|
pass
|
|
|
return False
|
|
|
|
|
|
def start(self, sessions: list[str], db_path: str, dry_run: bool = False) -> None:
|
|
|
self._db_path = db_path # set FIRST so _pid_file() uses the right dir
|
|
|
if self.running:
|
|
|
requested_sessions = [s.strip() for s in sessions if s and s.strip()]
|
|
|
saved = load_orb_saved_state(self._db_path) or {}
|
|
|
current_sessions = [
|
|
|
str(s).strip()
|
|
|
for s in saved.get("sessions", [])
|
|
|
if str(s).strip()
|
|
|
]
|
|
|
current_dry_run = bool(saved.get("dry_run", False))
|
|
|
if current_sessions == requested_sessions and current_dry_run == dry_run:
|
|
|
return # daemon already alive with the requested configuration
|
|
|
|
|
|
# A no-op here leaves the old daemon pinned to stale sessions. Restart
|
|
|
# so "start selected sessions" from the UI actually updates trading.
|
|
|
self.stop()
|
|
|
deadline = time.time() + 5.0
|
|
|
while self.running and time.time() < deadline:
|
|
|
time.sleep(0.1)
|
|
|
if self.running:
|
|
|
raise RuntimeError("ORB daemon is still running after stop request")
|
|
|
|
|
|
cmd = [
|
|
|
sys.executable, "-m", "apps.orb_trader.daemon",
|
|
|
"--db-path", db_path,
|
|
|
]
|
|
|
if sessions:
|
|
|
cmd += ["--sessions", ",".join(sessions)]
|
|
|
if dry_run:
|
|
|
cmd.append("--dry-run")
|
|
|
|
|
|
self._proc = subprocess.Popen(
|
|
|
cmd,
|
|
|
cwd=str(_PROJECT_ROOT),
|
|
|
start_new_session=True, # survives parent death
|
|
|
stdout=subprocess.DEVNULL,
|
|
|
stderr=subprocess.DEVNULL,
|
|
|
)
|
|
|
|
|
|
# Wait up to 2 s for daemon to write PID file
|
|
|
for _ in range(20):
|
|
|
time.sleep(0.1)
|
|
|
if self.running:
|
|
|
break
|
|
|
|
|
|
def stop(self) -> None:
|
|
|
"""Send SIGTERM to daemon and clean up state files."""
|
|
|
pid_file = self._pid_file()
|
|
|
if pid_file.exists():
|
|
|
try:
|
|
|
pid = int(pid_file.read_text().strip())
|
|
|
os.kill(pid, signal.SIGTERM)
|
|
|
except Exception:
|
|
|
pass
|
|
|
try:
|
|
|
pid_file.unlink()
|
|
|
except Exception:
|
|
|
pass
|
|
|
# Clear .orb_auto_state.json so lifespan doesn't auto-restart
|
|
|
try:
|
|
|
sf = _state_file_path(self._db_path)
|
|
|
if sf.exists():
|
|
|
sf.unlink()
|
|
|
except Exception:
|
|
|
pass
|
|
|
# Clear schedule state
|
|
|
try:
|
|
|
ssf = Path(self._db_path).parent / ".orb_schedule_state.json"
|
|
|
if ssf.exists():
|
|
|
ssf.unlink()
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
def shutdown(self) -> None:
|
|
|
"""Web server is shutting down — no-op; daemon survives."""
|
|
|
pass
|
|
|
|
|
|
def clear_log(self) -> None:
|
|
|
"""Delete the log file."""
|
|
|
try:
|
|
|
lf = Path(self._db_path).parent / "orb_scheduler.log"
|
|
|
if lf.exists():
|
|
|
lf.unlink()
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
def run_session_now(self, session_name: str) -> None:
|
|
|
"""Write a trigger file; daemon picks it up within 3 seconds."""
|
|
|
trigger_file = self._trigger_dir() / f".orb_trigger_{uuid.uuid4().hex[:8]}.json"
|
|
|
trigger_file.write_text(json.dumps({
|
|
|
"command": "run_session_now",
|
|
|
"session": session_name,
|
|
|
"ts": time.time(),
|
|
|
}))
|
|
|
|
|
|
def get_status(self) -> dict[str, Any]:
|
|
|
"""Read status from files written by the daemon."""
|
|
|
now_et = dt.datetime.now(tz=_TZ_ET)
|
|
|
|
|
|
# .orb_auto_state.json — sessions / dry_run
|
|
|
state = load_orb_saved_state(self._db_path) or {}
|
|
|
sessions = state.get("sessions", [])
|
|
|
dry_run = state.get("dry_run", False)
|
|
|
|
|
|
# .orb_schedule_state.json — schedule + completed
|
|
|
schedule_view: list[dict[str, Any]] = []
|
|
|
completed: set[str] = set()
|
|
|
try:
|
|
|
ssf = Path(self._db_path).parent / ".orb_schedule_state.json"
|
|
|
if ssf.exists():
|
|
|
sched_state = json.loads(ssf.read_text())
|
|
|
completed = set(sched_state.get("completed", []))
|
|
|
for ev in sched_state.get("schedule", []):
|
|
|
ev_dt = dt.datetime.fromisoformat(ev["et_iso"])
|
|
|
past = ev_dt <= now_et
|
|
|
wait = (ev_dt - now_et).total_seconds()
|
|
|
schedule_view.append({
|
|
|
"name": ev["name"],
|
|
|
"kind": ev["kind"],
|
|
|
"session": ev.get("session", ""),
|
|
|
"label": ev["label"],
|
|
|
"et_time": ev_dt.strftime("%H:%M ET"),
|
|
|
"et_iso": ev["et_iso"],
|
|
|
"past": past,
|
|
|
"done": ev["name"] in completed,
|
|
|
"wait_secs": max(0.0, wait),
|
|
|
})
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
# orb_scheduler.log — log tail
|
|
|
log_lines: list[str] = []
|
|
|
try:
|
|
|
lf = Path(self._db_path).parent / "orb_scheduler.log"
|
|
|
if lf.exists():
|
|
|
log_lines = lf.read_text(encoding="utf-8").splitlines()
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
return {
|
|
|
"running": self.running,
|
|
|
"sessions": sessions,
|
|
|
"dry_run": dry_run,
|
|
|
"schedule": schedule_view,
|
|
|
"log_tail": log_lines,
|
|
|
"log_line_count": len(log_lines),
|
|
|
}
|
|
|
|
|
|
|
|
|
# ── Module-level singleton ────────────────────────────────────────────────────
|
|
|
|
|
|
orb_auto_scheduler = ORBDaemonController()
|