File size: 3,638 Bytes
fc115d5 | 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 | """
Global Portfolio Manager
Enforces cross-asset correlation limits and portfolio-wide risk rules.
"""
import logging
from typing import Dict, Optional
from datetime import datetime
logger = logging.getLogger(__name__)
class GlobalPortfolioManager:
"""
Manages global portfolio risk across multiple independent trading agents.
Prevents correlated wipeouts by limiting concurrent directional exposure.
"""
def __init__(self, max_correlated_positions: int = 2):
"""
Initialize the global portfolio manager.
Args:
max_correlated_positions: Maximum number of concurrent positions
allowed in the exact same direction.
"""
self.max_correlated_positions = max_correlated_positions
# Track active positions across all agents. Format: {symbol: direction (1 or -1)}
self.active_positions: Dict[str, int] = {}
def register_position(self, symbol: str, direction: int):
"""Register that an agent successfully opened a position."""
if direction not in [1, -1]:
return
self.active_positions[symbol] = direction
logger.info(
f"🌐 GlobalPortfolioManager: Registered {symbol} {'LONG' if direction == 1 else 'SHORT'}. "
f"Active Portfolio: {self.get_portfolio_summary()}"
)
def clear_position(self, symbol: str):
"""Clear a position when an agent exits a trade."""
if symbol in self.active_positions:
direction = self.active_positions.pop(symbol)
logger.info(
f"🌐 GlobalPortfolioManager: Cleared {symbol} {'LONG' if direction == 1 else 'SHORT'}. "
f"Active Portfolio: {self.get_portfolio_summary()}"
)
def can_open_position(self, symbol: str, proposed_direction: int) -> bool:
"""
Check if an agent is allowed to open a new position.
Args:
symbol: The asset wanting to trade (e.g. BTCUSDT)
proposed_direction: 1 for LONG, -1 for SHORT
Returns:
True if the position is allowed under portfolio correlation limits.
"""
if proposed_direction not in [1, -1]:
return False
# Count how many existing positions are in this exact same direction
correlated_count = sum(
1 for s, d in self.active_positions.items()
if d == proposed_direction and s != symbol
)
if correlated_count >= self.max_correlated_positions:
logger.warning(
f"🌐 GlobalPortfolioManager: BLOCKED proposed {symbol} {'LONG' if proposed_direction == 1 else 'SHORT'}. "
f"Correlation Limit Reached ({correlated_count}/{self.max_correlated_positions} active). "
f"Active Portfolio: {self.get_portfolio_summary()}"
)
return False
return True
def get_portfolio_summary(self) -> str:
"""Returns a string describing the current directional exposure."""
if not self.active_positions:
return "Flat"
longs = [s for s, d in self.active_positions.items() if d == 1]
shorts = [s for s, d in self.active_positions.items() if d == -1]
summary = []
if longs:
summary.append(f"LONGs: {','.join(longs)}")
if shorts:
summary.append(f"SHORTs: {','.join(shorts)}")
return " | ".join(summary)
|