File size: 4,723 Bytes
c89a139 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 |
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)
|