app.py

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()