You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

413 lines
15 KiB
Python

This file contains ambiguous Unicode characters!

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

"""ORB Paper Trading API endpoints."""
from __future__ import annotations
import datetime as dt
import os
from typing import Any
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
router = APIRouter(prefix="/orb", tags=["orb-trading"])
# ---------------------------------------------------------------------------
# DB path resolution
# ---------------------------------------------------------------------------
def _db_path() -> str:
from pathlib import Path
project_root = Path(__file__).parent.parent.parent.parent
env_path = os.environ.get("ORB_TRADER_DB", "data/paper/orb.db")
path = Path(env_path) if Path(env_path).is_absolute() else project_root / env_path
path.parent.mkdir(parents=True, exist_ok=True)
return str(path)
def _state() -> Any:
from apps.orb_trader.state import ORBStateManager
return ORBStateManager(_db_path())
# ---------------------------------------------------------------------------
# Request/Response models
# ---------------------------------------------------------------------------
class CreateSessionRequest(BaseModel):
name: str
config: str # config path (relative to project root)
capital: float = 10000.0
class AutoStartRequest(BaseModel):
sessions: list[str] = []
dry_run: bool = False
# ---------------------------------------------------------------------------
# Sessions
# ---------------------------------------------------------------------------
@router.get("/sessions")
def list_sessions() -> dict[str, Any]:
state = _state()
sessions = state.list_sessions()
today = dt.date.today().isoformat()
results = []
for s in sessions:
daily = state.get_daily_state(s.session_id, today)
# phase starts as "idle" (default); engine sets it to "orb_detection"
# or beyond when it actually runs today
ran_today = daily.phase not in ("idle", "", None)
results.append({
"session_id": s.session_id,
"session_name": s.session_name,
"config_path": s.config_path,
"initial_equity": s.initial_equity,
"created_at": s.created_at,
"status": s.status,
"ran_today": ran_today,
})
return {"sessions": results}
@router.post("/sessions")
def create_session(req: CreateSessionRequest) -> dict[str, Any]:
from pathlib import Path
import yaml
project_root = Path(__file__).parent.parent.parent.parent
config_path = req.config if Path(req.config).is_absolute() else str(project_root / req.config)
if not Path(config_path).exists():
raise HTTPException(status_code=404, detail=f"Config not found: {config_path}")
# Validate it's an ORB strategy config
try:
with open(config_path) as f:
raw = yaml.safe_load(f)
if raw.get("strategy_mode") != "orb":
raise HTTPException(
status_code=400,
detail=f"Config strategy_mode must be 'orb', got '{raw.get('strategy_mode')}'"
)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=400, detail=f"Invalid config: {e}")
state = _state()
# Check duplicate name
if state.get_session(req.name) is not None:
raise HTTPException(status_code=409, detail=f"Session '{req.name}' already exists")
session_id = state.create_session(req.name, config_path, req.capital)
return {
"session_id": session_id,
"session_name": req.name,
"config_path": config_path,
"initial_equity": req.capital,
}
@router.get("/sessions/{session_id}")
def get_session(session_id: str) -> dict[str, Any]:
session = _state().get_session(session_id)
if session is None:
raise HTTPException(status_code=404, detail="Session not found")
equity = _state().get_equity(session.session_id) or session.initial_equity
return {
"session_id": session.session_id,
"session_name": session.session_name,
"config_path": session.config_path,
"initial_equity": session.initial_equity,
"current_equity": equity,
"total_return_pct": (equity - session.initial_equity) / session.initial_equity * 100,
"created_at": session.created_at,
"status": session.status,
}
@router.post("/sessions/{session_id}/pause")
def pause_session(session_id: str) -> dict[str, Any]:
state = _state()
session = state.get_session(session_id)
if session is None:
raise HTTPException(status_code=404, detail="Session not found")
state.set_session_status(session.session_id, "paused")
return {"session_id": session.session_id, "status": "paused"}
@router.post("/sessions/{session_id}/resume")
def resume_session(session_id: str) -> dict[str, Any]:
state = _state()
session = state.get_session(session_id)
if session is None:
raise HTTPException(status_code=404, detail="Session not found")
state.set_session_status(session.session_id, "active")
return {"session_id": session.session_id, "status": "active"}
@router.post("/sessions/{session_id}/run_today")
async def run_session_today(session_id: str) -> dict[str, Any]:
"""Immediately run ORB detection for a session that missed the morning window.
Requires the auto-scheduler to be running. After detection, the session's
remaining today-events (breakout checks, stop checks, EOD exit) are injected
into the live schedule automatically.
"""
import asyncio as _asyncio
from apps.web.orb_trading_service import orb_auto_scheduler
state = _state()
session = state.get_session(session_id)
if session is None:
raise HTTPException(status_code=404, detail="Session not found")
if not orb_auto_scheduler.running:
raise HTTPException(
status_code=400,
detail="스케줄러가 실행 중이 아닙니다. 먼저 자동 스케줄러를 시작하세요.",
)
# Fire-and-forget — detection takes ~12 min; we return immediately
_asyncio.create_task(orb_auto_scheduler.run_session_now(session.session_name))
return {
"session_id": session.session_id,
"session_name": session.session_name,
"status": "started",
"note": "ORB 감지가 백그라운드에서 시작되었습니다. 스케줄러 로그를 확인하세요.",
}
@router.delete("/sessions/{session_id}")
def close_session(session_id: str) -> dict[str, Any]:
state = _state()
session = state.get_session(session_id)
if session is None:
raise HTTPException(status_code=404, detail="Session not found")
# Close all open Alpaca positions across all dates
positions_closed = 0
close_errors: list[str] = []
try:
from apps.paper_trader.alpaca_broker import AlpacaBroker
broker = AlpacaBroker.from_env()
open_pos = state.get_all_open_positions(session.session_id)
for pos in open_pos:
try:
# Always close only this session's shares (partial close if other
# sessions also hold the same ticker in the same Alpaca account)
broker.close_position(pos.ticker, qty=int(pos.shares))
positions_closed += 1
except Exception as e:
close_errors.append(f"{pos.ticker}: {e}")
except Exception as e:
# Broker init failed — refuse to delete so user can investigate
raise HTTPException(status_code=500, detail=f"Alpaca broker error: {e}")
# Always delete the session; report any per-ticker failures in response
state.delete_session(session.session_id)
return {
"deleted": True,
"positions_closed": positions_closed,
"close_errors": close_errors, # empty list if all succeeded
}
# ---------------------------------------------------------------------------
# Data endpoints
# ---------------------------------------------------------------------------
@router.get("/sessions/{session_id}/positions")
def get_positions(session_id: str) -> dict[str, Any]:
state = _state()
session = state.get_session(session_id)
if session is None:
raise HTTPException(status_code=404, detail="Session not found")
today = dt.date.today().isoformat()
positions = state.get_open_positions(session.session_id, today)
# Enrich with real-time prices from Oracle/Alpaca snapshot API
from libs.oracle_client.alpaca import get_snapshots, AlpacaSnapshot
snapshots: dict[str, AlpacaSnapshot] = {}
if positions:
tickers = [p.ticker for p in positions]
try:
snapshots = get_snapshots(tickers)
except Exception:
pass
def _pos_to_dict(p: Any) -> dict[str, Any]:
snap = snapshots.get(p.ticker)
current_price = (snap.price if snap and snap.price else None) or p.entry_price
bid = snap.bid if snap else None
ask = snap.ask if snap else None
change_pct = snap.change_pct if snap else None
if p.direction == "long":
unrealized_pnl = (current_price - p.entry_price) * p.shares
r_multiple = (current_price - p.entry_price) / p.stop_distance if p.stop_distance > 0 else 0
else:
unrealized_pnl = (p.entry_price - current_price) * p.shares
r_multiple = (p.entry_price - current_price) / p.stop_distance if p.stop_distance > 0 else 0
return {
"ticker": p.ticker,
"direction": p.direction,
"entry_price": p.entry_price,
"entry_time": p.entry_time,
"shares": p.shares,
"current_stop": p.current_stop,
"peak_price": p.peak_price,
"trailing_active": p.trailing_active,
"atr_at_entry": p.atr_at_entry,
"stop_distance": p.stop_distance,
"rvol": p.rvol,
"composite_score": p.composite_score,
"current_price": current_price,
"bid": bid,
"ask": ask,
"change_pct": change_pct,
"unrealized_pnl": round(unrealized_pnl, 2),
"r_multiple": round(r_multiple, 3),
}
return {"positions": [_pos_to_dict(p) for p in positions], "date": today}
@router.get("/sessions/{session_id}/trades")
def get_trades(session_id: str, last: int | None = None) -> dict[str, Any]:
state = _state()
session = state.get_session(session_id)
if session is None:
raise HTTPException(status_code=404, detail="Session not found")
trades = state.list_trades(session.session_id, limit=last)
return {"trades": trades, "total": len(trades)}
@router.get("/sessions/{session_id}/equity")
def get_equity(session_id: str) -> dict[str, Any]:
state = _state()
session = state.get_session(session_id)
if session is None:
raise HTTPException(status_code=404, detail="Session not found")
snapshots = state.list_snapshots(session.session_id)
return {"snapshots": snapshots, "initial_equity": session.initial_equity}
@router.get("/sessions/{session_id}/candidates")
def get_candidates(session_id: str) -> dict[str, Any]:
state = _state()
session = state.get_session(session_id)
if session is None:
raise HTTPException(status_code=404, detail="Session not found")
today = dt.date.today().isoformat()
candidates = state.list_candidates(session.session_id, today)
return {"candidates": candidates, "date": today}
# ---------------------------------------------------------------------------
# Auto scheduler
# ---------------------------------------------------------------------------
@router.get("/auto")
def get_auto_status() -> dict[str, Any]:
from apps.web.orb_trading_service import orb_auto_scheduler
return orb_auto_scheduler.get_status()
@router.post("/auto/start")
async def start_auto(req: AutoStartRequest) -> dict[str, Any]:
from apps.web.orb_trading_service import orb_auto_scheduler
if orb_auto_scheduler.running:
raise HTTPException(status_code=409, detail="Scheduler already running")
try:
orb_auto_scheduler.start(
sessions=req.sessions,
db_path=_db_path(),
dry_run=req.dry_run,
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
return {"started": True, "dry_run": req.dry_run, "sessions": req.sessions}
@router.post("/auto/clear-log")
def clear_auto_log() -> dict[str, Any]:
from apps.web.orb_trading_service import orb_auto_scheduler
orb_auto_scheduler.clear_log()
return {"cleared": True}
@router.post("/auto/stop")
async def stop_auto() -> dict[str, Any]:
from apps.web.orb_trading_service import orb_auto_scheduler
orb_auto_scheduler.stop()
return {"stopped": True}
# ---------------------------------------------------------------------------
# Strategies — list ORB configs with config_path for session creation
# ---------------------------------------------------------------------------
@router.get("/strategies")
def list_orb_strategies() -> dict[str, Any]:
"""List all ORB strategy configs available for paper trading sessions.
Returns both built-in presets and every *.yaml under
configs/intraday/strategies/ that has strategy_mode=orb.
Always includes config_path so the frontend can pass it to POST /sessions.
"""
from pathlib import Path
import yaml as _yaml
from apps.web.routers.intraday import _BUILTIN_STRATEGIES, get_project_root
strategies: list[dict[str, Any]] = []
# 1. Built-in presets (always include, they are all ORB)
for strat in _BUILTIN_STRATEGIES.values():
strategies.append({
"slug": strat["slug"],
"name": strat["name"],
"description": strat.get("description", ""),
"builtin": True,
"config_path": strat["config_path"],
"orb_minutes": strat.get("orb_minutes"),
"sim_bar_minutes": strat.get("sim_bar_minutes"),
"entry_direction": strat.get("entry_direction", "long_only"),
"risk_per_trade_pct": strat.get("risk_per_trade_pct"),
"atr_stop_multiplier": strat.get("atr_stop_multiplier"),
})
# 2. YAMLs in configs/intraday/strategies/ not already listed as builtins
builtin_paths = {s["config_path"] for s in _BUILTIN_STRATEGIES.values()}
strategies_dir = Path(get_project_root()) / "configs" / "intraday" / "strategies"
if strategies_dir.exists():
for yaml_file in sorted(strategies_dir.glob("*.yaml")):
rel_path = str(yaml_file.relative_to(get_project_root()))
if rel_path in builtin_paths:
continue
try:
raw = _yaml.safe_load(yaml_file.read_text()) or {}
if raw.get("strategy_mode") != "orb":
continue
meta = raw.get("_meta", {})
orb = raw.get("orb_strategy", {})
slug = yaml_file.stem
strategies.append({
"slug": slug,
"name": meta.get("name", slug),
"description": meta.get("description", ""),
"builtin": False,
"config_path": rel_path,
"orb_minutes": orb.get("orb_minutes"),
"sim_bar_minutes": orb.get("sim_bar_minutes"),
"entry_direction": orb.get("entry_direction", "long_only"),
"risk_per_trade_pct": orb.get("risk_per_trade_pct"),
"atr_stop_multiplier": orb.get("atr_stop_multiplier"),
})
except Exception:
continue
return {"strategies": strategies}