Odos MCP Server

""" Track historical portfolio performance and returns analysis. """ from datetime import datetime, timedelta from typing import Any, Dict, List, Optional from pydantic import BaseModel from ...constants import ODOS_ROUTER_ADDRESS from ...helpers import make_zerion_request from ...mcp import mcp from .get_past_transactions import apply_filters, parse_transaction class OdosPerformanceArgs(BaseModel): wallet_address: str days_back: int = 30 input_tokens: Optional[List[str]] = None output_tokens: Optional[List[str]] = None token_addresses: Optional[List[str]] = None min_trade_value: Optional[float] = None max_trade_value: Optional[float] = None methods: Optional[List[str]] = None @mcp.tool( name="get_past_performance", description="Analyze Odos trading performance with optional filtering", ) async def get_past_performance(args: OdosPerformanceArgs) -> str: """Analyze Odos performance with filtering""" try: cutoff_timestamp = int( (datetime.now() - timedelta(days=args.days_back)).timestamp() * 1000 ) params: Dict[str, str] = { "filter[operation_types]": "trade", "filter[contract_addresses]": ODOS_ROUTER_ADDRESS, "filter[min_mined_at]": str(cutoff_timestamp), "page[size]": "50", } data = await make_zerion_request( f"/wallets/{args.wallet_address}/transactions", params ) if not data or "data" not in data: return "❌ No transaction data found" trades = data["data"] if not trades: return f"📊 No Odos trades found in the last {args.days_back} days" # Apply filters # We've verified that OdosPerformanceArgs is structurally compatible # with the subset of TransactionArgs fields used by apply_filters. filtered_trades = apply_filters(trades, args) # type: ignore[arg-type] if not filtered_trades: return f"📊 No trades match your filters. Found {len(trades)} total Odos trades." # Analyze performance analysis = analyze_trades(filtered_trades) return format_performance(args.wallet_address, analysis, args.days_back) except ConnectionError as e: return f"❌ API Error in get_past_performance: {str(e)}" except Exception as e: # pylint: disable=broad-except return f"❌ Unexpected Error in get_past_performance: {str(e)}" def _process_trade_status(parsed: Dict[str, Any]) -> Dict[str, int]: """Process trade status and return counts""" status = parsed.get("status", "unknown") counts = {"confirmed": 0, "failed": 0, "pending": 0} if status == "confirmed": counts["confirmed"] = 1 elif status == "failed": counts["failed"] = 1 else: counts["pending"] = 1 return counts def _process_gas_info(parsed: Dict[str, Any]) -> Dict[str, float]: """Process gas information from parsed transaction""" fee_info = parsed.get("fee", {}) gas_data: Dict[str, float] = {"gas_usd": 0.0, "gas_eth": 0.0} if fee_info: gas_usd = fee_info.get("value_usd", 0) gas_eth = fee_info.get("amount_eth", 0) if gas_usd > 0: gas_data["gas_usd"] = float(gas_usd) if gas_eth > 0: gas_data["gas_eth"] = float(gas_eth) return gas_data def _process_volume_info(parsed: Dict[str, Any]) -> Dict[str, float]: """Process volume information from parsed transaction""" trade_summary = parsed.get("trade_summary", {}) volume_data: Dict[str, float] = { "input_vol": 0.0, "output_vol": 0.0, "net_change": 0.0, } if trade_summary: volume_data["input_vol"] = float(trade_summary.get("input_total_usd", 0)) volume_data["output_vol"] = float(trade_summary.get("output_total_usd", 0)) volume_data["net_change"] = float(trade_summary.get("net_change_usd", 0)) return volume_data def _process_method_info(parsed: Dict[str, Any]) -> str: """Process method information from parsed transaction""" app_info = parsed.get("application", {}) if app_info: method_info = app_info.get("method", {}) if method_info: return str(method_info.get("name", "Unknown")) return "Unknown" def _process_tokens_info(parsed: Dict[str, Any]) -> set[str]: """Process token information from parsed transaction""" tokens_traded: set[str] = set() transfers = parsed.get("transfers", []) for transfer in transfers: token = transfer.get("token", {}) if token: symbol = token.get("symbol", "UNK") tokens_traded.add(str(symbol)) return tokens_traded def _process_chain_info(parsed: Dict[str, Any]) -> str: """Process chain information from parsed transaction""" chain_info = parsed.get("chain", {}) if chain_info: return str(chain_info.get("id", "unknown")) return "unknown" def _calculate_statistics(data: Dict[str, Any]) -> Dict[str, float]: """Calculate various statistics for the analysis""" stats: Dict[str, float] = {} total = data["total"] confirmed = data["confirmed"] failed = data["failed"] gas_costs = data["gas_costs"] trade_volumes = data["trade_volumes"] total_gas_usd = data["total_gas_usd"] total_gas_eth = data["total_gas_eth"] total_input_volume = data["total_input_volume"] # Confirmation and failure rates stats["confirmation_rate"] = (confirmed / total * 100) if total > 0 else 0 stats["failure_rate"] = (failed / total * 100) if total > 0 else 0 # Gas statistics stats["average_gas_cost_usd"] = total_gas_usd / total if total > 0 else 0 stats["average_gas_cost_eth"] = total_gas_eth / total if total > 0 else 0 stats["min_gas_cost_usd"] = min(gas_costs) if gas_costs else 0 stats["max_gas_cost_usd"] = max(gas_costs) if gas_costs else 0 # Volume statistics stats["average_trade_size_usd"] = ( sum(trade_volumes) / len(trade_volumes) if trade_volumes else 0 ) stats["min_trade_size_usd"] = min(trade_volumes) if trade_volumes else 0 stats["max_trade_size_usd"] = max(trade_volumes) if trade_volumes else 0 # Gas to volume ratio stats["gas_to_volume_ratio"] = ( (total_gas_usd / total_input_volume * 100) if total_input_volume > 0 else 0 ) return stats def _process_single_trade( trade: Dict, totals: Dict[str, Any], collections: Dict[str, Any] ) -> None: """Process a single trade and update totals and collections""" try: parsed = parse_transaction(trade) if not parsed: return # Process status status_counts = _process_trade_status(parsed) totals["confirmed"] += status_counts["confirmed"] totals["failed"] += status_counts["failed"] totals["pending"] += status_counts["pending"] # Process gas information gas_data = _process_gas_info(parsed) totals["total_gas_usd"] += gas_data["gas_usd"] totals["total_gas_eth"] += gas_data["gas_eth"] if gas_data["gas_usd"] > 0: collections["gas_costs"].append(gas_data["gas_usd"]) # Process volume information volume_data = _process_volume_info(parsed) totals["total_input_volume"] += volume_data["input_vol"] totals["total_output_volume"] += volume_data["output_vol"] totals["total_net_change"] += volume_data["net_change"] if volume_data["input_vol"] > 0: collections["trade_volumes"].append(volume_data["input_vol"]) # Process method information method = _process_method_info(parsed) collections["methods_used"][method] = ( collections["methods_used"].get(method, 0) + 1 ) # Process token information trade_tokens = _process_tokens_info(parsed) collections["tokens_traded"].update(trade_tokens) # Process chain information chain_id = _process_chain_info(parsed) collections["chains_used"].add(chain_id) except (KeyError, ValueError, TypeError) as e: print(f"Error parsing trade: {e}") def analyze_trades(trades: List[Dict]) -> Dict[str, Any]: """Comprehensive trade analysis with error handling""" # Initialize totals totals: Dict[str, Any] = { "confirmed": 0, "failed": 0, "pending": 0, "total_gas_usd": 0, "total_gas_eth": 0, "total_input_volume": 0, "total_output_volume": 0, "total_net_change": 0, } # Initialize collections collections: Dict[str, Any] = { "gas_costs": [], "trade_volumes": [], "methods_used": {}, "tokens_traded": set(), "chains_used": set(), } # Process each trade for trade in trades: _process_single_trade(trade, totals, collections) # Calculate statistics stats_data = { "total": len(trades), "confirmed": totals["confirmed"], "failed": totals["failed"], "gas_costs": collections["gas_costs"], "trade_volumes": collections["trade_volumes"], "total_gas_usd": totals["total_gas_usd"], "total_gas_eth": totals["total_gas_eth"], "total_input_volume": totals["total_input_volume"], } stats = _calculate_statistics(stats_data) return { "total_transactions": len(trades), "confirmed_transactions": totals["confirmed"], "failed_transactions": totals["failed"], "pending_transactions": totals["pending"], "confirmation_rate": stats["confirmation_rate"], "failure_rate": stats["failure_rate"], "total_gas_cost_usd": totals["total_gas_usd"], "total_gas_cost_eth": totals["total_gas_eth"], "average_gas_cost_usd": stats["average_gas_cost_usd"], "average_gas_cost_eth": stats["average_gas_cost_eth"], "min_gas_cost_usd": stats["min_gas_cost_usd"], "max_gas_cost_usd": stats["max_gas_cost_usd"], "total_input_volume_usd": totals["total_input_volume"], "total_output_volume_usd": totals["total_output_volume"], "total_net_change_usd": totals["total_net_change"], "average_trade_size_usd": stats["average_trade_size_usd"], "min_trade_size_usd": stats["min_trade_size_usd"], "max_trade_size_usd": stats["max_trade_size_usd"], "total_trade_count": len(collections["trade_volumes"]), "methods_used": dict(collections["methods_used"]), "unique_tokens_traded": list(collections["tokens_traded"]), "unique_chains_used": list(collections["chains_used"]), "gas_to_volume_ratio": stats["gas_to_volume_ratio"], } def format_performance( wallet_address: str, analysis: Dict[str, Any], days_back: int ) -> str: """Format comprehensive performance analysis""" output = "ODOS PERFORMANCE ANALYSIS\n" output += f"Wallet: {wallet_address}\n" output += f"Analysis period: {days_back} days\n" output += f"Router contract: {ODOS_ROUTER_ADDRESS}\n\n" output += "TRANSACTION STATISTICS:\n" output += f" Total transactions: {analysis['total_transactions']}\n" output += f" Confirmed transactions: {analysis['confirmed_transactions']}\n" output += f" Failed transactions: {analysis['failed_transactions']}\n" output += f" Pending transactions: {analysis['pending_transactions']}\n" output += f" Confirmation rate: {analysis['confirmation_rate']:.2f}%\n" output += f" Failure rate: {analysis['failure_rate']:.2f}%\n\n" output += "GAS COST ANALYSIS:\n" output += f" Total gas cost (USD): ${analysis['total_gas_cost_usd']:.6f}\n" output += f" Total gas cost (ETH): {analysis['total_gas_cost_eth']:.12f}\n" output += f" Average gas cost (USD): ${analysis['average_gas_cost_usd']:.6f}\n" output += f" Average gas cost (ETH): {analysis['average_gas_cost_eth']:.12f}\n" output += f" Minimum gas cost (USD): ${analysis['min_gas_cost_usd']:.6f}\n" output += f" Maximum gas cost (USD): ${analysis['max_gas_cost_usd']:.6f}\n\n" output += "TRADING VOLUME ANALYSIS:\n" output += f" Total input volume (USD): ${analysis['total_input_volume_usd']:.6f}\n" output += ( f" Total output volume (USD): ${analysis['total_output_volume_usd']:.6f}\n" ) output += f" Total net change (USD): ${analysis['total_net_change_usd']:.6f}\n" output += f" Average trade size (USD): ${analysis['average_trade_size_usd']:.6f}\n" output += f" Minimum trade size (USD): ${analysis['min_trade_size_usd']:.6f}\n" output += f" Maximum trade size (USD): ${analysis['max_trade_size_usd']:.6f}\n" output += f" Gas to volume ratio: {analysis['gas_to_volume_ratio']:.6f}%\n\n" output += "METHOD USAGE:\n" methods = analysis.get("methods_used", {}) for method, count in methods.items(): percentage = ( (count / analysis["total_transactions"] * 100) if analysis["total_transactions"] > 0 else 0 ) output += f" {method}: {count} transactions ({percentage:.2f}%)\n" output += "\n" output += "TOKENS TRADED:\n" tokens = analysis.get("unique_tokens_traded", []) output += f" Unique tokens: {len(tokens)}\n" output += f" Token symbols: {', '.join(sorted(tokens))}\n\n" output += "CHAINS USED:\n" chains = analysis.get("unique_chains_used", []) output += f" Unique chains: {len(chains)}\n" output += f" Chain IDs: {', '.join(sorted(chains))}\n" return output