Codex
155 lines
4.2 KiB
Python
155 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
|
|
def utc_now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Candle:
|
|
timestamp: int
|
|
open: float
|
|
high: float
|
|
low: float
|
|
close: float
|
|
volume: float
|
|
turnover: float = 0.0
|
|
ema_20: float | None = None
|
|
ema_50: float | None = None
|
|
ema_200: float | None = None
|
|
rsi_14: float | None = None
|
|
atr_14: float | None = None
|
|
macd: float | None = None
|
|
macd_signal: float | None = None
|
|
macd_hist: float | None = None
|
|
volume_ma_20: float | None = None
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Ticker:
|
|
symbol: str
|
|
last_price: float
|
|
bid: float
|
|
ask: float
|
|
turnover_24h: float
|
|
volume_24h: float
|
|
change_24h: float
|
|
updated_at: datetime = field(default_factory=utc_now)
|
|
|
|
@property
|
|
def spread_percent(self) -> float:
|
|
if self.bid <= 0 or self.ask <= 0:
|
|
return 0.0
|
|
mid = (self.ask + self.bid) / 2
|
|
return ((self.ask - self.bid) / mid) * 100 if mid else 0.0
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
data = asdict(self)
|
|
data["updated_at"] = self.updated_at.isoformat()
|
|
data["spread_percent"] = self.spread_percent
|
|
return data
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Signal:
|
|
symbol: str
|
|
action: str
|
|
confidence: float
|
|
reason: str
|
|
diagnostics: dict[str, Any] = field(default_factory=dict)
|
|
created_at: datetime = field(default_factory=utc_now)
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
data = asdict(self)
|
|
data["created_at"] = self.created_at.isoformat()
|
|
return data
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Position:
|
|
id: int | None
|
|
symbol: str
|
|
qty: float
|
|
entry_price: float
|
|
notional_usdt: float
|
|
entry_fee_usdt: float
|
|
stop_loss: float
|
|
take_profit: float
|
|
highest_price: float
|
|
opened_at: datetime = field(default_factory=utc_now)
|
|
entry_reason: str = ""
|
|
entry_confidence: float = 0.0
|
|
entry_pattern: str = ""
|
|
|
|
def mark_price(self, price: float) -> float:
|
|
return self.qty * price
|
|
|
|
def unrealized_pnl(self, price: float) -> float:
|
|
return (price - self.entry_price) * self.qty - self.entry_fee_usdt
|
|
|
|
def trailing_stop(self, percent: float) -> float | None:
|
|
stop = self.highest_price * (1 - percent)
|
|
return stop if stop > self.entry_price else None
|
|
|
|
def as_dict(self, mark_price: float | None = None) -> dict[str, Any]:
|
|
data = asdict(self)
|
|
data["opened_at"] = self.opened_at.isoformat()
|
|
if mark_price is not None:
|
|
data["mark_price"] = mark_price
|
|
data["market_value"] = self.mark_price(mark_price)
|
|
data["unrealized_pnl"] = self.unrealized_pnl(mark_price)
|
|
data["unrealized_pnl_percent"] = (
|
|
self.unrealized_pnl(mark_price) / self.notional_usdt * 100
|
|
if self.notional_usdt
|
|
else 0.0
|
|
)
|
|
return data
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Trade:
|
|
id: int | None
|
|
symbol: str
|
|
side: str
|
|
qty: float
|
|
entry_price: float | None = None
|
|
exit_price: float | None = None
|
|
gross_pnl: float = 0.0
|
|
fee_usdt: float = 0.0
|
|
net_pnl: float = 0.0
|
|
reason: str = ""
|
|
entry_pattern: str = ""
|
|
entry_confidence: float = 0.0
|
|
opened_at: datetime | None = None
|
|
closed_at: datetime | None = None
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
data = asdict(self)
|
|
data["opened_at"] = self.opened_at.isoformat() if self.opened_at else None
|
|
data["closed_at"] = self.closed_at.isoformat() if self.closed_at else None
|
|
return data
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class BotStatus:
|
|
running: bool
|
|
mode: str
|
|
live_trading_ready: bool
|
|
symbols: list[str]
|
|
started_at: datetime | None
|
|
last_loop_at: datetime | None
|
|
message: str = ""
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
data = asdict(self)
|
|
data["started_at"] = self.started_at.isoformat() if self.started_at else None
|
|
data["last_loop_at"] = self.last_loop_at.isoformat() if self.last_loop_at else None
|
|
return data
|