PositionManager#

tradeexecutor.strategy.pandas_trader.position_manager.PositionManager Python class in Trading Strategy framework.

class PositionManager[source]#

Bases: object

An utility class to open and close new trade positions.

PositionManager hides away the complex logic reason about trades. It is designed to be used in a trading strategy’s decide_trades() function as an utility class to generate trades a list of TradeExecution objects.

It offers a simple interface for trading for people who are used to TradingView’s Pine Script or similar limited trade scripting environment.

PositionManager helps about

How to have up-to-date price information
Setting take profit/stop loss parameters for positions
Converting between US dollar prices, crypto prices
Converting between quantity and value of a trade
Caring whether we have an existing position open for the trading pair already
Shortcut methods for trading strategies that trade only a single trading pair

PositionManager takes the price feed and current execution state as an input and produces the execution instructions to change positions.

Below are some recipes how to use position manager.

Position manager is usually instiated at your decide_trades function as the following:

from typing import List, Dict

from tradeexecutor.state.visualisation import PlotKind
from tradeexecutor.state.trade import TradeExecution
from tradeexecutor.strategy.pricing_model import PricingModel
from tradeexecutor.strategy.pandas_trader.position_manager import PositionManager
from tradeexecutor.state.state import State
from tradingstrategy.universe import Universe


def decide_trades(
        timestamp: pd.Timestamp,
        universe: Universe,
        state: State,
        pricing_model: PricingModel,
        cycle_debug_data: Dict) -> List[TradeExecution]:

    # Create a position manager helper class that allows us easily to create
    # opening/closing trades for different positions
    position_manager = PositionManager(timestamp, universe, state, pricing_model)

How to check if you have an open position using is_any_open() and then open a new position:

# List of any trades we decide on this cycle.
# Because the strategy is simple, there can be
# only zero (do nothing) or 1 (open or close) trades
# decides
trades = []

if not position_manager.is_any_open():
    buy_amount = cash * position_size
    trades += position_manager.open_1x_long(pair, buy_amount)

return trades

How to check the entry price and open quantity of your latest position. See also decimal.Decimal about arbitrary precision decimal numbers in Python.

# Will throw an exception if there is no position open
current_position = position_manager.get_current_position()

# Quantity is the open amount in tokens.
# This is expressed in Python Decimal class,
# because Ethereum token balances are accurate up to 18 decimals
# and this kind of accuracy cannot be expressed in floating point numbers.
quantity = current_position.get_quantity()
assert quantity == Decimal('0.03045760003971992547285959728')

# The current price is the price of the trading pair
# that was recorded on the last price feed sync.
# This is a 64-bit floating point, as the current price
# is always approximation based on market conditions.
price = current_position.get_current_price()
assert price == 1641.6263899583264

# The opening price is the price of the first trade
# that was made for this position. This is the actual
# executed price of the trade, expressed as floating
# point for the convenience.
price = current_position.get_opening_price()
assert price == 1641.6263899583264

__init__(timestamp, universe, state, pricing_model)[source]#

Parameters:

timestamp (Union[datetime, Timestamp]) –
universe (Universe) –
state (State) –
pricing_model (PricingModel) –

Methods

`__init__`(timestamp, universe, state, ...)
`adjust_position`(pair, dollar_amount_delta, ...)	Adjust holdings for a certain position.
`close_all`()	Close all open positions.
`close_position`(position[, trade_type, ...])	Close a single position.
`get_current_portfolio`()	Return the active portfolio of the strategy.
`get_current_position`()	Get the current single position.
`get_current_position_for_pair`(pair)	Get the current open position for a specific trading pair.
`get_last_closed_position`()	Get the position that was last closed.
`get_pair_fee`([pair])	Estimate the trading/LP fees for a trading pair.
`get_trading_pair`(pair_id)	Get a trading pair identifier by its internal id.
`is_any_open`()	Do we have any positions open.
`open_1x_long`(pair, value[, take_profit_pct, ...])	Open a long.

__init__(timestamp, universe, state, pricing_model)[source]#

Parameters:

timestamp (Union[datetime, Timestamp]) –
universe (Universe) –
state (State) –
pricing_model (PricingModel) –

is_any_open()[source]#

Do we have any positions open.

Return type:: bool

get_current_position()[source]#

Get the current single position.

This is a shortcut function for trading strategies that operate only a single trading pair and a single position.

Returns:: Currently open trading position
Raises:: NoSingleOpenPositionError – If you do not have a position open or there are multiple positions open.
Return type:: TradingPosition

get_current_position_for_pair(pair)[source]#

Get the current open position for a specific trading pair.

Returns:

Currently open trading position.

If there is no open position return None.

Parameters:

pair (TradingPairIdentifier) –

Return type:

Optional[TradingPosition]

get_last_closed_position()[source]#

Get the position that was last closed.

If multiple positions are closed at the same time, return a random position.

Example:

last_position = position_manager.get_last_closed_position()
if last_position:
    ago = timestamp - last_position.closed_at
    print(f"Last position was closed {ago}")
else:
    print("Strategy has not decided any position before")

Returns:: None if the strategy has not closed any positions
Return type:: Optional[TradingPosition]

get_current_portfolio()[source]#

Return the active portfolio of the strategy.

Return type:: Portfolio

get_trading_pair(pair_id)[source]#

Get a trading pair identifier by its internal id.

Note that internal integer ids are not stable over multiple trade cycles and might be reset. Always use (chain id, smart contract) for persistent pair identifier.

Returns:: Trading pair information
Parameters:: pair_id (int) –
Return type:: TradingPairIdentifier

get_pair_fee(pair=None)[source]#

Estimate the trading/LP fees for a trading pair.

This information can come either from the exchange itself (Uni v2 compatibles), or from the trading pair (Uni v3).

The return value is used to fill the fee values for any newly opened trades.

Parameters:

pair (Optional[TradingPairIdentifier]) –

Trading pair for which we want to have the fee.

Can be left empty if the underlying exchange is always offering the same fee.

Returns:

The estimated trading fee, expressed as %.

Returns None if the fee information is not available. This can be different from zero fees.

Return type:

Optional[float]

open_1x_long(pair, value, take_profit_pct=None, stop_loss_pct=None, notes=None)[source]#

Open a long.

For simple buy and hold trades
Open a spot market buy.
Checks that there is not existing position - cannot increase position

Parameters:

pair (Union[DEXPair, TradingPairIdentifier]) – Trading pair where we take the position
value (float) – How large position to open, in US dollar terms
take_profit_pct (Optional[float]) – If set, set the position take profit relative to the current market price. 1.0 is the current market price. If asset opening price is $1000, take_profit_pct=1.05 will sell the asset when price reaches $1050.
stop_loss_pct (Optional[float]) – If set, set the position to trigger stop loss relative to the current market price. 1.0 is the current market price. If asset opening price is $1000, stop_loss_pct=0.95 will sell the asset when price reaches 950.
notes (Optional[str]) – Human readable notes for this trade

Returns:

A list of new trades. Opening a position may general several trades for complex DeFi positions, though usually the result contains only a single trade.

Return type:

List[TradeExecution]

adjust_position(pair, dollar_amount_delta, weight, stop_loss=None, take_profit=None)[source]#

Adjust holdings for a certain position.

Used to rebalance positions.

A new position is opened if no existing position is open. If everything is sold, the old position is closed

If the rebalance is sell (dollar_amount_delta is negative), then calculate the quantity of the asset to sell based on the latest available market price on the position.

This method is called by rebalance_portfolio().

Parameters:

pair (TradingPairIdentifier) – Trading pair which position we adjust
dollar_amount_delta (float) – How much we want to increase/decrease the position
weight (float) – What is the weight of the asset in the new target portfolio 0….1. Currently only used to detect condition “sell all” instead of trying to match quantity/price conversion.
stop_loss (Optional[float]) –
Set the stop loss for the position.

Use 0…1 based on the current mid price. E.g. 0.98 = 2% stop loss under the current mid price.
take_profit (Optional[float]) –
Set the take profit for the position.

Use 0…1 based on the current mid price. E.g. 1.02 = 2% take profit over the current mid-price.

Returns:

List of trades to be executed to get to the desired position level.

Return type:

List[TradeExecution]

close_position(position, trade_type=TradeType.rebalance, notes=None, trades_as_list=False)[source]#

Close a single position.

The position may already have piled up selling trades. In this case calling close_position() again on the same position does nothing and None is returned.

Parameters:

position (TradingPosition) – Position to be closed
trade_type (TradeType) – What’s the reason to close the position
notes (Optional[str]) – Human readable notes for this trade
trades_as_list – A migration parameter for the future signature where we are always returning a list of trades.

Returns:

Get list of trades needed to close this position.

If trades_as_list is False. A trade that will close the position fully. If there is nothing left to close, return None.

Otherwise return list of trades.

Return type:

Union[TradeExecution, None, List[TradeExecution]]

close_all()[source]#

Close all open positions.

Returns:: List of trades that will close existing positions
Return type:: List[TradeExecution]