296 lines
9.5 KiB
Python
296 lines
9.5 KiB
Python
"""
|
|
Portfolio Manager - Position Sizing and Risk Management
|
|
|
|
Enforces portfolio constraints:
|
|
- Max 8% per stock
|
|
- Max X% in risky trades
|
|
- Position size based on signal strength
|
|
- Portfolio utilization tracking
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Dict, Optional, List
|
|
from datetime import datetime
|
|
import json
|
|
|
|
|
|
@dataclass
|
|
class Position:
|
|
"""Represents a single position in the portfolio"""
|
|
ticker: str
|
|
shares: int
|
|
entry_price: float
|
|
entry_date: datetime
|
|
signal_score: float # 0-100
|
|
position_type: str # "momentum", "pump", "fundamentals"
|
|
|
|
@property
|
|
def market_value(self, current_price: float) -> float:
|
|
return self.shares * current_price
|
|
|
|
|
|
class PortfolioManager:
|
|
"""
|
|
Manages portfolio constraints and position sizing.
|
|
|
|
Config options:
|
|
- max_position_pct: Max % of portfolio in one stock (default: 8%)
|
|
- max_risky_pct: Max % in high-risk trades (pump/momentum) (default: 25%)
|
|
- max_positions: Max number of open positions (default: 10)
|
|
- portfolio_cash: Starting cash (default: $10,000)
|
|
- min_position_size: Minimum $ per trade (default: $100)
|
|
- max_position_size: Maximum $ per trade (default: 2000)
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
portfolio_cash: float = 10000.0,
|
|
max_position_pct: float = 0.08, # 8%
|
|
max_risky_pct: float = 0.25, # 25% in risky trades
|
|
max_positions: int = 10,
|
|
min_position_size: float = 100.0,
|
|
max_position_size: float = 2000.0,
|
|
):
|
|
self.initial_cash = portfolio_cash
|
|
self.cash = portfolio_cash
|
|
self.portfolio_value = portfolio_cash
|
|
|
|
self.max_position_pct = max_position_pct
|
|
self.max_risky_pct = max_risky_pct
|
|
self.max_positions = max_positions
|
|
self.min_position_size = min_position_size
|
|
self.max_position_size = max_position_size
|
|
|
|
self.positions: Dict[str, Position] = {}
|
|
self.trade_history: List[Dict] = []
|
|
|
|
def calculate_position_size(
|
|
self,
|
|
current_price: float,
|
|
signal_score: float,
|
|
position_type: str = "momentum",
|
|
) -> Optional[Dict]:
|
|
"""
|
|
Calculate position size based on signal strength and portfolio constraints.
|
|
|
|
Args:
|
|
current_price: Current stock price
|
|
signal_score: Signal strength 0-100
|
|
position_type: "momentum", "pump", "fundamentals"
|
|
|
|
Returns:
|
|
Dict with shares, position_value, or None if trade not allowed
|
|
"""
|
|
|
|
# Check if we can trade
|
|
if not self._can_enter_trade():
|
|
return None
|
|
|
|
# Calculate max position value
|
|
max_position_value = self.portfolio_value * self.max_position_pct
|
|
|
|
# Scale position size by signal strength
|
|
# Stronger signals get bigger positions (up to max)
|
|
signal_multiplier = signal_score / 100.0 # 0-1
|
|
position_value = max_position_value * signal_multiplier
|
|
|
|
# Enforce min/max position size
|
|
position_value = max(self.min_position_size, position_value)
|
|
position_value = min(self.max_position_size, position_value)
|
|
|
|
# Check if we have enough cash
|
|
if position_value > self.cash:
|
|
position_value = self.cash
|
|
|
|
# Check risky position limits
|
|
risky_value = self._calculate_risky_exposure()
|
|
if position_type in ["momentum", "pump"]:
|
|
max_risky_value = self.portfolio_value * self.max_risky_pct
|
|
if risky_value + position_value > max_risky_value:
|
|
# Reduce position to fit within risky limit
|
|
position_value = max(
|
|
self.min_position_size,
|
|
max_risky_value - risky_value
|
|
)
|
|
|
|
# Calculate shares
|
|
shares = int(position_value / current_price)
|
|
|
|
if shares < 1:
|
|
return None
|
|
|
|
return {
|
|
"shares": shares,
|
|
"position_value": shares * current_price,
|
|
"signal_multiplier": signal_multiplier,
|
|
"position_pct": (shares * current_price) / self.portfolio_value,
|
|
}
|
|
|
|
def add_position(
|
|
self,
|
|
ticker: str,
|
|
shares: int,
|
|
entry_price: float,
|
|
signal_score: float,
|
|
position_type: str,
|
|
) -> bool:
|
|
"""Add a new position to the portfolio"""
|
|
position_value = shares * entry_price
|
|
|
|
if ticker in self.positions:
|
|
return False # Already have position
|
|
|
|
if position_value > self.cash:
|
|
return False # Not enough cash
|
|
|
|
# Update cash
|
|
self.cash -= position_value
|
|
|
|
# Add position
|
|
self.positions[ticker] = Position(
|
|
ticker=ticker,
|
|
shares=shares,
|
|
entry_price=entry_price,
|
|
entry_date=datetime.now(),
|
|
signal_score=signal_score,
|
|
position_type=position_type,
|
|
)
|
|
|
|
# Record trade
|
|
self.trade_history.append({
|
|
"action": "BUY",
|
|
"ticker": ticker,
|
|
"shares": shares,
|
|
"price": entry_price,
|
|
"timestamp": datetime.now().isoformat(),
|
|
"signal_score": signal_score,
|
|
"position_type": position_type,
|
|
})
|
|
|
|
return True
|
|
|
|
def close_position(
|
|
self,
|
|
ticker: str,
|
|
exit_price: float,
|
|
reason: str,
|
|
) -> Optional[Dict]:
|
|
"""Close an existing position"""
|
|
if ticker not in self.positions:
|
|
return None
|
|
|
|
position = self.positions[ticker]
|
|
exit_value = position.shares * exit_price
|
|
entry_value = position.shares * position.entry_price
|
|
profit = exit_value - entry_value
|
|
profit_pct = (profit / entry_value) * 100
|
|
|
|
# Update cash
|
|
self.cash += exit_value
|
|
|
|
# Record trade
|
|
self.trade_history.append({
|
|
"action": "SELL",
|
|
"ticker": ticker,
|
|
"shares": position.shares,
|
|
"price": exit_price,
|
|
"timestamp": datetime.now().isoformat(),
|
|
"profit": profit,
|
|
"profit_pct": profit_pct,
|
|
"hold_days": (datetime.now() - position.entry_date).days,
|
|
"reason": reason,
|
|
})
|
|
|
|
# Remove position
|
|
del self.positions[ticker]
|
|
|
|
return {
|
|
"ticker": ticker,
|
|
"profit": profit,
|
|
"profit_pct": profit_pct,
|
|
"hold_days": (datetime.now() - position.entry_date).days,
|
|
}
|
|
|
|
def get_portfolio_status(self) -> Dict:
|
|
"""Get current portfolio status"""
|
|
positions_value = sum(
|
|
p.shares * p.entry_price for p in self.positions.values()
|
|
)
|
|
self.portfolio_value = self.cash + positions_value
|
|
|
|
risky_value = self._calculate_risky_exposure()
|
|
|
|
return {
|
|
"total_value": self.portfolio_value,
|
|
"cash": self.cash,
|
|
"positions_value": positions_value,
|
|
"num_positions": len(self.positions),
|
|
"max_positions": self.max_positions,
|
|
"cash_utilization": (positions_value / self.portfolio_value) * 100,
|
|
"risky_exposure": (risky_value / self.portfolio_value) * 100,
|
|
"max_risky_pct": self.max_risky_pct * 100,
|
|
"positions": {
|
|
t: {
|
|
"shares": p.shares,
|
|
"entry_price": p.entry_price,
|
|
"signal_score": p.signal_score,
|
|
"position_type": p.position_type,
|
|
}
|
|
for t, p in self.positions.items()
|
|
},
|
|
}
|
|
|
|
def _can_enter_trade(self) -> bool:
|
|
"""Check if we can enter a new trade"""
|
|
# Check max positions
|
|
if len(self.positions) >= self.max_positions:
|
|
return False
|
|
|
|
# Check minimum cash buffer
|
|
if self.cash < self.min_position_size:
|
|
return False
|
|
|
|
return True
|
|
|
|
def _calculate_risky_exposure(self) -> float:
|
|
"""Calculate total value in risky positions (momentum/pump)"""
|
|
risky_value = 0.0
|
|
for ticker, pos in self.positions.items():
|
|
if pos.position_type in ["momentum", "pump"]:
|
|
risky_value += pos.shares * pos.entry_price
|
|
return risky_value
|
|
|
|
def to_dict(self) -> Dict:
|
|
"""Serialize portfolio to dict"""
|
|
return {
|
|
"cash": self.cash,
|
|
"portfolio_value": self.portfolio_value,
|
|
"positions": {
|
|
t: {
|
|
"shares": p.shares,
|
|
"entry_price": p.entry_price,
|
|
"entry_date": p.entry_date.isoformat(),
|
|
"signal_score": p.signal_score,
|
|
"position_type": p.position_type,
|
|
}
|
|
for t, p in self.positions.items()
|
|
},
|
|
"trade_history": self.trade_history,
|
|
}
|
|
|
|
def save_portfolio(self, filepath: str):
|
|
"""Save portfolio state to JSON"""
|
|
with open(filepath, 'w') as f:
|
|
json.dump(self.to_dict(), f, indent=2)
|
|
|
|
def load_portfolio(self, filepath: str):
|
|
"""Load portfolio state from JSON"""
|
|
with open(filepath, 'r') as f:
|
|
data = json.load(f)
|
|
|
|
self.cash = data["cash"]
|
|
self.portfolio_value = data["portfolio_value"]
|
|
self.trade_history = data["trade_history"]
|
|
|
|
# Note: positions would need to be reconstructed from historical data
|