from datetime import UTC, datetime
from decimal import Decimal
from typing import Any, Literal
from pydantic import BaseModel, Field
class DateTimeResponse(BaseModel):
"""Response model for get_current_datetime tool"""
timestamp: str = Field(
description="ISO8601 timestamp of now",
default_factory=lambda: datetime.now(tz=UTC).isoformat(),
)
class BinanceServerTimeResponse(BaseModel):
"""Response model for binance_server_time tool"""
server_time: int = Field(
description="Unix timestamp in milliseconds representing the Binance server time", alias="serverTime"
)
class BinanceConnectivityResponse(BaseModel):
"""Response model for binance_test_connectivity tool"""
timestamp: str = Field(
description="ISO8601 timestamp of the connectivity check",
default_factory=lambda: datetime.now(tz=UTC).isoformat(),
)
reachable: bool = Field(description="Binance API connectivity status")
type ContractType = Literal["PERPETUAL", "CURRENT_MONTH", "NEXT_MONTH", "CURRENT_QUARTER", "NEXT_QUARTER"]
type ContractStatus = Literal[
"PENDING_TRADING",
"TRADING",
"PRE_DELIVERING",
"DELIVERING",
"DELIVERED",
"CANCELLED",
"PRE_SETTLE",
"SETTLING",
"CLOSE",
]
type UnderlyingType = Literal["COIN", "INDEX"]
type OrderType = Literal[
"LIMIT",
"LIMIT_MAKER",
"MARKET",
"STOP_LOSS",
"STOP_LOSS_LIMIT",
"TAKE_PROFIT",
"TAKE_PROFIT_LIMIT",
]
type OrderTimeInForce = Literal["GTC", "IOC", "FOK", "GTX", "GTE_GTC", "GTD"]
class SymbolPriceFilter(BaseModel):
rate_limit_type: Literal["REQUEST_WEIGHT", "ORDERS", "RAW_REQUESTS"] = Field(alias="rateLimitType")
interval: Literal["SECOND", "MINUTE", "DAY"]
interval_num: int = Field(alias="intervalNum")
limit: int
class SymbolLotSizeFilter(BaseModel):
filter_type: Literal["LOT_SIZE"] = Field(alias="filterType")
min_qty: Decimal = Field(alias="minQty")
max_qty: Decimal = Field(alias="maxQty")
step_size: Decimal = Field(alias="stepSize")
class FuturesSymbolPercentPriceFilter(BaseModel):
filter_type: Literal["PERCENT_PRICE"] = Field(alias="filterType")
multiplier_up: Decimal = Field(alias="multiplierUp")
multiplier_down: Decimal = Field(alias="multiplierDown")
multiplier_decimal: int = Field(alias="multiplierDecimal")
class FuturesSymbolMinNotionalFilter(BaseModel):
filter_type: Literal["MIN_NOTIONAL"] = Field(alias="filterType")
notional: Decimal
class SymbolIcebergPartsFilter(BaseModel):
filter_type: Literal["ICEBERG_PARTS"] = Field(alias="filterType")
limit: int
class SymbolMarketLotSizeFilter(BaseModel):
filter_type: Literal["MARKET_LOT_SIZE"] = Field(alias="filterType")
min_qty: Decimal = Field(alias="minQty")
max_qty: Decimal = Field(alias="maxQty")
step_size: Decimal = Field(alias="stepSize")
class FuturesSymbolMaxOrdersFilter(BaseModel):
filter_type: Literal["MAX_NUM_ORDERS"] = Field(alias="filterType")
limit: int
class FuturesSymbolMaxAlgoOrdersFilter(BaseModel):
filter_type: Literal["MAX_NUM_ALGO_ORDERS"] = Field(alias="filterType")
limit: int
class SymbolMaxIcebergOrdersFilter(BaseModel):
filter_type: Literal["MAX_NUM_ICEBERG_ORDERS"] = Field(alias="filterType")
max_num_iceberg_orders: int = Field(alias="maxNumIcebergOrders")
class SymbolMaxPositionFilter(BaseModel):
filter_type: Literal["MAX_POSITION"] = Field(alias="filterType")
max_position: Decimal
type FutureSymbolFilter = (
SymbolPriceFilter
| FuturesSymbolPercentPriceFilter
| SymbolLotSizeFilter
| FuturesSymbolMinNotionalFilter
| SymbolIcebergPartsFilter
| SymbolMarketLotSizeFilter
| FuturesSymbolMaxOrdersFilter
| FuturesSymbolMaxAlgoOrdersFilter
| SymbolMaxIcebergOrdersFilter
| SymbolMaxPositionFilter
)
class FuturesSymbolExchangeInfo(BaseModel):
symbol: str
pair: str
contract_type: str = Field(alias="contractType")
delivery_date: int = Field(alias="deliveryDate")
onboard_date: int = Field(alias="onboardDate")
status: ContractStatus
maint_margin_percent: Decimal = Field(alias="maintMarginPercent")
required_margin_percent: Decimal = Field(alias="requiredMarginPercent")
base_asset: str = Field(alias="baseAsset")
quote_asset: str = Field(alias="quoteAsset")
margin_asset: str = Field(alias="marginAsset")
price_precision: int = Field(alias="pricePrecision")
quantity_precision: int = Field(alias="quantityPrecision")
base_asset_precision: int = Field(alias="baseAssetPrecision")
quote_precision: int = Field(alias="quotePrecision")
underlying_type: UnderlyingType = Field(alias="underlyingType")
settle_plan: int | None = Field(None, alias="settlePlan")
trigger_protect: Decimal = Field(alias="triggerProtect")
filters: list[dict[str, Any]]
order_type: list[OrderType] | None = Field(None, alias="OrderType")
time_in_force: list[OrderTimeInForce] = Field(alias="timeInForce")
liquidation_fee: Decimal = Field(alias="liquidationFee")
market_take_bound: Decimal = Field(alias="marketTakeBound")
contract_size: Decimal | None = Field(None, alias="contractSize")
class ExchangeMaxNumOrdersFilter(BaseModel):
filter_type: Literal["EXCHANGE_MAX_NUM_ORDERS"] = Field(alias="filterType")
max_num_orders: int = Field(alias="maxNumOrders")
class ExchangeMaxAlgoOrdersFilter(BaseModel):
filter_type: Literal["EXCHANGE_MAX_ALGO_ORDERS"] = Field(alias="filterType")
maxNumAlgoOrders: int = Field(alias="maxNumAlgoOrders")
type ExchangeFilter = ExchangeMaxNumOrdersFilter | ExchangeMaxAlgoOrdersFilter
class RateLimiter(BaseModel):
rate_limit_type: Literal["REQUEST_WEIGHT", "ORDERS", "RAW_REQUESTS"] = Field(alias="rateLimitType")
interval: Literal["SECOND", "MINUTE", "DAY"]
interval_num: int = Field(alias="intervalNum")
limit: int
class FuturesExchangeInfo(BaseModel):
exchange_filters: list[ExchangeFilter] = Field(alias="exchangeFilters")
rate_limits: list[RateLimiter] = Field(alias="rateLimits")
server_time: int = Field(alias="serverTime")
assets: list[Any]
symbols: list[FuturesSymbolExchangeInfo]
timezone: str
class HistoricOpenInterest(BaseModel):
symbol: str
sum_open_interest: Decimal = Field(alias="sumOpenInterest")
sum_open_interest_value: Decimal = Field(alias="sumOpenInterestValue")
timestamp: int
type Kline = list[
int,
int,
Decimal,
Decimal,
Decimal,
Decimal,
Decimal,
int,
Decimal,
int,
Decimal,
Decimal,
]
class Candle(BaseModel):
open_time: int
open: Decimal
high: Decimal
low: Decimal
close: Decimal
volume: Decimal
close_time: int
quote_asset_volume: Decimal
number_of_trades: int
taker_buy_base_asset_volume: Decimal
taker_buy_quote_asset_volume: Decimal
volume_ma: Decimal | None = None
taker_buy_volume_ma: Decimal | None = None
@classmethod
def from_kline(cls, kline: Kline) -> "Candle":
return cls(
open_time=kline[0],
open=kline[1],
high=kline[2],
low=kline[3],
close=kline[4],
volume=kline[5],
close_time=kline[6],
quote_asset_volume=kline[7],
number_of_trades=kline[8],
taker_buy_base_asset_volume=kline[9],
taker_buy_quote_asset_volume=kline[10],
)
@property
def buy_sell_ratio(self) -> Decimal:
if self.quote_asset_volume == 0:
return Decimal(0)
if self.taker_buy_quote_asset_volume == 0:
return Decimal(0)
return self.taker_buy_quote_asset_volume / self.quote_asset_volume
@property
def hl2(self) -> Decimal:
return (self.high + self.low) / Decimal(2)
@property
def is_green(self) -> bool:
return self.close > self.open
@property
def is_red(self) -> bool:
return self.close < self.open
@property
def volume_delta(self) -> Decimal:
return self.quote_asset_volume - self.taker_buy_quote_asset_volume
class LongShortRatio(BaseModel):
symbol: str
long_short_ratio: Decimal = Field(alias="longShortRatio")
long_account: Decimal = Field(alias="longAccount")
short_account: Decimal = Field(alias="shortAccount")
timestamp: int
class AggregatedTrade(BaseModel):
a: int
p: Decimal
q: Decimal
f: int
l: int # noqa: E741
T: int
m: bool
@property
def id(self) -> int:
return self.a
@property
def price(self) -> Decimal:
return self.p
@property
def quantity(self) -> Decimal:
return self.q
@property
def first_trade_id(self) -> int:
return self.f
@property
def last_trade_id(self) -> int:
return self.l
@property
def timestamp(self) -> int:
return self.T
@property
def maker(self) -> bool:
return self.m
type ValidPeriods = Literal["1m", "5m", "15m", "30m", "1h", "2h", "4h", "6h", "12h", "1d"]
class Candidate(BaseModel):
symbol: str
direction: Literal["long", "short"]
momentum: Literal["bullish", "bearish"] | None = None
momentum_strength: Decimal | None = None
oi_zscore: Decimal | None = None
ls_zscore: Decimal | None = None
oi_ls_weight: Decimal | None = None
oi_ls_weight_zscored: Decimal | None = None
trend_strength: float | None = None
volatility_factor: float | None = None
composite_z: float | None = None
funding_rate: float | None = None
class BookDepth(BaseModel):
e: Literal["depthUpdate"]
E: int
T: int
s: str
U: int
u: int
pu: int
b: list[list[float]]
a: list[list[float]]
class FundingRate(BaseModel):
symbol: str
funding_rate: Decimal = Field(alias="fundingRate")
funding_time: int = Field(alias="fundingTime")
mark_price: Decimal = Field(alias="markPrice")
class OrderBook(BaseModel):
last_update_id: int = Field(alias="lastUpdateId")
E: int
T: int
bids: list[list[Decimal]]
asks: list[list[Decimal]]
class Ticker(BaseModel):
symbol: str
price_change: Decimal = Field(alias="priceChange")
price_change_percent: Decimal = Field(alias="priceChangePercent")
weighted_avg_price: Decimal = Field(alias="weightedAvgPrice")
last_price: Decimal = Field(alias="lastPrice")
last_qty: Decimal = Field(alias="lastQty")
open_price: Decimal = Field(alias="openPrice")
high_price: Decimal = Field(alias="highPrice")
low_price: Decimal = Field(alias="lowPrice")
volume: Decimal
quote_volume: Decimal = Field(alias="quoteVolume")
open_time: int = Field(alias="openTime")
close_time: int = Field(alias="closeTime")
first_id: int = Field(alias="firstId")
last_id: int = Field(alias="lastId")
count: int