56. Smart Trade Anomaly Detection System¶

services/trade-anomaly-center/
├── src/
│   ├── main.py
│   ├── listener/
│   │   ├── trade_stream_listener.py
│   │   ├── order_event_listener.py
│   │   └── market_data_listener.py
│   ├── detector/
│   │   ├── anomaly_detector.py
│   │   ├── pattern_analyzer.py
│   │   ├── ml_anomaly_detector.py
│   │   └── correlation_detector.py
│   ├── responder/
│   │   ├── risk_responder.py
│   │   ├── alert_manager.py
│   │   ├── strategy_controller.py
│   │   └── account_freezer.py
│   ├── logger/
│   │   ├── anomaly_logger.py
│   │   ├── audit_trail.py
│   │   └── performance_metrics.py
│   ├── api/
│   │   ├── anomaly_api.py
│   ├── config.py
│   └── requirements.txt
├── Dockerfile
└── tests/

class TradeStreamListener:
    def __init__(self, order_event_listener, market_data_listener):
        self.order_event_listener = order_event_listener
        self.market_data_listener = market_data_listener
        self.trade_history = []
        self.order_history = []
        self.active_monitors = {}

    async def start_monitoring(self):
        """Start monitoring all trading activities"""

        # Monitor trade executions
        await self.monitor_trade_executions()

        # Monitor order events
        await self.monitor_order_events()

        # Monitor market data
        await self.monitor_market_data()

        # Monitor system events
        await self.monitor_system_events()

    async def monitor_trade_executions(self):
        """Monitor trade execution events"""

        trade_monitor = TradeExecutionMonitor()
        await trade_monitor.start_monitoring()

        async for trade in trade_monitor.get_trades():
            await self.process_trade_event(trade)

    async def monitor_order_events(self):
        """Monitor order lifecycle events"""

        order_monitor = OrderEventMonitor()
        await order_monitor.start_monitoring()

        async for order_event in order_monitor.get_order_events():
            await self.process_order_event(order_event)

    async def monitor_market_data(self):
        """Monitor market data for price anomalies"""

        market_monitor = MarketDataMonitor()
        await market_monitor.start_monitoring()

        async for market_update in market_monitor.get_market_updates():
            await self.process_market_update(market_update)

    async def monitor_system_events(self):
        """Monitor system-level events"""

        system_monitor = SystemEventMonitor()
        await system_monitor.start_monitoring()

        async for system_event in system_monitor.get_system_events():
            await self.process_system_event(system_event)

    async def process_trade_event(self, trade_event):
        """Process a trade execution event"""

        trade_record = {
            "timestamp": datetime.now().isoformat(),
            "trade_id": trade_event.get("trade_id"),
            "symbol": trade_event.get("symbol"),
            "side": trade_event.get("side"),
            "quantity": trade_event.get("quantity"),
            "price": trade_event.get("price"),
            "amount": trade_event.get("amount"),
            "account_id": trade_event.get("account_id"),
            "strategy_id": trade_event.get("strategy_id"),
            "counterparty": trade_event.get("counterparty"),
            "venue": trade_event.get("venue"),
            "order_id": trade_event.get("order_id")
        }

        # Store trade record
        await self.store_trade_record(trade_record)

        # Trigger anomaly detection
        await self.trigger_trade_anomaly_detection(trade_record)

    async def process_order_event(self, order_event):
        """Process an order lifecycle event"""

        order_record = {
            "timestamp": datetime.now().isoformat(),
            "order_id": order_event.get("order_id"),
            "symbol": order_event.get("symbol"),
            "side": order_event.get("side"),
            "order_type": order_event.get("order_type"),
            "quantity": order_event.get("quantity"),
            "price": order_event.get("price"),
            "status": order_event.get("status"),
            "account_id": order_event.get("account_id"),
            "strategy_id": order_event.get("strategy_id"),
            "action": order_event.get("action"),  # new, modify, cancel, fill
            "venue": order_event.get("venue")
        }

        # Store order record
        await self.store_order_record(order_record)

        # Trigger order anomaly detection
        await self.trigger_order_anomaly_detection(order_record)

    async def process_market_update(self, market_update):
        """Process market data update"""

        market_record = {
            "timestamp": datetime.now().isoformat(),
            "symbol": market_update.get("symbol"),
            "bid": market_update.get("bid"),
            "ask": market_update.get("ask"),
            "bid_size": market_update.get("bid_size"),
            "ask_size": market_update.get("ask_size"),
            "last_price": market_update.get("last_price"),
            "volume": market_update.get("volume"),
            "venue": market_update.get("venue")
        }

        # Store market record
        await self.store_market_record(market_record)

        # Trigger market anomaly detection
        await self.trigger_market_anomaly_detection(market_record)

    async def store_trade_record(self, trade_record):
        """Store trade record"""

        # Store in database
        await self.save_trade_to_database(trade_record)

        # Add to history
        self.trade_history.append(trade_record)

        # Keep history manageable
        if len(self.trade_history) > 10000:
            self.trade_history = self.trade_history[-10000:]

    async def store_order_record(self, order_record):
        """Store order record"""

        # Store in database
        await self.save_order_to_database(order_record)

        # Add to history
        self.order_history.append(order_record)

        # Keep history manageable
        if len(self.order_history) > 10000:
            self.order_history = self.order_history[-10000:]

    async def get_recent_trades(self, minutes: int = 60) -> List[Dict]:
        """Get recent trades"""

        cutoff_time = datetime.now() - timedelta(minutes=minutes)

        recent_trades = [
            trade for trade in self.trade_history
            if datetime.fromisoformat(trade["timestamp"]) > cutoff_time
        ]

        return recent_trades

    async def get_recent_orders(self, minutes: int = 60) -> List[Dict]:
        """Get recent orders"""

        cutoff_time = datetime.now() - timedelta(minutes=minutes)

        recent_orders = [
            order for order in self.order_history
            if datetime.fromisoformat(order["timestamp"]) > cutoff_time
        ]

        return recent_orders

class AnomalyDetector:
    def __init__(self, pattern_analyzer, ml_anomaly_detector):
        self.pattern_analyzer = pattern_analyzer
        self.ml_anomaly_detector = ml_anomaly_detector
        self.detection_rules = self.load_detection_rules()
        self.anomaly_history = []

    async def detect_trade_anomalies(self, trade_record: Dict) -> List[Dict]:
        """Detect anomalies in trade execution"""

        anomalies = []

        # Large trade detection
        large_trade_anomaly = await self.detect_large_trade(trade_record)
        if large_trade_anomaly:
            anomalies.append(large_trade_anomaly)

        # Price jump detection
        price_jump_anomaly = await self.detect_price_jump(trade_record)
        if price_jump_anomaly:
            anomalies.append(price_jump_anomaly)

        # Unusual timing detection
        timing_anomaly = await self.detect_unusual_timing(trade_record)
        if timing_anomaly:
            anomalies.append(timing_anomaly)

        # Counterparty behavior detection
        counterparty_anomaly = await self.detect_counterparty_anomaly(trade_record)
        if counterparty_anomaly:
            anomalies.append(counterparty_anomaly)

        # ML-based anomaly detection
        ml_anomaly = await self.ml_anomaly_detector.detect_anomaly(trade_record)
        if ml_anomaly:
            anomalies.append(ml_anomaly)

        return anomalies

    async def detect_order_anomalies(self, order_record: Dict) -> List[Dict]:
        """Detect anomalies in order behavior"""

        anomalies = []

        # High-frequency cancellation detection
        hf_cancel_anomaly = await self.detect_high_frequency_cancellation(order_record)
        if hf_cancel_anomaly:
            anomalies.append(hf_cancel_anomaly)

        # Order size anomaly detection
        size_anomaly = await self.detect_order_size_anomaly(order_record)
        if size_anomaly:
            anomalies.append(size_anomaly)

        # Order pattern anomaly detection
        pattern_anomaly = await self.detect_order_pattern_anomaly(order_record)
        if pattern_anomaly:
            anomalies.append(pattern_anomaly)

        # Market manipulation detection
        manipulation_anomaly = await self.detect_market_manipulation(order_record)
        if manipulation_anomaly:
            anomalies.append(manipulation_anomaly)

        return anomalies

    async def detect_large_trade(self, trade_record: Dict) -> Optional[Dict]:
        """Detect unusually large trades"""

        symbol = trade_record["symbol"]
        amount = trade_record["amount"]

        # Get average trade amount for symbol
        avg_amount = await self.get_average_trade_amount(symbol)

        # Check if trade is significantly larger than average
        if avg_amount > 0 and amount > avg_amount * self.detection_rules["large_trade_threshold"]:
            return {
                "type": "large_trade_anomaly",
                "severity": "medium",
                "description": f"Trade amount {amount} exceeds {self.detection_rules['large_trade_threshold']}x average ({avg_amount})",
                "trade_record": trade_record,
                "threshold": avg_amount * self.detection_rules["large_trade_threshold"],
                "average_amount": avg_amount
            }

        return None

    async def detect_price_jump(self, trade_record: Dict) -> Optional[Dict]:
        """Detect unusual price movements"""

        symbol = trade_record["symbol"]
        trade_price = trade_record["price"]

        # Get current market prices
        market_data = await self.get_current_market_data(symbol)

        if market_data:
            best_bid = market_data.get("bid", 0)
            best_ask = market_data.get("ask", 0)

            # Check if trade price is outside normal range
            if best_bid > 0 and best_ask > 0:
                spread = best_ask - best_bid
                mid_price = (best_bid + best_ask) / 2

                # Check if price is outside normal range
                if trade_price > best_ask * (1 + self.detection_rules["price_jump_threshold"]) or \
                   trade_price < best_bid * (1 - self.detection_rules["price_jump_threshold"]):

                    return {
                        "type": "price_jump_anomaly",
                        "severity": "high",
                        "description": f"Trade price {trade_price} outside normal range [{best_bid}, {best_ask}]",
                        "trade_record": trade_record,
                        "market_data": market_data,
                        "price_deviation": abs(trade_price - mid_price) / mid_price
                    }

        return None

    async def detect_high_frequency_cancellation(self, order_record: Dict) -> Optional[Dict]:
        """Detect high-frequency order cancellations"""

        account_id = order_record["account_id"]
        action = order_record["action"]

        if action == "cancel":
            # Get recent cancellations for this account
            recent_cancellations = await self.get_recent_cancellations(account_id, minutes=1)

            if len(recent_cancellations) > self.detection_rules["max_cancellations_per_minute"]:
                return {
                    "type": "high_frequency_cancellation_anomaly",
                    "severity": "high",
                    "description": f"High-frequency cancellations detected: {len(recent_cancellations)} in 1 minute",
                    "order_record": order_record,
                    "cancellation_count": len(recent_cancellations),
                    "threshold": self.detection_rules["max_cancellations_per_minute"]
                }

        return None

    async def detect_order_size_anomaly(self, order_record: Dict) -> Optional[Dict]:
        """Detect unusual order sizes"""

        symbol = order_record["symbol"]
        quantity = order_record["quantity"]

        # Get average order size for symbol
        avg_size = await self.get_average_order_size(symbol)

        # Check if order size is unusual
        if avg_size > 0 and quantity > avg_size * self.detection_rules["large_order_threshold"]:
            return {
                "type": "large_order_anomaly",
                "severity": "medium",
                "description": f"Order size {quantity} exceeds {self.detection_rules['large_order_threshold']}x average ({avg_size})",
                "order_record": order_record,
                "threshold": avg_size * self.detection_rules["large_order_threshold"],
                "average_size": avg_size
            }

        return None

    async def detect_market_manipulation(self, order_record: Dict) -> Optional[Dict]:
        """Detect potential market manipulation"""

        symbol = order_record["symbol"]
        account_id = order_record["account_id"]

        # Check for layering (placing orders on both sides)
        layering_detected = await self.detect_layering(account_id, symbol)
        if layering_detected:
            return {
                "type": "market_manipulation_anomaly",
                "severity": "high",
                "description": "Potential layering detected",
                "order_record": order_record,
                "manipulation_type": "layering"
            }

        # Check for spoofing (placing large orders to move price)
        spoofing_detected = await self.detect_spoofing(account_id, symbol)
        if spoofing_detected:
            return {
                "type": "market_manipulation_anomaly",
                "severity": "high",
                "description": "Potential spoofing detected",
                "order_record": order_record,
                "manipulation_type": "spoofing"
            }

        return None

    async def detect_layering(self, account_id: str, symbol: str) -> bool:
        """Detect layering behavior"""

        # Get recent orders for this account and symbol
        recent_orders = await self.get_recent_orders_by_account_symbol(account_id, symbol, minutes=5)

        # Check if orders are placed on both sides
        buy_orders = [order for order in recent_orders if order["side"] == "buy"]
        sell_orders = [order for order in recent_orders if order["side"] == "sell"]

        return len(buy_orders) > 0 and len(sell_orders) > 0

    async def detect_spoofing(self, account_id: str, symbol: str) -> bool:
        """Detect spoofing behavior"""

        # Get recent large orders that were cancelled
        recent_large_cancelled = await self.get_recent_large_cancelled_orders(account_id, symbol, minutes=10)

        # Check if large orders were placed and quickly cancelled
        return len(recent_large_cancelled) > self.detection_rules["spoofing_threshold"]

    def load_detection_rules(self) -> Dict:
        """Load detection rules"""

        return {
            "large_trade_threshold": 5.0,  # 5x average trade size
            "large_order_threshold": 10.0,  # 10x average order size
            "price_jump_threshold": 0.01,   # 1% outside bid/ask
            "max_cancellations_per_minute": 10,
            "spoofing_threshold": 3,  # 3 large cancelled orders
            "unusual_timing_threshold": 300  # 5 minutes outside normal hours
        }

class RiskResponder:
    def __init__(self, alert_manager, strategy_controller, account_freezer):
        self.alert_manager = alert_manager
        self.strategy_controller = strategy_controller
        self.account_freezer = account_freezer
        self.response_rules = self.load_response_rules()

    async def respond_to_anomalies(self, anomalies: List[Dict]):
        """Respond to detected anomalies"""

        for anomaly in anomalies:
            await self.respond_to_anomaly(anomaly)

    async def respond_to_anomaly(self, anomaly: Dict):
        """Respond to a single anomaly"""

        anomaly_type = anomaly["type"]
        severity = anomaly["severity"]

        # Get response actions based on type and severity
        response_actions = self.get_response_actions(anomaly_type, severity)

        # Execute response actions
        for action in response_actions:
            await self.execute_response_action(action, anomaly)

    def get_response_actions(self, anomaly_type: str, severity: str) -> List[str]:
        """Get response actions for anomaly type and severity"""

        if severity == "high":
            if anomaly_type in ["price_jump_anomaly", "market_manipulation_anomaly"]:
                return ["alert_team", "freeze_account", "stop_strategy"]
            elif anomaly_type == "high_frequency_cancellation_anomaly":
                return ["alert_team", "freeze_account"]
            else:
                return ["alert_team", "stop_strategy"]

        elif severity == "medium":
            if anomaly_type in ["large_trade_anomaly", "large_order_anomaly"]:
                return ["alert_team", "monitor_account"]
            else:
                return ["alert_team"]

        else:  # low severity
            return ["log_anomaly"]

    async def execute_response_action(self, action: str, anomaly: Dict):
        """Execute a specific response action"""

        if action == "alert_team":
            await self.alert_team(anomaly)

        elif action == "freeze_account":
            await self.freeze_account(anomaly)

        elif action == "stop_strategy":
            await self.stop_strategy(anomaly)

        elif action == "monitor_account":
            await self.monitor_account(anomaly)

        elif action == "log_anomaly":
            await self.log_anomaly(anomaly)

    async def alert_team(self, anomaly: Dict):
        """Send alert to risk management team"""

        alert_message = {
            "timestamp": datetime.now().isoformat(),
            "anomaly_type": anomaly["type"],
            "severity": anomaly["severity"],
            "description": anomaly["description"],
            "entity_id": self.get_entity_id(anomaly),
            "action_required": True
        }

        await self.alert_manager.send_alert(alert_message)

    async def freeze_account(self, anomaly: Dict):
        """Freeze trading account"""

        account_id = self.get_account_id(anomaly)
        if account_id:
            await self.account_freezer.freeze_account(account_id, anomaly)

    async def stop_strategy(self, anomaly: Dict):
        """Stop trading strategy"""

        strategy_id = self.get_strategy_id(anomaly)
        if strategy_id:
            await self.strategy_controller.stop_strategy(strategy_id, anomaly)

    async def monitor_account(self, anomaly: Dict):
        """Increase monitoring for account"""

        account_id = self.get_account_id(anomaly)
        if account_id:
            await self.account_freezer.increase_monitoring(account_id)

    async def log_anomaly(self, anomaly: Dict):
        """Log anomaly for record keeping"""

        await self.anomaly_logger.log_anomaly(anomaly)

    def get_entity_id(self, anomaly: Dict) -> str:
        """Get entity ID from anomaly"""

        if "trade_record" in anomaly:
            return anomaly["trade_record"].get("account_id", "")
        elif "order_record" in anomaly:
            return anomaly["order_record"].get("account_id", "")
        else:
            return ""

    def get_account_id(self, anomaly: Dict) -> Optional[str]:
        """Get account ID from anomaly"""

        return self.get_entity_id(anomaly)

    def get_strategy_id(self, anomaly: Dict) -> Optional[str]:
        """Get strategy ID from anomaly"""

        if "trade_record" in anomaly:
            return anomaly["trade_record"].get("strategy_id")
        elif "order_record" in anomaly:
            return anomaly["order_record"].get("strategy_id")
        else:
            return None

    def load_response_rules(self) -> Dict:
        """Load response rules"""

        return {
            "high_severity_actions": ["alert_team", "freeze_account", "stop_strategy"],
            "medium_severity_actions": ["alert_team", "monitor_account"],
            "low_severity_actions": ["log_anomaly"],
            "auto_freeze_threshold": 3,  # 3 high-severity anomalies
            "auto_stop_threshold": 5     # 5 medium-severity anomalies
        }

@router.get("/anomalies/recent")
async def get_recent_anomalies(minutes: int = 60):
    """Get recent anomalies"""

    cutoff_time = datetime.now() - timedelta(minutes=minutes)

    anomalies = await anomaly_detector.get_recent_anomalies(cutoff_time)

    return {
        "anomalies": anomalies,
        "total_count": len(anomalies),
        "high_severity": len([a for a in anomalies if a["severity"] == "high"]),
        "medium_severity": len([a for a in anomalies if a["severity"] == "medium"]),
        "low_severity": len([a for a in anomalies if a["severity"] == "low"])
    }

@router.get("/anomalies/account/{account_id}")
async def get_account_anomalies(account_id: str, hours: int = 24):
    """Get anomalies for specific account"""

    anomalies = await anomaly_detector.get_anomalies_by_account(account_id, hours)

    return {
        "account_id": account_id,
        "anomalies": anomalies,
        "total_count": len(anomalies),
        "anomaly_types": list(set(a["type"] for a in anomalies))
    }

@router.get("/anomalies/strategy/{strategy_id}")
async def get_strategy_anomalies(strategy_id: str, hours: int = 24):
    """Get anomalies for specific strategy"""

    anomalies = await anomaly_detector.get_anomalies_by_strategy(strategy_id, hours)

    return {
        "strategy_id": strategy_id,
        "anomalies": anomalies,
        "total_count": len(anomalies),
        "anomaly_types": list(set(a["type"] for a in anomalies))
    }

@router.post("/anomalies/acknowledge/{anomaly_id}")
async def acknowledge_anomaly(anomaly_id: str, user_id: str):
    """Acknowledge an anomaly"""

    await anomaly_detector.acknowledge_anomaly(anomaly_id, user_id)

    return {"status": "acknowledged", "anomaly_id": anomaly_id}

@router.get("/anomalies/statistics")
async def get_anomaly_statistics(days: int = 7):
    """Get anomaly statistics"""

    statistics = await anomaly_detector.get_anomaly_statistics(days)

    return {
        "period_days": days,
        "statistics": statistics,
        "trends": await anomaly_detector.get_anomaly_trends(days)
    }

@router.get("/trades/suspicious")
async def get_suspicious_trades(hours: int = 24):
    """Get suspicious trades"""

    suspicious_trades = await trade_stream_listener.get_suspicious_trades(hours)

    return {
        "suspicious_trades": suspicious_trades,
        "total_count": len(suspicious_trades),
        "time_period_hours": hours
    }

const AnomalyDetectionView: React.FC = () => {
  const [recentAnomalies, setRecentAnomalies] = useState<Anomaly[]>([]);
  const [anomalyStatistics, setAnomalyStatistics] = useState<AnomalyStatistics | null>(null);
  const [suspiciousTrades, setSuspiciousTrades] = useState<Trade[]>([]);
  const [selectedAccount, setSelectedAccount] = useState<string>("");

  return (
    <div className="anomaly-detection-dashboard">
      {/* Real-time Anomaly Alerts */}
      <AnomalyAlertsPanel 
        anomalies={recentAnomalies}
        onAnomalySelect={handleAnomalySelect}
      />

      {/* Anomaly Statistics */}
      <AnomalyStatisticsPanel 
        statistics={anomalyStatistics}
      />

      {/* Anomaly Timeline */}
      <AnomalyTimelinePanel 
        anomalies={recentAnomalies}
      />

      {/* Suspicious Trades */}
      <SuspiciousTradesPanel 
        trades={suspiciousTrades}
        onTradeSelect={handleTradeSelect}
      />

      {/* Account Anomaly Analysis */}
      <AccountAnomalyPanel 
        selectedAccount={selectedAccount}
        anomalies={recentAnomalies}
      />

      {/* Anomaly Type Breakdown */}
      <AnomalyTypeBreakdownPanel 
        anomalies={recentAnomalies}
      />

      {/* Risk Response Actions */}
      <RiskResponsePanel 
        anomalies={recentAnomalies}
        onActionExecute={handleActionExecute}
      />

      {/* Anomaly Trends */}
      <AnomalyTrendsPanel 
        statistics={anomalyStatistics}
      />
    </div>
  );
};