Spaces:
Running
Running
File size: 3,678 Bytes
d361784 | 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 | from __future__ import annotations
from aureon_baton_link import link_system as _baton_link; _baton_link(__name__)
import random
import threading
import time
from dataclasses import dataclass, field
from typing import List
from .config_manager import TradingConfig
@dataclass
class Position:
symbol: str
size: float
entry_price: float
@dataclass
class RuntimeState:
running: bool = False
pnl: float = 0.0
positions: List[Position] = field(default_factory=list)
logs: List[str] = field(default_factory=list)
class TradingService:
def __init__(self, config: TradingConfig):
self.config = config
self.state = RuntimeState()
self._worker: threading.Thread | None = None
self._stop_event = threading.Event()
self._current_price = 100.0
def start(self) -> None:
if self.state.running:
self.state.logs.append("Trading already running.")
return
self.state.running = True
self.state.logs.append(
f"Trading service started in {self.config.mode} mode for {self.config.base_asset}/{self.config.quote_asset}."
)
self._stop_event.clear()
self._worker = threading.Thread(target=self._run_loop, daemon=True)
self._worker.start()
def stop(self) -> None:
if not self.state.running:
return
self.state.logs.append("Stopping trading service...")
self._stop_event.set()
if self._worker and self._worker.is_alive():
self._worker.join(timeout=2)
self.state.running = False
self.state.logs.append("Trading stopped.")
def _run_loop(self) -> None:
while not self._stop_event.is_set():
jitter = random.uniform(-0.8, 1.5)
self._current_price = max(1.0, self._current_price + jitter)
# Mark-to-market for open positions
if self.state.positions:
pos = self.state.positions[0]
unrealized = (self._current_price - pos.entry_price) * pos.size * 0.01
self.state.pnl = round(unrealized, 2)
# Occasionally open or adjust a position
if random.random() > 0.75:
# Defensive check: trade_size is validated as positive in setup_wizard.py (lines 52-53),
# but this check is necessary in case the config is modified externally.
if self.config.trade_size <= 0:
self.state.logs.append("Trade size must be positive; adjust config to resume trading.")
else:
position = Position(
symbol=f"{self.config.base_asset}/{self.config.quote_asset}",
size=self.config.trade_size,
entry_price=round(self._current_price, 2),
)
self.state.positions = [position]
self.state.logs.append(
f"Opened {position.size} {position.symbol} @ {position.entry_price}"
)
if random.random() > 0.9 and self.state.positions:
closed = self.state.positions.pop()
realized = (self._current_price - closed.entry_price) * closed.size * 0.01
self.state.pnl = round(realized, 2)
self.state.logs.append(
f"Closed {closed.size} {closed.symbol} at {self._current_price:.2f} | P&L: {self.state.pnl}"
)
time.sleep(1)
__all__ = ["TradingService", "RuntimeState", "Position"]
|