115 lines
4.3 KiB
Python
115 lines
4.3 KiB
Python
import logging
|
|
from datetime import datetime, timezone
|
|
from config import MAX_POSITIONS, MIN_POSITION_USD, STOP_LOSS_PCT, TAKE_PROFIT_1_PCT, TAKE_PROFIT_2_PCT
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class PortfolioManager:
|
|
def __init__(self, initial_capital: float = 200.0):
|
|
self.initial_capital = initial_capital
|
|
self.cash = initial_capital
|
|
self.positions: dict[str, dict] = {}
|
|
self.trades: list[dict] = []
|
|
|
|
def buy(self, symbol: str, price: float, score: float) -> bool:
|
|
if symbol in self.positions:
|
|
return False
|
|
if len(self.positions) >= MAX_POSITIONS:
|
|
return False
|
|
amount = self._position_size(score)
|
|
if amount < MIN_POSITION_USD:
|
|
return False
|
|
if amount > self.cash:
|
|
amount = self.cash
|
|
if amount < MIN_POSITION_USD:
|
|
return False
|
|
quantity = amount / price
|
|
self.cash -= amount
|
|
self.positions[symbol] = {
|
|
"entry_price": price, "quantity": quantity,
|
|
"invested_usd": amount, "tp1_hit": False,
|
|
"opened_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
self.trades.append({
|
|
"coin": symbol, "side": "BUY", "price": price,
|
|
"quantity": quantity, "amount_usd": amount,
|
|
"timestamp": datetime.now(timezone.utc).isoformat(), "reason": "signal",
|
|
})
|
|
return True
|
|
|
|
def sell(self, symbol: str, price: float, reason: str = "signal", partial: float = 1.0):
|
|
if symbol not in self.positions:
|
|
return
|
|
pos = self.positions[symbol]
|
|
sell_qty = pos["quantity"] * partial
|
|
sell_usd = sell_qty * price
|
|
self.cash += sell_usd
|
|
self.trades.append({
|
|
"coin": symbol, "side": "SELL", "price": price,
|
|
"quantity": sell_qty, "amount_usd": sell_usd,
|
|
"timestamp": datetime.now(timezone.utc).isoformat(), "reason": reason,
|
|
})
|
|
if partial >= 1.0:
|
|
del self.positions[symbol]
|
|
else:
|
|
pos["quantity"] -= sell_qty
|
|
|
|
def check_exit(self, symbol: str, current_price: float):
|
|
if symbol not in self.positions:
|
|
return
|
|
pos = self.positions[symbol]
|
|
entry = pos["entry_price"]
|
|
change_pct = (current_price - entry) / entry
|
|
if change_pct <= STOP_LOSS_PCT:
|
|
self.sell(symbol, current_price, reason="stop-loss")
|
|
return
|
|
if change_pct >= TAKE_PROFIT_2_PCT:
|
|
self.sell(symbol, current_price, reason="take-profit-2")
|
|
return
|
|
if change_pct >= TAKE_PROFIT_1_PCT and not pos["tp1_hit"]:
|
|
pos["tp1_hit"] = True
|
|
self.sell(symbol, current_price, reason="take-profit-1", partial=0.5)
|
|
|
|
def _position_size(self, score: float) -> float:
|
|
if score >= 90:
|
|
pct = 0.30
|
|
elif score >= 80:
|
|
pct = 0.20
|
|
else:
|
|
pct = 0.15
|
|
return round(self.cash * pct, 2)
|
|
|
|
def get_portfolio_value(self, current_prices: dict[str, float]) -> dict:
|
|
holdings_value = sum(
|
|
pos["quantity"] * current_prices.get(sym, pos["entry_price"])
|
|
for sym, pos in self.positions.items()
|
|
)
|
|
total_value = self.cash + holdings_value
|
|
total_pnl = total_value - self.initial_capital
|
|
pnl_pct = (total_pnl / self.initial_capital) * 100
|
|
|
|
winning = sum(1 for t in self.trades if t["side"] == "SELL" and self._trade_pnl(t) > 0)
|
|
total_sells = sum(1 for t in self.trades if t["side"] == "SELL")
|
|
win_rate = (winning / total_sells * 100) if total_sells > 0 else 0
|
|
|
|
return {
|
|
"total_value": round(total_value, 2),
|
|
"cash": round(self.cash, 2),
|
|
"holdings_value": round(holdings_value, 2),
|
|
"total_pnl": round(total_pnl, 2),
|
|
"pnl_pct": round(pnl_pct, 2),
|
|
"win_rate": round(win_rate, 1),
|
|
"open_positions": len(self.positions),
|
|
}
|
|
|
|
def _trade_pnl(self, sell_trade: dict) -> float:
|
|
matching_buys = [
|
|
t for t in self.trades
|
|
if t["coin"] == sell_trade["coin"] and t["side"] == "BUY"
|
|
and t["timestamp"] <= sell_trade["timestamp"]
|
|
]
|
|
if matching_buys:
|
|
latest_buy = matching_buys[-1]
|
|
return (sell_trade["price"] - latest_buy["price"]) * sell_trade["quantity"]
|
|
return 0
|