portfolioManagement.py

# portfolio management, handles trades and equity management

import pandas as pd
from trades import Trades

# --- Module 5: Trade and equity management ---
class PortfolioManagement:
    """
    Class to manage cash, positions, trades, and equity. Contains methods to open, close trades,
    update PNL, stop loss, take profit. (in progress) Able to manage trade positions in multiple assets

    Will be responsible in calculation of **final** position sizes, which depends on risk management
    there are two scenarios where the module has to calculate (final) position sizes
        1. signals generated by strategy module are only 0, -1 and 1:
            this means that the only factor affecting the final position size is the risk tolerance and SL
            level, and current price
        2. signals generated are floats with varying values:
            the reason for position sizes in strategy module is to execute based on confidence levels/arbitrage hedge
            the position sizing given is **relative** ie if 1 XYZ is bought, 2 ABC will be sold. however, the actual
            number of shares of XYZ bought is 5, so 10 ABC will be sold.
            in summary: (relative) position sizing from strategy module -> **final** position sizes (to be executed)

    Tracks current liquid cash, as well as total equity (including unrealized PNL from open trades).
    Most functions will be called within main backtesting methods.
    Will also contain and handle lists of open trades (trade class objects), as well as closed trades (dicts).

    Args:
        initial capital - given by user when calling main backtesting method
        commissions pct - percentage of commissions, also given by user
    """

    def __init__(self, initial_capital: float = 10000.0, commission_pct: float = 0.01):

        self.initial_capital = initial_capital
        self.commission_pct = commission_pct
        self.pct_capital = 0.01  # max risk for capital when opening a trade

        self.cash = initial_capital # current liquid cash (excludes equity from open trades)
        self.open_trades = [] # contains trades objects
        self.closed_trades = [] # contains dicts (of all closed trades for statistics)
        self.equity_history = [] # contains date and equity history
        self.trade_counter = 0
        # self.SL_hit = 0
        # self.TP_hit = 0
        # self.end_of_backtest_hit = 0
        # self.leverage = 2 #NOT IN USE

    def record_equity(self, date: pd.Timestamp, market_prices: dict):
        """
        Records the total equity of the portfolio at a point in time.
        This now requires a dictionary of the current market prices for all open positions.

        Args:
            date (pd.Timestamp): The current date of the backtest.
            market_prices (dict): A dictionary mapping symbols to their current price.
                                  e.g., {'EURUSD=X': 1.08, 'AUDUSD=X': 0.66}
        """
        unrealized_pnl = 0
        for trade in self.open_trades:

            # 1. Look up the current price for the specific symbol of the trade, passed as a dict
            # example of dict: 'BTC-USD' = 21903.01
            current_price = market_prices.get(trade.symbol)

            # 2. Safety check: If for some reason the price is missing, skip this trade's PnL calculation
            if current_price is None:
                continue

            # 3. Calculate unrealized PnL based on the trade's direction and its specific market price
            if trade.direction == "LONG":
                unrealized_pnl += (current_price - trade.entry_price) * trade.quantity + trade.entry_price * trade.quantity
            elif trade.direction == "SHORT":
                unrealized_pnl += (trade.entry_price - current_price) * trade.quantity + trade.entry_price * trade.quantity

        # The rest of the logic remains the same
        total_equity = self.cash + unrealized_pnl
        self.equity_history.append({'date': date, 'equity': total_equity})

    def open_trade(self, entry_date: pd.Timestamp, symbol: str, entry_price: float, signal_value: float, trade_ID: int, trade_type: str, stop_loss: float, take_profit: float):
        """
        Opens trade given parameters by user
        IMPT: need to ensure correct row iteration for row['date']
              need to increment trade_id counter in the overarching iterator
        Args: current price, entry date (input row['date'], SL/TP price, quantity, trade id, trade direction, trade type)
                relative position size (scaling factor for position size), signal : to determine direction.

        will handle both long and short trades, as given by backtest run methods. currently handles trade creation, position
        sizing, as well as risk management (stop loss and take profit levels).

        final position sizing depends on the value of current liquid cash, as well as relative position sizes
        """

        if signal_value == 0:
            return  # No trade to open

        direction = "LONG" if signal_value > 0 else "SHORT"
        risk_per_unit = 0

        # 1. Calculate per-unit risk based on the stop loss
        if direction == "LONG":
            risk_per_unit = entry_price - stop_loss
        elif direction == "SHORT":
            risk_per_unit = stop_loss - entry_price

        if risk_per_unit <= 0:
            print("invalid risk per unit")
            return  # Invalid stop loss, cannot calculate position size

        # 2. Calculate a base quantity based on risk appetite
        dollar_risk_amount = self.cash * self.pct_capital
        base_quantity = dollar_risk_amount / risk_per_unit

        # 3. Scale the final quantity by the relative signal value
        final_quantity = base_quantity * abs(signal_value)

        # 4. Calculate the full notional cost of the trade
        notional_cost = final_quantity * entry_price
        commission = self.commission_pct * notional_cost
        total_cost = notional_cost + commission

        # 5. Check if you have enough cash to cover the full cost (no leverage)
        if self.cash < total_cost:
            print(f'INSUFFICIENT CASH, current cash: {self.cash}, cost of trade: {total_cost}')
            return  # Insufficient cash

        elif self.cash > total_cost:

            # 6. Deduct the full cost from cash and open the trade
            print(f'current cash: {self.cash}, cost of trade: {total_cost}, %: {total_cost / self.cash}')
            self.cash -= total_cost

            new_trade_data = {
                'symbol': symbol, 'entry_price': entry_price, 'entry_date': entry_date,
                'quantity': final_quantity, 'stop_loss_price': stop_loss, 'take_profit_price': take_profit,
                'tradeID': trade_ID, 'commission': commission, 'commission_initial': commission, 'trade_type': trade_type,
                'direction': direction
            }

            self.open_trades.append(Trades.create_from_dict(new_trade_data))
            # print("Trade created, ", new_trade_data) # logging purposes

    def close_trade(self, trade: Trades, exit_date: pd.Timestamp, exit_price: float, exit_reason: str):
        """
        Closes trade given parameters by user
        closes using close_trades method in Trades class, converts to dict and appends to closed_trades,
        removed from open_trades, update current liquid cash, update trade final PNL to match total proceeds
        Args: exit date (input row['date'], exit price (close?), exit reason (e.g. STOP LOSS)
        """
        # 1. Calculate gross PnL
        if trade.direction == "LONG":
            gross_pnl = (exit_price - trade.entry_price) * trade.quantity
        else:  # SHORT
            gross_pnl = (trade.entry_price - exit_price) * trade.quantity

        # 2. Calculate final commission and net PnL
        commission_final = self.commission_pct * (exit_price * trade.quantity)
        net_pnl = gross_pnl - trade.commission_initial - commission_final

        # 3. Calculate the initial capital committed to the trade
        initial_capital_committed = trade.entry_price * trade.quantity + trade.commission_initial

        # 4. Update cash: Add the initial capital back, plus the net profit (or minus the net loss)
        self.cash += initial_capital_committed + net_pnl

        # 5. Finalize the trade object for logging
        trade.close_trade(exit_price, exit_date, exit_reason, commission_final)
        trade.current_pnl = net_pnl
        self.closed_trades.append(trade.to_dict())
        self.open_trades.remove(trade)
        self.trade_counter += 1
        print("trade closed", trade.to_dict())

    def update_SL(self, trade: Trades, current_price: float): #not in use (yet)
        """
        Applies to trades of type trailing stop. Will be called at every iteration when during backtest run.
        Logic: if current prices are higher than the entry price, SL will be updated according to the difference.
        In the first scenario, we will only update the SL, and ignore the TP for now. In backtesting module, checks
        for SL hit happens before updating of SL
        """

        if trade.trade_type == "TRAILING_SL_FIXED_TP" and trade.entry_price < current_price and trade.direction == "LONG":

            if (current_price - trade.entry_price) + trade.stop_loss > trade.stop_loss:
                trade.update_sl((current_price - trade.entry_price) + trade.stop_loss)

        elif trade.trade_type == "TRAILING_SL_FIXED_TP" and trade.entry_price > current_price and trade.direction == "SHORT":

            if trade.stop_loss - (current_price - trade.entry_price) < trade.stop_loss:
                trade.update_sl(trade.stop_loss - (current_price - trade.entry_price))