The AI Hedge Fund: Running 50 Specialized Trading Agents in Parallel

import asyncio, httpx
from dataclasses import dataclass

PURPLE_FLEA_KEY = "your-api-key"

@dataclass
class TradeSignal:
    symbol: str
    side: str           # "long" or "short"
    confidence: float   # 0.0 - 1.0
    size_usd: float
    strategy: str

class HedgeFundOrchestrator:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.portfolio_var_limit = 0.02   # max 2% daily VaR
        self.max_single_position = 0.05   # max 5% in one asset
        self.total_capital = 100_000      # $100k AUM
        self.strategies = {}

    async def run_signal_agents(self) -> list[TradeSignal]:
        """Run all signal generators in parallel."""
        tasks = [
            self.momentum_agent(["BTC", "ETH", "SOL"]),
            self.mean_reversion_agent(["AVAX", "ARB", "OP"]),
            self.funding_rate_agent(["BTC", "ETH"]),
            self.sentiment_agent(["DOGE", "PEPE", "WIF"]),
            self.onchain_flow_agent(["BTC", "ETH"]),
        ]
        signal_lists = await asyncio.gather(*tasks)
        # Flatten nested lists
        return [s for signals in signal_lists for s in signals]

    async def risk_filter(self, signals: list[TradeSignal]) -> list[TradeSignal]:
        """Risk manager vetoes signals that breach portfolio limits."""
        approved = []
        current_exposure = await self.get_current_exposure()

        for signal in signals:
            # Check position limit
            current_pos = current_exposure.get(signal.symbol, 0)
            new_pos = (current_pos + signal.size_usd) / self.total_capital

            if new_pos > self.max_single_position:
                # Reduce size to stay within limit
                max_size = (self.max_single_position * self.total_capital) - current_pos
                signal.size_usd = max(0, max_size)

            # Only pass signals with sufficient confidence
            if signal.confidence >= 0.65 and signal.size_usd > 100:
                approved.append(signal)

        print(f"Risk filter: {len(signals)} signals → {len(approved)} approved")
        return approved

    async def execute_trades(self, approved_signals: list[TradeSignal]):
        """Execute approved signals in parallel."""
        tasks = [
            self.execute_single_trade(signal)
            for signal in approved_signals
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        successes = [r for r in results if not isinstance(r, Exception)]
        failures = [r for r in results if isinstance(r, Exception)]

        print(f"Executed: {len(successes)} trades, {len(failures)} failed")
        return successes

    async def execute_single_trade(self, signal: TradeSignal) -> dict:
        async with httpx.AsyncClient() as client:
            r = await client.post(
                "https://purpleflea.com/api/trade",
                json={
                    "symbol": signal.symbol,
                    "side": signal.side,
                    "size_usd": signal.size_usd,
                    "leverage": 1.0  # hedge fund uses conservative leverage
                },
                headers={"X-API-Key": self.api_key}
            )
            return r.json()

    async def run_cycle(self):
        """One complete fund cycle: generate → filter → execute."""
        print("\n--- Fund Cycle Start ---")
        signals = await self.run_signal_agents()
        print(f"Generated {len(signals)} signals")

        approved = await self.risk_filter(signals)
        results = await self.execute_trades(approved)

        print(f"--- Cycle Complete: {len(results)} trades executed ---\n")

# Run the hedge fund continuously
fund = HedgeFundOrchestrator(PURPLE_FLEA_KEY)
while True:
    asyncio.run(fund.run_cycle())
    asyncio.get_event_loop().run_until_complete(asyncio.sleep(300))  # 5 min cycle