Source code for tradeexecutor.strategy.stop_loss

"""Stop loss trade logic.

Logic for managing position stop loss/take profit signals.
import datetime
import logging
from decimal import Decimal
from io import StringIO
from typing import List, Dict

from tradingstrategy.candle import CandleSampleUnavailable

from tradeexecutor.state.position import TradingPosition, TriggerPriceUpdate, CLOSED_POSITION_DUST_EPSILON
from tradeexecutor.state.state import State
from import TradeExecution, TradeType
from tradeexecutor.strategy.pandas_trader.position_manager import PositionManager
from tradeexecutor.strategy.pricing_model import PricingModel

logger = logging.getLogger(__name__)

[docs]def report_position_triggered( position: TradingPosition, condition: TradeType, trigger_price: float, mid_price: float, expected_price: float, ): """Write a trade logging output report on a position trigger.""" buf = StringIO() name = "Stop loss" if condition == TradeType.stop_loss else "Take profit" size = position.get_quantity() print(f"{name} triggered", file=buf) print("", file=buf) print(f" Pair: {position.pair}", file=buf) print(f" Size: {size} {position.pair.base.token_symbol}", file=buf) print(f" Trigger price: {trigger_price} USD", file=buf) print(f" Current mid price: {mid_price} USD", file=buf) print(f" Expected avg closing price: {expected_price} USD", file=buf)
[docs]def check_position_triggers( position_manager: PositionManager, ) -> List[TradeExecution]: """Generate trades that depend on real-time price signals. - Stop loss - Take profit What does this do: - Get the real-time price of an assets that are currently hold in the portfolio - Update dynamic stop loss/take profits like trailing stop loss - Use mid-price to check for the trigger price threshold - Generate stop loss/take profit signals for trades See related position attributes - :py:attr:`tradeexecutor.state.position.TradingPosition.stop_loss` - :py:attr:`tradeexecutor.state.position.TradingPosition.take_profit` - :py:attr:`tradeexecutor.state.position.TradingPosition.trailing_stop_loss` :param position_manager: Encapsulates the current state, universe for closing positions :param epsilon: The rounding error to zero :return: List of triggered trades for all positions, like market sell on a stop loss triggered. """ ts: datetime.datetime = position_manager.timestamp state: State = position_manager.state pricing_model: PricingModel = position_manager.pricing_model positions = state.portfolio.get_open_positions() trades = [] for p in positions: if not p.has_trigger_conditions(): # This position does not have take profit/stop loss set continue assert any([ p.is_long(), p.is_short() and p.is_leverage(), ]), "Trigger only supports long and leveraged short positions" size = p.get_quantity() if size == 0: logger.warning("Encountered open position without token quantity: %s. Quantity is %s.", p, size) continue # TODO: Tracking down math bug if not isinstance(size, Decimal): logger.warning("Got bad size %s: %s", size, size.__class__) size = Decimal(size) spot_pair = p.pair.get_pricing_pair() try: mid_price = pricing_model.get_mid_price(ts, spot_pair) expected_sell_price = pricing_model.get_sell_price(ts, spot_pair, abs(size)) except CandleSampleUnavailable: # Backtest does not have price available for this timestamp, # because there has not been any trades. # Because there has not been any trades, we assume price has not moved # and any position trigger does not need to be executed. continue assert type(mid_price) == float, f"Received bad mid-price: {mid_price} {type(mid_price)}" trigger_type = trigger_price = None stop_loss_before = stop_loss_after = None # Check for trailing stop loss updates if p.trailing_stop_loss_pct: stop_loss_before = p.stop_loss if p.is_long(): new_stop_loss = mid_price * p.trailing_stop_loss_pct else: new_stop_loss = mid_price * (2 - p.trailing_stop_loss_pct) if any([ not p.stop_loss, p.is_long() and new_stop_loss > p.stop_loss, p.is_short() and new_stop_loss < p.stop_loss, ]): stop_loss_before = p.stop_loss stop_loss_after = new_stop_loss # Update dynamic triggers if needed if stop_loss_after is not None: assert stop_loss_after > 0 if p.is_long(): assert stop_loss_after > stop_loss_before else: assert stop_loss_after < stop_loss_before trigger_update = TriggerPriceUpdate( ts, mid_price, stop_loss_before, stop_loss_after, None, # No trailing take profits yet None, ) p.trigger_updates.append(trigger_update) p.stop_loss = stop_loss_after # Check we need to close position for take profit if p.take_profit: if any([ p.is_long() and mid_price >= p.take_profit, p.is_short() and mid_price <= p.take_profit, ]): trigger_type = TradeType.take_profit trigger_price = p.take_profit trades.extend(position_manager.close_position(p, TradeType.take_profit)) # Check we need to close position for stop loss if p.stop_loss: if any([ p.is_long() and mid_price <= p.stop_loss, p.is_short() and mid_price >= p.stop_loss, ]): trigger_type = TradeType.stop_loss trigger_price = p.stop_loss trades.extend(position_manager.close_position(p, TradeType.stop_loss)) if trigger_type: # We got triggered report_position_triggered( p, trigger_type, trigger_price, mid_price, expected_sell_price.price, ) return trades