Portfolio Attribution for AI Agents

Why Attribution Matters

Most agents know their total P&L. Few know why they made or lost it. Without attribution, you cannot improve — you are flying blind. A 15% monthly return that comes 90% from a lucky coin toss (the casino) and 10% from a losing trading strategy tells a very different story than the same return split evenly across all services with consistent factor contributions.

Portfolio attribution answers the question: given my total return, how much came from each decision I made? It decomposes total P&L into additive components so you can identify what to scale up, what to fix, and what to kill.

For agents operating across Purple Flea's six services, attribution is especially critical because the return sources have completely different risk profiles: casino returns are stochastic, trading returns are skill-dependent, and referral income is quasi-passive. Mixing them without attribution creates a distorted picture of your actual edge.

The Brinson-Hood-Beebower Model

The Brinson-Hood-Beebower (BHB) model, originally designed for equity portfolio managers, decomposes active return relative to a benchmark into three additive components: allocation effect, selection effect, and interaction effect.

For a Purple Flea agent, the "benchmark" can be defined as a simple 70/30 BTC/USDT hold with no active services. Any return above that benchmark represents genuine active skill — and the BHB model tells you exactly which service or strategy decision generated it.

bhb_attribution.pyfrom dataclasses import dataclass
from typing import Dict, List
import numpy as np

@dataclass
class SegmentData:
    name: str
    portfolio_weight: float    # w_p
    benchmark_weight: float    # w_b
    portfolio_return: float    # r_p,i
    benchmark_return: float    # r_b,i

@dataclass
class BHBResult:
    segment: str
    allocation: float
    selection: float
    interaction: float
    total_active: float

def bhb_attribution(segments: List[SegmentData],
                    benchmark_total_return: float) -> List[BHBResult]:
    """
    Brinson-Hood-Beebower attribution for a list of portfolio segments.
    benchmark_total_return: R_b = sum(w_b,i * r_b,i)
    """
    results = []
    R_b = benchmark_total_return
    for seg in segments:
        w_p = seg.portfolio_weight
        w_b = seg.benchmark_weight
        r_p = seg.portfolio_return
        r_b = seg.benchmark_return
        allocation  = (w_p - w_b) * (r_b - R_b)
        selection   = w_b * (r_p - r_b)
        interaction = (w_p - w_b) * (r_p - r_b)
        total_active = allocation + selection + interaction
        results.append(BHBResult(
            segment=seg.name,
            allocation=round(allocation, 6),
            selection=round(selection, 6),
            interaction=round(interaction, 6),
            total_active=round(total_active, 6)
        ))
    # Verify: sum of all total_active should equal portfolio_return - benchmark_return
    active_sum = sum(r.total_active for r in results)
    portfolio_return = sum(s.portfolio_weight * s.portfolio_return for s in segments)
    assert abs(active_sum - (portfolio_return - R_b)) < 1e-9, \
        f"Attribution doesn't add up: {active_sum} vs {portfolio_return - R_b}"
    return results

# Example: Purple Flea agent with 4 service segments
segments = [
    SegmentData("Casino",   portfolio_weight=0.30, benchmark_weight=0.00,
                portfolio_return=0.08,  benchmark_return=0.00),
    SegmentData("Trading",  portfolio_weight=0.40, benchmark_weight=0.70,
                portfolio_return=0.12,  benchmark_return=0.07),  # benchmark = BTC hold
    SegmentData("Wallet",   portfolio_weight=0.15, benchmark_weight=0.30,
                portfolio_return=0.035, benchmark_return=0.03),
    SegmentData("Referral", portfolio_weight=0.15, benchmark_weight=0.00,
                portfolio_return=0.05,  benchmark_return=0.00),
]
R_b = sum(s.benchmark_weight * s.benchmark_return for s in segments)  # = 0.07*0.7 + 0.03*0.3 = 0.058
results = bhb_attribution(segments, R_b)
for r in results:
    print(f"{r.segment:10s}  alloc={r.allocation:+.4f}  "
          f"select={r.selection:+.4f}  interact={r.interaction:+.4f}  "
          f"total={r.total_active:+.4f}")

Factor Attribution

BHB attribution explains which services generated returns. Factor attribution explains why the trading component performed as it did — which market factors the strategy was implicitly betting on.

The three primary factors in crypto trading are:

• Momentum: Recent winners tend to continue winning. Strategies that buy relative strength and sell relative weakness earn momentum premium.
• Mean-reversion: Overextended moves tend to reverse. Contrarian strategies earn mean-reversion premium, especially on shorter timeframes.
• Carry: In crypto, carry comes from funding rate harvesting (being long spot while shorting perps when funding is positive), or from yield-bearing vault positions.

factor_attribution.pyimport numpy as np
import pandas as pd
from sklearn.linear_model import LinearRegression

def compute_factor_returns(prices: pd.DataFrame,
                           funding_rates: pd.Series,
                           lookback_momentum: int = 20,
                           lookback_reversion: int = 5) -> pd.DataFrame:
    """
    Compute daily factor returns for crypto portfolio attribution.
    prices: DataFrame with columns as assets (e.g., BTC, ETH, SOL, etc.)
    funding_rates: Series of BTC perp funding rate (8h, annualized)
    """
    returns = prices.pct_change().dropna()

    # Momentum factor: cross-sectional z-score of 20-day returns
    mom_scores = returns.rolling(lookback_momentum).sum()
    mom_factor = mom_scores.sub(mom_scores.mean(axis=1), axis=0)
    mom_factor = mom_factor.div(mom_factor.std(axis=1), axis=0)
    momentum_return = (mom_factor.shift(1) * returns).mean(axis=1)

    # Mean-reversion factor: negative of 5-day return z-score
    rev_scores = returns.rolling(lookback_reversion).sum()
    rev_factor = -rev_scores.sub(rev_scores.mean(axis=1), axis=0)
    rev_factor = rev_factor.div(rev_factor.std(axis=1), axis=0)
    reversion_return = (rev_factor.shift(1) * returns).mean(axis=1)

    # Carry factor: proxy from funding rates (positive funding = long bias profitable)
    carry_return = funding_rates.reindex(returns.index).fillna(0) / 365

    factors = pd.DataFrame({
        "momentum":   momentum_return,
        "reversion":  reversion_return,
        "carry":      carry_return,
    })
    return factors

def factor_regression(portfolio_returns: pd.Series,
                      factor_returns: pd.DataFrame) -> dict:
    """
    OLS regression of portfolio returns on factor returns.
    Returns betas, t-stats, R-squared, and factor contribution to return.
    """
    X = factor_returns.values
    y = portfolio_returns.reindex(factor_returns.index).fillna(0).values

    model = LinearRegression(fit_intercept=True)
    model.fit(X, y)

    residuals = y - model.predict(X)
    ss_res = (residuals ** 2).sum()
    ss_tot = ((y - y.mean()) ** 2).sum()
    r2 = 1 - ss_res / (ss_tot + 1e-12)

    # t-statistics
    n, k = X.shape
    se = np.sqrt(np.diag(np.linalg.pinv(X.T @ X) * ss_res / (n - k - 1) + 1e-12))

    factor_means = factor_returns.mean()
    contributions = {f: model.coef_[i] * factor_means.iloc[i]
                     for i, f in enumerate(factor_returns.columns)}

    return {
        "alpha_daily":     model.intercept_,
        "alpha_annual":    model.intercept_ * 252,
        "betas":           dict(zip(factor_returns.columns, model.coef_)),
        "t_stats":         dict(zip(factor_returns.columns, model.coef_ / (se + 1e-12))),
        "r_squared":       r2,
        "factor_contrib":  contributions,
        "alpha_contrib":   model.intercept_,
    }

# Usage
# factors = compute_factor_returns(prices_df, funding_series)
# result = factor_regression(my_strategy_returns, factors)
# print(f"Alpha: {result['alpha_annual']:.2%} annualized")
# print(f"R²: {result['r_squared']:.3f}")
# for factor, beta in result['betas'].items():
#     contrib = result['factor_contrib'][factor]
#     print(f"  {factor}: beta={beta:.3f}, daily contrib={contrib:.4%}")

Fee Drag Quantification

Fees are the silent killer of trading strategies. A strategy with a 20% annualized gross alpha can be reduced to 12% after fees — and if your turnover is high enough, fees can turn a profitable strategy into a net loser. Attribution must make fees visible and attribute them to the strategies and services that incur them.

fee_attribution.pyfrom dataclasses import dataclass, field
from typing import List, Dict
import pandas as pd

@dataclass
class Trade:
    ts: float
    strategy: str
    service: str     # "trading" | "casino" | "escrow"
    notional: float  # USD value
    fee_rate: float  # as decimal (e.g., 0.001 = 0.1%)
    slippage: float  # actual - expected price, as decimal

@dataclass
class FeeAttributionReport:
    total_fees: float
    total_slippage: float
    total_cost: float
    by_strategy: Dict[str, float]
    by_service: Dict[str, float]
    fee_drag_pct: float   # as % of gross return
    breakeven_return: float  # gross return needed to break even

def attribute_fees(trades: List[Trade],
                   portfolio_value: float,
                   gross_return: float) -> FeeAttributionReport:
    """
    Decompose total costs (fees + slippage) by strategy and service.
    portfolio_value: current portfolio value in USD (for fee drag %)
    gross_return: total gross portfolio return for the period (as decimal)
    """
    by_strategy: Dict[str, float] = {}
    by_service: Dict[str, float] = {}
    total_fees = 0.0
    total_slippage = 0.0

    for t in trades:
        fee = t.notional * t.fee_rate
        slip = abs(t.notional * t.slippage)

        total_fees += fee
        total_slippage += slip

        by_strategy[t.strategy] = by_strategy.get(t.strategy, 0) + fee + slip
        by_service[t.service]   = by_service.get(t.service, 0) + fee + slip

    total_cost = total_fees + total_slippage
    fee_drag_pct = total_cost / portfolio_value if portfolio_value > 0 else 0.0
    breakeven_return = fee_drag_pct  # gross return just to cover costs

    return FeeAttributionReport(
        total_fees=round(total_fees, 4),
        total_slippage=round(total_slippage, 4),
        total_cost=round(total_cost, 4),
        by_strategy={k: round(v, 4) for k, v in by_strategy.items()},
        by_service={k: round(v, 4) for k, v in by_service.items()},
        fee_drag_pct=round(fee_drag_pct, 6),
        breakeven_return=round(breakeven_return, 6)
    )

Slippage Attribution

Slippage — the difference between your expected execution price and the actual price you received — erodes returns in a way that is easy to overlook because it never appears as an explicit line item on your account statement. It is embedded in your trade prices.

For systematic agents, slippage attribution breaks down into four components:

• Market impact: Your order moved the price against you before it was fully filled. Proportional to order size relative to available liquidity.
• Timing slippage: The price moved between when you decided to trade and when your order reached the exchange. Reduces with lower latency.
• Spread cost: The bid-ask spread paid on market orders. Fixed cost per trade.
• Information leakage: Front-running or order detection by HFT. Reduces with randomized order timing and smaller child orders.

slippage_attribution.pyfrom dataclasses import dataclass
from typing import List, Optional
import numpy as np

@dataclass
class OrderRecord:
    strategy: str
    decision_price: float   # VWAP at decision time
    arrival_price: float    # best ask/bid when order arrived at exchange
    execution_price: float  # average fill price
    size: float             # in base currency
    side: str               # "buy" | "sell"
    spread_at_exec: float   # bid-ask spread at execution time (as decimal)

def decompose_slippage(order: OrderRecord) -> dict:
    """
    Decompose slippage into: timing, spread, market_impact.
    All values as basis points (1 bp = 0.01%).
    Positive = cost, negative = improvement (rare).
    """
    direction = 1 if order.side == "buy" else -1

    # Timing slippage: price moved between decision and arrival
    timing_slip = direction * (order.arrival_price - order.decision_price) / order.decision_price
    # Spread cost: half-spread paid by aggressive order
    spread_cost = order.spread_at_exec / 2
    # Market impact: price moved during execution
    market_impact = direction * (order.execution_price - order.arrival_price) / order.arrival_price

    total_slip = timing_slip + spread_cost + market_impact

    return {
        "timing_bp":       round(timing_slip * 10000, 2),
        "spread_bp":       round(spread_cost * 10000, 2),
        "market_impact_bp": round(market_impact * 10000, 2),
        "total_bp":        round(total_slip * 10000, 2),
        "total_usd":       round(total_slip * order.size * order.execution_price, 4)
    }

def monthly_slippage_report(orders: List[OrderRecord]) -> dict:
    """Aggregate slippage across all orders for a monthly report."""
    by_strategy: dict = {}
    total_bp = 0.0
    total_usd = 0.0
    count = len(orders)
    for o in orders:
        decomp = decompose_slippage(o)
        total_bp += decomp["total_bp"]
        total_usd += decomp["total_usd"]
        s = by_strategy.setdefault(o.strategy, {"bp": 0.0, "usd": 0.0, "n": 0})
        s["bp"] += decomp["total_bp"]
        s["usd"] += decomp["total_usd"]
        s["n"] += 1
    return {
        "avg_slippage_bp":  round(total_bp / count if count else 0, 2),
        "total_slippage_usd": round(total_usd, 2),
        "by_strategy": by_strategy
    }

Purple Flea Multi-Service Attribution

An agent active across Purple Flea's six services — Casino, Trading, Wallet, Domains, Faucet, and Escrow — generates returns from fundamentally different mechanisms. The AttributionEngine below provides a unified view across all services, separating active returns from passive income and accounting for all cost drag.

attribution_engine.pyfrom dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime, date
import json

@dataclass
class ServicePnL:
    service: str
    gross_pnl: float
    fees_paid: float
    slippage: float

    @property
    def net_pnl(self) -> float:
        return self.gross_pnl - self.fees_paid - self.slippage

@dataclass
class AttributionReport:
    period_start: date
    period_end: date
    portfolio_value_start: float
    portfolio_value_end: float
    services: List[ServicePnL]
    factor_attribution: Dict[str, float] = field(default_factory=dict)
    benchmark_return: float = 0.0

    @property
    def total_gross_pnl(self) -> float:
        return sum(s.gross_pnl for s in self.services)

    @property
    def total_net_pnl(self) -> float:
        return sum(s.net_pnl for s in self.services)

    @property
    def total_fees(self) -> float:
        return sum(s.fees_paid for s in self.services)

    @property
    def total_slippage(self) -> float:
        return sum(s.slippage for s in self.services)

    @property
    def gross_return(self) -> float:
        return self.total_gross_pnl / self.portfolio_value_start

    @property
    def net_return(self) -> float:
        return self.total_net_pnl / self.portfolio_value_start

    @property
    def active_return(self) -> float:
        return self.net_return - self.benchmark_return

    def to_report_dict(self) -> dict:
        return {
            "period": f"{self.period_start} to {self.period_end}",
            "portfolio_start": self.portfolio_value_start,
            "portfolio_end": self.portfolio_value_end,
            "returns": {
                "gross": f"{self.gross_return:.2%}",
                "net": f"{self.net_return:.2%}",
                "benchmark": f"{self.benchmark_return:.2%}",
                "active_alpha": f"{self.active_return:.2%}",
            },
            "costs": {
                "total_fees": self.total_fees,
                "total_slippage": self.total_slippage,
                "fee_drag_pct": f"{self.total_fees / self.portfolio_value_start:.2%}",
            },
            "services": [
                {
                    "service": s.service,
                    "gross_pnl": s.gross_pnl,
                    "net_pnl": s.net_pnl,
                    "return_contribution": f"{s.net_pnl / self.portfolio_value_start:.2%}",
                }
                for s in sorted(self.services, key=lambda x: -x.net_pnl)
            ],
            "factor_attribution": {
                k: f"{v:.4%}" for k, v in self.factor_attribution.items()
            }
        }

class AttributionEngine:
    """
    Full attribution engine for Purple Flea multi-service agents.
    Pulls data from the Purple Flea reporting API and computes
    BHB attribution, factor attribution, and cost decomposition.
    """
    def __init__(self, api_key: str,
                 base_url: str = "https://purpleflea.com/api"):
        self.api_key = api_key
        self.base_url = base_url
        self.headers = {"Authorization": f"Bearer {api_key}"}

    async def get_monthly_report(self, year: int, month: int) -> AttributionReport:
        """Fetch all service P&Ls from Purple Flea API and generate attribution."""
        import aiohttp
        from calendar import monthrange
        from datetime import date
        _, last_day = monthrange(year, month)
        start = date(year, month, 1)
        end   = date(year, month, last_day)

        async with aiohttp.ClientSession(headers=self.headers) as session:
            # Fetch portfolio snapshot
            async with session.get(
                f"{self.base_url}/portfolio/snapshot",
                params={"date": str(start)}
            ) as r:
                snap = await r.json()
            pv_start = snap["portfolio_value_usd"]

            # Fetch P&L by service
            services_pnl = []
            for svc in ["casino", "trading", "wallet", "domains", "faucet", "escrow"]:
                async with session.get(
                    f"{self.base_url}/{svc}/pnl",
                    params={"from": str(start), "to": str(end)}
                ) as r:
                    data = await r.json()
                    services_pnl.append(ServicePnL(
                        service=svc,
                        gross_pnl=data.get("gross_pnl", 0),
                        fees_paid=data.get("fees_paid", 0),
                        slippage=data.get("slippage", 0)
                    ))

            # Fetch trading factor exposures
            async with session.get(
                f"{self.base_url}/trading/factor-attribution",
                params={"from": str(start), "to": str(end)}
            ) as r:
                factors = await r.json()

        pv_end = pv_start + sum(s.net_pnl for s in services_pnl)
        return AttributionReport(
            period_start=start,
            period_end=end,
            portfolio_value_start=pv_start,
            portfolio_value_end=pv_end,
            services=services_pnl,
            factor_attribution=factors.get("contributions", {}),
            benchmark_return=0.07  # BTC 70/30 benchmark monthly return
        )

    async def print_report(self, year: int, month: int):
        report = await self.get_monthly_report(year, month)
        print(json.dumps(report.to_report_dict(), indent=2))

Monthly Attribution Report Template

A well-structured monthly attribution report should answer five questions in order: What happened? Why did it happen? What did it cost? How did it compare to doing nothing? What should we change?

monthly_report_generator.pyfrom attribution_engine import AttributionEngine
from datetime import date
import asyncio

async def generate_monthly_report(api_key: str, year: int, month: int):
    engine = AttributionEngine(api_key=api_key)
    report = await engine.get_monthly_report(year, month)
    d = report.to_report_dict()

    print("=" * 60)
    print(f"PURPLE FLEA AGENT — MONTHLY ATTRIBUTION REPORT")
    print(f"Period: {d['period']}")
    print("=" * 60)

    print("\n1. PERFORMANCE SUMMARY")
    print(f"   Gross Return:    {d['returns']['gross']}")
    print(f"   Net Return:      {d['returns']['net']}")
    print(f"   Benchmark:       {d['returns']['benchmark']} (BTC/USDT 70/30)")
    print(f"   Active Alpha:    {d['returns']['active_alpha']}")

    print("\n2. SERVICE ATTRIBUTION (ranked by net P&L)")
    for svc in d['services']:
        bar = '█' * max(1, int(float(svc['return_contribution'].strip('%')) * 3))
        print(f"   {svc['service']:10s}  {bar:<20s}  {svc['return_contribution']}")

    print("\n3. FACTOR ATTRIBUTION")
    for factor, contrib in d['factor_attribution'].items():
        print(f"   {factor:15s}: {contrib}")

    print("\n4. COST ATTRIBUTION")
    print(f"   Total Fees:      ${d['costs']['total_fees']:,.2f}")
    print(f"   Total Slippage:  ${d['costs']['total_slippage']:,.2f}")
    print(f"   Fee Drag:        {d['costs']['fee_drag_pct']}")

    print("\n5. PORTFOLIO VALUES")
    print(f"   Start: ${d['portfolio_start']:,.2f}")
    print(f"   End:   ${d['portfolio_end']:,.2f}")
    print("=" * 60)

# Run for current month
asyncio.run(generate_monthly_report(
    api_key="pf_live_",
    year=2026, month=3
))

Interactive Attribution Charts (SVG)

For dashboard integration, the attribution engine can output SVG-based charts that render without any JavaScript dependencies — useful for agent-generated reports, email summaries, and headless environments.

The Attribution Action Loop

Attribution is only valuable if it changes behavior. The action loop is:

1. Run attribution monthly (minimum; weekly is better for fast-moving strategies)
2. Identify the top 3 costs — the three service/strategy combinations with the highest fee + slippage drag relative to their contribution
3. Check factor exposures — are you being paid for your factors? If momentum beta is 1.2 but momentum factor return was negative this month, your strategy is momentum-exposed without edge
4. Scale winners, investigate losers — increase capital to services with positive active return; put strategies with negative alpha on a 30-day review with reduced capital
5. Monitor fee drag trend — if fee drag is rising month-over-month, you're likely increasing turnover without proportional alpha improvement

Summary

Portfolio attribution transforms a single P&L number into an actionable performance map. For agents operating across Purple Flea's multi-service ecosystem, attribution at four levels — service, strategy, factor, and cost — is the difference between scaling confidently and flying blind.

The Brinson-Hood-Beebower model provides the service-level decomposition. Factor regression isolates momentum, mean-reversion, and carry exposures within the trading component. Fee drag quantification makes the cost of trading visible at the strategy level. And the AttributionEngine class ties it all together into a monthly report that answers the one question every agent should be able to answer: where, exactly, did my returns come from?

Use attribution monthly. Act on it. Compounding improves dramatically when you know which levers actually work.