Codex
312 lines
13 KiB
Python
312 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from datetime import datetime
|
|
|
|
from crypto_spot_bot.config import Settings
|
|
from crypto_spot_bot.execution import LiveBroker, PaperBroker
|
|
from crypto_spot_bot.learning import TradeLearner
|
|
from crypto_spot_bot.market_data import MarketData
|
|
from crypto_spot_bot.models import BotStatus, Signal, Ticker, utc_now
|
|
from crypto_spot_bot.patterns import PatternAnalyzer
|
|
from crypto_spot_bot.strategy import SpotStrategy
|
|
from crypto_spot_bot.storage import Storage
|
|
from crypto_spot_bot.time_series import TimeSeriesForecaster
|
|
|
|
|
|
class CryptoSpotBot:
|
|
def __init__(
|
|
self,
|
|
settings: Settings,
|
|
storage: Storage,
|
|
market: MarketData,
|
|
broker: PaperBroker | LiveBroker,
|
|
strategy: SpotStrategy,
|
|
pattern_analyzer: PatternAnalyzer,
|
|
learner: TradeLearner,
|
|
forecaster: TimeSeriesForecaster | None = None,
|
|
llm_advisor=None,
|
|
):
|
|
self.settings = settings
|
|
self.storage = storage
|
|
self.market = market
|
|
self.broker = broker
|
|
self.strategy = strategy
|
|
self.pattern_analyzer = pattern_analyzer
|
|
self.learner = learner
|
|
self.forecaster = forecaster
|
|
self.llm_advisor = llm_advisor
|
|
self.running = False
|
|
self.started_at: datetime | None = None
|
|
self.last_loop_at: datetime | None = None
|
|
self.message = "бот остановлен"
|
|
self._entry_cooldown_until: dict[str, datetime] = {}
|
|
self._loop_task: asyncio.Task | None = None
|
|
self._ws_task: asyncio.Task | None = None
|
|
|
|
async def start(self) -> None:
|
|
if self.running:
|
|
return
|
|
self.market.reset_stop()
|
|
if not self.market.symbols:
|
|
await self.market.bootstrap()
|
|
self._close_paper_positions_outside_symbol_universe()
|
|
self._update_patterns()
|
|
self._update_forecasts()
|
|
self.learner.refresh()
|
|
self.running = True
|
|
self.started_at = utc_now()
|
|
self.message = "бот работает"
|
|
self.storage.event("Бот запущен")
|
|
if self.settings.websocket_enabled:
|
|
self._ws_task = asyncio.create_task(self.market.websocket_loop())
|
|
self._loop_task = asyncio.create_task(self._run_loop())
|
|
|
|
async def stop(self) -> None:
|
|
self.running = False
|
|
self.message = "бот остановлен"
|
|
self.market.stop()
|
|
tasks = [task for task in (self._loop_task, self._ws_task) if task]
|
|
for task in tasks:
|
|
task.cancel()
|
|
if tasks:
|
|
await asyncio.gather(*tasks, return_exceptions=True)
|
|
self.storage.event("Бот остановлен")
|
|
|
|
async def _run_loop(self) -> None:
|
|
while self.running:
|
|
try:
|
|
rest_refresh_seconds = self._rest_refresh_seconds()
|
|
if self._needs_rest_refresh(rest_refresh_seconds):
|
|
await asyncio.to_thread(self.market.refresh_rest)
|
|
self.broker.update_highs(self.market.tickers)
|
|
self._update_patterns()
|
|
self._update_forecasts()
|
|
self.learner.refresh()
|
|
await self._process_exits()
|
|
await self._process_entries()
|
|
self.broker.mark_equity(self.market.prices())
|
|
self.last_loop_at = utc_now()
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as exc:
|
|
self.message = f"ошибка цикла: {exc}"
|
|
self.storage.event(self.message, "ERROR")
|
|
await asyncio.sleep(self.settings.effective_loop_interval_seconds)
|
|
|
|
def _needs_rest_refresh(self, rest_refresh_seconds: float) -> bool:
|
|
if self.market.last_rest_refresh_at is None:
|
|
return True
|
|
age = (utc_now() - self.market.last_rest_refresh_at).total_seconds()
|
|
return age >= rest_refresh_seconds
|
|
|
|
def _rest_refresh_seconds(self) -> float:
|
|
if self.settings.websocket_enabled:
|
|
return 10.0 if self.settings.fast_trading_enabled else 20.0
|
|
return self.settings.effective_loop_interval_seconds
|
|
|
|
async def _process_exits(self) -> None:
|
|
prices = self.market.prices()
|
|
reduction_candidate_id = self._reduction_candidate_id(prices)
|
|
for position in list(self.broker.open_positions()):
|
|
ticker = self.market.tickers.get(position.symbol)
|
|
candles = self.market.candles.get(position.symbol, [])
|
|
forecast = self.market.forecasts.get(position.symbol, {})
|
|
adaptive_rules = self._with_exposure_context(self.learner.rules_for(position.symbol, position.entry_pattern))
|
|
adaptive_rules["reduce_now"] = position.id is not None and position.id == reduction_candidate_id
|
|
learning = {"adaptive_rules": adaptive_rules}
|
|
signal = self.strategy.exit_signal(position, candles, ticker, learning, forecast)
|
|
self.storage.insert_signal(signal)
|
|
if signal.action == "SELL" and ticker is not None:
|
|
self.broker.sell(position, ticker, signal.reason)
|
|
self._entry_cooldown_until[position.symbol] = utc_now()
|
|
|
|
async def _process_entries(self) -> None:
|
|
prices = self.market.prices()
|
|
for symbol in self.market.symbols:
|
|
cooldown_since = self._entry_cooldown_until.get(symbol)
|
|
if cooldown_since:
|
|
age = (utc_now() - cooldown_since).total_seconds()
|
|
cooldown_seconds = self.settings.effective_entry_cooldown_seconds
|
|
if age < cooldown_seconds:
|
|
self.storage.insert_signal(
|
|
Signal(
|
|
symbol,
|
|
"HOLD",
|
|
0.0,
|
|
"пауза после закрытия позиции",
|
|
{"cooldown_remaining_seconds": cooldown_seconds - age},
|
|
)
|
|
)
|
|
continue
|
|
self._entry_cooldown_until.pop(symbol, None)
|
|
ticker = self.market.tickers.get(symbol)
|
|
candles = self.market.candles.get(symbol, [])
|
|
trend_candles = self.market.trend_candles.get(symbol, [])
|
|
open_count = len(self.broker.positions_for_symbol(symbol))
|
|
pattern = self.market.patterns.get(symbol, {})
|
|
forecast = self.market.forecasts.get(symbol, {})
|
|
learning = self.learner.adjustment_for(symbol, str(pattern.get("label", ""))).as_dict()
|
|
learning["adaptive_rules"] = self._with_exposure_context(learning.get("adaptive_rules") or {})
|
|
account = self.broker.account_state(prices)
|
|
llm = {}
|
|
if (
|
|
self.settings.llm_advisor_enabled
|
|
and self.llm_advisor is not None
|
|
and ticker is not None
|
|
and len(candles) >= 200
|
|
):
|
|
llm = (
|
|
await asyncio.to_thread(
|
|
self.llm_advisor.advice_for,
|
|
symbol=symbol,
|
|
candles=candles,
|
|
ticker=ticker,
|
|
pattern=pattern,
|
|
learning=learning,
|
|
open_positions_for_symbol=open_count,
|
|
account=account,
|
|
)
|
|
).as_dict()
|
|
signal = self.strategy.entry_signal(
|
|
symbol,
|
|
candles,
|
|
ticker,
|
|
open_count,
|
|
pattern,
|
|
learning,
|
|
llm,
|
|
forecast,
|
|
account,
|
|
trend_candles,
|
|
)
|
|
self.storage.insert_signal(signal)
|
|
if signal.action == "BUY" and ticker is not None:
|
|
self.broker.buy(
|
|
signal,
|
|
ticker,
|
|
self.market.instruments.get(symbol),
|
|
prices,
|
|
)
|
|
|
|
def _with_exposure_context(self, rules: dict) -> dict:
|
|
enriched = dict(rules)
|
|
current_exposure = self.broker.exposure()
|
|
target_total = float(enriched.get("target_total_exposure_usdt", self.settings.max_total_exposure_usdt) or 0.0)
|
|
target_total = max(0.0, min(self.settings.max_total_exposure_usdt, target_total))
|
|
enriched["current_total_exposure_usdt"] = round(current_exposure, 6)
|
|
enriched["target_total_exposure_usdt"] = round(target_total, 6)
|
|
enriched["over_target_exposure"] = current_exposure > target_total + self.settings.min_position_usdt
|
|
return enriched
|
|
|
|
def _close_paper_positions_outside_symbol_universe(self) -> None:
|
|
if self.settings.strategy_mode != "trend_macd" or self.settings.trading_mode != "paper":
|
|
return
|
|
allowed_symbols = set(self.market.symbols or self.settings.symbols)
|
|
for position in list(self.broker.open_positions()):
|
|
if position.symbol in allowed_symbols:
|
|
continue
|
|
synthetic_ticker = Ticker(
|
|
symbol=position.symbol,
|
|
last_price=position.entry_price,
|
|
bid=position.entry_price,
|
|
ask=position.entry_price,
|
|
turnover_24h=0.0,
|
|
volume_24h=0.0,
|
|
change_24h=0.0,
|
|
)
|
|
self.broker.sell(
|
|
position,
|
|
synthetic_ticker,
|
|
"trend_macd: закрыта старая paper-позиция вне списка разрешенных пар",
|
|
)
|
|
self.storage.event(
|
|
f"{position.symbol}: старая paper-позиция закрыта при переходе на trend_macd"
|
|
)
|
|
|
|
def _reduction_candidate_id(self, prices: dict[str, float]) -> int | None:
|
|
rules = self._with_exposure_context(self.learner.state.adaptive_rules or {})
|
|
if not rules.get("reduce_exposure") or not rules.get("over_target_exposure"):
|
|
return None
|
|
positions = self.broker.open_positions()
|
|
if not positions:
|
|
return None
|
|
|
|
def loss_ratio(position) -> float:
|
|
mark = prices.get(position.symbol, position.entry_price)
|
|
if position.notional_usdt <= 0:
|
|
return 0.0
|
|
return position.unrealized_pnl(mark) / position.notional_usdt
|
|
|
|
worst = min(positions, key=loss_ratio)
|
|
return worst.id
|
|
|
|
def _update_patterns(self) -> None:
|
|
if self.settings.strategy_mode == "trend_macd" or not self.settings.pattern_analysis_enabled:
|
|
self.market.patterns = {}
|
|
return
|
|
patterns: dict[str, dict] = {}
|
|
for symbol in self.market.symbols:
|
|
result = self.pattern_analyzer.analyze(
|
|
self.market.candles.get(symbol, []),
|
|
self.market.tickers.get(symbol),
|
|
)
|
|
patterns[symbol] = result.as_dict()
|
|
self.market.patterns = patterns
|
|
|
|
def _update_forecasts(self) -> None:
|
|
if (
|
|
self.settings.strategy_mode == "trend_macd"
|
|
or self.forecaster is None
|
|
or not self.settings.time_series_forecast_enabled
|
|
):
|
|
self.market.forecasts = {}
|
|
return
|
|
forecasts: dict[str, dict] = {}
|
|
for symbol in self.market.symbols:
|
|
forecasts[symbol] = self.forecaster.forecast(
|
|
self.market.candles.get(symbol, []),
|
|
symbol=symbol,
|
|
).as_dict()
|
|
self.market.forecasts = forecasts
|
|
|
|
def status(self) -> BotStatus:
|
|
return BotStatus(
|
|
running=self.running,
|
|
mode=self.settings.trading_mode,
|
|
live_trading_ready=self.settings.live_ready,
|
|
symbols=self.market.symbols,
|
|
started_at=self.started_at,
|
|
last_loop_at=self.last_loop_at,
|
|
message=self.message,
|
|
)
|
|
|
|
def account_snapshot(self) -> dict[str, float]:
|
|
prices = self.market.prices()
|
|
state = self.broker.account_state(prices)
|
|
state["starting_balance"] = self.settings.starting_balance_usdt
|
|
state["net_pnl"] = state["equity"] - self.settings.starting_balance_usdt
|
|
state["net_pnl_percent"] = (
|
|
(state["net_pnl"] / self.settings.starting_balance_usdt) * 100
|
|
if self.settings.starting_balance_usdt
|
|
else 0.0
|
|
)
|
|
return state
|
|
|
|
def positions_snapshot(self) -> list[dict]:
|
|
prices = self.market.prices()
|
|
return [
|
|
position.as_dict(mark_price=prices.get(position.symbol, position.entry_price))
|
|
for position in self.broker.open_positions()
|
|
]
|
|
|
|
def learning_snapshot(self) -> dict:
|
|
snapshot = self.learner.state.as_dict()
|
|
snapshot["adaptive_rules"] = self._with_exposure_context(snapshot.get("adaptive_rules") or {})
|
|
return snapshot
|
|
|
|
def llm_snapshot(self) -> dict:
|
|
if self.llm_advisor is None:
|
|
return {"enabled": False, "items": []}
|
|
return self.llm_advisor.snapshot()
|