| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657 |
- from __future__ import annotations
- from dataclasses import dataclass, field
- from typing import Any
- from .exec_client import list_open_orders, cancel_all_orders, cancel_order, place_order, get_account_info
- from .news_client import call_news_tool
- from .crypto_client import call_crypto_tool
- @dataclass(frozen=True)
- class StrategyContext:
- id: str
- account_id: str
- client_id: str | None = field(default=None, repr=False)
- mode: str = "off"
- market_symbol: str | None = None
- base_currency: str | None = None
- counter_currency: str | None = None
- minimum_order_value: float | None = None
- def __getattr__(self, name: str):
- if name == "mode":
- return "active"
- raise AttributeError(name)
- def get_open_orders(self) -> Any:
- payload = list_open_orders(self.account_id, self.client_id)
- if isinstance(payload, dict) and isinstance(payload.get("orders"), list):
- return payload["orders"]
- return payload
- def cancel_all_orders(self) -> Any:
- return cancel_all_orders(self.account_id, self.client_id)
- def cancel_order(self, order_id: str) -> Any:
- return cancel_order(self.account_id, order_id)
- def place_order(self, **kwargs: Any) -> Any:
- mode = getattr(self, "mode", "active")
- if mode != "active":
- raise RuntimeError(f"place_order not allowed in {mode} mode")
- kwargs.setdefault("account_id", self.account_id)
- kwargs.setdefault("client_id", self.client_id)
- return place_order(kwargs)
- def get_price(self, symbol: str) -> Any:
- return call_crypto_tool("get_price", {"symbol": symbol})
- def get_regime(self, symbol: str, timeframe: str = "1h") -> Any:
- return call_crypto_tool("get_regime", {"symbol": symbol, "timeframe": timeframe})
- def get_account_info(self) -> Any:
- return get_account_info(self.account_id)
- def get_news(self, **kwargs: Any) -> Any:
- return call_news_tool("search", kwargs)
|
