Add initial stop loss and take profit support

Author: MDUYN

investing_algorithm_framework/app/algorithm.py

@@ -1413,6 +1413,51 @@ def add_trailing_stop_loss(self, trade, percentage: int) -> None:
         """
         self.trade_service.add_trailing_stop_loss(trade, percentage=percentage)
 
+    def add_take_profit(self, trade, percentage: int) -> None:
+        """
+        Function to add a take profit to a trade. This function will add a
+        take profit to the specified trade. If the take profit is triggered,
+        the trade will be closed.
+
+        Example of take profit:
+            * You buy BTC at $40,000.
+            * You set a TP of 5% → TP level at $42,000 (40,000 + 5%).
+            * BTC rises to $42,000 → TP level reached, trade
+                closes, securing profit.
+
+        Args:
+            trade: Trade - The trade to add the take profit to
+            percentage: int - The take profit of the trade
+
+        Returns:
+            None
+        """
+        self.trade_service.add_take_profit(trade, percentage=percentage)
+
+    def add_trailing_take_profit(self, trade, percentage: int) -> None:
+        """
+        Function to add a trailing take profit to a trade. This function will
+        add a trailing take profit to the specified trade. If the trailing
+        take profit is triggered, the trade will be closed.
+
+        Example of trailing take profit:
+            * You buy BTC at $40,000.
+            * You set a TTP of 5%.
+            * BTC rises to $42,000 → New TTP level at $39,900 (42,000 - 5%).
+            * BTC rises to $45,000 → New TTP level at $42,750.
+            * BTC drops to $42,750 → Trade closes, securing profit.
+
+        Args:
+            trade: Trade - The trade to add the trailing take profit to
+            percentage: int - The trailing take profit of the trade
+
+        Returns:
+            None
+        """
+        self.trade_service.add_trailing_take_profit(
+            trade, percentage=percentage
+        )
+
     def close_trade(self, trade, precision=None) -> None:
         """
         Function to close a trade. This function will close a trade by

investing_algorithm_framework/dependency_container.py

@@ -4,7 +4,8 @@
 from investing_algorithm_framework.infrastructure import SQLOrderRepository, \
     SQLPositionRepository, SQLPortfolioRepository, \
     SQLPortfolioSnapshotRepository, SQLTradeRepository, \
-    SQLPositionSnapshotRepository, PerformanceService, CCXTMarketService
+    SQLPositionSnapshotRepository, PerformanceService, CCXTMarketService, \
+    SQLTradeStopLossRepository, SQLTradeTakeProfitRepository
 from investing_algorithm_framework.services import OrderService, \
     PositionService, PortfolioService, StrategyOrchestratorService, \
     PortfolioConfigurationService, MarketDataSourceService, BacktestService, \
@@ -41,6 +42,10 @@ class DependencyContainer(containers.DeclarativeContainer):
         SQLPortfolioSnapshotRepository
     )
     trade_repository = providers.Factory(SQLTradeRepository)
+    trade_take_profit_repository = providers\
+        .Factory(SQLTradeTakeProfitRepository)
+    trade_stop_loss_repository = providers.Factory(SQLTradeStopLossRepository)
+
     market_service = providers.Factory(
         CCXTMarketService,
         market_credential_service=market_credential_service,
@@ -75,6 +80,8 @@ class DependencyContainer(containers.DeclarativeContainer):
     )
     trade_service = providers.Factory(
         TradeService,
+        trade_take_profit_repository=trade_take_profit_repository,
+        trade_stop_loss_repository=trade_stop_loss_repository,
         configuration_service=configuration_service,
         trade_repository=trade_repository,
         portfolio_repository=portfolio_repository,

investing_algorithm_framework/domain/__init__.py

@@ -19,7 +19,7 @@
     BacktestReport, PortfolioSnapshot, StrategyProfile, \
     BacktestPosition, Trade, MarketCredential, PositionSnapshot, \
     BacktestReportsEvaluation, AppMode, BacktestDateRange, DateRange, \
-    MarketDataType
+    MarketDataType, TradeRiskType, TradeTakeProfit, TradeStopLoss
 from .services import TickerMarketDataSource, OrderBookMarketDataSource, \
     OHLCVMarketDataSource, BacktestMarketDataSource, MarketDataSource, \
     MarketService, MarketCredentialService, AbstractPortfolioSyncService, \
@@ -124,5 +124,8 @@
     "DEFAULT_LOGGING_CONFIG",
     "DATABASE_DIRECTORY_NAME",
     "BACKTESTING_INITIAL_AMOUNT",
-    "MarketDataType"
+    "MarketDataType",
+    "TradeRiskType",
+    "TradeTakeProfit",
+    "TradeStopLoss",
 ]

investing_algorithm_framework/domain/models/__init__.py

@@ -9,7 +9,8 @@
 from .time_frame import TimeFrame
 from .time_interval import TimeInterval
 from .time_unit import TimeUnit
-from .trade import Trade, TradeStatus
+from .trade import Trade, TradeStatus, TradeStopLoss, TradeTakeProfit, \
+    TradeRiskType
 from .trading_data_types import TradingDataType
 from .trading_time_frame import TradingTimeFrame
 from .date_range import DateRange
@@ -40,5 +41,8 @@
     "AppMode",
     "BacktestDateRange",
     "DateRange",
-    "MarketDataType"
+    "MarketDataType",
+    "TradeStopLoss",
+    "TradeTakeProfit",
+    "TradeRiskType",
 ]

investing_algorithm_framework/domain/models/trade/__init__.py

@@ -1,4 +1,13 @@
 from .trade import Trade
 from .trade_status import TradeStatus
+from .trade_stop_loss import TradeStopLoss
+from .trade_take_profit import TradeTakeProfit
+from .trade_risk_type import TradeRiskType
 
-__all__ = ["Trade", "TradeStatus"]
+__all__ = [
+    "Trade",
+    "TradeStatus",
+    "TradeStopLoss",
+    "TradeTakeProfit",
+    "TradeRiskType",
+]

investing_algorithm_framework/domain/models/trade/trade.py

@@ -36,10 +36,8 @@ class Trade(BaseModel):
         created_at (datetime): the datetime when the trade was created
         updated_at (datetime): the datetime when the trade was last updated
         status (str): the status of the trade
-        stop_loss_percentage (float): the stop loss percentage of
-            the trade
-        trailing_stop_loss_percentage (float): the trailing stop
-            loss percentage
+        stop_losses (List[TradeStopLoss]): the stop losses of the trade
+        take_profits (List[TradeTakeProfit]): the take profits of the trade
     """
 
     def __init__(
@@ -59,9 +57,8 @@ def __init__(
         last_reported_price=None,
         high_water_mark=None,
         updated_at=None,
-        stop_loss_percentage=None,
-        trailing_stop_loss_percentage=None,
-        stop_loss_triggered=False,
+        stop_losses=None,
+        take_profits=None,
     ):
         self.id = id
         self.orders = orders
@@ -78,9 +75,8 @@ def __init__(
         self.high_water_mark = high_water_mark
         self.status = status
         self.updated_at = updated_at
-        self.stop_loss_percentage = stop_loss_percentage
-        self.trailing_stop_loss_percentage = trailing_stop_loss_percentage
-        self.stop_loss_triggered = stop_loss_triggered
+        self.stop_losses = stop_losses
+        self.take_profits = take_profits
 
     @property
     def closed_prices(self):
@@ -228,6 +224,49 @@ def is_trailing_stop_loss_triggered(self):
 
         return self.last_reported_price <= stop_loss_price
 
+    def is_take_profit_triggered(self):
+
+        if self.take_profit_percentage is None:
+            return False
+
+        if self.last_reported_price is None:
+            return False
+
+        if self.open_price is None:
+            return False
+
+        take_profit_price = self.open_price * \
+            (1 + (self.take_profit_percentage / 100))
+
+        return self.last_reported_price >= take_profit_price
+
+    def is_trailing_take_profit_triggered(self):
+        """
+        Function to check if the trailing take profit is triggered.
+        The trailing take profit is triggered when the last reported price
+        is greater than or equal to the high water mark times the trailing
+        take profit percentage.
+        """
+
+        if self.trailing_take_profit_percentage is None:
+            return False
+
+        if self.last_reported_price is None:
+            return False
+
+        if self.high_water_mark is None:
+
+            if self.open_price is not None:
+                self.high_water_mark = self.open_price
+            else:
+                return False
+
+        take_profit_price = self.high_water_mark * \
+            (1 + (self.trailing_take_profit_percentage / 100))
+
+        return self.last_reported_price >= take_profit_price
+
+
     def to_dict(self, datetime_format=None):
 
         if datetime_format is not None:

investing_algorithm_framework/domain/models/trade/trade_risk_type.py

@@ -0,0 +1,34 @@
+from enum import Enum
+
+
+class TradeRiskType(Enum):
+    FIXED = "FIXED"
+    TRAILING = "TRAILING"
+
+    @staticmethod
+    def from_string(value: str):
+
+        if isinstance(value, str):
+            for status in TradeRiskType:
+
+                if value.upper() == status.value:
+                    return status
+
+        raise ValueError("Could not convert value to TradeRiskType")
+
+    @staticmethod
+    def from_value(value):
+
+        if isinstance(value, TradeRiskType):
+            for risk_type in TradeRiskType:
+
+                if value == risk_type:
+                    return risk_type
+
+        elif isinstance(value, str):
+            return TradeRiskType.from_string(value)
+
+        raise ValueError("Could not convert value to TradeRiskType")
+
+    def equals(self, other):
+        return TradeRiskType.from_value(other) == self

investing_algorithm_framework/domain/models/trade/trade_stop_loss.py

@@ -0,0 +1,128 @@
+from investing_algorithm_framework.domain.models.base_model import BaseModel
+from investing_algorithm_framework.domain.models.trade.trade_risk_type import \
+    TradeRiskType
+from investing_algorithm_framework.domain.models.trade.trade import Trade
+
+
+class TradeStopLoss(BaseModel):
+    """
+    TradeStopLoss represents a stop loss strategy for a trade.
+
+    Attributes:
+        trade: Trade - the trade that the take profit is for
+        take_profit: float - the take profit percentage
+        take_risk_type: TradeRiskType - the type of trade risk, either
+            trailing or fixed
+        percentage: float - the stop loss percentage
+        sell_percentage: float - the percentage of the trade to sell when the
+            take profit is hit. Default is 100% of the trade. If the take profit percentage is lower than 100% a check must be made that
+            the combined sell percentage of all take profits is less or
+            equal than 100%.
+        sell_amount: float - the amount to sell when the stop loss triggers
+        sold_amount: float - the amount that has been sold
+        high_water_mark: float - the highest price of the trade
+        stop_loss_price: float - the price at which the stop loss triggers
+
+    if trade_risk_type is fixed, the stop loss price is calculated as follows:
+        You buy a stock at $100.
+        You set a 5% stop loss, meaning you will sell if
+            the price drops to $95.
+        If the price rises to $120, the stop loss is not triggered.
+        But if the price keeps falling to $95, the stop loss triggers,
+            and you exit with a $5 loss.
+
+    if trade_risk_type is trailing, the stop loss price is calculated as follows:
+        You buy a stock at $100.
+        You set a 5% trailing stop loss, meaning you will sell if
+            the price drops 5% from its peak at $96
+        If the price rises to $120, the stop loss adjusts
+            to $114 (5% below $120).
+        If the price falls to $114, the position is
+            closed, securing a $14 profit.
+        But if the price keeps rising to $150, the stop
+            loss moves up to $142.50.
+        If the price drops from $150 to $142.50, the stop
+            loss triggers, and you exit with a $42.50 profit.
+    """
+
+    def __init__(
+        self,
+        trade_id: int,
+        take_risk_type: TradeRiskType,
+        percentage: float,
+        open_price: float,
+        total_amount_trade: float,
+        sell_percentage: float = 100,
+        active: bool = True
+    ):
+        self.trade_id = trade_id
+        self.take_risk_type = take_risk_type
+        self.percentage = percentage
+        self.sell_percentage = sell_percentage
+        self.high_water_mark = open_price
+        self.stop_loss_price = self.high_water_mark * \
+            (1 - (self.percentage / 100))
+        self.sell_amount = total_amount_trade * (self.sell_percentage / 100)
+        self.sold_amount = 0
+        self.active = active
+
+    def has_triggered(self, current_price: float) -> bool:
+        """
+        Function to check if the stop loss has triggered.
+        Function always returns False if the stop loss is not active or
+            the sold amount is equal to the sell amount.
+
+        Args:
+            current_price: float - the current price of the trade
+
+        Returns:
+            bool - True if the stop loss has triggered, False otherwise
+        """
+
+        if not self.active or self.sold_amount == self.sell_amount:
+            return False
+
+        if TradeRiskType.FIXED.equals(self.take_risk_type):
+            # Check if the current price is less than the high water mark
+            return current_price <= self.stop_loss_price
+        else:
+            # Check if the current price is less than the stop loss price
+            if current_price <= self.stop_loss_price:
+                return True
+            elif current_price > self.high_water_mark:
+                self.high_water_mark = current_price
+                self.stop_loss_price = self.high_water_mark * \
+                    (1 - (self.percentage / 100))
+
+        return False
+
+    def get_sell_amount(self, trade: Trade) -> float:
+        """
+        Function to calculate the amount to sell based on the
+        sell percentage and the remaining amount of the trade.
+        Keep in mind the moment the take profit triggers, the remaining
+            amount of the trade is used to calculate the sell amount.
+        If the remaining amount is smaller then the trade amount, the
+            trade stop loss stays active. The client that uses the
+            trade stop loss is responsible for setting the trade stop
+            loss to inactive.
+
+        Args:
+            trade: Trade - the trade to calculate the sell amount for
+
+        """
+
+        if not self.active:
+            return 0
+
+        return trade.remaining * (self.sell_percentage / 100)
+
+    def __str__(self) -> str:
+        return (
+            f"TradeStopLoss(trade_id={self.trade_id}, "
+            f"take_profit_type={self.take_profit_type}, "
+            f"percentage={self.percentage}, "
+            f"sell_percentage={self.sell_percentage})"
+            f"high_water_mark={self.high_water_mark}, "
+            f"stop_loss_price={self.stop_loss_price}"
+        )