from __future__ import annotations import hashlib import hmac from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from typing import Any from urllib.parse import urlencode import requests BASE_URLS = ( "https://api.binance.com", "https://api1.binance.com", "https://api2.binance.com", "https://api3.binance.com", "https://api4.binance.com", ) class BinanceReadOnlyError(RuntimeError): """A sanitized error that never contains API credentials or raw responses.""" def normalize_proxy(value: object) -> str | None: """Validate an optional HTTP(S) proxy URL; None means "no proxy".""" if value is None: return None text = str(value).strip() if not text: return None lowered = text.lower() if not (lowered.startswith("http://") or lowered.startswith("https://")): raise BinanceReadOnlyError("Proxy must start with http:// or https://.") return text def as_decimal(value: object) -> Decimal: try: return Decimal(str(value)) except (InvalidOperation, TypeError, ValueError): return Decimal("0") def decimal_text(value: Decimal, places: int = 8) -> str: text = format(value, f".{places}f").rstrip("0").rstrip(".") return text or "0" class BinanceReadOnlyClient: """Strictly read-only Binance Spot client.""" def __init__( self, api_key: str, api_secret: str, timeout: int = 15, proxy: str | None = None, ) -> None: if not api_key or not api_secret: raise BinanceReadOnlyError("An API key or secret is missing from .env.") self._api_key = api_key self._api_secret = api_secret self._timeout = timeout self._proxy = normalize_proxy(proxy) self._session = requests.Session() if self._proxy: self._session.proxies.update({"http": self._proxy, "https": self._proxy}) self._base_url: str | None = None def _request_json( self, url: str, *, headers: dict[str, str] | None = None, parameters: dict[str, object] | None = None, ) -> Any: try: response = self._session.get( url, headers=headers, params=parameters, timeout=self._timeout, ) except requests.RequestException as exc: raise BinanceReadOnlyError( f"Binance connection failed ({type(exc).__name__})." ) from None try: data = response.json() except ValueError: raise BinanceReadOnlyError( f"Binance returned unreadable data (HTTP {response.status_code})." ) from None if response.ok: return data error_code = data.get("code") if isinstance(data, dict) else None if response.status_code == 451: raise BinanceReadOnlyError( "Binance refused this network location (HTTP 451). Check the active VPN server." ) if error_code == -2015: raise BinanceReadOnlyError( "Binance rejected the API key. Check read permission and allowed IP addresses." ) if error_code == -1021: raise BinanceReadOnlyError( "Windows time is not synchronized. Enable automatic time and retry." ) raise BinanceReadOnlyError( f"Binance request failed (HTTP {response.status_code}, code {error_code})." ) def _select_base_url(self) -> tuple[str, int]: if self._base_url is not None: try: data = self._request_json(f"{self._base_url}/api/v3/time") if isinstance(data, dict) and "serverTime" in data: return self._base_url, int(data["serverTime"]) except BinanceReadOnlyError: self._base_url = None failures: list[str] = [] for candidate in BASE_URLS: try: data = self._request_json(f"{candidate}/api/v3/time") if isinstance(data, dict) and "serverTime" in data: self._base_url = candidate return candidate, int(data["serverTime"]) failures.append("Invalid response") except BinanceReadOnlyError as exc: failures.append(str(exc)) detail = failures[-1] if failures else "Unknown network error." raise BinanceReadOnlyError(f"Binance could not be reached. {detail}") def _public_get( self, path: str, parameters: dict[str, object] | None = None, ) -> Any: base_url, _ = self._select_base_url() return self._request_json(f"{base_url}{path}", parameters=parameters) def _signed_get( self, path: str, parameters: dict[str, object] | None = None, ) -> Any: base_url, server_time = self._select_base_url() payload = dict(parameters or {}) payload["timestamp"] = server_time payload["recvWindow"] = 5000 query = urlencode(payload) payload["signature"] = hmac.new( self._api_secret.encode("utf-8"), query.encode("utf-8"), hashlib.sha256, ).hexdigest() return self._request_json( f"{base_url}{path}", headers={"X-MBX-APIKEY": self._api_key}, parameters=payload, ) def permission_status(self) -> dict[str, bool | None]: data = self._signed_get("/sapi/v1/account/apiRestrictions") if not isinstance(data, dict): raise BinanceReadOnlyError("Permission check returned an unexpected response.") return { "reading": data.get("enableReading"), "trading": data.get("enableSpotAndMarginTrading"), "withdrawals": data.get("enableWithdrawals"), "ip_restricted": data.get("ipRestrict"), "proxy_active": self._proxy is not None, } @staticmethod def assert_read_only(permissions: dict[str, bool | None]) -> None: if permissions.get("reading") is not True: raise BinanceReadOnlyError("SAFETY STOP: Read permission is not enabled.") if permissions.get("trading") is not False: raise BinanceReadOnlyError( "SAFETY STOP: Spot/Margin trading permission must be disabled." ) if permissions.get("withdrawals") is not False: raise BinanceReadOnlyError( "SAFETY STOP: Withdrawal permission must be disabled." ) def account_balances(self) -> list[dict[str, Any]]: data = self._signed_get( "/api/v3/account", {"omitZeroBalances": "true"}, ) if not isinstance(data, dict) or not isinstance(data.get("balances"), list): raise BinanceReadOnlyError("Account read returned an unexpected response.") return [item for item in data["balances"] if isinstance(item, dict)] def open_orders(self) -> list[dict[str, Any]]: data = self._signed_get("/api/v3/openOrders") if not isinstance(data, list): raise BinanceReadOnlyError("Open orders read returned an unexpected response.") return [item for item in data if isinstance(item, dict)] def recent_trades(self, symbol: str, limit: int = 1000) -> list[dict[str, Any]]: clean_symbol = symbol.strip().upper() if not clean_symbol or not clean_symbol.isalnum() or not clean_symbol.endswith("USDT"): raise BinanceReadOnlyError("Invalid USDT Spot symbol.") data = self._signed_get( "/api/v3/myTrades", {"symbol": clean_symbol, "limit": max(1, min(int(limit), 1000))}, ) if not isinstance(data, list): raise BinanceReadOnlyError("Trade history read returned an unexpected response.") return [item for item in data if isinstance(item, dict)] def ticker_price(self, symbol: str) -> Decimal: clean_symbol = symbol.strip().upper() if not clean_symbol or not clean_symbol.isalnum() or not clean_symbol.endswith("USDT"): raise BinanceReadOnlyError("Invalid USDT Spot symbol.") data = self._public_get("/api/v3/ticker/price", {"symbol": clean_symbol}) if not isinstance(data, dict): raise BinanceReadOnlyError("Price read returned an unexpected response.") price = as_decimal(data.get("price")) if price <= 0: raise BinanceReadOnlyError("A valid current price was not returned.") return price @staticmethod def trade_cycle_summary( symbol: str, trades: list[dict[str, Any]], *, current_price: Decimal, current_quantity: Decimal, ) -> dict[str, Any]: """Summarize actual fills around the most recent sell; never creates an order.""" clean_symbol = symbol.strip().upper() base_asset = clean_symbol[:-4] valid: list[dict[str, Any]] = [] for item in trades: quantity = as_decimal(item.get("qty")) price = as_decimal(item.get("price")) timestamp = int(item.get("time", 0) or 0) if quantity <= 0 or price <= 0 or timestamp <= 0: continue valid.append( { **item, "qty_decimal": quantity, "quote_decimal": as_decimal(item.get("quoteQty")) or quantity * price, "price_decimal": price, "time_integer": timestamp, "buyer_boolean": bool(item.get("isBuyer")), } ) def aggregate(fills: list[dict[str, Any]]) -> dict[str, Any] | None: if not fills: return None quantity = sum((item["qty_decimal"] for item in fills), Decimal("0")) quote = sum((item["quote_decimal"] for item in fills), Decimal("0")) if quantity <= 0: return None timestamp = max(int(item["time_integer"]) for item in fills) return { "average_price": decimal_text(quote / quantity), "quantity": decimal_text(quantity), "quote_total": format(quote, ".2f"), "executed_at": datetime.fromtimestamp( timestamp / 1000, tz=timezone.utc ).isoformat(), "order_id": str(fills[-1].get("orderId", "")), } orders: dict[tuple[bool, str], list[dict[str, Any]]] = {} for item in valid: order_id = str(item.get("orderId", item.get("id", ""))) orders.setdefault((bool(item["buyer_boolean"]), order_id), []).append(item) ordered_groups = sorted( orders.values(), key=lambda fills: max(int(item["time_integer"]) for item in fills), ) sell_groups = [fills for fills in ordered_groups if not fills[0]["buyer_boolean"]] buy_groups = [fills for fills in ordered_groups if fills[0]["buyer_boolean"]] latest_sell_fills = sell_groups[-1] if sell_groups else [] latest_buy_fills = buy_groups[-1] if buy_groups else [] latest_sell = aggregate(latest_sell_fills) latest_buy = aggregate(latest_buy_fills) last_sell_time = ( max(int(item["time_integer"]) for item in latest_sell_fills) if latest_sell_fills else None ) cycle_buys = [ item for item in valid if item["buyer_boolean"] and last_sell_time is not None and int(item["time_integer"]) > last_sell_time ] cycle_quantity = sum( (item["qty_decimal"] for item in cycle_buys), Decimal("0") ) cycle_quote = sum( (item["quote_decimal"] for item in cycle_buys), Decimal("0") ) average_buy = cycle_quote / cycle_quantity if cycle_quantity > 0 else Decimal("0") current_value = current_quantity * current_price unrealized = ( current_quantity * (current_price - average_buy) if average_buy > 0 else None ) unrealized_percent = ( ((current_price - average_buy) / average_buy) * Decimal("100") if average_buy > 0 else None ) last_sell_price = ( as_decimal(latest_sell.get("average_price")) if latest_sell else Decimal("0") ) current_vs_sell = ( ((current_price - last_sell_price) / last_sell_price) * Decimal("100") if last_sell_price > 0 else None ) new_buy_vs_sell = ( ((average_buy - last_sell_price) / last_sell_price) * Decimal("100") if average_buy > 0 and last_sell_price > 0 else None ) other_fee_assets = sorted( { str(item.get("commissionAsset", "")) for item in valid if as_decimal(item.get("commission")) > 0 and str(item.get("commissionAsset", "")) not in {"", base_asset, "USDT"} } ) tracked_quantity = Decimal("0") tracked_cost = Decimal("0") for item in sorted(valid, key=lambda trade: int(trade["time_integer"])): quantity = item["qty_decimal"] quote = item["quote_decimal"] commission = as_decimal(item.get("commission")) commission_asset = str(item.get("commissionAsset", "")).strip().upper() if item["buyer_boolean"]: acquired = quantity - commission if commission_asset == base_asset else quantity spent = quote + commission if commission_asset == "USDT" else quote if acquired > 0: tracked_quantity += acquired tracked_cost += spent continue disposed = quantity + commission if commission_asset == base_asset else quantity if tracked_quantity <= 0: tracked_cost = Decimal("0") continue average_cost_before_sell = tracked_cost / tracked_quantity removed = min(disposed, tracked_quantity) tracked_quantity -= removed tracked_cost -= average_cost_before_sell * removed if tracked_quantity <= 0: tracked_quantity = Decimal("0") tracked_cost = Decimal("0") current_average_cost = ( tracked_cost / tracked_quantity if tracked_quantity > 0 else None ) cost_tolerance = max(Decimal("0.00000001"), current_quantity * Decimal("0.000001")) cost_basis_complete = ( abs(tracked_quantity - current_quantity) <= cost_tolerance if current_quantity > 0 else tracked_quantity == 0 ) return { "symbol": clean_symbol, "latest_sell": latest_sell, "latest_buy": latest_buy, "new_cycle": { "started": bool(latest_sell and cycle_quantity > 0), "bought_quantity": decimal_text(cycle_quantity), "quote_spent": format(cycle_quote, ".2f"), "average_buy_price": decimal_text(average_buy) if average_buy > 0 else None, "current_quantity": decimal_text(current_quantity), "current_price": decimal_text(current_price), "current_value_usdt": format(current_value, ".2f"), "unrealized_pnl_usdt": format(unrealized, ".2f") if unrealized is not None else None, "unrealized_pnl_percent": ( format(unrealized_percent, ".2f") if unrealized_percent is not None else None ), "current_vs_last_sell_percent": ( format(current_vs_sell, ".2f") if current_vs_sell is not None else None ), "new_buy_vs_last_sell_percent": ( format(new_buy_vs_sell, ".2f") if new_buy_vs_sell is not None else None ), }, "position": { "current_quantity": decimal_text(current_quantity), "cost_basis_quantity": decimal_text(tracked_quantity), "average_cost": ( decimal_text(current_average_cost) if current_average_cost is not None else None ), "cost_basis_complete": cost_basis_complete, }, "other_fee_assets": other_fee_assets, "manual_execution_only": True, } def ticker_prices(self) -> dict[str, Decimal]: data = self._public_get("/api/v3/ticker/price") if not isinstance(data, list): raise BinanceReadOnlyError("Price read returned an unexpected response.") prices: dict[str, Decimal] = {} for item in data: if not isinstance(item, dict): continue symbol = item.get("symbol") price = as_decimal(item.get("price")) if isinstance(symbol, str) and price > 0: prices[symbol] = price return prices def ticker_24hr(self) -> list[dict[str, Any]]: data = self._public_get("/api/v3/ticker/24hr") if not isinstance(data, list): raise BinanceReadOnlyError("24-hour market scan returned an unexpected response.") return [item for item in data if isinstance(item, dict)] @staticmethod def discover_usdt_symbols( tickers: list[dict[str, Any]], *, limit: int = 5, min_quote_volume_usdt: Decimal = Decimal("5000000"), ) -> list[str]: """Select liquid Spot candidates without chasing extreme one-day moves.""" stable_bases = {"USDC", "FDUSD", "TUSD", "USDP", "DAI", "EUR", "TRY", "BRL"} leveraged_suffixes = ("UP", "DOWN", "BULL", "BEAR") eligible: list[tuple[str, Decimal, Decimal]] = [] for item in tickers: symbol = str(item.get("symbol", "")).strip().upper() if not symbol.endswith("USDT") or not symbol.isalnum(): continue base = symbol[:-4] if not base or base in stable_bases or base.endswith(leveraged_suffixes): continue volume = as_decimal(item.get("quoteVolume")) change = as_decimal(item.get("priceChangePercent")) if volume < min_quote_volume_usdt or abs(change) > Decimal("20"): continue eligible.append((symbol, volume, change)) positive = sorted( (item for item in eligible if Decimal("1") <= item[2] <= Decimal("12")), key=lambda item: (item[2], item[1]), reverse=True, ) liquid = sorted(eligible, key=lambda item: item[1], reverse=True) selected: list[str] = [] positive_slots = max(1, min(int(limit), 3)) for symbol, _, _ in positive[:positive_slots]: if symbol not in selected: selected.append(symbol) for symbol, _, _ in liquid: if symbol not in selected: selected.append(symbol) if len(selected) >= max(1, int(limit)): break return selected[: max(1, int(limit))] @staticmethod def market_breadth_context( tickers: list[dict[str, Any]], btc_timeframes: list[dict[str, Any]], *, min_quote_volume_usdt: Decimal, min_breadth_percent: Decimal, ) -> dict[str, Any]: stable_bases = {"USDC", "FDUSD", "TUSD", "USDP", "DAI", "EUR", "TRY", "BRL"} leveraged_suffixes = ("UP", "DOWN", "BULL", "BEAR") eligible_changes: list[Decimal] = [] for item in tickers: symbol = str(item.get("symbol", "")).strip().upper() if not symbol.endswith("USDT") or not symbol.isalnum(): continue base = symbol[:-4] if not base or base in stable_bases or base.endswith(leveraged_suffixes): continue if as_decimal(item.get("quoteVolume")) < min_quote_volume_usdt: continue eligible_changes.append(as_decimal(item.get("priceChangePercent"))) advancing = sum(1 for change in eligible_changes if change > 0) breadth_percent = ( (Decimal(advancing) / Decimal(len(eligible_changes))) * Decimal("100") if eligible_changes else Decimal("0") ) btc_by_interval = { str(item.get("interval")): str(item.get("trend")) for item in btc_timeframes } btc_favorable = ( btc_by_interval.get("1h") != "BEARISH" and btc_by_interval.get("4h") == "BULLISH" and btc_by_interval.get("1d") == "BULLISH" ) breadth_favorable = ( len(eligible_changes) >= 10 and breadth_percent >= min_breadth_percent ) return { "favorable": btc_favorable and breadth_favorable, "btc_favorable": btc_favorable, "btc_trends": btc_by_interval, "breadth_favorable": breadth_favorable, "breadth_percent": format(breadth_percent, ".2f"), "advancing_symbols": advancing, "eligible_symbols": len(eligible_changes), "minimum_breadth_percent": format(min_breadth_percent, ".2f"), } def klines(self, symbol: str, interval: str, limit: int = 250) -> list[list[Any]]: if interval not in {"15m", "1h", "4h", "1d"}: raise BinanceReadOnlyError(f"Unsupported analysis interval: {interval}.") clean_symbol = symbol.strip().upper() if not clean_symbol or not clean_symbol.isalnum(): raise BinanceReadOnlyError("Invalid analysis symbol.") safe_limit = max(210, min(int(limit), 500)) data = self._public_get( "/api/v3/klines", {"symbol": clean_symbol, "interval": interval, "limit": safe_limit}, ) if not isinstance(data, list): raise BinanceReadOnlyError( f"Kline read returned an unexpected response for {clean_symbol} {interval}." ) return [row for row in data if isinstance(row, list)] def available_timeframe_analyses( self, symbol: str, intervals: list[str] | tuple[str, ...], ) -> tuple[list[dict[str, Any]], list[str]]: """Analyze usable timeframes without failing a newly listed symbol.""" from .analysis import timeframe_analysis results: list[dict[str, Any]] = [] unavailable: list[str] = [] for interval in intervals: try: rows = self.klines(symbol, interval, 250) results.append(timeframe_analysis(interval, rows)) except ValueError as exc: if str(exc) != f"Not enough closed candles for {interval} analysis.": raise unavailable.append(interval) return results, unavailable @staticmethod def trade_round_trips( symbol: str, trades: list[dict[str, Any]], *, max_trips: int = 20, ) -> dict[str, Any]: """Pair actual fills into buy→sell round trips using weighted-average cost matching (the same cost basis the exchange's own PnL view uses). Completed trips expose the average buy price vs the sell price so the owner can compare a later re-entry against the previous exit. Any quantity bought but not yet sold is reported as an open position with its average buy price. This is pure read-only calculation. """ clean_symbol = symbol.strip().upper() base_asset = clean_symbol[:-4] valid: list[dict[str, Any]] = [] for item in trades: quantity = as_decimal(item.get("qty")) price = as_decimal(item.get("price")) timestamp = int(item.get("time", 0) or 0) if quantity <= 0 or price <= 0 or timestamp <= 0: continue valid.append( { "qty_decimal": quantity, "quote_decimal": as_decimal(item.get("quoteQty")) or quantity * price, "price_decimal": price, "time_integer": timestamp, "buyer_boolean": bool(item.get("isBuyer")), "commission_decimal": as_decimal(item.get("commission")), "commission_asset": str(item.get("commissionAsset", "")).strip().upper(), } ) lots: list[dict[str, Any]] = [] trips: list[dict[str, Any]] = [] total_acquired = sum( ( item["qty_decimal"] - ( item["commission_decimal"] if item["commission_asset"] == base_asset else Decimal("0") ) ) for item in valid if item["buyer_boolean"] ) total_disposed = sum( ( item["qty_decimal"] + ( item["commission_decimal"] if item["commission_asset"] == base_asset else Decimal("0") ) ) for item in valid if not item["buyer_boolean"] ) # When materially more was sold than bought, the account received an # external balance credit (token migration, redenomination, or # airdrop). In that case pairing across the migration boundary is # invalid: an old lot priced on the pre-migration scale would # fabricate huge fake losses (or gains) against post-migration fills. # The 0.5% margin absorbs rounding dust so tiny mismatches (e.g. fee # rounding) do not trigger the quarantine. external_credit = total_disposed > total_acquired * Decimal("1.005") legacy_lots: list[dict[str, Any]] = [] unmatched_sells: list[dict[str, Any]] = [] for item in sorted(valid, key=lambda trade: int(trade["time_integer"])): quantity = item["qty_decimal"] quote = item["quote_decimal"] commission = item["commission_decimal"] commission_asset = item["commission_asset"] if item["buyer_boolean"]: acquired = ( quantity - commission if commission_asset == base_asset else quantity ) cost = quote + commission if commission_asset == "USDT" else quote if acquired > 0 and cost > 0: lots.append( { "qty": acquired, "cost": cost, "time": int(item["time_integer"]), } ) continue disposed = ( quantity + commission if commission_asset == base_asset else quantity ) proceeds = ( quote - commission if commission_asset == "USDT" else quote ) if disposed <= 0 or proceeds <= 0: continue remaining = disposed sell_time = int(item["time_integer"]) sell_price = item["price_decimal"] if external_credit: # Quarantine pre-migration lots whose unit cost sits on a # wildly different price scale than this sell; matching them # would invent fake round trips. while lots: lot = lots[0] unit_cost = lot["cost"] / lot["qty"] if ( sell_price < unit_cost * Decimal("0.25") or sell_price > unit_cost * Decimal("4") ): legacy_lots.append(lots.pop(0)) else: break if len(lots) > 1: # Weighted-average cost: blend every open lot into one pool # before consuming, exactly like the exchange's own PnL view. # The owner compares his average cost against each sell, so a # single sell closes against the blended pool instead of # cherry-picking the oldest (and often dearest) lot first. blended_qty = sum((lot["qty"] for lot in lots), Decimal("0")) blended_cost = sum((lot["cost"] for lot in lots), Decimal("0")) lots[:] = [ { "qty": blended_qty, "cost": blended_cost, "time": min(int(lot["time"]) for lot in lots), } ] while remaining > 0 and lots: lot = lots[0] matched = min(remaining, lot["qty"]) cost_part = lot["cost"] * matched / lot["qty"] proceeds_part = proceeds * matched / disposed pnl = proceeds_part - cost_part trips.append( { "symbol": clean_symbol, "status": "CLOSED", "quantity": decimal_text(matched), "buy_avg_price": decimal_text(cost_part / matched), "sell_price": decimal_text(sell_price), "buy_time": datetime.fromtimestamp( lot["time"] / 1000, tz=timezone.utc ).isoformat(), "sell_time": datetime.fromtimestamp( sell_time / 1000, tz=timezone.utc ).isoformat(), "pnl_usdt": format(pnl, ".4f"), "return_percent": ( format((pnl / cost_part) * Decimal("100"), ".2f") if cost_part > 0 else "0.00" ), "_cost": cost_part, } ) lot["qty"] -= matched lot["cost"] -= cost_part if lot["qty"] <= Decimal("0.00000001"): lots.pop(0) remaining -= matched if remaining > 0: # Sold quantity with no matching buy: a free balance credit # (airdrop/token migration) converted to USDT. It is real # profit and must not vanish from the report. unmatched_sells.append( { "quantity": decimal_text(remaining), "sell_price": decimal_text(sell_price), "proceeds_usdt": format( proceeds * remaining / disposed, ".4f" ), "sell_time": datetime.fromtimestamp( sell_time / 1000, tz=timezone.utc ).isoformat(), } ) # Merge the legs of a single sell order (same price, fills within ten # seconds) into one row with a cost-weighted average buy price. A user # who sells one batch sees one trade, not one fabricated losing trip # per historical buy lot. grouped: list[dict[str, Any]] = [] for trip in trips: last = grouped[-1] if grouped else None same_order = False if last is not None: last_price = Decimal(str(last["sell_price"])) trip_price = Decimal(str(trip["sell_price"])) seconds_apart = abs( ( datetime.fromisoformat(str(trip["sell_time"])) - datetime.fromisoformat(str(last["sell_time"])) ).total_seconds() ) price_close = trip_price == last_price or ( abs(trip_price - last_price) / max(trip_price, last_price) <= Decimal("0.001") ) same_order = seconds_apart <= 10 and price_close if same_order: merged_qty = Decimal(last["quantity"]) + Decimal(trip["quantity"]) merged_cost = Decimal(last["_cost"]) + Decimal(trip["_cost"]) merged_pnl = Decimal(last["pnl_usdt"]) + Decimal(trip["pnl_usdt"]) last["quantity"] = decimal_text(merged_qty) last["_cost"] = merged_cost last["pnl_usdt"] = format(merged_pnl, ".4f") last["buy_avg_price"] = decimal_text(merged_cost / merged_qty) last["return_percent"] = ( format(merged_pnl / merged_cost * Decimal("100"), ".2f") if merged_cost > 0 else "0.00" ) if str(trip["buy_time"]) < str(last["buy_time"]): last["buy_time"] = trip["buy_time"] else: trip = dict(trip) grouped.append(trip) trips = grouped open_quantity = sum((lot["qty"] for lot in lots), Decimal("0")) open_cost = sum((lot["cost"] for lot in lots), Decimal("0")) trips.sort(key=lambda trip: str(trip["sell_time"]), reverse=True) kept_trips = trips[: max(1, int(max_trips))] for trip in kept_trips: trip.pop("_cost", None) fills = [ { "side": "BUY" if item["buyer_boolean"] else "SELL", "price": decimal_text(item["price_decimal"]), "quantity": decimal_text(item["qty_decimal"]), "quote_usdt": format(item["quote_decimal"], ".4f"), "time": datetime.fromtimestamp( int(item["time_integer"]) / 1000, tz=timezone.utc ).isoformat(), } for item in sorted(valid, key=lambda trade: int(trade["time_integer"]))[ :60 ] ] return { "symbol": clean_symbol, "trips": kept_trips, "fills": fills, "unmatched_sells": unmatched_sells, "external_credit": external_credit, "legacy_lots": [ { "quantity": decimal_text(lot["qty"]), "average_buy_price": decimal_text(lot["cost"] / lot["qty"]), "opened_at": datetime.fromtimestamp( int(lot["time"]) / 1000, tz=timezone.utc ).isoformat(), } for lot in legacy_lots ], "open_position": ( { "quantity": decimal_text(open_quantity), "average_buy_price": decimal_text(open_cost / open_quantity), "cost_usdt": format(open_cost, ".2f"), "opened_at": datetime.fromtimestamp( min(int(lot["time"]) for lot in lots) / 1000, tz=timezone.utc ).isoformat(), } if open_quantity > 0 else None ), } @staticmethod def usdt_rate(asset: str, prices: dict[str, Decimal]) -> Decimal | None: if asset == "USDT": return Decimal("1") direct = prices.get(f"{asset}USDT") if direct is not None: return direct reverse = prices.get(f"USDT{asset}") if reverse: return Decimal("1") / reverse for bridge in ("USDC", "FDUSD", "BTC", "BNB", "ETH"): if asset == bridge: continue asset_bridge = prices.get(f"{asset}{bridge}") bridge_usdt = prices.get(f"{bridge}USDT") if asset_bridge is not None and bridge_usdt is not None: return asset_bridge * bridge_usdt bridge_asset = prices.get(f"{bridge}{asset}") if bridge_asset and bridge_usdt is not None: return bridge_usdt / bridge_asset return None @staticmethod def order_distance_percent(order_price: Decimal, market_price: Decimal) -> Decimal | None: if order_price <= 0 or market_price <= 0: return None return ((order_price - market_price) / market_price) * Decimal("100") def dashboard( self, watchlist: list[str], *, analysis_symbols: list[str] | None = None, analysis_timeframes: list[str] | None = None, risk_percent: Decimal = Decimal("0.50"), max_allocation_percent: Decimal = Decimal("20"), auto_scan_enabled: bool = True, auto_scan_count: int = 5, min_quote_volume_usdt: Decimal = Decimal("5000000"), min_signal_score: Decimal = Decimal("0.65"), min_confidence_percent: Decimal = Decimal("79"), min_volume_ratio: Decimal = Decimal("1.20"), min_market_breadth_percent: Decimal = Decimal("50"), max_daily_move_percent: Decimal = Decimal("12"), max_stop_risk_percent: Decimal = Decimal("7"), min_reward_risk: Decimal = Decimal("2"), include_account_candidates: bool = True, ) -> dict[str, Any]: from .analysis import trade_plan, timeframe_analysis permissions = self.permission_status() self.assert_read_only(permissions) balances_raw = self.account_balances() orders_raw = self.open_orders() prices = self.ticker_prices() tickers_24hr = self.ticker_24hr() tickers_by_symbol = { str(item.get("symbol", "")).strip().upper(): item for item in tickers_24hr } discovered_symbols: list[str] = [] if auto_scan_enabled: discovered_symbols = self.discover_usdt_symbols( tickers_24hr, limit=max(1, min(int(auto_scan_count), 100)), min_quote_volume_usdt=min_quote_volume_usdt, ) balances: list[dict[str, Any]] = [] estimated_total = Decimal("0") free_usdt = Decimal("0") unpriced_assets: list[str] = [] for item in balances_raw: asset = item.get("asset") if not isinstance(asset, str): continue free = as_decimal(item.get("free")) locked = as_decimal(item.get("locked")) total = free + locked if total <= 0: continue rate = self.usdt_rate(asset, prices) estimated_value = total * rate if rate is not None else None if estimated_value is None: unpriced_assets.append(asset) else: estimated_total += estimated_value if asset == "USDT": free_usdt = free balances.append( { "asset": asset, "free": decimal_text(free), "locked": decimal_text(locked), "total": decimal_text(total), "estimated_usdt": ( format(estimated_value, ".2f") if estimated_value is not None else None ), } ) balances.sort( key=lambda item: as_decimal(item.get("estimated_usdt")), reverse=True, ) orders: list[dict[str, Any]] = [] for item in orders_raw: symbol = str(item.get("symbol", "")) order_price = as_decimal(item.get("price")) market_price = prices.get(symbol, Decimal("0")) distance = self.order_distance_percent(order_price, market_price) orders.append( { "symbol": symbol, "side": str(item.get("side", "")), "type": str(item.get("type", "")), "status": str(item.get("status", "")), "order_price": decimal_text(order_price), "market_price": decimal_text(market_price) if market_price > 0 else None, "original_quantity": decimal_text(as_decimal(item.get("origQty"))), "executed_quantity": decimal_text(as_decimal(item.get("executedQty"))), "distance_percent": format(distance, ".2f") if distance is not None else None, } ) market: list[dict[str, str | None]] = [] for symbol in watchlist: clean_symbol = symbol.strip().upper() if not clean_symbol: continue price = prices.get(clean_symbol) ticker = tickers_by_symbol.get(clean_symbol, {}) high_price = as_decimal(ticker.get("highPrice")) low_price = as_decimal(ticker.get("lowPrice")) market.append( { "symbol": clean_symbol, "price": decimal_text(price) if price is not None else None, "high_price": decimal_text(high_price) if high_price > 0 else None, "low_price": decimal_text(low_price) if low_price > 0 else None, } ) candidates: list[str] = [] for symbol in analysis_symbols or []: candidates.append(symbol.strip().upper()) if include_account_candidates: for order in orders: candidates.append(str(order.get("symbol", "")).strip().upper()) for balance in balances: asset = str(balance.get("asset", "")).strip().upper() value = as_decimal(balance.get("estimated_usdt")) symbol = f"{asset}USDT" if asset not in {"USDT", ""} and value >= Decimal("10") and symbol in prices: candidates.append(symbol) candidates.extend(discovered_symbols) unique_candidates: list[str] = [] for symbol in candidates: if symbol and symbol.isalnum() and symbol not in unique_candidates: unique_candidates.append(symbol) # Keep the deeper candle analysis bounded so market data stays fresh. unique_candidates = unique_candidates[:20] held_symbols = { f"{str(balance.get('asset', '')).strip().upper()}USDT" for balance in balances if str(balance.get("asset", "")).strip().upper() not in {"", "USDT"} and as_decimal(balance.get("estimated_usdt")) >= Decimal("10") } requested_timeframes = [ item for item in (analysis_timeframes or ["1h", "4h", "1d"]) if item in {"15m", "1h", "4h", "1d"} ] timeframes = ["1h", "4h", "1d"] for interval in requested_timeframes: if interval not in timeframes: timeframes.append(interval) try: btc_timeframe_results = [ timeframe_analysis(interval, self.klines("BTCUSDT", interval, 250)) for interval in ("1h", "4h", "1d") ] except (BinanceReadOnlyError, ValueError): btc_timeframe_results = [] market_context = self.market_breadth_context( tickers_24hr, btc_timeframe_results, min_quote_volume_usdt=min_quote_volume_usdt, min_breadth_percent=min_market_breadth_percent, ) analyses: list[dict[str, Any]] = [] for symbol in unique_candidates: try: if symbol == "BTCUSDT" and tuple(timeframes) == ("1h", "4h", "1d"): timeframe_results = btc_timeframe_results unavailable_timeframes: list[str] = [] else: timeframe_results, unavailable_timeframes = self.available_timeframe_analyses( symbol, timeframes, ) if "1d" in unavailable_timeframes and "15m" not in timeframes: fallback_results, fallback_unavailable = self.available_timeframe_analyses( symbol, ("15m",), ) timeframe_results.extend(fallback_results) unavailable_timeframes.extend(fallback_unavailable) if not timeframe_results: raise ValueError( "Not enough closed candles for any supported timeframe analysis." ) symbol_ticker = tickers_by_symbol.get(symbol, {}) symbol_quote_volume = as_decimal(symbol_ticker.get("quoteVolume")) symbol_price_change = as_decimal(symbol_ticker.get("priceChangePercent")) plan_market_context = { **market_context, "symbol_liquid": symbol_quote_volume >= min_quote_volume_usdt, "symbol_quote_volume_usdt": format(symbol_quote_volume, ".2f"), "symbol_price_change_percent": format(symbol_price_change, ".2f"), } plan = trade_plan( symbol, timeframe_results, account_value_usdt=estimated_total, free_usdt=free_usdt, risk_percent=risk_percent, max_allocation_percent=max_allocation_percent, position_held=symbol in held_symbols, current_price=prices.get(symbol), market_context=plan_market_context, min_combined_score=min_signal_score, min_confidence_percent=min_confidence_percent, min_volume_ratio=min_volume_ratio, max_daily_move_percent=max_daily_move_percent, max_stop_risk_percent=max_stop_risk_percent, min_reward_risk=min_reward_risk, ) plan["analysis_limited"] = bool(unavailable_timeframes) plan["unavailable_timeframes"] = unavailable_timeframes if unavailable_timeframes: plan["analysis_mode"] = "NEW_COIN" plan["decision"] = "WAIT" plan["quality_gates"]["required_timeframes_present"] = False plan["explanation"].append( "New-coin mode: unavailable closed history for " + ", ".join(unavailable_timeframes) + "; strong recommendations are disabled." ) else: plan["analysis_mode"] = "STANDARD" plan["source"] = "MARKET_SCAN" if symbol in discovered_symbols else "ACCOUNT_OR_FAVORITES" analyses.append(plan) except (BinanceReadOnlyError, ValueError) as exc: analyses.append( { "symbol": symbol, "decision": "ERROR", "error": str(exc), "timeframes": [], } ) return { "mode": "READ_ONLY", "updated_at": datetime.now(timezone.utc).isoformat(), "permissions": permissions, "summary": { "estimated_total_usdt": format(estimated_total, ".2f"), "asset_count": len(balances), "open_order_count": len(orders), "unpriced_assets": unpriced_assets, }, "balances": balances, "orders": orders, "market": market, "analysis": analyses, "analysis_policy": { "execution": "NONE_READ_ONLY", "risk_percent": format(risk_percent, ".2f"), "max_allocation_percent": format(max_allocation_percent, ".2f"), "auto_scan_enabled": auto_scan_enabled, "auto_scan_count": len(discovered_symbols), "min_quote_volume_usdt": format(min_quote_volume_usdt, ".0f"), "required_timeframes": ["1h", "4h", "1d"], "min_signal_score": format(min_signal_score, ".2f"), "min_confidence_percent": format(min_confidence_percent, ".0f"), "min_volume_ratio": format(min_volume_ratio, ".2f"), "min_market_breadth_percent": format(min_market_breadth_percent, ".2f"), "max_daily_move_percent": format(max_daily_move_percent, ".2f"), "max_stop_risk_percent": format(max_stop_risk_percent, ".2f"), "min_reward_risk": format(min_reward_risk, ".2f"), "market_context": market_context, "discovered_symbols": discovered_symbols, "trade_permission_required": False, "withdrawal_permission_required": False, "manual_purchase_only": True, "disclaimer": "Decision support only. Signals can be wrong and do not guarantee profit.", }, }