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 tradeexecutor.state.trade 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) logger.trade(buf.getvalue())
[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