# Exchange Trade Reporting > "Real-time trade reporting and execution analytics for monitoring and" optimization - Skill: `paulpas/exchange-trade-reporting` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/exchange-trade-reporting` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/exchange-trade-reporting/raw - Safety review: PASS (external: skill-scanner PASS, skillspector PASS) - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Data & Analytics - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/exchange-trade-reporting --- **Role:** Generate and analyze trade reports for performance monitoring and regulatory compliance **Philosophy:** Execution quality directly impacts PnL; trade reporting provides the feedback loop for continuous improvement ## Key Principles 1. **Event Streaming**: Real-time trade events for immediate feedback 2. **Aggregated Metrics**: Volume-weighted, time-weighted, and volume-profile metrics 3. **Performance Attribution**: Break down PnL by source (alpha, slippage, fees) 4. **Anomaly Detection**: Flag unusual execution patterns for review 5. **Compliance Tracking**: Maintain audit trail for regulatory requirements ## Implementation Guidelines ### Structure - Core logic: reporting/trade_reporter.py - Metrics: reporting/metrics_calculator.py - Tests: tests/test_trade_reporting.py ### Patterns to Follow - Use event-driven architecture for real-time reporting - Support batch and streaming report generation - Implement customizable report templates - Track reporting latency ## Adherence Checklist Before completing your task, verify: - [ ] Trade events are processed in real-time - [ ] Slippage calculations use benchmark prices - [ ] Reports are generated within SLA timeframes - [ ] Anomaly detection triggers alerts - [ ] Audit trail is immutable Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory. ## Python Implementation ```python import time import uuid from typing import Dict, List, Optional, Callable from dataclasses import dataclass, field from enum import Enum from datetime import datetime import logging class TradeSide(Enum): BUY = "buy" SELL = "sell" class TradeType(Enum): MARKET = "market" LIMIT = "limit" STOP = "stop" OTHER = "other" @dataclass class Trade: """Represents a single trade execution.""" id: str order_id: str symbol: str side: TradeSide quantity: float price: float timestamp: float commission: float = 0.0 fees: float = 0.0 exchange: str = "" trade_type: TradeType = TradeType.MARKET tags: Dict[str, str] = field(default_factory=dict) @property def notional_value(self) -> float: return self.quantity * self.price @property def total_cost(self) -> float: return self.notional_value + self.commission + self.fees @dataclass class ExecutionReport: """Execution report for a single order.""" order_id: str symbol: str total_quantity: float filled_quantity: float avg_price: float remaining_quantity: float filled_notional: float total_commission: float total_fees: float start_time: float end_time: float trades: List[Trade] = field(default_factory=list) @property def fill_percentage(self) -> float: if self.total_quantity == 0: return 0.0 return (self.filled_quantity / self.total_quantity) * 100 @property def slippage(self) -> Optional[float]: """Calculate slippage if benchmark price is available.""" if self.benchmark_price and self.avg_price: if self.trades and self.trades[0].side == TradeSide.BUY: return self.avg_price - self.benchmark_price return self.benchmark_price - self.avg_price return None benchmark_price: Optional[float] = None class TradeReporter: """Generates and manages trade reports.""" def __init__(self, event_callback: Optional[Callable] = None): self.trades: List[Trade] = [] self.reports: Dict[str, ExecutionReport] = {} self.order_executions: Dict[str, List[Trade]] = {} self.event_callback = event_callback self.alert_callbacks: List[Callable] = [] def record_trade(self, trade: Trade): """Record a new trade execution.""" self.trades.append(trade) # Group by order if trade.order_id not in self.order_executions: self.order_executions[trade.order_id] = [] self.order_executions[trade.order_id].append(trade) # Update report self._update_report(trade) # Trigger event if self.event_callback: self.event_callback("trade_recorded", trade) def _update_report(self, trade: Trade): """Update execution report with new trade.""" if trade.order_id not in self.reports: self.reports[trade.order_id] = ExecutionReport( order_id=trade.order_id, symbol=trade.symbol, total_quantity=trade.quantity, filled_quantity=0, avg_price=0, remaining_quantity=trade.quantity, filled_notional=0, total_commission=0, total_fees=0, start_time=trade.timestamp, end_time=trade.timestamp ) report = self.reports[trade.order_id] report.trades.append(trade) # Update aggregates report.filled_quantity += trade.quantity report.filled_notional += trade.notional_value report.total_commission += trade.commission report.total_fees += trade.fees report.end_time = trade.timestamp # Recalculate average price if report.filled_quantity > 0: report.avg_price = report.filled_notional / report.filled_quantity report.remaining_quantity = report.total_quantity - report.filled_quantity def get_order_report(self, order_id: str) -> Optional[ExecutionReport]: """Get execution report for specific order.""" return self.reports.get(order_id) def get_order_trades(self, order_id: str) -> List[Trade]: """Get all trades for specific order.""" return self.order_executions.get(order_id, []) def get_execution_metrics(self, order_id: str) -> Dict: """Get detailed execution metrics for an order.""" report = self.reports.get(order_id) if not report: return {} metrics = { "order_id": order_id, "symbol": report.symbol, "fill_rate": report.fill_percentage, "avg_price": report.avg_price, "total_cost": report.filled_notional + report.total_commission + report.total_fees, "duration_seconds": report.end_time - report.start_time, "trade_count": len(report.trades), "slippage": report.slippage } # Per-trade metrics if report.trades: metrics.update({ "min_fill_size": min(t.quantity for t in report.trades), "max_fill_size": max(t.quantity for t in report.trades), "avg_fill_size": sum(t.quantity for t in report.trades) / len(report.trades) }) return metrics def generate_summary_report(self, start_time: Optional[float] = None, end_time: Optional[float] = None) -> Dict: """Generate summary report for time period.""" trades = self.trades if start_time: trades = [t for t in trades if t.timestamp >= start_time] if end_time: trades = [t for t in trades if t.timestamp <= end_time] if not trades: return {"error": "No trades in period"} total_volume = sum(t.quantity for t in trades) total_notional = sum(t.notional_value for t in trades) total_commission = sum(t.commission for t in trades) total_fees = sum(t.fees for t in trades) # Breakdown by side buys = [t for t in trades if t.side == TradeSide.BUY] sells = [t for t in trades if t.side == TradeSide.SELL] return { "period_start": start_time or trades[0].timestamp, "period_end": end_time or trades[-1].timestamp, "total_trades": len(trades), "total_volume": total_volume, "total_notional": total_notional, "total_cost": total_notional + total_commission + total_fees, "total_commission": total_commission, "total_fees": total_fees, "buy_volume": sum(t.quantity for t in buys), "sell_volume": sum(t.quantity for t in sells), "avg_fill_size": total_volume / len(trades), "trades_by_symbol": self._group_by_symbol(trades), "trades_by_exchange": self._group_by_exchange(trades) } def _group_by_symbol(self, trades: List[Trade]) -> Dict[str, Dict]: """Group trades by symbol.""" groups: Dict[str, List[Trade]] = {} for trade in trades: if trade.symbol not in groups: groups[trade.symbol] = [] groups[trade.symbol].append(trade) return { symbol: { "volume": sum(t.quantity for t in trades), "notional": sum(t.notional_value for t in trades), "count": len(trades) } for symbol, trades in groups.items() } def _group_by_exchange(self, trades: List[Trade]) -> Dict[str, Dict]: """Group trades by exchange.""" groups: Dict[str, List[Trade]] = {} for trade in trades: ex = trade.exchange or "unknown" if ex not in groups: groups[ex] = [] groups[ex].append(trade) return { ex: { "volume": sum(t.quantity for t in trades), "notional": sum(t.notional_value for t in trades), "count": len(trades) } for ex, trades in groups.items() } def register_alert_callback(self, callback: Callable): """Register callback for anomaly alerts.""" self.alert_callbacks.append(callback) def check_anomalies(self, trade: Trade) -> List[str]: """Check trade for anomalies.""" anomalies = [] # Large fill check (2x average) if len(self.trades) > 10: avg_fill = sum(t.quantity for t in self.trades[-10:]) / 10 if trade.quantity > avg_fill * 2: anomalies.append(f"Large fill: {trade.quantity} (avg: {avg_fill:.2f})") # Price deviation check (5% from mid) if trade.tags.get("expected_price") and trade.tags.get("mid_price"): expected = float(trade.tags["expected_price"]) mid = float(trade.tags["mid_price"]) deviation = abs(trade.price - mid) / mid if deviation > 0.05: anomalies.append(f"Large price deviation: {deviation:.2%}") # Timestamp anomaly (before start) if trade.timestamp < (time.time() - 3600): # More than 1 hour old anomalies.append("Stale trade timestamp") return anomalies class PerformanceTracker: """Tracks execution performance metrics.""" def __init__(self): self.volume_by_time: Dict[str, float] = {} self.vwap_by_time: Dict[str, float] = {} self.slippage_by_time: Dict[str, float] = {} self.start_time = time.time() def record_execution(self, trade: Trade, benchmark_price: Optional[float] = None): """Record execution for tracking.""" hour_key = datetime.fromtimestamp(trade.timestamp).strftime("%Y-%m-%d %H") if hour_key not in self.volume_by_time: self.volume_by_time[hour_key] = 0 self.vwap_by_time[hour_key] = 0 self.slippage_by_time[hour_key] = 0 # Update volume-weighted average old_volume = self.volume_by_time[hour_key] new_volume = old_volume + trade.quantity self.volume_by_time[hour_key] = new_volume old_vwap = self.vwap_by_time[hour_key] self.vwap_by_time[hour_key] = ( (old_vwap * old_volume + trade.price * trade.quantity) / new_volume ) # Track slippage if benchmark_price: slippage = abs(trade.price - benchmark_price) / benchmark_price self.slippage_by_time[hour_key] += slippage * trade.quantity def get_performance_summary(self) -> Dict: """Get performance summary.""" if not self.volume_by_time: return {"error": "No execution data"} total_volume = sum(self.volume_by_time.values()) total_slippage = sum(self.slippage_by_time.values()) return { "start_time": self.start_time, "total_executions": len(self.volume_by_time), "total_volume": total_volume, "avg_vwap": sum(self.vwap_by_time.values()) / len(self.vwap_by_time), "avg_slippage_bps": (total_slippage / total_volume * 10000) if total_volume > 0 else 0, "volume_by_hour": dict(sorted(self.volume_by_time.items())), "vwap_by_hour": dict(sorted(self.vwap_by_time.items())) } ``` --- --- ### Pattern 2: Risk-Managed Trading Logic with Validation ```python from __future__ import annotations import logging from dataclasses import dataclass from typing import Optional logger = logging.getLogger(__name__) @dataclass(frozen=True) class TradeSignal: """Immutable trade signal with all required validation constraints.""" symbol: str side: str # "buy" or "sell" price: float quantity: float confidence: float # 0.0 to 1.0 reason: str def validate(self) -> bool: """Validate that the trade signal meets all business constraints.""" if self.quantity <= 0: raise ValueError(f"Quantity must be positive, got {self.quantity}") if self.price <= 0: raise ValueError(f"Price must be positive, got {self.price}") if not 0.0 <= self.confidence <= 1.0: raise ValueError(f"Confidence must be between 0 and 1, got {self.confidence}") return True def generate_trade_signal( symbol: str, side: str, price: float, quantity: float, confidence: float, reason: str, ) -> TradeSignal: """Generate a validated trade signal with guard clause checks.""" if side not in ("buy", "sell"): raise ValueError(f"Invalid side '{side}', must be 'buy' or 'sell'") signal = TradeSignal( symbol=symbol, side=side, price=price, quantity=quantity, confidence=confidence, reason=reason, ) signal.validate() logger.info("Trade signal generated: %s %s %.4f @ %.2f (confidence=%.2f)", symbol, side, quantity, price, confidence) return signal def execute_with_risk_check(signal: TradeSignal, max_position_pct: float = 0.05) -> dict: """Execute a trade signal after applying risk management checks.""" adjusted_quantity = signal.quantity if signal.side == "buy" and signal.quantity > max_position_pct: logger.warning("Position %s exceeds max %.1f%% — capping to %.4f", signal.symbol, max_position_pct * 100, max_position_pct) adjusted_quantity = max_position_pct return { "symbol": signal.symbol, "side": signal.side, "price": signal.price, "quantity": adjusted_quantity, "capped": adjusted_quantity < signal.quantity, "confidence": signal.confidence, "status": "submitted", } ``` ## Constraints ### MUST DO - Implement a unified adapter interface across all exchange integrations to standardize order placement, cancellation, and querying - Handle rate limiting proactively with token bucket or leaky bucket algorithms — never wait for 429 responses before slowing down - Maintain local order state as the source of truth; reconcile with exchange state periodically via webhook events and polling - Implement heartbeat monitoring per exchange connection with automatic failover to a secondary data feed on timeout - Log all API interactions including request/response IDs, timing, and status codes for audit and debugging ### MUST NOT DO - Do not trust exchange-reported order states without local confirmation — always reconcile after every state change - Avoid sending multiple orders for the same position simultaneously across different adapters or sessions - Never store API keys or secrets in code — use environment variables or a secrets manager with automatic rotation - Do not assume all exchanges support the same order types — implement graceful degradation with clear capability negotiation - Avoid polling-based price updates when WebSocket/streaming APIs are available — polling creates unnecessary load and latency ## Live References > Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content. - [Trade Reports Tutorial](https://docs.quantconnect.com/tutorials/trade-reports) - [Trade Execution Confirmation](https://www.investopedia.com/terms/f/fill.asp) - [Regulatory Trade Reporting Requirements](https://en.wikipedia.org/wiki/Trade_reporting) - [Order Lifecycle Tracking](https://docs.quantconnect.com/tutorials/algorithms) - [Post-Trade Data Management](https://www.investopedia.com/terms/t/trade.asp)