Crypto Tax Guide for AI Agents: Wash Sales, Harvesting, and Automated Reporting

High-frequency AI agents can generate thousands of taxable events per day. Without automated tax tracking, you are flying blind at year-end — and almost certainly overpaying. This guide covers lot selection, wash sale analysis, and a complete Python TaxTracker implementation.

Crypto Tax Basics for Agents

In most jurisdictions, cryptocurrency is treated as property. Every sale, swap, or conversion is a taxable event. For AI agents executing hundreds of trades per day, this creates massive record-keeping obligations.

Key tax events your agent generates:

• Spot sale: Selling crypto for USDC or fiat — triggers capital gain/loss
• Crypto-to-crypto swap: Trading BTC for ETH — each swap is a disposal
• Futures settlement: Realized P&L on futures contracts
• Staking rewards / faucet claims: Ordinary income at fair market value when received
• Referral income: Ordinary income when credited to your wallet

Lot Selection Methods: HIFO, FIFO, LIFO

When you sell a fraction of your holdings, you must identify which specific units you are selling. This choice directly determines your gain or loss. The three standard methods are:

For most high-frequency AI agents, HIFO minimizes total capital gains across the year. However, HIFO must be combined with wash sale analysis to ensure harvested losses are not disallowed.

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Literal
import heapq

@dataclass
class TaxLot:
    lot_id: str
    symbol: str
    quantity: float
    cost_basis: float       # per-unit cost in USD
    acquired_at: datetime
    source: str             # 'purchase', 'faucet', 'referral'

class LotSelector:
    def __init__(self, method: Literal["HIFO", "FIFO", "LIFO"] = "HIFO"):
        self.method = method
        self.lots: dict[str, list[TaxLot]] = {}

    def add_lot(self, lot: TaxLot):
        self.lots.setdefault(lot.symbol, []).append(lot)

    def select_lots(self, symbol: str, quantity: float) -> list[tuple[TaxLot, float]]:
        """Return list of (lot, qty_used) to cover the sale."""
        available = [l for l in self.lots.get(symbol, []) if l.quantity > 0]
        if self.method == "HIFO":
            available.sort(key=lambda l: -l.cost_basis)
        elif self.method == "FIFO":
            available.sort(key=lambda l: l.acquired_at)
        elif self.method == "LIFO":
            available.sort(key=lambda l: -l.acquired_at.timestamp())

        selected = []
        remaining = quantity
        for lot in available:
            if remaining <= 0:
                break
            used = min(lot.quantity, remaining)
            selected.append((lot, used))
            remaining -= used

        if remaining > 1e-9:
            raise ValueError(f"Insufficient {symbol} lots: need {quantity}, have {quantity-remaining}")
        return selected

Wash Sale Rules for Crypto

Wash sale rules (selling at a loss and buying back the same asset within 30 days) were historically not applied to cryptocurrency in many jurisdictions because crypto was not classified as a "security." However, regulatory changes are closing this gap. Check your jurisdiction's current rules.

Regardless of legal requirements, tracking potential wash sales is important for two reasons: (1) staying compliant when rules change, and (2) understanding your true economic position vs. your tax position.

def detect_wash_sale(
    loss_sale_date: datetime,
    loss_symbol: str,
    purchases: list[tuple[datetime, str, float]],   # (date, symbol, qty)
    window_days: int = 30
) -> list[dict]:
    """
    Find purchases of the same asset within the wash sale window.
    Window is ±30 days around the loss sale date.
    """
    wash_start = loss_sale_date - timedelta(days=window_days)
    wash_end = loss_sale_date + timedelta(days=window_days)
    flagged = []
    for pdate, sym, qty in purchases:
        if sym == loss_symbol and wash_start <= pdate <= wash_end:
            flagged.append({
                "purchase_date": pdate.isoformat(),
                "symbol": sym,
                "quantity": qty,
                "wash_sale_flag": True,
            })
    return flagged

# If flagged, the loss may be disallowed and added to the cost basis
# of the replacement shares (increasing future deferred gains)

Loss Harvesting Timing

Tax loss harvesting is the practice of intentionally realizing losses to offset capital gains. For AI agents with hundreds of positions, systematic harvesting can save significant tax dollars annually.

The optimal harvesting triggers are:

• Year-end harvesting: December sweep to realize any remaining losses before year close
• Threshold harvesting: Harvest when a position crosses a loss threshold (e.g., -10%)
• Gain-offset harvesting: Harvest losses immediately after realizing a large gain
• Time-based harvesting: Monthly or quarterly systematic review of underwater positions

from datetime import datetime
import json

class HarvestingEngine:
    def __init__(
        self,
        loss_threshold: float = -0.10,   # harvest at -10% loss
        min_harvest_amount: float = 100,   # minimum $100 loss to bother
    ):
        self.loss_threshold = loss_threshold
        self.min_harvest_amount = min_harvest_amount
        self.harvested_this_year: float = 0
        self.realized_gains_this_year: float = 0

    def should_harvest(
        self,
        lot: TaxLot,
        current_price: float,
        year_end_approaching: bool = False
    ) -> bool:
        unrealized = (current_price - lot.cost_basis) * lot.quantity
        pct_change = (current_price - lot.cost_basis) / lot.cost_basis

        if unrealized >= 0:
            return False   # no loss to harvest

        if abs(unrealized) < self.min_harvest_amount:
            return False   # too small to bother

        if pct_change <= self.loss_threshold:
            return True    # below threshold: harvest

        if year_end_approaching and unrealized < 0:
            return True    # December: harvest everything red

        # Offset against gains realized this year
        if self.realized_gains_this_year > 0 and abs(unrealized) > 500:
            return True

        return False

    def harvest(self, lot: TaxLot, current_price: float) -> dict:
        loss = (current_price - lot.cost_basis) * lot.quantity
        self.harvested_this_year += abs(loss)
        return {
            "action": "harvest",
            "symbol": lot.symbol,
            "quantity": lot.quantity,
            "cost_basis": lot.cost_basis,
            "sale_price": current_price,
            "realized_loss": loss,
            "harvested_at": datetime.utcnow().isoformat(),
            "ytd_harvested": self.harvested_this_year,
        }

Cost Basis Tracking with Purple Flea Wallet

The Purple Flea Wallet API provides a complete transaction history endpoint that your TaxTracker can consume. Every deposit, withdrawal, trade settlement, faucet claim, and referral credit is recorded with timestamps and USD values.

import requests

class WalletTaxImporter:
    BASE = "https://purpleflea.com/api/v1"

    def __init__(self, api_key: str):
        self.headers = {"Authorization": f"Bearer {api_key}"}

    def fetch_transactions(self, year: int) -> list[dict]:
        """Fetch all transactions for a tax year from the wallet API."""
        resp = requests.get(
            f"{self.BASE}/wallet/transactions",
            headers=self.headers,
            params={
                "from": f"{year}-01-01T00:00:00Z",
                "to": f"{year}-12-31T23:59:59Z",
                "limit": 10000
            }
        )
        return resp.json()["transactions"]

    def classify(self, txn: dict) -> str:
        """Classify transaction type for tax purposes."""
        t = txn.get("type")
        if t == "faucet_claim": return "ordinary_income"
        if t == "referral_credit": return "ordinary_income"
        if t == "trade_settlement": return "capital_event"
        if t == "escrow_release": return "ordinary_income"
        if t == "casino_win": return "gambling_income"
        return "unknown"

Python TaxTracker Class

The complete TaxTracker integrates lot selection, wash sale detection, and loss harvesting into a single class that persists to SQLite for year-round tracking.

import sqlite3
from datetime import datetime

class TaxTracker:
    def __init__(self, db_path: str = "tax_tracker.db", method: str = "HIFO"):
        self.conn = sqlite3.connect(db_path)
        self.selector = LotSelector(method)
        self.harvester = HarvestingEngine()
        self._init_db()

    def _init_db(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS tax_events (
                id TEXT PRIMARY KEY,
                event_type TEXT,
                symbol TEXT,
                quantity REAL,
                proceeds REAL,
                cost_basis REAL,
                gain_loss REAL,
                holding_period TEXT,
                acquired_at TEXT,
                disposed_at TEXT,
                wash_sale_flag INTEGER DEFAULT 0,
                notes TEXT
            )
        """)
        self.conn.commit()

    def record_purchase(self, lot: TaxLot):
        self.selector.add_lot(lot)

    def record_sale(self, symbol: str, quantity: float, sale_price: float,
                    sale_date: datetime) -> list[dict]:
        """
        Match sale against lots, record tax events, return gain/loss breakdown.
        """
        lots_used = self.selector.select_lots(symbol, quantity)
        events = []

        for lot, qty_used in lots_used:
            proceeds = qty_used * sale_price
            basis = qty_used * lot.cost_basis
            gain_loss = proceeds - basis
            days_held = (sale_date - lot.acquired_at).days
            holding = "long_term" if days_held >= 365 else "short_term"

            event_id = f"{symbol}-{sale_date.timestamp():.0f}-{lot.lot_id}"
            self.conn.execute("""
                INSERT OR IGNORE INTO tax_events
                (id, event_type, symbol, quantity, proceeds, cost_basis,
                 gain_loss, holding_period, acquired_at, disposed_at)
                VALUES (?,?,?,?,?,?,?,?,?,?)
            """, (event_id, "sale", symbol, qty_used, proceeds, basis,
                     gain_loss, holding, lot.acquired_at.isoformat(),
                     sale_date.isoformat()))
            lot.quantity -= qty_used
            events.append({
                "lot_id": lot.lot_id, "qty": qty_used,
                "gain_loss": gain_loss, "holding": holding,
                "days_held": days_held,
            })

        self.conn.commit()
        return events

    def annual_summary(self, year: int) -> dict:
        """Generate IRS Form 8949-style summary."""
        rows = self.conn.execute("""
            SELECT holding_period, SUM(proceeds), SUM(cost_basis), SUM(gain_loss)
            FROM tax_events
            WHERE disposed_at LIKE ? AND event_type = 'sale'
            GROUP BY holding_period
        """, (f"{year}%",)).fetchall()

        summary = {"year": year, "short_term": {}, "long_term": {}}
        for holding, proceeds, basis, gl in rows:
            summary[holding] = {
                "proceeds": proceeds, "cost_basis": basis, "net_gain_loss": gl
            }
        return summary

Automated Report Generation

At year end, your TaxTracker should generate a complete report in CSV format compatible with common tax software (TurboTax, TaxAct, Koinly-import format).

import csv
from io import StringIO

def export_form_8949_csv(tracker: TaxTracker, year: int) -> str:
    """Export CSV compatible with most tax software import formats."""
    rows = tracker.conn.execute("""
        SELECT symbol, quantity, proceeds, cost_basis, gain_loss,
               holding_period, acquired_at, disposed_at, wash_sale_flag
        FROM tax_events
        WHERE disposed_at LIKE ? AND event_type = 'sale'
        ORDER BY disposed_at
    """, (f"{year}%",)).fetchall()

    output = StringIO()
    writer = csv.writer(output)
    writer.writerow([
        "Description", "Date Acquired", "Date Sold",
        "Proceeds", "Cost Basis", "Gain/Loss",
        "Term", "Wash Sale"
    ])
    for sym, qty, proc, basis, gl, holding, acq, disp, wash in rows:
        writer.writerow([
            f"{qty:.8f} {sym}", acq[:10], disp[:10],
            f"{proc:.2f}", f"{basis:.2f}", f"{gl:.2f}",
            "Long-term" if holding == "long_term" else "Short-term",
            "Yes" if wash else "No"
        ])
    return output.getvalue()

Edge Cases and Pitfalls

• Faucet claims: The $1 USDC from Purple Flea's faucet is ordinary income at time of receipt. Record it as income, not as a zero-cost purchase.
• Referral income: The 10-20% referral credits you earn are ordinary income when credited. Track each credit separately.
• Escrow fees: The 1% escrow fee paid to Purple Flea may be deductible as a business expense. Keep separate records.
• Stablecoin swaps: Even USDC-to-USDT swaps are technically taxable disposal events in many jurisdictions.
• Hard forks and airdrops: Generally treated as ordinary income at FMV on receipt date.

Related reading: Fee Optimization Guide • Bankroll Management • Wallet API Complete Guide