Source code for tradeexecutor.strategy.execution_model

"""Strategy execution model.

Currently supported models

- Backtesting via :py:mod:`tradeexecutor.backtest.backtest_execution`

- Live execution against Uniswap v2 via :py:mod:`tradeexecutor.ethereum.uniswap_v2_execution`
"""
import abc
import datetime
import enum
from typing import List

from tradeexecutor.state.state import State
from tradeexecutor.state.trade import TradeExecution
from tradeexecutor.strategy.routing import RoutingModel, RoutingState


class AutoClosingOrderUnsupported(Exception):
    """Raised when trade execution does not support stop loss/take profit.

    Stop loss handling requires special support from the trade execution engine.
    See :py:meth:`ExecutionModel.is_stop_loss_supported` for more details.
    """


[docs]class ExecutionModel(abc.ABC): """Define how trades are executed. See also :py:class:`tradeexecutor.strategy.mode.ExecutionMode`. """
[docs] @abc.abstractmethod def preflight_check(self): """Check that we can start the trade executor :raise: AssertionError if something is a miss """
[docs] @abc.abstractmethod def initialize(self): """Set up the execution model ready to make trades. Read any on-chain, etc., data to get synced. - Read EVM nonce for the hot wallet from the chain """
[docs] @abc.abstractmethod def get_routing_state_details(self) -> object: """Get needed details to establish a routing state. TODO: API Unfinished """
[docs] @abc.abstractmethod def is_stop_loss_supported(self) -> bool: """Do we support stop-loss/take profit functionality with this execution model? - For backtesting we need to have data stream for candles used to calculate stop loss - For production execution, we need to have special oracle data streams for checking real-time stop loss """
[docs] @abc.abstractmethod def execute_trades(self, ts: datetime.datetime, state: State, trades: List[TradeExecution], routing_model: RoutingModel, routing_state: RoutingState, max_slippage=0.005, check_balances=False, ): """Execute the trades determined by the algo on a designed Uniswap v2 instance. :param ts: Timestamp of the trade cycle. :param universe: Current trading universe for this cycle. :param state: State of the trade executor. :param trades: List of trades decided by the strategy. Will be executed and modified in place. :param routing_model: Routing model how to execute the trades :param routing_state: State of already made on-chain transactions and such on this cycle :param max_slippage: Max slippage % allowed on trades before trade execution fails. :param check_balances: Check that on-chain accounts have enough balance before creating transaction objects. Useful during unit tests to spot issues in trade routing. """
[docs] @abc.abstractmethod def repair_unconfirmed_trades(self, state: State) -> List[TradeExecution]: """Repair unconfirmed trades. Repair trades that failed to properly broadcast or confirm due to blockchain node issues. :return: List of fixed trades """
[docs]class TradeExecutionType(enum.Enum): """Default execution options. What kind of trade instruction execution model the strategy does. Give options for command line parameters and such. TODO: Clean up unused options. """ #: Does not make any trades, just captures and logs them dummy = "dummy" #: Server-side normal Ethereum private eky account uniswap_v2_hot_wallet = "uniswap_v2_hot_wallet" #: Trading using Enzyme Protocol pool, single oracle mode single_oracle_pooled = "single_oracle_pooled" #: Trading using oracle network, oracles form a consensus using a judge smart contract multi_oracle_judged = "multi_oracle_judged" #: Simulate execution using backtest data backtest = "backtest"