55. Smart Capital Flow Monitoring System¶

services/capital-flow-center/
├── src/
│   ├── main.py
│   ├── listener/
│   │   ├── fund_change_listener.py
│   │   ├── balance_monitor.py
│   │   └── event_collector.py
│   ├── classifier/
│   │   ├── fund_change_classifier.py
│   │   ├── flow_categorizer.py
│   │   └── source_identifier.py
│   ├── analyzer/
│   │   ├── fund_flow_analyzer.py
│   │   ├── flow_statistics.py
│   │   └── trend_analyzer.py
│   ├── detector/
│   │   ├── anomaly_detector.py
│   │   ├── threshold_monitor.py
│   │   └── alert_manager.py
│   ├── visualizer/
│   │   ├── flow_visualizer.py
│   │   └── chart_generator.py
│   ├── api/
│   │   ├── capital_flow_api.py
│   ├── config.py
│   └── requirements.txt
├── Dockerfile
└── tests/

class FundChangeListener:
    def __init__(self, balance_monitor, event_collector):
        self.balance_monitor = balance_monitor
        self.event_collector = event_collector
        self.change_history = []

    async def start_monitoring(self):
        """Start monitoring all capital movements"""

        # Monitor account balances
        await self.monitor_account_balances()

        # Monitor trading PnL changes
        await self.monitor_trading_pnl()

        # Monitor external transfers
        await self.monitor_external_transfers()

    async def process_capital_change(self, change_event):
        """Process a capital change event"""

        change_record = {
            "timestamp": datetime.now().isoformat(),
            "account_id": change_event.get("account_id"),
            "change_type": change_event.get("type"),
            "amount": change_event.get("amount"),
            "currency": change_event.get("currency"),
            "source": change_event.get("source"),
            "description": change_event.get("description"),
            "balance_before": change_event.get("balance_before"),
            "balance_after": change_event.get("balance_after")
        }

        # Store change record
        await self.store_change_record(change_record)

        # Trigger classification and anomaly detection
        await self.trigger_analysis(change_record)

    async def get_recent_changes(self, hours: int = 24) -> List[Dict]:
        """Get recent capital changes"""

        cutoff_time = datetime.now() - timedelta(hours=hours)
        recent_changes = [
            change for change in self.change_history
            if datetime.fromisoformat(change["timestamp"]) > cutoff_time
        ]

        return recent_changes

class FundChangeClassifier:
    async def classify_change(self, change_record: Dict) -> Dict:
        """Classify a capital change"""

        # Identify source
        source = await self.identify_source(change_record)

        # Categorize flow type
        flow_type = await self.categorize_flow_type(change_record, source)

        # Determine priority and risk level
        priority = self.determine_priority(change_record, flow_type)
        risk_level = self.assess_risk_level(change_record, flow_type)

        classification = {
            "change_id": change_record.get("reference_id"),
            "timestamp": change_record["timestamp"],
            "account_id": change_record["account_id"],
            "source": source,
            "flow_type": flow_type,
            "priority": priority,
            "amount": change_record["amount"],
            "currency": change_record["currency"],
            "risk_level": risk_level
        }

        return classification

    async def identify_source(self, change_record: Dict) -> str:
        """Identify the source of capital change"""

        source = change_record.get("source", "")
        description = change_record.get("description", "").lower()

        if "trading" in source or "pnl" in source:
            return "trading_pnl"
        elif "deposit" in source or "deposit" in description:
            return "manual_deposit"
        elif "withdrawal" in source or "withdrawal" in description:
            return "manual_withdrawal"
        elif "bank" in source or "wire" in description:
            return "bank_transfer"
        elif "crypto" in source:
            return "crypto_transfer"
        else:
            return "other"

    async def categorize_flow_type(self, change_record: Dict, source: str) -> str:
        """Categorize the flow type"""

        amount = change_record["amount"]

        if amount > 0:
            if source == "trading_pnl":
                return "trading_profit"
            elif source in ["manual_deposit", "bank_transfer", "crypto_transfer"]:
                return "capital_inflow"
            else:
                return "other_inflow"
        else:
            if source == "trading_pnl":
                return "trading_loss"
            elif source in ["manual_withdrawal", "bank_transfer", "crypto_transfer"]:
                return "capital_outflow"
            else:
                return "other_outflow"

class FundFlowAnalyzer:
    async def analyze_flows(self, classified_changes: List[Dict]) -> Dict:
        """Analyze capital flows"""

        # Calculate basic statistics
        basic_stats = await self.calculate_basic_statistics(classified_changes)

        # Analyze trends
        trend_analysis = await self.analyze_trends(classified_changes)

        # Generate flow summary
        flow_summary = await self.generate_flow_summary(classified_changes)

        return {
            "basic_statistics": basic_stats,
            "trend_analysis": trend_analysis,
            "flow_summary": flow_summary,
            "timestamp": datetime.now().isoformat()
        }

    async def calculate_basic_statistics(self, changes: List[Dict]) -> Dict:
        """Calculate basic flow statistics"""

        if not changes:
            return {
                "total_inflow": 0,
                "total_outflow": 0,
                "net_flow": 0,
                "transaction_count": 0
            }

        inflows = [change["amount"] for change in changes if change["amount"] > 0]
        outflows = [abs(change["amount"]) for change in changes if change["amount"] < 0]

        total_inflow = sum(inflows)
        total_outflow = sum(outflows)
        net_flow = total_inflow - total_outflow

        return {
            "total_inflow": total_inflow,
            "total_outflow": total_outflow,
            "net_flow": net_flow,
            "transaction_count": len(changes),
            "inflow_count": len(inflows),
            "outflow_count": len(outflows)
        }

    async def generate_flow_summary(self, changes: List[Dict]) -> Dict:
        """Generate flow summary by type"""

        flow_types = {}
        for change in changes:
            flow_type = change["flow_type"]
            if flow_type not in flow_types:
                flow_types[flow_type] = []
            flow_types[flow_type].append(change)

        summary = {}
        for flow_type, type_changes in flow_types.items():
            total_amount = sum(change["amount"] for change in type_changes)
            count = len(type_changes)

            summary[flow_type] = {
                "total_amount": total_amount,
                "count": count,
                "average_amount": total_amount / count if count > 0 else 0
            }

        return summary

class AnomalyDetector:
    def __init__(self, threshold_monitor, alert_manager):
        self.threshold_monitor = threshold_monitor
        self.alert_manager = alert_manager
        self.detection_rules = {
            "daily_change_threshold": 0.05,  # 5% of account balance
            "transaction_threshold": 0.10,   # 10% of account balance
            "rapid_transaction_threshold": 300  # 5 minutes
        }

    async def detect_anomalies(self, change_record: Dict, account_context: Dict) -> List[Dict]:
        """Detect anomalies in capital changes"""

        anomalies = []

        # Check threshold violations
        threshold_anomalies = await self.check_threshold_violations(change_record, account_context)
        anomalies.extend(threshold_anomalies)

        # Check fraud indicators
        fraud_anomalies = await self.check_fraud_indicators(change_record, account_context)
        anomalies.extend(fraud_anomalies)

        # Check timing anomalies
        timing_anomalies = await self.check_timing_anomalies(change_record, account_context)
        anomalies.extend(timing_anomalies)

        # Trigger alerts for high-priority anomalies
        await self.trigger_alerts(anomalies)

        return anomalies

    async def check_threshold_violations(self, change_record: Dict, account_context: Dict) -> List[Dict]:
        """Check for threshold violations"""

        anomalies = []
        amount = abs(change_record["amount"])
        account_balance = account_context.get("current_balance", 0)

        # Daily change threshold
        daily_change = account_context.get("daily_change", 0)
        daily_threshold = account_balance * self.detection_rules["daily_change_threshold"]

        if abs(daily_change) > daily_threshold:
            anomalies.append({
                "type": "threshold_violation",
                "severity": "high",
                "description": f"Daily change {daily_change} exceeds threshold {daily_threshold}",
                "change_record": change_record
            })

        # Single transaction threshold
        transaction_threshold = account_balance * self.detection_rules["transaction_threshold"]

        if amount > transaction_threshold:
            anomalies.append({
                "type": "threshold_violation",
                "severity": "medium",
                "description": f"Transaction amount {amount} exceeds threshold {transaction_threshold}",
                "change_record": change_record
            })

        return anomalies

    async def check_fraud_indicators(self, change_record: Dict, account_context: Dict) -> List[Dict]:
        """Check for fraud indicators"""

        anomalies = []

        # Unusual source
        unusual_sources = ["unknown", "external_api"]
        if change_record.get("source") in unusual_sources:
            anomalies.append({
                "type": "fraud_indicator",
                "severity": "medium",
                "description": f"Unusual source: {change_record.get('source')}",
                "change_record": change_record
            })

        # Unusual amount patterns
        if self.is_unusual_amount_pattern(change_record):
            anomalies.append({
                "type": "fraud_indicator",
                "severity": "medium",
                "description": "Unusual amount pattern detected",
                "change_record": change_record
            })

        return anomalies

    async def check_timing_anomalies(self, change_record: Dict, account_context: Dict) -> List[Dict]:
        """Check for timing anomalies"""

        anomalies = []
        timestamp = datetime.fromisoformat(change_record["timestamp"])

        # After-hours activity
        if self.is_after_hours(timestamp):
            anomalies.append({
                "type": "timing_anomaly",
                "severity": "low",
                "description": "After-hours activity detected",
                "change_record": change_record
            })

        # Weekend activity
        if self.is_weekend(timestamp):
            anomalies.append({
                "type": "timing_anomaly",
                "severity": "low",
                "description": "Weekend activity detected",
                "change_record": change_record
            })

        return anomalies

    def is_after_hours(self, timestamp: datetime) -> bool:
        """Check if timing is after hours"""
        hour = timestamp.hour
        return hour < 9 or hour > 18

    def is_weekend(self, timestamp: datetime) -> bool:
        """Check if timestamp is on weekend"""
        return timestamp.weekday() >= 5

    def is_unusual_amount_pattern(self, change_record: Dict) -> bool:
        """Check for unusual amount patterns"""
        amount = change_record["amount"]

        # Check for round numbers (potential fraud indicator)
        if amount % 1000 == 0 and amount > 10000:
            return True

        # Check for very small amounts (potential testing)
        if 0 < amount < 1:
            return True

        return False

@router.get("/flows/recent")
async def get_recent_flows(hours: int = 24):
    """Get recent capital flows"""

    changes = await fund_change_listener.get_recent_changes(hours)
    classified_changes = []

    for change in changes:
        classification = await fund_change_classifier.classify_change(change)
        classified_changes.append(classification)

    analysis = await fund_flow_analyzer.analyze_flows(classified_changes)

    return {
        "flows": classified_changes,
        "analysis": analysis,
        "timestamp": datetime.now().isoformat()
    }

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

    changes = await fund_change_listener.get_account_changes(account_id, hours)
    classified_changes = []

    for change in changes:
        classification = await fund_change_classifier.classify_change(change)
        classified_changes.append(classification)

    return {
        "account_id": account_id,
        "flows": classified_changes,
        "total_changes": len(classified_changes)
    }

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

    changes = await fund_change_listener.get_recent_changes(hours)
    all_anomalies = []

    for change in changes:
        account_context = await get_account_context(change["account_id"])
        anomalies = await anomaly_detector.detect_anomalies(change, account_context)
        all_anomalies.extend(anomalies)

    return {
        "anomalies": all_anomalies,
        "total_anomalies": len(all_anomalies),
        "high_severity": len([a for a in all_anomalies if a["severity"] == "high"])
    }

@router.get("/flows/summary")
async def get_flow_summary(days: int = 7):
    """Get flow summary for period"""

    hours = days * 24
    changes = await fund_change_listener.get_recent_changes(hours)
    classified_changes = []

    for change in changes:
        classification = await fund_change_classifier.classify_change(change)
        classified_changes.append(classification)

    analysis = await fund_flow_analyzer.analyze_flows(classified_changes)

    return {
        "period_days": days,
        "summary": analysis["flow_summary"],
        "statistics": analysis["basic_statistics"],
        "trends": analysis["trend_analysis"]
    }

const CapitalFlowView: React.FC = () => {
  const [recentFlows, setRecentFlows] = useState<CapitalFlow[]>([]);
  const [flowAnalysis, setFlowAnalysis] = useState<FlowAnalysis | null>(null);
  const [anomalies, setAnomalies] = useState<Anomaly[]>([]);

  return (
    <div className="capital-flow-dashboard">
      {/* Flow Overview */}
      <FlowOverviewPanel 
        analysis={flowAnalysis}
        onTimeRangeChange={handleTimeRangeChange}
      />

      {/* Real-time Flow Chart */}
      <FlowChartPanel 
        flows={recentFlows}
      />

      {/* Anomaly Alerts */}
      <AnomalyAlertsPanel 
        anomalies={anomalies}
        onAnomalySelect={handleAnomalySelect}
      />

      {/* Flow Details Table */}
      <FlowDetailsTable 
        flows={recentFlows}
        onFlowSelect={handleFlowSelect}
      />

      {/* Account Balance Trends */}
      <BalanceTrendsPanel />

      {/* Flow Statistics */}
      <FlowStatisticsPanel 
        analysis={flowAnalysis}
      />
    </div>
  );
};