Marketplace Automation - How to Handle 1,000+ Daily Transactions Without Burning Out Your Team

Transaction Occurs
    ↓
    [Event Stream - Kafka/RabbitMQ]
    ↓
    ┌─────────────────┬─────────────────┬─────────────────┐
    ↓                 ↓                 ↓                 ↓
Buyer Automation  Seller Automation  Platform Automation  Analytics
    ↓                 ↓                 ↓                 ↓
    └─────────────────┴─────────────────┴─────────────────┘
                        ↓
                 [Notifications]
                        ↓
              Buyers, Sellers, Ops Team

class OrderProcessor:
    def __init__(self):
        self.payment_gateway = Stripe()
        self.notification_service = NotificationService()
        self.fraud_detector = FraudDetector()

    async def process_order(self, order_data):
        """
        Fully automated order processing
        """

        # Step 1: Validate order
        validation = self.validate_order(order_data)
        if not validation.is_valid:
            return self.handle_invalid_order(order_data, validation.errors)

        # Step 2: Fraud check (instant)
        fraud_score = self.fraud_detector.score(order_data)
        if fraud_score > 80:
            return self.flag_for_review(order_data, fraud_score)

        # Step 3: Hold payment (don't charge yet)
        payment_intent = self.payment_gateway.create_intent(
            amount=order_data['total'],
            customer=order_data['buyer_id'],
            description=f"Order #{order_data['id']}"
        )

        # Step 4: Notify seller (instant)
        await self.notification_service.notify_seller(
            seller_id=order_data['seller_id'],
            message=f"New order #{order_data['id']}",
            action_required="confirm_fulfillment",
            deadline="24_hours"
        )

        # Step 5: Notify buyer (instant)
        await self.notification_service.notify_buyer(
            buyer_id=order_data['buyer_id'],
            message=f"Order #{order_data['id']} confirmed",
            estimated_delivery=self.calculate_delivery_date(order_data)
        )

        # Step 6: Set up automated follow-ups
        self.schedule_follow_ups(order_data)

        # Step 7: Update analytics
        self.analytics.track_order(order_data)

        return {
            'status': 'processing',
            'order_id': order_data['id'],
            'estimated_delivery': self.calculate_delivery_date(order_data)
        }

    def schedule_follow_ups(self, order_data):
        """
        Set up automatic nudges and checks
        """

        # If seller hasn't confirmed in 12 hours → nudge
        schedule_task(
            task='nudge_seller_confirmation',
            order_id=order_data['id'],
            delay_hours=12
        )

        # If seller hasn't confirmed in 24 hours → escalate
        schedule_task(
            task='escalate_no_confirmation',
            order_id=order_data['id'],
            delay_hours=24
        )

        # If not delivered in expected timeframe → check status
        schedule_task(
            task='check_delivery_status',
            order_id=order_data['id'],
            delay_hours=order_data['estimated_delivery_hours']
        )

        # 2 days after delivery → request review
        schedule_task(
            task='request_buyer_review',
            order_id=order_data['id'],
            delay_hours=order_data['estimated_delivery_hours'] + 48
        )

class FraudDetector:
    def __init__(self):
        self.model = self.load_trained_model()

    def score(self, order_data):
        """
        Real-time fraud scoring
        """

        # Extract features
        features = self.extract_features(order_data)

        # ML prediction
        fraud_probability = self.model.predict_proba([features])[0][1]

        # Rule-based red flags
        red_flags = self.check_red_flags(order_data)

        # Combine ML + rules
        final_score = (
            0.7 * fraud_probability * 100 +
            0.3 * len(red_flags) * 20
        )

        return min(final_score, 100)

    def extract_features(self, order_data):
        """
        Feature engineering for fraud detection
        """

        buyer = get_buyer(order_data['buyer_id'])
        seller = get_seller(order_data['seller_id'])

        return {
            # Buyer features
            'buyer_account_age_days': buyer.account_age_days,
            'buyer_total_orders': buyer.order_count,
            'buyer_dispute_ratio': buyer.dispute_ratio,
            'buyer_avg_order_value': buyer.avg_order_value,

            # Order features
            'order_value': order_data['total'],
            'value_vs_buyer_avg': order_data['total'] / buyer.avg_order_value,
            'shipping_country_risk': get_country_risk(order_data['shipping_country']),
            'payment_method_risk': get_payment_risk(order_data['payment_method']),

            # Seller features
            'seller_account_age_days': seller.account_age_days,
            'seller_rating': seller.average_rating,
            'seller_dispute_ratio': seller.dispute_ratio,

            # Behavioral features
            'time_since_last_order_hours': calculate_time_since_last(buyer),
            'multiple_orders_same_day': check_multiple_orders(buyer, today),
            'new_shipping_address': check_new_address(buyer, order_data['shipping_address']),
            'rush_order': order_data.get('rush', False),

            # Context features
            'hour_of_day': datetime.now().hour,
            'day_of_week': datetime.now().weekday(),
        }

    def check_red_flags(self, order_data):
        """
        Rule-based red flag detection
        """

        red_flags = []

        # Buyer account created < 24 hours ago
        if get_buyer(order_data['buyer_id']).account_age_days < 1:
            red_flags.append('new_account')

        # First order > $500
        buyer = get_buyer(order_data['buyer_id'])
        if buyer.order_count == 0 and order_data['total'] > 500:
            red_flags.append('high_value_first_order')

        # Shipping to high-risk country
        if order_data['shipping_country'] in HIGH_RISK_COUNTRIES:
            red_flags.append('high_risk_destination')

        # Different billing and shipping addresses (first order)
        if buyer.order_count == 0:
            if order_data['billing_address'] != order_data['shipping_address']:
                red_flags.append('address_mismatch_new_buyer')

        # Unusual purchase velocity
        recent_orders = get_recent_orders(buyer, hours=24)
        if len(recent_orders) >= 5:
            red_flags.append('high_velocity')

        return red_flags

class PayoutProcessor:
    def __init__(self):
        self.stripe = Stripe()

    async def process_payouts(self, eligible_sellers):
        """
        Automated daily payout processing
        """

        for seller in eligible_sellers:
            try:
                payout = await self.calculate_payout(seller)

                if payout['amount'] > 0:
                    # Transfer funds
                    transfer = self.stripe.transfer(
                        amount=payout['amount'],
                        destination=seller.stripe_account_id,
                        metadata={
                            'seller_id': seller.id,
                            'period_start': payout['period_start'],
                            'period_end': payout['period_end'],
                            'transaction_count': payout['transaction_count']
                        }
                    )

                    # Generate invoice
                    invoice = self.generate_invoice(seller, payout)

                    # Send notification
                    await self.notify_seller(seller, payout, invoice)

                    # Log for tax purposes
                    self.log_payout(seller, payout, transfer)

            except Exception as e:
                # Alert ops team
                self.alert_payout_failure(seller, e)

    async def calculate_payout(self, seller):
        """
        Calculate seller payout
        """

        # Get completed transactions in payout period
        transactions = get_completed_transactions(
            seller_id=seller.id,
            start_date=seller.last_payout_date,
            end_date=datetime.now()
        )

        # Calculate gross revenue
        gross_revenue = sum(t['seller_amount'] for t in transactions)

        # Platform fee
        platform_fee = gross_revenue * seller.platform_fee_rate

        # Payment processing fee
        processing_fee = sum(t['processing_fee'] for t in transactions)

        # Adjustments (refunds, disputes, etc.)
        adjustments = self.calculate_adjustments(seller, transactions)

        # Tax withholding (if applicable)
        tax_withholding = self.calculate_tax_withholding(seller, gross_revenue)

        # Net payout
        net_payout = (
            gross_revenue
            - platform_fee
            - processing_fee
            + adjustments['credits']
            - adjustments['debits']
            - tax_withholding
        )

        return {
            'seller_id': seller.id,
            'period_start': seller.last_payout_date,
            'period_end': datetime.now(),
            'transaction_count': len(transactions),
            'gross_revenue': gross_revenue,
            'platform_fee': platform_fee,
            'processing_fee': processing_fee,
            'adjustments': adjustments,
            'tax_withholding': tax_withholding,
            'amount': max(net_payout, 0) # Don't pay negative amounts
        }

class DisputeResolver:
    async def handle_dispute(self, dispute_data):
        """
        Automated dispute investigation and resolution
        """

        # Step 1: Collect evidence automatically
        evidence = await self.collect_evidence(dispute_data)

        # Step 2: AI analysis
        analysis = await self.analyze_dispute(dispute_data, evidence)

        # Step 3: Auto-resolve if confidence is high
        if analysis['confidence'] > 0.90:
            resolution = await self.auto_resolve(dispute_data, analysis)
            await self.notify_parties(dispute_data, resolution)
            return resolution

        # Step 4: Otherwise, prepare for human review
        else:
            case_summary = self.prepare_case_summary(
                dispute_data,
                evidence,
                analysis
            )
            await self.escalate_to_human(case_summary)

    async def collect_evidence(self, dispute_data):
        """
        Automatically gather all relevant evidence
        """

        evidence = {}

        # Transaction history
        evidence['transaction'] = get_transaction(dispute_data['transaction_id'])

        # Communication history
        evidence['messages'] = get_messages(
            buyer_id=dispute_data['buyer_id'],
            seller_id=dispute_data['seller_id'],
            order_id=dispute_data['order_id']
        )

        # Delivery confirmation
        if dispute_data['type'] == 'not_received':
            evidence['delivery'] = get_delivery_proof(dispute_data['order_id'])

        # Product description vs. received
        if dispute_data['type'] == 'not_as_described':
            evidence['listing'] = get_original_listing(dispute_data['product_id'])
            evidence['photos'] = dispute_data.get('buyer_photos', [])

        # Seller history
        evidence['seller_stats'] = {
            'rating': get_seller_rating(dispute_data['seller_id']),
            'dispute_rate': get_seller_dispute_rate(dispute_data['seller_id']),
            'total_sales': get_seller_total_sales(dispute_data['seller_id'])
        }

        # Buyer history
        evidence['buyer_stats'] = {
            'order_count': get_buyer_order_count(dispute_data['buyer_id']),
            'dispute_rate': get_buyer_dispute_rate(dispute_data['buyer_id'])
        }

        # Similar disputes
        evidence['similar_disputes'] = find_similar_disputes(dispute_data)

        return evidence

    async def analyze_dispute(self, dispute_data, evidence):
        """
        AI-powered dispute analysis
        """

        prompt = f"""Analyze this marketplace dispute:

Dispute Type: {dispute_data['type']}
Buyer Claim: {dispute_data['buyer_claim']}
Seller Response: {dispute_data.get('seller_response', 'No response yet')}

Evidence:
{format_evidence(evidence)}

Determine:
1. Who is likely correct? (buyer/seller/unclear)
2. Recommended resolution
3. Confidence level (0-100)
4. Reasoning

Return as JSON.
"""

        response = await call_gpt4(prompt)
        analysis = json.loads(response)

        return analysis

    async def auto_resolve(self, dispute_data, analysis):
        """
        Automatically resolve dispute based on analysis
        """

        if analysis['recommended_winner'] == 'buyer':
            # Refund buyer
            refund = self.process_refund(
                transaction_id=dispute_data['transaction_id'],
                amount=dispute_data['amount'],
                reason='dispute_resolved_buyer_favor'
            )

            # Deduct from seller (or absorb as marketplace)
            self.handle_seller_deduction(
                seller_id=dispute_data['seller_id'],
                amount=dispute_data['amount'],
                reason='dispute_loss'
            )

            resolution = 'buyer_refunded'

        elif analysis['recommended_winner'] == 'seller':
            # No refund, seller keeps payment
            resolution = 'seller_wins'

        else:
            # Partial refund
            partial_amount = dispute_data['amount'] * 0.5
            refund = self.process_refund(
                transaction_id=dispute_data['transaction_id'],
                amount=partial_amount,
                reason='dispute_resolved_partial'
            )
            resolution = 'partial_refund'

        # Log resolution
        self.log_dispute_resolution(dispute_data, analysis, resolution)

        return {
            'resolution': resolution,
            'reasoning': analysis['reasoning'],
            'auto_resolved': True
        }

class QualityMonitor:
    async def monitor_listings(self):
        """
        Continuous quality monitoring
        """

        # Check new listings
        new_listings = get_recent_listings(hours=24)

        for listing in new_listings:
            issues = []

            # AI content moderation
            moderation = await self.moderate_content(listing)
            if moderation['flags']:
                issues.extend(moderation['flags'])

            # Quality checks
            quality_score = self.assess_listing_quality(listing)
            if quality_score < 70:
                issues.append(f'low_quality_score: {quality_score}')

            # Policy compliance
            policy_violations = self.check_policy_compliance(listing)
            if policy_violations:
                issues.extend(policy_violations)

            # Action based on severity
            if len(issues) > 0:
                await self.handle_listing_issues(listing, issues)

    async def moderate_content(self, listing):
        """
        AI-powered content moderation
        """

        # Check text for prohibited content
        text_flags = await check_text_moderation(listing['description'])

        # Check images for prohibited content
        image_flags = await check_image_moderation(listing['images'])

        return {
            'flags': text_flags + image_flags,
            'safe': len(text_flags + image_flags) == 0
        }

    def assess_listing_quality(self, listing):
        """
        Multi-factor quality scoring
        """

        score = 100

        # Title quality
        if len(listing['title']) < 20:
            score -= 15

        # Description quality
        if len(listing['description']) < 100:
            score -= 20

        # Images
        if len(listing['images']) == 0:
            score -= 30
        elif len(listing['images']) < 3:
            score -= 10

        # Pricing reasonableness
        if self.is_price_suspicious(listing):
            score -= 25

        # Category accuracy
        if not self.is_category_appropriate(listing):
            score -= 15

        return max(score, 0)

    async def handle_listing_issues(self, listing, issues):
        """
        Take action based on issue severity
        """

        severity = self.calculate_severity(issues)

        if severity == 'critical':
            # Auto-delist
            await self.delist_listing(listing, reason=issues)
            await self.notify_seller_violation(listing, issues)

        elif severity == 'high':
            # Flag for review + notify seller
            await self.flag_for_review(listing, issues)
            await self.notify_seller_issues(listing, issues)

        else:
            # Just notify seller of improvements needed
            await self.notify_seller_suggestions(listing, issues)

class MarketplaceAnalytics:
    def generate_daily_report(self):
        """
        Automated daily executive report
        """

        report = {
            'date': datetime.now().date(),

            # GMV
            'gmv': {
                'today': calculate_gmv(period='today'),
                'mtd': calculate_gmv(period='month'),
                'vs_yesterday': calculate_change('gmv', 'yesterday'),
                'vs_last_month': calculate_change('gmv', 'last_month')
            },

            # Transaction metrics
            'transactions': {
                'count': count_transactions(period='today'),
                'avg_value': calculate_avg_transaction_value('today'),
                'completion_rate': calculate_completion_rate('today')
            },

            # User metrics
            'users': {
                'new_buyers': count_new_users('buyer', 'today'),
                'new_sellers': count_new_users('seller', 'today'),
                'active_buyers': count_active_users('buyer', 'today'),
                'active_sellers': count_active_users('seller', 'today')
            },

            # Quality metrics
            'quality': {
                'buyer_satisfaction': calculate_satisfaction('buyer', 'week'),
                'seller_satisfaction': calculate_satisfaction('seller', 'week'),
                'dispute_rate': calculate_dispute_rate('week'),
                'fraud_rate': calculate_fraud_rate('week')
            },

            # Alerts
            'alerts': self.generate_alerts(),

            # Recommendations
            'recommendations': self.generate_recommendations()
        }

        return report

    def generate_alerts(self):
        """
        Proactive alerting on anomalies
        """

        alerts = []

        # GMV drop
        if self.is_significant_drop('gmv', threshold=0.15):
            alerts.append({
                'severity': 'high',
                'metric': 'GMV',
                'message': 'GMV down 18% vs. yesterday',
                'possible_causes': self.analyze_gmv_drop(),
                'action': 'Investigate immediately'
            })

        # Fraud spike
        fraud_rate_today = calculate_fraud_rate('today')
        fraud_rate_avg = calculate_fraud_rate('last_30_days')
        if fraud_rate_today > fraud_rate_avg * 2:
            alerts.append({
                'severity': 'critical',
                'metric': 'Fraud Rate',
                'message': f'Fraud rate {fraud_rate_today:.1%} (avg: {fraud_rate_avg:.1%})',
                'action': 'Review fraud detection rules'
            })

        # Seller churn
        seller_churn = calculate_seller_churn('week')
        if seller_churn > 0.05:
            alerts.append({
                'severity': 'medium',
                'metric': 'Seller Churn',
                'message': f'Seller churn at {seller_churn:.1%} this week',
                'top_reasons': get_top_churn_reasons('seller'),
                'action': 'Review seller satisfaction'
            })

        return alerts