import os import gradio as gr import importlib import asyncio import openai import pandas as pd import hype_accounts_server from datetime import datetime from memory_utils import load_memories, save_memory, load_memories_df from agents.mcp import MCPServerStdio from agents import Agent, Runner, trace, Tool importlib.reload(hype_accounts_server) openai.api_key = os.getenv("OPENAI_API_KEY") hyperliquid_trader_mcp_server_params = [{"command": "python", "args": ["-u", "hype_accounts_server.py"], "env": {"HYPERLIQUID_API_KEY": os.getenv("HYPERLIQUID_API_KEY"), "HYPERLIQUID_PRIVATE_KEY": os.getenv("HYPERLIQUID_PRIVATE_KEY"), "HYPERLIQUID_ACCOUNT_ADDRESS": os.getenv("HYPERLIQUID_ACCOUNT_ADDRESS")}}] crypto_news_mcp_server_params = [{"command": "python", "args": ["-u", "cryptopanic_news_server.py"], "env": {"CRYPTOPANIC_API_KEY": os.getenv("CRYPTOPANIC_API_KEY")}}] technical_analyst_mcp_server_params = [{"command": "node", "args": ["crypto-indicators-mcp/index.js"], "env": {"EXCHANGE_NAME": "binance"}}] hyperliquid_trader_mcp_servers = [MCPServerStdio(params, client_session_timeout_seconds=30) for params in hyperliquid_trader_mcp_server_params] crypto_news_mcp_servers = [MCPServerStdio(params, client_session_timeout_seconds=30) for params in crypto_news_mcp_server_params] technical_analyst_mcp_servers = [MCPServerStdio(params, client_session_timeout_seconds=30) for params in technical_analyst_mcp_server_params] mcp_servers = hyperliquid_trader_mcp_servers + crypto_news_mcp_servers + technical_analyst_mcp_servers # putting agents together async def get_crypto_news_researcher(mcp_servers) -> Agent: instructions = f"""You are a cryptocurrency researcher. You have the tools to search for interesting cryptocurrency news, look for possible trading opportunities, and help with research. Default None if there are no specific cryptocurrency news or opportunities in the user request. Otherwise, indicate the symbol of specific cryptocurrency if there are specific requests. eg. 'HYPE', 'BTC', 'ETH', 'XRP', etc. Based on the request, you carry out necessary research and respond with your findings. Take time to make multiple searches to get a comprehensive overview, and then summarize your findings. If there isn't a specific request, then just respond with long/short opportunities based on searching latest news. The current datetime is {datetime.now().strftime("%Y-%m-%d %H:%M:%S")} """ researcher = Agent( name="Crypto news researcher", instructions=instructions, model="gpt-4.1-mini", mcp_servers=mcp_servers, ) return researcher async def get_crypto_news_researcher_tool(mcp_servers) -> Tool: researcher = await get_crypto_news_researcher(mcp_servers) return researcher.as_tool( tool_name="crypto_news_researcher", tool_description="This tool researches online for cryptocurrency news and opportunities, \ either based on your specific request to look into a certain cryptocurrency, \ or generally for notable cryptocurrency news and opportunities. \ Describe what kind of research you're looking for." ) async def get_technical_analyst_researcher(mcp_servers) -> Agent: instructions = """You are a cryptocurrency perpetuals technical trading researcher. You have the tools for trend/momentum/volatility/volume technical indicators and strategies. Default analysis interval is 1h, and default lookback period is 36. Default indicators/strategies you have are: - EMA (20, 200) - MACD (12, 26, 9) - stochastic RSI (14, 14, 3, 3) - Accumulation/Distribution (ADL) - Volume Based on the indicators, Look for possible long/short opportunities, and come up with a strategy. The current datetime is {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}""" researcher = Agent( name="Crypto technical researcher", instructions=instructions, model="gpt-4.1-mini", mcp_servers=mcp_servers, ) return researcher async def get_technical_analyst_researcher_tool(mcp_servers) -> Tool: researcher = await get_crypto_news_researcher(mcp_servers) return researcher.as_tool( tool_name="crypto_technical_researcher", tool_description="This tool researches technical indicators for trading opportunities, \ either based on your specific request to look into a certain cryptocurrency or indicator, \ or generally for notable cryptocurrency news and technical indicators EMA, MACD, stochastic RSI, Accumulation/Distribution (ADL), Volume. \ Describe what cryptocurrency and indicators you're looking for." ) async def run_trader(): """Run one trading cycle and update memory.""" try: account_details = await hype_accounts_server.get_account_details() past_memories = load_memories(3) memory_text = "\n".join(past_memories) if past_memories else "No past insights." instructions = f""" You are a cryptocurrency perpetuals trader. Past Insights: {memory_text} Your current holdings and balance is: {account_details} You have the tools to perform a search for relevant cryptocurrency news. You have tools to check cryptocurrency technical indicators and prices. You have tools to check your current holdings and balance. You have tools to long and short cryptocurrency perpetuals. Please make use of these tools to manage your portfolio. Carry out trades as you see fit; do not wait for instructions or ask for confirmation. """ prompt = f""" Use your tools to make decisions about your portfolio. Step 1: look into your current holdings and balance. Step 2: look into the latest cryptocurrency news and relevant long/short opportunities. select 3-5 cryptocurrencies to focus on. Step 3: look into the current price, and technical indicators for these cryptocurrencies. Recommend a few trading strategies. Step 4: make trades to long/short these cryptocurrencies based on your research and strategies. You do not have to make any trades if there are no good opportunities. Respond with detailed information and analysis for each step - 1) news research 2) detailed technical analysis with indicator values 3) trade strategy 4) trade executions. The current datetime is {datetime.now().strftime("%Y-%m-%d %H:%M")} """ # Connect servers for server in mcp_servers: await server.connect() # Load tools crypto_news_researcher_tool = await get_crypto_news_researcher_tool(crypto_news_mcp_servers) technical_analyst_researcher_tool = await get_technical_analyst_researcher_tool(technical_analyst_mcp_servers) trader = Agent( name="crypto_trader", instructions=instructions, tools=[crypto_news_researcher_tool, technical_analyst_researcher_tool], mcp_servers=hyperliquid_trader_mcp_servers, model="gpt-4.1-mini", ) with trace("crypto_trader"): result = await Runner.run(trader, prompt, max_turns=30) # Summarize and store save_memory(result.final_output) # Build the latest DF (e.g., show last 10 entries) mem_df = load_memories_df(10) if len(mem_df) == 0: mem_df = pd.DataFrame(columns=["ts","summary"]) output = f"""### ✅ Trade Completed {datetime.now().strftime("%Y-%m-%d %H:%M")}\n\n{result.final_output}""" return output, mem_df except Exception as e: return f"❌ Error during trading cycle: {e}", pd.DataFrame(columns=["ts","summary"]) with gr.Blocks() as demo: gr.Markdown("# 💹 Crypto Trader Agent") gr.Markdown( """ [agent tracer](https://platform.openai.com/logs?api=traces)\n [hyperliquid platform](https://app.hyperliquid.xyz/trade) """, ) out = gr.Markdown() mem_table = gr.Dataframe( headers=["ts", "summary"] ) run_btn = gr.Button("▶️ Run Trading Cycle Now") stop_btn = gr.Button("🛑 Stop Auto Trading") resume_btn = gr.Button("🔄 Resume Auto Trading") # Create timer but keep reference timer = gr.Timer(1800) timer_active = gr.State(True) # track if currently running # --- Functions --- async def run_and_display(): return await run_trader() def stop_auto_trading(timer_active): timer_active = False timer.stop() # stop the 30-min loop return "🛑 Auto trading stopped.", timer_active def resume_auto_trading(timer_active): timer_active = True timer.resume() # restart the loop return "🔄 Auto trading resumed.", timer_active # --- Button bindings --- run_btn.click(run_and_display, outputs=[out, mem_table]) stop_btn.click(stop_auto_trading, inputs=[timer_active], outputs=[out, timer_active]) resume_btn.click(resume_auto_trading, inputs=[timer_active], outputs=[out, timer_active]) # 🔁 Auto-run every 30 minutes (default ON) timer.tick(run_and_display, outputs=[out, mem_table]) if __name__ == "__main__": demo.launch()