Trade Attribution and Performance Analysis for AI Agents

An AI agent that generates 18% returns over three months might be a genius or might have gotten lucky in a bull market. Without attribution, you cannot tell. Trade attribution decomposes performance into its constituent sources — alpha, beta, timing, selection, and execution — so that agents and their principals can understand what is actually working, iterate on what is not, and account for results honestly.

1. Why Attribution Matters: Separating Skill from Luck

Returns without attribution are noise. A strategy that made money may have made it because:

• The market went up (beta) — not skill
• The agent happened to be in the right sector at the right time (timing) — partially skill
• The agent selected the right assets within a sector (selection) — genuine alpha
• Execution was particularly good or bad (interaction effects)
• Pure luck — no replicable edge

Without decomposing returns, agents iterate on the wrong things. A strategy that appears to work because of favorable beta exposure gets replaced by a new agent that also just rides beta — and both fail when the market turns. True alpha, by contrast, is persistent and deserves to be scaled.

Separating Skill from Luck

The statistical challenge: market returns are noisy enough that even a skilled agent will have long runs of underperformance, and an unskilled agent will have long runs of outperformance. Standard approaches to separate them:

• Information ratio (IR): Active return / Active risk. Sustainable IR above 0.5 is significant; above 1.0 is exceptional
• t-test on alpha: Is the estimated alpha statistically different from zero at 95% confidence? Requires many observations
• Bootstrap analysis: Randomly shuffle trade order and compute return distribution — does the actual strategy beat random?
• Holdout validation: Attribution on out-of-sample periods that were not used for strategy development

Three Levels of Attribution

2. Return Decomposition: Alpha, Beta, Timing, Selection

The standard decomposition of an active portfolio's returns uses a factor model framework. The simplest decomposition is the classic CAPM split:

For AI agents running multi-strategy books, a more nuanced decomposition is essential:

The key insight: the 14% gross return has very different quality depending on its source. Beta return is cheap (buy an index fund). Execution cost is wasteful and improvable. Alpha via selection and timing is what makes the agent worth running.

3. Brinson Attribution Framework for Multi-Strategy Agents

The Brinson-Hood-Beebower (BHB) model from 1986 is the industry-standard framework for attribution analysis. Originally designed for equity portfolio managers, it adapts naturally to multi-strategy AI agents where each strategy is analogous to a sector, and each trade within a strategy is a security selection.

The Mathematics

Given a portfolio of N strategies, each with weight w_p,i (actual) and return r_p,i, compared to a benchmark with weights w_b,i and returns r_b,i:

4. Python: TradeAttributor Class

The following TradeAttributor class implements full return decomposition, Brinson attribution, factor attribution, and execution analysis. It connects to the Purple Flea wallet API to fetch return data.

# trade_attributor.py
# Complete trade attribution and performance analysis for AI agents
# Requires: numpy, pandas, scipy, requests

import numpy as np
import pandas as pd
from scipy import stats
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
import requests
from datetime import datetime, timedelta
import logging

logger = logging.getLogger("TradeAttributor")


@dataclass
class Trade:
    """Single trade record."""
    trade_id: str
    strategy: str
    symbol: str
    side: str              # 'buy' or 'sell'
    quantity: float
    decision_price: float  # price when signal fired (t-decision)
    arrival_price: float   # market price at order arrival (t-arrival)
    fill_price: float      # actual execution price
    close_price: float     # price at position close or end of period
    timestamp_open: float
    timestamp_close: float
    fees: float = 0.0


@dataclass
class StrategyPeriodReturn:
    """Return data for a single strategy over a period."""
    strategy: str
    portfolio_weight: float    # actual weight
    benchmark_weight: float    # target/benchmark weight
    portfolio_return: float    # actual return
    benchmark_return: float    # strategy benchmark return


class TradeAttributor:
    """
    Full trade attribution engine for AI trading agents.
    Implements Brinson attribution, factor decomposition,
    and execution shortfall analysis.
    """

    PURPLE_FLEA_API = "https://purpleflea.com/api/v1"

    def __init__(self, api_key: str, agent_id: str):
        self.api_key = api_key
        self.agent_id = agent_id
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        })

    def decompose_returns(
        self,
        portfolio_returns: np.ndarray,
        benchmark_returns: np.ndarray,
        factor_returns: Optional[Dict[str, np.ndarray]] = None,
    ) -> Dict:
        """
        Decompose portfolio returns into alpha, beta, and factor components.
        Uses OLS regression with optional multi-factor model.

        Args:
            portfolio_returns: Array of periodic portfolio returns
            benchmark_returns: Array of benchmark returns (same periods)
            factor_returns: Optional dict of {factor_name: returns_array}

        Returns:
            Dict with alpha, betas, R², information ratio, and residuals
        """
        n = len(portfolio_returns)
        if n < 10:
            raise ValueError("Need at least 10 observations for attribution")

        # Build factor matrix
        X_cols = {"benchmark": benchmark_returns}
        if factor_returns:
            X_cols.update(factor_returns)

        X = np.column_stack([np.ones(n)] + [v for v in X_cols.values()])
        y = portfolio_returns

        # OLS regression: y = alpha + beta_1*X_1 + ... + beta_k*X_k + eps
        result = np.linalg.lstsq(X, y, rcond=None)
        coefficients = result[0]
        alpha = coefficients[0]
        betas = {name: coefficients[i+1] for i, name in enumerate(X_cols)}

        # Fitted values and residuals
        fitted = X @ coefficients
        residuals = y - fitted

        # R-squared
        ss_tot = np.sum((y - np.mean(y))**2)
        ss_res = np.sum(residuals**2)
        r_squared = 1 - ss_res / ss_tot if ss_tot > 0 else 0

        # Information ratio (active return / active risk)
        active_returns = portfolio_returns - benchmark_returns
        ir = np.mean(active_returns) / np.std(active_returns) if np.std(active_returns) > 0 else 0

        # Statistical significance of alpha
        std_err = np.sqrt(np.var(residuals) / n)
        alpha_t_stat = alpha / std_err if std_err > 0 else 0
        alpha_p_value = 2 * (1 - stats.t.cdf(abs(alpha_t_stat), df=n-len(coefficients)))

        # Factor return contribution
        factor_contributions = {}
        for i, (name, factor_ret) in enumerate(X_cols.items()):
            factor_contributions[name] = float(coefficients[i+1] * np.mean(factor_ret))

        return {
            "alpha_annualized": float(alpha * 252),
            "alpha_per_period": float(alpha),
            "betas": {k: float(v) for k, v in betas.items()},
            "r_squared": float(r_squared),
            "information_ratio": float(ir),
            "alpha_t_stat": float(alpha_t_stat),
            "alpha_p_value": float(alpha_p_value),
            "alpha_significant": alpha_p_value < 0.05,
            "factor_contributions": factor_contributions,
            "residuals": residuals.tolist(),
            "total_portfolio_return": float(np.sum(portfolio_returns)),
            "total_benchmark_return": float(np.sum(benchmark_returns)),
            "total_active_return": float(np.sum(active_returns)),
        }

    def timing_attribution(
        self,
        trades: List[Trade],
        market_returns: Optional[Dict[str, List[float]]] = None,
    ) -> Dict:
        """
        Attribute returns to entry/exit timing quality.
        Compares actual entry/exit to hypothetical passive hold.

        Timing alpha = Actual return - Buy-and-hold return over same period.
        Positive = agent entered/exited at better times than passive.
        """
        timing_pnl = 0.0
        timing_records = []

        for trade in trades:
            # Actual return on the trade
            if trade.side == "buy":
                actual_return = (trade.close_price - trade.fill_price) / trade.fill_price
                passive_return = (trade.close_price - trade.decision_price) / trade.decision_price
            else:
                actual_return = (trade.fill_price - trade.close_price) / trade.fill_price
                passive_return = (trade.decision_price - trade.close_price) / trade.decision_price

            timing_alpha = actual_return - passive_return
            timing_pnl += timing_alpha * trade.quantity * trade.fill_price

            timing_records.append({
                "trade_id": trade.trade_id,
                "strategy": trade.strategy,
                "symbol": trade.symbol,
                "actual_return": float(actual_return),
                "passive_return": float(passive_return),
                "timing_alpha": float(timing_alpha),
                "timing_pnl": float(timing_alpha * trade.quantity * trade.fill_price),
            })

        timing_alphas = [r["timing_alpha"] for r in timing_records]
        return {
            "total_timing_pnl": timing_pnl,
            "mean_timing_alpha": np.mean(timing_alphas) if timing_alphas else 0,
            "timing_win_rate": sum(1 for a in timing_alphas if a > 0) / len(timing_alphas) if timing_alphas else 0,
            "records": timing_records,
        }

    def selection_attribution(
        self,
        trades: List[Trade],
        strategy_benchmark_returns: Dict[str, float],
    ) -> Dict:
        """
        Attribute returns to asset selection skill within each strategy.
        Selection alpha = trade return - strategy benchmark return.
        Positive = agent selected better-than-average assets within the strategy.
        """
        by_strategy: Dict[str, List] = {}
        for trade in trades:
            strat_bm = strategy_benchmark_returns.get(trade.strategy, 0.0)
            if trade.side == "buy":
                trade_ret = (trade.close_price - trade.fill_price) / trade.fill_price
            else:
                trade_ret = (trade.fill_price - trade.close_price) / trade.fill_price

            selection_alpha = trade_ret - strat_bm
            trade_notional = trade.quantity * trade.fill_price

            record = {
                "trade_id": trade.trade_id,
                "symbol": trade.symbol,
                "trade_return": float(trade_ret),
                "strategy_bm_return": float(strat_bm),
                "selection_alpha": float(selection_alpha),
                "selection_pnl": float(selection_alpha * trade_notional),
                "notional": trade_notional,
            }
            by_strategy.setdefault(trade.strategy, []).append(record)

        summary = {}
        for strat, records in by_strategy.items():
            alphas = [r["selection_alpha"] for r in records]
            summary[strat] = {
                "total_selection_pnl": sum(r["selection_pnl"] for r in records),
                "mean_selection_alpha": float(np.mean(alphas)),
                "selection_win_rate": sum(1 for a in alphas if a > 0) / len(alphas),
                "n_trades": len(records),
                "records": records,
            }
        return summary

    def brinson_attribution(
        self,
        strategies: List[StrategyPeriodReturn],
    ) -> Dict:
        """
        Full Brinson-Hood-Beebower attribution across strategies.
        Returns allocation, selection, and interaction effects per strategy.
        """
        # Portfolio benchmark return (weighted average of strategy benchmarks)
        r_b = sum(s.benchmark_weight * s.benchmark_return for s in strategies)

        total_allocation = 0.0
        total_selection = 0.0
        total_interaction = 0.0
        strategy_results = []

        for s in strategies:
            allocation = (s.portfolio_weight - s.benchmark_weight) * (s.benchmark_return - r_b)
            selection = s.benchmark_weight * (s.portfolio_return - s.benchmark_return)
            interaction = (s.portfolio_weight - s.benchmark_weight) * (s.portfolio_return - s.benchmark_return)

            total_allocation += allocation
            total_selection += selection
            total_interaction += interaction

            strategy_results.append({
                "strategy": s.strategy,
                "portfolio_weight": s.portfolio_weight,
                "benchmark_weight": s.benchmark_weight,
                "active_weight": s.portfolio_weight - s.benchmark_weight,
                "portfolio_return": s.portfolio_return,
                "benchmark_return": s.benchmark_return,
                "allocation_effect": float(allocation),
                "selection_effect": float(selection),
                "interaction_effect": float(interaction),
                "total_contribution": float(allocation + selection + interaction),
            })

        return {
            "total_allocation_effect": float(total_allocation),
            "total_selection_effect": float(total_selection),
            "total_interaction_effect": float(total_interaction),
            "total_active_return": float(total_allocation + total_selection + total_interaction),
            "portfolio_benchmark_return": float(r_b),
            "strategies": strategy_results,
        }

5. Factor Attribution: What's Driving Returns

Factor attribution answers: is my agent's return explained by known systematic factors (momentum, mean reversion, volatility), or is there genuine idiosyncratic alpha?

Crypto-Relevant Factors

    def factor_attribution(
        self,
        portfolio_returns: np.ndarray,
        factor_data: Dict[str, np.ndarray],
        lookback: int = 60,
    ) -> Dict:
        """
        Multi-factor attribution using rolling regression.
        Measures how much of the portfolio return is explained
        by each factor and what remains as alpha.

        Args:
            portfolio_returns: Daily or period returns of portfolio
            factor_data: Dict of {factor_name: returns_array}
            lookback: Rolling window for regression (periods)

        Returns:
            Factor betas, return contributions, and residual alpha
        """
        n = len(portfolio_returns)
        factor_names = list(factor_data.keys())
        factor_matrix = np.column_stack([factor_data[f] for f in factor_names])

        # Full-period regression
        X = np.column_stack([np.ones(n), factor_matrix])
        coeffs, _, _, _ = np.linalg.lstsq(X, portfolio_returns, rcond=None)

        alpha = coeffs[0]
        betas = {name: coeffs[i+1] for i, name in enumerate(factor_names)}

        # Factor return contributions (beta × mean factor return)
        contributions = {
            name: float(coeffs[i+1] * np.mean(factor_data[name]))
            for i, name in enumerate(factor_names)
        }

        # Rolling betas (detect regime changes)
        rolling_betas = []
        for t in range(lookback, n + 1):
            window_y = portfolio_returns[t-lookback:t]
            window_X = X[t-lookback:t]
            try:
                w_coeffs, _, _, _ = np.linalg.lstsq(window_X, window_y, rcond=None)
                rolling_betas.append({
                    "t": t,
                    "alpha": float(w_coeffs[0]),
                    **{name: float(w_coeffs[i+1]) for i, name in enumerate(factor_names)}
                })
            except:
                pass

        # Residuals
        fitted = X @ coeffs
        residuals = portfolio_returns - fitted
        unexplained_return = float(np.mean(residuals))

        total_explained = sum(contributions.values())
        total_return = float(np.mean(portfolio_returns))

        return {
            "alpha_per_period": float(alpha),
            "factor_betas": {k: float(v) for k, v in betas.items()},
            "factor_return_contributions": contributions,
            "total_explained_return": float(total_explained),
            "unexplained_alpha": float(unexplained_return),
            "total_portfolio_return": total_return,
            "pct_explained": total_explained / total_return if abs(total_return) > 1e-9 else 0,
            "rolling_betas": rolling_betas,
        }

6. Execution Attribution: Implementation Shortfall Decomposition

Implementation shortfall (IS) is the gap between the paper portfolio return (using the decision price) and the actual portfolio return (using actual fill prices). It is the most rigorous framework for measuring execution quality.

IS decomposes into three parts:

1. Delay cost: Market moved between decision and order arrival (signal decay)
2. Market impact: Price moved because of our own order (impact cost)
3. Timing cost / missed trades: Orders that were not executed (opportunity cost)

    def implementation_shortfall(
        self,
        trades: List[Trade],
    ) -> Dict:
        """
        Compute implementation shortfall and its components for a list of trades.
        Requires: decision_price, arrival_price, fill_price per trade.

        Returns:
            Total IS, delay cost, market impact, and per-trade breakdown.
        """
        total_is = 0.0
        total_delay = 0.0
        total_impact = 0.0
        records = []

        for trade in trades:
            direction = 1 if trade.side == "buy" else -1
            notional = trade.quantity * trade.decision_price

            # Delay cost: cost of waiting from decision to arrival
            delay_cost = direction * (trade.arrival_price - trade.decision_price) * trade.quantity

            # Market impact: fill price vs. arrival price
            impact_cost = direction * (trade.fill_price - trade.arrival_price) * trade.quantity

            # Total IS for this trade (excluding fees)
            trade_is = delay_cost + impact_cost + trade.fees
            trade_is_bps = (trade_is / notional * 10000) if notional > 0 else 0

            total_is += trade_is
            total_delay += delay_cost
            total_impact += impact_cost

            records.append({
                "trade_id": trade.trade_id,
                "symbol": trade.symbol,
                "strategy": trade.strategy,
                "notional": float(notional),
                "delay_cost": float(delay_cost),
                "impact_cost": float(impact_cost),
                "fees": float(trade.fees),
                "total_is": float(trade_is),
                "total_is_bps": float(trade_is_bps),
            })

        total_notional = sum(r["notional"] for r in records)
        return {
            "total_is": float(total_is),
            "total_is_bps": float(total_is / total_notional * 10000) if total_notional > 0 else 0,
            "total_delay_cost": float(total_delay),
            "total_impact_cost": float(total_impact),
            "delay_fraction": float(total_delay / total_is) if total_is != 0 else 0,
            "impact_fraction": float(total_impact / total_is) if total_is != 0 else 0,
            "n_trades": len(trades),
            "trades": records,
        }

    def execution_quality_score(self, is_result: Dict) -> Dict:
        """
        Score execution quality based on IS components.
        Returns letter grade and actionable recommendations.
        """
        is_bps = is_result["total_is_bps"]
        delay_frac = is_result["delay_fraction"]
        impact_frac = is_result["impact_fraction"]

        # Grading: <5bps = A, 5-15 = B, 15-30 = C, >30 = D
        if is_bps < 5: grade = "A"
        elif is_bps < 15: grade = "B"
        elif is_bps < 30: grade = "C"
        else: grade = "D"

        recommendations = []
        if delay_frac > 0.4:
            recommendations.append("Reduce signal-to-order latency; use pre-staged orders or faster routing")
        if impact_frac > 0.5:
            recommendations.append("Reduce order size or use algo execution (VWAP/TWAP) to minimize footprint")
        if is_bps > 20:
            recommendations.append("Consider switching to limit orders with post-only flag to earn maker rebates")

        return {
            "grade": grade,
            "is_bps": is_bps,
            "dominant_cost": "delay" if delay_frac > impact_frac else "impact",
            "recommendations": recommendations,
        }

7. Reporting to Principals: Human-Readable Attribution Reports

AI agents operating within multi-agent systems or under human supervision need to communicate attribution results clearly. The goal is to answer three questions concisely:

1. What happened? (return summary)
2. Why? (attribution decomposition)
3. What should change? (actionable recommendations)

    def generate_attribution_report(
        self,
        period_start: str,
        period_end: str,
        decompose_result: Dict,
        brinson_result: Dict,
        is_result: Dict,
        factor_result: Dict,
    ) -> str:
        """
        Generate a human-readable attribution report for a principal or monitoring system.
        Returns formatted text report.
        """
        lines = [
            f"╔══════════════════════════════════════════════════╗",
            f"║     TRADE ATTRIBUTION REPORT — {period_start} to {period_end}      ║",
            f"╚══════════════════════════════════════════════════╝",
            "",
            "── RETURN SUMMARY ─────────────────────────────────",
            f"  Portfolio Return:     {decompose_result['total_portfolio_return']*100:+.2f}%",
            f"  Benchmark Return:     {decompose_result['total_benchmark_return']*100:+.2f}%",
            f"  Active Return:        {decompose_result['total_active_return']*100:+.2f}%",
            f"  Information Ratio:    {decompose_result['information_ratio']:.3f}",
            f"  Alpha (annualized):   {decompose_result['alpha_annualized']*100:+.2f}%",
            f"  Alpha significant:    {'YES ✓' if decompose_result['alpha_significant'] else 'NO (p=' + f\"{decompose_result['alpha_p_value']:.2f})\"}",
            "",
            "── BRINSON ATTRIBUTION ────────────────────────────",
            f"  Allocation Effect:   {brinson_result['total_allocation_effect']*100:+.3f}%",
            f"  Selection Effect:    {brinson_result['total_selection_effect']*100:+.3f}%",
            f"  Interaction Effect:  {brinson_result['total_interaction_effect']*100:+.3f}%",
            f"  Total Active:        {brinson_result['total_active_return']*100:+.3f}%",
            "",
        ]

        lines.append("  Strategy-level breakdown:")
        for s in brinson_result["strategies"]:
            total_contrib = s["total_contribution"] * 100
            sign = "+" if total_contrib >= 0 else ""
            lines.append(f"    {s['strategy']:<20} {sign}{total_contrib:.3f}%")

        lines += [
            "",
            "── FACTOR ATTRIBUTION ─────────────────────────────",
        ]
        for factor, contrib in factor_result["factor_return_contributions"].items():
            beta = factor_result["factor_betas"][factor]
            lines.append(f"  {factor:<22} β={beta:+.3f}  contrib={contrib*100:+.3f}%")

        lines += [
            f"  Residual alpha:      {factor_result['unexplained_alpha']*100:+.4f}%/period",
            f"  R-squared:           {factor_result.get('r_squared', decompose_result['r_squared']):.3f}",
            "",
            "── EXECUTION QUALITY ──────────────────────────────",
            f"  Implementation Shortfall: {is_result['total_is_bps']:.1f} bps",
            f"  Delay Cost:               {is_result['total_delay_cost']:.2f}",
            f"  Market Impact Cost:       {is_result['total_impact_cost']:.2f}",
            f"  Number of Trades:         {is_result['n_trades']}",
            "",
            "── VERDICT ────────────────────────────────────────",
        ]

        alpha_ann = decompose_result["alpha_annualized"]
        ir = decompose_result["information_ratio"]
        if alpha_ann > 0.05 and ir > 0.5:
            verdict = "STRONG ALPHA — strategy generating genuine skill-based returns"
        elif alpha_ann > 0:
            verdict = "MARGINAL ALPHA — positive but may be noise; increase sample size"
        elif alpha_ann > -0.02:
            verdict = "BETA PLAYER — returns driven by market exposure; no clear edge"
        else:
            verdict = "UNDERPERFORMING — destroying value vs. benchmark; review strategy"

        lines.append(f"  {verdict}")
        lines.append("")
        return "\n".join(lines)

    def post_report_to_principal(
        self,
        report_text: str,
        principal_webhook: Optional[str] = None,
    ) -> bool:
        """Send report to principal via Purple Flea message API or webhook."""
        payload = {
            "agent_id": self.agent_id,
            "message_type": "attribution_report",
            "content": report_text,
            "timestamp": datetime.utcnow().isoformat(),
        }

        if principal_webhook:
            resp = requests.post(principal_webhook, json=payload, timeout=10)
            return resp.status_code == 200

        # Default: post via Purple Flea escrow/message API
        resp = self.session.post(
            f"{self.PURPLE_FLEA_API}/agent/messages",
            json=payload,
        )
        return resp.status_code == 200

Example Report Output

8. Integration with Purple Flea Wallet API for Return Data

The Purple Flea wallet API provides the raw trade history and PnL data needed for attribution. Here is how to fetch, normalize, and run attribution in a production loop.

class PurpleFleatAttributionPipeline:
    """
    End-to-end attribution pipeline connecting to Purple Flea APIs.
    Fetches trade history, benchmarks, and outputs attribution reports.
    """

    def __init__(self, api_key: str, agent_id: str):
        self.attributor = TradeAttributor(api_key, agent_id)
        self.api_key = api_key
        self.agent_id = agent_id
        self.session = self.attributor.session

    def fetch_trade_history(
        self,
        start_ts: float,
        end_ts: float,
    ) -> List[Trade]:
        """Fetch raw trade records from Purple Flea wallet API."""
        resp = self.session.get(
            f"{TradeAttributor.PURPLE_FLEA_API}/wallet/trades",
            params={
                "agent_id": self.agent_id,
                "start": start_ts,
                "end": end_ts,
                "include_fees": True,
            }
        )
        if resp.status_code != 200:
            logger.error(f"Failed to fetch trades: {resp.status_code}")
            return []

        raw_trades = resp.json()["trades"]
        return [
            Trade(
                trade_id=t["id"],
                strategy=t.get("strategy", "default"),
                symbol=t["symbol"],
                side=t["side"],
                quantity=t["quantity"],
                decision_price=t.get("decision_price", t["price"]),
                arrival_price=t.get("arrival_price", t["price"]),
                fill_price=t["price"],
                close_price=t.get("close_price", t["price"]),
                timestamp_open=t["timestamp"],
                timestamp_close=t.get("close_timestamp", t["timestamp"]),
                fees=t.get("fees", 0.0),
            )
            for t in raw_trades
        ]

    def fetch_portfolio_returns(
        self,
        start_ts: float,
        end_ts: float,
        frequency: str = "1d",
    ) -> Tuple[np.ndarray, np.ndarray]:
        """
        Fetch portfolio and benchmark (BTC) daily returns from Purple Flea.
        Returns (portfolio_returns, benchmark_returns) arrays.
        """
        resp = self.session.get(
            f"{TradeAttributor.PURPLE_FLEA_API}/wallet/performance",
            params={"agent_id": self.agent_id, "start": start_ts, "end": end_ts, "freq": frequency}
        )
        if resp.status_code != 200:
            return np.array([]), np.array([])

        data = resp.json()
        portfolio_rets = np.array(data["portfolio_returns"])
        benchmark_rets = np.array(data["benchmark_returns"])
        return portfolio_rets, benchmark_rets

    def run_full_attribution(
        self,
        start_ts: float,
        end_ts: float,
        strategy_config: List[Dict],
        principal_webhook: Optional[str] = None,
    ) -> Dict:
        """
        Run the full attribution pipeline and generate report.

        Args:
            start_ts: Period start (unix timestamp)
            end_ts: Period end (unix timestamp)
            strategy_config: List of {strategy, benchmark_weight, benchmark_return}
            principal_webhook: Optional URL to POST the report to

        Returns:
            Complete attribution results dict
        """
        start_str = datetime.fromtimestamp(start_ts).strftime("%Y-%m-%d")
        end_str = datetime.fromtimestamp(end_ts).strftime("%Y-%m-%d")

        logger.info(f"Running attribution for {start_str} to {end_str}")

        trades = self.fetch_trade_history(start_ts, end_ts)
        port_rets, bench_rets = self.fetch_portfolio_returns(start_ts, end_ts)

        if len(port_rets) < 10:
            return {"error": "Insufficient return data for attribution"}

        # Core decomposition
        decompose = self.attributor.decompose_returns(port_rets, bench_rets)

        # Brinson attribution
        strat_periods = []
        for cfg in strategy_config:
            strat_trades = [t for t in trades if t.strategy == cfg["strategy"]]
            if strat_trades:
                port_ret = np.mean([
                    (t.close_price - t.fill_price) / t.fill_price for t in strat_trades
                ])
                strat_periods.append(StrategyPeriodReturn(
                    strategy=cfg["strategy"],
                    portfolio_weight=cfg["portfolio_weight"],
                    benchmark_weight=cfg["benchmark_weight"],
                    portfolio_return=float(port_ret),
                    benchmark_return=cfg["benchmark_return"],
                ))

        brinson = self.attributor.brinson_attribution(strat_periods) if strat_periods else {}

        # Factor attribution (using BTC momentum as proxy factors)
        factor_data = {
            "market": bench_rets,
            "momentum": np.roll(bench_rets, 5),   # lagged
        }
        factor = self.attributor.factor_attribution(port_rets, factor_data)

        # Execution attribution
        is_result = self.attributor.implementation_shortfall(trades) if trades else {}
        exec_score = self.attributor.execution_quality_score(is_result) if is_result else {}

        # Generate report
        if brinson and is_result:
            report = self.attributor.generate_attribution_report(
                start_str, end_str, decompose, brinson, is_result, factor
            )
            self.attributor.post_report_to_principal(report, principal_webhook)

        return {
            "period": {"start": start_str, "end": end_str},
            "decomposition": decompose,
            "brinson": brinson,
            "factor": factor,
            "implementation_shortfall": is_result,
            "execution_score": exec_score,
            "n_trades": len(trades),
        }


# ─── USAGE EXAMPLE ───────────────────────────────────────────────────────────

def run_monthly_attribution():
    API_KEY = "pf_live_your_key_here"
    AGENT_ID = "agent_abc123"

    pipeline = PurpleFleatAttributionPipeline(API_KEY, AGENT_ID)

    now = datetime.utcnow()
    start = (now - timedelta(days=30)).timestamp()
    end = now.timestamp()

    strategy_config = [
        {"strategy": "momentum", "portfolio_weight": 0.40, "benchmark_weight": 0.33, "benchmark_return": 0.08},
        {"strategy": "mean_reversion", "portfolio_weight": 0.35, "benchmark_weight": 0.33, "benchmark_return": 0.04},
        {"strategy": "carry", "portfolio_weight": 0.25, "benchmark_weight": 0.34, "benchmark_return": 0.02},
    ]

    results = pipeline.run_full_attribution(
        start_ts=start,
        end_ts=end,
        strategy_config=strategy_config,
        principal_webhook=None,  # or your webhook URL
    )

    logger.info(f"Attribution complete. IR={results['decomposition']['information_ratio']:.2f}")
    logger.info(f"Execution grade: {results['execution_score'].get('grade', 'N/A')}")
    return results


if __name__ == "__main__":
    run_monthly_attribution()

Summary

Trade attribution is not optional for serious AI trading agents — it is the difference between systematic improvement and randomness dressed up as a strategy. The key takeaways:

• Return decomposition separates alpha from beta — only alpha justifies the cost of running an active agent
• Brinson attribution shows whether value comes from capital allocation decisions or from asset selection quality within each strategy
• Factor attribution reveals whether returns are driven by known systematic factors or genuine idiosyncratic skill
• Implementation shortfall decomposes execution cost into delay and market impact — both are improvable with better execution logic
• Automated reporting to principals builds trust and enables oversight of autonomous agents at scale
• Purple Flea's wallet API provides the trade history and return data needed to run attribution without external data vendors

An agent that attributes its returns honestly — and acts on those insights — will compound its edge over time. An agent that ignores attribution will compound its mistakes.

Related reading: Market Microstructure Trading · HFT for AI Agents · Agent Portfolio Theory · Risk Management