Opinion.trade MCP Server

"""Trading client for Opinion.trade using the official SDK.""" import logging from typing import Any, Dict, Optional logger = logging.getLogger(__name__) class TradingAPIError(Exception): """Exception raised for Opinion.trade trading API errors.""" def __init__(self, message: str, error_code: Optional[str] = None): self.message = message self.error_code = error_code super().__init__(self.message) class TradingClient: """Client for Opinion.trade trading operations using opinion_clob_sdk. Wraps the official Opinion.trade SDK for: - Order placement (limit/market) - Order cancellation - Position management - Balance queries - Trade history The SDK handles EIP712 signing automatically. """ def __init__(self, config): """Initialize the trading client with opinion_clob_sdk. Args: config: OpinionConfig with api_key, private_key, chain_id Raises: ImportError: If opinion_clob_sdk is not installed TradingAPIError: If SDK initialization fails """ try: from opinion_clob_sdk import Client as OpinionSDKClient except ImportError: raise ImportError( "opinion-clob-sdk is required for trading mode. " "Install with: pip install opinion-clob-sdk" ) try: self.client = OpinionSDKClient( host=config.api_host, apikey=config.api_key, private_key=config.private_key, chain_id=config.chain_id ) logger.info("Opinion.trade SDK client initialized") except Exception as e: logger.error(f"Failed to initialize SDK client: {e}") raise TradingAPIError(f"SDK initialization failed: {str(e)}", "INIT_ERROR") async def place_order( self, token_id: str, side: str, amount: float, price: Optional[float] = None, order_type: str = "LIMIT" ) -> Dict[str, Any]: """Place a limit or market order. Args: token_id: Token ID to trade side: Order side (BUY/SELL) amount: Order size price: Limit price (required for LIMIT orders) order_type: Order type (LIMIT/MARKET) Returns: Order confirmation with order_id Raises: TradingAPIError: If order placement fails """ try: # Create order object for SDK order = { "token_id": token_id, "side": side.upper(), "amount": str(amount), "type": order_type.upper() } if order_type.upper() == "LIMIT": if price is None: raise ValueError("Price required for LIMIT orders") # Format price with max 2 decimals as string order["price"] = f"{price:.2f}" # SDK handles EIP712 signing automatically result = self.client.place_order(order) logger.info(f"Order placed: {result.get('order_id')}") return result except Exception as e: logger.error(f"Order placement failed: {e}") raise TradingAPIError(f"Failed to place order: {str(e)}", "ORDER_FAILED") async def cancel_order(self, order_id: str) -> Dict[str, Any]: """Cancel a specific order. Args: order_id: Order ID to cancel Returns: Cancellation confirmation Raises: TradingAPIError: If cancellation fails """ try: result = self.client.cancel_order(order_id) logger.info(f"Order cancelled: {order_id}") return result except Exception as e: logger.error(f"Order cancellation failed: {e}") raise TradingAPIError(f"Failed to cancel order: {str(e)}", "CANCEL_FAILED") async def cancel_all_orders(self, market_id: Optional[str] = None) -> Dict[str, Any]: """Cancel all open orders (optionally filtered by market). Args: market_id: Optional market filter Returns: Cancellation summary Raises: TradingAPIError: If cancellation fails """ try: # Get open orders first open_orders = await self.get_open_orders(market_id) orders = open_orders if isinstance(open_orders, list) else open_orders.get("list", []) cancelled = [] for order in orders: try: await self.cancel_order(order.get("order_id")) cancelled.append(order.get("order_id")) except Exception as e: logger.warning(f"Failed to cancel order {order.get('order_id')}: {e}") return { "cancelled_count": len(cancelled), "cancelled_orders": cancelled } except Exception as e: logger.error(f"Failed to cancel all orders: {e}") raise TradingAPIError(f"Failed to cancel all orders: {str(e)}", "CANCEL_ALL_FAILED") async def get_open_orders(self, market_id: Optional[str] = None) -> Dict[str, Any]: """Get user's open orders. Args: market_id: Optional market filter Returns: List of open orders Raises: TradingAPIError: If retrieval fails """ try: params = {} if market_id: params["market_id"] = market_id result = self.client.get_my_orders(**params) return result except Exception as e: logger.error(f"Failed to get open orders: {e}") raise TradingAPIError(f"Failed to get open orders: {str(e)}", "GET_ORDERS_FAILED") async def get_positions(self, market_id: Optional[str] = None) -> Dict[str, Any]: """Get current positions with P&L. Args: market_id: Optional market filter Returns: List of positions Raises: TradingAPIError: If retrieval fails """ try: params = {} if market_id: params["market_id"] = market_id result = self.client.get_my_positions(**params) return result except Exception as e: logger.error(f"Failed to get positions: {e}") raise TradingAPIError(f"Failed to get positions: {str(e)}", "GET_POSITIONS_FAILED") async def get_trade_history( self, market_id: Optional[str] = None, limit: int = 100, start_time: Optional[int] = None, end_time: Optional[int] = None ) -> Dict[str, Any]: """Get executed trades history. Args: market_id: Optional market filter limit: Maximum trades to return start_time: Start timestamp (ms) end_time: End timestamp (ms) Returns: List of trades Raises: TradingAPIError: If retrieval fails """ try: params = {"limit": limit} if market_id: params["market_id"] = market_id if start_time: params["start_time"] = start_time if end_time: params["end_time"] = end_time result = self.client.get_my_trades(**params) return result except Exception as e: logger.error(f"Failed to get trade history: {e}") raise TradingAPIError(f"Failed to get trade history: {str(e)}", "GET_TRADES_FAILED") async def get_balances(self) -> Dict[str, Any]: """Get account balances (available + locked). Returns: Balance information Raises: TradingAPIError: If retrieval fails """ try: result = self.client.get_my_balances() return result except Exception as e: logger.error(f"Failed to get balances: {e}") raise TradingAPIError(f"Failed to get balances: {str(e)}", "GET_BALANCES_FAILED")