ml_engine/pnl_tracker.py

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Optional

class PnLTracker:
    """
    Tracks hypothetical P&L for all predictions to demonstrate value.
    Simulates a portfolio that invests based on model confidence.
    """
    
    def __init__(self, initial_capital: float = 1_000_000.0):
        self.initial_capital = initial_capital
        self.current_capital = initial_capital
        self.predictions = []  # Log of all predictions made
        self.positions = []    # Active "investments"
        self.closed_trades = [] # Completed trades
        self.win_count = 0
        self.loss_count = 0
        
        # Performance metrics
        self.total_pnl = 0.0
        self.roi_pct = 0.0
        
    def calculate_position_size(self, confidence: float) -> float:
        """
        Kelly Criterion-inspired position sizing based on confidence.
        Higher confidence = larger bet size.
        """
        if confidence < 0.5:
            return 0.0
            
        # Simple scaling: 50% conf = 0% size, 100% conf = 10% of capital
        # This is a conservative simulation
        max_position_pct = 0.10
        scale_factor = (confidence - 0.5) * 2  # 0.0 to 1.0
        
        return self.current_capital * max_position_pct * scale_factor

    def record_prediction(self, prediction_id: str, vertical: str, target: str, 
                         predicted_value: any, confidence: float, 
                         expected_timeline_days: int):
        """
        Log a new prediction and "open" a hypothetical position.
        """
        position_size = self.calculate_position_size(confidence)
        
        record = {
            'id': prediction_id,
            'date': datetime.now().isoformat(),
            'vertical': vertical,
            'target': target,
            'prediction': predicted_value,
            'confidence': confidence,
            'position_size': position_size,
            'status': 'OPEN',
            'expected_close_date': (datetime.now() + timedelta(days=expected_timeline_days)).isoformat()
        }
        
        self.predictions.append(record)
        if position_size > 0:
            self.positions.append(record)
            
        return position_size

    def resolve_prediction(self, prediction_id: str, actual_value: any, success: bool, pnl_pct: float):
        """
        Close a position based on real-world outcome.
        pnl_pct: The simulated return on the position (e.g., 0.20 for 20% gain)
        """
        # Find the position
        position = next((p for p in self.positions if p['id'] == prediction_id), None)
        
        if not position:
            return
            
        # Calculate P&L
        invested_amount = position['position_size']
        pnl_amount = invested_amount * pnl_pct
        
        self.current_capital += pnl_amount
        self.total_pnl += pnl_amount
        self.roi_pct = (self.current_capital - self.initial_capital) / self.initial_capital * 100
        
        if success:
            self.win_count += 1
        else:
            self.loss_count += 1
            
        # Move to closed
        position['status'] = 'CLOSED'
        position['actual_value'] = actual_value
        position['pnl_amount'] = pnl_amount
        position['pnl_pct'] = pnl_pct
        position['close_date'] = datetime.now().isoformat()
        
        self.closed_trades.append(position)
        self.positions.remove(position)

    def get_performance_metrics(self) -> Dict:
        """
        Return comprehensive performance stats for the dashboard.
        """
        total_trades = self.win_count + self.loss_count
        win_rate = (self.win_count / total_trades * 100) if total_trades > 0 else 0.0
        
        return {
            'current_capital': self.current_capital,
            'total_pnl': self.total_pnl,
            'roi_pct': round(self.roi_pct, 2),
            'win_rate': round(win_rate, 1),
            'total_trades': total_trades,
            'active_positions': len(self.positions),
            'avg_win_pct': self._calculate_avg_pnl(wins_only=True),
            'avg_loss_pct': self._calculate_avg_pnl(losses_only=True)
        }
        
    def _calculate_avg_pnl(self, wins_only=False, losses_only=False) -> float:
        trades = self.closed_trades
        if wins_only:
            trades = [t for t in trades if t['pnl_amount'] > 0]
        if losses_only:
            trades = [t for t in trades if t['pnl_amount'] <= 0]
            
        if not trades:
            return 0.0
            
        avg = sum(t['pnl_pct'] for t in trades) / len(trades) * 100
        return round(avg, 1)