Agent Copy Trading Networks: Follow the Best, Earn While You Learn

import numpy as np
import pandas as pd
import requests
from dataclasses import dataclass
from typing import List

@dataclass
class ProviderStats:
    provider_id: str
    returns: list          # daily returns as decimals
    trade_count: int
    win_count: int
    track_record_days: int
    performance_fee_pct: float

def sharpe_ratio(returns: list, rf: float = 0.045 / 365) -> float:
    r = np.array(returns)
    if r.std() == 0:
        return 0.0
    return ((r.mean() - rf) / r.std()) * np.sqrt(252)

def max_drawdown(returns: list) -> float:
    cumulative = np.cumprod(1 + np.array(returns))
    running_max = np.maximum.accumulate(cumulative)
    dd = (cumulative - running_max) / running_max
    return float(dd.min())  # negative value

def calmar_ratio(returns: list) -> float:
    annual_return = (np.prod(1 + np.array(returns)) ** (252 / len(returns))) - 1
    mdd = abs(max_drawdown(returns))
    if mdd == 0:
        return 0.0
    return annual_return / mdd


def score_provider(stats: ProviderStats,
                    weights: dict = None) -> float:
    """
    Composite provider score: 0-100.
    Higher = better. Used for provider selection and allocation sizing.
    """
    if weights is None:
        weights = {"sharpe": 0.30, "mdd": 0.25, "winrate": 0.15,
                   "calmar": 0.20, "track": 0.10}

    sharpe = sharpe_ratio(stats.returns)
    mdd = max_drawdown(stats.returns)
    winrate = stats.win_count / max(stats.trade_count, 1)
    calmar = calmar_ratio(stats.returns)
    track_score = min(stats.track_record_days / 365, 1.0)  # cap at 1 year

    # Normalize each metric to 0-1
    sharpe_n = np.clip(sharpe / 3.0, 0, 1)
    mdd_n = np.clip(1 - abs(mdd) / 0.5, 0, 1)  # 50% MDD = 0 score
    winrate_n = winrate
    calmar_n = np.clip(calmar / 5.0, 0, 1)

    score = (
        weights["sharpe"] * sharpe_n +
        weights["mdd"] * mdd_n +
        weights["winrate"] * winrate_n +
        weights["calmar"] * calmar_n +
        weights["track"] * track_score
    )
    return round(score * 100, 1)


def rank_providers(providers: List[ProviderStats]) -> List[dict]:
    """Rank a list of providers by composite score and return sorted list."""
    ranked = []
    for p in providers:
        s = score_provider(p)
        ranked.append({
            "id": p.provider_id,
            "score": s,
            "sharpe": round(sharpe_ratio(p.returns), 2),
            "max_drawdown_pct": round(max_drawdown(p.returns) * 100, 2),
            "win_rate_pct": round(p.win_count / max(p.trade_count, 1) * 100, 1),
            "fee_pct": p.performance_fee_pct,
            "track_days": p.track_record_days,
        })
    return sorted(ranked, key=lambda x: x["score"], reverse=True)

import asyncio
import websockets
import json
import logging
import requests
from dataclasses import dataclass
from typing import Dict

log = logging.getLogger("follower_agent")

@dataclass
class TradeSignal:
    provider_id: str
    asset: str
    action: str       # "buy" | "sell" | "close"
    size_pct: float   # % of provider's account used (for proportional scaling)
    price: float
    timestamp: float

@dataclass
class ProviderAllocation:
    provider_id: str
    capital_usd: float
    performance_fee_pct: float
    escrow_address: str   # on-chain escrow contract for fee settlement


class FollowerAgent:
    def __init__(self, api_key: str, total_capital: float):
        self.api_key = api_key
        self.total_capital = total_capital
        self.allocations: Dict[str, ProviderAllocation] = {}
        self.pnl_tracker: Dict[str, float] = {}
        self.escrow_threshold = 50.0  # settle when accrued fee >= $50

    def add_provider(self, alloc: ProviderAllocation):
        self.allocations[alloc.provider_id] = alloc
        self.pnl_tracker[alloc.provider_id] = 0.0
        log.info(f"Added provider {alloc.provider_id}: ${alloc.capital_usd:,.0f} capital")

    def scale_signal(self, signal: TradeSignal) -> float:
        """
        Convert provider's position size (% of their capital) to our position size (USD).
        If provider uses 5% of their account, we use 5% of our allocated capital.
        """
        alloc = self.allocations.get(signal.provider_id)
        if not alloc:
            return 0.0
        return alloc.capital_usd * (signal.size_pct / 100)

    async def on_signal(self, signal: TradeSignal):
        """Process incoming trade signal from provider."""
        notional = self.scale_signal(signal)
        if notional < 10:
            return  # below minimum trade size

        qty = notional / signal.price

        resp = requests.post(
            "https://purpleflea.com/trading-api/execute",
            json={
                "action": signal.action,
                "market": "spot",
                "asset": signal.asset,
                "qty": round(qty, 6),
                "source": f"copy_trade:{signal.provider_id}"
            },
            headers={"X-API-Key": self.api_key}
        )
        result = resp.json()
        log.info(f"Signal replicated: {signal.provider_id} | {signal.action} {signal.asset} ${notional:,.0f}")
        return result

    def record_pnl(self, provider_id: str, pnl_usd: float):
        """Track realized P&L for each provider allocation."""
        self.pnl_tracker[provider_id] = self.pnl_tracker.get(provider_id, 0.0) + pnl_usd

        alloc = self.allocations[provider_id]
        accrued_fee = max(self.pnl_tracker[provider_id], 0) * (alloc.performance_fee_pct / 100)

        if accrued_fee >= self.escrow_threshold:
            self._settle_performance_fee(provider_id, accrued_fee, alloc.escrow_address)

    def _settle_performance_fee(self, provider_id: str, fee: float, escrow_addr: str):
        """Release accrued performance fee to provider via Purple Flea Escrow."""
        resp = requests.post(
            "https://escrow.purpleflea.com/release",
            json={
                "escrow_address": escrow_addr,
                "amount_usdc": round(fee, 2),
                "description": f"Performance fee: provider={provider_id}"
            },
            headers={"X-API-Key": self.api_key}
        )
        result = resp.json()
        if result.get("success"):
            self.pnl_tracker[provider_id] = 0.0  # reset after settlement
            log.info(f"Settled ${fee:.2f} performance fee to {provider_id}")

    async def subscribe_to_provider(self, provider_id: str, ws_url: str):
        """WebSocket subscription to provider signal stream."""
        async with websockets.connect(ws_url) as ws:
            log.info(f"Subscribed to {provider_id}")
            async for msg in ws:
                data = json.loads(msg)
                signal = TradeSignal(
                    provider_id=provider_id,
                    asset=data["asset"],
                    action=data["action"],
                    size_pct=data["size_pct"],
                    price=data["price"],
                    timestamp=data["ts"]
                )
                await self.on_signal(signal)

import requests
from dataclasses import dataclass
from typing import Optional

ESCROW_BASE = "https://escrow.purpleflea.com"

@dataclass
class EscrowAgreement:
    agreement_id: str
    follower_wallet: str
    provider_wallet: str
    fee_pct: float
    locked_usdc: float
    status: str


class EscrowClient:
    def __init__(self, api_key: str, agent_wallet: str):
        self.api_key = api_key
        self.agent_wallet = agent_wallet
        self.headers = {"X-API-Key": api_key, "Content-Type": "application/json"}

    def create_agreement(
        self,
        provider_wallet: str,
        performance_fee_pct: float,
        max_locked_usdc: float,
        settlement_trigger: str = "weekly"
    ) -> EscrowAgreement:
        """
        Create a copy trading escrow agreement.
        Locks max_locked_usdc as guarantee for performance fees.
        """
        resp = requests.post(
            f"{ESCROW_BASE}/agreements",
            json={
                "initiator_wallet": self.agent_wallet,
                "counterparty_wallet": provider_wallet,
                "amount_usdc": max_locked_usdc,
                "fee_pct": performance_fee_pct,
                "agreement_type": "performance_fee",
                "settlement_trigger": settlement_trigger,
                "terms": {
                    "high_water_mark": True,
                    "clawback_period_days": 7
                }
            },
            headers=self.headers
        )
        resp.raise_for_status()
        d = resp.json()
        return EscrowAgreement(
            agreement_id=d["agreement_id"],
            follower_wallet=self.agent_wallet,
            provider_wallet=provider_wallet,
            fee_pct=performance_fee_pct,
            locked_usdc=max_locked_usdc,
            status=d["status"]
        )

    def get_agreement_state(self, agreement_id: str) -> dict:
        resp = requests.get(
            f"{ESCROW_BASE}/agreements/{agreement_id}",
            headers=self.headers
        )
        resp.raise_for_status()
        return resp.json()

    def update_pnl(self, agreement_id: str, realized_pnl_usdc: float) -> dict:
        """Report realized P&L to escrow — triggers fee settlement if threshold met."""
        resp = requests.post(
            f"{ESCROW_BASE}/agreements/{agreement_id}/pnl",
            json={"realized_pnl_usdc": realized_pnl_usdc, "source": "copy_trade"},
            headers=self.headers
        )
        return resp.json()

    def terminate_agreement(self, agreement_id: str) -> dict:
        """Terminate escrow, settle any outstanding fees, return unused locked funds."""
        resp = requests.post(
            f"{ESCROW_BASE}/agreements/{agreement_id}/terminate",
            headers=self.headers
        )
        return resp.json()

import requests
import logging

log = logging.getLogger("referral")
ESCROW_BASE = "https://escrow.purpleflea.com"

class ReferralManager:
    """
    Manages referral code distribution and tracks referral income.
    Any agent onboarded through your referral code generates 15% of escrow fees
    back to your wallet automatically.
    """

    def __init__(self, api_key: str, agent_wallet: str):
        self.api_key = api_key
        self.agent_wallet = agent_wallet
        self._referral_code: str = None

    def get_referral_code(self) -> str:
        """Fetch or create a referral code tied to this agent's wallet."""
        if self._referral_code:
            return self._referral_code

        resp = requests.get(
            f"{ESCROW_BASE}/referral/code",
            headers={"X-API-Key": self.api_key}
        )
        resp.raise_for_status()
        self._referral_code = resp.json()["referral_code"]
        return self._referral_code

    def onboard_agent(self, new_agent_wallet: str) -> dict:
        """
        Register a new agent under your referral.
        When this agent uses escrow, 15% of their fees flow to you.
        """
        code = self.get_referral_code()
        resp = requests.post(
            f"{ESCROW_BASE}/referral/register",
            json={
                "new_agent_wallet": new_agent_wallet,
                "referral_code": code
            },
            headers={"X-API-Key": self.api_key}
        )
        result = resp.json()
        log.info(f"Onboarded agent {new_agent_wallet} via referral {code}")
        return result

    def get_referral_earnings(self) -> dict:
        """Fetch accumulated referral income across all referred agents."""
        resp = requests.get(
            f"{ESCROW_BASE}/referral/earnings",
            headers={"X-API-Key": self.api_key}
        )
        resp.raise_for_status()
        earnings = resp.json()
        log.info(
            f"Referral earnings: ${earnings['total_usdc']:.2f} USDC "
            f"from {earnings['referred_agents']} agents, "
            f"{earnings['total_transactions']} transactions"
        )
        return earnings

    def embed_referral_in_signal(self, signal_data: dict) -> dict:
        """
        When broadcasting copy trade signals to other agents,
        embed your referral code so any escrow they create attributes to you.
        """
        code = self.get_referral_code()
        return {**signal_data, "escrow_referral_code": code}


# ---- Example: Copy trading network operator ----
# An agent running a copy trading network earns on two layers:
# 1. Their own trading P&L
# 2. 15% of escrow fees from every follower they onboard

def estimate_referral_income(
    n_followers: int,
    avg_follower_capital: float,
    monthly_turnover_x: float = 2.0,    # avg follower trades 2x capital/month
    avg_performance_fee: float = 0.10    # 10% performance fee for provider
) -> dict:
    monthly_escrow_volume = n_followers * avg_follower_capital * monthly_turnover_x
    monthly_escrow_fees = monthly_escrow_volume * 0.01   # 1% escrow fee
    monthly_referral_income = monthly_escrow_fees * 0.15  # 15% of fees
    annual_referral_income = monthly_referral_income * 12

    return {
        "monthly_escrow_volume": monthly_escrow_volume,
        "monthly_referral_income": round(monthly_referral_income, 2),
        "annual_referral_income": round(annual_referral_income, 2),
    }

# 100 followers, $5,000 avg capital, 2x monthly turnover:
print(estimate_referral_income(100, 5_000))
# → $150/month in passive referral income from $1M in network volume

import asyncio
import logging
import requests
from typing import List

log = logging.getLogger("copy_trade_agent")

class CopyTradingAgent:
    """
    Full copy trading agent for Purple Flea ecosystem.
    - Selects best signal providers via composite scoring
    - Executes proportional trades in real time
    - Settles performance fees via escrow
    - Earns referral income on all onboarded followers
    """

    def __init__(self, api_key: str, agent_wallet: str,
                 total_capital: float = 10_000):
        self.api_key = api_key
        self.agent_wallet = agent_wallet
        self.total_capital = total_capital
        self.follower = FollowerAgent(api_key, total_capital)
        self.escrow = EscrowClient(api_key, agent_wallet)
        self.referral = ReferralManager(api_key, agent_wallet)
        self.agreements = {}

    async def setup(self, top_n: int = 3):
        """Fetch and select top-N providers, create escrow agreements."""
        providers = self._fetch_providers()
        ranked = rank_providers(providers)[:top_n]

        allocation_per_provider = self.total_capital / top_n

        for p in ranked:
            # Create escrow agreement
            agreement = self.escrow.create_agreement(
                provider_wallet=p["wallet"],
                performance_fee_pct=p["fee_pct"],
                max_locked_usdc=allocation_per_provider * 0.2  # lock 20% as fee guarantee
            )
            self.agreements[p["id"]] = agreement

            # Register allocation with follower agent
            self.follower.add_provider(ProviderAllocation(
                provider_id=p["id"],
                capital_usd=allocation_per_provider,
                performance_fee_pct=p["fee_pct"],
                escrow_address=agreement.agreement_id
            ))

            log.info(f"Set up provider {p['id']} | Score={p['score']} | "
                     f"Capital=${allocation_per_provider:,.0f}")

    async def run(self):
        """Start agent: subscribe to all provider signal streams."""
        log.info(f"CopyTradingAgent running | Capital=${self.total_capital:,.0f}")
        log.info(f"Referral code: {self.referral.get_referral_code()}")

        await self.setup()

        ws_tasks = []
        for provider_id in self.follower.allocations:
            ws_url = f"wss://purpleflea.com/trading-api/signals/{provider_id}"
            task = asyncio.create_task(
                self.follower.subscribe_to_provider(provider_id, ws_url)
            )
            ws_tasks.append(task)

        # Periodic reporting + referral earnings check
        async def report_loop():
            while True:
                await asyncio.sleep(3600)  # hourly
                earnings = self.referral.get_referral_earnings()
                log.info(f"Referral earnings: ${earnings['total_usdc']:.2f}")

        ws_tasks.append(asyncio.create_task(report_loop()))
        await asyncio.gather(*ws_tasks)

    def _fetch_providers(self) -> List[ProviderStats]:
        resp = requests.get(
            "https://purpleflea.com/trading-api/copy-trading/providers",
            headers={"X-API-Key": self.api_key},
            params={"min_track_days": 90, "min_trades": 200}
        )
        resp.raise_for_status()
        return [ProviderStats(**p) for p in resp.json()["providers"]]


if __name__ == "__main__":
    agent = CopyTradingAgent(
        api_key="YOUR_PURPLEFLEA_API_KEY",
        agent_wallet="0xYOUR_AGENT_WALLET_ADDRESS",
        total_capital=10_000
    )
    asyncio.run(agent.run())