diff --git a/src/backend/app/api/bots.py b/src/backend/app/api/bots.py index 051a628..a37740a 100644 --- a/src/backend/app/api/bots.py +++ b/src/backend/app/api/bots.py @@ -16,7 +16,7 @@ from ..db.schemas import ( ) from ..db.models import Bot, BotConversation, User from ..services.ai_agent.crew import get_trading_crew -from ..services.ai_agent.conversational import get_conversational_agent +from ..services.ai_agent import get_conversational_agent router = APIRouter() MAX_BOTS_PER_USER = 3 @@ -190,7 +190,7 @@ def chat( ] user_message = request.message - + # Use ConversationalAgent for natural chat with tool-calling agent = get_conversational_agent(bot_id=bot_id) result = agent.chat(user_message, history_for_agent) @@ -224,7 +224,9 @@ def chat( strategy_config=bot.strategy_config if result.get("strategy_updated") else None, success=result.get("success", False), strategy_needs_confirmation=result.get("strategy_needs_confirmation", False), - strategy_data=result.get("strategy_data") if result.get("strategy_needs_confirmation") else None, + strategy_data=result.get("strategy_data") + if result.get("strategy_needs_confirmation") + else None, token_search_results=result.get("token_search_results"), ) diff --git a/src/backend/app/services/ai_agent/__init__.py b/src/backend/app/services/ai_agent/__init__.py index 4060524..e3153d4 100644 --- a/src/backend/app/services/ai_agent/__init__.py +++ b/src/backend/app/services/ai_agent/__init__.py @@ -1,4 +1,29 @@ +"""AI Agent module for conversational trading.""" + +from .agent import ConversationalAgent, get_conversational_agent +from .client import MiniMaxClient +from .tools import get_tool_registry, TOOL_REGISTRY +from .help import ( + format_tools_list, + format_general_help, + format_tool_help, + format_skill_acknowledgment, +) from .crew import TradingCrew, get_trading_crew from .llm_connector import MiniMaxLLM, MiniMaxConnector -__all__ = ["TradingCrew", "get_trading_crew", "MiniMaxLLM", "MiniMaxConnector"] +__all__ = [ + "ConversationalAgent", + "get_conversational_agent", + "MiniMaxClient", + "get_tool_registry", + "TOOL_REGISTRY", + "format_tools_list", + "format_general_help", + "format_tool_help", + "format_skill_acknowledgment", + "TradingCrew", + "get_trading_crew", + "MiniMaxLLM", + "MiniMaxConnector", +] diff --git a/src/backend/app/services/ai_agent/conversational.py b/src/backend/app/services/ai_agent/agent.py similarity index 69% rename from src/backend/app/services/ai_agent/conversational.py rename to src/backend/app/services/ai_agent/agent.py index d893bc5..a01130d 100644 --- a/src/backend/app/services/ai_agent/conversational.py +++ b/src/backend/app/services/ai_agent/agent.py @@ -1,498 +1,47 @@ -""" -Conversational Trading Agent - -This agent can: -1. Have normal conversations with users -2. Update trading strategies when user provides specific instructions - -Uses MiniMax extended thinking API for proper thinking/reasoning separation. -""" +"""Conversational trading agent.""" import json import re -import requests from typing import List, Optional, Dict, Any from datetime import timedelta +from .client import MiniMaxClient, SYSTEM_PROMPT_WITH_TOOLS, TOOLS +from .help import ( + format_tools_list, + format_general_help, + format_tool_help, + format_skill_acknowledgment, +) +from .tools import get_tool_registry from ...core.config import get_settings from ...db.models import Bot, Simulation -TOOL_REGISTRY = { - "randebu": [ - { - "name": "backtest", - "description": "Run strategy backtest", - "category": "Randebu Built-in", - "command": "/backtest", - "details": { - "description": "Run a backtest to evaluate how the current trading strategy would have performed historically.", - "usage": "/backtest [token_address] [--timeframe 1d|4h|1h|15m] [--start YYYY-MM-DD] [--end YYYY-MM-DD]", - "example": "Run a backtest on PEPE for the last 30 days", - }, - }, - { - "name": "simulate", - "description": "Start/stop simulation", - "category": "Randebu Built-in", - "command": "/simulate", - "details": { - "description": "Start or stop trading simulations that run on real-time klines.", - "usage": "/simulate start|stop|status|results [token_address]", - "example": "Start a simulation on PEPE", - }, - }, - { - "name": "strategy", - "description": "View/update strategy", - "category": "Randebu Built-in", - "command": "/strategy", - "details": { - "description": "View your current trading strategy or update it with new parameters.", - "usage": "Describe your strategy in plain English, e.g., 'Buy PEPE when price drops 5%'", - "example": "Buy PEPE when it drops 10% within 1 hour", - }, - }, - ], - "ave": [ - { - "name": "search", - "description": "Token search", - "category": "AVE Cloud Skills", - "command": "/search", - "details": { - "description": "Find tokens by keyword, symbol, or contract address on BSC.", - "usage": "search [--chain bsc] [--limit 20]", - "example": "search PEPE\nsearch 0x1234... --chain bsc", - }, - }, - { - "name": "trending", - "description": "Popular tokens", - "category": "AVE Cloud Skills", - "command": "/trending", - "details": { - "description": "Get list of trending/popular tokens on BSC.", - "usage": "trending [--chain bsc] [--limit 20]", - "example": "trending --chain bsc\ntrending --limit 10", - }, - }, - { - "name": "risk", - "description": "Honeypot detection", - "category": "AVE Cloud Skills", - "command": "/risk", - "details": { - "description": "Get risk analysis for a token contract including honeypot assessment.", - "usage": "risk [--chain bsc]", - "example": "risk 0x6982508145454Ce125dDE157d8d64a26D53f60a2", - }, - }, - { - "name": "token", - "description": "Token details", - "category": "AVE Cloud Skills", - "command": "/token", - "details": { - "description": "Get detailed information about a specific token including price, market cap, and pairs.", - "usage": "token
[--chain bsc]", - "example": "token 0x6982508145454Ce125dDE157d8d64a26D53f60a2", - }, - }, - { - "name": "price", - "description": "Batch prices", - "category": "AVE Cloud Skills", - "command": "/price", - "details": { - "description": "Get current price(s) for multiple tokens.", - "usage": "price ,,... (e.g., PEPE-bsc,TRUMP-bsc)", - "example": "price PEPE-bsc,TRUMP-bsc", - }, - }, - ], -} - - -# Skill emojis mapping -SKILL_EMOJIS = { - "backtest": "šŸ“Š", - "simulate": "šŸŽ®", - "strategy": "šŸ“", - "search": "šŸ”", - "trending": "šŸ“ˆ", - "risk": "šŸ“‰", - "token": "šŸŖ™", - "price": "šŸ’°", -} - - -def get_tool_registry() -> Dict[str, Any]: - """Return the tool registry for slash command help.""" - return TOOL_REGISTRY - - -def format_tools_list() -> str: - """Format the tool registry as a help message.""" - message = "šŸ“‹ Available Tools\n\n" - - for category in ["randebu", "ave"]: - tools = TOOL_REGISTRY.get(category, []) - if category == "randedu": - message += "šŸ¤– Randebu Built-in:\n" - else: - message += "ā˜ļø AVE Cloud Skills:\n" - - for tool in tools: - message += f" ā€¢ {tool['command']} - {tool['description']}\n" - message += "\n" - - message = ( - message.rstrip() + "\n\nType / for detailed help on a specific tool." - ) - return message - - -def format_skill_acknowledgment(tool_name: str, description: str) -> str: - """Format a brief acknowledgment when a skill is activated.""" - emoji = SKILL_EMOJIS.get(tool_name.lower(), "āœØ") - return f"{emoji} **{tool_name}** loaded. Ready for *{description}*, ask me away!" - - -def format_tool_help(tool_name: str) -> str: - """Format detailed help for a specific tool.""" - tool_name = tool_name.lstrip("/") - - for category in ["randebu", "ave"]: - for tool in TOOL_REGISTRY.get(category, []): - if tool["name"].lower() == tool_name.lower(): - cat_label = ( - "Randebu Built-in" if category == "randebu" else "AVE Cloud Skill" - ) - details = tool["details"] - message = ( - f"šŸ” {tool['command']} - {details['description']} ({cat_label})\n\n" - ) - message += f"**Description:** {details['description']}\n" - message += f"**Commands:**\n {details['usage']}\n\n" - message += f"**Example:**\n```\n{details['example']}\n```" - return message - - return f"Tool '{tool_name}' not found. Type / to see all available tools." - - -def format_general_help() -> str: - """Format general help about Randebu.""" - return """šŸ¤– **Randebu - AI Trading Assistant** - -Randebu is your AI trading assistant that helps you manage your trading bots on BSC (Binance Smart Chain). - -**Getting Started:** -1. Create a bot on the dashboard -2. Describe your trading strategy in plain English -3. Run backtests to validate your strategy -4. Start simulations to see live trading - -**Example Strategies:** -- "Buy PEPE when it drops 5%" -- "Sell if price rises 10% within 1 hour" -- "Buy when volume spikes by 200%" - -**Slash Commands:** -- `/` - Show all available tools -- `/help` - Show this help message -- `/` - Get help on a specific tool - -**Natural Language:** -You can also just describe what you want in natural language. For example: -- "What's the price of PEPE?" -- "Run a backtest on 0x... token" -- "Start a simulation on TRUMP" -""" - - -SYSTEM_PROMPT = """You are a helpful AI trading assistant named Randebu. You help users manage their trading bots. - -IMPORTANT CHAIN LIMITATION: -- We ONLY support BSC (Binance Smart Chain) blockchain -- If user asks about any other chain (Solana, ETH, Base, etc.), respond with: "Currently we only support BSC (Binance Smart Chain). All trading strategies and token searches are performed on BSC." -- Never search or recommend tokens on other chains -- The search_tokens tool defaults to BSC, never change this - -Your response must be valid JSON with exactly this structure: -{ - "thinking": "Your internal reasoning and analysis (what you're thinking about)", - "response": "Your actual response to the user (be concise and helpful)", - "strategy_update": null or { - "conditions": [{"type": "price_drop" | "price_rise" | "volume_spike" | "price_level", "token": "TOKEN_SYMBOL", "token_address": null, "threshold": number, ...}], - "actions": [{"type": "buy" | "sell" | "hold", "amount_percent": number, ...}], - "risk_management": {"stop_loss_percent": number, "take_profit_percent": number} - } -}" - -Guidelines: -- "thinking" should be detailed reasoning about the user's request -- "response" should be conversational and clear -- "strategy_update" should be populated ONLY when the user provides specific trading parameters (percentages, tokens, conditions, etc.) -- IMPORTANT: When a token is mentioned, set "token_address": null and ask user to confirm the token address before saving. Your response should say something like: "I need to confirm the token address. Could you provide the contract address for [TOKEN]?" -- If no strategy parameters are provided, set "strategy_update" to null -- Be friendly, concise, and helpful in your response - -Example 1 (no strategy update): -User: "What can this bot do?" -{ - "thinking": "The user is asking about the bot's capabilities. I should explain the main features.", - "response": "Randebu is your AI trading assistant! It can monitor cryptocurrency prices and execute trades based on your configured strategies. Tell me your trading parameters and I'll set them up for you.", - "strategy_update": null -} - -Example 2 (token needs confirmation): -User: "I want to buy PEPE when it drops 10%" -{ - "thinking": "User wants to buy PEPE. I need the token contract address to proceed. I should ask for confirmation.", - "response": "I'd be happy to set up a buy order for PEPE! However, I need to confirm the token contract address. Could you provide the BSC contract address for PEPE? (It usually starts with 0x...)", - "strategy_update": { - "conditions": [{"type": "price_drop", "token": "PEPE", "token_address": null, "threshold": 10}], - "actions": [{"type": "buy", "amount_percent": 100}], - "risk_management": null - } -} - -Example 3 (with token address provided by user): -User: "Buy 0x6982508145454Ce125dDE157d8d64a26D53f60a2 when it drops 10%" -{ - "thinking": "User provided a contract address, I can use it directly.", - "response": "Perfect! I've configured your strategy to buy the token when it drops 10%.", - "strategy_update": { - "conditions": [{"type": "price_drop", "token": "TOKEN", "token_address": "0x6982508145454Ce125dDE157d8d64a26D53f60a2", "threshold": 10}], - "actions": [{"type": "buy", "amount_percent": 100}], - "risk_management": null - } -}""" - - -# Tool definitions for the agent -TOOLS = [ - { - "type": "function", - "function": { - "name": "search_tokens", - "description": "Search for tokens by keyword on BSC blockchain. Use this when user asks to search for a specific token or find tokens by name/symbol.", - "parameters": { - "type": "object", - "properties": { - "keyword": { - "type": "string", - "description": "Token symbol or name to search for (e.g., 'PEPE', 'BTC')", - }, - "limit": { - "type": "integer", - "description": "Number of tokens to return (default: 10)", - "default": 10, - }, - }, - "required": ["keyword"], - }, - }, - }, - { - "type": "function", - "function": { - "name": "get_token", - "description": "Get detailed information about a specific token including price, market cap, and pairs. Use when user asks for token details or wants to find a specific token by contract address.", - "parameters": { - "type": "object", - "properties": { - "address": { - "type": "string", - "description": "Token contract address (e.g., '0x6982508145454Ce125dDE157d8d64a26D53f60a2')", - }, - "chain": { - "type": "string", - "description": "Blockchain chain (default: bsc)", - "default": "bsc", - }, - }, - "required": ["address"], - }, - }, - }, - { - "type": "function", - "function": { - "name": "get_price", - "description": "Get current price(s) for tokens. Use when user asks for token price or wants to compare prices of multiple tokens.", - "parameters": { - "type": "object", - "properties": { - "token_ids": { - "type": "string", - "description": "Comma-separated list of token IDs with chain suffix (e.g., 'PEPE-bsc,TRUMP-bsc')", - } - }, - "required": ["token_ids"], - }, - }, - }, - { - "type": "function", - "function": { - "name": "get_risk", - "description": "Get risk analysis for a token contract. Use when user asks about token risk, honeypot analysis, or safety assessment before trading.", - "parameters": { - "type": "object", - "properties": { - "address": { - "type": "string", - "description": "Token contract address (e.g., '0x6982508145454Ce125dDE157d8d64a26D53f60a2')", - }, - "chain": { - "type": "string", - "description": "Blockchain chain (default: bsc)", - "default": "bsc", - }, - }, - "required": ["address"], - }, - }, - }, - { - "type": "function", - "function": { - "name": "get_trending", - "description": "Get trending tokens on a blockchain. Use when user asks what's trending, top tokens, or popular tokens right now.", - "parameters": { - "type": "object", - "properties": { - "chain": { - "type": "string", - "description": "Blockchain chain (default: bsc)", - "default": "bsc", - }, - "limit": { - "type": "integer", - "description": "Number of trending tokens to return (default: 10, max: 50)", - "default": 10, - }, - }, - }, - }, - }, - { - "type": "function", - "function": { - "name": "run_backtest", - "description": "Run a backtest to evaluate how the current trading strategy would have performed historically. Returns key metrics like ROI, win rate, max drawdown, etc. Use this when user asks to backtest, test strategy, or check historical performance.", - "parameters": { - "type": "object", - "properties": { - "token_address": { - "type": "string", - "description": "The BSC contract address of the token to backtest (required)", - }, - "timeframe": { - "type": "string", - "description": "Timeframe for klines: '1d' (1 day), '4h' (4 hours), '1h' (1 hour), '15m' (15 minutes)", - "default": "1d", - }, - "start_date": { - "type": "string", - "description": "Start date for backtest in YYYY-MM-DD format (e.g., '2024-01-01')", - }, - "end_date": { - "type": "string", - "description": "End date for backtest in YYYY-MM-DD format (e.g., '2024-12-01')", - }, - }, - "required": ["token_address"], - }, - }, - }, - { - "type": "function", - "function": { - "name": "manage_simulation", - "description": "Manage trading simulations: start, stop, or check status. Simulations run on real-time klines and show live portfolio updates. Use when user asks to run simulation, check simulation status, or stop simulation.", - "parameters": { - "type": "object", - "properties": { - "action": { - "type": "string", - "enum": ["start", "stop", "status", "results"], - "description": "Action to perform: 'start' (begin new simulation), 'stop' (stop running simulation), 'status' (check if simulation is running), 'results' (get results from current or latest simulation)", - }, - "token_address": { - "type": "string", - "description": "Token contract address for simulation (required for 'start' action)", - }, - "kline_interval": { - "type": "string", - "description": "Kline interval: '1m', '5m', '15m', '1h' (default: '1m')", - "default": "1m", - }, - }, - "required": ["action"], - }, - }, - }, -] - -SYSTEM_PROMPT_WITH_TOOLS = ( - SYSTEM_PROMPT - + """ - -You have access to tools: -- search_tokens(keyword, limit): Search for tokens by keyword. Use it when user asks to search for a token or find tokens by name/symbol. -- get_token(address, chain): Get detailed information about a specific token. Use when user asks for token details. -- get_price(token_ids): Get current price(s) for tokens. Use when user asks for token price. -- get_risk(address, chain): Get risk analysis for a token. Use when user asks about token safety or honeypot analysis. -- get_trending(chain, limit): Get trending tokens on a blockchain. Use when user asks what's trending, top tokens, or popular tokens. -- run_backtest(token_address, timeframe, start_date, end_date): Run a backtest on historical data. Returns performance metrics. Use when user asks to backtest or check historical performance. -- manage_simulation(action, token_address, kline_interval): Manage trading simulations. Actions: 'start' (begin new), 'stop' (stop running), 'status' (check if running), 'results' (get current/latest results). - -When you want to use a tool, respond with: -{ - "thinking": "...", - "response": "Running backtest...", - "tool_call": {"name": "run_backtest", "arguments": {"token_address": "0x...", "timeframe": "1d", "start_date": "2024-01-01", "end_date": "2024-12-01"}} -} -""" -) - - class ConversationalAgent: def __init__(self, api_key: str, model: str = "MiniMax-M2.7", bot_id: str = None): self.api_key = api_key self.model = model self.bot_id = bot_id - # Extended thinking endpoint - self.thinking_endpoint = "https://api.minimax.io/v1/text/chatcompletion_v2" + self.client = MiniMaxClient(api_key, model) - # Track pending command after acknowledgment self.pending_command = None - - # Track recent search results for context - self.recent_search_results = [] # List of {symbol, name, address} + self.recent_search_results = [] def _is_error_output(self, code: int, output: str) -> bool: """Check if the command output contains an error.""" if code != 0: return True - # Check for common error patterns in output - if output.startswith("Error:") or "API error" in output or "api key invalid" in output.lower(): + if ( + output.startswith("Error:") + or "API error" in output + or "api key invalid" in output.lower() + ): return True return False def _handle_slash_command(self, user_message: str) -> Dict[str, Any]: - """Handle slash command help requests. - - Args: - user_message: The slash command message (e.g., '/', '/help', '/search') - - Returns: - Dict with 'response', 'thinking', and other fields - """ + """Handle slash command help requests.""" cmd = user_message.strip().lower() if cmd == "/": @@ -512,30 +61,26 @@ class ConversationalAgent: "success": True, } elif cmd.startswith("/"): - # Check if it's a known skill (only look at first word after /) parts = cmd[1:].split() tool_name = parts[0].lower() if parts else "" has_args = len(parts) > 1 for category in ["randebu", "ave"]: - for tool in TOOL_REGISTRY.get(category, []): + for tool in get_tool_registry().get(category, []): if tool["name"].lower() == tool_name: - # Special handling for /strategy - fetch current strategy from DB if tool_name == "strategy" and not has_args: return self._get_strategy_response() - # Special handling for /trending - execute trending directly if tool_name == "trending" and not has_args: return self._execute_trending() - # Special handling for /backtest - execute directly if tool_name == "backtest": - return self._execute_backtest_direct(user_message if has_args else "") - # Special handling for /simulate - execute directly + return self._execute_backtest_direct( + user_message if has_args else "" + ) if tool_name == "simulate": - return self._execute_simulate_direct(user_message if has_args else "") - # For commands that need params (/search, /risk, /token, /price) - # execute immediately if args provided, otherwise set pending + return self._execute_simulate_direct( + user_message if has_args else "" + ) if not has_args: - # Set pending command for next message self.pending_command = tool_name return { "response": format_skill_acknowledgment( @@ -546,9 +91,8 @@ class ConversationalAgent: "strategy_needs_confirmation": False, "success": True, } - return None # Has args - let AI handle it + return None - # Unknown skill return { "response": f"Unknown command '{tool_name}'. Type / to see available tools.", "thinking": None, @@ -578,6 +122,7 @@ class ConversationalAgent: try: from ...core.database import get_db + db = next(get_db()) try: bot = db.query(Bot).filter(Bot.id == self.bot_id).first() @@ -600,14 +145,12 @@ class ConversationalAgent: "success": True, } - # Format the strategy nicely conditions = strategy_config.get("conditions", []) actions = strategy_config.get("actions", []) risk = strategy_config.get("risk_management", {}) response = "šŸ“ **Your Current Strategy**\n\n" - # Format conditions if conditions: response += "**Conditions:**\n" for cond in conditions: @@ -630,16 +173,16 @@ class ConversationalAgent: response += "\n" response += "\n" - # Format actions if actions: response += "**Actions:**\n" for action in actions: action_type = action.get("type", "unknown") amount = action.get("amount_percent", 0) - response += f"- {action_type.capitalize()} {amount}% of balance\n" + response += ( + f"- {action_type.capitalize()} {amount}% of balance\n" + ) response += "\n" - # Format risk management if risk: response += "**Risk Management:**\n" stop_loss = risk.get("stop_loss_percent", 0) @@ -686,7 +229,6 @@ class ConversationalAgent: } try: data = json.loads(output) - # Handle both dict with 'tokens' key and direct list data_field = data.get("data", []) if isinstance(data_field, list): tokens = data_field @@ -754,31 +296,33 @@ class ConversationalAgent: } try: data = json.loads(output) - # Handle both dict with 'tokens' key and direct list data_field = data.get("data", []) if isinstance(data_field, list): tokens = data_field else: tokens = data_field.get("tokens", []) if tokens: - # Store search results for context self.recent_search_results = [] token_list = "" for t in tokens[:10]: addr = t.get("token", "") symbol = t.get("symbol", "") name = t.get("name", "") - # Search API uses price_change_24h, trending uses token_price_change_24h - price_change = t.get("price_change_24h") or t.get("token_price_change_24h") or "N/A" + price_change = ( + t.get("price_change_24h") + or t.get("token_price_change_24h") + or "N/A" + ) mc = t.get("market_cap", "N/A") - # Store for context if addr and symbol: - self.recent_search_results.append({ - "symbol": symbol, - "name": name, - "address": addr, - "chain": "bsc" - }) + self.recent_search_results.append( + { + "symbol": symbol, + "name": name, + "address": addr, + "chain": "bsc", + } + ) try: mc_str = f"${float(mc):,.0f}" except (ValueError, TypeError): @@ -844,15 +388,12 @@ class ConversationalAgent: risk_score = risk_data.get("risk_score", "N/A") token_symbol = risk_data.get("token_symbol", "") token_name = risk_data.get("token_name", "") - - # Format token label + if token_symbol: token_label = f"**{token_symbol}** ({token_name}) - `{address}`" else: token_label = f"`{address}`" - - # Convert is_honeypot to string - # -1 = unknown/could not determine, 0 = false, 1 = true + if isinstance(is_honeypot, bool): is_honeypot_str = str(is_honeypot).lower() elif isinstance(is_honeypot, int): @@ -863,31 +404,49 @@ class ConversationalAgent: else: is_honeypot_str = "Unknown (could not determine)" else: - is_honeypot_str = str(is_honeypot).lower() if is_honeypot else "Unknown (could not determine)" - - # Convert tax values + is_honeypot_str = ( + str(is_honeypot).lower() + if is_honeypot + else "Unknown (could not determine)" + ) + try: - buy_tax_val = float(buy_tax) if buy_tax not in (None, "N/A") else 0 + buy_tax_val = ( + float(buy_tax) if buy_tax not in (None, "N/A") else 0 + ) except (ValueError, TypeError): buy_tax_val = 0 try: - sell_tax_val = float(sell_tax) if sell_tax not in (None, "N/A") else 0 + sell_tax_val = ( + float(sell_tax) if sell_tax not in (None, "N/A") else 0 + ) except (ValueError, TypeError): sell_tax_val = 0 - - # Determine risk level label - risk_level_str = "Low" if risk_level == 0 else "Medium" if risk_level == 1 else "High" if risk_level == 2 else "Unknown" - + + risk_level_str = ( + "Low" + if risk_level == 0 + else "Medium" + if risk_level == 1 + else "High" + if risk_level == 2 + else "Unknown" + ) + risk_text = f"šŸ›”ļø **Risk Analysis for {token_label}**\n\n" - risk_text += f"- Risk Level: {risk_level_str} (Score: {risk_score})\n" + risk_text += ( + f"- Risk Level: {risk_level_str} (Score: {risk_score})\n" + ) risk_text += f"- Honeypot: {is_honeypot_str}\n" risk_text += f"- Buy Tax: {buy_tax}%\n" risk_text += f"- Sell Tax: {sell_tax}%\n" - + if is_honeypot_str == "true": risk_text += "\nāš ļø **Warning: This token appears to be a honeypot. Do not buy!**" elif buy_tax_val > 10 or sell_tax_val > 10: - risk_text += "\nāš ļø **Warning: High tax detected. Trade with caution!**" + risk_text += ( + "\nāš ļø **Warning: High tax detected. Trade with caution!**" + ) else: risk_text += "\nāœ… This token appears safe to trade." return { @@ -967,20 +526,44 @@ class ConversationalAgent: "strategy_needs_confirmation": False, "success": True, } - price = token_info.get("current_price_usd") or token_info.get("price_usd") or token_info.get("price") or token_data.get("price") or "N/A" - mc = token_info.get("market_cap") or token_info.get("fdv") or token_data.get("market_cap") or "N/A" - vol = token_info.get("tx_volume_u_24h") or token_info.get("volume_24h") or token_data.get("volume_24h") or "N/A" - pairs = token_info.get("top_pairs") or token_data.get("top_pairs") or [] + price = ( + token_info.get("current_price_usd") + or token_info.get("price_usd") + or token_info.get("price") + or token_data.get("price") + or "N/A" + ) + mc = ( + token_info.get("market_cap") + or token_info.get("fdv") + or token_data.get("market_cap") + or "N/A" + ) + vol = ( + token_info.get("tx_volume_u_24h") + or token_info.get("volume_24h") + or token_data.get("volume_24h") + or "N/A" + ) + pairs = ( + token_info.get("top_pairs") or token_data.get("top_pairs") or [] + ) pairs_text = "" if pairs: pairs_text = "\n**Top Pairs:**\n" for p in pairs[:3]: - liq = p.get('liquidity', 'N/A') + liq = p.get("liquidity", "N/A") try: - liq_str = f"${float(liq):,.0f}" if liq and liq != "N/A" else liq + liq_str = ( + f"${float(liq):,.0f}" + if liq and liq != "N/A" + else liq + ) except (ValueError, TypeError): liq_str = str(liq) - pairs_text += f"- {p.get('pair', 'N/A')}: {liq_str} liquidity\n" + pairs_text += ( + f"- {p.get('pair', 'N/A')}: {liq_str} liquidity\n" + ) try: mc_str = f"${float(mc):,.0f}" if mc != "N/A" else mc except (ValueError, TypeError): @@ -1034,17 +617,17 @@ class ConversationalAgent: "success": True, } - # Check if input matches recent search results token_input = tokens_list[0].lower() matched_address = None for result in self.recent_search_results: - if (result["symbol"].lower() == token_input or - result["name"].lower() == token_input or - result["address"].lower() == token_input): + if ( + result["symbol"].lower() == token_input + or result["name"].lower() == token_input + or result["address"].lower() == token_input + ): matched_address = f"{result['address']}-{result['chain']}" break - # Use matched address or original input price_tokens = [matched_address] if matched_address else tokens_list code, output = self._call_ave_script( @@ -1067,11 +650,27 @@ class ConversationalAgent: if prices: price_text = "šŸ’° **Token Prices:**\n" for token_id, price_data in prices.items(): - price = price_data.get("price", "N/A") if isinstance(price_data, dict) else "N/A" - change_24h = price_data.get("token_price_change_24h", "N/A") if isinstance(price_data, dict) else "N/A" - mc = price_data.get("market_cap", "N/A") if isinstance(price_data, dict) else "N/A" + price = ( + price_data.get("price", "N/A") + if isinstance(price_data, dict) + else "N/A" + ) + change_24h = ( + price_data.get("token_price_change_24h", "N/A") + if isinstance(price_data, dict) + else "N/A" + ) + mc = ( + price_data.get("market_cap", "N/A") + if isinstance(price_data, dict) + else "N/A" + ) try: - price_str = f"${float(price):,.6f}" if price and price != "N/A" else price + price_str = ( + f"${float(price):,.6f}" + if price and price != "N/A" + else price + ) except (ValueError, TypeError): price_str = str(price) if price else "N/A" try: @@ -1121,15 +720,13 @@ class ConversationalAgent: def _execute_backtest_direct(self, message: str) -> Dict[str, Any]: """Execute backtest directly using token from strategy or message.""" - # Extract token address from message if provided parts = message.split() token_address = None timeframe = "1d" start_date = None end_date = None - # Parse arguments from message - for i, part in enumerate(parts[1:], 1): # Skip /backtest + for i, part in enumerate(parts[1:], 1): if part.startswith("0x") and len(part) > 20: token_address = part elif part in ["1d", "4h", "1h", "15m"]: @@ -1140,10 +737,10 @@ class ConversationalAgent: else: end_date = part - # If no token address in message, try to get from strategy if not token_address and self.bot_id: try: from ...core.database import get_db + db = next(get_db()) try: bot = db.query(Bot).filter(Bot.id == self.bot_id).first() @@ -1168,20 +765,17 @@ class ConversationalAgent: "success": True, } - # Look up token details first token_info = self._get_token_info(token_address) token_label = f"`{token_address}`" if token_info.get("symbol"): token_label = f"**{token_info['symbol']}** ({token_info.get('name', 'Unknown')}) - `{token_address}`" - # Execute backtest result = self._execute_backtest( token_address=token_address, timeframe=timeframe, start_date=start_date, end_date=end_date, ) - # Prepend token info to backtest result return { "response": f"šŸ“Š **Backtest for {token_label}**\n\n{result}", "thinking": None, @@ -1192,14 +786,12 @@ class ConversationalAgent: def _execute_simulate_direct(self, message: str) -> Dict[str, Any]: """Execute simulate directly using token from strategy or message.""" - # Extract parameters from message parts = message.split() action = None token_address = None kline_interval = "1m" - # Parse arguments from message - for i, part in enumerate(parts[1:], 1): # Skip /simulate + for i, part in enumerate(parts[1:], 1): if part in ["start", "stop", "status", "results"]: action = part elif part.startswith("0x") and len(part) > 20: @@ -1207,10 +799,10 @@ class ConversationalAgent: elif part in ["1m", "5m", "15m", "1h", "4h"]: kline_interval = part - # If no token in message and action is start, try to get from strategy if not token_address and self.bot_id and action == "start": try: from ...core.database import get_db + db = next(get_db()) try: bot = db.query(Bot).filter(Bot.id == self.bot_id).first() @@ -1244,7 +836,6 @@ class ConversationalAgent: "success": True, } - # Execute simulation result = self._manage_simulation( action=action, token_address=token_address, @@ -1261,29 +852,17 @@ class ConversationalAgent: def chat( self, user_message: str, conversation_history: List[Dict] = None ) -> Dict[str, Any]: - """Process a user message and return a structured response. - - Args: - user_message: The user's message - conversation_history: Optional list of previous messages - - Returns: - Dict with 'response', 'thinking', and 'strategy_updated' - """ + """Process a user message and return a structured response.""" try: - # Handle slash commands if user_message.startswith("/"): result = self._handle_slash_command(user_message) - # If None returned, it means a skill was recognized but has args - let AI handle it if result is not None: return result - # Check if there's a pending command from previous message if self.pending_command: pending = self.pending_command - self.pending_command = None # Clear pending + self.pending_command = None - # Auto-execute the pending command with user's message as param if pending == "search": return self._execute_search(user_message) elif pending == "risk": @@ -1293,52 +872,38 @@ class ConversationalAgent: elif pending == "price": return self._execute_price(user_message) - # Check for backtest/simulate with args in message if user_message.startswith("/backtest"): return self._execute_backtest_direct(user_message) elif user_message.startswith("/simulate"): return self._execute_simulate_direct(user_message) - # Build messages array with system prompt and conversation history - messages = [{"role": "system", "content": SYSTEM_PROMPT_WITH_TOOLS}] + messages = [{"role": "user", "content": user_message}] - # Add conversation history (last 10 messages) if conversation_history: for msg in conversation_history[-10:]: role = "assistant" if msg.get("role") == "assistant" else "user" - messages.append({"role": role, "content": msg.get("content", "")}) + messages.insert( + 0, {"role": role, "content": msg.get("content", "")} + ) - # Add current user message - messages.append({"role": "user", "content": user_message}) - - # Make API call to extended thinking endpoint - resp = requests.post( - self.thinking_endpoint, - headers={ - "Authorization": f"Bearer {self.api_key}", - "Content-Type": "application/json", - }, - json={ - "model": self.model, - "messages": messages, - "temperature": 0.7, - "max_tokens": 2000, - "thinking": {"type": "human", "budget_tokens": 1500}, - "tools": TOOLS, - }, + resp = self.client.chat( + messages=messages, + system_prompt=SYSTEM_PROMPT_WITH_TOOLS, + tools=TOOLS, ) - result = resp.json() or {} + result = resp - # Extract thinking from reasoning_content + # Initialize thinking to None to handle cases where API response + # doesn't have the expected message structure (intermittent issue) thinking = None + if result.get("choices") and len(result.get("choices", [])) > 0: choice = result["choices"][0] if "message" in choice: message = choice["message"] thinking = message.get("reasoning_content") - # Check for native function calls (tool_calls) tool_calls = message.get("tool_calls", []) if tool_calls: for tool_call in tool_calls: @@ -1364,7 +929,6 @@ class ConversationalAgent: if code == 0: try: data = json.loads(output) - # Handle both dict with 'tokens' key and direct list data_field = data.get("data", []) if isinstance(data_field, list): tokens = data_field @@ -1376,8 +940,11 @@ class ConversationalAgent: addr = t.get("token", "") symbol = t.get("symbol", "") name = t.get("name", "") - # Search API uses price_change_24h, trending uses token_price_change_24h - price_change = t.get("price_change_24h") or t.get("token_price_change_24h") or "N/A" + price_change = ( + t.get("price_change_24h") + or t.get("token_price_change_24h") + or "N/A" + ) mc = t.get("market_cap", "N/A") try: mc_str = f"${float(mc):,.0f}" @@ -1389,7 +956,7 @@ class ConversationalAgent: response_text = f"No tokens found for '{keyword}'. Try a different keyword." except json.JSONDecodeError: response_text = ( - f"Failed to parse search results." + "Failed to parse search results." ) else: response_text = f"Failed to search tokens: {output}" @@ -1413,47 +980,74 @@ class ConversationalAgent: try: data = json.loads(output) data_field = data.get("data") - # Handle both dict and list responses - token_data = data_field if isinstance(data_field, dict) else {} - # Token details may be nested in 'token' key + token_data = ( + data_field + if isinstance(data_field, dict) + else {} + ) token_info = token_data.get("token", token_data) - # Check if token has valid symbol/name (not None, not 'N/A') - symbol = token_info.get("symbol") or token_data.get("symbol") - name = token_info.get("name") or token_data.get("name") - if not symbol or symbol == "N/A" or not name or name == "N/A": + symbol = token_info.get( + "symbol" + ) or token_data.get("symbol") + name = token_info.get("name") or token_data.get( + "name" + ) + if ( + not symbol + or symbol == "N/A" + or not name + or name == "N/A" + ): response_text = f"Token not found for {address}. Raw response: {output[:500]}" else: - # Try different price field names - price = (token_info.get("current_price_usd") - or token_info.get("price_usd") - or token_info.get("price") - or token_data.get("price") - or "N/A") - mc = (token_info.get("market_cap") - or token_info.get("fdv") - or token_data.get("market_cap") - or "N/A") - vol = (token_info.get("tx_volume_u_24h") - or token_info.get("volume_24h") - or token_data.get("volume_24h") - or "N/A") - pairs = token_info.get("top_pairs") or token_data.get("top_pairs") or [] + price = ( + token_info.get("current_price_usd") + or token_info.get("price_usd") + or token_info.get("price") + or token_data.get("price") + or "N/A" + ) + mc = ( + token_info.get("market_cap") + or token_info.get("fdv") + or token_data.get("market_cap") + or "N/A" + ) + vol = ( + token_info.get("tx_volume_u_24h") + or token_info.get("volume_24h") + or token_data.get("volume_24h") + or "N/A" + ) + pairs = ( + token_info.get("top_pairs") + or token_data.get("top_pairs") + or [] + ) pairs_text = "" if pairs: pairs_text = "\n**Top Pairs:**\n" for p in pairs[:3]: - liq = p.get('liquidity', 'N/A') + liq = p.get("liquidity", "N/A") try: liq_str = f"${float(liq):,.0f}" except (ValueError, TypeError): liq_str = str(liq) pairs_text += f"- {p.get('pair', 'N/A')}: {liq_str} liquidity\n" try: - mc_str = f"${float(mc):,.0f}" if mc != "N/A" else "N/A" + mc_str = ( + f"${float(mc):,.0f}" + if mc != "N/A" + else "N/A" + ) except (ValueError, TypeError): mc_str = str(mc) try: - vol_str = f"${float(vol):,.0f}" if vol != "N/A" else "N/A" + vol_str = ( + f"${float(vol):,.0f}" + if vol != "N/A" + else "N/A" + ) except (ValueError, TypeError): vol_str = str(vol) response_text = f"**{symbol}** ({name})\n\nPrice: ${price}\nMarket Cap: {mc_str}\n24h Volume: {vol_str}{pairs_text}" @@ -1486,7 +1080,6 @@ class ConversationalAgent: try: data = json.loads(output) prices = data.get("data", {}) - # Ensure prices is a dict if not isinstance(prices, dict): prices = {} if prices: @@ -1535,8 +1128,11 @@ class ConversationalAgent: try: data = json.loads(output) data_field = data.get("data") - # Handle both dict and list responses - risk_data = data_field if isinstance(data_field, dict) else {} + risk_data = ( + data_field + if isinstance(data_field, dict) + else {} + ) if risk_data: is_honeypot = risk_data.get( "is_honeypot", "unknown" @@ -1544,38 +1140,53 @@ class ConversationalAgent: buy_tax = risk_data.get("buy_tax", 0) sell_tax = risk_data.get("sell_tax", 0) status = risk_data.get("status", "unknown") - # Convert is_honeypot to string for comparison if isinstance(is_honeypot, bool): - is_honeypot_str = str(is_honeypot).lower() + is_honeypot_str = str( + is_honeypot + ).lower() elif isinstance(is_honeypot, int): if is_honeypot == 1: is_honeypot_str = "true" elif is_honeypot == 0: is_honeypot_str = "false" else: - is_honeypot_str = "unknown" # -1 or other means couldn't determine + is_honeypot_str = "unknown" else: - is_honeypot_str = str(is_honeypot).lower() if is_honeypot else "unknown" - - # Format honeypot display value + is_honeypot_str = ( + str(is_honeypot).lower() + if is_honeypot + else "unknown" + ) + if is_honeypot_str == "unknown": - honeypot_display = "Unknown (could not determine)" + honeypot_display = ( + "Unknown (could not determine)" + ) else: honeypot_display = is_honeypot_str - # Convert tax values to float for comparison try: - buy_tax_val = float(buy_tax) if buy_tax not in (None, "N/A") else 0 + buy_tax_val = ( + float(buy_tax) + if buy_tax not in (None, "N/A") + else 0 + ) except (ValueError, TypeError): buy_tax_val = 0 try: - sell_tax_val = float(sell_tax) if sell_tax not in (None, "N/A") else 0 + sell_tax_val = ( + float(sell_tax) + if sell_tax not in (None, "N/A") + else 0 + ) except (ValueError, TypeError): sell_tax_val = 0 risk_text = ( f"**Risk Analysis for {address}**\n\n" ) risk_text += f"- Status: {status}\n" - risk_text += f"- Honeypot: {honeypot_display}\n" + risk_text += ( + f"- Honeypot: {honeypot_display}\n" + ) risk_text += f"- Buy Tax: {buy_tax}%\n" risk_text += f"- Sell Tax: {sell_tax}%\n" if is_honeypot_str == "true": @@ -1608,21 +1219,31 @@ class ConversationalAgent: code, output = self._call_ave_script( "trending", - ["--chain", chain, "--page-size", str(min(limit, 50))], + [ + "--chain", + chain, + "--page-size", + str(min(limit, 50)), + ], ) if code == 0: try: data = json.loads(output) data_field = data.get("data") - # Handle both dict and list responses - tokens = data_field if isinstance(data_field, list) else data_field.get("tokens", []) + tokens = ( + data_field + if isinstance(data_field, list) + else data_field.get("tokens", []) + ) if tokens: token_list = "" for t in tokens[:limit]: addr = t.get("token", "") symbol = t.get("symbol", "") name = t.get("name", "") - price_change = t.get("token_price_change_24h", "N/A") + price_change = t.get( + "token_price_change_24h", "N/A" + ) mc = t.get("market_cap", "N/A") try: mc_str = f"${float(mc):,.0f}" @@ -1635,7 +1256,9 @@ class ConversationalAgent: except json.JSONDecodeError: response_text = "Failed to parse trending data." else: - response_text = f"Failed to get trending tokens: {output}" + response_text = ( + f"Failed to get trending tokens: {output}" + ) return { "response": response_text, @@ -1651,7 +1274,6 @@ class ConversationalAgent: start_date = args.get("start_date") end_date = args.get("end_date") - # Execute backtest backtest_result = self._execute_backtest( token_address=token_address, timeframe=timeframe, @@ -1672,7 +1294,6 @@ class ConversationalAgent: token_address = args.get("token_address") kline_interval = args.get("kline_interval", "1m") - # Execute simulation management sim_result = self._manage_simulation( action=action, token_address=token_address, @@ -1687,24 +1308,20 @@ class ConversationalAgent: "success": True, } - # Get the main response content content = ( result.get("choices", [{}])[0].get("message", {}).get("content", "") ) - # Parse JSON from the content thinking_field = None response_text = content strategy_update = None - # Try to extract JSON from the content json_match = re.search( r"```(?:json)?\s*(\{.*?\})\s*```", content, re.DOTALL ) if json_match: json_str = json_match.group(1) else: - # Try to find JSON object directly json_match = re.search(r"\{.*\}", content, re.DOTALL) if json_match: json_str = json_match.group(0) @@ -1718,14 +1335,12 @@ class ConversationalAgent: response_text = parsed.get("response", content) strategy_update = parsed.get("strategy_update") - # Handle tool call tool_call = parsed.get("tool_call") if tool_call and tool_call.get("name") == "search_tokens": args = tool_call.get("arguments", {}) keyword = args.get("keyword", "") limit = args.get("limit", 10) - # Execute the tool using ave-cloud-skill CLI code, output = self._call_ave_script( "search", [ @@ -1740,7 +1355,6 @@ class ConversationalAgent: if code == 0: try: data = json.loads(output) - # Handle both dict with 'tokens' key and direct list data_field = data.get("data", []) if isinstance(data_field, list): tokens = data_field @@ -1772,25 +1386,21 @@ class ConversationalAgent: strategy_update = None except json.JSONDecodeError: - pass # Use defaults + pass - # Use the native thinking from API if available, otherwise use parsed thinking final_thinking = thinking or thinking_field - # Check if token_address is missing in strategy_update strategy_needs_confirmation = False token_search_results = None if strategy_update: - # Extract token name from conditions token_name = None - for cond in (strategy_update.get("conditions") or []): + for cond in strategy_update.get("conditions") or []: if not cond.get("token_address") and cond.get("token"): token_name = cond.get("token") strategy_needs_confirmation = True break - # Search for token if name is found if strategy_needs_confirmation and token_name: try: code, output = self._call_ave_script( @@ -1799,7 +1409,6 @@ class ConversationalAgent: ) if code == 0: data = json.loads(output) - # Handle both dict with 'tokens' key and direct list data_field = data.get("data", []) if isinstance(data_field, list): tokens = data_field @@ -1818,10 +1427,7 @@ class ConversationalAgent: except Exception as e: print(f"Token search error: {e}") - # Only update strategy if token_address is provided if strategy_update and strategy_needs_confirmation: - # Don't auto-save - user needs to confirm token address - # Return response but with strategy_update as None return { "response": response_text, "thinking": final_thinking, @@ -1832,7 +1438,6 @@ class ConversationalAgent: "success": True, } - # Update strategy in database if provided if strategy_update and self.bot_id: self._update_strategy(strategy_update) @@ -1872,18 +1477,15 @@ class ConversationalAgent: settings = get_settings() db = next(get_db()) - # Get the bot bot = db.query(Bot).filter(Bot.id == self.bot_id).first() if not bot: return "I couldn't find the bot. Please try again." - # Default dates if not provided (last 30 days) if not end_date: end_date = datetime.now().strftime("%Y-%m-%d") if not start_date: start_date = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d") - # Create backtest engine backtest_config = { "bot_id": self.bot_id, "token": token_address, @@ -1900,7 +1502,6 @@ class ConversationalAgent: engine = BacktestEngine(backtest_config) results = asyncio.run(engine.run()) - # Format results for display if "error" in results: return f"Backtest failed: {results['error']}" @@ -1911,7 +1512,6 @@ class ConversationalAgent: sharpe_ratio = results.get("sharpe_ratio", 0) final_balance = results.get("final_balance", 10000) - # Format return with emoji indicators return_emoji = "šŸ“ˆ" if total_return >= 0 else "šŸ“‰" return_str = ( f"+{total_return:.2f}%" if total_return >= 0 else f"{total_return:.2f}%" @@ -1957,7 +1557,6 @@ Would you like me to adjust the strategy parameters based on these results?""" settings = get_settings() try: - # Get the bot bot = db.query(Bot).filter(Bot.id == self.bot_id).first() if not bot: return "I couldn't find the bot. Please try again." @@ -1966,7 +1565,6 @@ Would you like me to adjust the strategy parameters based on these results?""" if not token_address: return "I need a token address to start a simulation. Which token would you like to simulate?" - # Check if there's already a running simulation running_sim = ( db.query(Simulation) .filter( @@ -1977,10 +1575,8 @@ Would you like me to adjust the strategy parameters based on these results?""" ) if running_sim: - # Stop the existing one first self._stop_simulation_db(running_sim.id) - # Create new simulation sim_id = str(uuid.uuid4()) simulation = Simulation( id=sim_id, @@ -1999,7 +1595,6 @@ Would you like me to adjust the strategy parameters based on these results?""" db.add(simulation) db.commit() - # Start the simulation in background sim_config = { "bot_id": self.bot_id, "token": token_address, @@ -2013,7 +1608,6 @@ Would you like me to adjust the strategy parameters based on these results?""" "initial_balance": 10000.0, } - # Run simulation in background thread def run_sim(): asyncio.run( self._run_simulation_sync( @@ -2028,7 +1622,6 @@ Would you like me to adjust the strategy parameters based on these results?""" return f"Started simulation on {token_address} using {kline_interval} klines. The simulation is running and will process up to 100 candles. Ask me for status or results anytime!" elif action == "stop": - # Find running simulation running_sim = ( db.query(Simulation) .filter( @@ -2043,7 +1636,6 @@ Would you like me to adjust the strategy parameters based on these results?""" self._stop_simulation_db(running_sim.id) - # Get final results portfolio = running_sim.portfolio or {} current_balance = portfolio.get("current_balance", 10000) initial_balance = portfolio.get("initial_balance", 10000) @@ -2055,7 +1647,6 @@ Would you like me to adjust the strategy parameters based on these results?""" return f"Simulation stopped!\n\nFinal Results:\nšŸ’° Final Balance: ${current_balance:,.2f}\nšŸ“ˆ P&L: {'+' if pnl >= 0 else ''}${pnl:,.2f} ({'+' if pnl_pct >= 0 else ''}{pnl_pct:.2f}%)\nšŸ“Š Trades: {len(running_sim.trade_log or [])}" elif action == "status": - # Find running simulation running_sim = ( db.query(Simulation) .filter( @@ -2088,7 +1679,6 @@ Would you like me to adjust the strategy parameters based on these results?""" return status elif action == "results": - # Find running or most recent simulation simulation = ( db.query(Simulation) .filter(Simulation.bot_id == self.bot_id) @@ -2124,7 +1714,7 @@ Would you like me to adjust the strategy parameters based on these results?""" if simulation.status == "running": results += ( - f"\nā³ Simulation still running... (refresh for latest)" + "\nā³ Simulation still running... (refresh for latest)" ) return results @@ -2220,14 +1810,17 @@ Would you like me to adjust the strategy parameters based on these results?""" def _call_ave_script(self, command: str, args: list) -> tuple[int, str]: """Call an ave-cloud-skill CLI script and return (status_code, stdout).""" - import json import os import subprocess from ...core.config import get_settings settings = get_settings() repo_root = os.path.dirname( - os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))) + os.path.dirname( + os.path.dirname( + os.path.dirname(os.path.dirname(os.path.dirname(__file__))) + ) + ) ) ave_skill_path = os.path.join( repo_root, "ave-cloud-skill", "scripts", "ave_data_rest.py" @@ -2246,7 +1839,6 @@ Would you like me to adjust the strategy parameters based on these results?""" env=env, timeout=30, ) - # Include stderr in output for debugging output = result.stdout if result.returncode != 0 and result.stderr: output = f"{output}\n{result.stderr}".strip() diff --git a/src/backend/app/services/ai_agent/client.py b/src/backend/app/services/ai_agent/client.py new file mode 100644 index 0000000..91a3a34 --- /dev/null +++ b/src/backend/app/services/ai_agent/client.py @@ -0,0 +1,308 @@ +"""MiniMax API client for the conversational agent.""" + +import requests +from typing import Dict, Any, Optional, List + + +SYSTEM_PROMPT = """You are a helpful AI trading assistant named Randebu. You help users manage their trading bots. + +IMPORTANT CHAIN LIMITATION: +- We ONLY support BSC (Binance Smart Chain) blockchain +- If user asks about any other chain (Solana, ETH, Base, etc.), respond with: "Currently we only support BSC (Binance Smart Chain). All trading strategies and token searches are performed on BSC." +- Never search or recommend tokens on other chains +- The search_tokens tool defaults to BSC, never change this + +Your response must be valid JSON with exactly this structure: +{ + "thinking": "Your internal reasoning and analysis (what you're thinking about)", + "response": "Your actual response to the user (be concise and helpful)", + "strategy_update": null or { + "conditions": [{"type": "price_drop" | "price_rise" | "volume_spike" | "price_level", "token": "TOKEN_SYMBOL", "token_address": null, "threshold": number, ...}], + "actions": [{"type": "buy" | "sell" | "hold", "amount_percent": number, ...}], + "risk_management": {"stop_loss_percent": number, "take_profit_percent": number} + } +} + +Guidelines: +- "thinking" should be detailed reasoning about the user's request +- "response" should be conversational and clear +- "strategy_update" should be populated ONLY when the user provides specific trading parameters (percentages, tokens, conditions, etc.) +- IMPORTANT: When a token is mentioned, set "token_address": null and ask user to confirm the token address before saving. Your response should say something like: "I need to confirm the token address. Could you provide the contract address for [TOKEN]?" +- If no strategy parameters are provided, set "strategy_update" to null +- Be friendly, concise, and helpful in your response + +Example 1 (no strategy update): +User: "What can this bot do?" +{ + "thinking": "The user is asking about the bot's capabilities. I should explain the main features.", + "response": "Randebu is your AI trading assistant! It can monitor cryptocurrency prices and execute trades based on your configured strategies. Tell me your trading parameters and I'll set them up for you.", + "strategy_update": null +} + +Example 2 (token needs confirmation): +User: "I want to buy PEPE when it drops 10%" +{ + "thinking": "User wants to buy PEPE. I need the token contract address to proceed. I should ask for confirmation.", + "response": "I'd be happy to set up a buy order for PEPE! However, I need to confirm the token contract address. Could you provide the BSC contract address for PEPE? (It usually starts with 0x...)", + "strategy_update": { + "conditions": [{"type": "price_drop", "token": "PEPE", "token_address": null, "threshold": 10}], + "actions": [{"type": "buy", "amount_percent": 100}], + "risk_management": null + } +} + +Example 3 (with token address provided by user): +User: "Buy 0x6982508145454Ce125dDE157d8d64a26D53f60a2 when it drops 10%" +{ + "thinking": "User provided a contract address, I can use it directly.", + "response": "Perfect! I've configured your strategy to buy the token when it drops 10%.", + "strategy_update": { + "conditions": [{"type": "price_drop", "token": "TOKEN", "token_address": "0x6982508145454Ce125dDE157d8d64a26D53f60a2", "threshold": 10}], + "actions": [{"type": "buy", "amount_percent": 100}], + "risk_management": null + } +}""" + + +TOOLS = [ + { + "type": "function", + "function": { + "name": "search_tokens", + "description": "Search for tokens by keyword on BSC blockchain. Use this when user asks to search for a specific token or find tokens by name/symbol.", + "parameters": { + "type": "object", + "properties": { + "keyword": { + "type": "string", + "description": "Token symbol or name to search for (e.g., 'PEPE', 'BTC')", + }, + "limit": { + "type": "integer", + "description": "Number of tokens to return (default: 10)", + "default": 10, + }, + }, + "required": ["keyword"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "get_token", + "description": "Get detailed information about a specific token including price, market cap, and pairs. Use when user asks for token details or wants to find a specific token by contract address.", + "parameters": { + "type": "object", + "properties": { + "address": { + "type": "string", + "description": "Token contract address (e.g., '0x6982508145454Ce125dDE157d8d64a26D53f60a2')", + }, + "chain": { + "type": "string", + "description": "Blockchain chain (default: bsc)", + "default": "bsc", + }, + }, + "required": ["address"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "get_price", + "description": "Get current price(s) for tokens. Use when user asks for token price or wants to compare prices of multiple tokens.", + "parameters": { + "type": "object", + "properties": { + "token_ids": { + "type": "string", + "description": "Comma-separated list of token IDs with chain suffix (e.g., 'PEPE-bsc,TRUMP-bsc')", + } + }, + "required": ["token_ids"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "get_risk", + "description": "Get risk analysis for a token contract. Use when user asks about token risk, honeypot analysis, or safety assessment before trading.", + "parameters": { + "type": "object", + "properties": { + "address": { + "type": "string", + "description": "Token contract address (e.g., '0x6982508145454Ce125dDE157d8d64a26D53f60a2')", + }, + "chain": { + "type": "string", + "description": "Blockchain chain (default: bsc)", + "default": "bsc", + }, + }, + "required": ["address"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "get_trending", + "description": "Get trending tokens on a blockchain. Use when user asks what's trending, top tokens, or popular tokens right now.", + "parameters": { + "type": "object", + "properties": { + "chain": { + "type": "string", + "description": "Blockchain chain (default: bsc)", + "default": "bsc", + }, + "limit": { + "type": "integer", + "description": "Number of trending tokens to return (default: 10, max: 50)", + "default": 10, + }, + }, + }, + }, + }, + { + "type": "function", + "function": { + "name": "run_backtest", + "description": "Run a backtest to evaluate how the current trading strategy would have performed historically. Returns key metrics like ROI, win rate, max drawdown, etc. Use this when user asks to backtest, test strategy, or check historical performance.", + "parameters": { + "type": "object", + "properties": { + "token_address": { + "type": "string", + "description": "The BSC contract address of the token to backtest (required)", + }, + "timeframe": { + "type": "string", + "description": "Timeframe for klines: '1d' (1 day), '4h' (4 hours), '1h' (1 hour), '15m' (15 minutes)", + "default": "1d", + }, + "start_date": { + "type": "string", + "description": "Start date for backtest in YYYY-MM-DD format (e.g., '2024-01-01')", + }, + "end_date": { + "type": "string", + "description": "End date for backtest in YYYY-MM-DD format (e.g., '2024-12-01')", + }, + }, + "required": ["token_address"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "manage_simulation", + "description": "Manage trading simulations: start, stop, or check status. Simulations run on real-time klines and show live portfolio updates. Use when user asks to run simulation, check simulation status, or stop simulation.", + "parameters": { + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["start", "stop", "status", "results"], + "description": "Action to perform: 'start' (begin new simulation), 'stop' (stop running simulation), 'status' (check if simulation is running), 'results' (get results from current or latest simulation)", + }, + "token_address": { + "type": "string", + "description": "Token contract address for simulation (required for 'start' action)", + }, + "kline_interval": { + "type": "string", + "description": "Kline interval: '1m', '5m', '15m', '1h' (default: '1m')", + "default": "1m", + }, + }, + "required": ["action"], + }, + }, + }, +] + +SYSTEM_PROMPT_WITH_TOOLS = ( + SYSTEM_PROMPT + + """ + +You have access to tools: +- search_tokens(keyword, limit): Search for tokens by keyword. Use it when user asks to search for a token or find tokens by name/symbol. +- get_token(address, chain): Get detailed information about a specific token. Use when user asks for token details. +- get_price(token_ids): Get current price(s) for tokens. Use when user asks for token price. +- get_risk(address, chain): Get risk analysis for a token. Use when user asks about token safety or honeypot analysis. +- get_trending(chain, limit): Get trending tokens on a blockchain. Use when user asks what's trending, top tokens, or popular tokens. +- run_backtest(token_address, timeframe, start_date, end_date): Run a backtest on historical data. Returns performance metrics. Use when user asks to backtest or check historical performance. +- manage_simulation(action, token_address, kline_interval): Manage trading simulations. Actions: 'start' (begin new), 'stop' (stop running), 'status' (check if running), 'results' (get current/latest results). + +When you want to use a tool, respond with: +{ + "thinking": "...", + "response": "Running backtest...", + "tool_call": {"name": "run_backtest", "arguments": {"token_address": "0x...", "timeframe": "1d", "start_date": "2024-01-01", "end_date": "2024-12-01"}} +} +""" +) + + +class MiniMaxClient: + """Client for MiniMax extended thinking API.""" + + def __init__(self, api_key: str, model: str = "MiniMax-M2.7"): + self.api_key = api_key + self.model = model + self.endpoint = "https://api.minimax.io/v1/text/chatcompletion_v2" + + def chat( + self, + messages: List[Dict[str, str]], + system_prompt: str, + tools: Optional[List[Dict[str, Any]]] = None, + temperature: float = 0.7, + max_tokens: int = 2000, + thinking_budget: int = 1500, + ) -> Dict[str, Any]: + """Send a chat request to MiniMax API.""" + headers = { + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + } + + all_messages = [{"role": "system", "content": system_prompt}] + messages + + payload = { + "model": self.model, + "messages": all_messages, + "temperature": temperature, + "max_tokens": max_tokens, + "thinking": {"type": "human", "budget_tokens": thinking_budget}, + } + + if tools: + payload["tools"] = tools + + resp = requests.post(self.endpoint, headers=headers, json=payload) + return resp.json() or {} + + def check_connection(self) -> bool: + """Check if API is reachable.""" + try: + resp = requests.post( + self.endpoint, + headers={"Authorization": f"Bearer {self.api_key}"}, + json={ + "model": self.model, + "messages": [{"role": "user", "content": "ping"}], + }, + timeout=10, + ) + return resp.status_code == 200 + except Exception: + return False diff --git a/src/backend/app/services/ai_agent/help.py b/src/backend/app/services/ai_agent/help.py new file mode 100644 index 0000000..68f8896 --- /dev/null +++ b/src/backend/app/services/ai_agent/help.py @@ -0,0 +1,83 @@ +"""Help formatters for slash commands and tool documentation.""" + +from typing import Optional +from .tools import get_tool_registry, SKILL_EMOJIS + + +def format_tools_list() -> str: + """Format the tool registry as a help message.""" + message = "šŸ“‹ Available Tools\n\n" + + for category in ["randebu", "ave"]: + tools = get_tool_registry().get(category, []) + if category == "randebu": + message += "šŸ¤– Randebu Built-in:\n" + else: + message += "ā˜ļø AVE Cloud Skills:\n" + + for tool in tools: + message += f" ā€¢ {tool['command']} - {tool['description']}\n" + message += "\n" + + message = ( + message.rstrip() + "\n\nType / for detailed help on a specific tool." + ) + return message + + +def format_skill_acknowledgment(tool_name: str, description: str) -> str: + """Format a brief acknowledgment when a skill is activated.""" + emoji = SKILL_EMOJIS.get(tool_name.lower(), "āœØ") + return f"{emoji} **{tool_name}** loaded. Ready for *{description}*, ask me away!" + + +def format_tool_help(tool_name: str) -> str: + """Format detailed help for a specific tool.""" + tool_name = tool_name.lstrip("/") + + for category in ["randebu", "ave"]: + for tool in get_tool_registry().get(category, []): + if tool["name"].lower() == tool_name.lower(): + cat_label = ( + "Randebu Built-in" if category == "randebu" else "AVE Cloud Skill" + ) + details = tool["details"] + message = ( + f"šŸ” {tool['command']} - {details['description']} ({cat_label})\n\n" + ) + message += f"**Description:** {details['description']}\n" + message += f"**Commands:**\n {details['usage']}\n\n" + message += f"**Example:**\n```\n{details['example']}\n```" + return message + + return f"Tool '{tool_name}' not found. Type / to see all available tools." + + +def format_general_help() -> str: + """Format general help about Randebu.""" + return """šŸ¤– **Randebu - AI Trading Assistant** + +Randebu is your AI trading assistant that helps you manage your trading bots on BSC (Binance Smart Chain). + +**Getting Started:** +1. Create a bot on the dashboard +2. Describe your trading strategy in plain English +3. Run backtests to validate your strategy +4. Start simulations to see live trading + +**Example Strategies:** +- "Buy PEPE when it drops 5%" +- "Sell if price rises 10% within 1 hour" +- "Buy when volume spikes by 200%" + +**Slash Commands:** +- `/` - Show all available tools +- `/help` - Show this help message +- `/` - Get help on a specific tool + +**Natural Language:** +You can also just describe what you want in natural language. For example: +- "What's the price of PEPE?" +- "Run a backtest on 0x... token" +- "Start a simulation on TRUMP" +""" diff --git a/src/backend/app/services/ai_agent/tools.py b/src/backend/app/services/ai_agent/tools.py new file mode 100644 index 0000000..fa714aa --- /dev/null +++ b/src/backend/app/services/ai_agent/tools.py @@ -0,0 +1,128 @@ +"""Tool registry and definitions for the conversational agent.""" + +from typing import Dict, Any, List + +TOOL_REGISTRY: Dict[str, Any] = { + "randebu": [ + { + "name": "backtest", + "description": "Run strategy backtest", + "category": "Randebu Built-in", + "command": "/backtest", + "details": { + "description": "Run a backtest to evaluate how the current trading strategy would have performed historically.", + "usage": "/backtest [token_address] [--timeframe 1d|4h|1h|15m] [--start YYYY-MM-DD] [--end YYYY-MM-DD]", + "example": "Run a backtest on PEPE for the last 30 days", + }, + }, + { + "name": "simulate", + "description": "Start/stop simulation", + "category": "Randebu Built-in", + "command": "/simulate", + "details": { + "description": "Start or stop trading simulations that run on real-time klines.", + "usage": "/simulate start|stop|status|results [token_address]", + "example": "Start a simulation on PEPE", + }, + }, + { + "name": "strategy", + "description": "View/update strategy", + "category": "Randebu Built-in", + "command": "/strategy", + "details": { + "description": "View your current trading strategy or update it with new parameters.", + "usage": "Describe your strategy in plain English, e.g., 'Buy PEPE when price drops 5%'", + "example": "Buy PEPE when it drops 10% within 1 hour", + }, + }, + ], + "ave": [ + { + "name": "search", + "description": "Token search", + "category": "AVE Cloud Skills", + "command": "/search", + "details": { + "description": "Find tokens by keyword, symbol, or contract address on BSC.", + "usage": "search [--chain bsc] [--limit 20]", + "example": "search PEPE\nsearch 0x1234... --chain bsc", + }, + }, + { + "name": "trending", + "description": "Popular tokens", + "category": "AVE Cloud Skills", + "command": "/trending", + "details": { + "description": "Get list of trending/popular tokens on BSC.", + "usage": "trending [--chain bsc] [--limit 20]", + "example": "trending --chain bsc\ntrending --limit 10", + }, + }, + { + "name": "risk", + "description": "Honeypot detection", + "category": "AVE Cloud Skills", + "command": "/risk", + "details": { + "description": "Get risk analysis for a token contract including honeypot assessment.", + "usage": "risk [--chain bsc]", + "example": "risk 0x6982508145454Ce125dDE157d8d64a26D53f60a2", + }, + }, + { + "name": "token", + "description": "Token details", + "category": "AVE Cloud Skills", + "command": "/token", + "details": { + "description": "Get detailed information about a specific token including price, market cap, and pairs.", + "usage": "token
[--chain bsc]", + "example": "token 0x6982508145454Ce125dDE157d8d64a26D53f60a2", + }, + }, + { + "name": "price", + "description": "Batch prices", + "category": "AVE Cloud Skills", + "command": "/price", + "details": { + "description": "Get current price(s) for multiple tokens.", + "usage": "price ,,... (e.g., PEPE-bsc,TRUMP-bsc)", + "example": "price PEPE-bsc,TRUMP-bsc", + }, + }, + ], +} + +SKILL_EMOJIS: Dict[str, str] = { + "backtest": "šŸ“Š", + "simulate": "šŸŽ®", + "strategy": "šŸ“", + "search": "šŸ”", + "trending": "šŸ“ˆ", + "risk": "šŸ“‰", + "token": "šŸŖ™", + "price": "šŸ’°", +} + + +def get_tool_registry() -> Dict[str, Any]: + """Return the tool registry for slash command help.""" + return TOOL_REGISTRY + + +def get_tools_by_category(category: str) -> List[Dict[str, Any]]: + """Get tools filtered by category.""" + return TOOL_REGISTRY.get(category, []) + + +def get_tool_by_name(tool_name: str) -> Dict[str, Any]: + """Get a tool by its name.""" + for category in ["randebu", "ave"]: + for tool in TOOL_REGISTRY.get(category, []): + if tool["name"].lower() == tool_name.lower(): + return tool + return None